typescript 50 lines · 7 steps

Processing background email jobs in NestJS

A BullMQ worker consumes queued email jobs, dispatches them by name, and reports progress and failures.

Explained by highlit

NestJS Intermediate job-queue background-processing dependency-injection event-handling

1import { Processor, WorkerHost, OnWorkerEvent } from '@nestjs/bullmq';
2import { Logger } from '@nestjs/common';
3import { Job } from 'bullmq';
4import { MailerService } from '../mailer/mailer.service';
5import { UsersService } from '../users/users.service';
6 
7interface WelcomeEmailData {
8  userId: string;
9  locale: string;
10}
11 
12@Processor('emails', { concurrency: 5 })
13export class EmailsProcessor extends WorkerHost {
14  private readonly logger = new Logger(EmailsProcessor.name);
15 
16  constructor(
17    private readonly mailer: MailerService,
18    private readonly users: UsersService,
19  ) {
20    super();
21  }
22 
23  async process(job: Job): Promise<void> {
24    switch (job.name) {
25      case 'welcome':
26        return this.handleWelcome(job as Job<WelcomeEmailData>);
27      default:
28        throw new Error(`Unhandled email job: ${job.name}`);
29    }
30  }
31 
32  private async handleWelcome(job: Job<WelcomeEmailData>): Promise<void> {
33    const { userId, locale } = job.data;
34    const user = await this.users.findByIdOrFail(userId);
35 
36    await job.updateProgress(50);
37    await this.mailer.send({
38      to: user.email,
39      template: 'welcome',
40      locale,
41      context: { name: user.firstName },
42    });
43    await job.updateProgress(100);
44  }
45 
46  @OnWorkerEvent('failed')
47  onFailed(job: Job, err: Error): void {
48    this.logger.error(`Job ${job.id} (${job.name}) failed: ${err.message}`, err.stack);
49  }
50}

01 / 01

STEP 01

Walkthrough

Space play ←→ step click any line

Three takeaways

1A single processor can route many job types by branching on the job's name.
2Reporting progress with updateProgress lets producers and dashboards track long-running work.
3Worker event hooks like failed centralize error logging across every job the queue handles.

Related explainers

python

from datetime import datetime
from fastapi import APIRouter, Depends, HTTPException, status
from pydantic import BaseModel, ConfigDict, EmailStr
from sqlalchemy.orm import Session

Building a users router in FastAPI

routing serialization dependency-injection

Intermediate 7 steps

typescript

type Event = Record<string, unknown>;
 
interface BatcherOptions {
  endpoint: string;

Batching analytics events in TypeScript

batching buffering async

Intermediate 8 steps

java

@Service
public class OrderCheckoutService {
 
    private final OrderRepository orderRepository;

How @Transactional guards a Spring checkout

dependency-injection transactions atomicity

Intermediate 6 steps

typescript

import { Injectable } from '@angular/core';
import { PreloadingStrategy, Route, Routes, RouterModule } from '@angular/router';
import { NgModule } from '@angular/core';
import { Observable, of, timer } from 'rxjs';

A custom preloading strategy in Angular

lazy-loading preloading rxjs

Intermediate 7 steps

java

@Service
public class OrderMetricsService {
 
    private final MeterRegistry registry;

Instrumenting orders with Micrometer in Spring

metrics micrometer observability

Intermediate 8 steps

typescript

import { Component, computed, inject, signal } from '@angular/core';
import { FormBuilder, ReactiveFormsModule, Validators } from '@angular/forms';
import { CommonModule } from '@angular/common';