feat: allow to disable queue workers (#507)

Author: fenos

src/config.ts

@@ -59,6 +59,7 @@ type StorageConfigType = {
   logflareApiKey?: string
   logflareSourceToken?: string
   pgQueueEnable: boolean
+  pgQueueEnableWorkers?: boolean
   pgQueueConnectionURL?: string
   pgQueueDeleteAfterDays?: number
   pgQueueArchiveCompletedAfterSeconds?: number
@@ -303,6 +304,7 @@ export function getConfig(options?: { reload?: boolean }): StorageConfigType {
 
     // Queue
     pgQueueEnable: getOptionalConfigFromEnv('PG_QUEUE_ENABLE', 'ENABLE_QUEUE_EVENTS') === 'true',
+    pgQueueEnableWorkers: getOptionalConfigFromEnv('PG_QUEUE_WORKERS_ENABLE') !== 'false',
     pgQueueConnectionURL: getOptionalConfigFromEnv('PG_QUEUE_CONNECTION_URL'),
     pgQueueDeleteAfterDays: parseInt(
       getOptionalConfigFromEnv('PG_QUEUE_DELETE_AFTER_DAYS') || '2',

src/queue/queue.ts

@@ -27,6 +27,7 @@ export abstract class Queue {
       pgQueueDeleteAfterDays,
       pgQueueArchiveCompletedAfterSeconds,
       pgQueueRetentionDays,
+      pgQueueEnableWorkers,
     } = getConfig()
 
     let url = pgQueueConnectionURL ?? databaseURL
@@ -52,7 +53,9 @@ export abstract class Queue {
       expireInHours: 48,
     })
 
-    registerWorkers()
+    if (pgQueueEnableWorkers) {
+      registerWorkers()
+    }
 
     Queue.pgBoss.on('error', (error) => {
       logSchema.error(logger, '[Queue] error', {

src/worker.ts

@@ -0,0 +1,54 @@
+import { Queue } from './queue'
+import { logger, logSchema } from './monitoring'
+import adminApp from './admin-app'
+import { getConfig } from './config'
+
+export async function main() {
+  const { requestTraceHeader, adminPort, host } = getConfig()
+
+  logger.info('[Queue] Starting Queue Worker')
+  const queue = await Queue.init()
+
+  const server = adminApp({
+    logger,
+    disableRequestLogging: true,
+    requestIdHeader: requestTraceHeader,
+  })
+
+  process.on('SIGTERM', async () => {
+    logger.info('[Worker] Stopping')
+    await server.close()
+    await Queue.stop()
+  })
+
+  await server.listen({ port: adminPort, host })
+
+  return new Promise<void>((resolve, reject) => {
+    queue.on('error', (err) => {
+      logger.info('[Queue] Error', err)
+      reject(err)
+    })
+
+    queue.on('stopped', () => {
+      logger.info('[Queue] Stopping')
+      resolve()
+    })
+  })
+}
+
+process.on('uncaughtException', (e) => {
+  logSchema.error(logger, 'uncaught exception', {
+    type: 'uncaughtException',
+    error: e,
+  })
+  logger.flush()
+  process.exit(1)
+})
+
+main()
+  .then(() => {
+    logger.info('[Queue] Worker Exited Successfully')
+  })
+  .catch(() => {
+    process.exit(1)
+  })