30631331327

Committed 31 Jul 2026 12:38PM UTC coverage: 80.392% (-0.1%) from 80.501%

Build # 30631331327

Build Type

Pull #1281

github

Committed by

web-flow

Commit Message

Merge 9cdd5dc95 into fe232f685

Pull Request Pull Request #1281: fix(vector): return the configured S3 vectors client instead of re-wrapping it

Coverage Stats

5633 of 7542 branches covered (74.69%)

Branch coverage included in aggregate %.

1 of 1 new or added line in 1 file covered. (100.0%)

20 existing lines in 3 files now uncovered.

10759 of 12848 relevant lines covered (83.74%)

414.11 hits per line

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

85.59

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

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