feat(notifications): upgrades pg-boss

Author: ygrishajev

apps/api/src/core/services/job-queue/job-queue.service.ts

@@ -37,7 +37,6 @@ export class JobQueueService implements Disposable {
       }
       seenJobs.add(queueName);
       await this.pgBoss.createQueue(queueName, {
-        name: queueName,
         retryLimit: 5,
         retryBackoff: true,
         retryDelayMax: 5 * 60

apps/notifications/package.json

@@ -73,7 +73,7 @@
     "murmurhash": "^2.0.1",
     "nestjs-zod": "^4.3.1",
     "pg": "^8.13.0",
-    "pg-boss": "^10.1.6",
+    "pg-boss": "^11.0.4",
     "pino": "^9.7.0",
     "reflect-metadata": "^0.2.2",
     "rxjs": "^7.8.1",

apps/notifications/src/infrastructure/broker/providers/pg-boss/pg-boss.provider.spec.ts

@@ -26,7 +26,6 @@ describe("createPgBoss", () => {
     expect(MockPgBoss).toHaveBeenCalledTimes(2);
     expect(MockPgBoss.mock.calls[0][0]).toEqual(config["broker.EVENT_BROKER_POSTGRES_URI"]);
     expect(MockPgBoss.mock.calls[1][0]).toEqual({
-      archiveCompletedAfterSeconds: config["broker.EVENT_BROKER_ARCHIVE_COMPLETED_AFTER_SECONDS"],
       db: {
         executeSql: expect.any(Function)
       }

apps/notifications/src/infrastructure/broker/providers/pg-boss/pg-boss.provider.ts

@@ -19,8 +19,7 @@ export const createPgBossFactory =
         executeSql(text: string, values: any[]): Promise<{ rows: any[] }> {
           return client.query(text, values);
         }
-      },
-      archiveCompletedAfterSeconds: config.getOrThrow("broker.EVENT_BROKER_ARCHIVE_COMPLETED_AFTER_SECONDS")
+      }
     });
 
     const logger = new LoggerService({ context: "PgBoss" });

apps/notifications/src/infrastructure/broker/services/broker/broker.service.spec.ts

@@ -46,7 +46,9 @@ describe(BrokerService.name, () => {
 
       await service.subscribe(eventName, options, handler);
 
-      expect(pgBoss.createQueue).toHaveBeenCalledWith(queueName);
+      expect(pgBoss.createQueue).toHaveBeenCalledWith(queueName, {
+        deleteAfterSeconds: configService.getOrThrow("broker.EVENT_BROKER_ARCHIVE_COMPLETED_AFTER_SECONDS")
+      });
       expect(pgBoss.subscribe).toHaveBeenCalledWith(eventName, queueName);
       expect(pgBoss.work).toHaveBeenCalledTimes(options.prefetchCount);
       expect(pgBoss.work).toHaveBeenCalledWith(queueName, expect.any(Function));

apps/notifications/src/infrastructure/broker/services/broker/broker.service.ts

@@ -36,7 +36,9 @@ export class BrokerService {
 
   async subscribe<ReqData>(eventName: string, options: { prefetchCount: number }, handler: SingleMsgWorkHandler<ReqData>) {
     const queueName = this.toQueueName(eventName);
-    await this.boss.createQueue(queueName);
+    await this.boss.createQueue(queueName, {
+      deleteAfterSeconds: this.configService.getOrThrow("broker.EVENT_BROKER_ARCHIVE_COMPLETED_AFTER_SECONDS")
+    });
     await this.boss.subscribe(eventName, queueName);
 
     await Promise.all(