19363040917

Committed 14 Nov 2025 11:21AM UTC coverage: 75.913% (-1.8%) from 77.76%

Build # 19363040917

Build Type

Pull #796

github

Committed by

web-flow

Commit Message

Merge 0bc327f4f into b9acc7ca1

Pull Request Pull Request #796: feat: add analytics bucket sharding

Coverage Stats

2074 of 3009 branches covered (68.93%)

Branch coverage included in aggregate %.

945 of 2098 new or added lines in 36 files covered. (45.04%)

50 existing lines in 7 files now uncovered.

25386 of 33164 relevant lines covered (76.55%)

94.44 hits per line

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

50.77

/src/internal/queue/database.ts

import { Db } from 'pg-boss'
import pg from 'pg'
import { ERRORS } from '@internal/errors'
import { Knex } from 'knex'
import EventEmitter from 'node:events'

export class QueueDB extends EventEmitter implements Db {
  opened = false
  isOurs = true
  events = {
    error: 'error',
  }
  protected config: pg.PoolConfig
  protected pool?: pg.Pool

  constructor(config: pg.PoolConfig) {
    super()

    config.application_name = config.application_name || 'pgboss'

    this.config = config
  }

  async open() {
    this.pool = new pg.Pool({ ...this.config, min: 0 })
    this.pool.on('error', (error) => this.emit('error', error))

    this.opened = true
  }

  async close() {
    this.opened = false
    await this.pool?.end()
  }

  async executeSql(text: string, values: any[]) {
    if (this.opened && this.pool) {
      return await this.pool.query(text, values)
    }

    throw ERRORS.InternalError(undefined, `QueueDB not opened ${this.opened} ${text}`)
  }
}

export class KnexQueueDB extends EventEmitter implements Db {
  events = {
    error: 'error',
  }

  constructor(protected readonly knex: Knex) {
    super()
  }

  async executeSql(text: string, values: any[]): Promise<{ rows: any[] }> {
    const knexQuery = text.replaceAll('$', ':')
    const params: Record<string, any> = {}

    values.forEach((value, index) => {
      const key = (index + 1).toString()
      params[key] = value === undefined ? null : value
    })
    const result = await this.knex.raw(knexQuery, params)
    return { rows: result.rows }
  }
}

1	import { Db } from 'pg-boss'	1✔
2	import pg from 'pg'	1✔
3	import { ERRORS } from '@internal/errors'	1✔
4	import { Knex } from 'knex'	1✔
5	import EventEmitter from 'node:events'	1✔
6		1✔
7	export class QueueDB extends EventEmitter implements Db {	1✔
8	opened = false	1✔
9	isOurs = true	1✔
10	events = {	1✔
11	error: 'error',	1✔
12	}	1✔
13	protected config: pg.PoolConfig	1✔
14	protected pool?: pg.Pool	1✔
15		1✔
16	constructor(config: pg.PoolConfig) {	1✔
17	super()	×
18		×
19	config.application_name = config.application_name \|\| 'pgboss'	×
20		×
21	this.config = config	×
22	}	×
23		1✔
24	async open() {	1✔
25	this.pool = new pg.Pool({ ...this.config, min: 0 })	×
26	this.pool.on('error', (error) => this.emit('error', error))	×
27		×
28	this.opened = true	×
29	}	×
30		1✔
31	async close() {	1✔
32	this.opened = false	×
33	await this.pool?.end()	×
34	}	×
35		1✔
36	async executeSql(text: string, values: any[]) {	1✔
37	if (this.opened && this.pool) {	×
38	return await this.pool.query(text, values)	×
39	}	×
40		×
41	throw ERRORS.InternalError(undefined, `QueueDB not opened ${this.opened} ${text}`)	×
42	}	×
43	}	1✔
44		1✔
45	export class KnexQueueDB extends EventEmitter implements Db {	1✔
46	events = {	1✔
47	error: 'error',	1✔
48	}	1✔
49		1✔
50	constructor(protected readonly knex: Knex) {	1✔
NEW 51	super()	×
NEW 52	}	×
53		1✔
54	async executeSql(text: string, values: any[]): Promise<{ rows: any[] }> {	1✔
NEW 55	const knexQuery = text.replaceAll('$', ':')	×
NEW 56	const params: Record<string, any> = {}	×
NEW 57		×
NEW 58	values.forEach((value, index) => {	×
NEW 59	const key = (index + 1).toString()	×
NEW 60	params[key] = value === undefined ? null : value	×
NEW 61	})	×
NEW 62	const result = await this.knex.raw(knexQuery, params)	×
NEW 63	return { rows: result.rows }	×
NEW 64	}	×
65	}	1✔

supabase / storage / 19363040917

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