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