fix: ensure queue exists added

wavezync · Sep 10, 2024 · cfb929d · cfb929d
1 parent a115473
commit cfb929d
Show file tree

Hide file tree

Showing 3 changed files with 26 additions and 5 deletions.
diff --git a/lib/handler-scanner.service.ts b/lib/handler-scanner.service.ts
@@ -52,14 +52,15 @@ export class HandlerScannerService {
         CRON_EXPRESSION,
         methodRef,
       );
-      const cronOptions = this.reflector.get<any>(CRON_OPTIONS, methodRef);
+      const cronOptions = this.reflector.get<PgBoss.ScheduleOptions>(
+        CRON_OPTIONS,
+        methodRef,
+      );
 
       if (jobName) {
         const boundHandler: WorkWithMetadataHandler<any> = async (job) => {
-          const jobData = {
-            ...job,
-          };
-          await methodRef.call(instance, jobData);
+          const extractedJob = this.normalizeJob(job);
+          await methodRef.call(instance, extractedJob);
         };
         try {
           if (cronExpression) {
@@ -85,4 +86,11 @@ export class HandlerScannerService {
       }
     }
   }
+
+  private normalizeJob(job: any) {
+    if (typeof job === "object" && "0" in job) {
+      return job[0];
+    }
+    return job;
+  }
 }
diff --git a/lib/pgboss.service.ts b/lib/pgboss.service.ts
@@ -20,6 +20,7 @@ export class PgBossService {
     data: TData,
     options?: PgBoss.SendOptions,
   ) {
+    await this.ensureQueueExists(name);
     await this.pgBoss.send(name, data, options);
   }
 
@@ -29,6 +30,7 @@ export class PgBossService {
     data?: TData,
     options?: PgBoss.ScheduleOptions,
   ) {
+    await this.ensureQueueExists(name);
     await this.pgBoss.schedule(name, cron, data ?? {}, options ?? {});
   }
 
@@ -39,6 +41,7 @@ export class PgBossService {
     data?: TData,
     options?: PgBoss.ScheduleOptions,
   ) {
+    await this.ensureQueueExists(name);
     await this.pgBoss.schedule(name, cron, data ?? {}, options ?? {});
     await this.pgBoss.work<TData>(
       name,
@@ -52,6 +55,7 @@ export class PgBossService {
     handler: WorkWithMetadataHandler<TData>,
     options?: PgBoss.WorkOptions,
   ) {
+    await this.ensureQueueExists(name);
     await this.pgBoss.work<TData>(
       name,
       { ...options, includeMetadata: true },
@@ -71,4 +75,11 @@ export class PgBossService {
 
     return transformedOptions;
   }
+
+  async ensureQueueExists(queueName: string) {
+    const currentQueue = await this.pgBoss.getQueue(queueName);
+    if (!currentQueue) {
+      await this.pgBoss.createQueue(queueName);
+    }
+  }
 }
diff --git a/test/pgboss.service.spec.ts b/test/pgboss.service.spec.ts
@@ -13,6 +13,8 @@ describe("PgBossService", () => {
       work: jest.fn(),
       send: jest.fn(),
       schedule: jest.fn(),
+      createQueue: jest.fn(),
+      getQueue: jest.fn(),
     } as any;
 
     const module: TestingModule = await Test.createTestingModule({