14758743434

Committed 30 Apr 2025 03:48PM UTC coverage: 77.886% (-0.4%) from 78.302%

Build # 14758743434

Build Type

push

github

Committed by

web-flow

Commit Message

fix: switch from jsonwebtoken to jose for jwt signing/verification (#678) (#679)

- Supports async jwt signing/verification using libuv
- Add support for OKP (ed25519/Ed448) jwks
- Cleaner implementation across jwt code

Co-authored-by: Lenny <itslenny@users.noreply.github.com>

Coverage Stats

1465 of 2052 branches covered (71.39%)

Branch coverage included in aggregate %.

304 of 500 new or added lines in 16 files covered. (60.8%)

1 existing line in 1 file now uncovered.

16885 of 21508 relevant lines covered (78.51%)

153.57 hits per line

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

67.42

/src/internal/database/pool.ts

import { getConfig } from '../../config'
import TTLCache from '@isaacs/ttlcache'
import { knex, Knex } from 'knex'
import { logger, logSchema } from '@internal/monitoring'
import { getSslSettings } from '@internal/database/util'
import { wait } from '@internal/concurrency'
import { JWTPayload } from 'jose'

const {
  isMultitenant,
  databaseSSLRootCert,
  databaseMaxConnections,
  databaseFreePoolAfterInactivity,
  databaseConnectionTimeout,
  dbSearchPath,
  dbPostgresVersion,
} = getConfig()

export interface TenantConnectionOptions {
  user: User
  superUser: User

  tenantId: string
  dbUrl: string
  isExternalPool?: boolean
  isSingleUse?: boolean
  idleTimeoutMillis?: number
  reapIntervalMillis?: number
  maxConnections: number
  clusterSize?: number
  headers?: Record<string, string | undefined | string[]>
  method?: string
  path?: string
  operation?: () => string | undefined
}

export interface User {
  jwt: string
  payload: { role?: string } & JWTPayload
}

export interface PoolStrategy {
  acquire(): Knex
  rebalance(options: { clusterSize: number }): void
  destroy(): Promise<void>
}

export const searchPath = ['storage', 'public', 'extensions', ...dbSearchPath.split(',')].filter(
  Boolean
)

const multiTenantLRUConfig = {
  ttl: 1000 * 10,
  updateAgeOnGet: true,
  checkAgeOnGet: true,
}

const tenantPools = new TTLCache<string, PoolStrategy>({
  ...(isMultitenant ? multiTenantLRUConfig : { max: 1, ttl: Infinity }),
  dispose: async (pool) => {
    if (!pool) return
    try {
      await pool.destroy()
    } catch (e) {
      logSchema.error(logger, 'pool was not able to be destroyed', {
        type: 'db',
        error: e,
      })
    }
  },
})

/**
 * PoolManager is a class that manages a pool of Knex connections.
 * It creates a new pool for each tenant and reuses existing pools.
 */
export class PoolManager {
  rebalanceAll(data: { clusterSize: number }) {
    for (const pool of tenantPools.values()) {
      pool.rebalance({
        clusterSize: data.clusterSize,
      })
    }
  }

  rebalance(tenantId: string, data: { clusterSize: number }) {
    const pool = tenantPools.get(tenantId)
    if (pool) {
      pool.rebalance({
        clusterSize: data.clusterSize,
      })
    }
  }

  getPool(settings: TenantConnectionOptions) {
    const existingPool = tenantPools.get(settings.tenantId)
    if (existingPool) {
      return existingPool
    }

    const newPool = this.newPool(settings)

    if ((settings.isSingleUse && !settings.isExternalPool) || !settings.isSingleUse) {
      tenantPools.set(settings.tenantId, newPool)
    }
    return newPool
  }

  destroy(tenantId: string) {
    const pool = tenantPools.get(tenantId)
    if (pool) {
      tenantPools.delete(tenantId)
      return pool.destroy()
    }
    return Promise.resolve()
  }

  destroyAll() {
    const promises: Promise<void>[] = []

    for (const [connectionString, pool] of tenantPools) {
      promises.push(pool.destroy())
      tenantPools.delete(connectionString)
    }

    return Promise.allSettled(promises)
  }

  protected newPool(settings: TenantConnectionOptions) {
    return new TenantPool(settings)
  }
}

/**
 * TenantPool create a new Knex pool for each tenant, with rebalance
 * functionality to adjust the number of connections based on the cluster size.
 */
class TenantPool implements PoolStrategy {
  protected pool?: Knex

  constructor(protected readonly options: TenantConnectionOptions) {}

  acquire() {
    if (this.pool) {
      return this.pool
    }

    this.pool = this.createKnexPool()
    return this.pool
  }

  destroy(): Promise<void> {
    if (!this.pool) {
      return Promise.resolve()
    }
    return this.drainPool(this.pool)
  }

  getSettings() {
    const isSingleUseExternalPool = this.options.isSingleUse && this.options.isExternalPool

    const clusterSize = this.options.clusterSize || 0
    let maxConnection = this.options.maxConnections || databaseMaxConnections

    if (clusterSize > 0) {
      maxConnection = Math.ceil(maxConnection / clusterSize)
    }

    if (isSingleUseExternalPool) {
      maxConnection = 1
    }

    return {
      ...this.options,
      searchPath: this.options.isExternalPool ? undefined : searchPath,
      idleTimeoutMillis: isSingleUseExternalPool ? 100 : databaseFreePoolAfterInactivity,
      reapIntervalMillis: isSingleUseExternalPool ? 50 : undefined,
      maxConnections: maxConnection,
    }
  }

  rebalance(options: { clusterSize: number }) {
    if (options.clusterSize === 0) {
      return
    }

    const originalPool = this.pool

    this.options.clusterSize = options.clusterSize
    this.pool = undefined

    if (originalPool) {
      this.drainPool(originalPool).catch((e) => {
        logger.error({ type: 'pool', error: e })
      })
    }
  }

  protected async drainPool(pool: Knex) {
    while (true) {
      let waiting = 0
      waiting += pool.client.pool.numPendingAcquires()
      waiting += pool.client.pool.numPendingValidations()
      waiting += pool.client.pool.numPendingCreates()

      if (waiting === 0) {
        break
      }

      await wait(200)
    }

    return pool.destroy()
  }

  protected createKnexPool() {
    const settings = this.getSettings()
    const sslSettings = getSslSettings({
      connectionString: settings.dbUrl,
      databaseSSLRootCert,
    })

    const maxConnections = settings.maxConnections

    return knex({
      client: 'pg',
      version: dbPostgresVersion,
      searchPath: settings.searchPath,
      pool: {
        min: 0,
        max: maxConnections,
        acquireTimeoutMillis: databaseConnectionTimeout,
        idleTimeoutMillis: settings.idleTimeoutMillis,
        reapIntervalMillis: 1000,
      },
      connection: {
        connectionString: settings.dbUrl,
        connectionTimeoutMillis: databaseConnectionTimeout,
        ssl: sslSettings ? { ...sslSettings } : undefined,
      },
      acquireConnectionTimeout: databaseConnectionTimeout,
    })
  }
}

1	import { getConfig } from '../../config'	1✔
2	import TTLCache from '@isaacs/ttlcache'	1✔
3	import { knex, Knex } from 'knex'	1✔
4	import { logger, logSchema } from '@internal/monitoring'	1✔
5	import { getSslSettings } from '@internal/database/util'	1✔
6	import { wait } from '@internal/concurrency'	1✔
7	import { JWTPayload } from 'jose'	1✔
8		1✔
9	const {	1✔
10	isMultitenant,	1✔
11	databaseSSLRootCert,	1✔
12	databaseMaxConnections,	1✔
13	databaseFreePoolAfterInactivity,	1✔
14	databaseConnectionTimeout,	1✔
15	dbSearchPath,	1✔
16	dbPostgresVersion,	1✔
17	} = getConfig()	1✔
18		1✔
19	export interface TenantConnectionOptions {	1✔
20	user: User	1✔
21	superUser: User	1✔
22		1✔
23	tenantId: string	1✔
24	dbUrl: string	1✔
25	isExternalPool?: boolean	1✔
26	isSingleUse?: boolean	1✔
27	idleTimeoutMillis?: number	1✔
28	reapIntervalMillis?: number	1✔
29	maxConnections: number	1✔
30	clusterSize?: number	1✔
31	headers?: Record<string, string \| undefined \| string[]>	1✔
32	method?: string	1✔
33	path?: string	1✔
34	operation?: () => string \| undefined	1✔
35	}	1✔
36		1✔
37	export interface User {	1✔
38	jwt: string	1✔
39	payload: { role?: string } & JWTPayload	1✔
40	}	1✔
41		1✔
42	export interface PoolStrategy {	1✔
43	acquire(): Knex	1✔
44	rebalance(options: { clusterSize: number }): void	1✔
45	destroy(): Promise<void>	1✔
46	}	1✔
47		1✔
48	export const searchPath = ['storage', 'public', 'extensions', ...dbSearchPath.split(',')].filter(	1✔
49	Boolean	1✔
50	)	1✔
51		1✔
52	const multiTenantLRUConfig = {	1✔
53	ttl: 1000 * 10,	1✔
54	updateAgeOnGet: true,	1✔
55	checkAgeOnGet: true,	1✔
56	}	1✔
57		1✔
58	const tenantPools = new TTLCache<string, PoolStrategy>({	1✔
59	...(isMultitenant ? multiTenantLRUConfig : { max: 1, ttl: Infinity }),	1!
60	dispose: async (pool) => {	1✔
NEW 61	if (!pool) return	×
NEW 62	try {	×
NEW 63	await pool.destroy()	×
NEW 64	} catch (e) {	×
NEW 65	logSchema.error(logger, 'pool was not able to be destroyed', {	×
NEW 66	type: 'db',	×
NEW 67	error: e,	×
NEW 68	})	×
NEW 69	}	×
70	},	1✔
71	})	1✔
72		1✔
73	/**	1✔
74	* PoolManager is a class that manages a pool of Knex connections.	1✔
75	* It creates a new pool for each tenant and reuses existing pools.	1✔
76	*/	1✔
77	export class PoolManager {	1✔
78	rebalanceAll(data: { clusterSize: number }) {	1✔
NEW 79	for (const pool of tenantPools.values()) {	×
NEW 80	pool.rebalance({	×
NEW 81	clusterSize: data.clusterSize,	×
NEW 82	})	×
NEW 83	}	×
NEW 84	}	×
85		1✔
86	rebalance(tenantId: string, data: { clusterSize: number }) {	1✔
NEW 87	const pool = tenantPools.get(tenantId)	×
NEW 88	if (pool) {	×
NEW 89	pool.rebalance({	×
NEW 90	clusterSize: data.clusterSize,	×
NEW 91	})	×
NEW 92	}	×
NEW 93	}	×
94		1✔
95	getPool(settings: TenantConnectionOptions) {	1✔
96	const existingPool = tenantPools.get(settings.tenantId)	135✔
97	if (existingPool) {	135✔
98	return existingPool	134✔
99	}	134✔
100		1✔
101	const newPool = this.newPool(settings)	1✔
102		1✔
103	if ((settings.isSingleUse && !settings.isExternalPool) \|\| !settings.isSingleUse) {	135!
104	tenantPools.set(settings.tenantId, newPool)	1✔
105	}	1✔
106	return newPool	1✔
107	}	1✔
108		1✔
109	destroy(tenantId: string) {	1✔
NEW 110	const pool = tenantPools.get(tenantId)	×
NEW 111	if (pool) {	×
NEW 112	tenantPools.delete(tenantId)	×
NEW 113	return pool.destroy()	×
NEW 114	}	×
NEW 115	return Promise.resolve()	×
NEW 116	}	×
117		1✔
118	destroyAll() {	1✔
NEW 119	const promises: Promise<void>[] = []	×
NEW 120		×
NEW 121	for (const [connectionString, pool] of tenantPools) {	×
NEW 122	promises.push(pool.destroy())	×
NEW 123	tenantPools.delete(connectionString)	×
NEW 124	}	×
NEW 125		×
NEW 126	return Promise.allSettled(promises)	×
NEW 127	}	×
128		1✔
129	protected newPool(settings: TenantConnectionOptions) {	1✔
130	return new TenantPool(settings)	1✔
131	}	1✔
132	}	1✔
133		1✔
134	/**	1✔
135	* TenantPool create a new Knex pool for each tenant, with rebalance	1✔
136	* functionality to adjust the number of connections based on the cluster size.	1✔
137	*/	1✔
138	class TenantPool implements PoolStrategy {	1✔
139	protected pool?: Knex	1✔
140		1✔
141	constructor(protected readonly options: TenantConnectionOptions) {}	1✔
142		1✔
143	acquire() {	1✔
144	if (this.pool) {	341✔
145	return this.pool	340✔
146	}	340✔
147		1✔
148	this.pool = this.createKnexPool()	1✔
149	return this.pool	1✔
150	}	1✔
151		1✔
152	destroy(): Promise<void> {	1✔
NEW 153	if (!this.pool) {	×
NEW 154	return Promise.resolve()	×
NEW 155	}	×
NEW 156	return this.drainPool(this.pool)	×
NEW 157	}	×
158		1✔
159	getSettings() {	1✔
160	const isSingleUseExternalPool = this.options.isSingleUse && this.options.isExternalPool	1✔
161		1✔
162	const clusterSize = this.options.clusterSize \|\| 0	1✔
163	let maxConnection = this.options.maxConnections \|\| databaseMaxConnections	1!
164		1✔
165	if (clusterSize > 0) {	1!
NEW 166	maxConnection = Math.ceil(maxConnection / clusterSize)	×
NEW 167	}	×
168		1✔
169	if (isSingleUseExternalPool) {	1!
NEW 170	maxConnection = 1	×
NEW 171	}	×
172		1✔
173	return {	1✔
174	...this.options,	1✔
175	searchPath: this.options.isExternalPool ? undefined : searchPath,	1!
176	idleTimeoutMillis: isSingleUseExternalPool ? 100 : databaseFreePoolAfterInactivity,	1!
177	reapIntervalMillis: isSingleUseExternalPool ? 50 : undefined,	1!
178	maxConnections: maxConnection,	1✔
179	}	1✔
180	}	1✔
181		1✔
182	rebalance(options: { clusterSize: number }) {	1✔
NEW 183	if (options.clusterSize === 0) {	×
NEW 184	return	×
NEW 185	}	×
NEW 186		×
NEW 187	const originalPool = this.pool	×
NEW 188		×
NEW 189	this.options.clusterSize = options.clusterSize	×
NEW 190	this.pool = undefined	×
NEW 191		×
NEW 192	if (originalPool) {	×
NEW 193	this.drainPool(originalPool).catch((e) => {	×
NEW 194	logger.error({ type: 'pool', error: e })	×
NEW 195	})	×
NEW 196	}	×
NEW 197	}	×
198		1✔
199	protected async drainPool(pool: Knex) {	1✔
NEW 200	while (true) {	×
NEW 201	let waiting = 0	×
NEW 202	waiting += pool.client.pool.numPendingAcquires()	×
NEW 203	waiting += pool.client.pool.numPendingValidations()	×
NEW 204	waiting += pool.client.pool.numPendingCreates()	×
NEW 205		×
NEW 206	if (waiting === 0) {	×
NEW 207	break	×
NEW 208	}	×
NEW 209		×
NEW 210	await wait(200)	×
NEW 211	}	×
NEW 212		×
NEW 213	return pool.destroy()	×
NEW 214	}	×
215		1✔
216	protected createKnexPool() {	1✔
217	const settings = this.getSettings()	1✔
218	const sslSettings = getSslSettings({	1✔
219	connectionString: settings.dbUrl,	1✔
220	databaseSSLRootCert,	1✔
221	})	1✔
222		1✔
223	const maxConnections = settings.maxConnections	1✔
224		1✔
225	return knex({	1✔
226	client: 'pg',	1✔
227	version: dbPostgresVersion,	1✔
228	searchPath: settings.searchPath,	1✔
229	pool: {	1✔
230	min: 0,	1✔
231	max: maxConnections,	1✔
232	acquireTimeoutMillis: databaseConnectionTimeout,	1✔
233	idleTimeoutMillis: settings.idleTimeoutMillis,	1✔
234	reapIntervalMillis: 1000,	1✔
235	},	1✔
236	connection: {	1✔
237	connectionString: settings.dbUrl,	1✔
238	connectionTimeoutMillis: databaseConnectionTimeout,	1✔
239	ssl: sslSettings ? { ...sslSettings } : undefined,	1!
240	},	1✔
241	acquireConnectionTimeout: databaseConnectionTimeout,	1✔
242	})	1✔
243	}	1✔
244	}	1✔

supabase / storage / 14758743434

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