/* ── SSE registry — shared between controllers ─────────────────────────── */ const clients = new Map(); // userId -> Set const guestClients = new Map(); // sessionId -> Set const db = require('./db/db'); function addClient(userId, res) { if (!clients.has(userId)) clients.set(userId, new Set()); clients.get(userId).add(res); } function removeClient(userId, res) { const set = clients.get(userId); if (!set) return; set.delete(res); if (set.size === 0) clients.delete(userId); } function emit(userId, data) { const conns = clients.get(userId); if (!conns?.size) return; const payload = `data: ${JSON.stringify(data)}\n\n`; for (const res of conns) { try { res.write(payload); } catch {} } } // Heartbeat: detect and remove dead connections every 30s setInterval(() => { for (const [userId, conns] of clients) { for (const res of conns) { try { if (res.writableEnded || res.destroyed) { conns.delete(res); continue; } res.write(': heartbeat\n\n'); } catch { conns.delete(res); } } if (conns.size === 0) clients.delete(userId); } for (const [sessionId, conns] of guestClients) { for (const res of conns) { try { if (res.writableEnded || res.destroyed) { conns.delete(res); continue; } res.write(': heartbeat\n\n'); } catch { conns.delete(res); } } if (conns.size === 0) guestClients.delete(sessionId); } }, 30_000).unref(); /* Broadcast to all members of a class */ function emitToClass(classId, data) { const members = db.prepare('SELECT user_id FROM class_members WHERE class_id=?').all(classId); for (const { user_id } of members) emit(user_id, data); } /* ── Guest SSE (session-scoped, no userId) ── */ function addGuestClient(sessionId, res) { if (!guestClients.has(sessionId)) guestClients.set(sessionId, new Set()); guestClients.get(sessionId).add(res); } function removeGuestClient(sessionId, res) { const set = guestClients.get(sessionId); if (!set) return; set.delete(res); if (set.size === 0) guestClients.delete(sessionId); } function emitToGuests(sessionId, data) { const set = guestClients.get(sessionId); if (!set?.size) return; const payload = `data: ${JSON.stringify(data)}\n\n`; for (const res of set) { try { res.write(payload); } catch {} } } /* Returns array of user IDs currently connected via SSE */ function getOnlineUserIds() { return [...clients.keys()]; } /* Сводка SSE-соединений для мониторинга: онлайн-пользователи, гости и суммарное число открытых стримов. */ function stats() { let conns = 0; for (const set of clients.values()) conns += set.size; let guestConns = 0; for (const set of guestClients.values()) guestConns += set.size; return { users: clients.size, guests: guestClients.size, connections: conns + guestConns }; } module.exports = { addClient, removeClient, emit, emitToClass, getOnlineUserIds, stats, addGuestClient, removeGuestClient, emitToGuests, };