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

Related explainers

typescript 
import { Component, computed, DestroyRef, inject, input, output, signal } from '@angular/core';
import { takeUntilDestroyed } from '@angular/core/rxjs-interop';
import { interval, map, takeWhile } from 'rxjs';
 

How a signal-driven countdown works in Angular

signals reactivity rxjs
Intermediate 8 steps
typescript 
type TokenType = "keyword" | "string" | "comment" | "number" | "text";
 
interface Token {
  type: TokenType;

How a regex tokenizer highlights code

tokenizer regex lexing
Intermediate 10 steps
python 
import secrets
 
from fastapi import Depends, FastAPI, HTTPException, Security, status
from fastapi.security import APIKeyHeader

API key authentication as a FastAPI dependency

authentication dependency-injection api-keys
Intermediate 8 steps
typescript 
import { Component, inject, signal } from '@angular/core';
import { CdkDragDrop, DragDropModule, moveItemInArray } from '@angular/cdk/drag-drop';
import { HttpClient } from '@angular/common/http';
import { finalize } from 'rxjs';

Drag-and-drop reordering with signals in Angular

drag-and-drop signals optimistic-update
Intermediate 8 steps
typescript 
type Middleware<TIn, TOut> = (ctx: TIn) => Promise<TOut> | TOut;
 
class Pipeline<TIn, TOut> {
  private constructor(private readonly run: Middleware<TIn, TOut>) {}

A type-safe async middleware pipeline

generics type-safety middleware
Advanced 9 steps
java 
@Component
public class HeaderMergeFilter {
 
    private static final String PER_REQUEST_HEADERS = HeaderMergeFilter.class.getName() + ".headers";

Merging default and per-request headers in Spring

webclient filters http-headers
Intermediate 8 steps

Share this explainer

Here's the card — post it anywhere.

Processing background email jobs in NestJS — share card
Made with highlit — turn any snippet into a walkthrough like this in about a minute.
Explain your code