30825450288

Committed 03 Aug 2026 03:00PM UTC coverage: 60.991% (-19.4%) from 80.366%

Build # 30825450288

Build Type

Pull #1293

github

Committed by

web-flow

Commit Message

Merge 838eef931 into 1e33d8708

Pull Request Pull Request #1293: fix: hardening for non-json object reply

Coverage Stats

3709 of 6900 branches covered (53.75%)

Branch coverage included in aggregate %.

22 of 32 new or added lines in 5 files covered. (68.75%)

2204 existing lines in 109 files now uncovered.

7686 of 11783 relevant lines covered (65.23%)

384.33 hits per line

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

55.15

/src/internal/queue/event.ts

import { type DatabaseExecutor, getTenantConfig } from '@internal/database'
import { ERRORS } from '@internal/errors'
import { logger, logSchema } from '@internal/monitoring'
import { queueJobScheduled, queueJobSchedulingTime } from '@internal/monitoring/metrics'
import { PgQueueDB } from '@internal/queue/database'
import PgBoss, { Job, Queue as PgBossQueue, SendOptions, WorkOptions } from 'pg-boss'
import { getConfig } from '../../config'
import { SYSTEM_TENANT_REF } from './constants'
import { PG_BOSS_SCHEMA, Queue } from './queue'

export interface BasePayload {
  $version?: string
  singletonKey?: string
  scheduleAt?: Date
  reqId?: string
  sbReqId?: string
  tenant: {
    ref: string
    host: string
  }
}

const { pgQueueEnable, region, isMultitenant } = getConfig()
type TransactionalQueueDb = DatabaseExecutor

function withPayloadVersion<TPayload extends BasePayload>(
  payload: TPayload,
  version: string
): TPayload {
  return {
    ...payload,
    $version: payload.$version ?? version,
  }
}

export type EventInputPayload = Omit<BasePayload, '$version'>
export type QueueEvent<T extends EventInputPayload = EventInputPayload> = Event<T>
export type StaticThis<TPayload extends BasePayload> = BaseEventConstructor<TPayload>

interface BaseEventConstructor<TPayload extends BasePayload> {
  version: string

  new (payload: TPayload): QueueEvent<Omit<TPayload, '$version'>>
}

/**
 * Base class for all events that are sent to the queue
 */
export class Event<T extends Omit<BasePayload, '$version'>> {
  public static readonly version: string = 'v1'
  protected static queueName = ''
  protected static allowSync = true

  constructor(public readonly payload: T & BasePayload) {}

  static eventName() {
    return this.name
  }

  static deadLetterQueueName() {
    return this.queueName + '-dead-letter'
  }

  static getQueueName() {
    if (!this.queueName) {
      throw new Error(`Queue name not set on ${this.constructor.name}`)
    }

    return this.queueName
  }

  static getQueueOptions(): PgBossQueue | undefined {
    return undefined
  }

  static getSendOptions(payload: BasePayload): SendOptions | undefined {
    return undefined
  }

  static getWorkerOptions(): WorkOptions & { concurrentTaskCount?: number } {
    return {}
  }

  static onClose() {
    // no-op
  }

  static onStart() {
    // no-op
  }

  static batchSend<TPayload extends BasePayload>(
    this: StaticThis<TPayload>,
    messages: Array<{ payload: TPayload; send(): Promise<string | void | null> }>
  ) {
    const eventClass = this as typeof Event

    if (!pgQueueEnable) {
      if (eventClass.allowSync) {
        return Promise.all(messages.map((message) => message.send()))
      } else {
        logger.warn(
          {
            type: 'queue',
            eventType: eventClass.eventName(),
          },
          '[Queue] skipped sending batch messages'
        )
        return
      }
    }

    return Queue.getInstance().insert(
      messages.map((message) => {
        const payloadWithVersion = withPayloadVersion(message.payload, eventClass.version)
        const sendOptions =
          (eventClass.getSendOptions(payloadWithVersion) as PgBoss.JobInsert) || {}

        if (payloadWithVersion.scheduleAt) {
          sendOptions.startAfter = new Date(payloadWithVersion.scheduleAt)
        }

        return {
          ...sendOptions,
          name: eventClass.getQueueName(),
          data: payloadWithVersion,
          deadLetter: eventClass.deadLetterQueueName(),
        }
      })
    )
  }

  static send<TPayload extends BasePayload>(
    this: StaticThis<TPayload>,
    payload: Omit<TPayload, '$version'>,
    opts?: SendOptions & { tnx?: TransactionalQueueDb }
  ) {
    const that = new this(withPayloadVersion(payload as TPayload, this.version))
    return that.send(opts)
  }

  static invoke<TPayload extends BasePayload>(
    this: StaticThis<TPayload>,
    payload: Omit<TPayload, '$version'>
  ) {
    const that = new this(withPayloadVersion(payload as TPayload, this.version))
    return that.invoke()
  }

  static invokeOrSend<TPayload extends BasePayload>(
    this: StaticThis<TPayload>,
    payload: Omit<TPayload, '$version'>,
    options?: SendOptions & { sendWhenError?: (error: unknown) => boolean }
  ) {
    const that = new this(withPayloadVersion(payload as TPayload, this.version))
    return that.invokeOrSend(options)
  }

  static handle(job: Job<BasePayload> | Job<BasePayload>[], opts?: { signal?: AbortSignal }) {
    throw new Error('not implemented')
  }

  static async shouldSend(payload: BasePayload) {
    if (isMultitenant && payload?.tenant?.ref) {
      // Do not send an event if disabled for this specific tenant
      const tenant = await getTenantConfig(payload.tenant.ref)
      const disabledEvents = tenant.disableEvents || []
      if (disabledEvents.includes(this.eventName())) {
        return false
      }
    }
    return true
  }

  /**
   * See issue https://github.com/timgit/pg-boss/issues/535
   * @param queueName
   * @param singletonKey
   * @param jobId
   */
  static async deleteIfActiveExists(queueName: string, singletonKey: string, jobId: string) {
    if (!pgQueueEnable) {
      return Promise.resolve()
    }

    await Queue.getDb().executeSql(
      `DELETE FROM ${PG_BOSS_SCHEMA}.job
       WHERE id = $1
       AND EXISTS(
          SELECT 1 FROM ${PG_BOSS_SCHEMA}.job
             WHERE id != $2
             AND state < 'active'
             AND name = $3
             AND singleton_key = $4
       )
      `,
      [jobId, jobId, queueName, singletonKey]
    )
  }

  async invokeOrSend(
    sendOptions?: SendOptions & { sendWhenError?: (error: unknown) => boolean }
  ): Promise<string | void | null> {
    const eventClass = this.constructor as typeof Event

    if (!eventClass.allowSync) {
      throw ERRORS.InternalError(undefined, 'Cannot send this event synchronously')
    }

    try {
      await this.invoke()
    } catch (e) {
      if (sendOptions?.sendWhenError && !sendOptions.sendWhenError(e)) {
        throw e
      }

      logSchema.error(logger, '[Queue] Error invoking event synchronously, sending to queue', {
        type: 'queue',
        project: this.payload.tenant?.ref || SYSTEM_TENANT_REF,
        error: e,
        metadata: JSON.stringify(this.payload),
        sbReqId: this.payload.sbReqId,
      })

      return this.send(sendOptions)
    }
  }

  async invoke(): Promise<string | void | null> {
    const eventClass = this.constructor as typeof Event

    if (!eventClass.allowSync) {
      throw ERRORS.InternalError(undefined, 'Cannot send this event synchronously')
    }

    await eventClass.handle({
      id: '__sync',
      expireInSeconds: 0,
      name: eventClass.getQueueName(),
      data: {
        region,
        ...this.payload,
        $version: eventClass.version,
      },
    })
  }

  async send(
    customSendOptions?: SendOptions & { tnx?: TransactionalQueueDb }
  ): Promise<string | void | null> {
    const eventClass = this.constructor as typeof Event

    const shouldSend = await eventClass.shouldSend(this.payload)

    if (!shouldSend) {
      return
    }

    if (!pgQueueEnable) {
      if (eventClass.allowSync) {
        return eventClass.handle({
          id: '__sync',
          expireInSeconds: 0,
          name: eventClass.getQueueName(),
          data: {
            region,
            ...this.payload,
            $version: eventClass.version,
          },
        })
      } else {
        logger.warn(
          {
            type: 'queue',
            eventType: eventClass.eventName(),
          },
          '[Queue] skipped sending message'
        )
        return
      }
    }

    const startTime = performance.now()
    const sendOptions = eventClass.getSendOptions(this.payload) || {}

    if (this.payload.scheduleAt) {
      sendOptions.startAfter = new Date(this.payload.scheduleAt)
    }

    sendOptions!.deadLetter = eventClass.deadLetterQueueName()

    try {
      const queue = customSendOptions?.tnx
        ? Queue.createPgBoss({
            enableWorkers: false,
            db: createTransactionQueueDB(customSendOptions.tnx),
          })
        : Queue.getInstance()

      const res = await queue.send({
        name: eventClass.getQueueName(),
        data: {
          region,
          ...this.payload,
          $version: eventClass.version,
        },
        options: {
          ...sendOptions,
          ...customSendOptions,
        },
      })

      // pg-boss returns null when the insert was dropped by a queue policy
      // (e.g. an exactly_once job with the same singleton key is still queued or active).
      if (res === null) {
        logSchema.info(logger, `[Queue Sender] Job not queued, dropped by queue policy`, {
          type: 'queue',
          project: this.payload.tenant?.ref || SYSTEM_TENANT_REF,
          metadata: JSON.stringify({ queue: eventClass.getQueueName() }),
          sbReqId: this.payload.sbReqId,
        })

        return res
      }

      queueJobScheduled.add(1, {
        name: eventClass.getQueueName(),
      })

      return res
    } catch (e) {
      // If we can't queue the message for some reason,
      // we run its handler right away.
      // This might create some latency with the benefit of being more fault-tolerant
      logSchema.warning(
        logger,
        `[Queue Sender] Error while sending job to queue, sending synchronously`,
        {
          type: 'queue',
          project: this.payload.tenant?.ref || SYSTEM_TENANT_REF,
          error: e,
          metadata: JSON.stringify(this.payload),
          sbReqId: this.payload.sbReqId,
        }
      )

      if (!eventClass.allowSync) {
        throw e
      }

      return eventClass.handle({
        id: '__sync',
        expireInSeconds: 0,
        name: eventClass.getQueueName(),
        data: {
          region,
          ...this.payload,
          $version: eventClass.version,
        },
      })
    } finally {
      const duration = (performance.now() - startTime) / 1000
      queueJobSchedulingTime.record(duration, {
        name: eventClass.getQueueName(),
      })
    }
  }
}

function createTransactionQueueDB(tnx: TransactionalQueueDb) {
  return new PgQueueDB(tnx)
}

1	import { type DatabaseExecutor, getTenantConfig } from '@internal/database'
2	import { ERRORS } from '@internal/errors'
3	import { logger, logSchema } from '@internal/monitoring'
4	import { queueJobScheduled, queueJobSchedulingTime } from '@internal/monitoring/metrics'
5	import { PgQueueDB } from '@internal/queue/database'
6	import PgBoss, { Job, Queue as PgBossQueue, SendOptions, WorkOptions } from 'pg-boss'
7	import { getConfig } from '../../config'
8	import { SYSTEM_TENANT_REF } from './constants'
9	import { PG_BOSS_SCHEMA, Queue } from './queue'
10
11	export interface BasePayload {
12	$version?: string
13	singletonKey?: string
14	scheduleAt?: Date
15	reqId?: string
16	sbReqId?: string
17	tenant: {
18	ref: string
19	host: string
20	}
21	}
22
23	const { pgQueueEnable, region, isMultitenant } = getConfig()	44✔
24	type TransactionalQueueDb = DatabaseExecutor
25
26	function withPayloadVersion<TPayload extends BasePayload>(
27	payload: TPayload,
28	version: string
29	): TPayload {
30	return {	1,807✔
31	...payload,
32	$version: payload.$version ?? version,	3,614✔
33	}
34	}
35
36	export type EventInputPayload = Omit<BasePayload, '$version'>
37	export type QueueEvent<T extends EventInputPayload = EventInputPayload> = Event<T>
38	export type StaticThis<TPayload extends BasePayload> = BaseEventConstructor<TPayload>
39
40	interface BaseEventConstructor<TPayload extends BasePayload> {
41	version: string
42
43	new (payload: TPayload): QueueEvent<Omit<TPayload, '$version'>>
44	}
45
46	/**
47	* Base class for all events that are sent to the queue
48	*/
49	export class Event<T extends Omit<BasePayload, '$version'>> {
50	public static readonly version: string = 'v1'	44✔
51	protected static queueName = ''	44✔
52	protected static allowSync = true	44✔
53
54	constructor(public readonly payload: T & BasePayload) {}	1,817✔
55
56	static eventName() {
57	return this.name	3✔
58	}
59
60	static deadLetterQueueName() {
61	return this.queueName + '-dead-letter'	1,077✔
62	}
63
64	static getQueueName() {
65	if (!this.queueName) {	6,141!
66	throw new Error(`Queue name not set on ${this.constructor.name}`)	×
67	}
68
69	return this.queueName	6,141✔
70	}
71
72	static getQueueOptions(): PgBossQueue \| undefined {
73	return undefined	×
74	}
75
76	static getSendOptions(payload: BasePayload): SendOptions \| undefined {
UNCOV 77	return undefined	×
78	}
79
80	static getWorkerOptions(): WorkOptions & { concurrentTaskCount?: number } {
81	return {}	×
82	}
83
84	static onClose() {
85	// no-op
86	}
87
88	static onStart() {
89	// no-op
90	}
91
92	static batchSend<TPayload extends BasePayload>(
93	this: StaticThis<TPayload>,
94	messages: Array<{ payload: TPayload; send(): Promise<string \| void \| null> }>
95	) {
96	const eventClass = this as typeof Event	2✔
97
98	if (!pgQueueEnable) {	2✔
99	if (eventClass.allowSync) {	1!
100	return Promise.all(messages.map((message) => message.send()))	10✔
101	} else {
102	logger.warn(	×
103	{
104	type: 'queue',
105	eventType: eventClass.eventName(),
106	},
107	'[Queue] skipped sending batch messages'
108	)
109	return	×
110	}
111	}
112
113	return Queue.getInstance().insert(	1✔
114	messages.map((message) => {
115	const payloadWithVersion = withPayloadVersion(message.payload, eventClass.version)	1✔
116	const sendOptions =
117	(eventClass.getSendOptions(payloadWithVersion) as PgBoss.JobInsert) \|\| {}	1!
118
119	if (payloadWithVersion.scheduleAt) {	1!
UNCOV 120	sendOptions.startAfter = new Date(payloadWithVersion.scheduleAt)	×
121	}
122
123	return {	1✔
124	...sendOptions,
125	name: eventClass.getQueueName(),
126	data: payloadWithVersion,
127	deadLetter: eventClass.deadLetterQueueName(),
128	}
129	})
130	)
131	}
132
133	static send<TPayload extends BasePayload>(
134	this: StaticThis<TPayload>,
135	payload: Omit<TPayload, '$version'>,
136	opts?: SendOptions & { tnx?: TransactionalQueueDb }
137	) {
138	const that = new this(withPayloadVersion(payload as TPayload, this.version))	1,781✔
139	return that.send(opts)	1,781✔
140	}
141
142	static invoke<TPayload extends BasePayload>(
143	this: StaticThis<TPayload>,
144	payload: Omit<TPayload, '$version'>
145	) {
146	const that = new this(withPayloadVersion(payload as TPayload, this.version))	3✔
147	return that.invoke()	3✔
148	}
149
150	static invokeOrSend<TPayload extends BasePayload>(
151	this: StaticThis<TPayload>,
152	payload: Omit<TPayload, '$version'>,
153	options?: SendOptions & { sendWhenError?: (error: unknown) => boolean }
154	) {
155	const that = new this(withPayloadVersion(payload as TPayload, this.version))	22✔
156	return that.invokeOrSend(options)	22✔
157	}
158
159	static handle(job: Job<BasePayload> \| Job<BasePayload>[], opts?: { signal?: AbortSignal }) {
160	throw new Error('not implemented')	×
161	}
162
163	static async shouldSend(payload: BasePayload) {
164	if (isMultitenant && payload?.tenant?.ref) {	63!
165	// Do not send an event if disabled for this specific tenant
166	const tenant = await getTenantConfig(payload.tenant.ref)	×
167	const disabledEvents = tenant.disableEvents \|\| []	×
168	if (disabledEvents.includes(this.eventName())) {	×
169	return false	×
170	}
171	}
172	return true	63✔
173	}
174
175	/**
176	* See issue https://github.com/timgit/pg-boss/issues/535
177	* @param queueName
178	* @param singletonKey
179	* @param jobId
180	*/
181	static async deleteIfActiveExists(queueName: string, singletonKey: string, jobId: string) {
182	if (!pgQueueEnable) {	×
183	return Promise.resolve()	×
184	}
185
186	await Queue.getDb().executeSql(	×
187	`DELETE FROM ${PG_BOSS_SCHEMA}.job
188	WHERE id = $1
189	AND EXISTS(
190	SELECT 1 FROM ${PG_BOSS_SCHEMA}.job
191	WHERE id != $2
192	AND state < 'active'
193	AND name = $3
194	AND singleton_key = $4
195	)
196	`,
197	[jobId, jobId, queueName, singletonKey]
198	)
199	}
200
201	async invokeOrSend(
202	sendOptions?: SendOptions & { sendWhenError?: (error: unknown) => boolean }
203	): Promise<string \| void \| null> {
204	const eventClass = this.constructor as typeof Event	22✔
205
206	if (!eventClass.allowSync) {	22!
207	throw ERRORS.InternalError(undefined, 'Cannot send this event synchronously')	×
208	}
209
210	try {	22✔
211	await this.invoke()	22✔
212	} catch (e) {
213	if (sendOptions?.sendWhenError && !sendOptions.sendWhenError(e)) {	×
214	throw e	×
215	}
216
217	logSchema.error(logger, '[Queue] Error invoking event synchronously, sending to queue', {	×
218	type: 'queue',
219	project: this.payload.tenant?.ref \|\| SYSTEM_TENANT_REF,	×
220	error: e,
221	metadata: JSON.stringify(this.payload),
222	sbReqId: this.payload.sbReqId,
223	})
224
225	return this.send(sendOptions)	×
226	}
227	}
228
229	async invoke(): Promise<string \| void \| null> {
230	const eventClass = this.constructor as typeof Event	25✔
231
232	if (!eventClass.allowSync) {	25!
233	throw ERRORS.InternalError(undefined, 'Cannot send this event synchronously')	×
234	}
235
236	await eventClass.handle({	25✔
237	id: '__sync',
238	expireInSeconds: 0,
239	name: eventClass.getQueueName(),
240	data: {
241	region,
242	...this.payload,
243	$version: eventClass.version,
244	},
245	})
246	}
247
248	async send(
249	customSendOptions?: SendOptions & { tnx?: TransactionalQueueDb }
250	): Promise<string \| void \| null> {
251	const eventClass = this.constructor as typeof Event	1,791✔
252
253	const shouldSend = await eventClass.shouldSend(this.payload)	1,791✔
254
255	if (!shouldSend) {	1,791!
256	return	×
257	}
258
259	if (!pgQueueEnable) {	1,791✔
260	if (eventClass.allowSync) {	715✔
261	return eventClass.handle({	712✔
262	id: '__sync',
263	expireInSeconds: 0,
264	name: eventClass.getQueueName(),
265	data: {
266	region,
267	...this.payload,
268	$version: eventClass.version,
269	},
270	})
271	} else {
272	logger.warn(	3✔
273	{
274	type: 'queue',
275	eventType: eventClass.eventName(),
276	},
277	'[Queue] skipped sending message'
278	)
279	return	3✔
280	}
281	}
282
283	const startTime = performance.now()	1,076✔
284	const sendOptions = eventClass.getSendOptions(this.payload) \|\| {}	1,076!
285
286	if (this.payload.scheduleAt) {	1,791!
287	sendOptions.startAfter = new Date(this.payload.scheduleAt)	×
288	}
289
290	sendOptions!.deadLetter = eventClass.deadLetterQueueName()	1,076✔
291
292	try {	1,076✔
293	const queue = customSendOptions?.tnx	1,076!
294	? Queue.createPgBoss({
295	enableWorkers: false,
296	db: createTransactionQueueDB(customSendOptions.tnx),
297	})
298	: Queue.getInstance()
299
300	const res = await queue.send({	1,791✔
301	name: eventClass.getQueueName(),
302	data: {
303	region,
304	...this.payload,
305	$version: eventClass.version,
306	},
307	options: {
308	...sendOptions,
309	...customSendOptions,
310	},
311	})
312
313	// pg-boss returns null when the insert was dropped by a queue policy
314	// (e.g. an exactly_once job with the same singleton key is still queued or active).
315	if (res === null) {	1,076!
316	logSchema.info(logger, `[Queue Sender] Job not queued, dropped by queue policy`, {	×
317	type: 'queue',
318	project: this.payload.tenant?.ref \|\| SYSTEM_TENANT_REF,	×
319	metadata: JSON.stringify({ queue: eventClass.getQueueName() }),
320	sbReqId: this.payload.sbReqId,
321	})
322
323	return res	×
324	}
325
326	queueJobScheduled.add(1, {	1,076✔
327	name: eventClass.getQueueName(),
328	})
329
330	return res	1,076✔
331	} catch (e) {
332	// If we can't queue the message for some reason,
333	// we run its handler right away.
334	// This might create some latency with the benefit of being more fault-tolerant
335	logSchema.warning(	×
336	logger,
337	`[Queue Sender] Error while sending job to queue, sending synchronously`,
338	{
339	type: 'queue',
340	project: this.payload.tenant?.ref \|\| SYSTEM_TENANT_REF,	×
341	error: e,
342	metadata: JSON.stringify(this.payload),
343	sbReqId: this.payload.sbReqId,
344	}
345	)
346
347	if (!eventClass.allowSync) {	×
348	throw e	×
349	}
350
351	return eventClass.handle({	×
352	id: '__sync',
353	expireInSeconds: 0,
354	name: eventClass.getQueueName(),
355	data: {
356	region,
357	...this.payload,
358	$version: eventClass.version,
359	},
360	})
361	} finally {
362	const duration = (performance.now() - startTime) / 1000	1,076✔
363	queueJobSchedulingTime.record(duration, {	1,076✔
364	name: eventClass.getQueueName(),
365	})
366	}
367	}
368	}
369
370	function createTransactionQueueDB(tnx: TransactionalQueueDb) {
UNCOV 371	return new PgQueueDB(tnx)	×
372	}

supabase / storage / 30825450288

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