import { Server } from 'socket.io'; // Abfragen pro Verbindung: kurzer Schwall erlaubt, danach 2 pro Sekunde const SOCKET_BURST = 20; const SOCKET_REFILL = 2; const SOCKET_REFILL_MS = 1000; /** * 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(async () => { io.to('stream-overview').emit('stream:overview', service.streamOverview()); // Zugänge mit Match-Schlüssel bekommen nur ihr Match for (const s of await io.in('stream-match').fetchSockets()) { s.emit('stream:overview', service.streamOverview({ matchId: s.data.streamMatch })); } }, 200); }; service.on('changed', (matchId) => { broadcastMatch(matchId); broadcastOverview(); broadcastStream(); }); io.on('connection', (socket) => { // Jede Nachricht kommt ungeprüft vom Client: Fehler im Handler dürfen nie das Backend beenden, // und Abfragen sind pro Verbindung begrenzt (jede baut einen vollständigen Zustand). let budget = SOCKET_BURST; const refill = setInterval(() => { budget = Math.min(SOCKET_BURST, budget + SOCKET_REFILL); }, SOCKET_REFILL_MS); const on = (event, handler) => socket.on(event, (payload) => { if (budget <= 0) { socket.emit('rate_limited', { error: 'Zu viele Anfragen – bitte kurz warten' }); return; } budget -= 1; try { handler(payload && typeof payload === 'object' && !Array.isArray(payload) ? payload : {}); } catch (err) { console.error(`[socket] ${event}: ${err.message}`); } }); on('overview:subscribe', () => { socket.join('overview'); socket.emit('overview', { matches: service.listMatches() }); }); on('stream:subscribe', ({ key }) => { const access = service.streamAccess(typeof key === 'string' ? key : ''); if (!access) { socket.emit('stream:error', { error: 'Für die Stream-Übersicht wird der Stream-Schlüssel benötigt' }); return; } socket.leave('stream-overview'); socket.leave('stream-match'); // Mit einem Match-Schlüssel gibt es nur die Daten dieses einen Matches socket.data.streamMatch = access.matchId ?? null; socket.join(access.matchId ? 'stream-match' : 'stream-overview'); socket.emit('stream:overview', service.streamOverview({ matchId: access.matchId })); }); on('match:subscribe', ({ matchId, token, playerToken }) => { if (typeof matchId !== 'string' || matchId.length > 64) 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 = { ...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', () => { clearInterval(refill); try { if (socket.data?.matchId) leave(socket.data.matchId); } catch (err) { console.error(`[socket] disconnect: ${err.message}`); } }); }); // 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; }