Task Queues: Bull Redis Job Processing - Async Task Handling

Problem

Modern applications require handling long-running operations without blocking user requests. Common challenges include:

• Blocking Operations: Heavy computations, file uploads, email sending block HTTP responses
• Reliability: Failed tasks disappear without retry mechanisms
• Scalability: Single-threaded servers can't handle concurrent heavy workloads
• Monitoring: No visibility into job status, progress, or failures
• Concurrency Control: Managing parallel job execution limits

Solution

Bull is a Node.js library that uses Redis as a message broker for job queues. It provides:

• Persistent Job Storage: Redis-backed queue survives crashes
• Automatic Retries: Configurable retry logic with exponential backoff
• Job Scheduling: Delayed jobs, recurring tasks, cron patterns
• Progress Tracking: Real-time job progress updates
• Concurrency Management: Control parallel job execution
• Dead Letter Queues: Handle permanently failed jobs
• Event-Driven Architecture: Listen to job lifecycle events

---

Code Implementation

1. Basic Setup & Installation

npm install bull redis

2. Queue Definition & Producer

// queues/emailQueue.js
const Queue = require('bull');
const redis = require('redis');

// Create queue instance
const emailQueue = new Queue('email-notifications', {
  redis: {
    host: process.env.REDIS_HOST || 'localhost',
    port: process.env.REDIS_PORT || 6379,
  },
});

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

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

emailQueue.on('error', (err) => {
  console.error('Queue error:', err);
});

module.exports = emailQueue;

3. Producer: Adding Jobs to Queue

// services/emailService.js
const emailQueue = require('../queues/emailQueue');

class EmailService {
  // Add job immediately
  static async sendWelcomeEmail(userId, email) {
    const job = await emailQueue.add(
      {
        userId,
        email,
        template: 'welcome',
      },
      {
        attempts: 3, // Retry 3 times
        backoff: {
          type: 'exponential',
          delay: 2000, // Start with 2s, exponentially increase
        },
        removeOnComplete: true, // Clean up after success
        removeOnFail: false, // Keep failed jobs for debugging
      }
    );

    console.log(`Email job queued: ${job.id}`);
    return job.id;
  }

  // Schedule job for later
  static async scheduleNewsletterEmail(userId, email, sendAt) {
    const job = await emailQueue.add(
      {
        userId,
        email,
        template: 'newsletter',
      },
      {
        delay: sendAt.getTime() - Date.now(),
        attempts: 5,
        backoff: {
          type: 'fixed',
          delay: 5000,
        },
      }
    );

    return job.id;
  }

  // Recurring job (cron)
  static async setupDailyDigest() {
    await emailQueue.add(
      { template: 'daily-digest' },
      {
        repeat: {
          cron: '0 9 * * *', // 9 AM daily
          tz: 'America/New_York',
        },
      }
    );
  }

  // Bulk add jobs
  static async sendBulkEmails(recipients) {
    const jobs = recipients.map((recipient) => ({
      name: 'send-email',
      data: {
        email: recipient.email,
        userId: recipient.id,
        template: 'promotional',
      },
      opts: {
        attempts: 3,
        backoff: { type: 'exponential', delay: 2000 },
      },
    }));

    const addedJobs = await emailQueue.addBulk(jobs);
    return addedJobs.map((job) => job.id);
  }
}

module.exports = EmailService;

4. Consumer: Processing Jobs

// workers/emailWorker.js
const emailQueue = require('../queues/emailQueue');
const nodemailer = require('nodemailer');

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

// Email templates
const templates = {
  welcome: (email) => ({
    subject: 'Welcome!',
    html: `<h1>Welcome ${email}</h1><p>Thanks for joining us!</p>`,
  }),
  newsletter: (email) => ({
    subject: 'Weekly Newsletter',
    html: `<h1>This Week's Updates</h1>...`,
  }),
  'daily-digest': () => ({
    subject: 'Your Daily Digest',
    html: `<h1>Today's Summary</h1>...`,
  }),
};

// Process jobs
emailQueue.process(5, async (job) => {
  const { email, template, userId } = job.data;

  try {
    // Update progress
    job.progress(25);

    // Validate email
    if (!email || !email.includes('@')) {
      throw new Error('Invalid email address');
    }

    job.progress(50);

    // Get template
    const emailContent = templates[template](email);

    job.progress(75);

    // Send email
    await transporter.sendMail({
      from: process.env.EMAIL_FROM,
      to: email,
      ...emailContent,
    });

    job.progress(100);

    // Return result
    return {
      success: true,
      email,
      userId,
      sentAt: new Date(),
    };
  } catch (error) {
    console.error(`Failed to send email to ${email}:`, error);
    throw error; // Bull will handle retry
  }
});

// Handle stalled jobs (jobs that didn't complete in time)
emailQueue.on('stalled', (job) => {
  console.warn(`Job ${job.id} stalled, will be retried`);
});

console.log('Email worker started, processing up to 5 concurrent jobs');

5. Express API Integration

// routes/emailRoutes.js
const express = require('express');
const router = express.Router();
const EmailService = require('../services/emailService');
const emailQueue = require('../queues/emailQueue');

// Send email
router.post('/send', async (req, res) => {
  try {
    const { userId, email } = req.body;
    const jobId = await EmailService.sendWelcomeEmail(userId, email);

    res.json({
      success: true,
      jobId,
      message: 'Email queued for sending',
    });
  } catch (error) {
    res.status(500).json({ error: error.message });
  }
});

// Get job status
router.get('/job/:jobId', async (req, res) => {
  try {
    const job = await emailQueue.getJob(req.params.jobId);

    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,
      data: job.data,
      attempts: job.attemptsMade,
      failedReason: job.failedReason,
    });
  } catch (error) {
    res.status(500).json({ error: error.message });
  }
});

// Get queue stats
router.get('/stats', async (req, res) => {
  try {
    const counts = await emailQueue.getJobCounts();
    const workers = emailQueue.workers.length;

    res.json({
      workers,
      ...counts,
    });
  } catch (error) {
    res.status(500).json({ error: error.message });
  }
});

// Retry failed job
router.post('/job/:jobId/retry', async (req, res) => {
  try {
    const job = await emailQueue.getJob(req.params.jobId);

    if (!job) {
      return res.status(404).json({ error: 'Job not found' });
    }

    await job.retry();
    res.json({ success: true, message: 'Job queued for retry' });
  } catch (error) {
    res.status(500).json({ error: error.message });
  }
});

module.exports = router;

6. Advanced: Multiple Queues & Processors

// queues/index.js
const Queue = require('bull');

const queues = {
  email: new Queue('email', { redis: { host: 'localhost' } }),
  image: new Queue('image-processing', { redis: { host: 'localhost' } }),
  report: new Queue('report-generation', { redis: { host: 'localhost' } }),
};

module.exports = queues;

// workers/imageWorker.js
const queues = require('../queues');
const sharp = require('sharp');
const fs = require('fs').promises;

queues.image.process(2, async (job) => {
  const { inputPath, outputPath, width, height } = job.data;

  try {
    job.progress(10);

    const image = sharp(inputPath);
    job.progress(40);

    await image
      .resize(width, height, { fit: 'cover' })
      .toFile(outputPath);

    job.progress(90);

    const stats = await fs.stat(outputPath);

    return {
      success: true,
      outputPath,
      size: stats.size,
    };
  } catch (error) {
    throw new Error(`Image processing failed: ${error.message}`);
  }
});

7. Main Application Entry Point

// app.js
const express = require('express');
const emailRoutes = require('./routes/emailRoutes');
const EmailService = require('./services/emailService');

const app = express();
app.use(express.json());

// Routes
app.use('/api/email', emailRoutes);

// Health check
app.get('/health', (req, res) => {
  res.json({ status: 'ok' });
});

// Example: Send email on user signup
app.post('/api/users/signup', async (req, res) => {
  try {
    const { email, name } = req.body;

    // Create user in DB
    const user = { id: Date.now(), email, name };

    // Queue welcome email
    await EmailService.sendWelcomeEmail(user.id, email);

    // Schedule newsletter for tomorrow
    const tomorrow = new Date();
    tomorrow.setDate(tomorrow.getDate() + 1);
    await EmailService.scheduleNewsletterEmail(user.id, email, tomorrow);

    res.json({ success: true, user });
  } catch (error) {
    res.status(500).json({ error: error.message });
  }
});

const PORT = process.env.PORT || 3000;
app.listen(PORT, () => {
  console.log(`Server running on port ${PORT}`);
});

---

Key Benefits

This architecture enables production-grade async task handling with minimal complexity.