Module P-9 — WebSockets and Real-Time Communication

HTTP vs WebSocket

HTTP is a half-duplex, stateless protocol. Every interaction is client-initiated: request → response → connection closes. The server cannot send data unless the client asks.

WebSocket is a full-duplex, stateful protocol over a single TCP connection. After the initial HTTP upgrade handshake, both sides can send frames at any time with minimal overhead (2–14 byte header per frame vs 200–800 bytes for an HTTP request).

HTTP polling (inefficient real-time):
Client → GET /messages?since=123        Server
Client ← { messages: [] }               (nothing new)
... wait 1 second ...
Client → GET /messages?since=123        Server
Client ← { messages: [] }               (still nothing)
... wait 1 second ...
Client → GET /messages?since=123        Server
Client ← { messages: [{ text: "Hi" }] } (finally)

WebSocket (efficient):
Client → HTTP Upgrade: websocket        Server
Client ← 101 Switching Protocols        (handshake)
                                        [connection stays open]
Server → { text: "Hi" }                 (pushed immediately when available)

The WebSocket handshake is a standard HTTP request with Connection: Upgrade and Upgrade: websocket headers. After the 101 response, the protocol switches and HTTP is no longer involved.

Raw WebSocket with the ws Library

bash
npm install ws
npm install -D @types/ws

typescript
// src/websocket/server.ts
import { WebSocketServer, WebSocket } from 'ws';
import { IncomingMessage } from 'http';
import { verifyAccessToken } from '../utils/jwt.js';
import logger from '../utils/logger.js';

interface AuthenticatedSocket extends WebSocket {
  userId?: number;
  isAlive?: boolean;
}

export function createWebSocketServer(httpServer: ReturnType<typeof import('http').createServer>) {
  const wss = new WebSocketServer({ server: httpServer });

  wss.on('connection', (ws: AuthenticatedSocket, req: IncomingMessage) => {
    // Authenticate via query string token (headers not available after upgrade in browsers)
    const url = new URL(req.url!, `http://${req.headers.host}`);
    const token = url.searchParams.get('token');

    try {
      const payload = verifyAccessToken(token ?? '');
      ws.userId = payload.sub as number;
    } catch {
      ws.send(JSON.stringify({ type: 'error', message: 'Unauthorized' }));
      ws.close(4001, 'Unauthorized');
      return;
    }

    ws.isAlive = true;
    logger.info({ userId: ws.userId }, 'WebSocket connected');

    ws.on('message', (data) => {
      try {
        const message = JSON.parse(data.toString());
        handleMessage(ws, message, wss);
      } catch {
        ws.send(JSON.stringify({ type: 'error', message: 'Invalid JSON' }));
      }
    });

    ws.on('pong', () => { ws.isAlive = true; });

    ws.on('close', (code, reason) => {
      logger.info({ userId: ws.userId, code }, 'WebSocket disconnected');
    });

    ws.send(JSON.stringify({ type: 'connected', userId: ws.userId }));
  });

  // Heartbeat — detect dead connections
  const heartbeat = setInterval(() => {
    wss.clients.forEach((ws: AuthenticatedSocket) => {
      if (!ws.isAlive) return ws.terminate();
      ws.isAlive = false;
      ws.ping();
    });
  }, 30_000);

  wss.on('close', () => clearInterval(heartbeat));

  return wss;
}

function handleMessage(
  sender: AuthenticatedSocket,
  message: { type: string; [key: string]: unknown },
  wss: WebSocketServer,
) {
  switch (message.type) {
    case 'chat:send':
      // Broadcast to all connected clients
      const outgoing = JSON.stringify({
        type: 'chat:message',
        from: sender.userId,
        text: message.text,
        ts: Date.now(),
      });
      wss.clients.forEach((client) => {
        if (client.readyState === WebSocket.OPEN) {
          client.send(outgoing);
        }
      });
      break;

    default:
      sender.send(JSON.stringify({ type: 'error', message: `Unknown type: ${message.type}` }));
  }
}

typescript
// src/index.ts — attach WebSocket server to the HTTP server
import http from 'http';
import app from './app.js';
import { createWebSocketServer } from './websocket/server.js';

const httpServer = http.createServer(app);
createWebSocketServer(httpServer);

httpServer.listen(env.PORT, () => {
  logger.info({ port: env.PORT }, 'Server started');
});

Socket.IO: Rooms, Namespaces, and Events

Socket.IO builds on WebSocket and adds rooms (group channels), namespaces (logical separation), automatic reconnection, event-based messaging, and a fallback to long polling when WebSocket is unavailable.

bash
npm install socket.io

typescript
// src/realtime/socket.ts
import { Server, Socket } from 'socket.io';
import http from 'http';
import { verifyAccessToken } from '../utils/jwt.js';
import { env } from '../config/env.js';

interface SocketData {
  userId: number;
  role: string;
}

export function createSocketServer(httpServer: http.Server) {
  const io = new Server<{}, {}, {}, SocketData>(httpServer, {
    cors: {
      origin: env.CORS_ORIGIN.split(','),
      credentials: true,
    },
    pingTimeout: 60_000,
    pingInterval: 25_000,
  });

  // Authentication middleware — runs before connection is established
  io.use((socket, next) => {
    const token = socket.handshake.auth.token ?? socket.handshake.query.token;
    try {
      const payload = verifyAccessToken(token as string);
      socket.data.userId = payload.sub as number;
      socket.data.role = payload.role as string;
      next();
    } catch {
      next(new Error('Authentication failed'));
    }
  });

  io.on('connection', (socket: Socket<{}, {}, {}, SocketData>) => {
    const { userId } = socket.data;

    // Join a personal room for direct messages
    socket.join(`user:${userId}`);

    socket.on('room:join', async (roomId: string) => {
      // Authorise — check if user can access this room
      const canAccess = await canUserAccessRoom(userId, roomId);
      if (!canAccess) {
        socket.emit('error', { message: 'Access denied' });
        return;
      }
      socket.join(`room:${roomId}`);
      socket.to(`room:${roomId}`).emit('room:user-joined', { userId, roomId });
    });

    socket.on('room:leave', (roomId: string) => {
      socket.leave(`room:${roomId}`);
      socket.to(`room:${roomId}`).emit('room:user-left', { userId, roomId });
    });

    socket.on('chat:message', async (data: { roomId: string; text: string }) => {
      const { roomId, text } = data;

      // Save to database
      const message = await messagesRepo.create({ roomId, userId, text });

      // Broadcast to everyone in the room (including sender)
      io.to(`room:${roomId}`).emit('chat:message', {
        id: message.id,
        roomId,
        userId,
        text,
        ts: message.createdAt,
      });
    });

    socket.on('disconnect', () => {
      io.to(`user:${userId}`).emit('user:offline', { userId });
    });
  });

  return io;
}

Emitting to specific targets

typescript
// To a single user (even if they have multiple tabs open)
io.to(`user:${userId}`).emit('notification', { message: 'Your order shipped!' });

// To all users in a room
io.to(`room:${roomId}`).emit('chat:message', message);

// To all connected clients
io.emit('announcement', { text: 'System maintenance in 5 minutes' });

// To everyone in a room except the sender
socket.to(`room:${roomId}`).emit('user:typing', { userId });

// From a REST controller — inject io and emit without a socket
// src/services/orders.service.ts
export async function confirmOrder(orderId: number) {
  const order = await ordersRepo.update(orderId, { status: 'confirmed' });
  io.to(`user:${order.userId}`).emit('order:confirmed', { orderId });
  return order;
}

Namespaces

Namespaces are like separate sub-applications on the same server — separate event handling, separate middleware:

typescript
// Admin namespace — different auth middleware
const adminNs = io.of('/admin');

adminNs.use((socket, next) => {
  // Stricter auth — must be admin
  if (socket.data.role !== 'admin') return next(new Error('Admin only'));
  next();
});

adminNs.on('connection', (socket) => {
  socket.on('broadcast:alert', (msg) => {
    io.emit('system:alert', msg); // broadcast to ALL users from admin namespace
  });
});

Server-Sent Events (SSE)

SSE is simpler than WebSocket for one-directional streams — server pushing to client, never the other way. It uses standard HTTP, works through proxies without configuration, and browsers reconnect automatically.

Good use cases: live dashboards, notification feeds, progress updates.

typescript
// src/routes/events.routes.ts
import { Router } from 'express';
import { authenticate } from '../middleware/auth.js';
import { asyncHandler } from '../utils/asyncHandler.js';

const router = Router();

router.get('/stream', authenticate, asyncHandler(async (req, res) => {
  // Set SSE headers
  res.setHeader('Content-Type', 'text/event-stream');
  res.setHeader('Cache-Control', 'no-cache');
  res.setHeader('Connection', 'keep-alive');
  res.setHeader('X-Accel-Buffering', 'no'); // disable nginx buffering
  res.flushHeaders();

  const userId = req.user!.id;

  // Helper to send an event
  const send = (event: string, data: unknown) => {
    res.write(`event: ${event}\n`);
    res.write(`data: ${JSON.stringify(data)}\n\n`);
  };

  // Register this client to receive events
  eventBus.on(`user:${userId}`, send);

  // Heartbeat — keep the connection alive through proxies (every 30s)
  const heartbeat = setInterval(() => {
    res.write(': heartbeat\n\n');
  }, 30_000);

  // Cleanup when client disconnects
  req.on('close', () => {
    clearInterval(heartbeat);
    eventBus.off(`user:${userId}`, send);
    res.end();
  });
}));

export default router;

Triggering SSE events from elsewhere in your app:

typescript
// src/utils/eventBus.ts
import { EventEmitter } from 'events';
export const eventBus = new EventEmitter();
eventBus.setMaxListeners(10_000); // one listener per connected user

// src/services/orders.service.ts
import { eventBus } from '../utils/eventBus.js';

export async function confirmOrder(orderId: number) {
  const order = await ordersRepo.update(orderId, { status: 'confirmed' });
  eventBus.emit(`user:${order.userId}`, 'order:confirmed', { orderId, status: 'confirmed' });
  return order;
}

Client-side (browser):

javascript
const es = new EventSource('/events/stream', { withCredentials: true });
es.addEventListener('order:confirmed', (e) => {
  const data = JSON.parse(e.data);
  console.log('Order confirmed:', data.orderId);
});
es.onerror = () => console.log('SSE connection lost, reconnecting...');
// Browsers reconnect automatically

Scaling WebSockets with the Redis Adapter

A single-instance app has one in-memory set of connected sockets. With two app servers, a user connected to server A cannot receive an event emitted on server B.

The Socket.IO Redis adapter solves this by routing events through Redis Pub/Sub:

bash
npm install @socket.io/redis-adapter

typescript
// src/realtime/socket.ts
import { createAdapter } from '@socket.io/redis-adapter';
import { createClient } from 'redis'; // node-redis, not ioredis
import { env } from '../config/env.js';

export async function createSocketServer(httpServer: http.Server) {
  const io = new Server(httpServer, { cors: { origin: env.CORS_ORIGIN.split(',') } });

  // Two separate Redis connections — one pub, one sub
  const pubClient = createClient({ url: env.REDIS_URL });
  const subClient = pubClient.duplicate();

  await Promise.all([pubClient.connect(), subClient.connect()]);

  io.adapter(createAdapter(pubClient, subClient));

  // Now io.to(...).emit(...) broadcasts across ALL server instances
  // ...
}

With the Redis adapter:

• Server A receives a message from a user
• Server A emits to room:xyz via Socket.IO
• Socket.IO publishes the event to Redis
• Redis fans it out to all subscribers (every app server)
• Each server delivers it to its locally connected clients in room:xyz

Choosing the Right Transport

WebSocket and SSE can coexist in the same app — use WebSocket for bidirectional real-time features, SSE for one-directional pushes.

Summary

• WebSocket upgrades HTTP to a persistent full-duplex TCP connection. Minimal frame overhead makes it ideal for high-frequency bidirectional messaging.
• ws library is the low-level WebSocket server. Authenticate via handshake.auth or query params (not headers — browsers don't allow custom WS headers). Heartbeat pings detect dead connections.
• Socket.IO adds rooms, namespaces, event names, automatic reconnection, and long-polling fallback. Authenticate in the io.use() middleware. Rooms enable targeted broadcasting without managing client sets manually.
• Server-Sent Events are plain HTTP — one-directional, proxy-friendly, auto-reconnecting. Perfect for notification feeds and live dashboards where the client never sends data over the stream.
• Redis adapter makes Socket.IO multi-server — events emitted on any instance reach clients on all instances. Two Redis connections required (pub + sub).

Next: REST API design principles, URL structure, versioning strategies, cursor-based pagination, and generating interactive OpenAPI documentation from your Express routes.