Background Jobs and Message Queues: BullMQ, Redis, and Resilient Workers

[ HTTP API Handler ]
                 |
                 v (queue.add)
     +-----------------------+
     |      BullMQ Queue     |
     +-----------------------+
                 |
                 v (Atomic Lua Script)
  +-------------------------------------+
  |             Redis State             |
  |  - zset: delayed (score: timestamp) |
  |  - list: wait (FIFO / LIFO)         |
  |  - hash: job metadata               |
  +-------------------------------------+
                 |
                 | Worker acquires lock
                 v (LMOVE + SET lock NX PX)
     +-----------------------+
     |    BullMQ Worker      | <----+ Heartbeat Lock Renewal
     | (concurrency: N)      |      | (lockRenewTime: 15s)
     +-----------------------+      |
                 |                  |
        Is Event Loop blocked? -----+
        Worker crashed (OOM)?
                 |
                 v Lock Expired (30s)
  +-------------------------------------+
  |     Stalled Job Recovery Loop       |
  |  Moves job back to 'wait' or 'fail' |
  +-------------------------------------+

+------------+
             |  Delayed   | (Scheduled timestamp reached)
             +------------+
                   |
                   v
             +------------+       Worker fetches
             |    Wait    | ------------------------+
             +------------+                         |
                   ^                                v
                   | Retry / Stalled          +------------+
                   +------------------------- |   Active   |
                                              +------------+
                                               /          \
                             Success Result   /            \   Failure / Max Attempts
                                             v              v
                                      +-----------+    +-----------+
                                      | Completed |    |  Failed   |
                                      +-----------+    +-----------+

import { Queue, Worker, QueueEvents, Job } from "bullmq";
import Redis from "ioredis";

export interface PaymentWebhookJobData {
  transactionId: string;
  amount: number;
  currency: string;
  source: string;
  timestamp: number;
}

export interface PaymentJobResult {
  processed: boolean;
  ledgerRecordId: string;
}

const redisConnection = new Redis(process.env.REDIS_URL || "redis://127.0.0.1:6379", {
  maxRetriesPerRequest: null, // MANDATORY requirement for BullMQ operation!
  enableReadyCheck: false,
});

export const PAYMENT_QUEUE_NAME = "payment-webhooks";

/**
 * Queue Producer Initialization
 */
export const paymentQueue = new Queue<PaymentWebhookJobData, PaymentJobResult>(
  PAYMENT_QUEUE_NAME,
  {
    connection: redisConnection,
    defaultJobOptions: {
      attempts: 5, // Up to 5 retries on failure
      backoff: {
        type: "exponential",
        delay: 2000, // Initial delay 2s, then 4s, 8s, 16s...
      },
      removeOnComplete: {
        age: 86400, // Keep completed jobs in Redis for 24 hours
        count: 5000, // Retain a maximum of 5,000 completed records
      },
      removeOnFail: {
        age: 604800, // Retain failed jobs for 7 days for audit and dead-letter review
      },
    },
  }
);

/**
 * Resilient Worker Consumer with Stalled Job Protection
 */
export function createPaymentWorker(): Worker<PaymentWebhookJobData, PaymentJobResult> {
  const worker = new Worker<PaymentWebhookJobData, PaymentJobResult>(
    PAYMENT_QUEUE_NAME,
    async (job: Job<PaymentWebhookJobData, PaymentJobResult>): Promise<PaymentJobResult> => {
      // Broadcast execution progress for live dashboard monitoring
      await job.updateProgress(10);

      if (job.data.amount <= 0) {
        throw new Error(`Invalid transaction amount: ${job.data.amount}`);
      }

      await job.updateProgress(50);

      // Business logic execution
      const ledgerRecordId = `led_${job.data.transactionId}`;
      await job.updateProgress(100);

      return {
        processed: true,
        ledgerRecordId,
      };
    },
    {
      connection: redisConnection,
      concurrency: 10, // Up to 10 concurrent jobs per worker process
      lockDuration: 30000, // Lock lease duration (30 seconds)
      lockRenewTime: 10000, // Renew lock lease every 10 seconds (Watchdog)
      maxStalledCount: 2, // Maximum 2 stalls before permanently marking as Failed
      stalledInterval: 30000, // Interval for scanning stalled jobs
    }
  );

  // Critical telemetry and lifecycle hooks
  worker.on("completed", (job: Job) => {
    console.info(`[Worker] Job ${job.id} completed successfully`);
  });

  worker.on("failed", (job: Job | undefined, err: Error) => {
    console.error(`[Worker] Job ${job?.id} failed on attempt ${job?.attemptsMade}: ${err.message}`);
  });

  worker.on("error", (err: Error) => {
    console.error("[Worker] Global worker error:", err.message);
  });

  worker.on("stalled", (jobId: string) => {
    console.warn(`[Worker] Job ${jobId} was marked as STALLED (lock expired)`);
  });

  return worker;
}

import { FlowProducer, FlowJob } from "bullmq";
import Redis from "ioredis";

const redis = new Redis(process.env.REDIS_URL || "redis://127.0.0.1:6379", {
  maxRetriesPerRequest: null,
});

export const flowProducer = new FlowProducer({ connection: redis });

/**
 * Creates a hierarchical dependency pipeline:
 * The parent aggregation job waits for two parallel data extraction jobs to resolve
 */
export async function createReportPipeline(reportId: string, companyId: string) {
  const flow: FlowJob = {
    name: "aggregate-final-report",
    queueName: "report-aggregator",
    data: { reportId, companyId },
    children: [
      {
        name: "extract-sales-data",
        queueName: "data-extraction",
        data: { reportId, source: "sales" },
        opts: { attempts: 3 },
      },
      {
        name: "extract-expenses-data",
        queueName: "data-extraction",
        data: { reportId, source: "expenses" },
        opts: { attempts: 3 },
      },
    ],
  };

  const tree = await flowProducer.add(flow);
  return tree.job.id;
}