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

import type { SSEvent } from './types.ts';
import { NEW_LINE, SSE_FIELD_VALUE_REGEX, SSE_LINE_ENDINGS, SSE_TRAILING_NEWLINE_REGEX } from './constants.ts';

/**
 * Optional encoders used when decoding incoming SSE bytes and re-encoding
 * any buffered remainder between chunks.
 */
export interface SSEEventParserOptions {

    /** Decodes raw stream bytes into text. Defaults to a new `TextDecoder`. */
    textDecoder: TextDecoder;
}

/**
 * Incrementally parses Server-Sent Events (SSE) from streamed byte chunks.
 *
 * SSE payloads are line-oriented: each event is a sequence of `field: value`
 * lines terminated by a blank line. This parser accepts arbitrary chunk
 * boundaries from a live HTTP response body and emits only complete events.
 *
 * Typical usage is one parser instance per connection, calling {@link parseEvents}
 * for each chunk received from the stream:
 *
 * ```ts
 * const parser = new SSEEventParser();
 *
 * for await (const chunk of response.body) {
 *   for (const event of parser.parseEvents(chunk)) {
 *     // handle event.data, event.event, event.id, event.retry
 *   }
 * }
 * ```
 *
 * Supported fields follow the SSE spec: `data`, `event`, `id`, and `retry`.
 * Multiple `data:` lines in one event are joined with `\n`. An event is only
 * emitted once a blank line is seen and at least one `data` field was collected.
 */
export class SSEEventParser {
    readonly #textDecoder: TextDecoder;

    /** Bytes from a partial line or incomplete event, carried over to the next chunk. */
    #messageBuffer: string = '';

    /**
     * Creates a parser for one SSE stream.
     *
     * Inject custom encoders in tests or when a non-default character encoding
     * is required; production callers can rely on the defaults.
     *
     * @param options - Optional text encoders for decode/encode of stream bytes.
     */
    constructor(options: Partial<SSEEventParserOptions> = {}) {
        this.#textDecoder = options.textDecoder ?? new TextDecoder();
    }

    /**
     * Clears any buffered bytes from a partial line or incomplete event.
     *
     * Call when abandoning a transport so the next connection does not prepend
     * stale bytes to incoming chunks.
     */
    public reset(): void {
        // Clear the message buffer
        this.#messageBuffer = '';

        // Reset the decoder to clear any buffered bytes
        this.#textDecoder.decode();
    }

    /**
     * Parses all complete SSE events contained in a newly received chunk.
     *
     * The chunk is appended to any bytes buffered from earlier calls. Complete
     * events (blank-line delimited blocks with at least one `data` field) are
     * returned immediately; any trailing partial line or in-progress event stays
     * in the internal buffer until a later chunk completes it.
     *
     * @param chunk - Newly received SSE stream bytes.
     * @returns Zero or more complete parsed SSE events from this chunk.
     */
    public parseEvents(chunk: Uint8Array): SSEvent[] {
        const lines = this.getBufferedLines(chunk);

        const eventLines = lines.slice(0, -1);

        const events: SSEvent[] = [];
        let event: Partial<SSEvent> = {};
        let processedLineCount = 0;

        for (const [ index, line ] of eventLines.entries()) {
            // A blank line indicates the end of an event. If we have received data, we can complete the event
            if (line === '') {
                if (event.data !== undefined) {
                    events.push(this.completeEvent(event));
                    event = {};
                    processedLineCount = index + 1;
                }

                continue;
            }

            this.parseLine(line, event);
        }

        this.storeRemainingLines(lines, processedLineCount);

        return events;
    }

    /**
     * Appends a new chunk to the buffered bytes and splits the combined payload
     * into lines.
     *
     * Accepts `\r\n`, `\r`, and `\n` line endings so events parse correctly
     * regardless of server or platform conventions.
     */
    private getBufferedLines(chunk: Uint8Array): string[] {
        this.#messageBuffer += this.#textDecoder.decode(chunk, { stream: true });

        return this.#messageBuffer.split(SSE_LINE_ENDINGS);
    }

    /**
     * Parses one SSE field line into an in-progress event.
     *
     * Lines without a colon are ignored. A single optional space after the colon
     * is stripped from the field value, per the SSE spec.
     */
    private parseLine(line: string, event: Partial<SSEvent>): void {
        const colonIndex = line.indexOf(':');
        if (colonIndex === -1) return;

        const field = line.slice(0, colonIndex);
        const value = line.slice(colonIndex + 1).replace(SSE_FIELD_VALUE_REGEX, '');

        switch (field) {
            case 'data':
                event.data = event.data ? `${event.data}${NEW_LINE}${value}` : value;

                return;

            case 'event':
                event.event = value;

                return;

            case 'id':
                event.id = value;

                return;

            case 'retry':
                this.parseRetry(value, event);

                return;
        }
    }

    /**
     * Applies a numeric `retry:` field to an in-progress event.
     *
     * Non-numeric values are ignored rather than failing the parse.
     */
    private parseRetry(value: string, event: Partial<SSEvent>): void {
        const retry = parseInt(value, 10);

        if (!isNaN(retry)) {
            event.retry = retry;
        }
    }

    /**
     * Constructs a completed SSE event from accumulated fields.
     *
     * Trims a trailing newline from multi-line `data` values so callers receive
     * the payload without an extra line break at the end.
     */
    private completeEvent(event: Partial<SSEvent>): SSEvent {
        return {
            ...event,
            data: event.data?.replace(SSE_TRAILING_NEWLINE_REGEX, ''),
        } as SSEvent;
    }

    /**
     * Preserves incomplete trailing lines for the next received chunk.
     *
     * Only lines that were fully processed (through a completed event boundary)
     * are discarded; the remainder is re-encoded into {@link messageBuffer}.
     */
    private storeRemainingLines(lines: string[], processedLineCount: number): void {
        this.#messageBuffer = lines.slice(processedLineCount).join(NEW_LINE);
    }
}