14354470908

Committed 09 Apr 2025 10:05AM UTC coverage: 77.658% (-0.1%) from 77.767%

Build # 14354470908

Build Type

push

github

Committed by

web-flow

Commit Message

fix: improve migrations of prefixes (#644)

Coverage Stats

1424 of 1987 branches covered (71.67%)

Branch coverage included in aggregate %.

30 of 98 new or added lines in 7 files covered. (30.61%)

1 existing line in 1 file now uncovered.

16407 of 20974 relevant lines covered (78.23%)

156.77 hits per line

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

48.48

/src/internal/queue/event.ts

import { Queue } from './queue'
import PgBoss, { BatchWorkOptions, Job, SendOptions, WorkOptions } 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
  }
}

export interface SlowRetryQueueOptions {
  retryLimit: number
  retryDelay: number
}

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 | BatchWorkOptions
}

/**
 * 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 getQueueName() {
    if (!this.queueName) {
      throw new Error(`Queue name not set on ${this.constructor.name}`)
    }

    return this.queueName
  }

  static getQueueOptions<T extends Event<any>>(payload: T['payload']): SendOptions | undefined {
    return undefined
  }

  static getWorkerOptions(): WorkOptions | BatchWorkOptions {
    return {}
  }

  static withSlowRetryQueue(): undefined | SlowRetryQueueOptions {
    return undefined
  }

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

    return this.queueName + '-slow'
  }

  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.getQueueOptions(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,
        }
      })
    )
  }

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

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

  static async shouldSend(payload: any) {
    if (isMultitenant) {
      // 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',
          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()
    let sendOptions = constructor.getQueueOptions(this.payload)

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

    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',
        name: constructor.getQueueName(),
        data: {
          region,
          ...this.payload,
          $version: constructor.version,
        },
      })
    } finally {
      timer({
        name: constructor.getQueueName(),
      })
    }
  }

  async sendSlowRetryQueue() {
    const constructor = this.constructor as typeof Event
    const slowRetryQueue = constructor.withSlowRetryQueue()

    if (!pgQueueEnable || !slowRetryQueue) {
      return
    }

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

    const res = await Queue.getInstance().send({
      name: constructor.getSlowRetryQueueName(),
      data: {
        region,
        ...this.payload,
        $version: constructor.version,
      },
      options: {
        retryBackoff: true,
        startAfter: 60 * 60 * 30, // 30 mins
        ...sendOptions,
        ...slowRetryQueue,
      },
    })

    timer({
      name: constructor.getSlowRetryQueueName(),
    })

    QueueJobScheduled.inc({
      name: constructor.getSlowRetryQueueName(),
    })

    return res
  }
}

1	import { Queue } from './queue'	1✔
2	import PgBoss, { BatchWorkOptions, Job, SendOptions, WorkOptions } 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	export interface SlowRetryQueueOptions {	1✔
20	retryLimit: number	1✔
21	retryDelay: number	1✔
22	}	1✔
23		1✔
24	const { pgQueueEnable, region, isMultitenant } = getConfig()	1✔
25		1✔
26	export type StaticThis<T extends Event<any>> = BaseEventConstructor<T>	1✔
27		1✔
28	interface BaseEventConstructor<Base extends Event<any>> {	1✔
29	version: string	1✔
30		1✔
31	new (...args: any): Base	1✔
32		1✔
33	send(	1✔
34	this: StaticThis<Base>,	1✔
35	payload: Omit<Base['payload'], '$version'>	1✔
36	): Promise<string \| void \| null>	1✔
37		1✔
38	eventName(): string	1✔
39	getWorkerOptions(): WorkOptions \| BatchWorkOptions	1✔
40	}	1✔
41		1✔
42	/**	1✔
43	* Base class for all events that are sent to the queue	1✔
44	*/	1✔
45	export class Event<T extends Omit<BasePayload, '$version'>> {	1✔
46	public static readonly version: string = 'v1'	1✔
47	protected static queueName = ''	1✔
48	protected static allowSync = true	1✔
49		1✔
50	constructor(public readonly payload: T & BasePayload) {}	1✔
51		1✔
52	static eventName() {	1✔
53	return this.name	×
54	}	×
55		1✔
56	static getQueueName() {	1✔
57	if (!this.queueName) {	10,042!
58	throw new Error(`Queue name not set on ${this.constructor.name}`)	×
59	}	×
60		10,042✔
61	return this.queueName	10,042✔
62	}	10,042✔
63		1✔
64	static getQueueOptions<T extends Event<any>>(payload: T['payload']): SendOptions \| undefined {	1✔
65	return undefined	5✔
66	}	5✔
67		1✔
68	static getWorkerOptions(): WorkOptions \| BatchWorkOptions {	1✔
69	return {}	×
70	}	×
71		1✔
72	static withSlowRetryQueue(): undefined \| SlowRetryQueueOptions {	1✔
73	return undefined	×
74	}	×
75		1✔
76	static getSlowRetryQueueName() {	1✔
77	if (!this.queueName) {	×
78	throw new Error(`Queue name not set on ${this.constructor.name}`)	×
79	}	×
80		×
81	return this.queueName + '-slow'	×
82	}	×
83		1✔
84	static onClose() {	1✔
85	// no-op	×
86	}	×
87		1✔
88	static onStart() {	1✔
89	// no-op	×
90	}	×
91		1✔
92	static batchSend<T extends Event<any>[]>(messages: T) {	1✔
93	if (!pgQueueEnable) {	1!
NEW 94	if (this.allowSync) {	×
NEW 95	return Promise.all(messages.map((message) => message.send()))	✔
NEW 96	} else {	×
NEW 97	logger.warn('[Queue] skipped sending batch messages', {	×
NEW 98	type: 'queue',	×
NEW 99	eventType: this.eventName(),	×
NEW 100	})	×
NEW 101	return	×
NEW 102	}	×
UNCOV 103	}	×
104		1✔
105	return Queue.getInstance().insert(	1✔
106	messages.map((message) => {	1✔
107	const sendOptions = (this.getQueueOptions(message.payload) as PgBoss.JobInsert) \|\| {}	1!
108	if (!message.payload.$version) {	1✔
109	;(message.payload as (typeof message)['payload']).$version = this.version	1✔
110	}	1✔
111		1✔
112	if (message.payload.scheduleAt) {	1!
113	sendOptions.startAfter = new Date(message.payload.scheduleAt)	×
114	}	×
115		1✔
116	return {	1✔
117	...sendOptions,	1✔
118	name: this.getQueueName(),	1✔
119	data: message.payload,	1✔
120	}	1✔
121	})	1✔
122	)	1✔
123	}	1✔
124		1✔
125	static send<T extends Event<any>>(this: StaticThis<T>, payload: Omit<T['payload'], '$version'>) {	1✔
126	if (!payload.$version) {	10,042✔
127	;(payload as T['payload']).$version = this.version	10,042✔
128	}	10,042✔
129	const that = new this(payload)	10,042✔
130	return that.send()	10,042✔
131	}	10,042✔
132		1✔
133	static sendSlowRetryQueue<T extends Event<any>>(	1✔
134	this: StaticThis<T>,	×
135	payload: Omit<T['payload'], '$version'>	×
136	) {	×
137	if (!payload.$version) {	×
138	;(payload as T['payload']).$version = this.version	×
139	}	×
140	const that = new this(payload)	×
141	return that.sendSlowRetryQueue()	×
142	}	×
143		1✔
144	static handle(job: Job<Event<any>['payload']> \| Job<Event<any>['payload']>[]) {	1✔
145	throw new Error('not implemented')	×
146	}	×
147		1✔
148	static async shouldSend(payload: any) {	1✔
149	if (isMultitenant) {	14!
150	// Do not send an event if disabled for this specific tenant	×
151	const tenant = await getTenantConfig(payload.tenant.ref)	×
152	const disabledEvents = tenant.disableEvents \|\| []	×
153	if (disabledEvents.includes(this.eventName())) {	×
154	return false	×
155	}	×
156	}	×
157	return true	14✔
158	}	14✔
159		1✔
160	async send(): Promise<string \| void \| null> {	1✔
161	const constructor = this.constructor as typeof Event	10,042✔
162		10,042✔
163	const shouldSend = await constructor.shouldSend(this.payload)	10,042✔
164		10,042✔
165	if (!shouldSend) {	10,042!
166	return	×
167	}	×
168		10,042✔
169	if (!pgQueueEnable) {	10,042✔
170	if (constructor.allowSync) {	10,042✔
171	return constructor.handle({	10,042✔
172	id: '__sync',	10,042✔
173	name: constructor.getQueueName(),	10,042✔
174	data: {	10,042✔
175	region,	10,042✔
176	...this.payload,	10,042✔
177	$version: constructor.version,	10,042✔
178	},	10,042✔
179	})	10,042✔
180	} else {	10,042!
NEW 181	logger.warn('[Queue] skipped sending message', {	×
NEW 182	type: 'queue',	×
NEW 183	eventType: constructor.eventName(),	×
NEW 184	})	×
NEW 185	return	×
NEW 186	}	×
187	}	10,042✔
188		×
189	const timer = QueueJobSchedulingTime.startTimer()	×
190	let sendOptions = constructor.getQueueOptions(this.payload)	×
191		×
192	if (this.payload.scheduleAt) {	×
193	if (!sendOptions) {	×
194	sendOptions = {}	×
195	}	×
196	sendOptions.startAfter = new Date(this.payload.scheduleAt)	×
197	}	×
198		×
199	try {	×
200	const res = await Queue.getInstance().send({	×
201	name: constructor.getQueueName(),	×
202	data: {	×
203	region,	×
204	...this.payload,	×
205	$version: constructor.version,	×
206	},	×
207	options: sendOptions,	×
208	})	×
209		×
210	QueueJobScheduled.inc({	×
211	name: constructor.getQueueName(),	×
212	})	×
213		×
214	return res	×
215	} catch (e) {	×
216	// If we can't queue the message for some reason,	×
217	// we run its handler right away.	×
218	// This might create some latency with the benefit of being more fault-tolerant	×
219	logSchema.warning(	×
220	logger,	×
221	`[Queue Sender] Error while sending job to queue, sending synchronously`,	×
222	{	×
223	type: 'queue',	×
224	error: e,	×
225	metadata: JSON.stringify(this.payload),	×
226	}	×
227	)	×
228	return constructor.handle({	×
229	id: '__sync',	×
230	name: constructor.getQueueName(),	×
231	data: {	×
232	region,	×
233	...this.payload,	×
234	$version: constructor.version,	×
235	},	×
236	})	×
237	} finally {	×
238	timer({	×
239	name: constructor.getQueueName(),	×
240	})	×
241	}	×
242	}	10,042✔
243		1✔
244	async sendSlowRetryQueue() {	1✔
245	const constructor = this.constructor as typeof Event	×
246	const slowRetryQueue = constructor.withSlowRetryQueue()	×
247		×
248	if (!pgQueueEnable \|\| !slowRetryQueue) {	×
249	return	×
250	}	×
251		×
252	const timer = QueueJobSchedulingTime.startTimer()	×
253	const sendOptions = constructor.getQueueOptions(this.payload) \|\| {}	×
254		×
255	const res = await Queue.getInstance().send({	×
256	name: constructor.getSlowRetryQueueName(),	×
257	data: {	×
258	region,	×
259	...this.payload,	×
260	$version: constructor.version,	×
261	},	×
262	options: {	×
263	retryBackoff: true,	×
264	startAfter: 60 * 60 * 30, // 30 mins	×
265	...sendOptions,	×
266	...slowRetryQueue,	×
267	},	×
268	})	×
269		×
270	timer({	×
271	name: constructor.getSlowRetryQueueName(),	×
272	})	×
273		×
274	QueueJobScheduled.inc({	×
275	name: constructor.getSlowRetryQueueName(),	×
276	})	×
277		×
278	return res	×
279	}	×
280	}	1✔

supabase / storage / 14354470908

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