/** * In-process SSE hub. Clients subscribe via GET /events; handlers call emit() * after a successful mutation so other tabs/users can refetch. */ /** @typedef {{ res: import('express').Response, companyId: number, userId: number, clientId: string|null, keepAlive: NodeJS.Timeout }} SseClient */ /** @type {Set} */ const clients = new Set(); const KEEP_ALIVE_MS = 20_000; /** * Register an open SSE response. Caller must write the initial headers. * @param {import('express').Response} res * @param {{ companyId: number, userId: number, clientId?: string|null }} meta * @param {import('express').Request} [req] * @returns {() => void} unsubscribe */ export function subscribe(res, { companyId, userId, clientId }, req) { const resolvedClientId = clientId || req?.headers?.['x-client-id'] || req?.query?.cid || null; /** @type {SseClient} */ const client = { res, companyId, userId, clientId: resolvedClientId ? String(resolvedClientId) : null, keepAlive: setInterval(() => { if (res.writableEnded) return; res.write(': heartbeat\n\n'); }, KEEP_ALIVE_MS), }; clients.add(client); const unsubscribe = () => { clearInterval(client.keepAlive); clients.delete(client); }; res.on('close', unsubscribe); req?.on('close', unsubscribe); return unsubscribe; } /** * End every open SSE stream so `server.close()` can finish (e.g. node --watch). * EventSource clients reconnect to the new process. */ export function closeAll() { for (const client of [...clients]) { clearInterval(client.keepAlive); clients.delete(client); if (!client.res.writableEnded) { client.res.end(); } } } /** * Checks whether an active SSE client connection exists matching the given criteria. * @param {{ companyId?: number, userId?: number, clientId?: string }} [options] * @returns {boolean} */ export function hasActiveClient({ companyId, userId, clientId } = {}) { for (const client of clients) { if (companyId != null && client.companyId !== companyId) continue; if (userId != null && client.userId !== userId) continue; if (clientId != null && client.clientId !== clientId) continue; if (!client.res.writableEnded) return true; } return false; } /** * Broadcast or target a named signal to subscribers. * If `targetClientId` or `targetUserId` is specified, sends only to matching clients of that user. * * @param {string} signal - SSE event name (e.g. `category`, `draft`, `modelconversation`) * @param {{ companyId: number, data?: string|object, targetClientId?: string, targetUserId?: number }} options */ export function emit(signal, { companyId, data, targetClientId, targetUserId }) { if (companyId == null) return; const payload = data == null ? '{}' : (typeof data === 'string' ? data : JSON.stringify(data)); const frame = `event: ${signal}\ndata: ${payload}\n\n`; if (targetClientId || targetUserId != null) { let sent = false; if (targetClientId) { for (const client of clients) { if (client.companyId !== companyId) continue; if (targetUserId != null && client.userId !== targetUserId) continue; if (client.clientId === targetClientId && !client.res.writableEnded) { client.res.write(frame); sent = true; } } } if (!sent && targetUserId != null) { for (const client of clients) { if (client.companyId !== companyId) continue; if (client.userId === targetUserId && !client.res.writableEnded) { client.res.write(frame); sent = true; } } } if (sent) return; // Targeted to a specific user but no active client was found: do not blast to other users if (targetUserId != null) return; } for (const client of clients) { if (client.companyId !== companyId) continue; if (client.res.writableEnded) continue; client.res.write(frame); } }