Queue events and monitoring in NestJS

Introduction

Queue events and monitoring help you watch what happens inside your task queues. This lets you know if tasks are waiting, running, or done, so you can fix problems fast.

You want to see if background jobs are working correctly in your NestJS app.

You need to know when a task fails so you can retry or alert someone.

You want to track how many jobs are waiting or completed to improve performance.

You want to log queue activity for debugging or auditing.

You want to build a dashboard showing queue status in real time.

Syntax

NestJS

import { Processor, Process, OnQueueEvent } from '@nestjs/bull';
import { Job } from 'bullmq';

@Processor('myQueue')
export class MyQueueProcessor {
  @Process()
  async handleJob(job: Job) {
    // job processing code here
  }

  @OnQueueEvent('completed')
  onJobCompleted(event: any) {
    console.log('Job completed:', event.jobId);
  }

  @OnQueueEvent('failed')
  onJobFailed(event: any) {
    console.log('Job failed:', event.jobId, 'Reason:', event.failedReason);
  }
}

Use @Processor('queueName') to define which queue to listen to.

@OnQueueEvent('eventName') listens to specific queue events like 'completed', 'failed', 'waiting', etc.

Examples

Basic example listening to 'completed' event for an email queue.

NestJS

import { Processor, Process, OnQueueEvent } from '@nestjs/bull';
import { Job } from 'bullmq';

@Processor('emailQueue')
export class EmailProcessor {
  @Process()
  async sendEmail(job: Job) {
    // send email logic
  }

  @OnQueueEvent('completed')
  onCompleted(event: any) {
    console.log(`Email job ${event.jobId} completed`);
  }
}

Example showing how to catch failed jobs and log the reason.

NestJS

import { Processor, Process, OnQueueEvent } from '@nestjs/bull';
import { Job } from 'bullmq';

@Processor('taskQueue')
export class TaskProcessor {
  @Process()
  async runTask(job: Job) {
    // task logic
  }

  @OnQueueEvent('failed')
  onFailed(event: any) {
    console.error(`Task job ${event.jobId} failed: ${event.failedReason}`);
  }
}

Handles the case when jobs are waiting in the queue, useful for empty or slow queues.

NestJS

import { Processor, Process, OnQueueEvent } from '@nestjs/bull';
import { Job } from 'bullmq';

@Processor('emptyQueue')
export class EmptyQueueProcessor {
  @Process()
  async processJob(job: Job) {
    // This queue might be empty sometimes
  }

  @OnQueueEvent('waiting')
  onWaiting(event: any) {
    console.log(`Job ${event.jobId} is waiting in the queue`);
  }
}

Sample Program

This processor handles notification jobs. It logs when a job starts, and listens for completion or failure events to log the results.

NestJS

import { Processor, Process, OnQueueEvent } from '@nestjs/bull';
import { Job } from 'bullmq';

@Processor('notificationQueue')
export class NotificationProcessor {
  @Process()
  async sendNotification(job: Job) {
    console.log(`Processing notification job ${job.id}`);
    // Simulate success or failure
    if (job.data.fail) {
      throw new Error('Notification failed');
    }
    return 'Notification sent';
  }

  @OnQueueEvent('completed')
  onCompleted(event: any) {
    console.log(`Job ${event.jobId} completed successfully.`);
  }

  @OnQueueEvent('failed')
  onFailed(event: any) {
    console.log(`Job ${event.jobId} failed with reason: ${event.failedReason}`);
  }
}

Important Notes

Queue events like 'completed' and 'failed' help you track job status in real time.

Listening to events does not block job processing; it runs alongside your job handler.

Common mistake: forgetting to handle errors inside the job, which can cause silent failures.

Summary

Queue events let you watch what happens to jobs in your queues.

Use @OnQueueEvent decorators to listen for events like 'completed' and 'failed'.

Monitoring queues helps you fix problems and improve your app's background work.