War Story: We Migrated from Socket.io 4.7 to Ably 2.0 and Cut Chat Latency 40% for Global Users

// Pre-migration Socket.io 4.7 chat server implementation
// Dependencies: socket.io@4.7.2, redis@4.6.12, @socket.io/redis-adapter@8.3.0
const { Server } = require('socket.io');
const { createClient } = require('redis');
const { createAdapter } = require('@socket.io/redis-adapter');

// Redis config for Socket.io horizontal scaling
const pubClient = createClient({ url: 'redis://redis-cluster.internal:6379' });
const subClient = pubClient.duplicate();

// Error handling for Redis connections
pubClient.on('error', (err) => {
  console.error(`[Redis Pub] Connection error: ${err.message}`);
  process.exit(1); // Crash if Redis is unavailable pre-migration (no retry logic)
});
subClient.on('error', (err) => {
  console.error(`[Redis Sub] Connection error: ${err.message}`);
  process.exit(1);
});

// Initialize Socket.io server with WebSocket + long-polling fallback
const io = new Server({
  cors: {
    origin: ['https://chat.example.com'],
    methods: ['GET', 'POST'],
    credentials: true
  },
  transports: ['websocket', 'polling'], // Polling first caused high latency for APAC users
  pingInterval: 25000,
  pingTimeout: 5000,
  maxHttpBufferSize: 1e6 // 1MB max message size
});

// Attach Redis adapter for multi-node scaling
io.adapter(createAdapter(pubClient, subClient));

// Connection handler for chat clients
io.on('connection', (socket) => {
  console.log(`[Socket.io] New connection: ${socket.id} from ${socket.handshake.address}`);

  // Authenticate user via JWT (simplified for example)
  const token = socket.handshake.auth.token;
  if (!token) {
    socket.emit('error', { code: 'AUTH_MISSING', message: 'Authentication token required' });
    socket.disconnect(true);
    return;
  }

  let userId;
  try {
    const decoded = verifyJWT(token); // Assume verifyJWT is imported
    userId = decoded.sub;
    socket.join(`user-${userId}`); // Join user's private room for DMs
  } catch (err) {
    socket.emit('error', { code: 'AUTH_INVALID', message: 'Invalid authentication token' });
    socket.disconnect(true);
    return;
  }

  // Handle chat message send
  socket.on('chat:send', async (message, ack) => {
    try {
      // Validate message payload
      if (!message?.content || typeof message.content !== 'string') {
        throw new Error('Invalid message content');
      }
      if (message.content.length > 1000) {
        throw new Error('Message exceeds 1000 character limit');
      }

      // Persist message to database (simplified)
      const persistedMessage = await persistMessage({
        senderId: userId,
        roomId: message.roomId,
        content: message.content,
        timestamp: Date.now()
      });

      // Broadcast to room via Redis adapter
      io.to(message.roomId).emit('chat:receive', persistedMessage);

      // Send acknowledgement to sender
      ack({ status: 'success', messageId: persistedMessage.id });
    } catch (err) {
      console.error(`[Socket.io] Failed to send message from ${userId}: ${err.message}`);
      ack({ status: 'error', message: err.message });
      socket.emit('error', { code: 'SEND_FAILED', message: 'Failed to send message' });
    }
  });

  // Handle disconnection
  socket.on('disconnect', (reason) => {
    console.log(`[Socket.io] Disconnected: ${socket.id} (${reason})`);
    // No cleanup logic pre-migration: caused ghost connections
  });
});

// Start server
const PORT = process.env.PORT || 3001;
io.listen(PORT, () => {
  console.log(`[Socket.io] Server listening on port ${PORT}`);
});

// Connect Redis clients (blocking, no retry)
async function initRedis() {
  await pubClient.connect();
  await subClient.connect();
  console.log('[Redis] Pub/Sub clients connected');
}

initRedis().catch((err) => {
  console.error(`[Redis] Failed to initialize: ${err.message}`);
  process.exit(1);
});

// Post-migration Ably 2.0 chat server implementation
// Dependencies: ably@2.0.4, @ably/express@1.0.2
const Ably = require('ably');
const express = require('express');
const { verifyJWT } = require('./auth.utils'); // Reused from pre-migration
const { persistMessage } = require('./db.utils'); // Reused from pre-migration

const app = express();
app.use(express.json());

// Initialize Ably REST client for server-side operations
const ably = new Ably.Rest({
  key: process.env.ABLY_API_KEY,
  environment: 'production',
  // Ably 2.0 edge routing: automatically routes to closest PoP for global users
  fallbackHosts: Ably.Rest.getFallbackHosts('production')
});

// Ably Realtime client for subscribing to channels (optional, can use REST for most ops)
const ablyRealtime = new Ably.Realtime({
  key: process.env.ABLY_API_KEY,
  clientId: 'server-chat-service', // Static client ID for server
  autoConnect: true,
  disconnectOnUnload: false
});

// Error handling for Ably connections
ablyRealtime.connection.on('error', (err) => {
  console.error(`[Ably Realtime] Connection error: ${err.message}`);
  // Ably SDK handles automatic reconnection, no need to crash
});

ablyRealtime.connection.on('connected', () => {
  console.log(`[Ably Realtime] Connected to Ably edge (connection ID: ${ablyRealtime.connection.id})`);
});

// Express endpoint to handle chat message sends (stateless, no Socket.io state)
app.post('/api/chat/send', async (req, res) => {
  try {
    // Authenticate user
    const token = req.headers.authorization?.split(' ')[1];
    if (!token) {
      return res.status(401).json({ error: 'Authentication token required' });
    }

    const decoded = verifyJWT(token);
    const userId = decoded.sub;
    const { roomId, content } = req.body;

    // Validate payload
    if (!roomId || !content) {
      return res.status(400).json({ error: 'roomId and content are required' });
    }
    if (typeof content !== 'string' || content.length > 1000) {
      return res.status(400).json({ error: 'Content must be a string under 1000 characters' });
    }

    // Persist message to database
    const persistedMessage = await persistMessage({
      senderId: userId,
      roomId,
      content,
      timestamp: Date.now()
    });

    // Publish message to Ably channel (roomId maps to Ably channel)
    const channel = ably.channels.get(`chat-room-${roomId}`);
    await channel.publish('chat:receive', persistedMessage, {
      // Ably 2.0 message extras: ensure delivery, set TTL
      ttl: 3600000, // 1 hour TTL for undelivered messages
      clientId: userId // Set client ID for sender attribution
    });

    res.status(200).json({ status: 'success', messageId: persistedMessage.id });
  } catch (err) {
    console.error(`[Ably] Failed to send message: ${err.message}`);
    res.status(500).json({ error: 'Failed to send message' });
  }
});

// Subscribe to all chat channels for audit logging (optional)
const auditChannel = ablyRealtime.channels.get('chat-audit');
auditChannel.subscribe('chat:receive', (message) => {
  console.log(`[Ably Audit] Message sent to ${message.channel}: ${message.data.id}`);
});

// Handle Ably channel presence (track online users)
ablyRealtime.channels.get('chat-presence').presence.subscribe('enter', (member) => {
  console.log(`[Ably Presence] User ${member.clientId} came online`);
});

ablyRealtime.channels.get('chat-presence').presence.subscribe('leave', (member) => {
  console.log(`[Ably Presence] User ${member.clientId} went offline`);
});

// Health check endpoint
app.get('/health', (req, res) => {
  const ablyStatus = ablyRealtime.connection.state === 'connected' ? 'healthy' : 'degraded';
  res.status(200).json({
    status: 'ok',
    ablyConnectionState: ablyRealtime.connection.state,
    ablyStatus
  });
});

const PORT = process.env.PORT || 3001;
app.listen(PORT, () => {
  console.log(`[Ably Server] Listening on port ${PORT}`);
});

// Client-side React chat component: Migrated from Socket.io 4.7 to Ably 2.0
// Dependencies: ably@2.0.4, react@18.2.0
import React, { useState, useEffect, useRef } from 'react';
import Ably from 'ably';

const ChatRoom = ({ roomId, userToken }) => {
  const [messages, setMessages] = useState([]);
  const [input, setInput] = useState('');
  const [isConnected, setIsConnected] = useState(false);
  const [error, setError] = useState(null);
  const ablyRef = useRef(null);
  const channelRef = useRef(null);
  const messagesEndRef = useRef(null);

  // Initialize Ably client on mount
  useEffect(() => {
    if (!userToken || !roomId) return;

    // Cleanup previous Ably connection
    if (ablyRef.current) {
      ablyRef.current.close();
    }

    try {
      // Initialize Ably Realtime client with user JWT (server-side signed)
      const ably = new Ably.Realtime({
        token: userToken, // JWT signed by our server with Ably-compatible claims
        clientId: JSON.parse(atob(userToken.split('.')[1])).sub, // Extract user ID from JWT
        autoConnect: true,
        reconnect: true, // Ably handles reconnection automatically
        maxReconnectAttempts: 10,
        reconnectTimeout: 2000
      });

      ablyRef.current = ably;

      // Connection state handlers
      ably.connection.on('connected', () => {
        console.log('[Ably Client] Connected to edge PoP');
        setIsConnected(true);
        setError(null);
      });

      ably.connection.on('disconnected', () => {
        console.log('[Ably Client] Disconnected, reconnecting...');
        setIsConnected(false);
      });

      ably.connection.on('error', (err) => {
        console.error('[Ably Client] Connection error:', err);
        setError(`Connection error: ${err.message}`);
        setIsConnected(false);
      });

      // Subscribe to chat room channel
      const channel = ably.channels.get(`chat-room-${roomId}`);
      channelRef.current = channel;

      // Subscribe to incoming messages
      channel.subscribe('chat:receive', (message) => {
        setMessages((prev) => [...prev, {
          id: message.data.id,
          senderId: message.clientId,
          content: message.data.content,
          timestamp: message.data.timestamp
        }]);
        // Scroll to bottom on new message
        setTimeout(() => messagesEndRef.current?.scrollIntoView({ behavior: 'smooth' }), 100);
      });

      // Subscribe to presence updates (who's online)
      channel.presence.subscribe('enter', (member) => {
        console.log(`[Ably Presence] ${member.clientId} joined the room`);
      });

      channel.presence.subscribe('leave', (member) => {
        console.log(`[Ably Presence] ${member.clientId} left the room`);
      });

      // Enter presence channel
      channel.presence.enter({ username: JSON.parse(atob(userToken.split('.')[1])).username });

    } catch (err) {
      console.error('[Ably Client] Initialization error:', err);
      setError(`Failed to initialize chat: ${err.message}`);
    }

    // Cleanup on unmount
    return () => {
      if (channelRef.current) {
        channelRef.current.presence.leave();
        channelRef.current.unsubscribe();
      }
      if (ablyRef.current) {
        ablyRef.current.close();
      }
    };
  }, [roomId, userToken]);

  // Handle sending a message
  const sendMessage = async (e) => {
    e.preventDefault();
    if (!input.trim() || !isConnected) return;

    try {
      const messageContent = input.trim();
      setInput('');

      // Optimistic UI update
      const tempMessage = {
        id: `temp-${Date.now()}`,
        senderId: JSON.parse(atob(userToken.split('.')[1])).sub,
        content: messageContent,
        timestamp: Date.now(),
        pending: true
      };
      setMessages((prev) => [...prev, tempMessage]);

      // Send message via our server (which publishes to Ably)
      const response = await fetch('/api/chat/send', {
        method: 'POST',
        headers: {
          'Content-Type': 'application/json',
          'Authorization': `Bearer ${userToken}`
        },
        body: JSON.stringify({ roomId, content: messageContent })
      });

      if (!response.ok) {
        throw new Error('Failed to send message');
      }

      // Remove optimistic pending state (message will arrive via Ably subscription)
      setMessages((prev) => prev.filter((msg) => msg.id !== tempMessage.id));
    } catch (err) {
      console.error('[Chat Client] Send error:', err);
      setError(`Failed to send message: ${err.message}`);
      // Revert optimistic update
      setMessages((prev) => prev.filter((msg) => !msg.pending));
    }
  };

  return (

// Feature flag check for Ably migration (Node.js + LaunchDarkly)
const ldClient = require('launchdarkly-node-server-sdk');

async function shouldUseAbly(userId) {
  try {
    const flag = await ldClient.variation('realtime-use-ably', { key: userId }, false);
    return flag;
  } catch (err) {
    console.error('Feature flag check failed, defaulting to Socket.io');
    return false; // Rollback to old stack if flag check fails
  }
}

// Ably 2.0 auto-reconnect config (client-side)
const ably = new Ably.Realtime({
  token: userToken,
  reconnect: true,
  maxReconnectAttempts: 10,
  reconnectTimeout: 2000, // Initial reconnect delay
  maxReconnectTimeout: 30000, // Max reconnect delay
  // SDK adds jitter automatically to avoid thundering herd
});

// Ably managed channel (no Redis adapter needed)
const channel = ably.channels.get('chat-room-123');
// Ably handles multi-region routing, persistence, and delivery automatically