feat: migrate to pgboss 10

lib/handler-scanner.service.ts

@@ -10,6 +10,7 @@ import {
 import { InstanceWrapper } from "@nestjs/core/injector/instance-wrapper";
 import PgBoss, { WorkWithMetadataHandler } from "pg-boss";
 import { LOGGER } from "./utils/consts";
+import { normalizeJob } from "./utils/helpers";
 
 @Injectable()
 export class HandlerScannerService {
@@ -52,14 +53,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 = normalizeJob(job);
+          await methodRef.call(instance, extractedJob);
         };
         try {
           if (cronExpression) {
@@ -75,7 +77,7 @@ export class HandlerScannerService {
             await this.pgBossService.registerJob(
               jobName,
               boundHandler,
-              jobOptions as PgBoss.BatchWorkOptions,
+              jobOptions,
             );
             this.logger.log(`Registered job: ${jobName}`);
           }

lib/pgboss.service.ts

@@ -2,6 +2,7 @@ import { Injectable } from "@nestjs/common";
 import PgBoss, { WorkWithMetadataHandler } from "pg-boss";
 import { Inject } from "@nestjs/common";
 import { PGBOSS_TOKEN } from "./utils/consts";
+import { transformOptions } from "./utils/helpers";
 
 @Injectable()
 export class PgBossService {
@@ -20,6 +21,7 @@ export class PgBossService {
     data: TData,
     options?: PgBoss.SendOptions,
   ) {
+    await this.ensureQueueExists(name);
     await this.pgBoss.send(name, data, options);
   }
 
@@ -29,6 +31,7 @@ export class PgBossService {
     data?: TData,
     options?: PgBoss.ScheduleOptions,
   ) {
+    await this.ensureQueueExists(name);
     await this.pgBoss.schedule(name, cron, data ?? {}, options ?? {});
   }
 
@@ -39,23 +42,32 @@ 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,
-      { ...options, includeMetadata: true },
+      { ...transformOptions(options), includeMetadata: true },
       handler,
     );
   }
 
   async registerJob<TData extends object>(
     name: string,
     handler: WorkWithMetadataHandler<TData>,
-    options?: PgBoss.BatchWorkOptions,
+    options?: PgBoss.WorkOptions,
   ) {
+    await this.ensureQueueExists(name);
     await this.pgBoss.work<TData>(
       name,
       { ...options, includeMetadata: true },
       handler,
     );
   }
+
+  async ensureQueueExists(queueName: string) {
+    const currentQueue = await this.pgBoss.getQueue(queueName);
+    if (!currentQueue) {
+      await this.pgBoss.createQueue(queueName);
+    }
+  }
 }

lib/utils/helpers.ts

@@ -0,0 +1,22 @@
+import PgBoss from "pg-boss";
+
+export function transformOptions(
+  options?: PgBoss.WorkOptions | PgBoss.ScheduleOptions,
+) {
+  if (!options) return {};
+
+  const transformedOptions: any = { ...options };
+
+  if (typeof options.priority === "number") {
+    transformedOptions.priority = options.priority > 0;
+  }
+
+  return transformedOptions;
+}
+
+export function normalizeJob(job: any) {
+  if (typeof job === "object" && "0" in job) {
+    return job[0];
+  }
+  return job;
+}

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@wavezync/nestjs-pgboss",
-  "version": "2.2.0",
+  "version": "3.0.0",
   "description": "A NestJS module that integrates pg-boss for job scheduling and handling.",
   "license": "MIT",
   "author": "[email protected]",
@@ -23,7 +23,7 @@
   "peerDependencies": {
     "@nestjs/common": "^9 || ^10",
     "@nestjs/core": "^9 || ^10",
-    "pg-boss": "^9",
+    "pg-boss": "^10",
     "reflect-metadata": "^0.1.13 || ^0.2.0",
     "rxjs": "^7.2.0"
   },

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({
@@ -32,13 +34,13 @@ describe("PgBossService", () => {
   describe("registerJob", () => {
     it("should call PgBoss work with correct parameters", async () => {
       const handler = jest.fn();
-      const options = { batchSize: 5 };
+      const options = {};
 
       await service.registerJob("test-job", handler, options);
 
       expect(mockPgBoss.work).toHaveBeenCalledWith(
         "test-job",
-        { batchSize: 5, includeMetadata: true },
+        { includeMetadata: true },
         handler,
       );
     });