Event-Driven Architecture with Redis and BullMQ

An eCommerce client's checkout took 6 seconds because the request handler sent the confirmation email, updated the search index, called the shipping API and posted to Slack before returning. When the shipping API had an outage, checkouts failed entirely. The fix was to make the request do one thing, write the order, and let everything else happen through a queue. Here is that system, built with BullMQ and Redis, and the rules that keep it reliable.

The shape

The request handler never talks to Redis directly. It writes an outbox row in the same transaction as the order; a relay process turns outbox rows into queue jobs. That is the transactional outbox pattern, and it is what guarantees you never send a confirmation email for an order whose transaction rolled back.

Setup

npm install bullmq ioredis
# Redis 6.2+ recommended. For local dev: docker run -p 6379:6379 redis:7-alpine

// lib/queues.ts
import { Queue, QueueEvents } from 'bullmq';
import IORedis from 'ioredis';

export const connection = new IORedis(process.env.REDIS_URL!, { maxRetriesPerRequest: null });

const defaultJobOptions = {
  attempts: 5,
  backoff: { type: 'exponential', delay: 3000 },     // 3s, 6s, 12s, 24s, 48s
  removeOnComplete: { age: 24 * 3600, count: 5000 },
  removeOnFail: false,                                // keep failures for the DLQ + inspection
};

export const emailQueue    = new Queue('email',    { connection, defaultJobOptions });
export const searchQueue   = new Queue('search',   { connection, defaultJobOptions });
export const shippingQueue = new Queue('shipping', { connection, defaultJobOptions: { ...defaultJobOptions, attempts: 8 } });
export const deadLetter    = new Queue('dlq',      { connection });

The outbox: publishing from the transaction

CREATE TABLE outbox (
  id          BIGSERIAL PRIMARY KEY,
  event_type  TEXT NOT NULL,             -- 'order.created'
  aggregate_id TEXT NOT NULL,            -- order id
  payload     JSONB NOT NULL,
  created_at  TIMESTAMPTZ NOT NULL DEFAULT now(),
  sent_at     TIMESTAMPTZ
);
CREATE INDEX outbox_unsent_idx ON outbox (created_at) WHERE sent_at IS NULL;   -- partial index: tiny

// modules/orders/service.ts (excerpt)
export async function createOrder(input: CreateOrderInput) {
  return withTransaction(async (tx) => {
    const order = await ordersRepo.insert(tx, input);
    await tx.query(
      `INSERT INTO outbox (event_type, aggregate_id, payload) VALUES ($1, $2, $3)`,
      ['order.created', order.id, JSON.stringify({ orderId: order.id, email: input.email, items: input.items })]
    );
    return order;                              // COMMIT happens in withTransaction
  });
}

// workers/outbox-relay.ts  -- one instance; polls every 500ms (or LISTEN/NOTIFY for lower latency)
const routes: Record<string, (p: any) => Promise<unknown>> = {
  'order.created': async (p) => Promise.all([
    emailQueue.add('order-confirmation', p, { jobId: `confirm:${p.orderId}` }),
    searchQueue.add('index-order', { orderId: p.orderId }, { jobId: `index:${p.orderId}` }),
    shippingQueue.add('create-shipment', p, { jobId: `ship:${p.orderId}`, delay: 60_000 }),   // give the customer a minute to cancel
  ]),
};

setInterval(async () => {
  const { rows } = await pool.query(`SELECT * FROM outbox WHERE sent_at IS NULL ORDER BY id LIMIT 100 FOR UPDATE SKIP LOCKED`);
  for (const row of rows) {
    await routes[row.event_type]?.(row.payload);
    await pool.query(`UPDATE outbox SET sent_at = now() WHERE id = $1`, [row.id]);
  }
}, 500);

The jobId on each add() is the deduplication key: if the relay crashes after enqueueing but before marking sent_at, the retry enqueues the same jobId and BullMQ ignores the duplicate. That is the first of two idempotency layers.

Workers: the second idempotency layer

// workers/email.worker.ts
import { Worker, type Job } from 'bullmq';
import { connection, deadLetter } from '@/lib/queues';

const worker = new Worker('email', async (job: Job) => {
  switch (job.name) {
    case 'order-confirmation': return sendOrderConfirmation(job.data);
    default: throw new Error(`unknown job ${job.name}`);
  }
}, {
  connection,
  concurrency: 5,
  limiter: { max: 50, duration: 1000 },           // respect the email provider's 50 req/s limit
});

async function sendOrderConfirmation({ orderId, email }: { orderId: string; email: string }) {
  // Idempotency: has this email already been sent? (retries and duplicates are normal)
  const [sent] = await sql`SELECT 1 FROM email_log WHERE order_id = ${orderId} AND kind = 'confirmation'`;
  if (sent) return { skipped: true };

  const order = await getOrderWithItems(orderId);
  if (!order) throw new UnrecoverableError(`order ${orderId} not found`);   // don't retry: it will never exist

  await mailer.send({ to: email, template: 'order-confirmation', data: order });
  await sql`INSERT INTO email_log (order_id, kind) VALUES (${orderId}, 'confirmation')`;
  return { sent: true };
}

worker.on('failed', async (job, err) => {
  if (!job) return;
  log.error({ jobId: job.id, attempts: job.attemptsMade, err: err.message }, 'email job failed');
  if (job.attemptsMade >= (job.opts.attempts ?? 1)) {
    await deadLetter.add(`email:${job.name}`, { original: job.data, error: err.message, failedAt: Date.now() }, { jobId: `dlq:${job.id}` });
  }
});

Three things to notice. The "already sent" check runs before the side effect and the log row is written after, so a crash between them causes at most one duplicate email, never zero emails. UnrecoverableError skips the remaining retries for errors that will never succeed. And the failed handler moves exhausted jobs to a dead-letter queue with the error attached, instead of leaving them in a failed state nobody looks at.

Dead-letter queue: the human loop

Jobs land in the DLQ for three reasons: a real bug, bad data, or an outage that outlasted the retries. Each needs a person. I run Bull Board behind the admin login so the team can see failures, read the error and click "replay". Replay just re-adds the original job to its source queue:

// admin/replay.ts
export async function replayDeadLetter(dlqJobId: string) {
  const job = await deadLetter.getJob(dlqJobId);
  if (!job) throw new Error('not found');
  const [queueName, jobName] = job.name.split(':');
  const queue = { email: emailQueue, search: searchQueue, shipping: shippingQueue }[queueName];
  await queue!.add(jobName, job.data.original, { jobId: `replay:${dlqJobId}:${Date.now()}` });
  await job.remove();
}

When the shipping API had its next outage, 212 jobs hit the DLQ over forty minutes. Once it recovered, replaying them took one click and ninety seconds. Checkouts never noticed.

Patterns that came up on real projects

Delayed and scheduled jobs

// Abandoned-cart email 2 hours after last activity; re-adding with the same jobId resets the timer
await emailQueue.add('abandoned-cart', { cartId }, { jobId: `abandon:${cartId}`, delay: 2 * 3600 * 1000 });
await emailQueue.remove(`abandon:${cartId}`);       // on checkout: cancel it

// Repeatable job: nightly report at 02:00 IST
await reportQueue.add('daily-sales', {}, { repeat: { pattern: '0 2 * * *', tz: 'Asia/Kolkata' } });

Parent-child flows

"Generate 40 product images, then build the catalogue PDF" is a parent job that completes only when its children do. BullMQ's FlowProducer models exactly that, and it is how I run the per-document steps in RAG ingestion.

Priorities and per-tenant fairness

One tenant importing 50,000 products should not block another tenant's password reset email. Separate queues per job class (already done above), and for shared queues use priority plus a per-tenant concurrency cap using BullMQ's group features or a Redis semaphore. The multi-tenant guide goes deeper on the fairness problem.

Operating it

• Metrics: queue depth, oldest waiting job age, failure rate, processing time p95, per queue. Alert on job age, not depth.
• Workers as separate processes, scaled independently of the API. In Docker Compose that is a second service sharing the image with a different command; in ECS it is a second task definition (see the deploy pipeline).
• Graceful shutdown: await worker.close() on SIGTERM waits for in-flight jobs. Without it, deploys create retries.
• Redis persistence: enable AOF (appendonly yes) or your queue disappears on a Redis restart. Managed Redis (ElastiCache, Upstash) handles this.
• Payload size: keep jobs under a few KB; pass IDs and let the worker load the data. Large payloads bloat Redis memory and slow every operation.

What it changed for the checkout

Queues are the second of the three building blocks in my system design fundamentals post, and the one that most changes how a product feels to use, because everything the user does becomes fast and everything that can fail becomes recoverable. If your checkout, signup or import flow is doing too much in the request, this is the refactor to plan.