Queues with BullMQ

A queue moves slow work out of the request into Redis, where a separate worker picks it up and a restart loses nothing.

Syntaxqueue.add(name: string, data: D, opts?: JobsOptions): Promise<Job<D, R, string>>

A request that sends an email, renders a PDF or calls a slow third-party API makes the client wait for work it does not need to watch finish. A queue breaks that link. Your handler writes a job — a name plus a JSON payload — into Redis and returns. Another process reads the job later and does the work.

The pieces

BullMQ is the queue library; @nestjs/bullmq wires it into Nest's injector. BullModule.forRoot() carries the Redis connection and is imported once, in the root module. BullModule.registerQueue({ name: 'email' }) creates that one queue and exports a provider for it — an object the injector owns and passes to constructors — under the token BullQueue_email. A token is the key the injector looks a provider up by. @InjectQueue('email') asks for that key.

Where it runs

queue.add() is called inside your handler, after pipes have validated the body and before response interceptors run. The job itself runs outside the request entirely: no guards, no interceptors, no filters, no request object.

What survives a restart

Waiting jobs are Redis records, so stopping the API loses none of them. A job that was mid-flight is different: its worker's lock expires, the stalled check moves it back to wait, and it runs again from the start.

What trips people up

The payload is JSON. A Date comes back a string and a class instance comes back a plain object. Store row ids, not rows — the row may have changed before the worker reads it.

Example

Example · typescript
// src/app.module.ts
import { Module } from '@nestjs/common';
import { BullModule } from '@nestjs/bullmq';
import { EmailModule } from './email/email.module';

@Module({
  imports: [
    // One Redis connection shared by every queue. Import forRoot once, here.
    BullModule.forRoot({
      connection: { host: process.env.REDIS_HOST ?? '127.0.0.1', port: 6379 },
      // Applied to every job on every queue unless the add() call overrides them.
      defaultJobOptions: {
        attempts: 3,
        backoff: { type: 'exponential', delay: 1000 },
      },
    }),
    EmailModule,
  ],
})
export class AppModule {}

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

@Module({
  // Creates the "email" queue and provides it under the token BullQueue_email.
  imports: [BullModule.registerQueue({ name: 'email' })],
  providers: [EmailProducer],
  exports: [EmailProducer],
})
export class EmailModule {}

// src/email/email.producer.ts
import { Injectable } from '@nestjs/common';
import { InjectQueue } from '@nestjs/bullmq';
import { Queue } from 'bullmq';

export interface WelcomeEmailJob {
  userId: string;
}

@Injectable()
export class EmailProducer {
  // Queue is BullMQ's class. InjectQueue resolves the BullQueue_email token.
  constructor(
    @InjectQueue('email') private readonly emailQueue: Queue<WelcomeEmailJob>,
  ) {}

  // Call this from a controller: it returns once Redis has the job.
  async queueWelcomeEmail(userId: string): Promise<string> {
    const job = await this.emailQueue.add('welcome', { userId });
    // id is assigned by Redis; it is typed optional on the Job class.
    return job.id as string;
  }
}

When to use it

  • Return 202 from an upload endpoint and transcode the file in the background.
  • Send a welcome email without the signup request waiting on your mail provider.
  • Call a flaky partner API from a worker so a timeout does not become a 504 for the user.
  • Fan out one "invoice run" request into one job per customer with addBulk.

More examples

Reading the Redis connection from config

forRoot takes a literal options object; forRootAsync takes a factory and resolves the same options at startup. Reach for it as soon as the host, port or password comes from the environment rather than from source, which is every deployment beyond your laptop. prefix namespaces every Redis key, so two environments can share one Redis instance.

Example · typescript
// src/app.module.ts
import { Module } from '@nestjs/common';
import { BullModule } from '@nestjs/bullmq';
import { ConfigModule, ConfigService } from '@nestjs/config';

@Module({
  imports: [
    ConfigModule.forRoot({ isGlobal: true }),
    // forRootAsync is the longhand of forRoot: same options object, but built
    // by a factory that can inject providers. Everything else is identical.
    BullModule.forRootAsync({
      imports: [ConfigModule],
      inject: [ConfigService],
      useFactory: (config: ConfigService) => ({
        connection: {
          host: config.getOrThrow<string>('REDIS_HOST'),
          port: config.getOrThrow<number>('REDIS_PORT'),
          password: config.get<string>('REDIS_PASSWORD'),
        },
        prefix: config.get<string>('QUEUE_PREFIX') ?? 'app',
      }),
    }),
  ],
})
export class AppModule {}

Injecting the queue the long way, and adding jobs in bulk

Shows that the decorator is nothing magical: it is @Inject with a generated string token. The longhand is what you write when the queue name is computed, and getQueueToken is also how you override a queue provider in a test module. addBulk matters when a scheduled sweep enqueues thousands of jobs at once.

Example · typescript
// src/email/digest.service.ts
import { Inject, Injectable } from '@nestjs/common';
import { getQueueToken } from '@nestjs/bullmq';
import { Queue } from 'bullmq';

@Injectable()
export class DigestService {
  // @InjectQueue('email') is shorthand for exactly this. getQueueToken('email')
  // returns the string 'BullQueue_email'; both forms resolve the same provider,
  // so you will meet either one in other people's code.
  constructor(
    @Inject(getQueueToken('email')) private readonly emailQueue: Queue,
  ) {}

  async queueWeeklyDigests(userIds: string[]): Promise<void> {
    // addBulk is one round trip to Redis instead of one per job.
    await this.emailQueue.addBulk(
      userIds.map((userId) => ({
        name: 'weekly-digest',
        data: { userId },
        opts: { attempts: 3 },
      })),
    );
  }
}

Per-job options: delay, priority and retention

Options passed to add() override the queue's defaultJobOptions for that job. Use delay for "remind them in an hour" work, and set retention explicitly: BullMQ keeps completed and failed jobs forever by default, which eventually fills Redis. Note the units differ - delay is milliseconds, the retention age is seconds.

Example · typescript
// src/orders/orders.service.ts
import { Injectable } from '@nestjs/common';
import { InjectQueue } from '@nestjs/bullmq';
import { Queue } from 'bullmq';

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

  async sendReceiptLater(orderId: string): Promise<void> {
    await this.emailQueue.add(
      'order-receipt',
      { orderId },
      {
        delay: 60_000, // becomes available a minute from now
        // 0 means unprioritised, and unprioritised jobs run before prioritised
        // ones. Among prioritised jobs, a lower number runs first. Max 2097151.
        priority: 1,
        attempts: 5,
        removeOnComplete: 1000, // shorthand for { count: 1000 }: keep the last 1000
        removeOnFail: { age: 7 * 24 * 60 * 60 }, // age is in SECONDS, not ms
      },
    );
  }
}

Related

nest queue processorsnest queue retriesnest queue idempotencynest cache redisnest dynamic modules

Discussion

  • Be the first to comment on this lesson.