16223502981

Committed 11 Jul 2025 03:16PM UTC coverage: 77.08% (-0.9%) from 77.981%

Build # 16223502981

Build Type

push

github

Committed by

web-flow

Commit Message

fix: analytic bucket support

Coverage Stats

1623 of 2339 branches covered (69.39%)

Branch coverage included in aggregate %.

3010 of 3976 new or added lines in 64 files covered. (75.7%)

84 existing lines in 10 files now uncovered.

20321 of 26130 relevant lines covered (77.77%)

102.97 hits per line

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

60.5

/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'
import { ERRORS } from '@internal/errors'

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

  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 invoke(): Promise<string | void | null> {
    const constructor = this.constructor as typeof Event

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

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

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

supabase / storage / 16223502981

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