Background job processing

import { Queue, JobsOptions } from 'bullmq';
import { v4 as uuidv4 } from 'uuid';

// Define strict types for job data
export interface EmailJobData {
  userId: string;
  template: string;
  idempotencyKey: string;
  metadata: Record<string, unknown>;
}

const emailQueue = new Queue<EmailJobData>('email-processing', {
  connection: {
    host: process.env.REDIS_HOST || 'localhost',
    port: Number(process.env.REDIS_PORT) || 6379,
    // Production: Enable TLS and auth
  },
  defaultJobOptions: {
    attempts: 5,
    backoff: {
      type: 'exponential',
      delay: 2000,
    },
    removeOnComplete: 100, // Keep last 100 for debugging
    removeOnFail: false,   // Retain failed jobs for DLQ analysis
  },
});

export async function enqueueEmail(userId: string, template: string): Promise<string> {
  // Generate idempotency key to prevent duplicates
  const idempotencyKey = `${userId}-${template}-${Date.now()}`;
  
  const jobOptions: JobsOptions = {
    jobId: idempotencyKey, // BullMQ uses jobId for deduplication if set
    priority: 1,
  };

  const job = await emailQueue.add(
    'send-welcome',
    { userId, template, idempotencyKey, metadata: {} },
    jobOptions
  );

  return job.id!;
}

Worker Implementation

Workers must handle errors gracefully, implement circuit breakers for external dependencies, and log structured data.

import { Worker, Job, FailedError } from 'bullmq';
import { EmailJobData } from './producer';
import { sendEmail } from './email-service';

const worker = new Worker<EmailJobData>(
  'email-processing',
  async (job: Job<EmailJobData>) => {
    const { userId, template, idempotencyKey } = job.data;

    // Idempotency check: Verify if job was already processed
    // This is a fallback; jobId deduplication handles most cases
    const isProcessed = await checkProcessedStatus(idempotencyKey);
    if (isProcessed) {
      return; 
    }

    try {
      // Execute business logic
      await sendEmail({ to: userId, template });
      
      // Mark as processed
      await markProcessed(idempotencyKey);
      
      // Update job progress
      job.updateProgress(100);
    } catch (error) {
      // Distinguish between transient and permanent failures
      if (isPermanentError(error)) {
        throw new FailedError('Permanent failure: ' + error.message);
      }
      // Transient errors trigger retry based on backoff config
      throw error;
    }
  },
  {
    connection: { host: 'localhost', port: 6379 },
    concurrency: 10, // Limit concurrent executions
    limiter: {
      max: 10,
      duration: 1000, // Rate limiting: 10 jobs per second
    },
  }
);

// Event listeners for observability
worker.on('failed', (job, err) => {
  console.error(`Job ${job?.id} failed: ${err.message}`, {
    attempt: job?.attemptsMade,
    queue: job?.queueName,
  });
});

worker.on('stalled', (jobId) => {
  console.warn(`Job ${jobId} stalled. Check for event loop blocking.`);
});

3. Transactional Outbox Pattern

For jobs triggered by database transactions, use the Outbox pattern to guarantee consistency.

1. Application writes data and a message to an outbox table in the same DB transaction.
2. A separate process polls the outbox table or uses CDC (Change Data Capture) to publish messages to the job queue.
3. This ensures a job is never enqueued unless the associated data is committed.

Pitfall Guide

1. Non-Idempotent Operations

Mistake: Jobs that modify state without checking current state, leading to duplicate charges or emails on retry. Fix: Always use jobId for deduplication in the queue and implement idempotency checks within the worker logic. External APIs must support idempotency keys.

2. Retry Storms

Mistake: Linear backoff or no backoff causes the system to hammer failing dependencies, cascading failures. Fix: Use exponential backoff with jitter. Implement circuit breakers for downstream services. If a dependency is down, pause the queue or reduce concurrency.

3. Event Loop Blocking

Mistake: Workers performing CPU-intensive tasks or synchronous I/O block the Node.js event loop, stalling all concurrent jobs. Fix: Offload CPU tasks to worker threads or separate services. Use setTimeout or setImmediate to yield control in long-running loops. Monitor stalled events.

4. Unbounded Payload Size

Mistake: Storing large objects in job payloads bloats Redis memory and slows serialization. Fix: Keep payloads minimal. Pass references (IDs) to data stored in a database or object store. Enforce a payload size limit (e.g., < 10KB).

5. Silent Failures

Mistake: Jobs failing without alerts, leading to data gaps. Fix: Implement DLQ monitoring. Alert when DLQ depth exceeds a threshold. Use structured logging with jobId and traceId for distributed tracing.

6. Graceful Shutdown Neglect

Mistake: Workers killed while processing jobs, causing jobs to be lost or processed twice. Fix: Implement SIGTERM handlers. Call worker.close() to finish current jobs before exiting. Ensure the broker recognizes the disconnect and requeues in-progress jobs.

7. Security in Queues

Mistake: Storing PII or secrets in job data visible in Redis. Fix: Encrypt sensitive fields before serialization. Use RBAC for Redis access. Avoid logging full job payloads in error traces.

Production Bundle

Action Checklist

• Define Idempotency Strategy: Ensure every job has a unique jobId and the worker logic verifies processing status.
• Configure Exponential Backoff: Set attempts and backoff in queue options; avoid linear retries.
• Implement DLQ Monitoring: Set up alerts for failed jobs and DLQ depth thresholds.
• Limit Payload Size: Refactor jobs to pass IDs rather than full objects; enforce size limits.
• Add Graceful Shutdown: Handle SIGTERM/SIGINT to call worker.close() and drain in-flight jobs.
• Enable Observability: Integrate metrics for queue depth, processing latency, and failure rates into your monitoring stack.
• Test Failure Scenarios: Simulate Redis outages, worker crashes, and dependency failures in staging.
• Review Concurrency Settings: Tune concurrency and rate limits based on downstream dependency capacity.

Decision Matrix

Scenario	Recommended Approach	Why	Cost Impact
Low Volume, Simple Tasks	In-Memory with Persistence	Simplicity; sufficient for < 1k jobs/hr.	Low infra, high risk on crash.
Standard Web App	Redis + BullMQ	Best DX, durability, and retry logic for TS stack.	Moderate ($45/mo).
High Throughput / Analytics	Kafka / RabbitMQ	Partitioning, massive scale, strict ordering.	High ($120+/mo), complex ops.
Transactional Consistency	Outbox Pattern + CDC	Guarantees job creation matches DB commit.	Adds infra complexity (CDC).

Configuration Template

Copy this template for a production-ready BullMQ worker setup.

// worker.config.ts
import { Queue, Worker, QueueEvents } from 'bullmq';

export const QUEUE_CONFIG = {
  connection: {
    host: process.env.REDIS_HOST!,
    port: parseInt(process.env.REDIS_PORT || '6379'),
    password: process.env.REDIS_PASSWORD,
    tls: process.env.NODE_ENV === 'production' ? {} : undefined,
    retryStrategy: (times: number) => Math.min(times * 50, 2000),
  },
  defaultJobOptions: {
    attempts: 3,
    backoff: { type: 'exponential', delay: 1000 },
    removeOnComplete: { age: 3600, count: 500 },
    removeOnFail: { age: 24 * 3600 }, // Keep failed for 24h
  },
};

export const WORKER_OPTIONS = {
  concurrency: parseInt(process.env.WORKER_CONCURRENCY || '10'),
  limiter: {
    max: parseInt(process.env.WORKER_RATE_LIMIT || '20'),
    duration: 1000,
  },
  lockDuration: 30000, // 30s lock, renew if job takes longer
  lockRenewTime: 15000,
};

// Queue Events for monitoring
export const queueEvents = new QueueEvents('my-queue', QUEUE_CONFIG);

queueEvents.on('failed', ({ jobId, failedReason }) => {
  // Send to Sentry/Datadog
  console.error(`Job ${jobId} failed: ${failedReason}`);
});

Quick Start Guide

1. Install Dependencies:

   npm install bullmq ioredis uuid
   npm install -D @types/redis
   

2. Start Redis: Ensure Redis is running with persistence enabled.

   docker run -d -p 6379:6379 redis:7-alpine redis-server --appendonly yes
   

3. Define and Run Worker: Create worker.ts using the Core Solution code. Run with:

   ts-node worker.ts
   

4. Enqueue a Job: From your application, call the producer function. Verify the job appears in Redis and is processed by the worker. Check logs for success or failure handling.

5. Verify Observability: Use bull-board or a custom dashboard to monitor queue metrics. Ensure DLQ captures failed jobs correctly.