BullMQ

import { Queue, Worker } from "bullmq";
import IORedis from "ioredis";
 
const connection = new IORedis(process.env.REDIS_URL!, { maxRetriesPerRequest: null });
 
export const reportQueue = new Queue("reports", { connection });
 
new Worker(
  "reports",
  async (job) => {
    await buildReport(job.data.reportId);
  },
  { connection: connection.duplicate(), concurrency: 3 }
);

// src/queues/connection.ts
import IORedis from "ioredis";
 
export function createBullConnection() {
  return new IORedis(process.env.REDIS_URL!, {
    maxRetriesPerRequest: null,
    enableReadyCheck: false,
  });
}
 
// src/queues/reports.ts
import { Queue, Worker, QueueEvents } from "bullmq";
import { createBullConnection } from "./connection";
import { buildReport } from "../services/reports";
 
const connection = createBullConnection();
export const reportsQueue = new Queue("reports", { connection });
 
export function startReportWorker() {
  const worker = new Worker(
    "reports",
    async (job) => {
      await job.updateProgress(10);
      const url = await buildReport(job.data.reportId);
      await job.updateProgress(100);
      return { url };
    },
    {
      connection: createBullConnection(),
      concurrency: 2,
      lockDuration: 30_000,
    }
  );
 
  worker.on("failed", (job, err) => {
    console.error({ msg: "job_failed", jobId: job?.id, err: err.message });
  });
 
  worker.on("stalled", (jobId) => {
    console.warn({ msg: "job_stalled", jobId });
  });
 
  return worker;
}
 
// src/api/reports-route.ts
import express from "express";
import { reportsQueue } from "../queues/reports";
 
const router = express.Router();
 
router.post("/", async (req, res) => {
  const job = await reportsQueue.add(
    "build",
    { reportId: req.body.reportId },
    { attempts: 3, backoff: { type: "exponential", delay: 1000 } }
  );
  res.status(202).json({ jobId: job.id });
});
 
export default router;

const worker = startReportWorker();
 
process.on("SIGTERM", async () => {
  await worker.close(); // waits for current job by default
  process.exit(0);
});

await reportsQueue.add(
  "nightly",
  {},
  { repeat: { pattern: "0 2 * * *" }, jobId: "nightly-report" }
);

import { FlowProducer } from "bullmq";
 
const flow = new FlowProducer({ connection });
await flow.add({
  name: "invoice-pdf",
  queueName: "pdf",
  data: { invoiceId },
  children: [{ name: "fetch-line-items", queueName: "data", data: { invoiceId } }],
});