29122668957

Committed 10 Jul 2026 08:48PM UTC coverage: 79.82% (+0.4%) from 79.376%

Build # 29122668957

Build Type

Pull #1043

github

Committed by

web-flow

Commit Message

Merge 8cf1b6acf into b6b815fbd

Pull Request Pull Request #1043: feat: add queue management handlers

Coverage Stats

5535 of 7498 branches covered (73.82%)

Branch coverage included in aggregate %.

279 of 308 new or added lines in 11 files covered. (90.58%)

61 existing lines in 5 files now uncovered.

10765 of 12923 relevant lines covered (83.3%)

417.26 hits per line

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

74.24

/src/internal/queue/database.ts

import EventEmitter from 'node:events'
import { ERRORS } from '@internal/errors'
import pg from 'pg'
import { Db } from 'pg-boss'
import {
  type PgBeginTransactionOptions,
  type PgExecutor,
  PgPoolExecutor,
} from '../database/pg-connection'

export { quoteIdentifier } from '../database/sql'

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()
    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()
  }

  beginTransaction(options?: PgBeginTransactionOptions) {
    if (!this.opened || !this.pool) {
      throw ERRORS.InternalError(undefined, `QueueDB not opened ${this.opened}`)
    }

    return new PgPoolExecutor(this.pool).beginTransaction(options)
  }

  protected async useTransaction<T>(fn: (client: pg.PoolClient) => Promise<T>): Promise<T> {
    if (!this.opened || !this.pool) {
      throw ERRORS.InternalError(undefined, `QueueDB not opened ${this.opened}`)
    }

    const client = await this.pool.connect()

    // Create a promise that rejects if the client emits an error
    // (e.g. connection lost, statement_timeout at the backend level)
    let clientError: Error | undefined
    const onError = (e: Error) => {
      clientError = e
    }
    client.on('error', onError)

    try {
      await client.query('BEGIN')

      if (this.config.statement_timeout && this.config.statement_timeout > 0) {
        await client.query(`SET LOCAL statement_timeout = ${this.config.statement_timeout}`)
      }

      const result = await fn(client)

      if (clientError) {
        throw clientError
      }

      await client.query('COMMIT')
      return result
    } catch (err) {
      const rollbackErr = await client.query('ROLLBACK').catch((e) => e as Error)

      const errors = [err as Error, clientError, rollbackErr].filter(
        (e): e is Error => e instanceof Error
      )

      if (errors.length === 1) throw errors[0]
      throw new AggregateError(errors, 'Queue transaction failed')
    } finally {
      client.off('error', onError)
      client.release(clientError)
    }
  }

  async executeSql(text: string, values: unknown[]): Promise<{ rows: unknown[] }> {
    if (this.opened && this.pool) {
      return this.useTransaction((client) => client.query(text, values))
    }

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

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

  constructor(protected readonly db: PgExecutor) {
    super()
  }

  async executeSql(text: string, values: unknown[]): Promise<{ rows: unknown[] }> {
    const result = await this.db.query({
      text,
      values: values.map((value) => (value === undefined ? null : value)),
    })

    return { rows: result.rows }
  }
}

1	import EventEmitter from 'node:events'
2	import { ERRORS } from '@internal/errors'
3	import pg from 'pg'
4	import { Db } from 'pg-boss'
5	import {
6	type PgBeginTransactionOptions,
7	type PgExecutor,
8	PgPoolExecutor,
9	} from '../database/pg-connection'
10
11	export { quoteIdentifier } from '../database/sql'
12
13	export class QueueDB extends EventEmitter implements Db {
14	opened = false	2✔
15	isOurs = true	2✔
16	events = {	2✔
17	error: 'error',
18	}
19	protected config: pg.PoolConfig
20	protected pool?: pg.Pool
21
22	constructor(config: pg.PoolConfig) {
23	super()	2✔
24	this.config = config	2✔
25	}
26
27	async open() {
28	this.pool = new pg.Pool({ ...this.config, min: 0 })	2✔
29	this.pool.on('error', (error) => this.emit('error', error))	2✔
30
31	this.opened = true	2✔
32	}
33
34	async close() {
35	this.opened = false	2✔
36	await this.pool?.end()	2✔
37	}
38
39	beginTransaction(options?: PgBeginTransactionOptions) {
40	if (!this.opened \|\| !this.pool) {	20!
NEW 41	throw ERRORS.InternalError(undefined, `QueueDB not opened ${this.opened}`)	×
42	}
43
44	return new PgPoolExecutor(this.pool).beginTransaction(options)	20✔
45	}
46
47	protected async useTransaction<T>(fn: (client: pg.PoolClient) => Promise<T>): Promise<T> {
48	if (!this.opened \|\| !this.pool) {	13!
49	throw ERRORS.InternalError(undefined, `QueueDB not opened ${this.opened}`)	×
50	}
51
52	const client = await this.pool.connect()	13✔
53
54	// Create a promise that rejects if the client emits an error
55	// (e.g. connection lost, statement_timeout at the backend level)
56	let clientError: Error \| undefined
57	const onError = (e: Error) => {	13✔
UNCOV 58	clientError = e	×
59	}
60	client.on('error', onError)	13✔
61
62	try {	13✔
63	await client.query('BEGIN')	13✔
64
65	if (this.config.statement_timeout && this.config.statement_timeout > 0) {	13!
66	await client.query(`SET LOCAL statement_timeout = ${this.config.statement_timeout}`)	13✔
67	}
68
69	const result = await fn(client)	13✔
70
71	if (clientError) {	13!
72	throw clientError	×
73	}
74
75	await client.query('COMMIT')	13✔
76	return result	13✔
77	} catch (err) {
78	const rollbackErr = await client.query('ROLLBACK').catch((e) => e as Error)	×
79
UNCOV 80	const errors = [err as Error, clientError, rollbackErr].filter(	×
81	(e): e is Error => e instanceof Error	×
82	)
83
84	if (errors.length === 1) throw errors[0]	×
85	throw new AggregateError(errors, 'Queue transaction failed')	×
86	} finally {
87	client.off('error', onError)	13✔
88	client.release(clientError)	13✔
89	}
90	}
91
92	async executeSql(text: string, values: unknown[]): Promise<{ rows: unknown[] }> {
93	if (this.opened && this.pool) {	13!
94	return this.useTransaction((client) => client.query(text, values))	13✔
95	}
96
UNCOV 97	throw ERRORS.InternalError(undefined, `QueueDB not opened ${this.opened} ${text}`)	×
98	}
99	}
100
101	export class PgQueueDB extends EventEmitter implements Db {
102	events = {	2✔
103	error: 'error',
104	}
105
106	constructor(protected readonly db: PgExecutor) {	2✔
107	super()	2✔
108	}
109
110	async executeSql(text: string, values: unknown[]): Promise<{ rows: unknown[] }> {
111	const result = await this.db.query({	1✔
112	text,
113	values: values.map((value) => (value === undefined ? null : value)),	3✔
114	})
115
116	return { rows: result.rows }	1✔
117	}
118	}

supabase / storage / 29122668957

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