Scaling WebSockets with Redis Pub/Sub

npm install ws ioredis

import { WebSocketServer, WebSocket } from "ws";
import Redis from "ioredis";
 
// In Redis, subscriber mode blocks normal commands.
// You MUST create two separate Redis client instances: one for publishing, one for subscribing.
const publisher = new Redis(process.env.REDIS_URL || "redis://localhost:6379");
const subscriber = new Redis(process.env.REDIS_URL || "redis://localhost:6379");
 
const serverId = process.env.SERVER_ID || `srv_${Math.random().toString(36).slice(2, 8)}`;
const PORT = process.env.PORT || 8080;
 
const wss = new WebSocketServer({ port: PORT });
console.log(`⚡ WebSocket instance ${serverId} running on port ${PORT}`);
 
// Map: ws -> Set<roomName>
const clientRooms = new Map();
 
// 1. Listen for cross-server messages forwarded by Redis
subscriber.on("message", (channel, messageString) => {
  const roomName = channel.replace("room:", "");
  const { originServerId, payload } = JSON.parse(messageString);
 
  // If this server published the message, we already delivered it locally
  if (originServerId === serverId) return;
 
  // Broadcast to all sockets connected to this local instance
  broadcastToLocalRoom(roomName, payload);
});
 
// Helper: Broadcast to local sockets in a specific room
function broadcastToLocalRoom(roomName, payload) {
  const data = JSON.stringify(payload);
  wss.clients.forEach((client) => {
    const userRooms = clientRooms.get(client);
    if (userRooms?.has(roomName) && client.readyState === WebSocket.OPEN) {
      client.send(data);
    }
  });
}
 
// Helper: Publish message to Redis
function publishToRoom(roomName, payload) {
  // 1. Send to Redis message bus for other server instances
  publisher.publish(
    `room:${roomName}`,
    JSON.stringify({ originServerId: serverId, payload })
  );
 
  // 2. Deliver to local clients immediately
  broadcastToLocalRoom(roomName, payload);
}
 
// 2. Handle Client WebSocket Connections
wss.on("connection", (ws) => {
  ws.userId = `usr_${Math.random().toString(36).substring(7)}`;
  clientRooms.set(ws, new Set());
 
  ws.on("message", async (raw) => {
    try {
      const { type, room, text } = JSON.parse(raw);
 
      if (type === "join") {
        const rooms = clientRooms.get(ws);
        rooms.add(room);
 
        // If this is the FIRST local client entering this room on this server,
        // subscribe this server instance to the Redis channel
        const count = countLocalClientsInRoom(room);
        if (count === 1) {
          await subscriber.subscribe(`room:${room}`);
          console.log(`📡 [${serverId}] Subscribed to Redis channel: room:${room}`);
        }
 
        publishToRoom(room, {
          type: "notification",
          text: `User ${ws.userId} joined #${room}`
        });
      }
 
      if (type === "chat") {
        publishToRoom(room, {
          type: "chat",
          from: ws.userId,
          room,
          text,
          timestamp: Date.now()
        });
      }
    } catch (err) {
      console.error("Message processing error:", err);
    }
  });
 
  ws.on("close", async () => {
    const rooms = clientRooms.get(ws);
    if (rooms) {
      for (const room of rooms) {
        rooms.delete(room);
        publishToRoom(room, {
          type: "notification",
          text: `User ${ws.userId} left #${room}`
        });
 
        // If no clients on this server remain in the room, unsubscribe from Redis
        if (countLocalClientsInRoom(room) === 0) {
          await subscriber.unsubscribe(`room:${room}`);
          console.log(`🔕 [${serverId}] Unsubscribed from Redis channel: room:${room}`);
        }
      }
    }
    clientRooms.delete(ws);
  });
});
 
function countLocalClientsInRoom(roomName) {
  let count = 0;
  wss.clients.forEach((c) => {
    if (clientRooms.get(c)?.has(roomName)) count++;
  });
  return count;
}

upstream websocket_cluster {
    # ip_hash ensures client IP stays sticky to the same server node
    ip_hash;
    server 10.0.0.1:8080;
    server 10.0.0.2:8080;
    server 10.0.0.3:8080;
}
 
server {
    listen 80;
    server_name ws.example.com;
 
    location /ws {
        proxy_pass http://websocket_cluster;
        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;
 
        # Keep persistent connections open without timing out
        proxy_read_timeout 86400s;
        proxy_send_timeout 86400s;
    }
}

import Redis from "ioredis";
const redis = new Redis(process.env.REDIS_URL);
 
// Mark user as active and record server location
export async function setUserOnline(userId, serverId) {
  await redis.hset("online_users", userId, serverId);
}
 
// Remove user on disconnect
export async function setUserOffline(userId) {
  await redis.hdel("online_users", userId);
}
 
// Get global list of all online users
export async function getOnlineUsers() {
  return await redis.hgetall("online_users");
}
 
// Check if specific user is currently online
export async function isUserOnline(userId) {
  return (await redis.hexists("online_users", userId)) === 1;
}

import { Server } from "socket.io";
import { createAdapter } from "@socket.io/redis-adapter";
import { createClient } from "redis";
 
const io = new Server(3000);
 
const pubClient = createClient({ url: process.env.REDIS_URL });
const subClient = pubClient.duplicate();
 
await Promise.all([pubClient.connect(), subClient.connect()]);
 
// Plug in the Redis adapter
io.adapter(createAdapter(pubClient, subClient));
 
io.on("connection", (socket) => {
  socket.on("join_room", (room) => socket.join(room));
 
  socket.on("chat_message", ({ room, text }) => {
    // Automatically routes across all cluster servers through Redis!
    io.to(room).emit("chat_message", { text, from: socket.id });
  });
});