Split View: WebSocket & 실시간 통신 완전 가이드 2025: SSE, Socket.IO, 채팅/알림/협업 구현

// Short Polling 구현
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);
    }
  };

  // 즉시 한 번 실행 후 인터벌 설정
  poll();
  return setInterval(poll, interval);
}

// 정리
const pollId = startPolling(3000);
// clearInterval(pollId);

// Long Polling 클라이언트
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초 타임아웃
    });

    if (response.status === 200) {
      const data = await response.json();
      handleNewData(data);
      // 즉시 다음 Long Poll 시작
      longPoll(data.lastEventId);
    } else if (response.status === 204) {
      // 데이터 없음 - 타임아웃 후 재시도
      longPoll(lastEventId);
    }
  } catch (error) {
    // 에러 시 점진적 재시도
    console.error('Long poll error:', error);
    setTimeout(() => longPoll(lastEventId), 3000);
  }
}

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

  const checkForData = () => {
    return new Promise((resolve) => {
      const timer = setTimeout(() => {
        resolve(null); // 타임아웃
      }, timeout);

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

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

// 클라이언트 요청
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

// 서버 응답
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 });

// 연결된 클라이언트 관리
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 하트비트
  ws.isAlive = true;
  ws.on('pong', () => {
    ws.isAlive = true;
  });

  // 메시지 수신
  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' }));
    }
  });

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

  // 에러 처리
  ws.on('error', (error) => {
    console.error(`WebSocket error for ${clientId}:`, error);
  });

  // 환영 메시지
  ws.send(JSON.stringify({
    type: 'welcome',
    clientId,
    timestamp: Date.now(),
  }));
});

// 하트비트 인터벌
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 message = JSON.stringify({
    type: 'broadcast',
    from: senderId,
    payload,
    timestamp: Date.now(),
  });

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

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;
      // 큐에 쌓인 메시지 전송
      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;
    }

    // 지수 백오프 + 지터
    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; // 재연결 방지
    this.ws.close(1000, 'Client closing');
  }
}

// 사용 예시
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 서버
import express from 'express';

const app = express();

// SSE 클라이언트 관리
const sseClients = new Map();
let eventCounter = 0;

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

  // SSE 헤더 설정
  res.writeHead(200, {
    'Content-Type': 'text/event-stream',
    'Cache-Control': 'no-cache',
    'Connection': 'keep-alive',
    'X-Accel-Buffering': 'no', // Nginx 버퍼링 비활성화
  });

  // Last-Event-ID 처리 (재연결 시 복구)
  const lastEventId = req.headers['last-event-id'];
  if (lastEventId) {
    // 놓친 이벤트 재전송
    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`);
    });
  }

  // 클라이언트 등록
  sseClients.set(clientId, res);

  // 재연결 간격 설정 (밀리초)
  res.write('retry: 5000\n\n');

  // 초기 연결 확인
  res.write(`id: ${++eventCounter}\n`);
  res.write('event: connected\n');
  res.write(`data: ${JSON.stringify({ clientId })}\n\n`);

  // Keep-alive (15초마다 코멘트 전송)
  const keepAlive = setInterval(() => {
    res.write(':keep-alive\n\n');
  }, 15000);

  // 연결 종료 처리
  req.on('close', () => {
    clearInterval(keepAlive);
    sseClients.delete(clientId);
    console.log(`SSE client disconnected: ${clientId}`);
  });
});

// 모든 SSE 클라이언트에 이벤트 브로드캐스트
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`);
  });
}

// 알림 발송 예시
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 활용
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가 자동으로 재연결을 시도함
    };

    // 커스텀 이벤트 등록
    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();
    }
  }
}

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

// 서버 설정
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,
  },
  // 트랜스포트 설정
  transports: ['websocket', 'polling'],
  // Ping/Pong 설정
  pingTimeout: 20000,
  pingInterval: 25000,
  // 최대 페이로드
  maxHttpBufferSize: 1e6, // 1MB
});

// 네임스페이스 - 논리적 분리
const chatNamespace = io.of('/chat');
const adminNamespace = io.of('/admin');

// 채팅 네임스페이스
chatNamespace.on('connection', (socket) => {
  console.log('User connected to chat:', socket.id);

  // 룸 참여
  socket.on('joinRoom', async (roomName) => {
    await socket.join(roomName);
    // 해당 룸의 다른 사용자에게 알림
    socket.to(roomName).emit('userJoined', {
      userId: socket.id,
      room: roomName,
    });

    // 룸 참여자 수 전송
    const sockets = await chatNamespace.in(roomName).fetchSockets();
    chatNamespace.to(roomName).emit('roomInfo', {
      room: roomName,
      members: sockets.length,
    });
  });

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

  // 룸 내 메시지 전송
  socket.on('message', (data) => {
    chatNamespace.to(data.room).emit('message', {
      from: socket.id,
      text: data.text,
      room: data.room,
      timestamp: Date.now(),
    });
  });
});

// 관리자 네임스페이스 - 인증 필요
adminNamespace.use((socket, next) => {
  const token = socket.handshake.auth.token;
  if (verifyAdminToken(token)) {
    next();
  } else {
    next(new Error('Unauthorized'));
  }
});

// 글로벌 미들웨어
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'));
  }
});

// 네임스페이스 미들웨어
chatNamespace.use((socket, next) => {
  // 추가 권한 확인
  if (socket.data.user.role === 'banned') {
    return next(new Error('User is banned'));
  }
  next();
});

// 연결 후 이벤트별 미들웨어
io.on('connection', (socket) => {
  // 모든 이벤트에 대한 로깅 미들웨어
  socket.use(([event, ...args], next) => {
    console.log(`Event: ${event}, User: ${socket.data.user.name}`);
    next();
  });

  // Rate limiting 미들웨어
  const rateLimiter = createRateLimiter(100, 60000); // 분당 100개
  socket.use(([event, ...args], next) => {
    if (rateLimiter.isAllowed(socket.id)) {
      next();
    } else {
      next(new Error('Rate limit exceeded'));
    }
  });
});

// 서버
io.on('connection', (socket) => {
  socket.on('createOrder', async (orderData, callback) => {
    try {
      const order = await processOrder(orderData);
      // 성공 응답
      callback({
        status: 'ok',
        orderId: order.id,
      });
    } catch (error) {
      // 에러 응답
      callback({
        status: 'error',
        message: error.message,
      });
    }
  });
});

// 클라이언트
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
socket.timeout(5000).emit('createOrder', orderData, (err, response) => {
  if (err) {
    // 서버가 5초 안에 응답하지 않음
    console.error('Timeout');
  } else {
    console.log('Response:', response);
  }
});

// 서버: 메시지 순서 관리
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',
    };

    // 영구 저장
    await this.persistMessage(message);

    return message;
  }

  async getMessagesSince(sequence, limit = 50) {
    // 특정 시퀀스 이후의 메시지 조회
    return await db.messages
      .find({
        roomId: this.roomId,
        sequence: { $gt: sequence },
      })
      .sort({ sequence: 1 })
      .limit(limit)
      .toArray();
  }
}

// Presence 시스템
class PresenceManager {
  constructor(io, redis) {
    this.io = io;
    this.redis = redis;
    this.TTL = 60; // 60초
  }

  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);

    // 친구에게 상태 변경 알림
    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' };
  }
}

// 클라이언트
let typingTimeout;

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

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

// 서버
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,
  });
});

// 서버: 읽음 확인 처리
socket.on('markRead', async (data) => {
  const { roomId, messageId, userId } = data;

  // DB에 읽음 상태 저장
  await db.readReceipts.upsert({
    roomId,
    userId,
    lastReadMessageId: messageId,
    readAt: new Date(),
  });

  // 룸 참여자에게 알림
  socket.to(roomId).emit('readReceipt', {
    userId,
    messageId,
    readAt: Date.now(),
  });
});

// 클라이언트: 읽음 확인 표시
socket.on('readReceipt', (receipt) => {
  updateMessageStatus(receipt.messageId, receipt.userId, 'read');
});

// 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 }
);

// 알림 서비스
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,
    };

    // 영구 저장
    await db.notifications.insertOne({
      userId,
      ...enriched,
    });

    // 실시간 전송
    this.io.to(`user:${userId}`).emit('notification', enriched);

    // 오프라인이면 푸시 알림 큐에 추가
    const isOnline = await this.isUserOnline(userId);
    if (!isOnline) {
      await this.queuePushNotification(userId, enriched);
    }

    return enriched;
  }

  async sendBatch(userIds, notification) {
    // 배치 처리로 대량 알림 효율적 전송
    const enriched = {
      id: generateId(),
      ...notification,
      createdAt: Date.now(),
      read: false,
    };

    // 벌크 인서트
    const docs = userIds.map((userId) => ({
      userId,
      ...enriched,
    }));
    await db.notifications.insertMany(docs);

    // 온라인 사용자에게 실시간 전송
    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() } }
    );

    // 읽지 않은 알림 수 업데이트
    const unreadCount = await db.notifications.countDocuments({
      userId,
      read: false,
    });

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

// 고빈도 이벤트 배치 처리
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);

    // 배치 크기 초과 시 즉시 전송
    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, []);

    // 중복 제거 및 그룹화
    const grouped = this.groupNotifications(notifications);
    io.to(`user:${userId}`).emit('notifications:batch', grouped);
  }

  groupNotifications(notifications) {
    // 같은 유형의 알림 그룹화
    const groups = {};
    notifications.forEach((n) => {
      const key = n.type;
      if (!groups[key]) groups[key] = [];
      groups[key].push(n);
    });
    return groups;
  }
}

// OT 기본 개념
// 두 사용자가 동시에 편집할 때:
// User A: insert('X', position=0) - 위치 0에 'X' 삽입
// User B: insert('Y', position=2) - 위치 2에 'Y' 삽입

// A의 연산을 B의 관점에서 변환:
// B가 위치 2에 삽입했으므로 A의 연산은 변경 불필요 (위치 0은 영향 없음)

// B의 연산을 A의 관점에서 변환:
// A가 위치 0에 삽입했으므로 B의 위치는 2+1=3으로 조정

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

// Y.js를 활용한 CRDT 기반 협업 편집
import * as Y from 'yjs';
import { WebsocketProvider } from 'y-websocket';

// 공유 문서 생성
const ydoc = new Y.Doc();

// WebSocket 프로바이더 연결
const provider = new WebsocketProvider(
  'wss://collaboration.example.com',
  'document-room-123',
  ydoc
);

// 공유 텍스트
const ytext = ydoc.getText('content');

// 텍스트 변경 감지
ytext.observe((event) => {
  console.log('Text changed:', event.delta);
});

// 텍스트 편집
ytext.insert(0, 'Hello, ');
ytext.insert(7, 'World!');

// 공유 맵 (키-값)
const ymap = ydoc.getMap('metadata');
ymap.set('title', 'My Document');
ymap.set('lastEdited', Date.now());

// 공유 배열
const yarray = ydoc.getArray('items');
yarray.push(['item1', 'item2']);

// 연결 상태 모니터링
provider.on('status', (event) => {
  console.log('Connection status:', event.status);
});

// Awareness (커서 위치, 사용자 정보)
const awareness = provider.awareness;
awareness.setLocalState({
  user: { name: 'Alice', color: '#ff0000' },
  cursor: { position: 42 },
});

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

// y-websocket 서버
import { WebSocketServer } from 'ws';
import { setupWSConnection } from 'y-websocket/bin/utils';

const wss = new WebSocketServer({ port: 1234 });

wss.on('connection', (ws, req) => {
  setupWSConnection(ws, req, {
    docname: req.url.slice(1), // URL 경로를 문서 이름으로 사용
    gc: true, // 가비지 컬렉션
  });
});

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

const io = new Server(httpServer);

// Redis 클라이언트 생성
const pubClient = createClient({ url: 'redis://redis-host:6379' });
const subClient = pubClient.duplicate();

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

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

// 이제 어떤 서버 인스턴스에서든 broadcast 가능
io.emit('globalEvent', { message: 'All servers receive this' });

// 특정 룸으로 전송 (모든 서버의 해당 룸 클라이언트에게)
io.to('room-123').emit('roomEvent', { data: 'value' });

# Nginx 설정
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;

        # 타임아웃 설정
        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();

    // 브로드캐스트 채널 구독
    const sub = this.nc.subscribe('ws.broadcast.>');
    (async () => {
      for await (const msg of sub) {
        const data = JSON.parse(msg.data.toString());
        // 자신이 보낸 메시지는 무시
        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,
    }));
  }
}

// 토큰 기반 인증
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 검증
    const payload = await verifyJWT(token);

    // 토큰 만료 확인
    if (payload.exp * 1000 < Date.now()) {
      return next(new Error('Token expired'));
    }

    // 사용자 정보 첨부
    socket.data.user = {
      id: payload.sub,
      roles: payload.roles,
    };

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

// 주기적 토큰 갱신
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();
  }
});

// 토큰 버킷 알고리즘
class TokenBucket {
  constructor(capacity, refillRate) {
    this.capacity = capacity;
    this.tokens = capacity;
    this.refillRate = refillRate; // 초당 토큰 추가량
    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;
  }
}

// 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 용량, 초당 10 리필
  }
  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,
  },
});

// 추가 Origin 검증 미들웨어
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 확장
import { WebSocketServer } from 'ws';

const wss = new WebSocketServer({
  port: 8080,
  perMessageDeflate: {
    zlibDeflateOptions: {
      chunkSize: 1024,
      memLevel: 7,
      level: 3, // 압축 레벨 (1-9)
    },
    zlibInflateOptions: {
      chunkSize: 10 * 1024,
    },
    // 임계값 이상만 압축
    threshold: 1024, // 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;
  }
}

// Protobuf 사용
import protobuf from 'protobufjs';

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

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

// 디코딩
function decodeMessage(buffer) {
  return WebSocketFrame.decode(new Uint8Array(buffer));
}

// WebSocket에서 바이너리 전송
ws.on('message', (data, isBinary) => {
  if (isBinary) {
    const frame = decodeMessage(data);
    handleFrame(frame);
  }
});

// 연결 수 제한
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);

    // 최대 연결 수 초과 시 가장 오래된 연결 종료
    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 서버 헬스 체크
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(),
  };

  // 연결 수 임계값 확인
  if (health.connections > 10000) {
    health.status = 'degraded';
    res.status(503);
  }

  res.json(health);
});

// 쿠버네티스 readiness/liveness 프로브
app.get('/ready', (req, res) => {
  if (isReady) {
    res.status(200).json({ ready: true });
  } else {
    res.status(503).json({ ready: false });
  }
});