dashback/src/api/sse.js

/**
 * 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|null, userId: number, clientId: string|null, keepAlive: NodeJS.Timeout }} SseClient */

/** @type {Set<SseClient>} */
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|null, userId: number, clientId?: string|null }} meta
 * @param {import('express').Request} [req]
 * @returns {() => void} unsubscribe
 */
export function subscribe(res, { companyId = null, userId, clientId }, req) {
    const resolvedClientId = clientId || req?.headers?.['x-client-id'] || req?.query?.cid || null;
    /** @type {SseClient} */
    const client = {
        res,
        companyId: companyId ?? null,
        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;
}

/**
 * Updates the active company scope for all open SSE streams of a user
 * (e.g. when the user selects or switches company on a pre-company stream)
 * and emits `user` and `company` events to those streams.
 * @param {number} userId
 * @param {number|null} companyId
 */
export function setUserCompany(userId, companyId) {
    if (userId == null) return;
    const nextCompanyId = companyId ?? null;
    const userFrame = `event: user\ndata: ${JSON.stringify({ selectedCompanyId: nextCompanyId })}\n\n`;
    const companyFrame = `event: company\ndata: ${JSON.stringify({ companyId: nextCompanyId })}\n\n`;
    for (const client of clients) {
        if (client.userId !== userId) continue;
        client.companyId = nextCompanyId;
        if (!client.res.writableEnded) {
            client.res.write(userFrame);
            client.res.write(companyFrame);
        }
    }
}

/**
 * Clears a deleted company from any open SSE streams bound to it
 * and emits `user` (`{ selectedCompanyId: null }`) and `company` (`{ companyId: null }`)
 * events to those streams.
 * @param {number} companyId
 */
export function clearDeletedCompany(companyId) {
    if (companyId == null) return;
    const userFrame = `event: user\ndata: ${JSON.stringify({ selectedCompanyId: null })}\n\n`;
    const companyFrame = `event: company\ndata: ${JSON.stringify({ companyId: null })}\n\n`;
    for (const client of clients) {
        if (client.companyId !== companyId) continue;
        client.companyId = null;
        if (!client.res.writableEnded) {
            client.res.write(userFrame);
            client.res.write(companyFrame);
        }
    }
}

/**
 * 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`, `articles`, `modelconversation`)
 * @param {{ companyId?: number|null, data?: string|object, targetClientId?: string, targetUserId?: number }} options
 */
export function emit(signal, { companyId, data, targetClientId, targetUserId }) {
    if (companyId == null && targetUserId == 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 (companyId != null && 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 (companyId != null && 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);
    }
}