Bullmq Patterns

BullMQ + NestJS Queue Processing Patterns

Module Setup

// feature.module.ts
@Module({
  imports: [
    BullModule.registerQueue({
      name: FEATURE_QUEUE_NAME,
      defaultJobOptions: {
        removeOnComplete: false,  // Keep completed jobs for debugging
        removeOnFail: false,      // Keep failed jobs for retry/analysis
      },
    }),
    // ... other imports
  ],
  providers: [FeatureProcessor, FeatureSchedulerService],
  exports: [BullModule], // Export so other modules can enqueue jobs
})
export class FeatureModule {}

AppModule BullMQ root config:

BullModule.forRoot({
  connection: {
    host: process.env.REDIS_HOST,
    port: parseInt(process.env.REDIS_PORT ?? '6379'),
    password: process.env.REDIS_PASSWORD,
  },
  defaultJobOptions: {
    removeOnComplete: false,
    removeOnFail: false,
  },
})

Processor Pattern (WorkerHost)

@Injectable()
@Processor(FEATURE_QUEUE_NAME, { concurrency: 1 })
export class FeatureProcessor extends WorkerHost {
  constructor(
    private readonly service: FeatureService,
    private readonly logger: OtelLogger,      // Or Logger from @nestjs/common
    private readonly metricsService: MetricsService,
  ) {
    super();
    this.logger.setContext(FeatureProcessor.name);
  }

  async process(job: Job<FeatureJobData>): Promise<void> {
    await withJobSpan(
      job,
      FeatureSpanName.PROCESS,
      this.metricsService,
      async () => {
        const { userId, featureId } = job.data;
        this.logger.log(`Processing feature ${featureId} for user ${userId}`);

        // Business logic here
        await this.service.processFeature(featureId, userId);
      },
    );
  }
}

Key rules:

• Always extend WorkerHost from @nestjs/bullmq
• Always wrap process() body in withJobSpan() for OpenTelemetry tracing
• Set concurrency explicitly ({ concurrency: 1 } for sequential, higher for parallel)
• Use typed Job<DataType> for type-safe job data
• Call this.logger.setContext() in constructor

Type-Safe Job Data

// feature.constants.ts
export type FeatureJobData = {
  featureId: string;
  userId: string;
  amountInMinor: number;
  metadata?: Record<string, unknown>;
};

// In processor
async process(job: Job<FeatureJobData>): Promise<void> {
  const { featureId, userId, amountInMinor } = job.data;
}

Constants Pattern

// feature.constants.ts

// Queue name — used in BullModule.registerQueue + @InjectQueue
export const FEATURE_QUEUE_NAME = 'feature-processing' as const;

// Job names — used when enqueuing (descriptive, for Bull Board UI)
export const FeatureJobName = {
  PROCESS: 'process-feature',
  CLEANUP: 'cleanup-expired',
} as const;

// Job IDs — for deduplication at enqueue time
export const FeatureJobId = {
  PROCESS: 'process-feature-job',
  CLEANUP: 'cleanup-expired-job',
} as const;

// OTel span names — dot-notation for tracing hierarchy
export const FeatureSpanName = {
  PROCESS: 'feature.process',
  CLEANUP: 'feature.cleanup',
} as const;

CRITICAL: @InjectQueue and BullModule.registerQueue must use the queue name constant (FEATURE_QUEUE_NAME), never a job-ID or job-name constant. These are different things:

• Queue name = the Redis queue identifier
• Job name = human-readable label for the job type
• Job ID = deduplication key

Enqueuing Jobs

@Injectable()
export class PaymentService {
  constructor(
    @InjectQueue(FEATURE_QUEUE_NAME) private readonly featureQueue: Queue,
    private readonly bullMQService: BullMQService,
  ) {}

  async afterPayment(payment: Payment) {
    // Standard job
    await this.bullMQService.addJob(
      this.featureQueue,
      FeatureJobName.PROCESS,
      { featureId: payment.featureId, userId: payment.userId },
      { jobId: payment.reference, attempts: 3 },
    );

    // Delayed job (execute after N ms)
    await this.bullMQService.addDelayedJob(
      this.featureQueue,
      FeatureJobName.CLEANUP,
      { featureId: payment.featureId },
      { delay: 3_600_000 }, // 1 hour
    );
  }
}

BullMQ Service (Trace Context Propagation)

@Injectable()
export class BullMQService {
  async addJob<T>(queue: Queue, name: string, data: T, opts?: JobsOptions) {
    // Capture current OTel trace context and inject into job data
    const context = this.captureTraceContext();
    await queue.add(name, { ...data, __otelContext: context }, opts);
  }

  async addUniqueCronJob<T>(queue: Queue, name: string, data: T, jobId: string) {
    const uniqueId = `cron-${jobId}-${Date.now()}`;
    await this.addJob(queue, name, data, { jobId: uniqueId });
  }
}

This makes processor spans children of the enqueuing request span in distributed traces.

Scheduler Pattern (Cron)

@Injectable()
export class FeatureSchedulerService {
  constructor(
    @InjectQueue(FEATURE_QUEUE_NAME) private readonly queue: Queue,
    private readonly bullMQService: BullMQService,
  ) {}

  @Cron(CronExpression.EVERY_30_MINUTES)
  async triggerPeriodicProcess() {
    await this.bullMQService.addUniqueCronJob(
      this.queue,
      FeatureJobName.PROCESS,
      {},
      FeatureJobId.PROCESS,
    );
  }

  @Cron(CronExpression.EVERY_DAY_AT_MIDNIGHT)
  async triggerDailyCleanup() {
    await this.bullMQService.addUniqueCronJob(
      this.queue,
      FeatureJobName.CLEANUP,
      {},
      FeatureJobId.CLEANUP,
    );
  }
}

Key: Use addUniqueCronJob which generates timestamp-based jobIds to prevent duplicates within the same cron cycle.

withJobSpan (OTel Tracing)

async function withJobSpan<T>(
  job: Job,
  spanName: string,
  metricsService: MetricsService,
  fn: () => Promise<T>,
): Promise<T> {
  // 1. Extract OTel context from job.data.__otelContext
  // 2. Create child span with job metadata (id, name, queueName)
  // 3. Execute fn() within span context
  // 4. Record success/failure metrics
  // 5. Set span status + exception on error
  return result;
}

This ensures:

• Distributed trace continuity (HTTP request -> queue -> processor)
• Automatic error recording in traces
• Job duration metrics

Bull Board Integration

// bootstrap function
export function setupBullBoard(app: NestFastifyApplication) {
  const serverAdapter = new FastifyAdapter();
  createBullBoard({
    queues: [
      new BullMQAdapter(app.get(getQueueToken(FEATURE_QUEUE_NAME))),
      new BullMQAdapter(app.get(getQueueToken(OTHER_QUEUE_NAME))),
    ],
    serverAdapter,
  });
  serverAdapter.setBasePath('/api/queues');
  app.register(serverAdapter.registerPlugin(), { prefix: '/api/queues' });
}

Idempotency

// In processor
async process(job: Job<LoyaltyJobData>): Promise<void> {
  await withJobSpan(job, spanName, this.metrics, async () => {
    // Check if already processed (idempotency)
    const exists = await this.repo.hasEntry(job.data.referenceId);
    if (exists) {
      this.logger.warn(`Duplicate job for reference ${job.data.referenceId}, skipping`);
      return;
    }
    // Process...
  });
}

Error Handling

• Failed jobs stay in the queue (removeOnFail: false) for manual inspection
• Use attempts option for automatic retries: { attempts: 3, backoff: { type: 'exponential', delay: 5000 } }
• Log errors with context before re-throwing
• Never silently swallow errors in processors