1---
2title: Use Message Queues for Background Jobs
3impact: MEDIUM-HIGH
4impactDescription: Queues enable reliable background processing
5tags: microservices, queues, bullmq, background-jobs
6---
7 
8## Use Message Queues for Background Jobs
9 
10Use `@nestjs/bullmq` for background job processing. Queues decouple long-running tasks from HTTP requests, enable retry logic, and distribute workload across workers. Use them for emails, file processing, notifications, and any task that shouldn't block user requests.
11 
12**Incorrect (long-running tasks in HTTP handlers):**
13 
14```typescript
15// Long-running tasks in HTTP handlers
16@Controller('reports')
17export class ReportsController {
18  @Post()
19  async generate(@Body() dto: GenerateReportDto): Promise<Report> {
20    // This blocks the request for potentially minutes
21    const data = await this.fetchLargeDataset(dto);
22    const report = await this.processData(data); // Slow!
23    await this.sendEmail(dto.email, report); // Can fail!
24    return report; // Client times out
25  }
26}
27 
28// Fire-and-forget without retry
29@Injectable()
30export class EmailService {
31  async sendWelcome(email: string): Promise<void> {
32    // If this fails, email is never sent
33    await this.mailer.send({ to: email, template: 'welcome' });
34    // No retry, no tracking, no visibility
35  }
36}
37 
38// Use setInterval for scheduled tasks
39setInterval(async () => {
40  await cleanupOldRecords();
41}, 60000); // No error handling, memory leaks
42```
43 
44**Correct (use BullMQ for background processing):**
45 
46```typescript
47// Configure BullMQ
48import { BullModule } from '@nestjs/bullmq';
49 
50@Module({
51  imports: [
52    BullModule.forRoot({
53      connection: {
54        host: 'localhost',
55        port: 6379,
56      },
57      defaultJobOptions: {
58        removeOnComplete: 1000,
59        removeOnFail: 5000,
60        attempts: 3,
61        backoff: {
62          type: 'exponential',
63          delay: 1000,
64        },
65      },
66    }),
67    BullModule.registerQueue(
68      { name: 'email' },
69      { name: 'reports' },
70      { name: 'notifications' },
71    ),
72  ],
73})
74export class QueueModule {}
75 
76// Producer: Add jobs to queue
77@Injectable()
78export class ReportsService {
79  constructor(
80    @InjectQueue('reports') private reportsQueue: Queue,
81  ) {}
82 
83  async requestReport(dto: GenerateReportDto): Promise<{ jobId: string }> {
84    // Return immediately, process in background
85    const job = await this.reportsQueue.add('generate', dto, {
86      priority: dto.urgent ? 1 : 10,
87      delay: dto.scheduledFor ? Date.parse(dto.scheduledFor) - Date.now() : 0,
88    });
89 
90    return { jobId: job.id };
91  }
92 
93  async getJobStatus(jobId: string): Promise<JobStatus> {
94    const job = await this.reportsQueue.getJob(jobId);
95    return {
96      status: await job.getState(),
97      progress: job.progress,
98      result: job.returnvalue,
99    };
100  }
101}
102 
103// Consumer: Process jobs
104@Processor('reports')
105export class ReportsProcessor {
106  private readonly logger = new Logger(ReportsProcessor.name);
107 
108  @Process('generate')
109  async generateReport(job: Job<GenerateReportDto>): Promise<Report> {
110    this.logger.log(`Processing report job ${job.id}`);
111 
112    // Update progress
113    await job.updateProgress(10);
114 
115    const data = await this.fetchData(job.data);
116    await job.updateProgress(50);
117 
118    const report = await this.processData(data);
119    await job.updateProgress(90);
120 
121    await this.saveReport(report);
122    await job.updateProgress(100);
123 
124    return report;
125  }
126 
127  @OnQueueActive()
128  onActive(job: Job) {
129    this.logger.log(`Processing job ${job.id}`);
130  }
131 
132  @OnQueueCompleted()
133  onCompleted(job: Job, result: any) {
134    this.logger.log(`Job ${job.id} completed`);
135  }
136 
137  @OnQueueFailed()
138  onFailed(job: Job, error: Error) {
139    this.logger.error(`Job ${job.id} failed: ${error.message}`);
140  }
141}
142 
143// Email queue with retry
144@Processor('email')
145export class EmailProcessor {
146  @Process('send')
147  async sendEmail(job: Job<SendEmailDto>): Promise<void> {
148    const { to, template, data } = job.data;
149 
150    try {
151      await this.mailer.send({
152        to,
153        template,
154        context: data,
155      });
156    } catch (error) {
157      // BullMQ will retry based on job options
158      throw error;
159    }
160  }
161}
162 
163// Usage
164@Injectable()
165export class NotificationService {
166  constructor(@InjectQueue('email') private emailQueue: Queue) {}
167 
168  async sendWelcome(user: User): Promise<void> {
169    await this.emailQueue.add(
170      'send',
171      {
172        to: user.email,
173        template: 'welcome',
174        data: { name: user.name },
175      },
176      {
177        attempts: 5,
178        backoff: { type: 'exponential', delay: 5000 },
179      },
180    );
181  }
182}
183 
184// Scheduled jobs
185@Injectable()
186export class ScheduledJobsService implements OnModuleInit {
187  constructor(@InjectQueue('maintenance') private queue: Queue) {}
188 
189  async onModuleInit(): Promise<void> {
190    // Clean up old reports daily at midnight
191    await this.queue.add(
192      'cleanup',
193      {},
194      {
195        repeat: { cron: '0 0 * * *' },
196        jobId: 'daily-cleanup', // Prevent duplicates
197      },
198    );
199 
200    // Send digest every hour
201    await this.queue.add(
202      'digest',
203      {},
204      {
205        repeat: { every: 60 * 60 * 1000 },
206        jobId: 'hourly-digest',
207      },
208    );
209  }
210}
211 
212@Processor('maintenance')
213export class MaintenanceProcessor {
214  @Process('cleanup')
215  async cleanup(): Promise<void> {
216    await this.cleanupOldReports();
217    await this.cleanupExpiredSessions();
218  }
219 
220  @Process('digest')
221  async sendDigest(): Promise<void> {
222    const users = await this.getUsersForDigest();
223    for (const user of users) {
224      await this.emailQueue.add('send', { to: user.email, template: 'digest' });
225    }
226  }
227}
228 
229// Queue monitoring with Bull Board
230import { BullBoardModule } from '@bull-board/nestjs';
231import { BullMQAdapter } from '@bull-board/api/bullMQAdapter';
232 
233@Module({
234  imports: [
235    BullBoardModule.forRoot({
236      route: '/admin/queues',
237      adapter: ExpressAdapter,
238    }),
239    BullBoardModule.forFeature({
240      name: 'email',
241      adapter: BullMQAdapter,
242    }),
243    BullBoardModule.forFeature({
244      name: 'reports',
245      adapter: BullMQAdapter,
246    }),
247  ],
248})
249export class AdminModule {}
250```
251 
252Reference: [NestJS Queues](https://docs.nestjs.com/techniques/queues)
253