xo-cash-utils/source/sse-session/sse-session.ts

import type { SSESessionOptions, SSESessionEventMap, SSEvent } from './types.ts';

import { EventEmitter } from 'eventemitter3'

import { HTTPError, ResponseBodyNullError } from '../errors.ts';
import { tryAsync } from '../misc.ts';
import { ExponentialBackoff } from '../exponential-backoff.ts';

import { SSEEventParser } from './sse-event-parser.ts';
import { AsyncPushIterator } from './async-push-iterator.ts';

/**
 * A fetch-based Server-Sent Events (SSE) client with reconnect and optional
 * browser tab visibility handling.
 *
 * Each session maintains one HTTP streaming connection at a time. Incoming
 * bytes are parsed into {@link SSEvent} objects and delivered through two
 * surfaces:
 *
 * - **Events** — `"connected"`, `"message"`, `"disconnected"`, `"error"`,
 *   and `"closed"` on the session itself (extends {@link EventEmitter}).
 * - **Messages** — {@link messages}, an async iterable for `for await...of`
 *   consumers.
 *
 * Typical usage:
 *
 * ```ts
 * const session = await SSESession.create("/events");
 *
 * session.on("message", (event) => console.log(event.data));
 *
 * for await (const event of session.messages) {
 *   handle(event);
 * }
 * ```
 *
 * ## Lifecycle
 *
 * - {@link connect} opens (or reopens) the transport. It resolves once the
 *   HTTP stream is established; reading continues in the background.
 * - {@link abort} stops the in-flight fetch without ending the session.
 *   Used internally for tab visibility. The {@link messages} iterator stays
 *   open so an existing consumer resumes when the tab becomes visible again.
 * - {@link disconnect} aborts the transport, closes {@link messages}, emits
 *   `"closed"`, and disables attached visibility handlers until the next
 *   manual {@link connect}.
 *
 * Automatic reconnect is controlled by {@link SSESessionOptions.persistent}
 * (server closed the stream) and
 * {@link SSESessionOptions.attemptReconnect} (transport error).
 *
 * ## Connection supersession
 *
 * Each {@link connect} or {@link abort} bumps an internal `connectionId`.
 * Background read loops capture their id at start and exit quietly when a
 * newer connection supersedes them, avoiding duplicate events or errors from
 * stale transports.
 */
export class SSESession extends EventEmitter<SSESessionEventMap> {
    /**
     * Creates a session and waits until the first connection is established.
     *
     * @param url - The SSE endpoint URL.
     * @param options - Configuration merged with instance defaults.
     * @returns A connected session.
     * @throws When the initial connection cannot be established.
     */
    static async create(url: string, options: Partial<SSESessionOptions> = {}): Promise<SSESession> {
        const client = new SSESession(url, options);
        await client.connect();

        return client;
    }

    /**
     * Enables SSE resume semantics by sending `Last-Event-ID` on reconnect.
     *
     * Listens for incoming `"message"` events and remembers the most recent
     * {@link SSEvent.id}. On every subsequent connect or reconnect, the session's
     * {@link onRequest} hook is wrapped so that header is attached when an id is
     * known, allowing the server to replay only events the client has not yet
     * received.
     *
     * The existing {@link onRequest} callback is preserved and runs after the
     * header is applied, so auth or other header mutations continue to work.
     *
     * Attach as early in the session lifetime as possible. When added after
     * {@link create}, the initial connection omits the header (no id yet);
     * all later reconnects include it. To instrument before the first connect,
     * call this on the session returned from {@link withBrowserVisibility}
     * before awaiting a separate {@link connect} when the tab starts hidden.
     *
     * ```ts
     * const session = await SSESession.create(url);
     * await SSESession.addLastEventIdReconnect(session);
     * // Reconnects send Last-Event-ID once an event with an id is received.
     * ```
     *
     * @param client - The session to instrument.
     * @returns The same session, for chaining.
     */
    static async addLastEventIdReconnect(client: SSESession): Promise<SSESession> {
        let lastEventId: string | undefined;

        client.on('message', (event) => {
            lastEventId = event.id;
        });

        const originalOnRequest = client.options.onRequest;

        client.options.onRequest = async (request: RequestInit): Promise<RequestInit> => {
            if (lastEventId) {
                request.headers = { ...request.headers, 'Last-Event-ID': lastEventId };
            }

            return originalOnRequest(request);
        };

        return client;
    }

    /**
     * Pauses and resumes a session based on browser tab visibility.
     *
     * Uses the Page Visibility API (`document.visibilitychange`):
     *
     * - **hidden** — {@link abort} stops the active fetch. {@link messages}
     *   stays open; `"disconnected"` fires but `"closed"` does not.
     * - **visible** — {@link connect} re-establishes the stream if needed.
     *
     * The listener is removed when {@link disconnect} emits `"closed"`, and
     * re-attached automatically on the next `"connected"` event.
     *
     * No-op in non-browser environments where `document` is undefined.
     *
     * @param client - The session to manage.
     */
    static addBrowserVisibilityHandler(client: SSESession): SSESession {
        if (typeof document === 'undefined') return client;

        const handleVisibilityChange = (): void => {
            if (document.visibilityState === 'hidden') {
                void client.abort();

                return;
            }

            client.connect().catch(() => {
                // connect() reports failures via onError and the "error" event.
            });
        };

        document.addEventListener('visibilitychange', handleVisibilityChange);

        // Stop managing visibility after an explicit disconnect; re-register
        // when the same instance is manually connected again.
        client.once('closed', () => {
            document.removeEventListener('visibilitychange', handleVisibilityChange);

            client.once('connected', () => {
                SSESession.addBrowserVisibilityHandler(client);
            });
        });

        return client;
    }

    /** SSE endpoint URL for this session. */
    private readonly url: string;

    /**
     * Per-instance configuration.
     *
     * Defaults live on the instance field (not a shared static) so each session
     * gets its own {@link SSEEventParser} and {@link ExponentialBackoff}.
     */
    public options: SSESessionOptions = {
        fetch: (...args) => fetch(...args),
        method: 'GET',
        headers: {
            Accept: 'text/event-stream',
            'Cache-Control': 'no-cache',
        },
        body: new FormData(),

        onRequest: (request) => Promise.resolve(request),
        onConnected: () => {},
        onDisconnected: () => {},
        onError: (error) => console.error('SSEClient error:', error),

        // Retry the initial fetch until it succeeds (maxAttempts: 0 = unlimited).
        retry: new ExponentialBackoff({
            baseDelay: 1000,
            maxDelay: 10000,
            maxAttempts: 0,
            growthRate: 1.3,
            jitter: 0.3,
        }),

        attemptReconnect: true,
        persistent: false,

        eventParser: new SSEEventParser(),
    };

    /** AbortController for the currently active fetch, if any. */
    private controller: AbortController | null = null;

    /**
     * Asynchronous stream of parsed SSE events for the active connection.
     *
     * Stays open across {@link abort} and automatic reconnects so an existing
     * `for await` consumer keeps receiving events after visibility resumes.
     *
     * Closes when:
     * - the server ends the stream and {@link SSESessionOptions.persistent}
     *   is false,
     * - {@link disconnect} is called, or
     * - a transport error occurs with
     *   {@link SSESessionOptions.attemptReconnect} disabled.
     *
     * A later {@link connect} replaces this with a new iterator when the
     * previous one was closed. Consumers should read from `session.messages`
     * rather than caching a reference across terminal disconnects.
     */
    public messages: AsyncPushIterator<SSEvent> = new AsyncPushIterator<SSEvent>();

    public constructor(url: string, options: Partial<SSESessionOptions> = {}) {
        super();

        this.url = url;
        this.options = {
            ...this.options,
            ...options,
            // Shallow merge would drop default headers when options.headers is set.
            headers: { ...this.options.headers, ...options.headers },
        };
    }

    /**
     * Connects or reconnects to the SSE endpoint.
     *
     * Resolves once the HTTP stream is established and `"connected"` has been
     * emitted. Body reading continues asynchronously in the background via
     * {@link readStream}.
     *
     * @throws When the fetch retry policy exhausts attempts or the connection
     *   is superseded before the reader is handed off (in the latter case the
     *   promise resolves without throwing).
     */
    public async connect(): Promise<void> {
        // If there is already a controller present, we are already connected.
        if (this.controller) return;

        // Prepare for a fresh transport. Parser state from an abandoned connection
        // must not bleed into the next one; reopen messages if a prior terminal
        // close ended the consumer's iteration loop.
        this.resetEventParser();
        this.ensureMessageStreamOpen();

        const controller = new AbortController();
        this.controller = controller;

        const { method, headers, body } = this.options;

        const fetchBody = method === 'POST' ? body : null;

        const fetchOptions: RequestInit = {
            method,
            headers: headers || {},
            body: fetchBody,
            signal: controller.signal,
            cache: 'no-store',
        };

        let reader: ReadableStreamDefaultReader<Uint8Array>;

        try {
            reader = await this.options.retry.run(() => this.createReader(fetchOptions));
        } catch (error) {
            // A newer abort/connect superseded this attempt — leave state to the winner.
            if (this.controller !== controller) return;

            this.controller = null;

            await this.notifyDisconnected();
            await this.notifyError(error);
            this.closeMessageStream();

            throw error;
        }

        // Connection succeeded but was already replaced (for example abort during fetch).
        if (this.controller !== controller) {
            await reader.cancel();

            return;
        }

        await tryAsync(
            () => this.options.onConnected(),
            (error) => this.options.onError(error),
        );
        this.emit('connected', undefined);

        // Fire-and-forget: connect() resolves while the stream is consumed.
        this.readStream(reader, controller).catch((error) => {
            this.options.onError(error);
        });
    }

    /**
     * Aborts only the currently active transport.
     *
     * The session remains reusable: {@link messages} stays open, visibility
     * handling stays attached, and {@link connect} can reopen the stream.
     * Partial parser state from the abandoned transport is discarded.
     *
     * Emits `"disconnected"` but not `"closed"`.
     */
    public async abort(): Promise<void> {
        if (!this.controller) return;

        // Grab the current controller to ensure we are aborting the correct one.
        const controller = this.controller;
        this.controller = null;

        // Invalidate any in-flight read loop and fetch for this transport.
        controller.abort();
        this.resetEventParser();

        await this.notifyDisconnected();
    }

    /**
     * Terminates the session and disables attached visibility handling until
     * the same instance is manually {@link connect connected} again.
     *
     * Closes {@link messages} and emits `"closed"`.
     */
    public async disconnect(): Promise<void> {
        this.closeMessageStream();
        this.emit('closed', undefined);

        if (this.controller) {
            await this.abort();
        } else {
            this.resetEventParser();
        }
    }

    /**
     * Performs the HTTP request and returns a reader for the response body.
     *
     * {@link SSESessionOptions.onRequest} may mutate headers (for example auth
     * tokens or `Last-Event-ID`) before the fetch runs.
     */
    private async createReader(fetchOptions: RequestInit): Promise<ReadableStreamDefaultReader<Uint8Array>> {
        const requestOptions = await this.options.onRequest(fetchOptions);
        const response = await this.options.fetch(this.url, requestOptions);

        if (!response.ok) {
            const responseCode = response.status;
            const responseText = await response.text();

            const error = new HTTPError(responseCode, responseText);
            void this.notifyError(error);
            throw error;
        }

        if (!response.body) {
            const error = new ResponseBodyNullError();
            void this.notifyError(error);
            throw error;
        }

        return response.body.getReader();
    }

    /**
     * Reads bytes from an established stream until it ends, errors, or is
     * superseded by a newer connection.
     */
    private async readStream(reader: ReadableStreamDefaultReader<Uint8Array>, controller: AbortController): Promise<void> {
        try {
            while (this.controller === controller) {
                const { done, value } = await reader.read();

                // abort() or a newer connect() may have landed while we were awaiting.
                if (this.controller !== controller) return;

                if (done) {
                    this.controller = null;

                    await this.notifyDisconnected();

                    if (this.options.persistent) {
                        // Server closed gracefully — reopen unless the consumer opted out.
                        await this.connect();
                    } else {
                        this.closeMessageStream();
                    }

                    return;
                }

                // Some environments yield `{ done: false, value: undefined }`.
                if (!value) continue;

                for (const event of this.options.eventParser.parseEvents(value)) {
                    this.emit('message', event);
                    this.messages.push(event);
                }
            }
        } catch (error) {
            // If the controller is different, we already started a new connection and it would be confusing to handle this error.
            if (controller !== this.controller) return;

            // Invalidate the current controller to allow for reconnection if needed
            this.controller = null;

            await this.notifyDisconnected();

            // Expected path for abort() — do not treat as an error or reconnect.
            if (controller.signal.aborted) return;

            await this.notifyError(error);

            if (this.options.attemptReconnect) {
                await this.connect();
            } else {
                this.closeMessageStream();
            }
        }
    }

    /** Clears partial SSE frames left over from an abandoned transport. */
    private resetEventParser(): void {
        this.options.eventParser.reset();
    }

    /**
     * Creates a new {@link messages} iterator when the previous one was closed
     * by a terminal disconnect or server stream end.
     */
    private ensureMessageStreamOpen(): void {
        if (!this.messages.closed) return;

        this.messages = new AsyncPushIterator<SSEvent>();
    }

    /** Ends the message iteration loop for the current connection span. */
    private closeMessageStream(): void {
        if (this.messages.closed) return;

        this.messages.close();
    }

    /** Invokes {@link SSESessionOptions.onDisconnected} and emits `"disconnected"`. */
    private async notifyDisconnected(): Promise<void> {
        await tryAsync(
            () => this.options.onDisconnected(),
            (error) => this.options.onError(error),
        );
        this.emit('disconnected', undefined);
    }

    /** Invokes {@link SSESessionOptions.onError} and emits `"error"`. */
    private async notifyError(error: unknown): Promise<void> {
        const errorInstance = error instanceof Error ? error : new Error(String(error));

        await tryAsync(
            () => this.options.onError(errorInstance),
            (callbackError) => console.error('SSESession error:', callbackError),
        );
        this.emit('error', errorInstance);
    }
}