/**
 * 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, 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, userId: number }} meta
 * @param {import('express').Request} [req]
 * @returns {() => void} unsubscribe
 */
export function subscribe(res, { companyId, userId }, req) {
    /** @type {SseClient} */
    const client = {
        res,
        companyId,
        userId,
        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();
        }
    }
}

/**
 * Broadcast a named signal to subscribers of the given company.
 * @param {string} signal - SSE event name (e.g. `category`)
 * @param {{ companyId: number }} options
 */
export function emit(signal, { companyId }) {
    if (companyId == null) return;

    const frame = `event: ${signal}\ndata: {}\n\n`;
    for (const client of clients) {
        if (client.companyId !== companyId) continue;
        if (client.res.writableEnded) continue;
        client.res.write(frame);
    }
}