WebSocket 완전 정복 12편 — 수평 확장: Pub/Sub 백플레인과 무중단 배포

 Client A ─┐                         ┌─ Client B
           ▼                         ▼
       [WS Server 1] ◀──▶ [Redis Pub/Sub] ◀──▶ [WS Server 2]

import { WebSocketServer, WebSocket } from "ws";
import { createClient } from "redis";
 
const pub = createClient({ url: process.env.REDIS_URL });
const sub = pub.duplicate();
await Promise.all([pub.connect(), sub.connect()]);
 
// 이 서버 로컬의 채널 → 소켓 매핑
const rooms = new Map<string, Set<WebSocket>>();
 
function join(room: string, ws: WebSocket) {
  if (!rooms.has(room)) {
    rooms.set(room, new Set());
    // 이 서버에 해당 방의 첫 구독자가 생길 때만 Redis 구독
    sub.subscribe(`room:${room}`, (message) => {
      for (const client of rooms.get(room) ?? []) {
        if (client.readyState === WebSocket.OPEN) client.send(message);
      }
    });
  }
  rooms.get(room)!.add(ws);
}
 
function leave(room: string, ws: WebSocket) {
  const set = rooms.get(room);
  if (!set) return;
  set.delete(ws);
  if (set.size === 0) {
    rooms.delete(room);
    sub.unsubscribe(`room:${room}`);
  }
}
 
const wss = new WebSocketServer({ port: Number(process.env.PORT ?? 8080) });
 
wss.on("connection", (ws) => {
  const joined = new Set<string>();
 
  ws.on("message", async (raw) => {
    let msg: any;
    try { msg = JSON.parse(raw.toString()); } catch { return ws.close(1007); }
 
    if (msg.type === "join" && typeof msg.room === "string") {
      join(msg.room, ws); joined.add(msg.room);
    } else if (msg.type === "chat" && joined.has(msg.room)) {
      await pub.publish(`room:${msg.room}`,
        JSON.stringify({ type: "chat", room: msg.room, text: String(msg.text).slice(0, 2000) }));
    }
  });
 
  ws.on("close", () => { for (const r of joined) leave(r, ws); });
});