WebSocket & Real-time Communication Complete Guide 2025: SSE, Socket.IO, Chat/Notifications/Collaboration

// Short Polling implementation
function startPolling(interval = 5000) {
  const poll = async () => {
    try {
      const response = await fetch('/api/messages');
      const data = await response.json();
      updateUI(data);
    } catch (error) {
      console.error('Polling error:', error);
    }
  };

  // Execute immediately then set interval
  poll();
  return setInterval(poll, interval);
}

// Cleanup
const pollId = startPolling(3000);
// clearInterval(pollId);

// Long Polling client
async function longPoll(lastEventId = null) {
  try {
    const url = lastEventId
      ? `/api/events?since=${lastEventId}`
      : '/api/events';

    const response = await fetch(url, {
      signal: AbortSignal.timeout(30000), // 30-second timeout
    });

    if (response.status === 200) {
      const data = await response.json();
      handleNewData(data);
      // Immediately start next long poll
      longPoll(data.lastEventId);
    } else if (response.status === 204) {
      // No data - retry after timeout
      longPoll(lastEventId);
    }
  } catch (error) {
    // Progressive retry on error
    console.error('Long poll error:', error);
    setTimeout(() => longPoll(lastEventId), 3000);
  }
}

// Long Polling server (Express)
app.get('/api/events', async (req, res) => {
  const since = req.query.since;
  const timeout = 25000; // 25 seconds

  const checkForData = () => {
    return new Promise((resolve) => {
      const timer = setTimeout(() => {
        resolve(null); // Timeout
      }, timeout);

      eventEmitter.once('newData', (data) => {
        clearTimeout(timer);
        resolve(data);
      });
    });
  };

  const data = await checkForData();
  if (data) {
    res.json(data);
  } else {
    res.status(204).end();
  }
});

// Client request
GET /chat HTTP/1.1
Host: example.com
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==
Sec-WebSocket-Version: 13
Sec-WebSocket-Protocol: chat, superchat

// Server response
HTTP/1.1 101 Switching Protocols
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=
Sec-WebSocket-Protocol: chat

 0                   1                   2                   3
 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+-+-+-+-+-------+-+-------------+-------------------------------+
|F|R|R|R| opcode|M| Payload len |    Extended payload length    |
|I|S|S|S|  (4)  |A|     (7)     |             (16/64)           |
|N|V|V|V|       |S|             |   (if payload len==126/127)   |
| |1|2|3|       |K|             |                               |
+-+-+-+-+-------+-+-------------+-------------------------------+

import { WebSocketServer } from 'ws';
import { createServer } from 'http';

const server = createServer();
const wss = new WebSocketServer({ server });

// Connected client management
const clients = new Map();

wss.on('connection', (ws, req) => {
  const clientId = generateId();
  clients.set(clientId, { ws, alive: true });

  console.log(`Client connected: ${clientId}`);

  // Ping/Pong heartbeat
  ws.isAlive = true;
  ws.on('pong', () => {
    ws.isAlive = true;
  });

  // Message reception
  ws.on('message', (data, isBinary) => {
    const message = isBinary ? data : data.toString();

    try {
      const parsed = JSON.parse(message);
      handleMessage(clientId, parsed);
    } catch (e) {
      ws.send(JSON.stringify({ error: 'Invalid message format' }));
    }
  });

  // Connection close
  ws.on('close', (code, reason) => {
    console.log(`Client disconnected: ${clientId}, code: ${code}`);
    clients.delete(clientId);
  });

  // Error handling
  ws.on('error', (error) => {
    console.error(`WebSocket error for ${clientId}:`, error);
  });

  // Welcome message
  ws.send(JSON.stringify({
    type: 'welcome',
    clientId,
    timestamp: Date.now(),
  }));
});

// Heartbeat interval
const heartbeat = setInterval(() => {
  wss.clients.forEach((ws) => {
    if (ws.isAlive === false) {
      return ws.terminate();
    }
    ws.isAlive = false;
    ws.ping();
  });
}, 30000);

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

function handleMessage(clientId, message) {
  switch (message.type) {
    case 'broadcast':
      broadcast(clientId, message.payload);
      break;
    case 'direct':
      sendToClient(message.targetId, message.payload);
      break;
    default:
      console.warn(`Unknown message type: ${message.type}`);
  }
}

function broadcast(senderId, payload) {
  const msg = JSON.stringify({
    type: 'broadcast',
    from: senderId,
    payload,
    timestamp: Date.now(),
  });

  clients.forEach((client, id) => {
    if (id !== senderId && client.ws.readyState === 1) {
      client.ws.send(msg);
    }
  });
}

function sendToClient(targetId, payload) {
  const target = clients.get(targetId);
  if (target && target.ws.readyState === 1) {
    target.ws.send(JSON.stringify({ type: 'direct', payload }));
  }
}

server.listen(8080, () => {
  console.log('WebSocket server running on port 8080');
});

class ReconnectingWebSocket {
  constructor(url, options = {}) {
    this.url = url;
    this.maxRetries = options.maxRetries || 10;
    this.baseDelay = options.baseDelay || 1000;
    this.maxDelay = options.maxDelay || 30000;
    this.retryCount = 0;
    this.handlers = new Map();
    this.messageQueue = [];
    this.connect();
  }

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

    this.ws.onopen = () => {
      console.log('WebSocket connected');
      this.retryCount = 0;
      // Send queued messages
      while (this.messageQueue.length > 0) {
        const msg = this.messageQueue.shift();
        this.ws.send(msg);
      }
      this.emit('open');
    };

    this.ws.onmessage = (event) => {
      try {
        const data = JSON.parse(event.data);
        this.emit('message', data);
        if (data.type) {
          this.emit(data.type, data);
        }
      } catch (e) {
        this.emit('message', event.data);
      }
    };

    this.ws.onclose = (event) => {
      this.emit('close', event);
      if (!event.wasClean) {
        this.reconnect();
      }
    };

    this.ws.onerror = (error) => {
      this.emit('error', error);
    };
  }

  reconnect() {
    if (this.retryCount >= this.maxRetries) {
      console.error('Max retries reached');
      this.emit('maxRetriesReached');
      return;
    }

    // Exponential backoff with jitter
    const delay = Math.min(
      this.baseDelay * Math.pow(2, this.retryCount) + Math.random() * 1000,
      this.maxDelay
    );

    console.log(`Reconnecting in ${delay}ms (attempt ${this.retryCount + 1})`);
    this.retryCount++;
    setTimeout(() => this.connect(), delay);
  }

  send(data) {
    const message = typeof data === 'string' ? data : JSON.stringify(data);
    if (this.ws.readyState === WebSocket.OPEN) {
      this.ws.send(message);
    } else {
      this.messageQueue.push(message);
    }
  }

  on(event, handler) {
    if (!this.handlers.has(event)) {
      this.handlers.set(event, []);
    }
    this.handlers.get(event).push(handler);
  }

  emit(event, data) {
    const handlers = this.handlers.get(event) || [];
    handlers.forEach((handler) => handler(data));
  }

  close() {
    this.maxRetries = 0; // Prevent reconnection
    this.ws.close(1000, 'Client closing');
  }
}

// Usage
const ws = new ReconnectingWebSocket('wss://api.example.com/ws');
ws.on('message', (data) => console.log('Received:', data));
ws.send({ type: 'join', room: 'general' });

// Express SSE server
import express from 'express';

const app = express();

// SSE client management
const sseClients = new Map();
let eventCounter = 0;

app.get('/api/events', (req, res) => {
  const clientId = Date.now().toString();

  // SSE headers
  res.writeHead(200, {
    'Content-Type': 'text/event-stream',
    'Cache-Control': 'no-cache',
    'Connection': 'keep-alive',
    'X-Accel-Buffering': 'no', // Disable Nginx buffering
  });

  // Handle Last-Event-ID (recovery on reconnect)
  const lastEventId = req.headers['last-event-id'];
  if (lastEventId) {
    // Resend missed events
    const missedEvents = getEventsSince(parseInt(lastEventId));
    missedEvents.forEach((event) => {
      res.write(`id: ${event.id}\n`);
      res.write(`event: ${event.type}\n`);
      res.write(`data: ${JSON.stringify(event.data)}\n\n`);
    });
  }

  // Register client
  sseClients.set(clientId, res);

  // Set reconnection interval (milliseconds)
  res.write('retry: 5000\n\n');

  // Initial connection confirmation
  res.write(`id: ${++eventCounter}\n`);
  res.write('event: connected\n');
  res.write(`data: ${JSON.stringify({ clientId })}\n\n`);

  // Keep-alive (send comment every 15s)
  const keepAlive = setInterval(() => {
    res.write(':keep-alive\n\n');
  }, 15000);

  // Handle disconnect
  req.on('close', () => {
    clearInterval(keepAlive);
    sseClients.delete(clientId);
    console.log(`SSE client disconnected: ${clientId}`);
  });
});

// Broadcast event to all SSE clients
function broadcastSSE(eventType, data) {
  const id = ++eventCounter;
  sseClients.forEach((res) => {
    res.write(`id: ${id}\n`);
    res.write(`event: ${eventType}\n`);
    res.write(`data: ${JSON.stringify(data)}\n\n`);
  });
}

// Notification endpoint
app.post('/api/notify', express.json(), (req, res) => {
  broadcastSSE('notification', {
    title: req.body.title,
    message: req.body.message,
    timestamp: Date.now(),
  });
  res.json({ success: true, clients: sseClients.size });
});

// EventSource API usage
class SSEClient {
  constructor(url) {
    this.url = url;
    this.handlers = {};
    this.connect();
  }

  connect() {
    this.source = new EventSource(this.url);

    this.source.onopen = () => {
      console.log('SSE connection opened');
    };

    this.source.onerror = (error) => {
      console.error('SSE error:', error);
      if (this.source.readyState === EventSource.CLOSED) {
        console.log('SSE connection closed');
      }
      // EventSource automatically attempts reconnection
    };

    // Register custom events
    Object.entries(this.handlers).forEach(([event, handler]) => {
      this.source.addEventListener(event, handler);
    });
  }

  on(event, handler) {
    const wrappedHandler = (e) => {
      try {
        const data = JSON.parse(e.data);
        handler(data, e);
      } catch {
        handler(e.data, e);
      }
    };

    this.handlers[event] = wrappedHandler;
    if (this.source) {
      this.source.addEventListener(event, wrappedHandler);
    }
  }

  close() {
    if (this.source) {
      this.source.close();
    }
  }
}

// Usage
const sse = new SSEClient('/api/events');
sse.on('notification', (data) => {
  showNotification(data.title, data.message);
});
sse.on('update', (data) => {
  refreshDashboard(data);
});

// Server setup
import { createServer } from 'http';
import { Server } from 'socket.io';

const httpServer = createServer();
const io = new Server(httpServer, {
  cors: {
    origin: ['https://example.com'],
    methods: ['GET', 'POST'],
    credentials: true,
  },
  // Transport configuration
  transports: ['websocket', 'polling'],
  // Ping/Pong settings
  pingTimeout: 20000,
  pingInterval: 25000,
  // Max payload
  maxHttpBufferSize: 1e6, // 1MB
});

// Namespaces - logical separation
const chatNamespace = io.of('/chat');
const adminNamespace = io.of('/admin');

// Chat namespace
chatNamespace.on('connection', (socket) => {
  console.log('User connected to chat:', socket.id);

  // Join room
  socket.on('joinRoom', async (roomName) => {
    await socket.join(roomName);
    // Notify other users in the room
    socket.to(roomName).emit('userJoined', {
      userId: socket.id,
      room: roomName,
    });

    // Send room member count
    const sockets = await chatNamespace.in(roomName).fetchSockets();
    chatNamespace.to(roomName).emit('roomInfo', {
      room: roomName,
      members: sockets.length,
    });
  });

  // Leave room
  socket.on('leaveRoom', async (roomName) => {
    await socket.leave(roomName);
    socket.to(roomName).emit('userLeft', {
      userId: socket.id,
      room: roomName,
    });
  });

  // Send message within room
  socket.on('message', (data) => {
    chatNamespace.to(data.room).emit('message', {
      from: socket.id,
      text: data.text,
      room: data.room,
      timestamp: Date.now(),
    });
  });
});

// Admin namespace - requires authentication
adminNamespace.use((socket, next) => {
  const token = socket.handshake.auth.token;
  if (verifyAdminToken(token)) {
    next();
  } else {
    next(new Error('Unauthorized'));
  }
});

// Global middleware
io.use((socket, next) => {
  const token = socket.handshake.auth.token;
  if (!token) {
    return next(new Error('Authentication required'));
  }

  try {
    const user = verifyToken(token);
    socket.data.user = user;
    next();
  } catch (err) {
    next(new Error('Invalid token'));
  }
});

// Namespace middleware
chatNamespace.use((socket, next) => {
  // Additional permission checks
  if (socket.data.user.role === 'banned') {
    return next(new Error('User is banned'));
  }
  next();
});

// Per-event middleware after connection
io.on('connection', (socket) => {
  // Logging middleware for all events
  socket.use(([event, ...args], next) => {
    console.log(`Event: ${event}, User: ${socket.data.user.name}`);
    next();
  });

  // Rate limiting middleware
  const rateLimiter = createRateLimiter(100, 60000); // 100 per minute
  socket.use(([event, ...args], next) => {
    if (rateLimiter.isAllowed(socket.id)) {
      next();
    } else {
      next(new Error('Rate limit exceeded'));
    }
  });
});

// Server
io.on('connection', (socket) => {
  socket.on('createOrder', async (orderData, callback) => {
    try {
      const order = await processOrder(orderData);
      // Success response
      callback({
        status: 'ok',
        orderId: order.id,
      });
    } catch (error) {
      // Error response
      callback({
        status: 'error',
        message: error.message,
      });
    }
  });
});

// Client
socket.emit('createOrder', { item: 'laptop', qty: 1 }, (response) => {
  if (response.status === 'ok') {
    console.log('Order created:', response.orderId);
  } else {
    console.error('Order failed:', response.message);
  }
});

// Emit with timeout
socket.timeout(5000).emit('createOrder', orderData, (err, response) => {
  if (err) {
    // Server did not respond within 5 seconds
    console.error('Timeout');
  } else {
    console.log('Response:', response);
  }
});

// Server: Message ordering management
class ChatRoom {
  constructor(roomId) {
    this.roomId = roomId;
    this.messageSequence = 0;
    this.pendingMessages = new Map();
  }

  async addMessage(userId, content) {
    const sequence = ++this.messageSequence;
    const message = {
      id: `${this.roomId}-${sequence}`,
      sequence,
      roomId: this.roomId,
      userId,
      content,
      timestamp: Date.now(),
      status: 'sent',
    };

    // Persist to storage
    await this.persistMessage(message);

    return message;
  }

  async getMessagesSince(sequence, limit = 50) {
    // Retrieve messages after a specific sequence
    return await db.messages
      .find({
        roomId: this.roomId,
        sequence: { $gt: sequence },
      })
      .sort({ sequence: 1 })
      .limit(limit)
      .toArray();
  }
}

// Presence system
class PresenceManager {
  constructor(io, redis) {
    this.io = io;
    this.redis = redis;
    this.TTL = 60; // 60 seconds
  }

  async setOnline(userId, socketId) {
    const key = `presence:${userId}`;
    await this.redis.hset(key, {
      status: 'online',
      socketId,
      lastSeen: Date.now().toString(),
    });
    await this.redis.expire(key, this.TTL);

    // Notify friends of status change
    const friends = await this.getFriends(userId);
    friends.forEach((friendId) => {
      this.io.to(`user:${friendId}`).emit('presence', {
        userId,
        status: 'online',
      });
    });
  }

  async setOffline(userId) {
    const key = `presence:${userId}`;
    await this.redis.hset(key, {
      status: 'offline',
      lastSeen: Date.now().toString(),
    });

    const friends = await this.getFriends(userId);
    friends.forEach((friendId) => {
      this.io.to(`user:${friendId}`).emit('presence', {
        userId,
        status: 'offline',
        lastSeen: Date.now(),
      });
    });
  }

  async heartbeat(userId) {
    const key = `presence:${userId}`;
    await this.redis.expire(key, this.TTL);
  }

  async getStatus(userId) {
    const data = await this.redis.hgetall(`presence:${userId}`);
    return data || { status: 'offline' };
  }
}

// Client
let typingTimeout;

messageInput.addEventListener('input', () => {
  socket.emit('typing', { room: currentRoom });

  clearTimeout(typingTimeout);
  typingTimeout = setTimeout(() => {
    socket.emit('stopTyping', { room: currentRoom });
  }, 2000);
});

// Server
socket.on('typing', (data) => {
  socket.to(data.room).emit('typing', {
    userId: socket.data.user.id,
    userName: socket.data.user.name,
  });
});

socket.on('stopTyping', (data) => {
  socket.to(data.room).emit('stopTyping', {
    userId: socket.data.user.id,
  });
});

// Server: Read receipt handling
socket.on('markRead', async (data) => {
  const { roomId, messageId, userId } = data;

  // Save read status to DB
  await db.readReceipts.upsert({
    roomId,
    userId,
    lastReadMessageId: messageId,
    readAt: new Date(),
  });

  // Notify room participants
  socket.to(roomId).emit('readReceipt', {
    userId,
    messageId,
    readAt: Date.now(),
  });
});

// Client: Display read receipts
socket.on('readReceipt', (receipt) => {
  updateMessageStatus(receipt.messageId, receipt.userId, 'read');
});

// Auto-detect message visibility with Intersection Observer
const observer = new IntersectionObserver(
  (entries) => {
    entries.forEach((entry) => {
      if (entry.isIntersecting) {
        const messageId = entry.target.dataset.messageId;
        socket.emit('markRead', {
          roomId: currentRoom,
          messageId,
          userId: currentUser.id,
        });
        observer.unobserve(entry.target);
      }
    });
  },
  { threshold: 0.5 }
);

// Notification service
class NotificationService {
  constructor(io, redis) {
    this.io = io;
    this.redis = redis;
  }

  async send(userId, notification) {
    const enriched = {
      id: generateId(),
      ...notification,
      createdAt: Date.now(),
      read: false,
    };

    // Persist to database
    await db.notifications.insertOne({
      userId,
      ...enriched,
    });

    // Real-time delivery
    this.io.to(`user:${userId}`).emit('notification', enriched);

    // Queue push notification if offline
    const isOnline = await this.isUserOnline(userId);
    if (!isOnline) {
      await this.queuePushNotification(userId, enriched);
    }

    return enriched;
  }

  async sendBatch(userIds, notification) {
    // Efficient batch processing for bulk notifications
    const enriched = {
      id: generateId(),
      ...notification,
      createdAt: Date.now(),
      read: false,
    };

    // Bulk insert
    const docs = userIds.map((userId) => ({
      userId,
      ...enriched,
    }));
    await db.notifications.insertMany(docs);

    // Real-time delivery to online users
    userIds.forEach((userId) => {
      this.io.to(`user:${userId}`).emit('notification', enriched);
    });
  }

  async markAsRead(userId, notificationIds) {
    await db.notifications.updateMany(
      { _id: { $in: notificationIds }, userId },
      { $set: { read: true, readAt: new Date() } }
    );

    // Update unread count
    const unreadCount = await db.notifications.countDocuments({
      userId,
      read: false,
    });

    this.io.to(`user:${userId}`).emit('unreadCount', unreadCount);
  }
}

// High-frequency event batching
class NotificationBatcher {
  constructor(flushInterval = 1000, maxBatchSize = 50) {
    this.buffer = new Map(); // userId -> notifications[]
    this.flushInterval = flushInterval;
    this.maxBatchSize = maxBatchSize;

    setInterval(() => this.flush(), this.flushInterval);
  }

  add(userId, notification) {
    if (!this.buffer.has(userId)) {
      this.buffer.set(userId, []);
    }

    const userBuffer = this.buffer.get(userId);
    userBuffer.push(notification);

    // Flush immediately if batch size exceeded
    if (userBuffer.length >= this.maxBatchSize) {
      this.flushUser(userId);
    }
  }

  flush() {
    this.buffer.forEach((notifications, userId) => {
      if (notifications.length > 0) {
        this.flushUser(userId);
      }
    });
  }

  flushUser(userId) {
    const notifications = this.buffer.get(userId) || [];
    if (notifications.length === 0) return;

    this.buffer.set(userId, []);

    // Deduplicate and group
    const grouped = this.groupNotifications(notifications);
    io.to(`user:${userId}`).emit('notifications:batch', grouped);
  }

  groupNotifications(notifications) {
    // Group notifications by type
    const groups = {};
    notifications.forEach((n) => {
      const key = n.type;
      if (!groups[key]) groups[key] = [];
      groups[key].push(n);
    });
    return groups;
  }
}

// OT basic concept
// When two users edit simultaneously:
// User A: insert('X', position=0) - Insert 'X' at position 0
// User B: insert('Y', position=2) - Insert 'Y' at position 2

// Transform A's operation from B's perspective:
// B inserted at position 2, so A's operation is unaffected (position 0)

// Transform B's operation from A's perspective:
// A inserted at position 0, so B's position shifts to 2+1=3

function transformInsert(opA, opB) {
  if (opA.position <= opB.position) {
    return { ...opB, position: opB.position + 1 };
  }
  return opB;
}

// CRDT-based collaborative editing with Y.js
import * as Y from 'yjs';
import { WebsocketProvider } from 'y-websocket';

// Create shared document
const ydoc = new Y.Doc();

// Connect WebSocket provider
const provider = new WebsocketProvider(
  'wss://collaboration.example.com',
  'document-room-123',
  ydoc
);

// Shared text
const ytext = ydoc.getText('content');

// Observe text changes
ytext.observe((event) => {
  console.log('Text changed:', event.delta);
});

// Edit text
ytext.insert(0, 'Hello, ');
ytext.insert(7, 'World!');

// Shared map (key-value)
const ymap = ydoc.getMap('metadata');
ymap.set('title', 'My Document');
ymap.set('lastEdited', Date.now());

// Shared array
const yarray = ydoc.getArray('items');
yarray.push(['item1', 'item2']);

// Connection status monitoring
provider.on('status', (event) => {
  console.log('Connection status:', event.status);
});

// Awareness (cursor positions, user info)
const awareness = provider.awareness;
awareness.setLocalState({
  user: { name: 'Alice', color: '#ff0000' },
  cursor: { position: 42 },
});

awareness.on('change', () => {
  const states = awareness.getStates();
  updateCursors(states);
});

import { Server } from 'socket.io';
import { createAdapter } from '@socket.io/redis-adapter';
import { createClient } from 'redis';

const io = new Server(httpServer);

// Create Redis clients
const pubClient = createClient({ url: 'redis://redis-host:6379' });
const subClient = pubClient.duplicate();

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

// Apply Redis Adapter
io.adapter(createAdapter(pubClient, subClient));

// Now broadcasting works across all server instances
io.emit('globalEvent', { message: 'All servers receive this' });

// Send to specific room (reaches clients on all servers)
io.to('room-123').emit('roomEvent', { data: 'value' });

# Nginx configuration
upstream websocket_backend {
    ip_hash;  # Sticky Session
    server backend1:3000;
    server backend2:3000;
    server backend3:3000;
}

server {
    listen 80;

    location /socket.io/ {
        proxy_pass http://websocket_backend;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;

        # Timeout settings
        proxy_read_timeout 86400s;
        proxy_send_timeout 86400s;
    }
}

import { connect } from 'nats';

class NATSAdapter {
  constructor(io) {
    this.io = io;
    this.serverId = process.env.SERVER_ID || generateId();
  }

  async connect() {
    this.nc = await connect({ servers: 'nats://nats-host:4222' });
    this.js = this.nc.jetstream();

    // Subscribe to broadcast channel
    const sub = this.nc.subscribe('ws.broadcast.>');
    (async () => {
      for await (const msg of sub) {
        const data = JSON.parse(msg.data.toString());
        // Ignore messages from self
        if (data.serverId === this.serverId) continue;

        const topic = msg.subject.split('.').slice(2).join('.');
        this.io.to(topic).emit(data.event, data.payload);
      }
    })();
  }

  broadcast(room, event, payload) {
    this.nc.publish(`ws.broadcast.${room}`, JSON.stringify({
      serverId: this.serverId,
      event,
      payload,
    }));
  }
}

// Token-based authentication
io.use(async (socket, next) => {
  try {
    const token = socket.handshake.auth.token
      || socket.handshake.query.token;

    if (!token) {
      return next(new Error('Authentication token required'));
    }

    // JWT verification
    const payload = await verifyJWT(token);

    // Check token expiration
    if (payload.exp * 1000 < Date.now()) {
      return next(new Error('Token expired'));
    }

    // Attach user info
    socket.data.user = {
      id: payload.sub,
      roles: payload.roles,
    };

    next();
  } catch (err) {
    next(new Error('Invalid authentication token'));
  }
});

// Periodic token refresh
socket.on('refreshToken', async (refreshToken, callback) => {
  try {
    const newToken = await refreshAccessToken(refreshToken);
    callback({ status: 'ok', token: newToken });
  } catch (err) {
    callback({ status: 'error', message: 'Token refresh failed' });
    socket.disconnect();
  }
});

// Token bucket algorithm
class TokenBucket {
  constructor(capacity, refillRate) {
    this.capacity = capacity;
    this.tokens = capacity;
    this.refillRate = refillRate; // Tokens added per second
    this.lastRefill = Date.now();
  }

  consume(count = 1) {
    this.refill();
    if (this.tokens >= count) {
      this.tokens -= count;
      return true;
    }
    return false;
  }

  refill() {
    const now = Date.now();
    const elapsed = (now - this.lastRefill) / 1000;
    this.tokens = Math.min(
      this.capacity,
      this.tokens + elapsed * this.refillRate
    );
    this.lastRefill = now;
  }
}

// Apply to Socket.IO
const rateLimiters = new Map();

io.use((socket, next) => {
  const userId = socket.data.user.id;
  if (!rateLimiters.has(userId)) {
    rateLimiters.set(userId, new TokenBucket(100, 10)); // 100 capacity, 10/sec refill
  }
  next();
});

socket.use(([event, ...args], next) => {
  const bucket = rateLimiters.get(socket.data.user.id);
  if (bucket.consume()) {
    next();
  } else {
    next(new Error('Rate limit exceeded'));
  }
});

const io = new Server(httpServer, {
  cors: {
    origin: (origin, callback) => {
      const allowedOrigins = [
        'https://app.example.com',
        'https://admin.example.com',
      ];

      if (!origin || allowedOrigins.includes(origin)) {
        callback(null, true);
      } else {
        callback(new Error('Not allowed by CORS'));
      }
    },
    credentials: true,
  },
});

// Additional origin validation middleware
io.use((socket, next) => {
  const origin = socket.handshake.headers.origin;
  const allowedOrigins = process.env.ALLOWED_ORIGINS?.split(',') || [];

  if (allowedOrigins.length > 0 && !allowedOrigins.includes(origin)) {
    return next(new Error('Origin not allowed'));
  }
  next();
});

// permessage-deflate extension
import { WebSocketServer } from 'ws';

const wss = new WebSocketServer({
  port: 8080,
  perMessageDeflate: {
    zlibDeflateOptions: {
      chunkSize: 1024,
      memLevel: 7,
      level: 3, // Compression level (1-9)
    },
    zlibInflateOptions: {
      chunkSize: 10 * 1024,
    },
    // Only compress above threshold
    threshold: 1024, // Only messages above 1KB
    clientNoContextTakeover: true,
    serverNoContextTakeover: true,
    serverMaxWindowBits: 10,
    concurrencyLimit: 10,
  },
});

// message.proto
syntax = "proto3";

message ChatMessage {
  string id = 1;
  string room_id = 2;
  string user_id = 3;
  string content = 4;
  int64 timestamp = 5;
  MessageType type = 6;

  enum MessageType {
    TEXT = 0;
    IMAGE = 1;
    FILE = 2;
    SYSTEM = 3;
  }
}

message WebSocketFrame {
  string event = 1;
  oneof payload {
    ChatMessage chat_message = 2;
    PresenceUpdate presence = 3;
    TypingIndicator typing = 4;
  }
}

// Using Protobuf
import protobuf from 'protobufjs';

const root = await protobuf.load('message.proto');
const ChatMessage = root.lookupType('ChatMessage');
const WebSocketFrame = root.lookupType('WebSocketFrame');

// Encoding
function encodeMessage(event, payload) {
  const frame = WebSocketFrame.create({ event, chatMessage: payload });
  return WebSocketFrame.encode(frame).finish();
}

// Decoding
function decodeMessage(buffer) {
  return WebSocketFrame.decode(new Uint8Array(buffer));
}

// Binary transmission over WebSocket
ws.on('message', (data, isBinary) => {
  if (isBinary) {
    const frame = decodeMessage(data);
    handleFrame(frame);
  }
});

// Connection limit per user
class ConnectionManager {
  constructor(maxConnectionsPerUser = 5) {
    this.connections = new Map(); // userId -> Set of socketIds
    this.maxPerUser = maxConnectionsPerUser;
  }

  addConnection(userId, socketId) {
    if (!this.connections.has(userId)) {
      this.connections.set(userId, new Set());
    }

    const userConnections = this.connections.get(userId);

    // Disconnect oldest if max connections exceeded
    if (userConnections.size >= this.maxPerUser) {
      const oldest = userConnections.values().next().value;
      io.sockets.sockets.get(oldest)?.disconnect();
      userConnections.delete(oldest);
    }

    userConnections.add(socketId);
  }

  removeConnection(userId, socketId) {
    const userConnections = this.connections.get(userId);
    if (userConnections) {
      userConnections.delete(socketId);
      if (userConnections.size === 0) {
        this.connections.delete(userId);
      }
    }
  }

  getConnectionCount() {
    let total = 0;
    this.connections.forEach((sockets) => {
      total += sockets.size;
    });
    return total;
  }
}

// WebSocket server health check
app.get('/health', (req, res) => {
  const health = {
    status: 'ok',
    uptime: process.uptime(),
    connections: io.engine.clientsCount,
    rooms: io.sockets.adapter.rooms.size,
    memory: process.memoryUsage(),
    timestamp: Date.now(),
  };

  // Check connection threshold
  if (health.connections > 10000) {
    health.status = 'degraded';
    res.status(503);
  }

  res.json(health);
});

// Kubernetes readiness/liveness probes
app.get('/ready', (req, res) => {
  if (isReady) {
    res.status(200).json({ ready: true });
  } else {
    res.status(503).json({ ready: false });
  }
});