15439109207

Committed 04 Jun 2025 09:46AM UTC coverage: 77.775% (-0.4%) from 78.159%

Build # 15439109207

Build Type

Pull #696

github

Committed by

web-flow

Commit Message

Merge 10a96e63f into d82ebecc1

Pull Request Pull Request #696: feat: pgboss v10

Coverage Stats

1538 of 2141 branches covered (71.84%)

Branch coverage included in aggregate %.

172 of 475 new or added lines in 28 files covered. (36.21%)

5 existing lines in 3 files now uncovered.

17387 of 22192 relevant lines covered (78.35%)

110.03 hits per line

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

56.28

/src/internal/queue/event.ts

import { Queue } from './queue'
import PgBoss, { Job, SendOptions, WorkOptions, Queue as PgBossQueue } from 'pg-boss'
import { getConfig } from '../../config'
import { QueueJobScheduled, QueueJobSchedulingTime } from '@internal/monitoring/metrics'
import { logger, logSchema } from '@internal/monitoring'
import { getTenantConfig } from '@internal/database'

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

const { pgQueueEnable, region, isMultitenant } = getConfig()

export type StaticThis<T extends Event<any>> = BaseEventConstructor<T>

interface BaseEventConstructor<Base extends Event<any>> {
  version: string

  new (...args: any): Base

  send(
    this: StaticThis<Base>,
    payload: Omit<Base['payload'], '$version'>
  ): Promise<string | void | null>

  eventName(): string
  getWorkerOptions(): WorkOptions
}

/**
 * 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<T extends Event<any>>(payload: T['payload']): SendOptions | undefined {
    return undefined
  }

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

  static onClose() {
    // no-op
  }

  static onStart() {
    // no-op
  }

  static batchSend<T extends Event<any>[]>(messages: T) {
    if (!pgQueueEnable) {
      if (this.allowSync) {
        return Promise.all(messages.map((message) => message.send()))
      } else {
        logger.warn('[Queue] skipped sending batch messages', {
          type: 'queue',
          eventType: this.eventName(),
        })
        return
      }
    }

    return Queue.getInstance().insert(
      messages.map((message) => {
        const sendOptions = (this.getSendOptions(message.payload) as PgBoss.JobInsert) || {}
        if (!message.payload.$version) {
          ;(message.payload as (typeof message)['payload']).$version = this.version
        }

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

        return {
          ...sendOptions,
          name: this.getQueueName(),
          data: message.payload,
          deadLetter: this.deadLetterQueueName(),
        }
      })
    )
  }

  static send<T extends Event<any>>(this: StaticThis<T>, payload: Omit<T['payload'], '$version'>) {
    if (!payload.$version) {
      ;(payload as T['payload']).$version = this.version
    }
    const that = new this(payload)
    return that.send()
  }

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

  static async shouldSend(payload: any) {
    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
  }

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

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

    if (!shouldSend) {
      return
    }

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

    const timer = QueueJobSchedulingTime.startTimer()
    const sendOptions = constructor.getSendOptions(this.payload) || {}

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

    sendOptions!.deadLetter = constructor.deadLetterQueueName()

    try {
      const res = await Queue.getInstance().send({
        name: constructor.getQueueName(),
        data: {
          region,
          ...this.payload,
          $version: constructor.version,
        },
        options: sendOptions,
      })

      QueueJobScheduled.inc({
        name: constructor.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',
          error: e,
          metadata: JSON.stringify(this.payload),
        }
      )
      return constructor.handle({
        id: '__sync',
        expireInSeconds: 0,
        name: constructor.getQueueName(),
        data: {
          region,
          ...this.payload,
          $version: constructor.version,
        },
      })
    } finally {
      timer({
        name: constructor.getQueueName(),
      })
    }
  }
}

1	import { Queue } from './queue'	1✔
2	import PgBoss, { Job, SendOptions, WorkOptions, Queue as PgBossQueue } from 'pg-boss'	1✔
3	import { getConfig } from '../../config'	1✔
4	import { QueueJobScheduled, QueueJobSchedulingTime } from '@internal/monitoring/metrics'	1✔
5	import { logger, logSchema } from '@internal/monitoring'	1✔
6	import { getTenantConfig } from '@internal/database'	1✔
7		1✔
8	export interface BasePayload {	1✔
9	$version?: string	1✔
10	singletonKey?: string	1✔
11	scheduleAt?: Date	1✔
12	reqId?: string	1✔
13	tenant: {	1✔
14	ref: string	1✔
15	host: string	1✔
16	}	1✔
17	}	1✔
18		1✔
19	const { pgQueueEnable, region, isMultitenant } = getConfig()	1✔
20		1✔
21	export type StaticThis<T extends Event<any>> = BaseEventConstructor<T>	1✔
22		1✔
23	interface BaseEventConstructor<Base extends Event<any>> {	1✔
24	version: string	1✔
25		1✔
26	new (...args: any): Base	1✔
27		1✔
28	send(	1✔
29	this: StaticThis<Base>,	1✔
30	payload: Omit<Base['payload'], '$version'>	1✔
31	): Promise<string \| void \| null>	1✔
32		1✔
33	eventName(): string	1✔
34	getWorkerOptions(): WorkOptions	1✔
35	}	1✔
36		1✔
37	/**	1✔
38	* Base class for all events that are sent to the queue	1✔
39	*/	1✔
40	export class Event<T extends Omit<BasePayload, '$version'>> {	1✔
41	public static readonly version: string = 'v1'	1✔
42	protected static queueName = ''	1✔
43	protected static allowSync = true	1✔
44		1✔
45	constructor(public readonly payload: T & BasePayload) {}	1✔
46		1✔
47	static eventName() {	1✔
48	return this.name	×
49	}	×
50		1✔
51	static deadLetterQueueName() {	1✔
52	return this.queueName + '-dead-letter'	1✔
53	}	1✔
54		1✔
55	static getQueueName() {	1✔
56	if (!this.queueName) {	10,042!
57	throw new Error(`Queue name not set on ${this.constructor.name}`)	×
58	}	×
59		10,042✔
60	return this.queueName	10,042✔
61	}	10,042✔
62		1✔
63	static getQueueOptions(): PgBossQueue \| undefined {	1✔
UNCOV 64	return undefined	×
UNCOV 65	}	×
66		1✔
67	static getSendOptions<T extends Event<any>>(payload: T['payload']): SendOptions \| undefined {	1✔
68	return undefined	×
69	}	×
70		1✔
71	static getWorkerOptions(): WorkOptions & { concurrentTaskCount?: number } {	1✔
NEW 72	return {}	×
UNCOV 73	}	×
74		1✔
75	static onClose() {	1✔
76	// no-op	×
77	}	×
78		1✔
79	static onStart() {	1✔
80	// no-op	×
81	}	×
82		1✔
83	static batchSend<T extends Event<any>[]>(messages: T) {	1✔
84	if (!pgQueueEnable) {	1!
85	if (this.allowSync) {	×
86	return Promise.all(messages.map((message) => message.send()))	✔
87	} else {	×
88	logger.warn('[Queue] skipped sending batch messages', {	×
89	type: 'queue',	×
90	eventType: this.eventName(),	×
91	})	×
92	return	×
93	}	×
94	}	×
95		1✔
96	return Queue.getInstance().insert(	1✔
97	messages.map((message) => {	1✔
98	const sendOptions = (this.getSendOptions(message.payload) as PgBoss.JobInsert) \|\| {}	1!
99	if (!message.payload.$version) {	1✔
100	;(message.payload as (typeof message)['payload']).$version = this.version	1✔
101	}	1✔
102		1✔
103	if (message.payload.scheduleAt) {	1!
104	sendOptions.startAfter = new Date(message.payload.scheduleAt)	×
105	}	×
106		1✔
107	return {	1✔
108	...sendOptions,	1✔
109	name: this.getQueueName(),	1✔
110	data: message.payload,	1✔
111	deadLetter: this.deadLetterQueueName(),	1✔
112	}	1✔
113	})	1✔
114	)	1✔
115	}	1✔
116		1✔
117	static send<T extends Event<any>>(this: StaticThis<T>, payload: Omit<T['payload'], '$version'>) {	1✔
118	if (!payload.$version) {	10,042✔
119	;(payload as T['payload']).$version = this.version	10,042✔
120	}	10,042✔
121	const that = new this(payload)	10,042✔
122	return that.send()	10,042✔
123	}	10,042✔
124		1✔
125	static handle(	1✔
NEW 126	job: Job<Event<any>['payload']> \| Job<Event<any>['payload']>[],	×
NEW 127	opts?: { signal?: AbortSignal }	×
128	) {	×
129	throw new Error('not implemented')	×
130	}	×
131		1✔
132	static async shouldSend(payload: any) {	1✔
133	if (isMultitenant && payload?.tenant?.ref) {	14!
134	// Do not send an event if disabled for this specific tenant	×
135	const tenant = await getTenantConfig(payload.tenant.ref)	×
136	const disabledEvents = tenant.disableEvents \|\| []	×
137	if (disabledEvents.includes(this.eventName())) {	×
138	return false	×
139	}	×
140	}	×
141	return true	14✔
142	}	14✔
143		1✔
144	async send(): Promise<string \| void \| null> {	1✔
145	const constructor = this.constructor as typeof Event	10,042✔
146		10,042✔
147	const shouldSend = await constructor.shouldSend(this.payload)	10,042✔
148		10,042✔
149	if (!shouldSend) {	10,042!
150	return	×
151	}	×
152		10,042✔
153	if (!pgQueueEnable) {	10,042✔
154	if (constructor.allowSync) {	10,042✔
155	return constructor.handle({	10,042✔
156	id: '__sync',	10,042✔
157	expireInSeconds: 0,	10,042✔
158	name: constructor.getQueueName(),	10,042✔
159	data: {	10,042✔
160	region,	10,042✔
161	...this.payload,	10,042✔
162	$version: constructor.version,	10,042✔
163	},	10,042✔
164	})	10,042✔
165	} else {	10,042!
166	logger.warn('[Queue] skipped sending message', {	×
167	type: 'queue',	×
168	eventType: constructor.eventName(),	×
169	})	×
170	return	×
171	}	×
172	}	10,042✔
173		×
174	const timer = QueueJobSchedulingTime.startTimer()	×
NEW 175	const sendOptions = constructor.getSendOptions(this.payload) \|\| {}	×
176		10,042✔
177	if (this.payload.scheduleAt) {	10,042!
178	sendOptions.startAfter = new Date(this.payload.scheduleAt)	×
179	}	×
180		×
NEW 181	sendOptions!.deadLetter = constructor.deadLetterQueueName()	×
NEW 182		×
183	try {	×
184	const res = await Queue.getInstance().send({	×
185	name: constructor.getQueueName(),	×
186	data: {	×
187	region,	×
188	...this.payload,	×
189	$version: constructor.version,	×
190	},	×
191	options: sendOptions,	×
192	})	×
193		×
194	QueueJobScheduled.inc({	×
195	name: constructor.getQueueName(),	×
196	})	×
197		×
198	return res	×
199	} catch (e) {	×
200	// If we can't queue the message for some reason,	×
201	// we run its handler right away.	×
202	// This might create some latency with the benefit of being more fault-tolerant	×
203	logSchema.warning(	×
204	logger,	×
205	`[Queue Sender] Error while sending job to queue, sending synchronously`,	×
206	{	×
207	type: 'queue',	×
208	error: e,	×
209	metadata: JSON.stringify(this.payload),	×
210	}	×
211	)	×
212	return constructor.handle({	×
213	id: '__sync',	×
NEW 214	expireInSeconds: 0,	×
215	name: constructor.getQueueName(),	×
216	data: {	×
217	region,	×
218	...this.payload,	×
219	$version: constructor.version,	×
220	},	×
221	})	×
222	} finally {	×
223	timer({	×
224	name: constructor.getQueueName(),	×
225	})	×
226	}	×
227	}	10,042✔
228	}	1✔

supabase / storage / 15439109207

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