27760253447

Committed 18 Jun 2026 12:43PM UTC coverage: 60.005% (-18.2%) from 78.232%

Build # 27760253447

Build Type

push

github

Committed by

web-flow

Commit Message

fix: limit for delete objects and sign urls (#1160)

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

Coverage Stats

3351 of 6375 branches covered (52.56%)

Branch coverage included in aggregate %.

25 of 31 new or added lines in 4 files covered. (80.65%)

2041 existing lines in 86 files now uncovered.

7151 of 11127 relevant lines covered (64.27%)

354.89 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 pg from 'pg'
import { Db } from 'pg-boss'
import { PgExecutor } 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()
  }

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

supabase / storage / 27760253447

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