21726771076

Committed 05 Feb 2026 08:10PM UTC coverage: 75.896% (-0.03%) from 75.923%

Build # 21726771076

Build Type

push

github

Committed by

web-flow

Commit Message

fix: aws stream buffer (#850)

Coverage Stats

2135 of 3105 branches covered (68.76%)

Branch coverage included in aggregate %.

9 of 13 new or added lines in 4 files covered. (69.23%)

6 existing lines in 2 files now uncovered.

26153 of 34167 relevant lines covered (76.54%)

94.87 hits per line

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

69.2

/src/http/plugins/db.ts

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

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', null)

    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',
            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',
            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',
            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', null)

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

supabase / storage / 21726771076

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