• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

supabase / storage / 23638991080

27 Mar 2026 09:06AM UTC coverage: 79.332% (+0.001%) from 79.331%
23638991080

push

github

web-flow
feat: support for watt process runner (#917)

5047 of 6664 branches covered (75.74%)

Branch coverage included in aggregate %.

14 of 18 new or added lines in 5 files covered. (77.78%)

3 existing lines in 2 files now uncovered.

28742 of 35928 relevant lines covered (80.0%)

587.97 hits per line

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

10.55
/src/internal/queue/queue.ts
1
import { ERRORS } from '@internal/errors'
2✔
2
import { QueueDB } from '@internal/queue/database'
2✔
3
import { Semaphore } from '@shopify/semaphore'
2✔
4
import PgBoss, { Db, Job, JobWithMetadata } from 'pg-boss'
2✔
5
import { getConfig } from '../../config'
2✔
6
import { logger, logSchema } from '../monitoring'
2✔
7
import { queueJobCompleted, queueJobError, queueJobRetryFailed } from '../monitoring/metrics'
2✔
8
import { Event } from './event'
2✔
9

2✔
10
type SubclassOfBaseClass = (new (
2✔
11
  payload: any
2✔
12
) => Event<any>) & {
2✔
13
  [K in keyof typeof Event]: (typeof Event)[K]
2✔
14
}
2✔
15

2✔
16
export const PG_BOSS_SCHEMA = 'pgboss_v10'
2✔
17

2✔
18
export abstract class Queue {
2✔
19
  protected static events: SubclassOfBaseClass[] = []
2✔
20
  private static pgBoss?: PgBoss
2✔
21
  private static pgBossDb?: PgBoss.Db
2✔
22

2✔
23
  static createPgBoss(opts: { db: Db; enableWorkers: boolean }) {
2✔
24
    const {
×
25
      isMultitenant,
×
26
      databaseURL,
×
27
      multitenantDatabasePoolUrl,
×
28
      multitenantDatabaseUrl,
×
29
      pgQueueConnectionURL,
×
30
      pgQueueArchiveCompletedAfterSeconds,
×
31
      pgQueueDeleteAfterDays,
×
32
      pgQueueDeleteAfterHours,
×
33
      pgQueueRetentionDays,
×
34
    } = getConfig()
×
35

×
36
    let url = pgQueueConnectionURL ?? databaseURL
×
37
    let migrate = true
×
38

×
39
    if (isMultitenant && !pgQueueConnectionURL) {
×
40
      if (!multitenantDatabaseUrl) {
×
41
        throw new Error(
×
42
          'running storage in multi-tenant but DB_MULTITENANT_DATABASE_URL is not set'
×
43
        )
×
44
      }
×
45
      url = multitenantDatabasePoolUrl || multitenantDatabaseUrl
×
46

×
47
      if (multitenantDatabasePoolUrl) {
×
48
        migrate = false
×
49
      }
×
50
    }
×
51

×
52
    return new PgBoss({
×
53
      connectionString: url,
×
54
      migrate,
×
55
      db: opts.db,
×
56
      schema: PG_BOSS_SCHEMA,
×
57
      ...(pgQueueDeleteAfterHours
×
58
        ? { deleteAfterHours: pgQueueDeleteAfterHours }
×
59
        : { deleteAfterDays: pgQueueDeleteAfterDays }),
×
60
      archiveCompletedAfterSeconds: pgQueueArchiveCompletedAfterSeconds,
×
61
      retentionDays: pgQueueRetentionDays,
×
62
      retryBackoff: true,
×
63
      retryLimit: 20,
×
64
      expireInHours: 23,
×
65
      maintenanceIntervalSeconds: 60 * 5,
×
66
      schedule: opts.enableWorkers,
×
67
      supervise: opts.enableWorkers,
×
68
    })
×
69
  }
×
70

2✔
71
  static async start(opts: {
2✔
72
    signal?: AbortSignal
×
73
    onMessage?: (job: Job) => void
×
74
    registerWorkers?: () => void
×
75
  }) {
×
76
    if (Queue.pgBoss) {
×
77
      return Queue.pgBoss
×
78
    }
×
79

×
80
    if (opts.signal?.aborted) {
×
81
      throw ERRORS.Aborted('Cannot start queue with aborted signal')
×
82
    }
×
83

×
84
    const {
×
85
      isMultitenant,
×
86
      databaseURL,
×
87
      multitenantDatabaseUrl,
×
88
      multitenantDatabasePoolUrl,
×
89
      pgQueueConnectionURL,
×
90
      pgQueueEnableWorkers,
×
91
      pgQueueReadWriteTimeout,
×
92
      pgQueueConcurrentTasksPerQueue,
×
93
      pgQueueMaxConnections,
×
94
    } = getConfig()
×
95

×
96
    let url = pgQueueConnectionURL || databaseURL
×
97

×
98
    if (isMultitenant && !pgQueueConnectionURL) {
×
99
      if (!multitenantDatabaseUrl && !multitenantDatabasePoolUrl) {
×
100
        throw new Error(
×
101
          'running storage in multi-tenant but DB_MULTITENANT_DATABASE_URL is not set'
×
102
        )
×
103
      }
×
104
      url = (multitenantDatabasePoolUrl || multitenantDatabaseUrl) as string
×
105
    }
×
106

×
107
    Queue.pgBossDb = new QueueDB({
×
108
      min: 0,
×
109
      max: pgQueueMaxConnections,
×
110
      connectionString: url,
×
111
      statement_timeout: pgQueueReadWriteTimeout > 0 ? pgQueueReadWriteTimeout : undefined,
×
112
    })
×
113

×
114
    Queue.pgBoss = this.createPgBoss({
×
115
      db: Queue.pgBossDb,
×
116
      enableWorkers: pgQueueEnableWorkers !== false,
×
117
    })
×
118

×
119
    Queue.pgBoss.on('error', (error) => {
×
120
      logSchema.error(logger, '[Queue] error', {
×
121
        type: 'queue',
×
122
        error,
×
123
      })
×
124
    })
×
125

×
126
    await Queue.pgBoss.start()
×
127

×
128
    if (opts.registerWorkers && pgQueueEnableWorkers) {
×
129
      opts.registerWorkers()
×
130
    }
×
131

×
132
    await Queue.callStart()
×
133
    await Queue.startWorkers({
×
134
      maxConcurrentTasks: pgQueueConcurrentTasksPerQueue,
×
135
      onMessage: opts.onMessage,
×
136
      signal: opts.signal,
×
137
    })
×
138

×
139
    if (opts.signal) {
×
140
      opts.signal.addEventListener(
×
141
        'abort',
×
142
        async () => {
×
143
          logSchema.info(logger, '[Queue] Stopping', {
×
144
            type: 'queue',
×
145
          })
×
146
          return Queue.stop()
×
147
            .then(async () => {
×
148
              logSchema.info(logger, '[Queue] Exited', {
×
149
                type: 'queue',
×
150
              })
×
151
            })
×
152
            .catch((e) => {
×
153
              logSchema.error(logger, '[Queue] Error while stopping queue', {
×
154
                error: e,
×
155
                type: 'queue',
×
156
              })
×
157
            })
×
158
            .finally(async () => {
×
159
              await Queue.callClose().catch(() => {
×
160
                // no-op
×
161
              })
×
162
            })
×
163
        },
×
164
        { once: true }
×
165
      )
×
166
    }
×
167

×
168
    return Queue.pgBoss
×
169
  }
×
170

2✔
171
  static getInstance() {
2✔
172
    if (!this.pgBoss) {
×
173
      throw new Error('pg boss not initialised')
×
174
    }
×
175

×
176
    return this.pgBoss
×
177
  }
×
178

2✔
179
  static getDb() {
2✔
180
    if (!this.pgBossDb) {
×
181
      throw new Error('pg boss not initialised')
×
182
    }
×
183

×
184
    return this.pgBossDb
×
185
  }
×
186

2✔
187
  static register<T extends SubclassOfBaseClass>(event: T) {
2✔
188
    Queue.events.push(event)
×
189
  }
×
190

2✔
191
  static async stop() {
2✔
192
    if (!this.pgBoss) {
×
193
      return
×
194
    }
×
195

×
196
    const boss = this.pgBoss
×
197
    const { isProduction } = getConfig()
×
198

×
199
    await boss.stop({
×
200
      timeout: 20 * 1000,
×
201
      graceful: isProduction,
×
202
      wait: true,
×
203
    })
×
204

×
NEW
205
    await this.callClose()
×
206

×
207
    Queue.pgBoss = undefined
×
208
  }
×
209

2✔
210
  protected static async startWorkers(opts: {
2✔
211
    maxConcurrentTasks: number
×
212
    signal?: AbortSignal
×
213
    onMessage?: (job: Job) => void
×
214
  }) {
×
215
    for (const event of Queue.events) {
×
216
      await Queue.registerTask(event, opts.maxConcurrentTasks, opts.onMessage, opts.signal)
×
217
    }
×
218
  }
×
219

2✔
220
  protected static callStart() {
2✔
221
    const events = Queue.events.map((event) => {
×
222
      return event.onStart()
×
223
    })
×
224

×
225
    return Promise.all(events)
×
226
  }
×
227

2✔
228
  protected static callClose() {
2✔
229
    const events = Queue.events.map((event) => {
×
230
      return event.onClose()
×
231
    })
×
232

×
233
    return Promise.all(events)
×
234
  }
×
235

2✔
236
  protected static async registerTask(
2✔
237
    event: SubclassOfBaseClass,
×
238
    maxConcurrentTasks: number,
×
239
    onMessage?: (job: Job) => void,
×
240
    signal?: AbortSignal
×
241
  ) {
×
242
    const queueName = event.getQueueName()
×
243
    const deadLetterName = event.deadLetterQueueName()
×
244

×
245
    const concurrentTaskCount = event.getWorkerOptions().concurrentTaskCount || maxConcurrentTasks
×
246
    try {
×
247
      // Create dead-letter queue and the normal queue
×
248
      const queueOptions = {
×
249
        policy: 'standard',
×
250
        ...event.getQueueOptions(),
×
251
      } as const
×
252

×
253
      // dead-letter
×
254
      await this.pgBoss?.createQueue(deadLetterName, {
×
255
        ...queueOptions,
×
256
        name: deadLetterName,
×
257
        retentionDays: 30,
×
258
        retryBackoff: true,
×
259
      })
×
260

×
261
      // // normal queue
×
262
      await this.pgBoss?.createQueue(queueName, {
×
263
        ...queueOptions,
×
264
        name: queueName,
×
265
        deadLetter: deadLetterName,
×
266
      })
×
267
    } catch {
×
268
      // no-op
×
269
    }
×
270

×
271
    return this.pollQueue(event, {
×
272
      concurrentTaskCount,
×
273
      onMessage,
×
274
      signal,
×
275
    })
×
276
  }
×
277

2✔
278
  protected static pollQueue(
2✔
279
    event: SubclassOfBaseClass,
×
280
    queueOpts: {
×
281
      concurrentTaskCount: number
×
282
      onMessage?: (job: Job) => void
×
283
      signal?: AbortSignal
×
284
    }
×
285
  ) {
×
286
    const semaphore = new Semaphore(queueOpts.concurrentTaskCount)
×
287
    const pollingInterval = (event.getWorkerOptions().pollingIntervalSeconds || 5) * 1000
×
288
    const batchSize =
×
289
      event.getWorkerOptions().batchSize ||
×
290
      queueOpts.concurrentTaskCount + Math.max(1, Math.floor(queueOpts.concurrentTaskCount * 1.2))
×
291

×
292
    logSchema.info(logger, `[Queue] Polling queue ${event.getQueueName()}`, {
×
293
      type: 'queue',
×
294
      metadata: JSON.stringify({
×
295
        queueName: event.getQueueName(),
×
296
        batchSize,
×
297
        pollingInterval,
×
298
      }),
×
299
    })
×
300

×
301
    let started = false
×
302
    let jobFetched = 0
×
303

×
304
    const interval = setInterval(async () => {
×
305
      if (started) {
×
306
        return
×
307
      }
×
308

×
309
      try {
×
310
        started = true
×
311
        const defaultFetch = {
×
312
          includeMetadata: true,
×
313
          batchSize,
×
314
        }
×
315

×
316
        const currentBatch = defaultFetch.batchSize - jobFetched
×
317

×
318
        if (currentBatch <= 0) {
×
319
          return
×
320
        }
×
321

×
322
        const jobs = await this.pgBoss?.fetch(event.getQueueName(), {
×
323
          ...event.getWorkerOptions(),
×
324
          ...defaultFetch,
×
325
          batchSize: currentBatch,
×
326
        })
×
327

×
328
        jobFetched += jobs?.length || 0
×
329

×
330
        if (jobFetched < defaultFetch.batchSize) {
×
331
          started = false
×
332
        }
×
333

×
334
        if (queueOpts.signal?.aborted) {
×
335
          started = false
×
336
          return
×
337
        }
×
338

×
339
        if (!jobs || (jobs && jobs.length === 0)) {
×
340
          started = false
×
341
          return
×
342
        }
×
343

×
344
        await Promise.allSettled(
×
345
          jobs.map(async (job) => {
×
346
            const lock = await semaphore.acquire()
×
347
            try {
×
348
              queueOpts.onMessage?.(job as Job)
×
349

×
350
              await event.handle(job, { signal: queueOpts.signal })
×
351

×
352
              await this.pgBoss?.complete(event.getQueueName(), job.id)
×
353
              queueJobCompleted.add(1, {
×
354
                name: event.getQueueName(),
×
355
              })
×
356
            } catch (e) {
×
357
              queueJobRetryFailed.add(1, {
×
358
                name: event.getQueueName(),
×
359
              })
×
360

×
361
              await this.pgBoss?.fail(event.getQueueName(), job.id)
×
362

×
363
              try {
×
364
                const dbJob: JobWithMetadata | null =
×
365
                  (job as JobWithMetadata).priority !== undefined
×
366
                    ? (job as JobWithMetadata)
×
367
                    : await Queue.getInstance().getJobById(event.getQueueName(), job.id)
×
368

×
369
                if (!dbJob) {
×
370
                  return
×
371
                }
×
372
                if (dbJob.retryCount >= dbJob.retryLimit) {
×
373
                  queueJobError.add(1, {
×
374
                    name: event.getQueueName(),
×
375
                  })
×
376
                }
×
377
              } catch (e) {
×
378
                logSchema.error(logger, `[Queue Handler] fetching job ${event.name}`, {
×
379
                  type: 'queue-task',
×
380
                  error: e,
×
381
                  metadata: JSON.stringify(job),
×
382
                })
×
383
              }
×
384

×
385
              logSchema.error(logger, `[Queue Handler] Error while processing job ${event.name}`, {
×
386
                type: 'queue-task',
×
387
                error: e,
×
388
                metadata: JSON.stringify(job),
×
389
              })
×
390

×
391
              throw e
×
392
            } finally {
×
393
              jobFetched = Math.max(0, jobFetched - 1)
×
394
              await lock.release()
×
395
            }
×
396
          })
×
397
        )
×
398
      } catch (e) {
×
399
        logSchema.error(logger, `[Queue] Error while polling queue ${event.name}`, {
×
400
          type: 'queue',
×
401
          error: e,
×
402
          metadata: JSON.stringify({
×
403
            queueName: event.getQueueName(),
×
404
            batchSize,
×
405
            pollingInterval,
×
406
          }),
×
407
        })
×
408
      } finally {
×
409
        started = false
×
410
      }
×
411
    }, pollingInterval)
×
412

×
413
    queueOpts.signal?.addEventListener('abort', () => {
×
414
      clearInterval(interval)
×
415
    })
×
416
  }
×
417
}
2✔
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2026 Coveralls, Inc