# Job Queues: Background Task Processing

# Job Queues: Background Task Processing with Bull Redis Workers

## Problem

Modern applications need to handle long-running operations without blocking user requests. Sending emails, processing images, generating reports, or syncing data shouldn't delay API responses. Traditional synchronous processing creates bottlenecks, poor user experience, and system instability.

**Key Challenges:**
- Long operations block request handlers
- No retry mechanism for failed tasks
- Difficult to track job status
- No priority handling
- Resource exhaustion from concurrent operations
- Lost jobs on server crashes

## Solution

**Bull** is a Node.js library that provides a robust job queue system using Redis as the backing store. It enables asynchronous task processing with:

- **Persistent job storage** in Redis
- **Automatic retries** with exponential backoff
- **Job prioritization** and scheduling
- **Progress tracking** and status monitoring
- **Concurrency control** and rate limiting
- **Dead letter queues** for failed jobs
- **Event-driven architecture** with listeners

## Code Implementation

### 1. Basic Setup

```javascript
// npm install bull redis

const Queue = require('bull');
const redis = require('redis');

// Create a queue
const emailQueue = new Queue('emails', {
  redis: {
    host: 'localhost',
    port: 6379
  }
});

// Alternative: Using Redis connection
const redisClient = redis.createClient();
const taskQueue = new Queue('tasks', redisClient);

module.exports = emailQueue;
```

### 2. Producer: Adding Jobs

```javascript
const emailQueue = require('./queue');

// Simple job
async function sendWelcomeEmail(userId, email) {
  const job = await emailQueue.add(
    { userId, email },
    {
      attempts: 3,
      backoff: {
        type: 'exponential',
        delay: 2000
      },
      removeOnComplete: true
    }
  );
  
  console.log(`Job ${job.id} added to queue`);
  return job.id;
}

// Delayed job (send after 1 hour)
async function scheduleReminder(userId, message) {
  await emailQueue.add(
    { userId, message },
    {
      delay: 3600000, // 1 hour
      attempts: 2
    }
  );
}

// Prioritized job
async function sendUrgentNotification(userId, message) {
  await emailQueue.add(
    { userId, message },
    {
      priority: 1, // Higher priority = processed first
      attempts: 5
    }
  );
}

// Batch jobs
async function sendBulkEmails(recipients) {
  const jobs = recipients.map(recipient =>
    emailQueue.add(recipient, { attempts: 3 })
  );
  
  return Promise.all(jobs);
}

module.exports = {
  sendWelcomeEmail,
  scheduleReminder,
  sendUrgentNotification,
  sendBulkEmails
};
```

### 3. Consumer: Processing Jobs

```javascript
const emailQueue = require('./queue');
const nodemailer = require('nodemailer');

// Configure email service
const transporter = nodemailer.createTransport({
  service: 'gmail',
  auth: {
    user: process.env.EMAIL_USER,
    pass: process.env.EMAIL_PASS
  }
});

// Process jobs
emailQueue.process(5, async (job) => {
  console.log(`Processing job ${job.id}:`, job.data);
  
  try {
    const { userId, email, message } = job.data;
    
    // Update progress
    job.progress(25);
    
    // Validate email
    if (!email || !email.includes('@')) {
      throw new Error('Invalid email address');
    }
    
    job.progress(50);
    
    // Send email
    await transporter.sendMail({
      from: process.env.EMAIL_USER,
      to: email,
      subject: 'Welcome!',
      html: `<h1>Hello ${userId}</h1><p>${message}</p>`
    });
    
    job.progress(100);
    
    return { success: true, userId, email };
  } catch (error) {
    console.error(`Job ${job.id} failed:`, error.message);
    throw error; // Bull will retry
  }
});

// Event listeners
emailQueue.on('completed', (job, result) => {
  console.log(`✓ Job ${job.id} completed:`, result);
});

emailQueue.on('failed', (job, err) => {
  console.error(`✗ Job ${job.id} failed:`, err.message);
});

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

emailQueue.on('stalled', (job) => {
  console.warn(`Job ${job.id} stalled`);
});

module.exports = emailQueue;
```

### 4. Advanced: Multiple Queues

```javascript
const Queue = require('bull');

// Different queues for different tasks
const emailQueue = new Queue('emails');
const imageQueue = new Queue('images');
const reportQueue = new Queue('reports');

// Email processor
emailQueue.process(10, async (job) => {
  // Send email logic
  return { sent: true };
});

// Image processor (fewer concurrent jobs)
imageQueue.process(2, async (job) => {
  const { imageUrl, format } = job.data;
  // Process image
  return { processed: true };
});

// Report processor (scheduled)
reportQueue.process(1, async (job) => {
  const { reportType, userId } = job.data;
  // Generate report
  return { reportId: 'report-123' };
});

// Add jobs from API
const express = require('express');
const app = express();

app.post('/api/send-email', async (req, res) => {
  const { email, subject } = req.body;
  const jobId = await emailQueue.add({ email, subject });
  res.json({ jobId, status: 'queued' });
});

app.post('/api/process-image', async (req, res) => {
  const { imageUrl, format } = req.body;
  const jobId = await imageQueue.add({ imageUrl, format });
  res.json({ jobId, status: 'queued' });
});

app.get('/api/job/:id', async (req, res) => {
  const job = await emailQueue.getJob(req.params.id);
  if (!job) return res.status(404).json({ error: 'Job not found' });
  
  const state = await job.getState();
  const progress = job._progress;
  
  res.json({ id: job.id, state, progress });
});

app.listen(3000);
```

### 5. Error Handling & Dead Letter Queue

```javascript
const Queue = require('bull');

const taskQueue = new Queue('tasks');
const deadLetterQueue = new Queue('dead-letters');

taskQueue.process(async (job) => {
  const { retryCount = 0 } = job.data;
  
  try {
    // Simulate failure
    if (Math.random() < 0.3) {
      throw new Error('Random failure');
    }
    return { success: true };
  } catch (error) {
    // Move to dead letter after max retries
    if (job.attemptsMade >= job.opts.attempts) {
      await deadLetterQueue.add({
        originalJob: job.data,
        error: error.message,
        attempts: job.attemptsMade,
        timestamp: new Date()
      });
    }
    throw error;
  }
});

// Monitor dead letter queue
deadLetterQueue.process(async (job) => {
  console.error('Dead letter:', job.data);
  // Send alert, log to database, etc.
});

// Retry dead letter jobs manually
async function retryDeadLetterJob(jobId) {
  const job = await deadLetterQueue.getJob(jobId);
  if (job) {
    await taskQueue.add(job.data.originalJob, { attempts: 3 });
    await job.remove();
  }
}

module.exports = { taskQueue, deadLetterQueue, retryDeadLetterJob };
```

### 6. Monitoring & Dashboard

```javascript
const Queue = require('bull');
const express = require('express');
const { createBullBoard } = require('@bull-board/express');
const { BullAdapter } = require('@bull-board/api/bullAdapter');

const app = express();

// Create queues
const emailQueue = new Queue('emails');
const imageQueue = new Queue('images');
const reportQueue = new Queue('reports');

// Setup Bull Board
const serverAdapter = new express.Router();
createBullBoard({
  queues: [
    new BullAdapter(emailQueue),
    new BullAdapter(imageQueue),
    new BullAdapter(reportQueue)
  ],
  serverAdapter
});

app.use('/admin/queues', serverAdapter);

// Custom monitoring endpoint
app.get('/api/queue-stats', async (req, res) => {
  const stats = await Promise.all([
    emailQueue.getJobCounts(),
    imageQueue.getJobCounts(),
    reportQueue.getJobCounts()
  ]);
  
  res.json({
    email: stats[0],
    image: stats[1],
    report: stats[2]
  });
});

app.listen(3000);
```

## Tips & Best Practices

### 1. **Job Design**
```javascript
// ✓ Good: Idempotent, small payload
await queue.add({ userId: 123, action: 'send-email' });

// ✗ Bad: Large data, non-idempotent
await queue.add({ largeFile: buffer, timestamp: Date.now() });
```

### 2. **Concurrency Control**
```javascript
// Process 5 jobs concurrently
queue.process(5, handler);

// Or dynamic based on load
queue.process(async (job) => {
  // Handler
});
```

### 3. **Timeout Configuration**
```javascript
const job = await queue.add(data, {
  timeout: 30000 // 30 seconds max
});
```

### 4. **Graceful Shutdown**
```javascript
process.on('SIGTERM', async () => {
  await queue.close();
  process.exit(0);
});
```

### 5. **Testing**
```javascript
// Use bull-testing for unit tests
const { mockQueue } = require('bull-testing');

const queue = mockQueue();
await queue.add({ test: true });
const jobs = await queue.getJobs();
```

### 6. **Production Checklist**
- Use Redis persistence (AOF/RDB)
- Monitor queue depth and processing time
- Set appropriate retry strategies
- Implement circuit breakers for external APIs
- Use separate Redis instances for different queues
- Enable Redis clustering for high availability
- Log all job failures
- Set up alerts for stalled jobs

Bull Redis workers provide enterprise-grade job processing for Node.js applications, enabling scalable, reliable background task handling.