bullmq / ioredis

// src/redis.ts - single connection factory
import IORedis from "ioredis";
 
export function createRedis() {
  return new IORedis(process.env.REDIS_URL!, {
    maxRetriesPerRequest: null, // required for BullMQ workers
    enableReadyCheck: false,
  });
}

// src/queues/email.queue.ts
import { Queue } from "bullmq";
import { createRedis } from "../redis.js";
 
const connection = createRedis();
 
export const emailQueue = new Queue("email", {
  connection,
  defaultJobOptions: {
    attempts: 5,
    backoff: { type: "exponential", delay: 2000 },
    removeOnComplete: 1000,
    removeOnFail: false, // keep for DLQ inspection
  },
});
 
await emailQueue.add("send-welcome", { userId: "abc", template: "welcome" });

// src/workers/email.worker.ts
import { Worker } from "bullmq";
import { createRedis } from "../redis.js";
 
const worker = new Worker(
  "email",
  async (job) => {
    await sendEmail(job.data);
  },
  { connection: createRedis(), concurrency: 10 }
);
 
worker.on("failed", (job, err) => {
  console.error({ jobId: job?.id, err }, "email_job_failed");
});

// src/cache.ts
import type IORedis from "ioredis";
 
const CACHE_PREFIX = "cache:";
 
export async function cacheGet<T>(
  redis: IORedis,
  key: string
): Promise<T | null> {
  const raw = await redis.get(`${CACHE_PREFIX}${key}`);
  return raw ? (JSON.parse(raw) as T) : null;
}
 
export async function cacheSet(
  redis: IORedis,
  key: string,
  value: unknown,
  ttlSeconds: number
): Promise<void> {
  await redis.set(
    `${CACHE_PREFIX}${key}`,
    JSON.stringify(value),
    "EX",
    ttlSeconds
  );
}

// HTTP handler: cache-aside pattern
import { createRedis } from "./redis.js";
import { cacheGet, cacheSet } from "./cache.js";
 
const redis = createRedis();
 
app.get("/products/:id", async (req, reply) => {
  const { id } = req.params;
  const cached = await cacheGet<Product>(redis, `product:${id}`);
  if (cached) return cached;
 
  const product = await db.findProduct(id);
  await cacheSet(redis, `product:${id}`, product, 300);
  return product;
});

// Dead letter after max attempts
import { Queue } from "bullmq";
 
export const dlq = new Queue("email-dlq", { connection: createRedis() });
 
worker.on("failed", async (job, err) => {
  if (job && job.attemptsMade >= (job.opts.attempts ?? 1)) {
    await dlq.add("failed", { original: job.data, error: err.message });
  }
});