30832958929

Committed 03 Aug 2026 04:36PM UTC coverage: 53.989% (-26.5%) from 80.494%

Build # 30832958929

Build Type

Pull #1292

github

Committed by

web-flow

Commit Message

Merge 477971401 into 42e86a3a7

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

Coverage Stats

3711 of 7412 branches covered (50.07%)

Branch coverage included in aggregate %.

140 of 337 new or added lines in 38 files covered. (41.54%)

3365 existing lines in 150 files now uncovered.

6941 of 12318 relevant lines covered (56.35%)

77.58 hits per line

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

80.87

/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 type { FastifyInstance } from 'fastify'
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)
      }
    })

    registerConnectionCleanupHooks(fastify)
  },
  { 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)
      }
    })

    registerConnectionCleanupHooks(fastify)
  },
  { name: 'db-superuser-init' }
)

function registerConnectionCleanupHooks(fastify: FastifyInstance) {
  fastify.addHook('onSend', (request, _reply, payload, done) => {
    request.db?.dispose()
    done(null, payload)
  })

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

  fastify.addHook('onRequestAbort', (request, done) => {
    request.db?.dispose()
    done()
  })
}

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

supabase / storage / 30832958929

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