Module 8: Real-Time Communication & Background Tasks

  • Home
  • Blog
  • Module 8: Real-Time Communication & Background Tasks
Featured

Module 8: Real-Time Communication & Background Tasks

November 9, 2026 in Node,js

WebSockets & Real-Time Communication (Socket.io)

HTTP is a stateless, request-response protocol initiated solely by the client. For use cases requiring bidirectional, low-latency updates (chats, live metrics, collaboration tools), the WebSocket protocol (ws://, wss://) establishes a persistent, full-duplex TCP connection.

Socket.io builds on top of WebSockets by providing automatic reconnection, connection fallback (HTTP long-polling), packet buffering, acknowledgments, and horizontal scaling abstractions (rooms and namespaces).

Client ──▶ HTTP Upgrade Request ──▶ Server Handshake ──▶ Persistent Full-Duplex TCP Socket

Bash

npm install socket.io

Core Concepts: Namespaces, Rooms, and Broadcasting

  • Namespaces (io.of('/custom')): Multiplexes a single shared TCP connection into isolated logical channels with independent middleware, event handlers, and rooms.
  • Rooms (socket.join('room-name')): Server-side-only subdivisions within a namespace. Sockets can join/leave rooms to receive targeted broadcasts without other clients receiving that traffic.
  • Broadcasting: Emitting an event to subsets of connected clients.

JavaScript

// server.js
import http from 'node:http';
import express from 'express';
import { Server } from 'socket.io';
import { verifyAccessToken } from './utils/jwt.js';

const app = express();
const server = http.createServer(app);

// Initialize Socket.io with strict CORS
const io = new Server(server, {
  cors: {
    origin: 'https://app.domain.com',
    methods: ['GET', 'POST'],
    credentials: true
  },
  pingTimeout: 20000,
  pingInterval: 25000
});

// Authentication Middleware for WebSockets
io.use((socket, next) => {
  const token = socket.handshake.auth?.token || socket.handshake.headers?.authorization?.split(' ')[1];
  
  if (!token) {
    return next(new Error('Authentication error: Missing token'));
  }

  try {
    const user = verifyAccessToken(token);
    socket.data.user = user; // Attach authenticated identity to socket instance
    next();
  } catch (err) {
    next(new Error('Authentication error: Invalid or expired token'));
  }
});

// Connection lifecycle & event dispatching
io.on('connection', (socket) => {
  const userId = socket.data.user.sub;
  console.log(`Socket connected: ${socket.id} (User: ${userId})`);

  // Auto-join personal room for private notifications
  socket.join(`user:${userId}`);

  // Joining a group room
  socket.on('room:join', (roomId) => {
    socket.join(roomId);
    // Broadcast to everyone ELSE in the room
    socket.to(roomId).emit('user:joined', { userId, timestamp: Date.now() });
  });

  // Bidirectional messaging with Acknowledgments
  socket.on('message:send', async (payload, callback) => {
    try {
      const messageRecord = {
        senderId: userId,
        text: payload.text,
        roomId: payload.roomId,
        createdAt: new Date().toISOString()
      };

      // Broadcast message to everyone in the target room (including the sender)
      io.in(payload.roomId).emit('message:received', messageRecord);

      // Fire callback acknowledgment to the original sender
      callback({ status: 'delivered', id: 'msg_abc123' });
    } catch (err) {
      callback({ status: 'error', message: err.message });
    }
  });

  // Leaving a room
  socket.on('room:leave', (roomId) => {
    socket.leave(roomId);
    socket.to(roomId).emit('user:left', { userId });
  });

  // Handle client disconnection
  socket.on('disconnect', (reason) => {
    console.log(`Socket ${socket.id} disconnected: ${reason}`);
  });
});

server.listen(3000, () => console.log('Realtime server running on :3000'));

Event Emission Cheat Sheet

JavaScript

// Send to the current socket only
socket.emit('event', data);

// Send to all connected clients EXCEPT the sender
socket.broadcast.emit('event', data);

// Send to all clients in 'roomA' EXCEPT the sender
socket.to('roomA').emit('event', data);

// Send to all clients in 'roomA' INCLUDING the sender
io.in('roomA').emit('event', data);

// Send to a single targeted user by their personal room
io.to(`user:${targetUserId}`).emit('notification', { alert: 'New message' });

// Send to everyone connected to the entire server
io.emit('system:alert', { message: 'Server restarting in 5m' });

Horizontal Scaling with Redis Streams Adapter

Because WebSockets are stateful, running multiple Node.js instances behind a load balancer will isolate sockets to individual processes. Use the @socket.io/redis-adapter to publish and synchronize events across instances through Redis:

Bash

npm install @socket.io/redis-adapter ioredis

JavaScript

import { createAdapter } from '@socket.io/redis-adapter';
import Redis from 'ioredis';

const pubClient = new Redis(process.env.REDIS_URL);
const subClient = pubClient.duplicate();

// Distribute socket events across any number of clustered Node nodes
io.adapter(createAdapter(pubClient, subClient));

Background Jobs & Message Queues

Running long-running or CPU-heavy tasks (video transcoding, PDF generation, CSV imports, batch emails) synchronously inside an HTTP handler ties up the main thread and causes client timeouts.

Offload these operations to background worker queues.

HTTP POST /reports ──▶ [ Producer ] ──▶ BullMQ Job pushed to Redis ──▶ 202 Accepted
                                                                               │
[ Background Worker Process ] ◀── Picks up job from Redis Queue ◀──────────────┘

1. BullMQ (Redis-Backed Distributed Queue)

BullMQ is the NodeJS standard for message queues, handling concurrency, exponential backoff retries, delayed jobs, parent-child flows, and metrics.

Bash

npm install bullmq ioredis
Step 1: Redis Connection & Queue Producer

JavaScript

// queues/email.queue.js
import { Queue } from 'bullmq';

const connection = {
  host: process.env.REDIS_HOST || '127.0.0.1',
  port: Number(process.env.REDIS_PORT) || 6379
};

export const emailQueue = new Queue('email-delivery', {
  connection,
  defaultJobOptions: {
    attempts: 4, // Retry up to 4 times on failure
    backoff: {
      type: 'exponential',
      delay: 2000 // 2s, 4s, 8s, 16s...
    },
    removeOnComplete: 1000, // Keep last 1,000 completed jobs in Redis for audit
    removeOnFail: 5000      // Keep failed jobs for inspection
  }
});

// Adding jobs to the queue from an Express controller
export async function scheduleWelcomeEmail(user) {
  return await emailQueue.add(
    'send-welcome',
    { userId: user.id, email: user.email, name: user.name },
    {
      priority: 2, // 1 is highest priority
      delay: 5000  // Deliver 5 seconds after invocation
    }
  );
}
Step 2: Dedicated Worker Processor

Run workers in a separate process or container so heavy compute operations never interfere with HTTP traffic.

JavaScript

// workers/email.worker.js
import { Worker } from 'bullmq';

const connection = {
  host: process.env.REDIS_HOST || '127.0.0.1',
  port: Number(process.env.REDIS_PORT) || 6379
};

export const emailWorker = new Worker(
  'email-delivery',
  async (job) => {
    console.log(`Processing job ${job.id} of type ${job.name}...`);

    switch (job.name) {
      case 'send-welcome': {
        const { email, name } = job.data;
        // Update job progress (0% - 100%) for UI trackers
        await job.updateProgress(25);
        
        // Execute the external operation
        await sendEmailViaProvider({ to: email, template: 'welcome', context: { name } });
        
        await job.updateProgress(100);
        return { delivered: true, sentAt: new Date().toISOString() };
      }

      default:
        throw new Error(`Unhandled job type: ${job.name}`);
    }
  },
  {
    connection,
    concurrency: 5 // Process up to 5 jobs concurrently per worker instance
  }
);

emailWorker.on('completed', (job, returnvalue) => {
  console.log(`Job ${job.id} completed. Result:`, returnvalue);
});

emailWorker.on('failed', (job, err) => {
  console.error(`Job ${job.id} failed after ${job.attemptsMade} attempts: ${err.message}`);
});

2. In-Process Cron Jobs (node-cron)

For simple, single-instance recurring scheduled tasks (e.g., nightly cache purges, daily report aggregation), node-cron provides a straightforward cron syntax parser.

Bash

npm install node-cron

JavaScript

// jobs/scheduler.js
import cron from 'node-cron';

// Cron Syntax: [second] [minute] [hour] [day-of-month] [month] [day-of-week]
// Runs every night at exactly 02:00 AM (server local time)
cron.schedule('0 0 2 * * *', async () => {
  console.log('Running daily database backup and orphan file cleanup...');
  try {
    await purgeExpiredSessions();
    console.log('Maintenance finished successfully');
  } catch (err) {
    console.error('Scheduled task failed:', err);
  }
}, {
  scheduled: true,
  timezone: 'UTC' // Always lock cron execution to UTC
});

Warning on Clustered Deployments: If running multiple replicas of your app behind a load balancer, standard node-cron schedules execute on every single running instance, causing duplicate executions. For multi-replica setups, use BullMQ Repeatable Jobs, which use Redis to enforce single execution across clusters:

JavaScript

await queue.add('nightly-sync', {}, { repeat: { pattern: '0 2 * * *' } });

Email & Transactional Notifications

Transactional emails (password resets, invoice PDFs, magic links) are critical delivery pathways. Node.js applications use SMTP via Nodemailer or modern HTTP-based REST APIs (Resend, SendGrid, Postmark).

Method 1: Nodemailer (SMTP Engine)

Nodemailer connects directly to an arbitrary SMTP server (AWS SES, Mailgun, or local Postfix).

Bash

npm install nodemailer

JavaScript

// services/nodemailer.service.js
import nodemailer from 'nodemailer';

// Create persistent transporter pool
const transporter = nodemailer.createTransport({
  host: process.env.SMTP_HOST,
  port: Number(process.env.SMTP_PORT) || 587,
  secure: false, // true for 465, false for 587 (STARTTLS)
  auth: {
    user: process.env.SMTP_USER,
    pass: process.env.SMTP_PASS
  },
  pool: true,     // Keep connections alive to reuse TCP sockets
  maxConnections: 5
});

export async function sendPasswordResetEmail({ to, token }) {
  const resetLink = `https://app.domain.com/reset-password?token=${token}`;

  const mailOptions = {
    from: '"Security Team" <security@domain.com>',
    to,
    subject: 'Action Required: Reset Your Password',
    text: `Reset your password by opening this URL: ${resetLink}`,
    html: `
      <div style="font-family: sans-serif; padding: 20px;">
        <h2>Password Reset Request</h2>
        <p>Click the link below to set a new password. This link expires in 15 minutes.</p>
        <a href="${resetLink}" style="display:inline-block;padding:10px 20px;background:#0070f3;color:#fff;text-decoration:none;border-radius:5px;">Reset Password</a>
      </div>
    `
  };

  const info = await transporter.sendMail(mailOptions);
  return { messageId: info.messageId };
}

Method 2: Modern HTTP APIs (Resend SDK)

Direct SMTP can be slow, vulnerable to firewall blocking, and requires maintaining persistent connections. Modern platforms use transactional HTTP APIs with lower overhead and built-in React template rendering.

Bash

npm install resend

JavaScript

// services/resend.service.js
import { Resend } from 'resend';

const resend = new Resend(process.env.RESEND_API_KEY);

export async function sendInvoiceEmail({ to, invoiceId, pdfBuffer }) {
  const { data, error } = await resend.emails.send({
    from: 'billing@domain.com',
    to: [to],
    subject: `Invoice #${invoiceId} is available`,
    html: '<p>Attached is your monthly billing statement. Thank you for your business!</p>',
    attachments: [
      {
        filename: `invoice-${invoiceId}.pdf`,
        content: pdfBuffer // Pass raw binary Buffer directly without writing to disk
      }
    ],
    tags: [
      { name: 'category', value: 'invoicing' }
    ]
  });

  if (error) {
    throw new Error(`Email delivery failed: ${error.message}`);
  }

  return data; // { id: '49af6ee4-4806-44c6-8d8d-29406085ff87' }
}

Architectural Pattern: Combining WebSockets, Queues, and Email

1. Client POST /api/reports/export
   │
   ├──▶ Controller pushes 'generate-pdf' task to BullMQ
   └──▶ Immediately replies: 202 Accepted { jobId: 'job_456' }

2. BullMQ Worker picks up 'generate-pdf'
   │
   ├──▶ Compiles heavy dynamic report
   ├──▶ Sends email with file attachment via Resend
   └──▶ Emits completion to Redis PubSub

3. Socket.io Cluster receives Redis PubSub
   │
   └──▶ Dispatches socket event to target user room:
        io.to(`user:${userId}`).emit('report:ready', { downloadUrl: '...' });

singh aditya