queue-job-processor

What this skill does

# 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

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

## Redis Connection

```typescript
// 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

```typescript
// 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

```typescript
// 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,
    },
  },
});
```

## Workers

### Email Worker

```typescript
// workers/email.worker.ts
import { Worker, Job } from 'bullmq';
import { redis } from '../lib/redis';
import { EmailJobData } from '../jobs/types';
import { sendEmail } from '../lib/email';

const emailWorker = new Worker<EmailJobData>(
  'email',
  async (job: Job<EmailJobData>) => {
    const { to, subject, template, variables } = job.data;

    console.log(`Processing email job ${job.id} to ${to}`);

    // Update progress
    await job.updateProgress(10);

    // Render template
    const html = await renderTemplate(template, variables);
    await job.updateProgress(50);

    // Send email
    const result = await sendEmail({
      to,
      subject,
      html,
    });

    await job.updateProgress(100);

    return { messageId: result.messageId, sentAt: new Date() };
  },
  {
    connection: redis,
    concurrency: 10, // Process 10 emails at a time
    limiter: {
      max: 100, // Max 100 jobs
      duration: 60000, // Per minute
    },
  }
);

// Event handlers
emailWorker.on('completed', (job, result) => {
  console.log(`Email job ${job.id} completed:`, result);
});

emailWorker.on('failed', (job, error) => {
  console.error(`Email job ${job?.id} failed:`, error);
});

emailWorker.on('progress', (job, progress) => {
  console.log(`Email job ${job.id} progress: ${progress}%`);
});

export { emailWorker };
```

### Image Processing Worker

```typescript
// workers/image.worker.ts
import { Worker, Job } from 'bullmq';
import { redis } from '../lib/redis';
import { ImageProcessingJobData } from '../jobs/types';
import sharp from 'sharp';
import { S3Client, PutObjectCommand } from '@aws-sdk/client-s3';

const s3 = new S3Client({ region: process.env.AWS_REGION });

const imageWorker = new Worker<ImageProcessingJobData>(
  'image-processing',
  async (job: Job<ImageProcessingJobData>) => {
    const { imageId, userId, operations } = job.data;

    console.log(`Processing image ${imageId} for user ${userId}`);

    // Download original image
    const originalBuffer = await downloadImage(imageId);
    let image = sharp(originalBuffer);

    // Apply operations
    for (let i = 0; i < operations.length; i++) {
      const op = operations[i];

      switch (op.type) {
        case 'resize':
          image = image.resize(op.params.width, op.params.height, {
            fit: op.params.fit || 'cover',
          });
          break;
        case 'crop':
          image = image.extract({
            left: op.params.left,
            top: op.params.top,
            width: op.params.width,
            height: op.params.height,
          });
          break;
        case 'watermark':
          image = image.composite([
            { input: op.params.watermarkPath, gravity: 'southeast' },
          ]);
          break;
      }

      await job.updateProgress(((i + 1) / operations.length) * 80);
    }

    // Convert and upload
    const processedBuffer = await image.webp({ quality: 85 }).toBuffer();

    const key = `processed/${userId}/${imageId}.webp`;
    await s3.send(
      new PutObjectCommand({
        Bucket: process.env.S3_BUCKET,
        Key: key,
        Body: processedBuffer,
        ContentType: 'image/webp',
      })
    );

    await job.updateProgress(100);

    return {
      url: `https://${process.env.S3_BUCKET}.s3.amazonaws.com/${key}`,
      size: processedBuffer.length,
    };
  },
  {
    connection: redis,
    concurrency: 5,
  }
);

imageWorker.on('failed', async (job, error) => {
  // Notify user of failure
  if (job) {
    await notifyUser(job.data.userId, {
      type: 'image-processing-failed',
      imageId: job.data.imageId,
      error: error.message,
    });
  }
});

export { imageWorker };
```

### Webhook Worker with Retries

```typescript
// workers/webhook.worker.ts
import { Worker, Job } from 'bullmq';
import { redis } from '../lib/redis';
import { WebhookJobData } from '../jobs/types';

const webhookWorker = new Worker<WebhookJobData>(
  'webhooks',
  async (job: Job<WebhookJobData>) => {
    const { url, payload, headers = {} } = job.data;

    const response = await fetch(url, {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        'X-Webhook-Signature': generateSignature(payload),
        ...headers,
      },
      body: JSON.stringify(payload),
      signal: AbortSignal.timeout(30000), // 30s timeout
    });

    if (!response.ok) {
      // Retry for server errors
      if (response.status >= 500) {
        throw new Error(`Webhook failed: ${response.status}`);
      }
      // Don't retry for client errors
      return {
        success: false,
        status: response.status,
        message: 'Client error, not retrying',
      };
    }

    return {
      success: true,
      status: response.status,
    };
  },
  {
    connection: redis,
    concurrency: 20,
  }
);

export { we