const express = require('express');
const { enqueueEmail } = require('./emailQueue');
const router = express.Router();
router.post('/users/:id/welcome-email', async (req, res, next) => {
try {
const userId = req.params.id;
const { email } = req.body;
if (!email) {
return res.status(422).json({ error: 'email is required' });
}
const job = await enqueueEmail({
type: 'welcome',
to: email,
userId,
dedupeKey: `welcome:${userId}`,
});
return res.status(202).json({ queued: true, jobId: job.id });
} catch (err) {
return next(err);
}
});
module.exports = router;
const { Queue } = require('bullmq');
const IORedis = require('ioredis');
const connection = new IORedis(process.env.REDIS_URL, {
maxRetriesPerRequest: null,
});
const emailQueue = new Queue('emails', {
connection,
defaultJobOptions: {
attempts: 5,
backoff: { type: 'exponential', delay: 2000 },
removeOnComplete: { age: 3600, count: 1000 },
removeOnFail: { age: 24 * 3600 },
},
});
async function enqueueEmail(data) {
const { dedupeKey } = data;
return emailQueue.add(data.type, data, {
jobId: dedupeKey, // duplicate ids are silently ignored by BullMQ
});
}
module.exports = { emailQueue, enqueueEmail, connection };
const { Worker } = require('bullmq');
const { connection } = require('./emailQueue');
const mailer = require('./mailer');
async function handleJob(job) {
const { type, to, userId } = job.data;
switch (type) {
case 'welcome':
return mailer.send({
to,
template: 'welcome',
vars: { userId },
});
case 'password_reset':
return mailer.send({ to, template: 'password_reset' });
default:
throw new Error(`Unknown email type: ${type}`);
}
}
const worker = new Worker('emails', handleJob, {
connection,
concurrency: 10,
});
worker.on('completed', (job) => {
console.log(`sent ${job.data.type} to ${job.data.to}`);
});
worker.on('failed', (job, err) => {
const exhausted = job && job.attemptsMade >= job.opts.attempts;
console.error(
`email ${job && job.id} failed (attempt ${job && job.attemptsMade}` +
`${exhausted ? ', giving up' : ''}):`,
err.message
);
});
process.on('SIGTERM', async () => {
await worker.close();
process.exit(0);
});
This snippet shows how a durable background-email pipeline is wired together with BullMQ on top of Redis, split across the three pieces that actually collaborate in production: a shared queue definition, the enqueue path used by request handlers, and the standalone worker process that drains the queue.
In emailQueue.js the queue is created once and exported so every part of the app shares a single BullMQ Queue instance backed by one Redis connection. The defaultJobOptions encode the reliability policy in one place: attempts with backoff set to exponential means a transient SMTP failure is retried on a widening schedule rather than dropped, while removeOnComplete keeps Redis from growing unbounded and removeOnFail retains recent failures for inspection. enqueueEmail wraps queue.add and passes a stable jobId derived from the caller's dedupeKey, which is the core idempotency trick — BullMQ refuses to insert a second job with the same id, so a retried HTTP request or an at-least-once upstream event cannot enqueue the same welcome email twice.
In emailController.js the Express handler does the minimum synchronous work: it validates input, then calls enqueueEmail and returns 202 Accepted immediately. The request never waits on the mail server, so latency stays low and a slow provider cannot tie up web workers. The dedupeKey is built from the user id and email type so repeated clicks collapse to one job.
emailWorker.js is a separate long-lived process, not part of the web server. The Worker binds to the same queue name and Redis, and its concurrency option controls how many messages are processed in parallel per worker — horizontal scale comes from running more worker processes. The processor switches on job.data.type, sends through a mailer, and lets thrown errors propagate so BullMQ can apply the retry policy; a handled error would wrongly mark the job complete. The failed and completed listeners provide observability, and job.attemptsMade distinguishes a first failure from an exhausted one.
The main trade-off is operational: a second process and a Redis dependency in exchange for decoupled, retryable, deduplicated delivery. This pattern is the right reach whenever a request triggers slow, failure-prone I/O that should not block the response.
Related snips
export interface RetryOptions {
retries: number;
baseMs: number;
maxMs: number;
signal?: AbortSignal;
onRetry?: (attempt: number, delay: number, err: unknown) => void;Exponential backoff with jitter for retries
package dbutil
import (
"context"
"github.com/jackc/pgconn"Retry Postgres serialization failures with bounded attempts
require "csv"
class PeopleCsvStream
include Enumerable
HEADERS = %w[id full_name email signed_up_at plan].freezeResilient CSV Export as a Streamed Response
use crossbeam::channel::unbounded;
use std::thread;
fn main() {
let (tx, rx) = unbounded();Crossbeam for advanced concurrent data structures
// Creating a Promise
const myPromise = new Promise((resolve, reject) => {
const success = true;
setTimeout(() => {
if (success) {Promises and async/await patterns for asynchronous JavaScript
use std::sync::mpsc;
use std::thread;
fn main() {
let (tx, rx) = mpsc::channel();Channels (mpsc) for message passing between threads
Share this code
Here's the card — post it anywhere.
