Chapter 19: WebSockets & Real-Time Communication: Socket.IO, Namespaces, Redis Adapter, and SSE

1. Short Polling:
   Client  ──► GET /status (Every 2s) ──► Server (No new data)
   Client  ──► GET /status (Every 2s) ──► Server (No new data)
   Client  ──► GET /status (Every 2s) ──► Server (200 OK: Data!)
   ⚠️ Massive overhead: Hundreds of redundant HTTP headers per minute.

2. Long Polling:
   Client  ──► GET /status ────────────► Server holds socket open...
   Client  ◄── 200 OK (After 15s) ◄───── Server emits event when ready
   Client  ──► Immediately re-opens ───► Server holds socket open...
   ⚠️ Reconnection latency and connection churn.

3. Server-Sent Events (SSE):
   Client  ──► GET /events (Accept: text/event-stream)
   Server  ──► Persistent HTTP/2 Stream (Server pushes events at will)
   ✅ Native browser auto-reconnect, lightweight, text-based.
   ❌ Unidirectional only (Server-to-Client).

4. WebSockets:
   Client  ──► GET /socket (Upgrade: websocket)
   Server  ◄── 101 Switching Protocols
   ══════════════════════════════════════════════════════════════════════════
   Bi-directional, Full-Duplex TCP Socket (Raw frames: 2-10 bytes overhead)
   ══════════════════════════════════════════════════════════════════════════
   ✅ True bi-directional, lowest latency, binary and text support.

Client (Browser)                                    Node.js Server
       │                                                   │
       │  1. HTTP GET /socket                              │
       │     Upgrade: websocket                            │
       │     Connection: Upgrade                           │
       │     Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==   │
       │     Sec-WebSocket-Version: 13                     │
       ───────────────────────────────────────────────────►│
       │                                                   │ (Validates key & auth)
       │  2. HTTP/1.1 101 Switching Protocols              │
       │     Upgrade: websocket                            │
       │     Connection: Upgrade                           │
       │     Sec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=
       │◄──────────────────────────────────────────────────│
       │                                                   │
       ═════════════════════════════════════════════════════
         PERSISTENT FULL-DUPLEX TCP FRAMES (Bidirectional)
       ═════════════════════════════════════════════════════

pnpm add socket.io @socket.io/redis-adapter ioredis

// src/common/websocket/socket-auth.middleware.ts
import { Socket } from 'socket.io';
import jwt from 'jsonwebtoken';
import { config } from '../config/env';

export interface AuthenticatedSocket extends Socket {
  user?: {
    id: string;
    email: string;
    role: string;
  };
}

/**
 * Socket.IO middleware that verifies JWT tokens before accepting connections.
 */
export const socketAuthMiddleware = (socket: Socket, next: (err?: Error) => void) => {
  // 1. Extract token from handshake auth object or query params
  const token =
    socket.handshake.auth?.token ||
    socket.handshake.headers.authorization?.replace(/^Bearer\s+/i, '');

  if (!token) {
    return next(new Error('Authentication failed: Missing token'));
  }

  try {
    // 2. Cryptographically verify token
    const decoded = jwt.verify(token, config.JWT_ACCESS_SECRET) as any;

    // 3. Attach authenticated identity to socket instance
    (socket as AuthenticatedSocket).user = {
      id: decoded.sub,
      email: decoded.email,
      role: decoded.role,
    };

    next(); // Connection allowed!
  } catch (error) {
    next(new Error('Authentication failed: Invalid or expired token'));
  }
};

                      Socket.IO Server
                             │
            ┌────────────────┴────────────────┐
            ▼                                 ▼
   Namespace: /problems              Namespace: /chat
            │                                 │
     ┌──────┴──────┐                   ┌──────┴──────┐
     ▼             ▼                   ▼             ▼
Room: "p_101"  Room: "p_102"     Room: "lobby"  Room: "help"
(Only users    (Only users       (Global chat)  (Support)
 on Two Sum)    on 3Sum)

// Client-side request with a 5000ms timeout guarantee:
try {
  const ack = await socket.timeout(5000).emitWithAck('run_test_case', {
    problemId: 'p_101',
    solutionCode: userCode,
  });
  console.log('Server confirmed test case queued:', ack.jobId);
} catch (err) {
  // Throws if the server does not acknowledge within 5000ms!
  console.error('Server timed out or unreachable');
}

// src/common/websocket/socket.server.ts
import { Server as HttpServer } from 'node:http';
import { Server as SocketIOServer } from 'socket.io';
import { socketAuthMiddleware, AuthenticatedSocket } from './socket-auth.middleware';
import { logger } from '../logger/logger';

export class RealtimeGateway {
  private static io: SocketIOServer;

  public static initialize(httpServer: HttpServer): SocketIOServer {
    this.io = new SocketIOServer(httpServer, {
      cors: {
        origin: ['http://localhost:3000', 'http://localhost:3002'],
        credentials: true,
      },
      pingTimeout: 20000,
      pingInterval: 25000,
    });

    // 🔒 Enforce authentication on all incoming connections
    this.io.use(socketAuthMiddleware);

    this.registerEventHandlers();
    return this.io;
  }

  private static registerEventHandlers(): void {
    this.io.on('connection', (rawSocket) => {
      const socket = rawSocket as AuthenticatedSocket;
      const user = socket.user!;

      logger.info({ socketId: socket.id, userId: user.id }, 'User connected to WebSocket');

      // Join a personal room for private notifications
      socket.join(`user:${user.id}`);

      // Handle joining collaborative problem workspace
      socket.on('join_problem', (problemSlug: string) => {
        const roomName = `problem:${problemSlug}`;
        socket.join(roomName);
        logger.info({ userId: user.id, roomName }, 'User joined problem room');

        // Notify others in the room
        socket.to(roomName).emit('user_joined', {
          userId: user.id,
          email: user.email,
        });
      });

      // Handle leaving problem workspace
      socket.on('leave_problem', (problemSlug: string) => {
        const roomName = `problem:${problemSlug}`;
        socket.leave(roomName);
        socket.to(roomName).emit('user_left', { userId: user.id });
      });

      // Handle test case execution with Acknowledgments
      socket.on('run_test_case', async (data: { problemSlug: string }, callback) => {
        logger.info({ userId: user.id, problem: data.problemSlug }, 'Test execution requested');
        // Acknowledge receipt back to the sender
        if (typeof callback === 'function') {
          callback({
            status: 'queued',
            jobId: `job_${Date.now()}`,
            timestamp: new Date().toISOString(),
          });
        }
      });

      // Handle disconnection
      socket.on('disconnect', (reason) => {
        logger.info({ socketId: socket.id, userId: user.id, reason }, 'User disconnected');
      });
    });
  }

  /**
   * Public helper to broadcast an event to a specific problem room from controllers or services.
   */
  public static broadcastToProblem(problemSlug: string, event: string, payload: any): void {
    if (this.io) {
      this.io.to(`problem:${problemSlug}`).emit(event, payload);
    }
  }

  /**
   * Public helper to dispatch a private notification to a specific user.
   */
  public static notifyUser(userId: string, event: string, payload: any): void {
    if (this.io) {
      this.io.to(`user:${userId}`).emit(event, payload);
    }
  }
}

❌ The Multi-Node Problem without Redis Adapter:

  Client A (User 1)               Client B (User 2)
         │                               │
         ▼                               ▼
  ┌──────────────┐                ┌──────────────┐
  │ Node.js      │                │ Node.js      │
  │ Instance 1   │                │ Instance 2   │
  └──────────────┘                └──────────────┘
         │                               ▲
         │ (Emits to "room_123")         │ (NEVER RECEIVES IT!)
         ▼                               │
  Only User 1 gets it! ──────────────────┘

✅ Multi-Node Scaling with Redis Pub/Sub:

  Client A (User 1)                               Client B (User 2)
         │                                               │
         ▼                                               ▼
  ┌──────────────┐                                ┌──────────────┐
  │ Node Server 1│                                │ Node Server 2│
  └──────┬───────┘                                └──────▲───────┘
         │                                               │
         │ 1. Publish to Redis                           │ 3. Forward to Client B
         ▼                                               │
  ┌──────────────────────────────────────────────────────────────┐
  │                   Redis Pub/Sub Channel                      │
  │                    (Cluster Backplane)                       │
  └──────────────────────────────────────────────────────────────┘

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

const pubClient = new Redis(process.env.REDIS_URL || 'redis://localhost:6379');
const subClient = pubClient.duplicate();

// Attach Redis Adapter to Socket.IO Server:
io.adapter(createAdapter(pubClient, subClient));

// src/features/submission/submission-stream.controller.ts
import { Request, Response } from 'express';

export function streamSubmissionProgress(req: Request, res: Response) {
  const { submissionId } = req.params;

  // 1. Set required SSE headers
  res.setHeader('Content-Type', 'text/event-stream');
  res.setHeader('Cache-Control', 'no-cache, no-transform');
  res.setHeader('Connection', 'keep-alive');
  res.setHeader('X-Accel-Buffering', 'no'); // Disable Nginx proxy buffering!

  res.flushHeaders(); // Send headers immediately to establish stream

  // 2. Helper to write SSE formatted data
  const sendEvent = (event: string, data: any) => {
    res.write(`event: ${event}\n`);
    res.write(`data: ${JSON.stringify(data)}\n\n`); // Must end with double newline!
  };

  // 3. Emit initial event
  sendEvent('status', { phase: 'queued', progress: 0 });

  // 4. Listen to progress events from event emitter or Redis
  const interval = setInterval(() => {
    sendEvent('progress', {
      phase: 'compiling',
      timestamp: Date.now(),
    });
  }, 1000);

  // 5. 🔒 CRITICAL: Clean up listeners on client disconnect
  req.on('close', () => {
    clearInterval(interval);
    res.end();
  });
}

┌────────────────────────────────────────────────────────────────────────────┐
│                  PRODUCTION REAL-TIME ARCHITECTURE CHECKLIST               │
├────────────────────────────────────────────────────────────────────────────┤
│ [ ] Handshake Authentication: JWT verification enforced before socket      │
│     upgrade; unauthenticated sockets rejected at the HTTP boundary.        │
│                                                                            │
│ [ ] Redis Adapter Configured: Multi-node scaling active via Redis Pub/Sub; │
│     events broadcast reliably across all container replicas.               │
│                                                                            │
│ [ ] Sticky Sessions on Reverse Proxy: Load balancer configured with        │
│     cookie-based affinity to support Socket.IO polling handshakes.         │
│                                                                            │
│ [ ] Connection Cleanup: Event listeners (`req.on('close')`, `disconnect`)  │
│     cleaned up promptly to prevent memory leaks and dangling sockets.      │
│                                                                            │
│ [ ] SSE for Unidirectional Data: Server-Sent Events used for AI streaming  │
│     and status updates instead of maintaining heavy bi-directional sockets.│
│                                                                            │
│ [ ] Ping/Pong Timeouts: Heartbeat timers configured (`pingTimeout: 20000`) │
│     to detect half-open dead TCP sockets from dropped mobile networks.     │
└────────────────────────────────────────────────────────────────────────────┘