30832958929

Committed 03 Aug 2026 04:36PM UTC coverage: 53.989% (-26.5%) from 80.494%

Build # 30832958929

Build Type

Pull #1292

github

Committed by

web-flow

Commit Message

Merge 477971401 into 42e86a3a7

Pull Request Pull Request #1292: feat: new queing system

Coverage Stats

3711 of 7412 branches covered (50.07%)

Branch coverage included in aggregate %.

140 of 337 new or added lines in 38 files covered. (41.54%)

3365 existing lines in 150 files now uncovered.

6941 of 12318 relevant lines covered (56.35%)

77.58 hits per line

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

11.65

/src/internal/queue/middleware.ts

import { getTenantConfig } from '@internal/database'
import { logger, logSchema } from '@internal/monitoring'
import { queueJobScheduled, queueJobSchedulingTime } from '@internal/monitoring/metrics'
import type {
  AnyMessageClass,
  ProduceCall,
  ProduceMessage,
  WaveMiddleware,
} from '@supabase-labs/wave-core'
import { MESSAGE_TYPE_HEADER } from '@supabase-labs/wave-core'
import { getConfig } from '../../config'
import { SYSTEM_TENANT_REF } from './constants'
import { isBasePayload, type QueueEventOptions } from './events'

type AnyWaveMiddleware = WaveMiddleware<unknown>

const typeOf = (m: ProduceMessage<unknown>): string | undefined => m.headers?.[MESSAGE_TYPE_HEADER]

/** Whether a bound message class carries the storage queue-middleware statics — declared by
 * `storageEvent` (events/base.ts), not part of wave's `AnyMessageClass` contract. A runtime
 * check, so a class from outside `storageEvent` simply gets no gating config rather than
 * being blindly trusted to have it. */
const isQueueEventClass = (cls: AnyMessageClass): cls is AnyMessageClass & QueueEventOptions =>
  typeof (cls as Partial<QueueEventOptions>).eventType === 'string'

/**
 * Resolve the message class actually bound to this produce call's topic, via `call.getTopic` —
 * no registry threaded in as a constructor argument. `allowSync`/`disableKeys`/`eventType` are
 * read directly off the class (see `storageEvent` in events/base.ts): the class IS the
 * single source of truth, not a parallel metadata object. Polymorphic topics (more than one
 * bound class) disambiguate by the wire `message-type` header; the single-class case (every
 * storage topic today) needs no disambiguation.
 */
const classOf = (
  call: ProduceCall,
  messageType: string | undefined
): QueueEventOptions | undefined => {
  const classes = call.getTopic(call.topic)?.classes?.filter(isQueueEventClass)

  if (classes === undefined || classes.length === 0) return undefined
  if (classes.length === 1) return classes[0]
  return classes.find((cls) => cls.eventType === messageType) ?? classes[0]
}

const tenantRefOf = (m: ProduceMessage<unknown>): string =>
  (isBasePayload(m.data) ? m.data.tenant.ref : undefined) || SYSTEM_TENANT_REF

/**
 * Per-tenant event disabling (v1 `shouldSend`): in multitenant mode, a message whose
 * disable keys intersect the tenant's `disableEvents` is silently dropped — the produce call
 * still resolves, exactly as v1's gated `send()` resolved to nothing. Applies on the real
 * wave AND the sync wave (v1 gates before the env-disabled branch).
 */
export function tenantDisableEvents(): AnyWaveMiddleware {
  const { isMultitenant } = getConfig()
  return {
    produce: (next) => async (call) => {
      if (!isMultitenant) return next(call)

      const kept: ProduceMessage<unknown>[] = []
      for (const m of call.messages) {
        const cls = classOf(call, typeOf(m))
        // System-produced messages have no tenant config row to consult — never gated.
        if (
          !isBasePayload(m.data) ||
          cls === undefined ||
          m.data.tenant.ref === SYSTEM_TENANT_REF
        ) {
          kept.push(m)
          continue
        }
        const disabled = (await getTenantConfig(m.data.tenant.ref)).disableEvents || []
        const keys = cls.disableKeys?.(m.data) ?? [cls.eventType]
        if (!keys.some((key) => disabled.includes(key))) kept.push(m)
      }

      if (kept.length === 0) return
      await next(kept.length === call.messages.length ? call : { ...call, messages: kept })
    },
  }
}

/** v1's scheduling metrics, verbatim names: `queue_job_scheduled` per landed message and
 * `queue_job_scheduled_time_seconds` around the whole produce (fallback included, as v1's
 * `finally` measured). */
export function schedulingMetrics(): AnyWaveMiddleware {
  return {
    produce: (next) => async (call) => {
      const startTime = performance.now()
      try {
        await next(call)
        queueJobScheduled.add(call.messages.length, { name: call.topic })
      } finally {
        const duration = (performance.now() - startTime) / 1000
        queueJobSchedulingTime.record(duration, { name: call.topic })
      }
    },
  }
}

/**
 * v1's fault-tolerance seam: a failed enqueue (timeout included) degrades to running the
 * handler synchronously via the sync wave — which dispatches on producer-only nodes too and
 * tolerates nested sends. Faithful to v1 in both directions: only single-message calls fall
 * back (v1's `batchSend` had no fallback — batch errors propagate raw), and `allowSync: false`
 * events rethrow instead. A fallen-back handler may ALSO run later from the queue (the append
 * outcome is unknown on timeout) and must tolerate duplicates — same contract as v1.
 */
export function syncFallback(): AnyWaveMiddleware {
  return {
    produce: (next) => async (call) => {
      try {
        await next(call)
      } catch (e) {
        const single = call.messages.length === 1 ? call.messages[0] : undefined
        const cls = single !== undefined ? classOf(call, typeOf(single)) : undefined
        if (single === undefined || cls === undefined) throw e

        logSchema.warning(
          logger,
          `[Queue Sender] Error while sending job to queue, sending synchronously`,
          {
            type: 'queue',
            project: tenantRefOf(single),
            error: e,
            metadata: JSON.stringify(single.data),
            sbReqId: isBasePayload(single.data) ? single.data.sbReqId : undefined,
          }
        )

        if (!(cls.allowSync ?? true)) throw e

        await call.invoke!(call.topic, single.data, { key: single.key, headers: single.headers })
      }
    },
  }
}

/**
 * The env-disabled posture for events that must never run in-process (v1 `allowSync: false`
 * under `!pgQueueEnable`): warn and skip, never throw. Composed into the SYNC wave only —
 * on the real wave the same events rethrow from the fallback instead.
 */
export function syncModeGuard(): AnyWaveMiddleware {
  return {
    produce: (next) => async (call) => {
      const kept = call.messages.filter((m) => {
        const cls = classOf(call, typeOf(m))
        if (cls === undefined || (cls.allowSync ?? true)) return true
        logger.warn({ type: 'queue', eventType: cls.eventType }, '[Queue] skipped sending message')
        return false
      })
      if (kept.length === 0) return
      await next(kept.length === call.messages.length ? call : { ...call, messages: kept })
    },
  }
}

1	import { getTenantConfig } from '@internal/database'
2	import { logger, logSchema } from '@internal/monitoring'
3	import { queueJobScheduled, queueJobSchedulingTime } from '@internal/monitoring/metrics'
4	import type {
5	AnyMessageClass,
6	ProduceCall,
7	ProduceMessage,
8	WaveMiddleware,
9	} from '@supabase-labs/wave-core'
10	import { MESSAGE_TYPE_HEADER } from '@supabase-labs/wave-core'
11	import { getConfig } from '../../config'
12	import { SYSTEM_TENANT_REF } from './constants'
13	import { isBasePayload, type QueueEventOptions } from './events'
14
15	type AnyWaveMiddleware = WaveMiddleware<unknown>
16
17	const typeOf = (m: ProduceMessage<unknown>): string \| undefined => m.headers?.[MESSAGE_TYPE_HEADER]	40✔
18
19	/** Whether a bound message class carries the storage queue-middleware statics — declared by
20	* `storageEvent` (events/base.ts), not part of wave's `AnyMessageClass` contract. A runtime
21	* check, so a class from outside `storageEvent` simply gets no gating config rather than
22	* being blindly trusted to have it. */
23	const isQueueEventClass = (cls: AnyMessageClass): cls is AnyMessageClass & QueueEventOptions =>	40✔
NEW 24	typeof (cls as Partial<QueueEventOptions>).eventType === 'string'	×
25
26	/**
27	* Resolve the message class actually bound to this produce call's topic, via `call.getTopic` —
28	* no registry threaded in as a constructor argument. `allowSync`/`disableKeys`/`eventType` are
29	* read directly off the class (see `storageEvent` in events/base.ts): the class IS the
30	* single source of truth, not a parallel metadata object. Polymorphic topics (more than one
31	* bound class) disambiguate by the wire `message-type` header; the single-class case (every
32	* storage topic today) needs no disambiguation.
33	*/
34	const classOf = (	40✔
35	call: ProduceCall,
36	messageType: string \| undefined
37	): QueueEventOptions \| undefined => {
NEW 38	const classes = call.getTopic(call.topic)?.classes?.filter(isQueueEventClass)	×
39
NEW 40	if (classes === undefined \|\| classes.length === 0) return undefined	×
NEW 41	if (classes.length === 1) return classes[0]	×
NEW 42	return classes.find((cls) => cls.eventType === messageType) ?? classes[0]	×
43	}
44
45	const tenantRefOf = (m: ProduceMessage<unknown>): string =>	40!
46	(isBasePayload(m.data) ? m.data.tenant.ref : undefined) \|\| SYSTEM_TENANT_REF	×
47
48	/**
49	* Per-tenant event disabling (v1 `shouldSend`): in multitenant mode, a message whose
50	* disable keys intersect the tenant's `disableEvents` is silently dropped — the produce call
51	* still resolves, exactly as v1's gated `send()` resolved to nothing. Applies on the real
52	* wave AND the sync wave (v1 gates before the env-disabled branch).
53	*/
54	export function tenantDisableEvents(): AnyWaveMiddleware {
NEW 55	const { isMultitenant } = getConfig()	×
NEW 56	return {	×
NEW 57	produce: (next) => async (call) => {	×
NEW 58	if (!isMultitenant) return next(call)	×
59
NEW 60	const kept: ProduceMessage<unknown>[] = []	×
NEW 61	for (const m of call.messages) {	×
NEW 62	const cls = classOf(call, typeOf(m))	×
63	// System-produced messages have no tenant config row to consult — never gated.
NEW 64	if (	×
65	!isBasePayload(m.data) \|\|	×
66	cls === undefined \|\|
67	m.data.tenant.ref === SYSTEM_TENANT_REF
68	) {
NEW 69	kept.push(m)	×
NEW 70	continue	×
71	}
NEW 72	const disabled = (await getTenantConfig(m.data.tenant.ref)).disableEvents \|\| []	×
NEW 73	const keys = cls.disableKeys?.(m.data) ?? [cls.eventType]	×
NEW 74	if (!keys.some((key) => disabled.includes(key))) kept.push(m)	×
75	}
76
NEW 77	if (kept.length === 0) return	×
NEW 78	await next(kept.length === call.messages.length ? call : { ...call, messages: kept })	×
79	},
80	}
81	}
82
83	/** v1's scheduling metrics, verbatim names: `queue_job_scheduled` per landed message and
84	* `queue_job_scheduled_time_seconds` around the whole produce (fallback included, as v1's
85	* `finally` measured). */
86	export function schedulingMetrics(): AnyWaveMiddleware {
87	return {	2✔
88	produce: (next) => async (call) => {	2✔
89	const startTime = performance.now()	2✔
90	try {	2✔
91	await next(call)	2✔
92	queueJobScheduled.add(call.messages.length, { name: call.topic })	1✔
93	} finally {
94	const duration = (performance.now() - startTime) / 1000	2✔
95	queueJobSchedulingTime.record(duration, { name: call.topic })	2✔
96	}
97	},
98	}
99	}
100
101	/**
102	* v1's fault-tolerance seam: a failed enqueue (timeout included) degrades to running the
103	* handler synchronously via the sync wave — which dispatches on producer-only nodes too and
104	* tolerates nested sends. Faithful to v1 in both directions: only single-message calls fall
105	* back (v1's `batchSend` had no fallback — batch errors propagate raw), and `allowSync: false`
106	* events rethrow instead. A fallen-back handler may ALSO run later from the queue (the append
107	* outcome is unknown on timeout) and must tolerate duplicates — same contract as v1.
108	*/
109	export function syncFallback(): AnyWaveMiddleware {
NEW 110	return {	×
NEW 111	produce: (next) => async (call) => {	×
NEW 112	try {	×
NEW 113	await next(call)	×
114	} catch (e) {
NEW 115	const single = call.messages.length === 1 ? call.messages[0] : undefined	×
NEW 116	const cls = single !== undefined ? classOf(call, typeOf(single)) : undefined	×
NEW 117	if (single === undefined \|\| cls === undefined) throw e	×
118
NEW 119	logSchema.warning(	×
120	logger,
121	`[Queue Sender] Error while sending job to queue, sending synchronously`,
122	{
123	type: 'queue',
124	project: tenantRefOf(single),
125	error: e,
126	metadata: JSON.stringify(single.data),
127	sbReqId: isBasePayload(single.data) ? single.data.sbReqId : undefined,	×
128	}
129	)
130
NEW 131	if (!(cls.allowSync ?? true)) throw e	×
132
NEW 133	await call.invoke!(call.topic, single.data, { key: single.key, headers: single.headers })	×
134	}
135	},
136	}
137	}
138
139	/**
140	* The env-disabled posture for events that must never run in-process (v1 `allowSync: false`
141	* under `!pgQueueEnable`): warn and skip, never throw. Composed into the SYNC wave only —
142	* on the real wave the same events rethrow from the fallback instead.
143	*/
144	export function syncModeGuard(): AnyWaveMiddleware {
NEW 145	return {	×
NEW 146	produce: (next) => async (call) => {	×
NEW 147	const kept = call.messages.filter((m) => {	×
NEW 148	const cls = classOf(call, typeOf(m))	×
NEW 149	if (cls === undefined \|\| (cls.allowSync ?? true)) return true	×
NEW 150	logger.warn({ type: 'queue', eventType: cls.eventType }, '[Queue] skipped sending message')	×
NEW 151	return false	×
152	})
NEW 153	if (kept.length === 0) return	×
NEW 154	await next(kept.length === call.messages.length ? call : { ...call, messages: kept })	×
155	},
156	}
157	}

supabase / storage / 30832958929

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