Real-Time & WebSocket

const WebSocket = require('ws');
const http = require('http');

const server = http.createServer(app);
const wss = new WebSocket.Server({ server });

// Room management: orderId → Set of client WebSockets
const orderRooms = new Map();

wss.on('connection', (ws, req) => {
  // Authenticate via JWT in query param (can't set headers in WS from browser)
  const token = new URL(req.url, 'ws://x').searchParams.get('token');
  let user;
  try {
    user = verifyToken(token);
  } catch {
    ws.close(4001, 'Unauthorized');
    return;
  }

  ws.user = user;
  ws.isAlive = true;

  // Client sends: { type: 'subscribe', orderId: 'abc123', lastEventId: '...' }
  ws.on('message', (data) => {
    const msg = JSON.parse(data);

    if (msg.type === 'subscribe') {
      // Join room
      if (!orderRooms.has(msg.orderId)) orderRooms.set(msg.orderId, new Set());
      orderRooms.get(msg.orderId).add(ws);
      ws.subscribedOrders = ws.subscribedOrders || new Set();
      ws.subscribedOrders.add(msg.orderId);

      // Event replay: send missed events since lastEventId
      if (msg.lastEventId) {
        replayMissedEvents(ws, msg.orderId, msg.lastEventId);
      }
    }
  });

  // Ping-pong heartbeat: mark connection dead if no pong received
  ws.on('pong', () => { ws.isAlive = true; });

  ws.on('close', () => {
    // Remove from all rooms on disconnect
    ws.subscribedOrders?.forEach(orderId => {
      orderRooms.get(orderId)?.delete(ws);
      if (orderRooms.get(orderId)?.size === 0) orderRooms.delete(orderId);
    });
  });
});

// Heartbeat interval: detect and close zombie connections
const heartbeat = setInterval(() => {
  wss.clients.forEach(ws => {
    if (!ws.isAlive) return ws.terminate(); // dead connection — remove
    ws.isAlive = false;
    ws.ping(); // client must respond with pong within next interval
  });
}, 30000); // every 30 seconds

Multi-instance WebSocket scaling: EC2 Instance A EC2 Instance B ┌───────────────┐ ┌───────────────┐ │ ws-server │ │ ws-server │ │ │ │ │ │ customer1 ──┐ │ │ merchant1 ──┐ │ │ customer2 ──┤ │ │ rider1 ──┤ │ └─────────────┼─┘ └─────────────┼─┘ │ │ └─────────┬─────────────────┘ │ Redis Pub/Sub channel: 'order:abc' │ When order updates: │ Any instance publishes │ → ALL instances receive to Redis → ALL │ → Each broadcasts to instances broadcast │ their local subscribers

// Publisher (any instance that processes an order update)
function broadcastOrderUpdate(order) {
  const payload = JSON.stringify({
    type: 'ORDER_UPDATE',
    orderId: order._id,
    status: order.status,
    tenantId: order.tenantId,
    eventId: generateEventId(), // for event replay
    timestamp: Date.now(),
  });

  // Store in Redis sorted set for replay (score = timestamp)
  redis.zadd(`events:order:${order._id}`, Date.now(), payload);
  redis.expire(`events:order:${order._id}`, 600); // 10 min TTL

  // Publish to all instances
  redisPub.publish(`order:${order._id}`, payload);
}

// Subscriber setup (each ws-server instance)
redisSub.subscribe('order:*'); // subscribe to all order channels
redisSub.on('message', (channel, message) => {
  const orderId = channel.split(':')[1];
  const clients = orderRooms.get(orderId) || new Set();
  clients.forEach(ws => {
    if (ws.readyState === WebSocket.OPEN) ws.send(message);
  });
});

Event Replay

Handling reconnects without missing updates

// Client sends lastEventId on reconnect
// { type: 'subscribe', orderId: 'abc', lastEventId: '1710000000000:abc' }

async function replayMissedEvents(ws, orderId, lastEventId) {
  const lastTimestamp = parseInt(lastEventId.split(':')[0]);

  // Fetch all events after lastTimestamp from Redis sorted set
  const events = await redis.zrangebyscore(
    `events:order:${orderId}`,
    lastTimestamp + 1,  // exclusive: after last seen
    '+inf',
    'WITHSCORES'
  );

  // Send missed events in order
  events.forEach(event => {
    if (ws.readyState === WebSocket.OPEN) {
      ws.send(event);
    }
  });
}

Automatic Reconnection (Flutter Client)

// Flutter WebSocket with exponential backoff reconnection
class OrderWebSocketService {
  WebSocketChannel? _channel;
  int _retryCount = 0;
  static const maxRetries = 10;
  String? _lastEventId;

  void connect(String orderId) {
    final uri = Uri.parse(
      'wss://api.zoober.in/ws?token=${authToken}&lastEventId=$_lastEventId'
    );
    _channel = WebSocketChannel.connect(uri);
    _channel!.stream.listen(
      (message) {
        _retryCount = 0; // reset on successful message
        final data = jsonDecode(message);
        _lastEventId = data['eventId']; // track for replay
        _handleMessage(data);
      },
      onDone: () => _scheduleReconnect(orderId),
      onError: (_) => _scheduleReconnect(orderId),
    );
    _channel!.sink.add(jsonEncode({
      'type': 'subscribe',
      'orderId': orderId,
      'lastEventId': _lastEventId,
    }));
  }

  void _scheduleReconnect(String orderId) {
    if (_retryCount >= maxRetries) return;
    // Exponential backoff with jitter to avoid thundering herd
    final delay = Duration(
      milliseconds: min(
        (pow(2, _retryCount) * 1000 + Random().nextInt(1000)).toInt(),
        30000, // cap at 30s
      ),
    );
    Future.delayed(delay, () => connect(orderId));
    _retryCount++;
  }
}