misc: use pg queue for audit logs when enabled

backend/src/queue/queue-service.ts

@@ -1,5 +1,6 @@
 import { Job, JobsOptions, Queue, QueueOptions, RepeatOptions, Worker, WorkerListener } from "bullmq";
 import Redis from "ioredis";
+import PgBoss, { WorkOptions } from "pg-boss";
 
 import { SecretEncryptionAlgo, SecretKeyEncoding } from "@app/db/schemas";
 import { TCreateAuditLogDTO } from "@app/ee/services/audit-log/audit-log-types";
@@ -184,17 +185,31 @@ export type TQueueJobTypes = {
 };
 
 export type TQueueServiceFactory = ReturnType<typeof queueServiceFactory>;
-export const queueServiceFactory = (redisUrl: string) => {
+export const queueServiceFactory = (redisUrl: string, dbConnectionUrl: string) => {
   const connection = new Redis(redisUrl, { maxRetriesPerRequest: null });
   const queueContainer = {} as Record<
     QueueName,
     Queue<TQueueJobTypes[QueueName]["payload"], void, TQueueJobTypes[QueueName]["name"]>
   >;
+
+  const pgBoss = new PgBoss({
+    connectionString: dbConnectionUrl,
+    archiveCompletedAfterSeconds: 30,
+    maintenanceIntervalSeconds: 30,
+    deleteAfterSeconds: 30
+  });
+
+  const queueContainerPg = {} as Record<QueueJobs, boolean>;
+
   const workerContainer = {} as Record<
     QueueName,
     Worker<TQueueJobTypes[QueueName]["payload"], void, TQueueJobTypes[QueueName]["name"]>
   >;
 
+  const initialize = async () => {
+    return pgBoss.start();
+  };
+
   const start = <T extends QueueName>(
     name: T,
     jobFn: (job: Job<TQueueJobTypes[T]["payload"], void, TQueueJobTypes[T]["name"]>, token?: string) => Promise<void>,
@@ -215,6 +230,27 @@ export const queueServiceFactory = (redisUrl: string) => {
     });
   };
 
+  const startPg = async <T extends QueueName>(
+    jobName: QueueJobs,
+    jobsFn: (jobs: PgBoss.Job<TQueueJobTypes[T]["payload"]>[]) => Promise<void>,
+    options: WorkOptions & {
+      workerCount: number;
+    }
+  ) => {
+    if (queueContainerPg[jobName]) {
+      throw new Error(`${jobName} queue is already initialized`);
+    }
+
+    await pgBoss.createQueue(jobName);
+    queueContainerPg[jobName] = true;
+
+    await Promise.all(
+      Array.from({ length: options.workerCount }).map(() =>
+        pgBoss.work<TQueueJobTypes[T]["payload"]>(jobName, options, jobsFn)
+      )
+    );
+  };
+
   const listen = <
     T extends QueueName,
     U extends keyof WorkerListener<TQueueJobTypes[T]["payload"], void, TQueueJobTypes[T]["name"]>
@@ -238,6 +274,18 @@ export const queueServiceFactory = (redisUrl: string) => {
     await q.add(job, data, opts);
   };
 
+  const queuePg = async <T extends QueueName>(
+    job: TQueueJobTypes[T]["name"],
+    data: TQueueJobTypes[T]["payload"],
+    opts?: PgBoss.SendOptions & { jobId?: string }
+  ) => {
+    await pgBoss.send({
+      name: job,
+      data,
+      options: opts
+    });
+  };
+
   const stopRepeatableJob = async <T extends QueueName>(
     name: T,
     job: TQueueJobTypes[T]["name"],
@@ -274,5 +322,17 @@ export const queueServiceFactory = (redisUrl: string) => {
     await Promise.all(Object.values(workerContainer).map((worker) => worker.close()));
   };
 
-  return { start, listen, queue, shutdown, stopRepeatableJob, stopRepeatableJobByJobId, clearQueue, stopJobById };
+  return {
+    initialize,
+    start,
+    listen,
+    queue,
+    shutdown,
+    stopRepeatableJob,
+    stopRepeatableJobByJobId,
+    clearQueue,
+    stopJobById,
+    startPg,
+    queuePg
+  };
 };