Real-time Web Patterns

// WRONG: WebSocket adds unnecessary complexity for server-to-client-only updates
const ws = new WebSocket('/api/dashboard-updates');
ws.onmessage = (e) => updateDashboard(JSON.parse(e.data));
// Now you must handle reconnection, heartbeats, etc. manually

// RIGHT: SSE handles reconnection automatically and works through proxies
const es = new EventSource('/api/dashboard-updates');
es.addEventListener('metrics', (e) => updateDashboard(JSON.parse(e.data)));
es.onerror = () => console.log('SSE reconnecting automatically...');

// src/sse.ts
import { Request, Response } from 'express';

interface SSEClient {
  id: string;
  res: Response;
  lastEventId: number;
}

const clients = new Map<string, SSEClient>();
let globalEventId = 0;

export function addClient(req: Request, res: Response): void {
  res.writeHead(200, {
    'Content-Type': 'text/event-stream',
    'Cache-Control': 'no-cache',
    'Connection': 'keep-alive',
    'X-Accel-Buffering': 'no',  // Disable nginx buffering
  });

  const clientId = crypto.randomUUID();

  // Send initial connection event
  res.write(`event: connected\ndata: ${JSON.stringify({ clientId, time: Date.now() })}\n\n`);

  // If client reconnected, replay missed events using Last-Event-ID
  const lastEventId = parseInt(req.headers['last-event-id'] as string, 10);
  if (!isNaN(lastEventId)) {
    replayEventsSince(res, lastEventId);
  }

  clients.set(clientId, { id: clientId, res, lastEventId: globalEventId });
  req.on('close', () => clients.delete(clientId));
}

export function broadcast(event: string, data: unknown): void {
  globalEventId++;
  // Include id: field so EventSource sends Last-Event-ID on reconnect
  const message = `id: ${globalEventId}\nevent: ${event}\ndata: ${JSON.stringify(data)}\n\n`;
  for (const client of clients.values()) {
    if (!client.res.writableEnded) {
      client.res.write(message);
    }
  }
}

// Heartbeat -- keeps connection alive through proxies and load balancers
setInterval(() => {
  for (const [clientId, client] of clients) {
    if (client.res.writableEnded) {
      clients.delete(clientId);
      continue;
    }
    client.res.write(': heartbeat\n\n');
  }
}, 30_000);

// WRONG: No event IDs means client cannot resume after reconnection
function broadcast(event: string, data: unknown): void {
  const message = `event: ${event}\ndata: ${JSON.stringify(data)}\n\n`;
  for (const client of clients) {
    client.write(message);
  }
}
// Client reconnects but misses all events sent during the disconnect gap

// RIGHT: Include id: field so EventSource sends Last-Event-ID on reconnect
function broadcast(event: string, data: unknown): void {
  eventId++;
  const message = `id: ${eventId}\nevent: ${event}\ndata: ${JSON.stringify(data)}\n\n`;
  for (const client of clients.values()) {
    client.res.write(message);
  }
}

// On reconnect, check Last-Event-ID header and replay missed events
app.get('/api/events', (req, res) => {
  const lastId = parseInt(req.headers['last-event-id'] as string, 10);
  if (!isNaN(lastId)) {
    replayEventsSince(res, lastId);
  }
  addClient(req, res);
});

function connectSSE(onEvent: (event: string, data: any) => void): () => void {
  const es = new EventSource('/api/events');

  es.addEventListener('order:updated', (e) => {
    onEvent('order:updated', JSON.parse(e.data));
  });

  es.addEventListener('order:new', (e) => {
    onEvent('order:new', JSON.parse(e.data));
  });

  es.onerror = () => {
    // EventSource auto-reconnects -- just log it
    console.log('SSE connection lost, reconnecting...');
  };

  // Return cleanup function
  return () => es.close();
}

import { Server as SocketIOServer } from 'socket.io';
import http from 'http';

const server = http.createServer(app);
const io = new SocketIOServer(server, {
  cors: { origin: process.env.FRONTEND_URL || 'http://localhost:5173' },
  pingInterval: 25000,   // How often to ping clients
  pingTimeout: 20000,    // How long to wait for pong before disconnect
});

// Authentication middleware -- verify on upgrade, not after
io.use((socket, next) => {
  const token = socket.handshake.auth.token;
  try {
    const user = verifyToken(token);
    socket.data.user = user;
    next();
  } catch (err) {
    next(new Error('Authentication failed'));
  }
});

io.on('connection', (socket) => {
  console.log(`Client connected: ${socket.id}, user: ${socket.data.user.id}`);

  // Join a room for targeted updates
  socket.on('watch:order', (orderId: string) => {
    // Authorize: only let user watch their own orders
    if (canUserWatchOrder(socket.data.user, orderId)) {
      socket.join(`order:${orderId}`);
    }
  });

  socket.on('unwatch:order', (orderId: string) => {
    socket.leave(`order:${orderId}`);
  });

  socket.on('disconnect', (reason) => {
    console.log(`Client disconnected: ${socket.id} (${reason})`);
  });
});

// Broadcast to specific room -- not to every client
function broadcastOrderUpdate(order: Order) {
  io.to(`order:${order.id}`).emit('order:status', {
    id: order.id,
    status: order.status,
    updatedAt: order.updatedAt,
  });
}

// WRONG: No auth check -- anyone can connect and listen to all events
io.on('connection', (socket) => {
  socket.on('watch:order', (orderId: string) => {
    socket.join(`order:${orderId}`);  // No authorization check
  });
});

// RIGHT: Verify token before connection is established
io.use((socket, next) => {
  const token = socket.handshake.auth.token;
  try {
    socket.data.user = verifyToken(token);
    next();
  } catch {
    next(new Error('Authentication failed'));
  }
});

import { io, Socket } from 'socket.io-client';

let socket: Socket | null = null;

export function connectSocket(authToken: string): Socket {
  if (socket?.connected) return socket;

  socket = io('/', {
    auth: { token: authToken },
    reconnection: true,
    reconnectionAttempts: Infinity,
    reconnectionDelay: 1000,
    reconnectionDelayMax: 30000,    // Cap at 30s
    randomizationFactor: 0.5,       // Jitter to avoid thundering herd
  });

  socket.on('connect', () => {
    console.log('Connected to server');
    // Re-subscribe to rooms after reconnection
    resubscribeAll(socket!);
  });

  socket.on('disconnect', (reason) => {
    console.log(`Disconnected: ${reason}`);
    // 'io server disconnect' means server kicked us -- won't auto-reconnect
    if (reason === 'io server disconnect') {
      socket!.connect();
    }
  });

  socket.on('connect_error', (err) => {
    if (err.message === 'Authentication failed') {
      // Don't keep retrying with a bad token -- refresh it
      refreshAuthToken().then((newToken) => {
        socket!.auth = { token: newToken };
        socket!.connect();
      });
    }
  });

  return socket;
}

function resubscribeAll(socket: Socket) {
  // After reconnection, re-join any rooms we were watching
  const watchedOrders = getWatchedOrderIds();  // From your app state
  for (const orderId of watchedOrders) {
    socket.emit('watch:order', orderId);
  }
}

// WRONG: Reconnects immediately, hammering the server
const ws = new WebSocket(url);
ws.onclose = () => {
  // Instant reconnect -- will DDoS the server if it's overloaded
  new WebSocket(url);
};

// WRONG: Fixed 1-second delay -- all clients reconnect at the same time
// causing "thundering herd" that can crash the server
ws.onclose = () => {
  setTimeout(() => new WebSocket(url), 1000);
};

function createReconnectingWebSocket(
  url: string,
  onMessage: (data: any) => void,
  onStateChange?: (state: ConnectionState) => void,
) {
  let ws: WebSocket | null = null;
  let attempt = 0;
  let disposed = false;

  type ConnectionState = 'connecting' | 'connected' | 'reconnecting' | 'disconnected';

  function setState(state: ConnectionState) {
    onStateChange?.(state);
  }

  function connect() {
    if (disposed) return;

    setState(attempt === 0 ? 'connecting' : 'reconnecting');
    ws = new WebSocket(url);

    ws.onopen = () => {
      attempt = 0;  // Reset on successful connection
      setState('connected');
    };

    ws.onmessage = (e) => {
      onMessage(JSON.parse(e.data));
    };

    ws.onclose = (event) => {
      if (disposed) return;

      // 4000-4099: application-level "do not reconnect" codes
      if (event.code >= 4000 && event.code < 4100) {
        setState('disconnected');
        return;
      }

      const delay = Math.min(1000 * Math.pow(2, attempt), 30000);
      const jitter = delay * (0.5 + Math.random() * 0.5);
      attempt++;
      console.log(`Reconnecting in ${Math.round(jitter)}ms (attempt ${attempt})`);
      setTimeout(connect, jitter);
    };

    ws.onerror = () => {
      // onerror is always followed by onclose -- handle reconnection there
    };
  }

  connect();

  return {
    send: (data: unknown) => {
      if (ws?.readyState === WebSocket.OPEN) {
        ws.send(JSON.stringify(data));
      } else {
        console.warn('Cannot send: WebSocket not open');
      }
    },
    dispose: () => {
      disposed = true;
      ws?.close();
      setState('disconnected');
    },
  };
}

// WRONG: User sees stale data with no indication the connection is dead
const ws = new WebSocket(url);
ws.onmessage = (e) => updateUI(JSON.parse(e.data));
// Connection dies... user stares at frozen screen

// React hook for connection state
type ConnectionState = 'connecting' | 'connected' | 'reconnecting' | 'disconnected';

function useConnectionStatus(): ConnectionState {
  const [status, setStatus] = useState<ConnectionState>('connecting');

  useEffect(() => {
    const socket = getSocket();

    const onConnect = () => setStatus('connected');
    const onDisconnect = () => setStatus('reconnecting');
    const onReconnectFailed = () => setStatus('disconnected');

    socket.on('connect', onConnect);
    socket.on('disconnect', onDisconnect);
    socket.on('reconnect_failed', onReconnectFailed);

    return () => {
      socket.off('connect', onConnect);
      socket.off('disconnect', onDisconnect);
      socket.off('reconnect_failed', onReconnectFailed);
    };
  }, []);

  return status;
}

// Connection banner component
function ConnectionBanner() {
  const status = useConnectionStatus();
  if (status === 'connected') return null;

  return (
    <div role="status" aria-live="polite" className="connection-banner">
      {status === 'reconnecting'
        ? 'Connection lost. Reconnecting...'
        : 'Unable to connect. Please check your network and refresh.'}
    </div>
  );
}

// WRONG: No heartbeats -- proxy kills the connection after 60s of idle time
// Client thinks it's still connected but receives nothing
const ws = new WebSocket(url);

// RIGHT: Heartbeat every 30s keeps the connection alive through proxies
setInterval(() => {
  for (const [clientId, client] of clients) {
    if (client.res.writableEnded) {
      clients.delete(clientId);
      continue;
    }
    client.res.write(': heartbeat\n\n');  // SSE comment -- ignored by EventSource
  }
}, 30_000);

// Server-side: detect dead clients
const HEARTBEAT_INTERVAL = 30_000;
const PONG_TIMEOUT = 10_000;

const clients = new Map<string, { ws: WebSocket; alive: boolean }>();

setInterval(() => {
  for (const [id, client] of clients) {
    if (!client.alive) {
      // Client did not respond to last ping -- terminate
      console.log(`Dead client detected: ${id}`);
      client.ws.terminate();
      clients.delete(id);
      continue;
    }
    client.alive = false;  // Mark as pending
    client.ws.ping();       // Send ping -- expect pong
  }
}, HEARTBEAT_INTERVAL);

// On new connection:
wss.on('connection', (ws) => {
  const id = crypto.randomUUID();
  clients.set(id, { ws, alive: true });

  ws.on('pong', () => {
    const client = clients.get(id);
    if (client) client.alive = true;
  });

  ws.on('close', () => clients.delete(id));
});

// WRONG: If a client is slow, messages pile up in the kernel buffer
// Eventually the server runs out of memory
function broadcast(data: unknown) {
  const message = JSON.stringify(data);
  for (const ws of clients) {
    ws.send(message);  // Never checks if the previous send completed
  }
}

const MAX_BUFFER_SIZE = 1024 * 1024;  // 1 MB

function broadcast(data: unknown) {
  const message = JSON.stringify(data);
  for (const [id, ws] of clients) {
    if (ws.readyState !== WebSocket.OPEN) continue;

    // If the send buffer is too large, this client is too slow
    if (ws.bufferedAmount > MAX_BUFFER_SIZE) {
      console.warn(`Dropping slow client ${id} (buffer: ${ws.bufferedAmount} bytes)`);
      ws.close(4001, 'Too slow');
      clients.delete(id);
      continue;
    }

    ws.send(message);
  }
}

function broadcastSSE(event: string, data: unknown) {
  const message = `id: ${++eventId}\nevent: ${event}\ndata: ${JSON.stringify(data)}\n\n`;
  for (const [clientId, client] of clients) {
    if (client.res.writableEnded) {
      clients.delete(clientId);
      continue;
    }
    // write() returns false when internal buffer is full (backpressure)
    const ok = client.res.write(message);
    if (!ok) {
      // Optionally: disconnect slow clients or skip messages for them
      console.warn(`Backpressure on client ${clientId}`);
    }
  }
}

// WRONG: Messages can arrive out of order after reconnection
socket.on('order:status', (data) => {
  setOrderStatus(data.status);  // May overwrite a newer status with an older one
});

// RIGHT: Only apply updates that are newer than what we have
const lastSeenSequence = new Map<string, number>();

socket.on('order:status', (data: { id: string; status: string; seq: number }) => {
  const lastSeq = lastSeenSequence.get(data.id) ?? -1;
  if (data.seq <= lastSeq) {
    // Stale or duplicate message -- ignore
    return;
  }
  lastSeenSequence.set(data.id, data.seq);
  setOrderStatus(data.id, data.status);
});

// RIGHT: Use a seen-set with TTL to deduplicate messages
const recentMessageIds = new Set<string>();

function handleMessage(msg: { messageId: string; payload: unknown }) {
  if (recentMessageIds.has(msg.messageId)) {
    return;  // Already processed this message
  }
  recentMessageIds.add(msg.messageId);

  // Clean up old entries periodically to avoid memory leak
  setTimeout(() => recentMessageIds.delete(msg.messageId), 60_000);

  processPayload(msg.payload);
}

// WRONG: Every connected client receives every order update
io.emit('order:updated', order);  // 10,000 clients get 10,000 irrelevant updates

// RIGHT: Only clients watching this specific order receive the update
io.to(`order:${order.id}`).emit('order:status', {
  id: order.id,
  status: order.status,
});

socket.on('disconnect', () => {
  // Socket.IO auto-removes from rooms on disconnect.
  // For custom implementations, explicitly clean up:
  for (const room of socket.rooms) {
    socket.leave(room);
  }
});

function connectRealtime(onUpdate: (data: any) => void): () => void {
  // Try SSE first
  if (typeof EventSource !== 'undefined') {
    const es = new EventSource('/api/events');
    let connected = false;

    es.onopen = () => { connected = true; };
    es.addEventListener('update', (e) => onUpdate(JSON.parse(e.data)));

    es.onerror = () => {
      if (!connected) {
        // SSE never connected -- fall back to polling
        es.close();
        return startPolling(onUpdate);
      }
      // Otherwise, EventSource will auto-reconnect
    };

    return () => es.close();
  }

  // EventSource not available -- poll
  return startPolling(onUpdate);
}

function startPolling(onUpdate: (data: any) => void): () => void {
  let active = true;
  let etag = '';

  async function poll() {
    while (active) {
      try {
        const res = await fetch('/api/current-state', {
          headers: etag ? { 'If-None-Match': etag } : {},
        });
        if (res.status === 200) {
          etag = res.headers.get('etag') || '';
          const data = await res.json();
          onUpdate(data);
        }
        // 304 Not Modified -- no new data
      } catch {
        // Network error -- wait longer before retrying
        await new Promise((r) => setTimeout(r, 10_000));
        continue;
      }
      await new Promise((r) => setTimeout(r, 5_000));
    }
  }

  poll();
  return () => { active = false; };
}

// WRONG: After reconnect, client shows stale data from before the disconnect
socket.on('connect', () => {
  console.log('Reconnected!');
  // ...and that's it. No state recovery. UI shows data from 5 minutes ago.
});

// RIGHT: After reconnection, fetch the current state to fill any gaps
socket.on('connect', () => {
  // Re-subscribe to rooms
  for (const orderId of watchedOrders) {
    socket.emit('watch:order', orderId);
  }

  // Fetch current state to fill any gaps during disconnect
  fetch('/api/orders/current-state')
    .then((res) => res.json())
    .then((orders) => {
      for (const order of orders) {
        updateOrderInUI(order);
      }
    });
});