25397491493

Committed 05 May 2026 07:25PM UTC coverage: 74.101% (-0.1%) from 74.25%

Build # 25397491493

Build Type

Pull #922

github

Committed by

web-flow

Commit Message

Merge fd2216ed5 into a6ded4fd0

Pull Request Pull Request #922: fix: no any rule

Coverage Stats

3970 of 5906 branches covered (67.22%)

Branch coverage included in aggregate %.

102 of 150 new or added lines in 23 files covered. (68.0%)

9 existing lines in 5 files now uncovered.

8004 of 10253 relevant lines covered (78.06%)

406.77 hits per line

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

0.0

/src/internal/queue/database.ts

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

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 || getConfig().pgQueueApplicationName

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

  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 KnexQueueDB extends EventEmitter implements Db {
  events = {
    error: 'error',
  }

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

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

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

supabase / storage / 25397491493

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