feat(nestjs-notifications): register BullMQ workers in forRootAsync path

The async module path created Queue instances but never registered
processor workers, leaving enqueued jobs unprocessed. This adds manual
Worker factories that bypass @processor decorator discovery (which would
crash before async options resolve) and a WorkerCleanupService for
graceful shutdown.

Author: JonatanSalas

packages/nestjs-notifications/src/index.ts

@@ -34,12 +34,18 @@ export { PreferenceController } from './preferences/preference.controller';
 // Templates
 export { TemplateService } from './templates/template.service';
 
+// Worker cleanup
+export { WorkerCleanupService } from './worker-cleanup.service';
+
 // Interfaces and constants
 export {
   NOTIFICATION_MODULE_OPTIONS,
   EMAIL_PROVIDER,
   SMS_PROVIDER,
   PUSH_PROVIDER,
+  EMAIL_WORKER,
+  SMS_WORKER,
+  PUSH_WORKER,
 } from './interfaces';
 
 // Auth integration

packages/nestjs-notifications/src/interfaces/index.ts

@@ -170,3 +170,6 @@ export const NOTIFICATION_MODULE_OPTIONS = 'NOTIFICATION_MODULE_OPTIONS';
 export const EMAIL_PROVIDER = 'EMAIL_PROVIDER';
 export const SMS_PROVIDER = 'SMS_PROVIDER';
 export const PUSH_PROVIDER = 'PUSH_PROVIDER';
+export const EMAIL_WORKER = 'EMAIL_WORKER';
+export const SMS_WORKER = 'SMS_WORKER';
+export const PUSH_WORKER = 'PUSH_WORKER';

packages/nestjs-notifications/src/notification.module.spec.ts

@@ -16,6 +16,7 @@ jest.mock('./channels/push/providers/firebase.provider', () => ({
 }));
 jest.mock('bullmq', () => ({
   Queue: jest.fn().mockImplementation(() => ({})),
+  Worker: jest.fn().mockImplementation(() => ({ close: jest.fn() })),
 }));
 jest.mock('@nestjs/bullmq', () => ({
   BullModule: {
@@ -48,8 +49,12 @@ import {
   EMAIL_PROVIDER,
   SMS_PROVIDER,
   PUSH_PROVIDER,
+  EMAIL_WORKER,
+  SMS_WORKER,
+  PUSH_WORKER,
   type NotificationModuleOptions,
 } from './interfaces';
+import { WorkerCleanupService } from './worker-cleanup.service';
 
 describe('NotificationModule', () => {
   describe('forRoot', () => {
@@ -368,6 +373,21 @@ describe('NotificationModule', () => {
       expect(optionsProvider.inject).toEqual([]);
     });
 
+    it('should include worker tokens and WorkerCleanupService in providers', () => {
+      const result = NotificationModule.forRootAsync({
+        useFactory: () => ({
+          channels: {},
+        }),
+        inject: [],
+      });
+
+      const providerTokens = extractProviderTokens(result.providers as any[]);
+      expect(providerTokens).toContain(EMAIL_WORKER);
+      expect(providerTokens).toContain(SMS_WORKER);
+      expect(providerTokens).toContain(PUSH_WORKER);
+      expect(providerTokens).toContain(WorkerCleanupService);
+    });
+
     it('should export all core services', () => {
       const result = NotificationModule.forRootAsync({
         useFactory: () => ({ channels: {} }),

packages/nestjs-notifications/src/notification.module.ts

@@ -1,13 +1,19 @@
 import { DynamicModule, Module, Provider, Type } from '@nestjs/common';
 import { BullModule, getQueueToken } from '@nestjs/bullmq';
-import { Queue } from 'bullmq';
+import { Queue, Worker, Job } from 'bullmq';
 import {
   NOTIFICATION_MODULE_OPTIONS,
   EMAIL_PROVIDER,
   SMS_PROVIDER,
   PUSH_PROVIDER,
+  EMAIL_WORKER,
+  SMS_WORKER,
+  PUSH_WORKER,
   type NotificationModuleOptions,
   type NotificationModuleAsyncOptions,
+  type EmailProvider,
+  type SmsProvider,
+  type PushProvider,
 } from './interfaces';
 import { NotificationService } from './notification.service';
 import { InAppService } from './channels/in-app/in-app.service';
@@ -24,6 +30,7 @@ import { SmtpEmailProvider } from './channels/email/providers/smtp.provider';
 import { SendGridEmailProvider } from './channels/email/providers/sendgrid.provider';
 import { TwilioSmsProvider } from './channels/sms/providers/twilio.provider';
 import { FirebasePushProvider } from './channels/push/providers/firebase.provider';
+import { WorkerCleanupService } from './worker-cleanup.service';
 
 @Module({})
 export class NotificationModule {
@@ -256,6 +263,84 @@ export class NotificationModule {
         },
         inject: [NOTIFICATION_MODULE_OPTIONS],
       },
+      // Worker factories — create BullMQ Workers manually to bypass @Processor
+      // decorator discovery which would start workers before async options resolve.
+      {
+        provide: EMAIL_WORKER,
+        useFactory: (
+          options: NotificationModuleOptions,
+          emailProvider: EmailProvider | null,
+          prisma: any,
+        ) => {
+          if (!options.channels.email?.enabled || !options.queue?.redis || !emailProvider) {
+            return null;
+          }
+          const processor = new EmailProcessor(emailProvider, prisma);
+          return new Worker(
+            'notifications-email',
+            async (job: Job) => processor.process(job),
+            {
+              connection: {
+                host: options.queue.redis.host,
+                port: options.queue.redis.port ?? 6379,
+                password: options.queue.redis.password,
+              },
+            },
+          );
+        },
+        inject: [NOTIFICATION_MODULE_OPTIONS, EMAIL_PROVIDER, 'PRISMA_SERVICE'],
+      },
+      {
+        provide: SMS_WORKER,
+        useFactory: (
+          options: NotificationModuleOptions,
+          smsProvider: SmsProvider | null,
+          prisma: any,
+        ) => {
+          if (!options.channels.sms?.enabled || !options.queue?.redis || !smsProvider) {
+            return null;
+          }
+          const processor = new SmsProcessor(smsProvider, prisma);
+          return new Worker(
+            'notifications-sms',
+            async (job: Job) => processor.process(job),
+            {
+              connection: {
+                host: options.queue.redis.host,
+                port: options.queue.redis.port ?? 6379,
+                password: options.queue.redis.password,
+              },
+            },
+          );
+        },
+        inject: [NOTIFICATION_MODULE_OPTIONS, SMS_PROVIDER, 'PRISMA_SERVICE'],
+      },
+      {
+        provide: PUSH_WORKER,
+        useFactory: (
+          options: NotificationModuleOptions,
+          pushProvider: PushProvider | null,
+          prisma: any,
+        ) => {
+          if (!options.channels.push?.enabled || !options.queue?.redis || !pushProvider) {
+            return null;
+          }
+          const processor = new PushProcessor(pushProvider, prisma);
+          return new Worker(
+            'notifications-push',
+            async (job: Job) => processor.process(job),
+            {
+              connection: {
+                host: options.queue.redis.host,
+                port: options.queue.redis.port ?? 6379,
+                password: options.queue.redis.password,
+              },
+            },
+          );
+        },
+        inject: [NOTIFICATION_MODULE_OPTIONS, PUSH_PROVIDER, 'PRISMA_SERVICE'],
+      },
+      WorkerCleanupService,
     ];
 
     const controllers: Type[] = [InAppController, DeviceTokenController, PreferenceController];

packages/nestjs-notifications/src/worker-cleanup.service.spec.ts

@@ -0,0 +1,59 @@
+import { WorkerCleanupService } from './worker-cleanup.service';
+
+describe('WorkerCleanupService', () => {
+  function createMockWorker() {
+    return { close: jest.fn().mockResolvedValue(undefined) };
+  }
+
+  it('should call close on all non-null workers', async () => {
+    const emailWorker = createMockWorker();
+    const smsWorker = createMockWorker();
+    const pushWorker = createMockWorker();
+
+    const service = new WorkerCleanupService(
+      emailWorker as any,
+      smsWorker as any,
+      pushWorker as any,
+    );
+
+    await service.onModuleDestroy();
+
+    expect(emailWorker.close).toHaveBeenCalledTimes(1);
+    expect(smsWorker.close).toHaveBeenCalledTimes(1);
+    expect(pushWorker.close).toHaveBeenCalledTimes(1);
+  });
+
+  it('should handle null workers gracefully', async () => {
+    const service = new WorkerCleanupService(null, null, null);
+
+    await expect(service.onModuleDestroy()).resolves.not.toThrow();
+  });
+
+  it('should handle mixed null and non-null workers', async () => {
+    const smsWorker = createMockWorker();
+
+    const service = new WorkerCleanupService(null, smsWorker as any, null);
+
+    await service.onModuleDestroy();
+
+    expect(smsWorker.close).toHaveBeenCalledTimes(1);
+  });
+
+  it('should not block other workers if one close fails', async () => {
+    const emailWorker = { close: jest.fn().mockRejectedValue(new Error('close failed')) };
+    const smsWorker = createMockWorker();
+    const pushWorker = createMockWorker();
+
+    const service = new WorkerCleanupService(
+      emailWorker as any,
+      smsWorker as any,
+      pushWorker as any,
+    );
+
+    await expect(service.onModuleDestroy()).resolves.not.toThrow();
+
+    expect(emailWorker.close).toHaveBeenCalledTimes(1);
+    expect(smsWorker.close).toHaveBeenCalledTimes(1);
+    expect(pushWorker.close).toHaveBeenCalledTimes(1);
+  });
+});

packages/nestjs-notifications/src/worker-cleanup.service.ts

@@ -0,0 +1,38 @@
+import { Injectable, Optional, Inject, OnModuleDestroy, Logger } from '@nestjs/common';
+import type { Worker } from 'bullmq';
+import { EMAIL_WORKER, SMS_WORKER, PUSH_WORKER } from './interfaces';
+
+@Injectable()
+export class WorkerCleanupService implements OnModuleDestroy {
+  private readonly logger = new Logger(WorkerCleanupService.name);
+
+  constructor(
+    @Optional() @Inject(EMAIL_WORKER) private readonly emailWorker: Worker | null,
+    @Optional() @Inject(SMS_WORKER) private readonly smsWorker: Worker | null,
+    @Optional() @Inject(PUSH_WORKER) private readonly pushWorker: Worker | null,
+  ) {}
+
+  async onModuleDestroy(): Promise<void> {
+    const workers = [
+      { name: 'email', worker: this.emailWorker },
+      { name: 'sms', worker: this.smsWorker },
+      { name: 'push', worker: this.pushWorker },
+    ];
+
+    const results = await Promise.allSettled(
+      workers
+        .filter(({ worker }) => worker != null)
+        .map(async ({ name, worker }) => {
+          this.logger.log(`Closing ${name} worker...`);
+          await worker!.close();
+          this.logger.log(`${name} worker closed`);
+        }),
+    );
+
+    for (const result of results) {
+      if (result.status === 'rejected') {
+        this.logger.error(`Failed to close worker: ${result.reason}`);
+      }
+    }
+  }
+}