30826909455

Committed 03 Aug 2026 03:18PM UTC coverage: 53.88% (-26.5%) from 80.366%

Build # 30826909455

Build Type

Pull #1292

github

Committed by

web-flow

Commit Message

Merge 478fb900a into 1e33d8708

Pull Request Pull Request #1292: feat: new queing system

Coverage Stats

3689 of 7393 branches covered (49.9%)

Branch coverage included in aggregate %.

140 of 332 new or added lines in 38 files covered. (42.17%)

3352 existing lines in 149 files now uncovered.

6920 of 12297 relevant lines covered (56.27%)

76.57 hits per line

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

81.36

/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	getConfig()	25✔
30
31	const migrationSingleFlight = createSingleFlightByKey<keyof typeof DBMigration>()	25✔
32
33	function resolveLatestMigration(
34	localLatest: keyof typeof DBMigration,
35	applied: keyof typeof DBMigration \| undefined
36	): keyof typeof DBMigration {
37	if (	22✔
38	applied &&	49✔
39	DBMigration[applied] !== undefined &&
40	DBMigration[applied] > DBMigration[localLatest]
41	) {
42	return applied	3✔
43	}
44	return localLatest	19✔
45	}
46
47	export const db = fastifyPlugin(	25✔
48	async function db(fastify) {
49	fastify.register(migrations)	33✔
50
51	fastify.decorateRequest('db')	33✔
52
53	fastify.addHook('preHandler', async (request) => {	33✔
54	const adminUser = await getServiceKeyUser(request.tenantId)	3✔
55	const userPayload = request.jwtPayload	3✔
56
57	if (!userPayload) {	3!
58	throw ERRORS.AccessDenied('JWT payload is missing')	×
59	}
60
61	request.db = await getPostgresConnection({	3✔
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,
UNCOV 72	operation: () => request.operation,	×
73	})
74
75	// Connect abort signal to DB connection for query cancellation
76	if (databaseEnableQueryCancellation && request.signals) {	3✔
77	request.db.setAbortSignal(request.signals.disconnect.signal)	1✔
78	}
79	})
80
81	fastify.addHook('onSend', async (request, reply, payload) => {	33✔
82	request.db?.dispose()	3✔
83	return payload	3✔
84	})
85
86	fastify.addHook('onTimeout', async (request) => {	33✔
87	request.db?.dispose()	×
88	})
89
90	fastify.addHook('onRequestAbort', async (request) => {	33✔
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>(	25✔
102	async function dbSuperUser(fastify, opts) {
103	fastify.register(migrations)	37✔
104	fastify.decorateRequest('db')	37✔
105
106	fastify.addHook('preHandler', async (request) => {	37✔
107	const adminUser = await getServiceKeyUser(request.tenantId)	20✔
108
109	request.db = await getPostgresConnection({	20✔
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,
UNCOV 118	operation: () => request.operation,	×
119	})
120
121	// Connect abort signal to DB connection for query cancellation
122	if (databaseEnableQueryCancellation && request.signals) {	20✔
123	request.db.setAbortSignal(request.signals.disconnect.signal)	1✔
124	}
125	})
126
127	fastify.addHook('onSend', async (request, reply, payload) => {	37✔
128	request.db?.dispose()	24✔
129	return payload	24✔
130	})
131
132	fastify.addHook('onTimeout', async (request) => {	37✔
133	request.db?.dispose()	×
134	})
135
136	fastify.addHook('onRequestAbort', async (request) => {	37✔
UNCOV 137	request.db?.dispose()	×
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(	25✔
147	async function migrations(fastify) {
148	fastify.addHook('preHandler', async (req) => {	70✔
149	if (isMultitenant) {	25✔
150	const { migrationVersion } = await getTenantConfig(req.tenantId)	18✔
151	req.latestMigration = migrationVersion	18✔
152	return	18✔
153	}
154
155	req.latestMigration = await lastLocalMigrationName()	7✔
156	})
157
158	if (dbMigrationStrategy === MultitenantMigrationStrategy.ON_REQUEST) {	70✔
159	fastify.addHook('preHandler', async (request) => {	63✔
160	// migrations are handled via async migrations
161	if (!isMultitenant) {	18!
UNCOV 162	return	×
163	}
164
165	const tenant = await getTenantConfig(request.tenantId)	18✔
166	if (tenant.syncMigrationsDone) {	18✔
167	request.latestMigration = resolveLatestMigration(	4✔
168	await lastLocalMigrationName(),
169	tenant.migrationVersion
170	)
171	return	4✔
172	}
173
174	const latestMigration = await migrationSingleFlight(request.tenantId, async () => {	14✔
175	const localLatest = await lastLocalMigrationName()	10✔
176	const migrationsUpToDate = await areMigrationsUpToDate(request.tenantId)	10✔
177
178	if (!migrationsUpToDate) {	10✔
179	await runMigrationsOnTenant({	9✔
180	databaseUrl: tenant.databaseUrl,
181	tenantId: request.tenantId,
182	upToMigration: dbMigrationFreezeAt,
183	})
184	}
185
186	const refreshedTenant = await getTenantConfig(request.tenantId)	9✔
187	const resolvedMigration = resolveLatestMigration(	9✔
188	resolveLatestMigration(localLatest, tenant.migrationVersion),
189	refreshedTenant.migrationVersion
190	)
191
192	if (!migrationsUpToDate) {	9✔
193	await updateTenantMigrationsState(request.tenantId, {	8✔
194	migration: resolvedMigration,
195	state: TenantMigrationStatus.COMPLETED,
196	})
197	}
198
199	refreshedTenant.migrationVersion = resolvedMigration	9✔
200	refreshedTenant.migrationStatus = TenantMigrationStatus.COMPLETED	9✔
201	refreshedTenant.syncMigrationsDone = true	9✔
202
203	return resolvedMigration	9✔
204	})
205
206	tenant.migrationVersion = latestMigration	12✔
207	tenant.migrationStatus = TenantMigrationStatus.COMPLETED	12✔
208	tenant.syncMigrationsDone = true	12✔
209	request.latestMigration = latestMigration	12✔
210	})
211	}
212
213	if (dbMigrationStrategy === MultitenantMigrationStrategy.PROGRESSIVE) {	70✔
214	fastify.addHook('preHandler', async (request) => {	7✔
215	if (!isMultitenant) {	7!
216	return	7✔
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 / 30826909455

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