Cs2-Mapveto/server/src/realtime.js

import { Server } from 'socket.io';
import { safeEqual } from './security.js';

/**
 * Echtzeit-Kanal. Clients senden nie Aktionen über den Socket – alle Änderungen laufen über die
 * REST-API. Der Socket liefert nach jeder Änderung den vollständigen, rollenabhängigen Match-Zustand.
 * Dadurch ist ein Client nach Neuladen oder Verbindungsabbruch sofort wieder synchron.
 */
export function attachRealtime(httpServer, { service, config }) {
  const io = new Server(httpServer, {
    path: '/socket.io',
    cors: { origin: config.publicUrl, credentials: false },
    connectionStateRecovery: { maxDisconnectionDuration: 2 * 60_000 },
  });

  // matchId -> Map(socketId -> presenceKey)
  const presence = new Map();
  service.presence = (matchId) => new Set((presence.get(matchId) ?? new Map()).values());

  const sendState = (socket) => {
    const { matchId, token, playerToken } = socket.data;
    try {
      const viewer = service.resolveViewer(matchId, [token, playerToken]);
      // Rolle kann sich geändert haben (z. B. beigetreten, entfernt) – Anwesenheit vor dem Snapshot nachziehen
      // und die anderen informieren, damit der Online-Status sofort stimmt.
      const map = presence.get(matchId);
      const key = viewer.playerId ?? `socket:${socket.id}`;
      const changed = map && map.get(socket.id) !== key;
      map?.set(socket.id, key);
      socket.emit('match:state', service.snapshot(matchId, viewer));
      if (changed) broadcastMatch(matchId);
    } catch (err) {
      socket.emit('match:error', { error: err.message, code: err.code ?? 'error' });
    }
  };

  let overviewTimer = null;
  const broadcastOverview = () => {
    clearTimeout(overviewTimer);
    overviewTimer = setTimeout(() => io.to('overview').emit('overview', { matches: service.listMatches() }), 150);
  };

  const pending = new Map();
  const broadcastMatch = (matchId) => {
    // Mehrere Änderungen innerhalb kurzer Zeit werden zu einem Update zusammengefasst.
    if (pending.has(matchId)) return;
    pending.set(matchId, setTimeout(async () => {
      pending.delete(matchId);
      const sockets = await io.in(`match:${matchId}`).fetchSockets();
      sockets.forEach(sendState);
    }, 50));
  };

  let streamTimer = null;
  const broadcastStream = () => {
    clearTimeout(streamTimer);
    streamTimer = setTimeout(() => io.to('stream-overview').emit('stream:overview', service.streamOverview()), 200);
  };

  service.on('changed', (matchId) => {
    broadcastMatch(matchId);
    broadcastOverview();
    broadcastStream();
  });

  io.on('connection', (socket) => {
    socket.on('overview:subscribe', () => {
      socket.join('overview');
      socket.emit('overview', { matches: service.listMatches() });
    });

    socket.on('stream:subscribe', ({ key } = {}) => {
      if (config.streamKey && !safeEqual(String(key ?? ''), config.streamKey)) {
        socket.emit('stream:error', { error: 'Für die Stream-Übersicht wird der Stream-Schlüssel benötigt' });
        return;
      }
      socket.join('stream-overview');
      socket.emit('stream:overview', service.streamOverview());
    });

    socket.on('match:subscribe', ({ matchId, token, playerToken } = {}) => {
      if (typeof matchId !== 'string') return;
      for (const room of socket.rooms) if (room.startsWith('match:')) leave(room.slice(6));
      const str = (v) => (typeof v === 'string' ? v : null);
      socket.data = { matchId, token: str(token), playerToken: str(playerToken) };
      try {
        service.getMatch(matchId);
      } catch (err) {
        socket.emit('match:error', { error: err.message, code: err.code });
        return;
      }
      socket.join(`match:${matchId}`);
      if (!presence.has(matchId)) presence.set(matchId, new Map());
      service.touch(matchId);
      sendState(socket);
      broadcastMatch(matchId); // Online-Status für alle aktualisieren
    });

    const leave = (matchId) => {
      socket.leave(`match:${matchId}`);
      const map = presence.get(matchId);
      if (map?.delete(socket.id)) {
        if (map.size === 0) presence.delete(matchId);
        service.touch(matchId);
        broadcastMatch(matchId);
      }
    };

    socket.on('disconnect', () => {
      if (socket.data?.matchId) leave(socket.data.matchId);
    });
  });

  // Beim Herunterfahren: ausstehende Sammel-Updates verwerfen, damit nichts mehr auf die Datenbank zugreift
  io.disposeRealtime = () => {
    clearTimeout(overviewTimer);
    clearTimeout(streamTimer);
    for (const t of pending.values()) clearTimeout(t);
    pending.clear();
    service.removeAllListeners('changed');
  };

  return io;
}