25738155938

Committed 12 May 2026 01:38PM UTC coverage: 39.251% (-35.1%) from 74.366%

Build # 25738155938

Build Type

Pull #1094

github

Committed by

web-flow

Commit Message

Merge 0f3efcca0 into defbbb616

Pull Request Pull Request #1094: feat: embedded vector store

Coverage Stats

2188 of 6152 branches covered (35.57%)

Branch coverage included in aggregate %.

88 of 280 new or added lines in 6 files covered. (31.43%)

3689 existing lines in 165 files now uncovered.

4312 of 10408 relevant lines covered (41.43%)

34.74 hits per line

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

13.93

/src/http/plugins/db.ts

import { createMutexByKey } from '@internal/concurrency'
import {
  getPostgresConnection,
  getServiceKeyUser,
  getTenantConfig,
  TenantConnection,
} from '@internal/database'
import {
  areMigrationsUpToDate,
  DBMigration,
  lastLocalMigrationName,
  progressiveMigrations,
  runMigrationsOnTenant,
  updateTenantMigrationsState,
} from '@internal/database/migrations'
import { ERRORS } from '@internal/errors'
import { logSchema } from '@internal/monitoring'
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()

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?.type,
      })

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

    fastify.addHook('onSend', async (request, reply, payload) => {
      if (request.db) {
        request.db.dispose().catch((e) => {
          logSchema.error(request.log, 'Error disposing db connection', {
            type: 'db-connection',
            tenantId: request.tenantId,
            project: request.tenantId,
            reqId: request.id,
            sbReqId: request.sbReqId,
            error: e,
          })
        })
      }
      return payload
    })

    fastify.addHook('onTimeout', async (request) => {
      if (request.db) {
        request.db.dispose().catch((e) => {
          logSchema.error(request.log, 'Error disposing db connection', {
            type: 'db-connection',
            tenantId: request.tenantId,
            project: request.tenantId,
            reqId: request.id,
            sbReqId: request.sbReqId,
            error: e,
          })
        })
      }
    })

    fastify.addHook('onRequestAbort', async (request) => {
      if (request.db) {
        request.db.dispose().catch((e) => {
          logSchema.error(request.log, 'Error disposing db connection', {
            type: 'db-connection',
            tenantId: request.tenantId,
            project: request.tenantId,
            reqId: request.id,
            sbReqId: request.sbReqId,
            error: e,
          })
        })
      }
    })
  },
  { name: 'db-init' }
)

interface DbSuperUserPluginOptions {
  disableHostCheck?: boolean
  maxConnections?: number
}

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,
        maxConnections: opts.maxConnections,
        operation: () => request.operation?.type,
      })

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

    fastify.addHook('onSend', async (request, reply, payload) => {
      if (request.db) {
        request.db.dispose().catch((e) => {
          logSchema.error(request.log, 'Error disposing db connection', {
            type: 'db-connection',
            tenantId: request.tenantId,
            project: request.tenantId,
            reqId: request.id,
            sbReqId: request.sbReqId,
            error: e,
          })
        })
      }

      return payload
    })

    fastify.addHook('onTimeout', async (request) => {
      if (request.db) {
        request.db.dispose().catch((e) => {
          logSchema.error(request.log, 'Error disposing db connection', {
            type: 'db-connection',
            tenantId: request.tenantId,
            project: request.tenantId,
            reqId: request.id,
            sbReqId: request.sbReqId,
            error: e,
          })
        })
      }
    })

    fastify.addHook('onRequestAbort', async (request) => {
      if (request.db) {
        request.db.dispose().catch((e) => {
          logSchema.error(request.log, 'Error disposing db connection', {
            type: 'db-connection',
            tenantId: request.tenantId,
            project: request.tenantId,
            reqId: request.id,
            sbReqId: request.sbReqId,
            error: e,
          })
        })
      }
    })
  },
  { 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) {
      const migrationsMutex = createMutexByKey<void>()

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

        const tenant = await getTenantConfig(request.tenantId)
        const migrationsUpToDate = await areMigrationsUpToDate(request.tenantId)

        if (tenant.syncMigrationsDone || migrationsUpToDate) {
          return
        }

        await migrationsMutex(request.tenantId, async () => {
          const tenant = await getTenantConfig(request.tenantId)

          if (tenant.syncMigrationsDone) {
            return
          }

          await runMigrationsOnTenant({
            databaseUrl: tenant.databaseUrl,
            tenantId: request.tenantId,
            upToMigration: dbMigrationFreezeAt,
          })
          await updateTenantMigrationsState(request.tenantId)
          tenant.syncMigrationsDone = true
        })
      })
    }

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

        const tenant = await getTenantConfig(request.tenantId)
        const migrationsUpToDate = await areMigrationsUpToDate(request.tenantId)

        // migrations are up to date
        if (tenant.syncMigrationsDone || migrationsUpToDate) {
          return
        }

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

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

supabase / storage / 25738155938

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