Websocket Realtime

WebSocket & Real-Time

This skill covers the server and client patterns for WebSocket and SSE connections in production — beyond the happy path.

When to Use

• Implementing chat, live notifications, or collaborative editing
• Streaming LLM responses to the frontend (SSE is ideal for this)
• Building presence indicators (who is online, who is typing)
• Any feature where the server needs to push updates to clients without polling
• An existing WebSocket implementation is dropping connections or failing under load

When NOT to Use

• Simple polling for infrequent updates — polling is simpler and sufficient for < 1 update/minute
• Webhook delivery to other servers — use HTTP POST callbacks
• Background job status that is checked once — use a regular REST endpoint

Protocol Choice

SSE runs over HTTP — no special proxy config, works through load balancers out of the box. WebSocket requires Upgrade header support (most modern load balancers support this).

Server-Sent Events (SSE)

SSE is the right choice for LLM streaming and unidirectional server push:

NestJS / Express SSE endpoint:

@Get('stream/:id')
@Sse()
streamDocument(@Param('id') id: string, @Req() req: Request): Observable<MessageEvent> {
  return new Observable((observer) => {
    const stream = this.documentService.streamAnalysis(id);

    stream.on('data', (chunk: string) => {
      observer.next({ data: JSON.stringify({ content: chunk }) });
    });

    stream.on('end', () => {
      observer.next({ data: JSON.stringify({ done: true }) });
      observer.complete();
    });

    stream.on('error', (err) => {
      observer.next({ data: JSON.stringify({ error: err.message }) });
      observer.complete();
    });

    // Clean up on client disconnect
    req.on('close', () => stream.destroy());
  });
}

FastAPI SSE endpoint (for thesis streaming):

from fastapi import Request
from fastapi.responses import StreamingResponse
import asyncio

@router.get("/documents/{doc_id}/stream")
async def stream_analysis(doc_id: str, request: Request):
    async def event_generator():
        async for chunk in document_service.stream_analysis(doc_id):
            if await request.is_disconnected():
                break
            yield f"data: {json.dumps({'content': chunk})}\n\n"
        yield f"data: {json.dumps({'done': True})}\n\n"

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",   # disable nginx buffering
        },
    )

SSE client with automatic reconnect:

function createSSEConnection(url: string, onMessage: (data: unknown) => void) {
  const eventSource = new EventSource(url, { withCredentials: true });

  eventSource.onmessage = (e) => {
    const data = JSON.parse(e.data);
    if (data.done) {
      eventSource.close();
      return;
    }
    onMessage(data);
  };

  eventSource.onerror = () => {
    // EventSource reconnects automatically — onerror does not mean permanent failure
    // Only close explicitly when done or auth fails
  };

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

The browser's EventSource API reconnects automatically with the Last-Event-ID header — no manual retry logic needed for basic cases.

WebSocket

Server (Node.js with ws library)

import { WebSocketServer, WebSocket } from 'ws';
import { IncomingMessage } from 'http';

const wss = new WebSocketServer({ noServer: true });

// Connection map for broadcasting
const connections = new Map<string, Set<WebSocket>>();

wss.on('connection', (ws: WebSocket, req: IncomingMessage, userId: string) => {
  // Register connection
  if (!connections.has(userId)) connections.set(userId, new Set());
  connections.get(userId)!.add(ws);

  // Heartbeat — detect stale connections
  (ws as any).isAlive = true;
  ws.on('pong', () => { (ws as any).isAlive = true; });

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

  ws.on('close', () => {
    connections.get(userId)?.delete(ws);
    if (connections.get(userId)?.size === 0) connections.delete(userId);
  });

  ws.on('error', (err) => {
    logger.error('WebSocket error', { userId, error: err.message });
    ws.terminate();
  });
});

// Heartbeat interval — terminate stale connections
const heartbeat = setInterval(() => {
  wss.clients.forEach((ws) => {
    if (!(ws as any).isAlive) return ws.terminate();
    (ws as any).isAlive = false;
    ws.ping();
  });
}, 30_000);

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

WebSocket Upgrade (HTTP server integration)

httpServer.on('upgrade', async (req, socket, head) => {
  try {
    // Authenticate before upgrading
    const userId = await authenticateWebSocketRequest(req);
    if (!userId) {
      socket.write('HTTP/1.1 401 Unauthorized\r\n\r\n');
      socket.destroy();
      return;
    }

    wss.handleUpgrade(req, socket, head, (ws) => {
      wss.emit('connection', ws, req, userId);
    });
  } catch {
    socket.destroy();
  }
});

Never skip authentication on WebSocket upgrade. The upgrade happens before any application-level auth middleware.

Client with Reconnection

class ReconnectingWebSocket {
  private ws: WebSocket | null = null;
  private reconnectDelay = 1000;
  private maxDelay = 30_000;
  private shouldReconnect = true;

  constructor(
    private url: string,
    private onMessage: (data: unknown) => void,
  ) {
    this.connect();
  }

  private connect() {
    this.ws = new WebSocket(this.url);

    this.ws.onopen = () => {
      this.reconnectDelay = 1000; // reset on successful connection
    };

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

    this.ws.onclose = (e) => {
      if (!this.shouldReconnect || e.code === 1008) return; // 1008 = policy violation (auth fail)
      const jitter = Math.random() * 500;
      setTimeout(() => this.connect(), this.reconnectDelay + jitter);
      this.reconnectDelay = Math.min(this.reconnectDelay * 2, this.maxDelay);
    };

    this.ws.onerror = () => {
      // onerror always precedes onclose — handle reconnect in onclose
    };
  }

  send(data: unknown) {
    if (this.ws?.readyState === WebSocket.OPEN) {
      this.ws.send(JSON.stringify(data));
    }
    // Drop silently if not connected — or queue if the app needs guaranteed delivery
  }

  close() {
    this.shouldReconnect = false;
    this.ws?.close(1000);
  }
}

Reconnection rules:

• Exponential backoff with jitter (prevents thundering herd after server restart)
• Do not reconnect on auth failure (code 1008) — it will never succeed
• Cap at 30s max delay
• Reset delay on successful connection

Backpressure

When the server produces data faster than the client can consume, the WebSocket send buffer fills up. Unchecked, this causes memory exhaustion.

function safeSend(ws: WebSocket, data: unknown): boolean {
  if (ws.readyState !== WebSocket.OPEN) return false;

  // bufferedAmount is the bytes queued but not yet sent
  if (ws.bufferedAmount > 1024 * 1024) { // 1MB threshold
    logger.warn('WebSocket backpressure — dropping message', {
      bufferedAmount: ws.bufferedAmount,
    });
    return false;
  }

  ws.send(JSON.stringify(data));
  return true;
}

For high-throughput streams, implement a draining loop:

async function drainSend(ws: WebSocket, messages: unknown[]) {
  for (const msg of messages) {
    ws.send(JSON.stringify(msg));
    if (ws.bufferedAmount > 512 * 1024) {
      // Wait for drain before continuing
      await new Promise<void>((resolve) => ws.once('drain', resolve));
    }
  }
}

Presence System

Track who is online and propagate state changes:

class PresenceManager {
  // roomId → Map<userId, { lastSeen, metadata }>
  private rooms = new Map<string, Map<string, PresenceEntry>>();

  join(roomId: string, userId: string, metadata: Record<string, unknown>) {
    if (!this.rooms.has(roomId)) this.rooms.set(roomId, new Map());
    this.rooms.get(roomId)!.set(userId, { lastSeen: Date.now(), metadata });
    this.broadcast(roomId, { type: 'presence', userId, status: 'online', metadata });
  }

  leave(roomId: string, userId: string) {
    this.rooms.get(roomId)?.delete(userId);
    if (this.rooms.get(roomId)?.size === 0) this.rooms.delete(roomId);
    this.broadcast(roomId, { type: 'presence', userId, status: 'offline' });
  }

  private broadcast(roomId: string, message: unknown) {
    const room = this.rooms.get(roomId);
    if (!room) return;
    const payload = JSON.stringify(message);
    for (const [uid] of room) {
      getConnectionsForUser(uid)?.forEach((ws) => {
        if (ws.readyState === WebSocket.OPEN) ws.send(payload);
      });
    }
  }
}

Scaling Across Instances

A single WebSocket server cannot broadcast to connections held by other instances. Use Redis pub/sub to fan out:

import { createClient } from 'redis';

const publisher = createClient({ url: process.env.REDIS_URL });
const subscriber = createClient({ url: process.env.REDIS_URL });

await publisher.connect();
await subscriber.connect();

// Subscribe to messages for this server's connections
await subscriber.subscribe('ws:broadcast', (message) => {
  const { roomId, payload } = JSON.parse(message);
  broadcastToLocalConnections(roomId, payload);
});

// Publish from any instance — all instances receive it
async function broadcastToRoom(roomId: string, payload: unknown) {
  await publisher.publish('ws:broadcast', JSON.stringify({ roomId, payload }));
}

For production scale: use socket.io with @socket.io/redis-adapter which handles this pattern including room management.

WebSocket / SSE Checklist

Connection:
- [ ] Authentication before upgrade (not inside message handler)
- [ ] Heartbeat/ping-pong to detect stale connections
- [ ] Graceful close on server shutdown (drain before kill)

Client:
- [ ] Reconnection with exponential backoff + jitter
- [ ] No reconnect on auth failure (1008)
- [ ] Cleanup on component unmount

Backpressure:
- [ ] bufferedAmount checked before sending
- [ ] Messages dropped or queued when buffer is full

Scaling:
- [ ] Redis pub/sub for multi-instance broadcasts
- [ ] Connection count monitored as a metric

Security:
- [ ] Origin validation on WebSocket upgrade
- [ ] Message size limit enforced
- [ ] Rate limiting on message frequency per connection

Anti-Patterns to Avoid

Connected Skills

• observability — track connection count, message rate, and reconnect frequency as metrics
• error-handling — WebSocket error handling follows the same boundary translation principles
• security-hardening — origin validation, rate limiting per connection, message size limits
• python-fastapi — FastAPI SSE streaming for thesis LLM response streaming
• message-queue — for guaranteed delivery, use a queue instead of direct WebSocket push