Learn_System/backend/src/sse.js

/* ── SSE registry — shared between controllers ─────────────────────────── */
const clients      = new Map(); // userId   -> Set<res>
const guestClients = new Map(); // sessionId -> Set<res>
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,
};