30105224435

Committed 24 Jul 2026 03:25PM UTC coverage: 59.634% (-20.8%) from 80.45%

Build # 30105224435

Build Type

Pull #1261

github

Committed by

web-flow

Commit Message

Merge 56e9321c4 into 42bb7886c

Pull Request Pull Request #1261: chore(deps): bump @opentelemetry/instrumentation-pg from 0.64.0 to 0.72.0

Coverage Stats

3705 of 7077 branches covered (52.35%)

Branch coverage included in aggregate %.

7681 of 12016 relevant lines covered (63.92%)

370.0 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

69.49

/src/http/plugins/db.ts

import { createSingleFlightByKey } from '@internal/concurrency'
import {
  getPostgresConnection,
  getServiceKeyUser,
  getTenantConfig,
  type TenantConnection,
  TenantMigrationStatus,
} from '@internal/database'
import {
  areMigrationsUpToDate,
  DBMigration,
  lastLocalMigrationName,
  progressiveMigrations,
  runMigrationsOnTenant,
  updateTenantMigrationsState,
} from '@internal/database/migrations'
import { ERRORS } from '@internal/errors'
import fastifyPlugin from 'fastify-plugin'
import { getConfig, MultitenantMigrationStrategy } from '../../config'

declare module 'fastify' {
  interface FastifyRequest {
    db: TenantConnection
    latestMigration?: keyof typeof DBMigration
  }
}

const { databaseEnableQueryCancellation, dbMigrationStrategy, isMultitenant, dbMigrationFreezeAt } =
  getConfig()

const migrationSingleFlight = createSingleFlightByKey<keyof typeof DBMigration>()

function resolveLatestMigration(
  localLatest: keyof typeof DBMigration,
  applied: keyof typeof DBMigration | undefined
): keyof typeof DBMigration {
  if (
    applied &&
    DBMigration[applied] !== undefined &&
    DBMigration[applied] > DBMigration[localLatest]
  ) {
    return applied
  }
  return localLatest
}

export const db = fastifyPlugin(
  async function db(fastify) {
    fastify.register(migrations)

    fastify.decorateRequest('db')

    fastify.addHook('preHandler', async (request) => {
      const adminUser = await getServiceKeyUser(request.tenantId)
      const userPayload = request.jwtPayload

      if (!userPayload) {
        throw ERRORS.AccessDenied('JWT payload is missing')
      }

      request.db = await getPostgresConnection({
        user: {
          payload: userPayload,
          jwt: request.jwt,
        },
        superUser: adminUser,
        tenantId: request.tenantId,
        host: request.headers['x-forwarded-host'] as string,
        headers: request.headers,
        path: request.url,
        method: request.method,
        operation: () => request.operation,
      })

      // Connect abort signal to DB connection for query cancellation
      if (databaseEnableQueryCancellation && request.signals) {
        request.db.setAbortSignal(request.signals.disconnect.signal)
      }
    })

    fastify.addHook('onSend', async (request, reply, payload) => {
      request.db?.dispose()
      return payload
    })

    fastify.addHook('onTimeout', async (request) => {
      request.db?.dispose()
    })

    fastify.addHook('onRequestAbort', async (request) => {
      request.db?.dispose()
    })
  },
  { name: 'db-init' }
)

interface DbSuperUserPluginOptions {
  disableHostCheck?: boolean
}

export const dbSuperUser = fastifyPlugin<DbSuperUserPluginOptions>(
  async function dbSuperUser(fastify, opts) {
    fastify.register(migrations)
    fastify.decorateRequest('db')

    fastify.addHook('preHandler', async (request) => {
      const adminUser = await getServiceKeyUser(request.tenantId)

      request.db = await getPostgresConnection({
        user: adminUser,
        superUser: adminUser,
        tenantId: request.tenantId,
        host: request.headers['x-forwarded-host'] as string,
        path: request.url,
        method: request.method,
        headers: request.headers,
        disableHostCheck: opts.disableHostCheck,
        operation: () => request.operation,
      })

      // Connect abort signal to DB connection for query cancellation
      if (databaseEnableQueryCancellation && request.signals) {
        request.db.setAbortSignal(request.signals.disconnect.signal)
      }
    })

    fastify.addHook('onSend', async (request, reply, payload) => {
      request.db?.dispose()
      return payload
    })

    fastify.addHook('onTimeout', async (request) => {
      request.db?.dispose()
    })

    fastify.addHook('onRequestAbort', async (request) => {
      request.db?.dispose()
    })
  },
  { name: 'db-superuser-init' }
)

/**
 * Handle database migration for multitenant applications when a request is made
 */
export const migrations = fastifyPlugin(
  async function migrations(fastify) {
    fastify.addHook('preHandler', async (req) => {
      if (isMultitenant) {
        const { migrationVersion } = await getTenantConfig(req.tenantId)
        req.latestMigration = migrationVersion
        return
      }

      req.latestMigration = await lastLocalMigrationName()
    })

    if (dbMigrationStrategy === MultitenantMigrationStrategy.ON_REQUEST) {
      fastify.addHook('preHandler', async (request) => {
        // migrations are handled via async migrations
        if (!isMultitenant) {
          return
        }

        const tenant = await getTenantConfig(request.tenantId)
        if (tenant.syncMigrationsDone) {
          request.latestMigration = resolveLatestMigration(
            await lastLocalMigrationName(),
            tenant.migrationVersion
          )
          return
        }

        const latestMigration = await migrationSingleFlight(request.tenantId, async () => {
          const localLatest = await lastLocalMigrationName()
          const migrationsUpToDate = await areMigrationsUpToDate(request.tenantId)

          if (!migrationsUpToDate) {
            await runMigrationsOnTenant({
              databaseUrl: tenant.databaseUrl,
              tenantId: request.tenantId,
              upToMigration: dbMigrationFreezeAt,
            })
          }

          const refreshedTenant = await getTenantConfig(request.tenantId)
          const resolvedMigration = resolveLatestMigration(
            resolveLatestMigration(localLatest, tenant.migrationVersion),
            refreshedTenant.migrationVersion
          )

          if (!migrationsUpToDate) {
            await updateTenantMigrationsState(request.tenantId, {
              migration: resolvedMigration,
              state: TenantMigrationStatus.COMPLETED,
            })
          }

          refreshedTenant.migrationVersion = resolvedMigration
          refreshedTenant.migrationStatus = TenantMigrationStatus.COMPLETED
          refreshedTenant.syncMigrationsDone = true

          return resolvedMigration
        })

        tenant.migrationVersion = latestMigration
        tenant.migrationStatus = TenantMigrationStatus.COMPLETED
        tenant.syncMigrationsDone = true
        request.latestMigration = latestMigration
      })
    }

    if (dbMigrationStrategy === MultitenantMigrationStrategy.PROGRESSIVE) {
      fastify.addHook('preHandler', async (request) => {
        if (!isMultitenant) {
          return
        }

        const tenant = await getTenantConfig(request.tenantId)
        if (tenant.syncMigrationsDone) {
          return
        }

        // migrations are up to date
        if (await areMigrationsUpToDate(request.tenantId)) {
          tenant.syncMigrationsDone = true
          return
        }

        progressiveMigrations.addTenant(request.tenantId)
      })
    }
  },
  { name: 'db-migrations' }
)

1	import { createSingleFlightByKey } from '@internal/concurrency'
2	import {
3	getPostgresConnection,
4	getServiceKeyUser,
5	getTenantConfig,
6	type TenantConnection,
7	TenantMigrationStatus,
8	} from '@internal/database'
9	import {
10	areMigrationsUpToDate,
11	DBMigration,
12	lastLocalMigrationName,
13	progressiveMigrations,
14	runMigrationsOnTenant,
15	updateTenantMigrationsState,
16	} from '@internal/database/migrations'
17	import { ERRORS } from '@internal/errors'
18	import fastifyPlugin from 'fastify-plugin'
19	import { getConfig, MultitenantMigrationStrategy } from '../../config'
20
21	declare module 'fastify' {
22	interface FastifyRequest {
23	db: TenantConnection
24	latestMigration?: keyof typeof DBMigration
25	}
26	}
27
28	const { databaseEnableQueryCancellation, dbMigrationStrategy, isMultitenant, dbMigrationFreezeAt } =	29✔
29	getConfig()
30
31	const migrationSingleFlight = createSingleFlightByKey<keyof typeof DBMigration>()	29✔
32
33	function resolveLatestMigration(
34	localLatest: keyof typeof DBMigration,
35	applied: keyof typeof DBMigration \| undefined
36	): keyof typeof DBMigration {
37	if (	8!
38	applied &&	12✔
39	DBMigration[applied] !== undefined &&
40	DBMigration[applied] > DBMigration[localLatest]
41	) {
42	return applied	×
43	}
44	return localLatest	8✔
45	}
46
47	export const db = fastifyPlugin(	29✔
48	async function db(fastify) {
49	fastify.register(migrations)	4,640✔
50
51	fastify.decorateRequest('db')	4,640✔
52
53	fastify.addHook('preHandler', async (request) => {	4,640✔
54	const adminUser = await getServiceKeyUser(request.tenantId)	1,085✔
55	const userPayload = request.jwtPayload	1,085✔
56
57	if (!userPayload) {	1,085!
58	throw ERRORS.AccessDenied('JWT payload is missing')	×
59	}
60
61	request.db = await getPostgresConnection({	1,085✔
62	user: {
63	payload: userPayload,
64	jwt: request.jwt,
65	},
66	superUser: adminUser,
67	tenantId: request.tenantId,
68	host: request.headers['x-forwarded-host'] as string,
69	headers: request.headers,
70	path: request.url,
71	method: request.method,
72	operation: () => request.operation,	2,400✔
73	})
74
75	// Connect abort signal to DB connection for query cancellation
76	if (databaseEnableQueryCancellation && request.signals) {	1,083!
77	request.db.setAbortSignal(request.signals.disconnect.signal)	×
78	}
79	})
80
81	fastify.addHook('onSend', async (request, reply, payload) => {	4,640✔
82	request.db?.dispose()	1,087✔
83	return payload	1,087✔
84	})
85
86	fastify.addHook('onTimeout', async (request) => {	4,640✔
87	request.db?.dispose()	×
88	})
89
90	fastify.addHook('onRequestAbort', async (request) => {	4,640✔
91	request.db?.dispose()	×
92	})
93	},
94	{ name: 'db-init' }
95	)
96
97	interface DbSuperUserPluginOptions {
98	disableHostCheck?: boolean
99	}
100
101	export const dbSuperUser = fastifyPlugin<DbSuperUserPluginOptions>(	29✔
102	async function dbSuperUser(fastify, opts) {
103	fastify.register(migrations)	1,763✔
104	fastify.decorateRequest('db')	1,763✔
105
106	fastify.addHook('preHandler', async (request) => {	1,763✔
107	const adminUser = await getServiceKeyUser(request.tenantId)	200✔
108
109	request.db = await getPostgresConnection({	200✔
110	user: adminUser,
111	superUser: adminUser,
112	tenantId: request.tenantId,
113	host: request.headers['x-forwarded-host'] as string,
114	path: request.url,
115	method: request.method,
116	headers: request.headers,
117	disableHostCheck: opts.disableHostCheck,
118	operation: () => request.operation,	65✔
119	})
120
121	// Connect abort signal to DB connection for query cancellation
122	if (databaseEnableQueryCancellation && request.signals) {	200!
123	request.db.setAbortSignal(request.signals.disconnect.signal)	×
124	}
125	})
126
127	fastify.addHook('onSend', async (request, reply, payload) => {	1,763✔
128	request.db?.dispose()	261✔
129	return payload	261✔
130	})
131
132	fastify.addHook('onTimeout', async (request) => {	1,763✔
133	request.db?.dispose()	×
134	})
135
136	fastify.addHook('onRequestAbort', async (request) => {	1,763✔
137	request.db?.dispose()	1✔
138	})
139	},
140	{ name: 'db-superuser-init' }
141	)
142
143	/**
144	* Handle database migration for multitenant applications when a request is made
145	*/
146	export const migrations = fastifyPlugin(	29✔
147	async function migrations(fastify) {
148	fastify.addHook('preHandler', async (req) => {	6,403✔
149	if (isMultitenant) {	1,285✔
150	const { migrationVersion } = await getTenantConfig(req.tenantId)	4✔
151	req.latestMigration = migrationVersion	4✔
152	return	4✔
153	}
154
155	req.latestMigration = await lastLocalMigrationName()	1,281✔
156	})
157
158	if (dbMigrationStrategy === MultitenantMigrationStrategy.ON_REQUEST) {	6,403!
159	fastify.addHook('preHandler', async (request) => {	6,403✔
160	// migrations are handled via async migrations
161	if (!isMultitenant) {	1,285✔
162	return	1,281✔
163	}
164
165	const tenant = await getTenantConfig(request.tenantId)	4✔
166	if (tenant.syncMigrationsDone) {	4!
167	request.latestMigration = resolveLatestMigration(	×
168	await lastLocalMigrationName(),
169	tenant.migrationVersion
170	)
171	return	×
172	}
173
174	const latestMigration = await migrationSingleFlight(request.tenantId, async () => {	4✔
175	const localLatest = await lastLocalMigrationName()	4✔
176	const migrationsUpToDate = await areMigrationsUpToDate(request.tenantId)	4✔
177
178	if (!migrationsUpToDate) {	4!
179	await runMigrationsOnTenant({	4✔
180	databaseUrl: tenant.databaseUrl,
181	tenantId: request.tenantId,
182	upToMigration: dbMigrationFreezeAt,
183	})
184	}
185
186	const refreshedTenant = await getTenantConfig(request.tenantId)	4✔
187	const resolvedMigration = resolveLatestMigration(	4✔
188	resolveLatestMigration(localLatest, tenant.migrationVersion),
189	refreshedTenant.migrationVersion
190	)
191
192	if (!migrationsUpToDate) {	4!
193	await updateTenantMigrationsState(request.tenantId, {	4✔
194	migration: resolvedMigration,
195	state: TenantMigrationStatus.COMPLETED,
196	})
197	}
198
199	refreshedTenant.migrationVersion = resolvedMigration	4✔
200	refreshedTenant.migrationStatus = TenantMigrationStatus.COMPLETED	4✔
201	refreshedTenant.syncMigrationsDone = true	4✔
202
203	return resolvedMigration	4✔
204	})
205
206	tenant.migrationVersion = latestMigration	4✔
207	tenant.migrationStatus = TenantMigrationStatus.COMPLETED	4✔
208	tenant.syncMigrationsDone = true	4✔
209	request.latestMigration = latestMigration	4✔
210	})
211	}
212
213	if (dbMigrationStrategy === MultitenantMigrationStrategy.PROGRESSIVE) {	6,403!
214	fastify.addHook('preHandler', async (request) => {	×
215	if (!isMultitenant) {	×
216	return	×
217	}
218
219	const tenant = await getTenantConfig(request.tenantId)	×
220	if (tenant.syncMigrationsDone) {	×
221	return	×
222	}
223
224	// migrations are up to date
225	if (await areMigrationsUpToDate(request.tenantId)) {	×
226	tenant.syncMigrationsDone = true	×
227	return	×
228	}
229
230	progressiveMigrations.addTenant(request.tenantId)	×
231	})
232	}
233	},
234	{ name: 'db-migrations' }
235	)

supabase / storage / 30105224435

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous