NestJS — Enterprise Node.js

Task Scheduling & Queues (BullMQ)

16 min Lesson 41 of 48

Task Scheduling & Queues (BullMQ)

Production applications routinely need to do work outside the request/response cycle: sending a welcome email after sign-up, generating a weekly report at midnight, or retrying a failed payment. NestJS solves these with two complementary tools: @nestjs/schedule for time-based triggers and @nestjs/bullmq for reliable, persistent background jobs backed by Redis.

Built-in scheduler — @nestjs/schedule

The schedule package wraps the node-cron library and lets you declare recurring work directly on service methods using decorators. Install and register the module once:

npm install @nestjs/schedule

// app.module.ts
import { Module } from '@nestjs/common';
import { ScheduleModule } from '@nestjs/schedule';
import { TasksService } from './tasks/tasks.service';

@Module({
  imports: [ScheduleModule.forRoot()],
  providers: [TasksService],
})
export class AppModule {}

Now any injectable service can host scheduled methods:

import { Injectable, Logger } from '@nestjs/common';
import { Cron, CronExpression, Interval, Timeout } from '@nestjs/schedule';

@Injectable()
export class TasksService {
  private readonly logger = new Logger(TasksService.name);

  // Fires every day at 03:00
  @Cron(CronExpression.EVERY_DAY_AT_3AM)
  purgeExpiredSessions(): void {
    this.logger.log('Purging expired sessions...');
    // ... database cleanup logic
  }

  // Fires every 10 seconds
  @Interval(10_000)
  pollExternalApi(): void {
    this.logger.log('Polling external API');
  }

  // Fires once, 5 seconds after bootstrap
  @Timeout(5_000)
  warmupCache(): void {
    this.logger.log('Cache warmed up');
  }
}

@Cron(expression) — Standard cron syntax (6 fields). Use CronExpression constants for readability.
@Interval(ms) — Fires repeatedly every N milliseconds, measured from the previous execution.
@Timeout(ms) — Fires exactly once after N milliseconds; useful for one-time startup work.

@nestjs/schedule does not survive crashes or horizontal scale-out. If the process dies mid-job, the job is lost. If you run three instances, all three fire the cron simultaneously. For anything that must not be lost or duplicated, use a queue.

Reliable queues — @nestjs/bullmq

BullMQ is a Redis-backed queue library. Jobs are persisted in Redis, so they survive restarts. Each job can be retried automatically on failure, delayed, prioritised, and monitored. @nestjs/bullmq integrates BullMQ as first-class NestJS providers.

npm install @nestjs/bullmq bullmq

// email.module.ts
import { Module } from '@nestjs/common';
import { BullModule } from '@nestjs/bullmq';
import { EmailProducer } from './email.producer';
import { EmailProcessor } from './email.processor';

@Module({
  imports: [
    BullModule.registerQueue({ name: 'email' }),
  ],
  providers: [EmailProducer, EmailProcessor],
})
export class EmailModule {}

Register BullModule.forRoot() once in AppModule to provide the Redis connection, then call BullModule.registerQueue() per feature. The root config accepts any ioredis connection options: { connection: { host: 'localhost', port: 6379 } }.

Producers — adding jobs to the queue

Inject the queue via @InjectQueue('email') and call queue.add(). The first argument is a job name (a discriminator for the processor), the second is the payload, and the third is options:

import { Injectable } from '@nestjs/common';
import { InjectQueue } from '@nestjs/bullmq';
import { Queue } from 'bullmq';

@Injectable()
export class EmailProducer {
  constructor(@InjectQueue('email') private readonly emailQueue: Queue) {}

  async sendWelcomeEmail(userId: string, email: string): Promise<void> {
    await this.emailQueue.add(
      'welcome',
      { userId, email },
      {
        attempts: 3,          // retry up to 3 times on failure
        backoff: { type: 'exponential', delay: 2_000 },
        removeOnComplete: true,
      },
    );
  }

  async sendDailyDigest(emails: string[]): Promise<void> {
    // Delay delivery by 1 hour
    await this.emailQueue.add(
      'digest',
      { emails },
      { delay: 60 * 60 * 1_000 },
    );
  }
}

Processors — consuming jobs

A processor is a class decorated with @Processor('email'). Each @Process('jobName') method handles one job type. BullMQ calls the method concurrently up to the configured concurrency limit and marks the job completed or failed based on whether the method resolves or throws:

import { Processor, WorkerHost } from '@nestjs/bullmq';
import { Process } from '@nestjs/bullmq';
import { Job } from 'bullmq';
import { Logger } from '@nestjs/common';

@Processor('email')
export class EmailProcessor extends WorkerHost {
  private readonly logger = new Logger(EmailProcessor.name);

  async process(job: Job): Promise<void> {
    switch (job.name) {
      case 'welcome':
        await this.handleWelcome(job.data);
        break;
      case 'digest':
        await this.handleDigest(job.data);
        break;
      default:
        this.logger.warn(`Unknown job: ${job.name}`);
    }
  }

  private async handleWelcome(data: { userId: string; email: string }) {
    this.logger.log(`Sending welcome email to ${data.email}`);
    // call mail service...
  }

  private async handleDigest(data: { emails: string[] }) {
    this.logger.log(`Sending digest to ${data.emails.length} subscribers`);
    // call mail service...
  }
}

Extend WorkerHost and implement a single process(job) method (the v2 API). Older tutorials show @Process() on individual methods — that still works but mixing both styles causes confusion. Pick one pattern per processor.

Summary

Use @nestjs/schedule for simple, time-based recurring tasks that are acceptable to lose (polling, cleanup). Use @nestjs/bullmq for anything that must not be dropped: transactional emails, payment processing, report generation. Producers enqueue jobs with queue.add(); processors consume them in WorkerHost.process(). BullMQ's Redis persistence, retry/backoff, and delay options make background jobs reliable by default.