Background Job Processing in Node.js

Some tasks don't belong in request handlers—sending emails, processing images, generating reports. Background jobs handle these asynchronously, improving response times and reliability.

When to Use Background Jobs#

Good candidates:
- Sending emails/notifications
- Processing uploads (images, videos)
- Generating reports
- Syncing with external services
- Batch data processing
- Scheduled tasks (cron)
- Webhooks delivery

BullMQ Setup#

import { Queue, Worker, Job } from 'bullmq';
import Redis from 'ioredis';

const connection = new Redis({
  host: process.env.REDIS_HOST,
  port: parseInt(process.env.REDIS_PORT || '6379'),
  maxRetriesPerRequest: null,
});

// Create queue
const emailQueue = new Queue('emails', { connection });
const imageQueue = new Queue('image-processing', { connection });

// Add job
async function sendWelcomeEmail(userId: string) {
  await emailQueue.add(
    'welcome-email',
    { userId },
    {
      attempts: 3,
      backoff: {
        type: 'exponential',
        delay: 1000,
      },
      removeOnComplete: 100,
      removeOnFail: 1000,
    }
  );
}

// Add delayed job
async function sendReminderEmail(userId: string) {
  await emailQueue.add(
    'reminder-email',
    { userId },
    {
      delay: 24 * 60 * 60 * 1000, // 24 hours
    }
  );
}

// Add scheduled job (cron)
async function setupDailyReport() {
  await emailQueue.add(
    'daily-report',
    {},
    {
      repeat: {
        pattern: '0 9 * * *', // 9 AM daily
      },
    }
  );
}

Worker Implementation#

// Email worker
const emailWorker = new Worker(
  'emails',
  async (job: Job) => {
    const { userId } = job.data;

    switch (job.name) {
      case 'welcome-email':
        await sendWelcomeEmailToUser(userId);
        break;
      case 'reminder-email':
        await sendReminderEmailToUser(userId);
        break;
      case 'daily-report':
        await generateAndSendDailyReport();
        break;
      default:
        throw new Error(`Unknown job type: ${job.name}`);
    }

    return { success: true };
  },
  {
    connection,
    concurrency: 5,
    limiter: {
      max: 100,
      duration: 60000, // 100 jobs per minute
    },
  }
);

// Event handlers
emailWorker.on('completed', (job) => {
  logger.info('Job completed', {
    jobId: job.id,
    name: job.name,
    duration: job.finishedOn! - job.processedOn!,
  });
});

emailWorker.on('failed', (job, error) => {
  logger.error('Job failed', {
    jobId: job?.id,
    name: job?.name,
    error: error.message,
    attempts: job?.attemptsMade,
  });
});

emailWorker.on('error', (error) => {
  logger.error('Worker error', { error: error.message });
});

Job Patterns#

Fan-Out#

// Process multiple items in parallel
async function processOrder(orderId: string) {
  const order = await prisma.order.findUnique({
    where: { id: orderId },
    include: { items: true },
  });

  // Create child jobs for each item
  const jobs = order.items.map((item) => ({
    name: 'process-item',
    data: { orderId, itemId: item.id },
  }));

  await orderQueue.addBulk(jobs);
}

Job Chaining#

// Sequential processing
const pipelineQueue = new Queue('pipeline', { connection });

async function startImagePipeline(imageId: string) {
  await pipelineQueue.add('download', { imageId });
}

const pipelineWorker = new Worker(
  'pipeline',
  async (job) => {
    const { imageId } = job.data;

    switch (job.name) {
      case 'download':
        const image = await downloadImage(imageId);
        await pipelineQueue.add('resize', { imageId, path: image.path });
        break;

      case 'resize':
        const resized = await resizeImage(job.data.path);
        await pipelineQueue.add('upload', { imageId, path: resized.path });
        break;

      case 'upload':
        const url = await uploadToS3(job.data.path);
        await prisma.image.update({
          where: { id: imageId },
          data: { url, status: 'processed' },
        });
        break;
    }
  },
  { connection }
);

Job Dependencies (Flow)#

import { FlowProducer } from 'bullmq';

const flowProducer = new FlowProducer({ connection });

async function createOrderFlow(orderId: string) {
  await flowProducer.add({
    name: 'complete-order',
    queueName: 'orders',
    data: { orderId },
    children: [
      {
        name: 'send-confirmation',
        queueName: 'emails',
        data: { orderId },
      },
      {
        name: 'update-inventory',
        queueName: 'inventory',
        data: { orderId },
      },
      {
        name: 'notify-shipping',
        queueName: 'shipping',
        data: { orderId },
      },
    ],
  });
}

// Parent job runs after all children complete

Progress Tracking#

// Update progress from worker
const worker = new Worker(
  'reports',
  async (job) => {
    const { items } = job.data;
    let processed = 0;

    for (const item of items) {
      await processItem(item);
      processed++;
      await job.updateProgress(Math.round((processed / items.length) * 100));
    }

    return { processed };
  },
  { connection }
);

// Check progress from API
app.get('/api/jobs/:id/progress', async (req, res) => {
  const job = await reportsQueue.getJob(req.params.id);

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

  res.json({
    id: job.id,
    state: await job.getState(),
    progress: job.progress,
    data: job.returnvalue,
  });
});

Error Handling#

const worker = new Worker(
  'critical-jobs',
  async (job) => {
    try {
      await processJob(job.data);
    } catch (error) {
      // Determine if retryable
      if (error instanceof TemporaryError) {
        throw error; // Will retry
      }

      // Permanent failure - don't retry
      if (error instanceof ValidationError) {
        await job.moveToFailed(error, job.token!);
        return;
      }

      // Log and rethrow
      logger.error('Job failed', {
        jobId: job.id,
        error: error.message,
      });
      throw error;
    }
  },
  {
    connection,
    settings: {
      backoffStrategy: (attemptsMade: number) => {
        // Custom backoff: 1s, 5s, 25s, 2m, 10m
        return Math.min(Math.pow(5, attemptsMade) * 1000, 600000);
      },
    },
  }
);

// Dead letter queue for failed jobs
const dlqWorker = new Worker(
  'emails',
  async (job) => {
    // Process...
  },
  { connection }
);

dlqWorker.on('failed', async (job, error) => {
  if (job.attemptsMade >= job.opts.attempts!) {
    // Move to dead letter queue
    await deadLetterQueue.add('failed-email', {
      originalJob: job.data,
      error: error.message,
      attempts: job.attemptsMade,
    });
  }
});

Monitoring Dashboard#

import { createBullBoard } from '@bull-board/api';
import { BullMQAdapter } from '@bull-board/api/bullMQAdapter';
import { ExpressAdapter } from '@bull-board/express';

const serverAdapter = new ExpressAdapter();
serverAdapter.setBasePath('/admin/queues');

createBullBoard({
  queues: [
    new BullMQAdapter(emailQueue),
    new BullMQAdapter(imageQueue),
    new BullMQAdapter(reportsQueue),
  ],
  serverAdapter,
});

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

Best Practices#

DO:
✓ Make jobs idempotent
✓ Store minimal data in job payload
✓ Set appropriate timeouts
✓ Monitor queue depths
✓ Handle job failures gracefully
✓ Use separate queues for priorities

DON'T:
✗ Store large payloads in jobs
✗ Rely on job order
✗ Ignore failed jobs
✗ Run workers without monitoring
✗ Use infinite retries

Conclusion#

Background jobs improve user experience and system reliability. Use queues for anything that can be deferred, implement proper error handling, and monitor your queues.

The key is making jobs idempotent—they should be safe to run multiple times.

Share this article