Message Queues & Event-Driven

When a user uploads a video, you need to compress it. If your API does this synchronously, the user will stare at a loading spinner for 5 minutes and the connection might timeout. This is where Message Queues come in: they allow your API to respond instantly, while the heavy lifting happens in the background.

Step 1 — The Architecture

A message queue decouples the 'Producer' (who creates the task) from the 'Consumer' (who executes the task).

Core Components

Producer: Your main API. It quickly pushes a 'message' onto the queue and returns a 200 OK to the user.
Message Broker: The server managing the queue (RabbitMQ, Kafka, AWS SQS, or Redis BullMQ).
Consumer (Worker): A separate background Node.js process that listens to the queue, pulls messages off one by one, and processes them.

Step 2 — Implementing a Queue with BullMQ

BullMQ is a popular, robust NodeJS message queue based on Redis.

const { Queue } = require('bullmq');

// Create a new queue instance connected to Redis
const videoQueue = new Queue('VideoEncoding', {
  connection: { host: 'localhost', port: 6379 }
});

// API Route (Producer)
app.post('/api/upload-video', async (req, res) => {
  const videoId = saveToS3(req.file);
  
  // Add a job to the queue. Returns instantly!
  await videoQueue.add('compress-video', {
    videoId: videoId,
    resolution: '1080p'
  });
  
  res.status(202).json({ message: 'Video uploaded! Processing in background.' });
});

Step 3 — The Worker Process

In a completely separate file (or even a separate server), you run the Worker. This script just listens to Redis and processes jobs.

const { Worker } = require('bullmq');

const worker = new Worker('VideoEncoding', async job => {
  console.log(`Processing video ${job.data.videoId}...`);
  
  // Simulate heavy processing (FFmpeg compression)
  await heavyVideoCompressionFunction(job.data.videoId, job.data.resolution);
  
  // Update database status
  await db.query('UPDATE videos SET status = "READY" WHERE id = ?', job.data.videoId);
  
  console.log('Done!');
}, {
  connection: { host: 'localhost', port: 6379 },
  concurrency: 5 // Process up to 5 videos simultaneously
});

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