29583752771

Committed 17 Jul 2026 01:23PM UTC coverage: 79.932% (+0.4%) from 79.574%

Build # 29583752771

Build Type

Pull #1043

github

Committed by

web-flow

Commit Message

Merge 77dd9d483 into e0684d992

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

Coverage Stats

5618 of 7594 branches covered (73.98%)

Branch coverage included in aggregate %.

283 of 312 new or added lines in 11 files covered. (90.71%)

17 existing lines in 1 file now uncovered.

10860 of 13021 relevant lines covered (83.4%)

418.62 hits per line

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

90.29

/src/internal/queue/overflow.ts

import { normalizeStringList } from '@internal/strings'
import type { QueryResultRow } from 'pg'
import type { PgExecutor, PgTransaction, PgTransactionalExecutor } from '../database/pg-connection'
import { quoteQualifiedIdentifier } from '../database/sql'
import { PG_BOSS_SCHEMA } from './constants'

const CREATED_STATE = 'created'
const QUEUE_OVERFLOW_ADVISORY_LOCK_KEY = '-5525285245963000612'
export const JOB_OVERFLOW_LIST_LIMIT_DEFAULT = 50
export const JOB_OVERFLOW_RESTORE_LIMIT_DEFAULT = 10000
export const QUEUE_OVERFLOW_UNSCOPED_BACKUP_MESSAGE =
  'Backup requires at least one queue, event type, or tenant filter unless confirmAll is true'

export type QueueOverflowSource = 'job' | 'backup'
export type QueueOverflowGroupBy = 'summary' | 'tenant'

export interface QueueOverflowFilters {
  eventTypes?: readonly string[]
  name?: string
  tenantRefs?: readonly string[]
}

interface QueueOverflowOperationOptions {
  signal?: AbortSignal
}

export interface ListQueueOverflowOptions
  extends QueueOverflowFilters,
    QueueOverflowOperationOptions {
  groupBy?: QueueOverflowGroupBy
  limit?: number
  source?: QueueOverflowSource
}

export interface MoveQueueOverflowOptions
  extends QueueOverflowFilters,
    QueueOverflowOperationOptions {
  limit?: number
}

export interface BackupQueueOverflowOptions extends MoveQueueOverflowOptions {
  confirmAll?: boolean
}

export interface QueueOverflowSummaryRow {
  count: number
  eventType: string | null
  name: string
}

export interface QueueOverflowTenantRow {
  count: number
  tenantRef: string | null
}

interface QueueOverflowWhereClause {
  sql: string
  values: unknown[]
}

interface QueueOverflowAggregateRow extends QueryResultRow {
  count: number | string
  group_count: number | string
  total_count: number | string
}

interface QueueOverflowTotalCountRow extends QueryResultRow {
  total_count: number | string
}

interface QueueOverflowSummaryDbRow extends QueueOverflowAggregateRow {
  event_type: string | null
  name: string
}

interface QueueOverflowTenantDbRow extends QueueOverflowAggregateRow {
  tenant_ref: string | null
}

interface QueueOverflowCountRow extends QueryResultRow {
  moved_count: number | string
}

interface QueueOverflowRestoreCountRow extends QueueOverflowCountRow {
  selected_count: number | string
}

interface QueueOverflowTableRow extends QueryResultRow {
  table_name: string | null
}

function normalizeOverflowString(value: string | undefined) {
  const normalized = value?.trim()
  return normalized ? normalized : undefined
}

export function normalizeQueueOverflowFilters(filters: QueueOverflowFilters): QueueOverflowFilters {
  return {
    name: normalizeOverflowString(filters.name),
    eventTypes: normalizeStringList(filters.eventTypes),
    tenantRefs: normalizeStringList(filters.tenantRefs),
  }
}

export function hasQueueOverflowFilters(filters: QueueOverflowFilters): boolean {
  const normalized = normalizeQueueOverflowFilters(filters)
  return Boolean(normalized.name || normalized.eventTypes?.length || normalized.tenantRefs?.length)
}

export function isQueueOverflowBackupScoped(options: BackupQueueOverflowOptions): boolean {
  return options.confirmAll === true || hasQueueOverflowFilters(options)
}

export function buildQueueOverflowWhereClause(
  filters: QueueOverflowFilters
): QueueOverflowWhereClause {
  const normalizedFilters = normalizeQueueOverflowFilters(filters)
  const clauses = ['state = $1']
  const values: unknown[] = [CREATED_STATE]

  if (normalizedFilters.name) {
    values.push(normalizedFilters.name)
    clauses.push(`name = $${values.length}`)
  }

  if (normalizedFilters.eventTypes?.length) {
    values.push(normalizedFilters.eventTypes)
    clauses.push(`data->'event'->>'type' = ANY($${values.length}::text[])`)
  }

  if (normalizedFilters.tenantRefs?.length) {
    values.push(normalizedFilters.tenantRefs)
    clauses.push(`data->'tenant'->>'ref' = ANY($${values.length}::text[])`)
  }

  return {
    sql: clauses.join(' AND '),
    values,
  }
}

export class QueueOverflowStorePg {
  private readonly backupTableName: string
  private readonly jobTable: string
  private readonly backupTable: string

  constructor(
    private readonly db: Pick<PgTransactionalExecutor, 'beginTransaction'>,
    schema = PG_BOSS_SCHEMA
  ) {
    const jobTableName = `${schema}.job`
    this.backupTableName = `${schema}.job_overflow_backup`
    this.jobTable = quoteQualifiedIdentifier(jobTableName)
    this.backupTable = quoteQualifiedIdentifier(this.backupTableName)
  }

  async countCreated(options: QueueOverflowOperationOptions = {}) {
    return this.withMaintenanceTransaction(options.signal, async (transaction) => {
      const result = await transaction.query<QueueOverflowTotalCountRow>(
        {
          text: `
            SELECT COUNT(*)::bigint AS total_count
            FROM ${this.jobTable}
            WHERE state = $1
          `,
          values: [CREATED_STATE],
        },
        { signal: options.signal }
      )

      return {
        totalCount: Number(result.rows[0]?.total_count ?? 0),
      }
    })
  }

  async list(options: ListQueueOverflowOptions) {
    const source = options.source ?? 'job'
    const groupBy = options.groupBy ?? 'summary'
    const limit = options.limit ?? JOB_OVERFLOW_LIST_LIMIT_DEFAULT
    const filters = normalizeQueueOverflowFilters(options)

    assertPositiveSafeInteger(limit, 'limit')

    return this.withMaintenanceTransaction(options.signal, async (transaction) => {
      const sourceTableExists =
        source === 'backup' ? await this.backupTableExists(transaction, options.signal) : true

      if (!sourceTableExists) {
        return {
          sourceTableExists,
          data: [] as QueueOverflowSummaryRow[] | QueueOverflowTenantRow[],
          filters,
          groupBy,
          groupCount: 0,
          hasMore: false,
          source,
          totalCount: 0,
        }
      }

      const tableName = source === 'backup' ? this.backupTable : this.jobTable
      const whereClause = buildQueueOverflowWhereClause(filters)
      const values = [...whereClause.values, limit]
      const limitPlaceholder = `$${values.length}`

      if (groupBy === 'tenant') {
        const result = await transaction.query<QueueOverflowTenantDbRow>(
          {
            text: `
              SELECT
                data->'tenant'->>'ref' AS tenant_ref,
                COUNT(*)::bigint AS count,
                COUNT(*) OVER ()::bigint AS group_count,
                SUM(COUNT(*)) OVER ()::bigint AS total_count
              FROM ${tableName}
              WHERE ${whereClause.sql}
              GROUP BY data->'tenant'->>'ref'
              ORDER BY count DESC, tenant_ref ASC
              LIMIT ${limitPlaceholder}
            `,
            values,
          },
          { signal: options.signal }
        )

        const groupCount = Number(result.rows[0]?.group_count ?? 0)

        return {
          sourceTableExists,
          data: result.rows.map((row) => ({
            count: Number(row.count),
            tenantRef: row.tenant_ref,
          })),
          filters,
          groupBy,
          groupCount,
          hasMore: groupCount > result.rows.length,
          source,
          totalCount: Number(result.rows[0]?.total_count ?? 0),
        }
      }

      const result = await transaction.query<QueueOverflowSummaryDbRow>(
        {
          text: `
            SELECT
              name,
              data->'event'->>'type' AS event_type,
              COUNT(*)::bigint AS count,
              COUNT(*) OVER ()::bigint AS group_count,
              SUM(COUNT(*)) OVER ()::bigint AS total_count
            FROM ${tableName}
            WHERE ${whereClause.sql}
            GROUP BY name, data->'event'->>'type'
            ORDER BY count DESC, name ASC, event_type ASC
            LIMIT ${limitPlaceholder}
          `,
          values,
        },
        { signal: options.signal }
      )

      const groupCount = Number(result.rows[0]?.group_count ?? 0)

      return {
        sourceTableExists,
        data: result.rows.map((row) => ({
          count: Number(row.count),
          eventType: row.event_type,
          name: row.name,
        })),
        filters,
        groupBy,
        groupCount,
        hasMore: groupCount > result.rows.length,
        source,
        totalCount: Number(result.rows[0]?.total_count ?? 0),
      }
    })
  }

  async backup(options: BackupQueueOverflowOptions) {
    if (!isQueueOverflowBackupScoped(options)) {
      throw new Error(QUEUE_OVERFLOW_UNSCOPED_BACKUP_MESSAGE)
    }

    assertOptionalPositiveSafeInteger(options.limit, 'limit')
    const filters = normalizeQueueOverflowFilters(options)

    return this.withMaintenanceTransaction(options.signal, async (transaction) => {
      await this.acquireMaintenanceLock(transaction, options.signal)
      const backupTableCreated = await this.ensureBackupTable(transaction, options.signal)

      await transaction.query(`LOCK TABLE ${this.jobTable} IN SHARE ROW EXCLUSIVE MODE`, {
        signal: options.signal,
      })

      const movedCount = await this.moveJobs(
        transaction,
        filters,
        this.jobTable,
        this.backupTable,
        options.limit,
        options.signal
      )

      return {
        backupTableCreated,
        filters,
        limit: options.limit ?? null,
        movedCount,
      }
    })
  }

  async restore(options: MoveQueueOverflowOptions) {
    const limit = options.limit ?? JOB_OVERFLOW_RESTORE_LIMIT_DEFAULT
    assertPositiveSafeInteger(limit, 'limit')
    const filters = normalizeQueueOverflowFilters(options)

    return this.withMaintenanceTransaction(options.signal, async (transaction) => {
      await this.acquireMaintenanceLock(transaction, options.signal)
      const backupTableExists = await this.backupTableExists(transaction, options.signal)

      if (!backupTableExists) {
        return {
          backupTableExists,
          conflictCount: 0,
          filters,
          hasMore: false,
          limit,
          movedCount: 0,
        }
      }

      const { conflictCount, hasMore, movedCount } = await this.restoreJobs(
        transaction,
        filters,
        limit,
        options.signal
      )

      return {
        backupTableExists,
        conflictCount,
        filters,
        hasMore,
        limit,
        movedCount,
      }
    })
  }

  private async backupTableExists(db: PgExecutor, signal?: AbortSignal) {
    const result = await db.query<QueueOverflowTableRow>(
      {
        text: 'SELECT to_regclass($1) AS table_name',
        values: [this.backupTableName],
      },
      { signal }
    )

    return Boolean(result.rows[0]?.table_name)
  }

  private async ensureBackupTable(db: PgExecutor, signal?: AbortSignal) {
    const existed = await this.backupTableExists(db, signal)

    await db.query(
      `CREATE TABLE IF NOT EXISTS ${this.backupTable} (LIKE ${this.jobTable} INCLUDING ALL)`,
      {
        signal,
      }
    )

    return !existed
  }

  private async moveJobs(
    db: PgExecutor,
    filters: QueueOverflowFilters,
    sourceTable: string,
    targetTable: string,
    limit?: number,
    signal?: AbortSignal
  ) {
    const whereClause = buildQueueOverflowWhereClause(filters)
    const values = [...whereClause.values]
    let selectionBoundary = ''

    if (limit !== undefined) {
      values.push(limit)
      selectionBoundary = `ORDER BY name, id LIMIT $${values.length}`
    }

    const result = await db.query<QueueOverflowCountRow>(
      {
        text: `
          WITH selected AS (
            SELECT name, id
            FROM ${sourceTable}
            WHERE ${whereClause.sql}
            ${selectionBoundary}
          ),
          moved AS (
            DELETE FROM ${sourceTable} AS source_job
            USING selected
            WHERE source_job.name = selected.name
              AND source_job.id = selected.id
            RETURNING source_job.*
          ),
          inserted AS (
            INSERT INTO ${targetTable}
            SELECT * FROM moved
            RETURNING 1
          )
          SELECT COUNT(*)::bigint AS moved_count FROM inserted
        `,
        values,
      },
      { signal }
    )

    return Number(result.rows[0]?.moved_count ?? 0)
  }

  private async restoreJobs(
    db: PgExecutor,
    filters: QueueOverflowFilters,
    limit: number,
    signal?: AbortSignal
  ) {
    const whereClause = buildQueueOverflowWhereClause(filters)
    const values = [...whereClause.values]
    values.push(limit)
    const limitPlaceholder = `$${values.length}`
    const result = await db.query<QueueOverflowRestoreCountRow>(
      {
        text: `
          WITH selected AS (
            SELECT *
            FROM ${this.backupTable}
            WHERE ${whereClause.sql}
            ORDER BY name, id
            LIMIT ${limitPlaceholder}
          ),
          inserted AS (
            INSERT INTO ${this.jobTable}
            SELECT * FROM selected
            ON CONFLICT DO NOTHING
            RETURNING 1
          ),
          deleted AS (
            DELETE FROM ${this.backupTable} AS source_job
            USING selected
            WHERE source_job.name = selected.name
              AND source_job.id = selected.id
            RETURNING 1
          )
          SELECT
            (SELECT COUNT(*) FROM selected) AS selected_count,
            (SELECT COUNT(*) FROM inserted) AS moved_count
        `,
        values,
      },
      { signal }
    )

    const selectedCount = Number(result.rows[0]?.selected_count ?? 0)
    const movedCount = Number(result.rows[0]?.moved_count ?? 0)

    return {
      conflictCount: selectedCount - movedCount,
      hasMore: selectedCount === limit,
      movedCount,
    }
  }

  private async acquireMaintenanceLock(db: PgExecutor, signal?: AbortSignal): Promise<void> {
    await db.query(
      {
        text: 'SELECT pg_advisory_xact_lock($1::bigint)',
        values: [QUEUE_OVERFLOW_ADVISORY_LOCK_KEY],
      },
      { signal }
    )
  }

  private async withMaintenanceTransaction<T>(
    signal: AbortSignal | undefined,
    fn: (transaction: PgTransaction) => Promise<T>
  ): Promise<T> {
    const transaction = await this.db.beginTransaction({ signal })

    try {
      await transaction.query(
        {
          text: `
            SELECT
              set_config('statement_timeout', $1, true),
              set_config('lock_timeout', $2, true)
          `,
          values: ['0', '30s'],
        },
        { signal }
      )

      const result = await fn(transaction)
      await transaction.commit()
      return result
    } catch (error) {
      try {
        await transaction.rollback()
      } catch (rollbackError) {
        await logQueueOverflowRollbackFailure(error, rollbackError)
      }

      throw error
    }
  }
}

async function logQueueOverflowRollbackFailure(
  originalError: unknown,
  rollbackError: unknown
): Promise<void> {
  try {
    const { logger, logSchema } = await import('@internal/monitoring')
    logSchema.warning(logger, '[QueueOverflow] Failed to rollback maintenance transaction', {
      type: 'pgboss',
      error: rollbackError,
      metadata: JSON.stringify({ originalError: String(originalError) }),
    })
  } catch (loggingError) {
    console.error('[QueueOverflow] Failed to log maintenance transaction rollback failure', {
      originalError,
      rollbackError,
      loggingError,
    })
  }
}

function assertOptionalPositiveSafeInteger(value: number | undefined, field: string): void {
  if (value !== undefined) {
    assertPositiveSafeInteger(value, field)
  }
}

function assertPositiveSafeInteger(value: number, field: string): void {
  if (!Number.isSafeInteger(value) || value <= 0) {
    throw new Error(`${field} must be a positive safe integer`)
  }
}

1	import { normalizeStringList } from '@internal/strings'
2	import type { QueryResultRow } from 'pg'
3	import type { PgExecutor, PgTransaction, PgTransactionalExecutor } from '../database/pg-connection'
4	import { quoteQualifiedIdentifier } from '../database/sql'
5	import { PG_BOSS_SCHEMA } from './constants'
6
7	const CREATED_STATE = 'created'	74✔
8	const QUEUE_OVERFLOW_ADVISORY_LOCK_KEY = '-5525285245963000612'	74✔
9	export const JOB_OVERFLOW_LIST_LIMIT_DEFAULT = 50	74✔
10	export const JOB_OVERFLOW_RESTORE_LIMIT_DEFAULT = 10000	74✔
11	export const QUEUE_OVERFLOW_UNSCOPED_BACKUP_MESSAGE =
12	'Backup requires at least one queue, event type, or tenant filter unless confirmAll is true'	74✔
13
14	export type QueueOverflowSource = 'job' \| 'backup'
15	export type QueueOverflowGroupBy = 'summary' \| 'tenant'
16
17	export interface QueueOverflowFilters {
18	eventTypes?: readonly string[]
19	name?: string
20	tenantRefs?: readonly string[]
21	}
22
23	interface QueueOverflowOperationOptions {
24	signal?: AbortSignal
25	}
26
27	export interface ListQueueOverflowOptions
28	extends QueueOverflowFilters,
29	QueueOverflowOperationOptions {
30	groupBy?: QueueOverflowGroupBy
31	limit?: number
32	source?: QueueOverflowSource
33	}
34
35	export interface MoveQueueOverflowOptions
36	extends QueueOverflowFilters,
37	QueueOverflowOperationOptions {
38	limit?: number
39	}
40
41	export interface BackupQueueOverflowOptions extends MoveQueueOverflowOptions {
42	confirmAll?: boolean
43	}
44
45	export interface QueueOverflowSummaryRow {
46	count: number
47	eventType: string \| null
48	name: string
49	}
50
51	export interface QueueOverflowTenantRow {
52	count: number
53	tenantRef: string \| null
54	}
55
56	interface QueueOverflowWhereClause {
57	sql: string
58	values: unknown[]
59	}
60
61	interface QueueOverflowAggregateRow extends QueryResultRow {
62	count: number \| string
63	group_count: number \| string
64	total_count: number \| string
65	}
66
67	interface QueueOverflowTotalCountRow extends QueryResultRow {
68	total_count: number \| string
69	}
70
71	interface QueueOverflowSummaryDbRow extends QueueOverflowAggregateRow {
72	event_type: string \| null
73	name: string
74	}
75
76	interface QueueOverflowTenantDbRow extends QueueOverflowAggregateRow {
77	tenant_ref: string \| null
78	}
79
80	interface QueueOverflowCountRow extends QueryResultRow {
81	moved_count: number \| string
82	}
83
84	interface QueueOverflowRestoreCountRow extends QueueOverflowCountRow {
85	selected_count: number \| string
86	}
87
88	interface QueueOverflowTableRow extends QueryResultRow {
89	table_name: string \| null
90	}
91
92	function normalizeOverflowString(value: string \| undefined) {
93	const normalized = value?.trim()	59✔
94	return normalized ? normalized : undefined	59✔
95	}
96
97	export function normalizeQueueOverflowFilters(filters: QueueOverflowFilters): QueueOverflowFilters {
98	return {	59✔
99	name: normalizeOverflowString(filters.name),
100	eventTypes: normalizeStringList(filters.eventTypes),
101	tenantRefs: normalizeStringList(filters.tenantRefs),
102	}
103	}
104
105	export function hasQueueOverflowFilters(filters: QueueOverflowFilters): boolean {
106	const normalized = normalizeQueueOverflowFilters(filters)	10✔
107	return Boolean(normalized.name \|\| normalized.eventTypes?.length \|\| normalized.tenantRefs?.length)	10✔
108	}
109
110	export function isQueueOverflowBackupScoped(options: BackupQueueOverflowOptions): boolean {
111	return options.confirmAll === true \|\| hasQueueOverflowFilters(options)	11✔
112	}
113
114	export function buildQueueOverflowWhereClause(
115	filters: QueueOverflowFilters
116	): QueueOverflowWhereClause {
117	const normalizedFilters = normalizeQueueOverflowFilters(filters)	24✔
118	const clauses = ['state = $1']	24✔
119	const values: unknown[] = [CREATED_STATE]	24✔
120
121	if (normalizedFilters.name) {	24✔
122	values.push(normalizedFilters.name)	19✔
123	clauses.push(`name = $${values.length}`)	19✔
124	}
125
126	if (normalizedFilters.eventTypes?.length) {	24✔
127	values.push(normalizedFilters.eventTypes)	4✔
128	clauses.push(`data->'event'->>'type' = ANY($${values.length}::text[])`)	4✔
129	}
130
131	if (normalizedFilters.tenantRefs?.length) {	24✔
132	values.push(normalizedFilters.tenantRefs)	8✔
133	clauses.push(`data->'tenant'->>'ref' = ANY($${values.length}::text[])`)	8✔
134	}
135
136	return {	24✔
137	sql: clauses.join(' AND '),
138	values,
139	}
140	}
141
142	export class QueueOverflowStorePg {
143	private readonly backupTableName: string
144	private readonly jobTable: string
145	private readonly backupTable: string
146
147	constructor(
148	private readonly db: Pick<PgTransactionalExecutor, 'beginTransaction'>,	26✔
149	schema = PG_BOSS_SCHEMA	26✔
150	) {
151	const jobTableName = `${schema}.job`	26✔
152	this.backupTableName = `${schema}.job_overflow_backup`	26✔
153	this.jobTable = quoteQualifiedIdentifier(jobTableName)	26✔
154	this.backupTable = quoteQualifiedIdentifier(this.backupTableName)	26✔
155	}
156
157	async countCreated(options: QueueOverflowOperationOptions = {}) {	4✔
158	return this.withMaintenanceTransaction(options.signal, async (transaction) => {	4✔
159	const result = await transaction.query<QueueOverflowTotalCountRow>(	4✔
160	{
161	text: `
162	SELECT COUNT(*)::bigint AS total_count
163	FROM ${this.jobTable}
164	WHERE state = $1
165	`,
166	values: [CREATED_STATE],
167	},
168	{ signal: options.signal }
169	)
170
171	return {	4✔
172	totalCount: Number(result.rows[0]?.total_count ?? 0),	4!
173	}
174	})
175	}
176
177	async list(options: ListQueueOverflowOptions) {
178	const source = options.source ?? 'job'	4✔
179	const groupBy = options.groupBy ?? 'summary'	4✔
180	const limit = options.limit ?? JOB_OVERFLOW_LIST_LIMIT_DEFAULT	4!
181	const filters = normalizeQueueOverflowFilters(options)	4✔
182
183	assertPositiveSafeInteger(limit, 'limit')	4✔
184
185	return this.withMaintenanceTransaction(options.signal, async (transaction) => {	4✔
186	const sourceTableExists =
187	source === 'backup' ? await this.backupTableExists(transaction, options.signal) : true	4✔
188
189	if (!sourceTableExists) {	4✔
190	return {	1✔
191	sourceTableExists,
192	data: [] as QueueOverflowSummaryRow[] \| QueueOverflowTenantRow[],
193	filters,
194	groupBy,
195	groupCount: 0,
196	hasMore: false,
197	source,
198	totalCount: 0,
199	}
200	}
201
202	const tableName = source === 'backup' ? this.backupTable : this.jobTable	3!
203	const whereClause = buildQueueOverflowWhereClause(filters)	4✔
204	const values = [...whereClause.values, limit]	4✔
205	const limitPlaceholder = `$${values.length}`	4✔
206
207	if (groupBy === 'tenant') {	4✔
208	const result = await transaction.query<QueueOverflowTenantDbRow>(	1✔
209	{
210	text: `
211	SELECT
212	data->'tenant'->>'ref' AS tenant_ref,
213	COUNT(*)::bigint AS count,
214	COUNT(*) OVER ()::bigint AS group_count,
215	SUM(COUNT(*)) OVER ()::bigint AS total_count
216	FROM ${tableName}
217	WHERE ${whereClause.sql}
218	GROUP BY data->'tenant'->>'ref'
219	ORDER BY count DESC, tenant_ref ASC
220	LIMIT ${limitPlaceholder}
221	`,
222	values,
223	},
224	{ signal: options.signal }
225	)
226
227	const groupCount = Number(result.rows[0]?.group_count ?? 0)	1!
228
229	return {	1✔
230	sourceTableExists,
231	data: result.rows.map((row) => ({	2✔
232	count: Number(row.count),
233	tenantRef: row.tenant_ref,
234	})),
235	filters,
236	groupBy,
237	groupCount,
238	hasMore: groupCount > result.rows.length,
239	source,
240	totalCount: Number(result.rows[0]?.total_count ?? 0),	1!
241	}
242	}
243
244	const result = await transaction.query<QueueOverflowSummaryDbRow>(	2✔
245	{
246	text: `
247	SELECT
248	name,
249	data->'event'->>'type' AS event_type,
250	COUNT(*)::bigint AS count,
251	COUNT(*) OVER ()::bigint AS group_count,
252	SUM(COUNT(*)) OVER ()::bigint AS total_count
253	FROM ${tableName}
254	WHERE ${whereClause.sql}
255	GROUP BY name, data->'event'->>'type'
256	ORDER BY count DESC, name ASC, event_type ASC
257	LIMIT ${limitPlaceholder}
258	`,
259	values,
260	},
261	{ signal: options.signal }
262	)
263
264	const groupCount = Number(result.rows[0]?.group_count ?? 0)	2!
265
266	return {	4✔
267	sourceTableExists,
268	data: result.rows.map((row) => ({	2✔
269	count: Number(row.count),
270	eventType: row.event_type,
271	name: row.name,
272	})),
273	filters,
274	groupBy,
275	groupCount,
276	hasMore: groupCount > result.rows.length,
277	source,
278	totalCount: Number(result.rows[0]?.total_count ?? 0),	4!
279	}
280	})
281	}
282
283	async backup(options: BackupQueueOverflowOptions) {
284	if (!isQueueOverflowBackupScoped(options)) {	11✔
285	throw new Error(QUEUE_OVERFLOW_UNSCOPED_BACKUP_MESSAGE)	2✔
286	}
287
288	assertOptionalPositiveSafeInteger(options.limit, 'limit')	9✔
289	const filters = normalizeQueueOverflowFilters(options)	9✔
290
291	return this.withMaintenanceTransaction(options.signal, async (transaction) => {	9✔
292	await this.acquireMaintenanceLock(transaction, options.signal)	9✔
293	const backupTableCreated = await this.ensureBackupTable(transaction, options.signal)	9✔
294
295	await transaction.query(`LOCK TABLE ${this.jobTable} IN SHARE ROW EXCLUSIVE MODE`, {	9✔
296	signal: options.signal,
297	})
298
299	const movedCount = await this.moveJobs(	9✔
300	transaction,
301	filters,
302	this.jobTable,
303	this.backupTable,
304	options.limit,
305	options.signal
306	)
307
308	return {	9✔
309	backupTableCreated,
310	filters,
311	limit: options.limit ?? null,	16✔
312	movedCount,
313	}
314	})
315	}
316
317	async restore(options: MoveQueueOverflowOptions) {
318	const limit = options.limit ?? JOB_OVERFLOW_RESTORE_LIMIT_DEFAULT	11✔
319	assertPositiveSafeInteger(limit, 'limit')	11✔
320	const filters = normalizeQueueOverflowFilters(options)	11✔
321
322	return this.withMaintenanceTransaction(options.signal, async (transaction) => {	11✔
323	await this.acquireMaintenanceLock(transaction, options.signal)	11✔
324	const backupTableExists = await this.backupTableExists(transaction, options.signal)	11✔
325
326	if (!backupTableExists) {	11✔
327	return {	1✔
328	backupTableExists,
329	conflictCount: 0,
330	filters,
331	hasMore: false,
332	limit,
333	movedCount: 0,
334	}
335	}
336
337	const { conflictCount, hasMore, movedCount } = await this.restoreJobs(	10✔
338	transaction,
339	filters,
340	limit,
341	options.signal
342	)
343
344	return {	8✔
345	backupTableExists,
346	conflictCount,
347	filters,
348	hasMore,
349	limit,
350	movedCount,
351	}
352	})
353	}
354
355	private async backupTableExists(db: PgExecutor, signal?: AbortSignal) {
356	const result = await db.query<QueueOverflowTableRow>(	21✔
357	{
358	text: 'SELECT to_regclass($1) AS table_name',
359	values: [this.backupTableName],
360	},
361	{ signal }
362	)
363
364	return Boolean(result.rows[0]?.table_name)	21✔
365	}
366
367	private async ensureBackupTable(db: PgExecutor, signal?: AbortSignal) {
368	const existed = await this.backupTableExists(db, signal)	9✔
369
370	await db.query(	9✔
371	`CREATE TABLE IF NOT EXISTS ${this.backupTable} (LIKE ${this.jobTable} INCLUDING ALL)`,
372	{
373	signal,
374	}
375	)
376
377	return !existed	9✔
378	}
379
380	private async moveJobs(
381	db: PgExecutor,
382	filters: QueueOverflowFilters,
383	sourceTable: string,
384	targetTable: string,
385	limit?: number,
386	signal?: AbortSignal
387	) {
388	const whereClause = buildQueueOverflowWhereClause(filters)	9✔
389	const values = [...whereClause.values]	9✔
390	let selectionBoundary = ''	9✔
391
392	if (limit !== undefined) {	9✔
393	values.push(limit)	2✔
394	selectionBoundary = `ORDER BY name, id LIMIT $${values.length}`	2✔
395	}
396
397	const result = await db.query<QueueOverflowCountRow>(	9✔
398	{
399	text: `
400	WITH selected AS (
401	SELECT name, id
402	FROM ${sourceTable}
403	WHERE ${whereClause.sql}
404	${selectionBoundary}
405	),
406	moved AS (
407	DELETE FROM ${sourceTable} AS source_job
408	USING selected
409	WHERE source_job.name = selected.name
410	AND source_job.id = selected.id
411	RETURNING source_job.*
412	),
413	inserted AS (
414	INSERT INTO ${targetTable}
415	SELECT * FROM moved
416	RETURNING 1
417	)
418	SELECT COUNT(*)::bigint AS moved_count FROM inserted
419	`,
420	values,
421	},
422	{ signal }
423	)
424
425	return Number(result.rows[0]?.moved_count ?? 0)	9!
426	}
427
428	private async restoreJobs(
429	db: PgExecutor,
430	filters: QueueOverflowFilters,
431	limit: number,
432	signal?: AbortSignal
433	) {
434	const whereClause = buildQueueOverflowWhereClause(filters)	10✔
435	const values = [...whereClause.values]	10✔
436	values.push(limit)	10✔
437	const limitPlaceholder = `$${values.length}`	10✔
438	const result = await db.query<QueueOverflowRestoreCountRow>(	10✔
439	{
440	text: `
441	WITH selected AS (
442	SELECT *
443	FROM ${this.backupTable}
444	WHERE ${whereClause.sql}
445	ORDER BY name, id
446	LIMIT ${limitPlaceholder}
447	),
448	inserted AS (
449	INSERT INTO ${this.jobTable}
450	SELECT * FROM selected
451	ON CONFLICT DO NOTHING
452	RETURNING 1
453	),
454	deleted AS (
455	DELETE FROM ${this.backupTable} AS source_job
456	USING selected
457	WHERE source_job.name = selected.name
458	AND source_job.id = selected.id
459	RETURNING 1
460	)
461	SELECT
462	(SELECT COUNT(*) FROM selected) AS selected_count,
463	(SELECT COUNT(*) FROM inserted) AS moved_count
464	`,
465	values,
466	},
467	{ signal }
468	)
469
470	const selectedCount = Number(result.rows[0]?.selected_count ?? 0)	8!
471	const movedCount = Number(result.rows[0]?.moved_count ?? 0)	10!
472
473	return {	10✔
474	conflictCount: selectedCount - movedCount,
475	hasMore: selectedCount === limit,
476	movedCount,
477	}
478	}
479
480	private async acquireMaintenanceLock(db: PgExecutor, signal?: AbortSignal): Promise<void> {
481	await db.query(	20✔
482	{
483	text: 'SELECT pg_advisory_xact_lock($1::bigint)',
484	values: [QUEUE_OVERFLOW_ADVISORY_LOCK_KEY],
485	},
486	{ signal }
487	)
488	}
489
490	private async withMaintenanceTransaction<T>(
491	signal: AbortSignal \| undefined,
492	fn: (transaction: PgTransaction) => Promise<T>
493	): Promise<T> {
494	const transaction = await this.db.beginTransaction({ signal })	28✔
495
496	try {	28✔
497	await transaction.query(	28✔
498	{
499	text: `
500	SELECT
501	set_config('statement_timeout', $1, true),
502	set_config('lock_timeout', $2, true)
503	`,
504	values: ['0', '30s'],
505	},
506	{ signal }
507	)
508
509	const result = await fn(transaction)	28✔
510	await transaction.commit()	26✔
511	return result	26✔
512	} catch (error) {
513	try {	2✔
514	await transaction.rollback()	2✔
515	} catch (rollbackError) {
NEW 516	await logQueueOverflowRollbackFailure(error, rollbackError)	×
517	}
518
519	throw error	2✔
520	}
521	}
522	}
523
524	async function logQueueOverflowRollbackFailure(
525	originalError: unknown,
526	rollbackError: unknown
527	): Promise<void> {
NEW 528	try {	×
NEW 529	const { logger, logSchema } = await import('@internal/monitoring')	×
NEW 530	logSchema.warning(logger, '[QueueOverflow] Failed to rollback maintenance transaction', {	×
531	type: 'pgboss',
532	error: rollbackError,
533	metadata: JSON.stringify({ originalError: String(originalError) }),
534	})
535	} catch (loggingError) {
NEW 536	console.error('[QueueOverflow] Failed to log maintenance transaction rollback failure', {	×
537	originalError,
538	rollbackError,
539	loggingError,
540	})
541	}
542	}
543
544	function assertOptionalPositiveSafeInteger(value: number \| undefined, field: string): void {
545	if (value !== undefined) {	9✔
546	assertPositiveSafeInteger(value, field)	2✔
547	}
548	}
549
550	function assertPositiveSafeInteger(value: number, field: string): void {
551	if (!Number.isSafeInteger(value) \|\| value <= 0) {	17!
NEW 552	throw new Error(`${field} must be a positive safe integer`)	×
553	}
554	}

supabase / storage / 29583752771

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