27200912234

Committed 09 Jun 2026 10:46AM UTC coverage: 78.038% (+1.6%) from 76.452%

Build # 27200912234

Build Type

push

github

Committed by

web-flow

Commit Message

fix: replace knex with pg directly (#1096)

Signed-off-by: ferhat elmas <elmas.ferhat@gmail.com>

Coverage Stats

4835 of 6782 branches covered (71.29%)

Branch coverage included in aggregate %.

1499 of 1771 new or added lines in 41 files covered. (84.64%)

5 existing lines in 5 files now uncovered.

9673 of 11809 relevant lines covered (81.91%)

416.34 hits per line

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

71.0

/src/http/plugins/db.ts

import { createMutexByKey } from '@internal/concurrency'
import {
  getPostgresConnection,
  getServiceKeyUser,
  getTenantConfig,
  PgTenantConnection,
} 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: PgTenantConnection
    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) => {
      disposeRequestConnections(request)
      return payload
    })

    fastify.addHook('onTimeout', async (request) => {
      disposeRequestConnections(request)
    })

    fastify.addHook('onRequestAbort', async (request) => {
      disposeRequestConnections(request)
    })
  },
  { 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) => {
      disposeRequestConnections(request)
      return payload
    })

    fastify.addHook('onTimeout', async (request) => {
      disposeRequestConnections(request)
    })

    fastify.addHook('onRequestAbort', async (request) => {
      disposeRequestConnections(request)
    })
  },
  { name: 'db-superuser-init' }
)

function disposeRequestConnections(request: {
  db?: PgTenantConnection
  log: Parameters<typeof logSchema.error>[0]
  tenantId: string
  id: string
  sbReqId?: string
}) {
  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,
    })
  })
}

/**
 * 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	PgTenantConnection,
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: PgTenantConnection
24	latestMigration?: keyof typeof DBMigration
25	}
26	}
27
28	const { databaseEnableQueryCancellation, dbMigrationStrategy, isMultitenant, dbMigrationFreezeAt } =	31✔
29	getConfig()
30
31	export const db = fastifyPlugin(	31✔
32	async function db(fastify) {
33	fastify.register(migrations)	4,811✔
34
35	fastify.decorateRequest('db')	4,811✔
36
37	fastify.addHook('preHandler', async (request) => {	4,811✔
38	const adminUser = await getServiceKeyUser(request.tenantId)	1,033✔
39	const userPayload = request.jwtPayload	1,033✔
40
41	if (!userPayload) {	1,033!
42	throw ERRORS.AccessDenied('JWT payload is missing')	×
43	}
44
45	request.db = await getPostgresConnection({	1,033✔
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,
56	operation: () => request.operation?.type,	2,438✔
57	})
58
59	// Connect abort signal to DB connection for query cancellation
60	if (request.signals?.disconnect?.signal && databaseEnableQueryCancellation) {	1,031!
61	request.db.setAbortSignal(request.signals.disconnect.signal)	×
62	}
63	})
64
65	fastify.addHook('onSend', async (request, reply, payload) => {	4,811✔
66	disposeRequestConnections(request)	1,029✔
67	return payload	1,029✔
68	})
69
70	fastify.addHook('onTimeout', async (request) => {	4,811✔
NEW 71	disposeRequestConnections(request)	×
72	})
73
74	fastify.addHook('onRequestAbort', async (request) => {	4,811✔
75	disposeRequestConnections(request)	1✔
76	})
77	},
78	{ name: 'db-init' }
79	)
80
81	interface DbSuperUserPluginOptions {
82	disableHostCheck?: boolean
83	maxConnections?: number
84	}
85
86	export const dbSuperUser = fastifyPlugin<DbSuperUserPluginOptions>(	31✔
87	async function dbSuperUser(fastify, opts) {
88	fastify.register(migrations)	1,723✔
89	fastify.decorateRequest('db')	1,723✔
90
91	fastify.addHook('preHandler', async (request) => {	1,723✔
92	const adminUser = await getServiceKeyUser(request.tenantId)	200✔
93
94	request.db = await getPostgresConnection({	200✔
95	user: adminUser,
96	superUser: adminUser,
97	tenantId: request.tenantId,
98	host: request.headers['x-forwarded-host'] as string,
99	path: request.url,
100	method: request.method,
101	headers: request.headers,
102	disableHostCheck: opts.disableHostCheck,
103	maxConnections: opts.maxConnections,
104	operation: () => request.operation?.type,	65✔
105	})
106
107	// Connect abort signal to DB connection for query cancellation
108	if (request.signals?.disconnect?.signal && databaseEnableQueryCancellation) {	200!
109	request.db.setAbortSignal(request.signals.disconnect.signal)	×
110	}
111	})
112
113	fastify.addHook('onSend', async (request, reply, payload) => {	1,723✔
114	disposeRequestConnections(request)	261✔
115	return payload	261✔
116	})
117
118	fastify.addHook('onTimeout', async (request) => {	1,723✔
NEW 119	disposeRequestConnections(request)	×
120	})
121
122	fastify.addHook('onRequestAbort', async (request) => {	1,723✔
123	disposeRequestConnections(request)	1✔
124	})
125	},
126	{ name: 'db-superuser-init' }
127	)
128
129	function disposeRequestConnections(request: {
130	db?: PgTenantConnection
131	log: Parameters<typeof logSchema.error>[0]
132	tenantId: string
133	id: string
134	sbReqId?: string
135	}) {
136	request.db?.dispose().catch((e) => {	1,292✔
NEW 137	logSchema.error(request.log, 'Error disposing db connection', {	×
138	type: 'db-connection',
139	tenantId: request.tenantId,
140	project: request.tenantId,
141	reqId: request.id,
142	sbReqId: request.sbReqId,
143	error: e,
144	})
145	})
146	}
147
148	/**
149	* Handle database migration for multitenant applications when a request is made
150	*/
151	export const migrations = fastifyPlugin(	31✔
152	async function migrations(fastify) {
153	fastify.addHook('preHandler', async (req) => {	6,534✔
154	if (isMultitenant) {	1,233✔
155	const { migrationVersion } = await getTenantConfig(req.tenantId)	3✔
156	req.latestMigration = migrationVersion	3✔
157	return	3✔
158	}
159
160	req.latestMigration = await lastLocalMigrationName()	1,230✔
161	})
162
163	if (dbMigrationStrategy === MultitenantMigrationStrategy.ON_REQUEST) {	6,534!
164	const migrationsMutex = createMutexByKey<void>()	6,534✔
165
166	fastify.addHook('preHandler', async (request) => {	6,534✔
167	// migrations are handled via async migrations
168	if (!isMultitenant) {	1,233✔
169	return	1,230✔
170	}
171
172	const tenant = await getTenantConfig(request.tenantId)	3✔
173	const migrationsUpToDate = await areMigrationsUpToDate(request.tenantId)	3✔
174
175	if (tenant.syncMigrationsDone \|\| migrationsUpToDate) {	3!
176	return	×
177	}
178
179	await migrationsMutex(request.tenantId, async () => {	3✔
180	const tenant = await getTenantConfig(request.tenantId)	3✔
181
182	if (tenant.syncMigrationsDone) {	3!
183	return	×
184	}
185
186	await runMigrationsOnTenant({	3✔
187	databaseUrl: tenant.databaseUrl,
188	tenantId: request.tenantId,
189	upToMigration: dbMigrationFreezeAt,
190	})
191	await updateTenantMigrationsState(request.tenantId)	3✔
192	tenant.syncMigrationsDone = true	3✔
193	})
194	})
195	}
196
197	if (dbMigrationStrategy === MultitenantMigrationStrategy.PROGRESSIVE) {	6,534!
198	fastify.addHook('preHandler', async (request) => {	×
199	if (!isMultitenant) {	×
200	return	×
201	}
202
203	const tenant = await getTenantConfig(request.tenantId)	×
204	const migrationsUpToDate = await areMigrationsUpToDate(request.tenantId)	×
205
206	// migrations are up to date
207	if (tenant.syncMigrationsDone \|\| migrationsUpToDate) {	×
208	return	×
209	}
210
211	progressiveMigrations.addTenant(request.tenantId)	×
212	})
213	}
214	},
215	{ name: 'db-migrations' }
216	)

supabase / storage / 27200912234

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