queue-job-processor

Queue Job Processor

Build robust background job processing with BullMQ and Redis.

Core Workflow

1. Setup Redis: Configure connection
2. Create queues: Define job queues
3. Implement workers: Process jobs
4. Add job types: Type-safe job definitions
5. Configure retries: Handle failures
6. Add monitoring: Dashboard and alerts

Installation

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

Redis Connection

// lib/redis.ts
import IORedis from 'ioredis';

export const redis = new IORedis(process.env.REDIS_URL!, {
  maxRetriesPerRequest: null, // Required for BullMQ
  enableReadyCheck: false,
});

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

Queue Setup

Define Job Types

// jobs/types.ts
export interface EmailJobData {
  to: string;
  subject: string;
  template: string;
  variables: Record<string, string>;
}

export interface ImageProcessingJobData {
  imageId: string;
  userId: string;
  operations: Array<{
    type: 'resize' | 'crop' | 'watermark';
    params: Record<string, any>;
  }>;
}

export interface ReportJobData {
  reportId: string;
  userId: string;
  type: 'daily' | 'weekly' | 'monthly';
  dateRange: {
    start: string;
    end: string;
  };
}

export interface WebhookJobData {
  url: string;
  payload: Record<string, any>;
  headers?: Record<string, string>;
  retryCount?: number;
}

export type JobData =
  | { type: 'email'; data: EmailJobData }
  | { type: 'image-processing'; data: ImageProcessingJobData }
  | { type: 'report'; data: ReportJobData }
  | { type: 'webhook'; data: WebhookJobData };

Create Queues

// queues/index.ts
import { Queue, QueueOptions } from 'bullmq';
import { redis } from '../lib/redis';
import {
  EmailJobData,
  ImageProcessingJobData,
  ReportJobData,
  WebhookJobData,
} from './types';

const defaultOptions: QueueOptions = {
  connection: redis,
  defaultJobOptions: {
    attempts: 3,
    backoff: {
      type: 'exponential',
      delay: 1000,
    },
    removeOnComplete: {
      count: 1000, // Keep last 1000 completed jobs
      age: 24 * 3600, // Keep for 24 hours
    },
    removeOnFail: {
      count: 5000, // Keep last 5000 failed jobs
    },
  },
};

export const emailQueue = new Queue<EmailJobData>('email', defaultOptions);

export const imageQueue = new Queue<ImageProcessingJobData>('image-processing', {
  ...defaultOptions,
  defaultJobOptions: {
    ...defaultOptions.defaultJobOptions,
    attempts: 5,
    timeout: 5 * 60 * 1000, // 5 minutes
  },
});

export const reportQueue = new Queue<ReportJobData>('reports', {
  ...defaultOptions,
  defaultJobOptions: {
    ...defaultOptions.defaultJobOptions,
    timeout: 30 * 60 * 1000, // 30 minutes
  },
});

export const webhookQueue = new Queue<WebhookJobData>('webhooks', {
  ...defaultOptions,
  defaultJobOptions: {
    ...defaultOptions.defaultJobOptions,
    attempts: 5,
    backoff: {
      type: 'exponential',
      delay: 5000,
    },
  },
});