diff --git a/eslint.config.mjs b/eslint.config.mjs index d892dda..85753b4 100644 --- a/eslint.config.mjs +++ b/eslint.config.mjs @@ -2,7 +2,7 @@ import baseConfig from '@xo-cash/eslint-config'; export default [ { - ignores: [ 'scripts/**', 'docs/**' ], + ignores: [ 'scripts/**' ], }, ...baseConfig, ]; diff --git a/package-lock.json b/package-lock.json index a3f163f..0cb6e63 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "@xo-cash/utils", - "version": "0.0.7", + "version": "0.0.8", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "@xo-cash/utils", - "version": "0.0.7", + "version": "0.0.8", "license": "MIT", "dependencies": { "@bitauth/libauth": "^3.1.0-next.8", diff --git a/package.json b/package.json index d7fce66..d50bf89 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@xo-cash/utils", - "version": "0.0.7", + "version": "0.0.8", "description": "XO Cash utilities", "type": "module", "types": "./dist/index.d.mts", diff --git a/sandbox/sandbox-llm.ts b/sandbox/sandbox-llm.ts new file mode 100644 index 0000000..b4e49b7 --- /dev/null +++ b/sandbox/sandbox-llm.ts @@ -0,0 +1,93 @@ +import { SSESession } from '../source/sse-session/index.ts'; + +// Because this is in a library, and we don't want to add the node types to this as it is intended to be used in a browser +// we will just declare the process object here locally so we don't get type errors from this script +declare const process: { + env: Record; + stdout: { + write: (data: string) => void; + }; +}; + +// Recommended URL: https://openrouter.ai/api/v1/chat/completions +// Recommended Model: nvidia/nemotron-3-nano-omni-30b-a3b-reasoning:free +// Example Command: API_KEY="your-api-key" URL="https://openrouter.ai/api/v1/chat/completions" MODEL="ibm-granite/granite-4.1-8b" PROMPT="Hello, Tell me a joke about robots?" npx tsx ./sandbox/sandbox-llm.ts + +// Read the Environment Variables for url, model, prompt and api key +const url = process.env.URL ?? 'https://openrouter.ai/api/v1/chat/completions'; +const model = process.env.MODEL ?? 'nvidia/nemotron-3-nano-omni-30b-a3b-reasoning:free'; +const prompt = process.env.PROMPT ?? 'Hello, Tell me a joke about robots?'; + +const apiKey = process.env.API_KEY ?? ''; + +// Throw an error if the api key is not set +if (!apiKey) { + throw new Error('API key is required'); +} + +// Create a function to get the auth header +const getAuthHeader = (): string => { + return `Bearer ${apiKey}`; +}; + +// Create our sse session +const sseSession = new SSESession(url, { + // LLMs use requests + method: 'POST', + + // Create the body of the request + body: JSON.stringify({ + model: model, + messages: [ + { + role: 'user', + content: prompt, + }, + ], + stream: true, + }), + + // Create a custom request handler to set the headers + onRequest: async (requestInit): Promise => { + requestInit.headers ??= {} as HeadersInit; + + // Handle typescript annoyances + const headers = requestInit.headers as Record; + + // Set our headers - We could also do this using the `headers` property in the SSESession constructor + // Doing it here to demonstrate dynamic headers, for example a signed timestamp could be used to authenticate the request. + headers.Authorization = getAuthHeader(); + headers['Content-Type'] = 'application/json'; + + return requestInit; + }, +}); + +// Connect to the SSESession +await sseSession.connect(); + +// Loop over the message chunks using `for await` +for await (const message of sseSession.messages) { + // Handle `[DONE]` (this may be specific to OpenRouter) + if (message.data === '[DONE]') { + continue; + } + + // First, we will parse the event to JSON + const responseJson = JSON.parse(message.data); + + // Then we will grab the relevant part of the response (we want to first grab the choices array) + const choices = responseJson.choices; + + // Then we will grab the first choice + const firstChoice = choices[0]; + + // Then we will grab the next chunk of text + const messageContent = firstChoice.delta?.content; + + // Then we will append the text to the console + process.stdout.write(messageContent || ''); +} + +// Just a terminal/node thing. If we dont put a new line, the console will overwrite the text with the `cwd` or next command input +process.stdout.write('\n'); diff --git a/sandbox/sandbox-price-oracle.ts b/sandbox/sandbox-price-oracle.ts new file mode 100644 index 0000000..959f32f --- /dev/null +++ b/sandbox/sandbox-price-oracle.ts @@ -0,0 +1,17 @@ +import { SSESession, type SSEvent } from '../source/sse-session/index.ts'; + +// Command: npx tsx ./sandbox/sandbox-price-oracle.ts + +// Use the GP Price Oracle API +const url = 'https://oracles.generalprotocols.com/sse/v1/messages'; + +// Create our sse session +const sseSession = await SSESession.create(url); + +// Create a message event handler +sseSession.on('message', (message: SSEvent) => { + console.log(message); +}); + +// Connect to the SSESession +await sseSession.connect(); diff --git a/source/index.ts b/source/index.ts index 3c809ad..1b15e23 100644 --- a/source/index.ts +++ b/source/index.ts @@ -2,6 +2,7 @@ export * from './errors.ts'; export * from './event-emitter.ts'; export * from './exponential-backoff/index.ts'; export * from './extended-json.ts'; +export * from './misc.ts'; export * from './script.ts'; export * from './sse-session/index.ts'; export * from './template/errors.ts'; diff --git a/source/misc.ts b/source/misc.ts index edd5854..46814c0 100644 --- a/source/misc.ts +++ b/source/misc.ts @@ -37,3 +37,15 @@ export const isWithinBounds = (value: number, min: number, max: number): boolean return true; }; + +/** + * Converts a non-error to an error + * @param error - The error to convert + * @returns The error + */ +export const normalizeError = (error: unknown): Error => { + // Convert the error to an error if it is not already an error + const errorInstance = error instanceof Error ? error : new Error(`${error}`); + + return errorInstance; +}; diff --git a/source/sse-session/constants.ts b/source/sse-session/constants.ts index 6ed2005..983ca5c 100644 --- a/source/sse-session/constants.ts +++ b/source/sse-session/constants.ts @@ -32,3 +32,26 @@ export const SSE_TRAILING_NEWLINE_REGEX = /\n$/; * buffered partial lines between streamed chunks before the next parse call. */ export const NEW_LINE = '\n'; + +/** + * Default exponential backoff options for the SSE session. + */ +export const SSE_SESSION_EXPONENTIAL_BACKOFF_DEFAULTS = { + baseDelay: 250, + maxDelay: 10000, + maxAttempts: 0, + growthRate: 1.3, + jitter: 0.3, +} as const; + +/** + * Default attempt reconnect flag for the SSE session. + * @default true + */ +export const SSE_SESSION_ATTEMPT_RECONNECT_DEFAULT = true; + +/** + * Default persistent flag for the SSE session. + * @default false + */ +export const SSE_SESSION_PERSISTENT_DEFAULT = false; diff --git a/source/sse-session/errors.ts b/source/sse-session/errors.ts new file mode 100644 index 0000000..0b44432 --- /dev/null +++ b/source/sse-session/errors.ts @@ -0,0 +1,31 @@ +/* eslint-disable max-classes-per-file */ + +/** + * Error thrown when a response body is null + */ +export class ResponseBodyNullError extends Error { + constructor() { + super('Response body is null'); + this.name = 'ResponseBodyNullError'; + } +} + +/** + * Error thrown when an HTTP error occurs + */ +export class HTTPError extends Error { + constructor(status: number, message: string) { + super(`HTTP error! Status: ${status} - ${message}`); + this.name = 'HTTPError'; + } +} + +/** + * Error thrown when a plugin's unsubscribe function is not a function + */ +export class SSESessionUnsubscribePluginNotAFunctionError extends Error { + constructor(pluginName: string) { + super(`${pluginName}'s unsubscribe function is not a function`); + this.name = 'SSESessionUnsubscribePluginNotAFunctionError'; + } +} diff --git a/source/sse-session/index.ts b/source/sse-session/index.ts index 48842e2..e00345e 100644 --- a/source/sse-session/index.ts +++ b/source/sse-session/index.ts @@ -1,2 +1,4 @@ export * from './async-push-iterator.ts'; +export * from './types.ts'; +export * from './sse-session.ts'; export * from './sse-event-parser.ts'; diff --git a/source/sse-session/sse-session.ts b/source/sse-session/sse-session.ts new file mode 100644 index 0000000..b3127a3 --- /dev/null +++ b/source/sse-session/sse-session.ts @@ -0,0 +1,635 @@ +import type { SSESessionOptions, SSESessionEventMap, SSEvent, SSESessionEventCallbacks, SSESessionOperationContext } from './types.ts'; +import { SSE_SESSION_EXPONENTIAL_BACKOFF_DEFAULTS, SSE_SESSION_ATTEMPT_RECONNECT_DEFAULT, SSE_SESSION_PERSISTENT_DEFAULT } from './constants.ts'; + +import { EventEmitter, type OffCallback } from '../event-emitter.ts'; +import { ExponentialBackoff } from '../exponential-backoff/index.ts'; +import { normalizeError } from '../misc.ts'; + +import { SSEEventParser } from './sse-event-parser.ts'; +import { AsyncPushIterator } from './async-push-iterator.ts'; +import { HTTPError, ResponseBodyNullError, SSESessionUnsubscribePluginNotAFunctionError } from './errors.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 disconnect} 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 close} aborts the transport, closes {@link messages}, and emits + * `"closed"`. + * + * Automatic reconnect is controlled by {@link SSESessionOptions.persistent} + * (server closed the stream) and + * {@link SSESessionOptions.attemptReconnect} (transport error). + * + * ## Connection supersession + * + * Each {@link connect} will create a new controller if one does not exist. Otherwise, + * it will return immediately. + */ +export class SSESession extends EventEmitter { + /** + * 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 = {}): Promise { + const session = new SSESession(url, options); + await session.connect(); + + return session; + } + + /** + * 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. + * + * Since this relies on listening to the `"message"` event, this should ideally + * be called before "{@link connect} is called so no messages (and no ids) are missed. + * + * ```ts + * const session = new SSESession(url); + * SSESession.addLastEventIdReconnect(session); + * await session.connect(); + * // Reconnects send Last-Event-ID once an event with an id is received. + * ``` + * + * @param session - The session to instrument. + * @returns The same session, for chaining. + */ + static addLastEventIdReconnect(session: SSESession): { session: SSESession; removeListener?: OffCallback } { + // If the plugin is already installed, we return the session and the remove listener function. + if (session.plugins.has(SSESession.addLastEventIdReconnect)) { + // Get the remove listener function from the session. + const removeListener = session.plugins.get(SSESession.addLastEventIdReconnect) as OffCallback; + + // If the plugin was added with a non-function listener, we throw an error. + if (typeof removeListener !== 'function') { + throw new SSESessionUnsubscribePluginNotAFunctionError('addLastEventIdReconnect'); + } + + // If the plugin is already installed, we return the session and the remove listener function. + return { session, removeListener }; + } + + // Store the last event id so we can send it on reconnect. + let lastEventId: string | undefined; + + // Define the message listener that will store the last event id and begin listening for 'message' events. + const messageListener = (event: SSEvent): void => { + // If the event has no id, we should not update the lastEventId and instead keep the old one. + lastEventId = event.id ?? lastEventId; + }; + + session.on('message', messageListener); + + // Store the original onRequest callback so we can extend its behavior to write the reconnect header. + const originalOnRequest = session.options.onRequest; + + // Whether we actively send the Last-Event-ID header on reconnect. + // Because the `onRequest` callback may be extended further by other code, we can't just roll-back to the `originalOnRequest` + // Instead, we just use this flag to determine if we should write the Last-Event-ID header. + let enabled = true; + + // Extend the onRequest callback to write the Last-Event-ID header if it is enabled. + session.options.onRequest = async (request: RequestInit): Promise => { + if (lastEventId && enabled) { + request.headers = { ...request.headers, 'Last-Event-ID': lastEventId }; + } + + return originalOnRequest(request); + }; + + // Define the remove listener function that will disable the Last-Event-ID header on reconnect. + const removeListener = (): void => { + enabled = false; + session.off('message', messageListener); + + // If the remove listener function is the same as the existing remove listener function, we remove the plugin from the session. + // This protects against the `removeListener` function being called onto after a new instance of this plugin is added + const existingRemoveListener = session.plugins.get(SSESession.addLastEventIdReconnect) as OffCallback; + if (existingRemoveListener === removeListener) { + session.plugins.delete(SSESession.addLastEventIdReconnect); + } + }; + + // Store the plugin in the session. + session.plugins.set(SSESession.addLastEventIdReconnect, removeListener); + + return { session, removeListener }; + } + + /** + * Pauses and resumes a session based on browser tab visibility. + * + * Uses the Page Visibility API (`document.visibilitychange`): + * + * - **hidden** — {@link disconnect} stops the active fetch. {@link messages} + * stays open; `"disconnected"` fires but `"closed"` does not. + * - **visible** — {@link connect} re-establishes the stream if needed. + * + * This controller will not re-connect if the session was disconnected by something else. + * + * No-op in non-browser environments where `document` is undefined. + * + * @param session - The session to manage. + */ + static addBrowserVisibilityHandler(session: SSESession): { session: SSESession; removeListener?: OffCallback } { + if (typeof document === 'undefined') return { session }; + + // If the plugin is already installed, we return the session and the remove listener function. + if (session.plugins.has(SSESession.addBrowserVisibilityHandler)) { + // Get the remove listener function from the session. + const removeListener = session.plugins.get(SSESession.addBrowserVisibilityHandler) as OffCallback; + + // If the plugin was added with a non-function listener, we throw an error. + if (typeof removeListener !== 'function') { + throw new SSESessionUnsubscribePluginNotAFunctionError('addBrowserVisibilityHandler'); + } + + // If the plugin is already installed, we return the session and the remove listener function. + return { session, removeListener }; + } + + // Create a listener that will reconnect the session + // This listener is added when the hidden->disconnect listener disconnects the session, but will not be added + // if the session was disconnected by something else, like a manual disconnect, server-disconnect, or another plugin. + const reconnectOnVisible = async (): Promise => { + if (document.visibilityState === 'visible') { + // Remove this listener + document.removeEventListener('visibilitychange', reconnectOnVisible); + + // Reconnect the session + await session.connect(); + } + }; + + // Create a listener that will disable the reconnect listener if the session is disconnected by something else. + const disableOnExternalDisconnect = ({ source }: SSESessionOperationContext): void => { + // If the source is not the browser visibility handler, we remove the listener. + if (source !== SSESession.addBrowserVisibilityHandler) { + document.removeEventListener('visibilitychange', reconnectOnVisible); + } + }; + + // Create a listener that will live as-long as this session is not closed + // This listener will create a new 'on visible' listener that will reconnect the session when it disconnects the session. + // This prevents it from trying to reconnect the session if the session was disconnected by something else + const handleVisibilityChange = async (): Promise => { + // If the session is not connected, we do nothing. + if (!session.active) { + return; + } + + // If the session is connected and the visibility is hidden, we disconnect it. + if (document.visibilityState === 'hidden') { + // Add the listener for connected event to reconnect the session. + document.addEventListener('visibilitychange', reconnectOnVisible); + + // Disconnect the session. + session.disconnect({ source: SSESession.addBrowserVisibilityHandler }); + } + }; + + // Add the listener for visibility change. + document.addEventListener('visibilitychange', handleVisibilityChange); + + // Add a listener for `beforeDisconnect` event to disable the reconnect listener if the session is disconnected by something else. + session.on('beforeDisconnect', disableOnExternalDisconnect); + + // Remove the listener for visibility change. + const removeListener = (): void => { + document.removeEventListener('visibilitychange', handleVisibilityChange); + document.removeEventListener('visibilitychange', reconnectOnVisible); + + session.off('beforeDisconnect', disableOnExternalDisconnect); + + // If the remove listener function is the same as the existing remove listener function, we remove the plugin from the session. + // This protects against the `removeListener` function being called onto after a new instance of this plugin is added + const existingRemoveListener = session.plugins.get(SSESession.addBrowserVisibilityHandler) as OffCallback; + if (existingRemoveListener === removeListener) { + session.plugins.delete(SSESession.addBrowserVisibilityHandler); + } + }; + + // Store the plugin in the session. + session.plugins.set(SSESession.addBrowserVisibilityHandler, removeListener); + + // Return the session and the remove listener function. + return { session, removeListener }; + } + + /** SSE endpoint URL for this session. */ + 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 options + fetch: (...args) => fetch(...args), + method: 'GET', + headers: { + Accept: 'text/event-stream', + 'Cache-Control': 'no-cache', + }, + body: new FormData(), + + // callback to mutate the fetch options before the fetch is made. + onRequest: (request) => Promise.resolve(request), + + // Retry the initial fetch. + // Use the spread operator so it's impossible for us to override the default options. (default is exponential backoff) + retry: new ExponentialBackoff({ + ...SSE_SESSION_EXPONENTIAL_BACKOFF_DEFAULTS, + }), + + // Reconnection options + attemptReconnect: SSE_SESSION_ATTEMPT_RECONNECT_DEFAULT, + persistent: SSE_SESSION_PERSISTENT_DEFAULT, + + // Event parser options + eventParser: new SSEEventParser(), + }; + + /** + * Registered plugins for the session. + * This can be used to prevent duplicate plugins from being added to the session. + * The value for a plugin is arbitrary, for example a function that removes the plugin. + */ + public readonly plugins: Map = new Map(); + + /** AbortController for the currently active fetch, if any. */ + #connectionController: AbortController | null = null; + + /** The server's requested retry interval in milliseconds (per SSE spec) if any */ + #retryInterval: number | null = null; + + /** + * Asynchronous stream of parsed SSE events for the active connection. + * + * Stays open across {@link disconnect} 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 close} 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 = new AsyncPushIterator(); + + public constructor(url: string, options: Partial = {}) { + super(); + + const { onConnected, onDisconnected, onError, ...restOptions } = options; + + this.#url = url; + this.options = { + ...this.options, + ...restOptions, + // Shallow merge would drop default headers when options.headers is set. + headers: { ...this.options.headers, ...options.headers }, + }; + + // Add the onConnected, onDisconnected, and onError as event listeners to the session. + // We accept these as options to the constructor so that we can both instantiate the session and add the event listeners at the same time, + // avoiding any potential race-conditions or async/await issues. + if (onConnected) { + this.on('connected', onConnected); + } + + if (onDisconnected) { + this.on('disconnected', onDisconnected); + } + + if (onError) { + this.on('error', onError); + } + } + + /** + * Returns true if the session is active by checking if the controller is not null. + */ + public get active(): boolean { + return this.#connectionController !== null; + } + + /** + * 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 the + * internal `#readStream` method. + * + * @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 { + // If there is already a controller present, we are already connected. + if (this.#connectionController) 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 connectionController = new AbortController(); + this.#connectionController = connectionController; + + const { method, headers, body } = this.options; + + // Create the fetch options. + const fetchOptions: RequestInit = { + method, + headers: headers || {}, + signal: connectionController.signal, + cache: 'no-store', + }; + + // If the method is POST, we will set the body of the fetch options. + // NOTE: RequestInit doesn't allow body to be `undefined`, so if its false-y we will set it to null. + if (method === 'POST') { + fetchOptions.body = body || null; + } + + // Create a reader for the response body. + let reader: ReadableStreamDefaultReader; + + try { + // Retry the fetch using the provided retry policy. (default is exponential backoff) + reader = await this.options.retry.run(() => this.#createReader(fetchOptions), { + signal: connectionController.signal, + }); + } catch (error) { + // A newer `connect()` superseded this attempt — leave state to the winner. + if (this.#connectionController !== connectionController) return; + + // Reset the controller to null to allow for reconnection if needed. + this.#connectionController = null; + + // Normalize the error so it can be passed to the event listeners. + const normalizedError = normalizeError(error); + + // Emit the disconnected and error events. + this.emit('disconnected', { reason: 'error', error: normalizedError }); + this.emit('error', normalizedError); + + // Close the message stream so `for await...of` consumers resolve. + this.#closeMessageStream(); + + // Throw the error so the caller can handle it. + throw error; + } + + // Connection succeeded but was already replaced (for example disconnect during fetch). + if (this.#connectionController !== connectionController) { + await reader.cancel(); + + return; + } + + this.emit('connected', undefined); + + // Fire-and-forget: connect() resolves while the stream is consumed. + // Data is emitted as a `message` event and from the `messages` iterator. + this.#readStream(reader, connectionController).catch((error) => { + this.emit('error', error); + }); + } + + /** + * Disconnects only the currently active transport. + * + * `beforeDisconnect` is emitted on every invocation, even when no + * transport is currently active. This allows observers to react to + * an explicit disconnect operation. + * + * @param context - Context describing the operation initiator. + * @emits `beforeDisconnect` with the supplied operation context. + * @emits `disconnected` with reason `"disconnect"` when an active + * transport is actually disconnected. + */ + public disconnect(context: SSESessionOperationContext = {}): void { + this.emit('beforeDisconnect', context); + + if (this.#connectionController) { + // Grab the current controller to ensure we are aborting the correct one. + const connectionController = this.#connectionController; + this.#connectionController = null; + + // Invalidate any in-flight read loop and fetch for this transport. + connectionController.abort(); + + // Emit the disconnected event. + this.emit('disconnected', { + reason: 'disconnect', + source: context.source, + }); + } + + // Reset the event parser buffer so data is not carried through resulting in malformed events. + this.#resetEventParser(); + } + + /** + * 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 close(): void { + // Disconnect the session. + this.disconnect(); + + // Close the message stream so `for await...of` consumers resolve. + this.#closeMessageStream(); + + // Emit the closed event. + this.emit('closed', undefined); + } + + /** + * 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. + */ + async #createReader(fetchOptions: RequestInit): Promise> { + const requestOptions = await this.options.onRequest(fetchOptions); + const response = await this.options.fetch(this.#url, requestOptions); + + // Handle a bad response from the server + if (!response.ok) { + // Get the response code and text. + const responseCode = response.status; + const responseText = await response.text(); + + // Create a new error and emit it. + const error = new HTTPError(responseCode, responseText); + this.emit('error', error); + + // Throw the error so the caller can handle it. + throw error; + } + + // Handle a response with no body (SSE still uses the response body) + if (!response.body) { + const error = new ResponseBodyNullError(); + this.emit('error', error); + throw error; + } + + // Return the reader for the response body. + return response.body.getReader(); + } + + /** + * Reads bytes from an established stream until it ends, errors, or is + * superseded by a newer connection. + */ + async #readStream(reader: ReadableStreamDefaultReader, connectionController: AbortController): Promise { + try { + while (this.#connectionController === connectionController) { + const { done, value } = await reader.read(); + + // `disconnect()` or a newer connect() may have landed while we were awaiting. + if (this.#connectionController !== connectionController) return; + + // If the stream is done, we've been disconnected. + // We need to clean up and either complete or reconnect the session. + if (done) { + // Reset the controller to null to allow for reconnection if needed. + this.#connectionController = null; + + // Emit the disconnected event. + this.emit('disconnected', { reason: 'remote' }); + + // If the session is persistent, we will attempt to reconnect. + // NOTE: in very rare cases, a malfunctioning server can make this code turn into an infinite loop. + if (this.options.persistent) { + // Server closed gracefully — reopen unless the consumer opted out. + await this.connect(); + } else { + // Close the message stream so `for await...of` consumers resolve. + this.#closeMessageStream(); + } + + return; + } + + // Some environments yield `{ done: false, value: undefined }`. + if (!value) continue; + + // Emit the parsed events to the event listeners and the messages iterator. + for (const event of this.options.eventParser.parseEvents(value)) { + // If the event has a retry interval, we will store it. + if (event.retry) { + this.#retryInterval = event.retry; + } + + 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. + // The controller can also be null here if the session was disconnected. + if (connectionController !== this.#connectionController) return; + + // Invalidate the current controller to allow for reconnection if needed + this.#connectionController = null; + + const normalizedError = normalizeError(error); + this.emit('disconnected', { reason: 'error', error: normalizedError }); + + // Expected path for `disconnect()` — do not treat as an error or reconnect. + if (connectionController.signal.aborted) return; + + this.emit('error', normalizedError); + + // If the session is configured to attempt reconnect, we will attempt to reconnect. + if (this.options.attemptReconnect) { + // If the server has requested a retry interval, we will wait for it before attempting to reconnect. + if (this.#retryInterval) { + await new Promise((resolve) => setTimeout(resolve, this.#retryInterval!)); + } + + // Attempt to reconnect. + await this.connect(); + } else { + // Close the message stream so `for await...of` consumers resolve. + this.#closeMessageStream(); + } + } + } + + /** Clears partial SSE frames left over from an abandoned transport. */ + #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. + */ + #ensureMessageStreamOpen(): void { + if (!this.messages.closed) return; + + this.messages = new AsyncPushIterator(); + } + + /** Ends the message iteration loop for the current connection span. */ + #closeMessageStream(): void { + if (this.messages.closed) return; + + this.messages.close(); + } +} diff --git a/source/sse-session/types.ts b/source/sse-session/types.ts index 224e048..9c4d08b 100644 --- a/source/sse-session/types.ts +++ b/source/sse-session/types.ts @@ -1,3 +1,173 @@ +import type { ExponentialBackoffRunOptions } from '../exponential-backoff/exponential-backoff'; + +export type SSERequestInit = { + + /** + * The HTTP method to use. + */ + method: 'GET' | 'POST'; + + /** + * Request headers sent on every connect and reconnect. + */ + headers?: Record; + + /** + * Request body for POST-based SSE endpoints. + */ + body?: string | FormData; +}; + +/** + * The fetch function to use. + * + * NOTE: This is compatible with Browser/Node's native "fetch" function. + * We use this in place of "typeof fetch" so that we can accept non-standard URLs ("url" is a "string" here). + * For example, a LibP2P adapter might not use a standardized URL format (and might only include "path"). + * This would cause a type error as native fetch expects type "URL". + */ +export type SSERequestFunction = { + fetch: (url: string, options: RequestInit) => Promise; +}; + +/** + * Lifecycle hooks invoked by {@link SSESession} during connect, read, and teardown. + */ +export type SSESessionCallbacks = { + + /** + * Called before each fetch so callers can attach auth headers, cookies, or + * a `Last-Event-ID` for resume semantics. + */ + onRequest: (request: RequestInit) => Promise; +}; + +export type SSESessionEventCallbacks = { + + /** + * Called after the HTTP stream is established and before body reading begins. + */ + onConnected: () => void; + + /** + * Called when the active transport ends — including {@link SSESession.disconnect}, + * server stream completion, and errors. Not paired with {@link SSESessionEventCallbacks.onConnected} + * when the initial connect never succeeds. + */ + onDisconnected: () => void; + + /** + * Called on fetch or read failures. Not invoked for intentional + * {@link SSESession.disconnect} disconnects. + */ + onError: (error: Error) => void; +}; + +export type SSESessionRetryInterface = { + + /** + * Retry policy used while establishing the HTTP connection in + * {@link SSESession.connect}. Defaults to {@link ExponentialBackoff} with + * unlimited attempts. + */ + retry: { + run(fn: () => Promise, options?: ExponentialBackoffRunOptions): Promise; + }; +}; + +export interface SSEParser { + + /** + * Incremental SSE frame parser for the response body. + * + * {@link SSEEventParser.reset} is called by the session when abandoning a + * transport so partial frames do not carry over to the next connection. + */ + eventParser: { + parseEvents(buffer: Uint8Array): SSEvent[]; + reset(): void; + }; +} + +export type SSELifecycleOptions = { + + /** + * When true, {@link SSESession} calls {@link SSESession.connect} again after + * a transport **error** (not an intentional disconnect). + * @default true + */ + attemptReconnect: boolean; + + /** + * When true, {@link SSESession} calls {@link SSESession.connect} again after + * the **server** closes the stream normally (`done`). + * @default false + */ + persistent: boolean; +}; + +/** + * Context from the source that triggered the event. + */ +export type SSESessionOperationContext = { + source?: unknown; +}; + +/** + * Discriminated context for the reason the session was disconnected internally. + */ +export type SSESessionDisconnectInternalContext = { + reason: 'disconnect'; +} & SSESessionOperationContext; + +/** + * Discriminated context for the reason the session was disconnected remotely. + */ +export type SSESessionDisconnectRemoteContext = { + reason: 'remote'; +}; + +/** + * Discriminated context for the reason the session was disconnected due to an error. + */ +export type SSESessionDisconnectErrorContext = { + reason: 'error'; + error: Error; +}; + +/** + * Discriminated context for the reason the session was disconnected. + */ +export type SSESessionDisconnectContext = SSESessionDisconnectInternalContext | SSESessionDisconnectRemoteContext | SSESessionDisconnectErrorContext; + +/** + * Events emitted by {@link SSESession}. + * + * - `"connected"` — HTTP stream established. + * - `"message"` — A complete SSE event was parsed. + * - `"disconnected"` — The active transport ended (including {@link SSESession.disconnect}). + * - `"error"` — An unexpected fetch or read failure. + * - `"closed"` — {@link SSESession.close} was called. + */ +export type SSESessionEventMap = { + connected: void; + beforeDisconnect: SSESessionOperationContext; + disconnected: SSESessionDisconnectContext; + error: Error; + message: SSEvent; + closed: void; +}; + +/** + * Configuration for {@link SSESession}. + */ +export type SSESessionOptions = SSESessionCallbacks + & SSERequestInit + & SSERequestFunction + & SSESessionRetryInterface + & SSELifecycleOptions + & SSEParser; + /** * Represents a Server-Sent Event. */ diff --git a/test/event-emitter.test.ts b/test/event-emitter.test.ts index 0f09bd8..9c51930 100644 --- a/test/event-emitter.test.ts +++ b/test/event-emitter.test.ts @@ -101,6 +101,9 @@ const testEventEmitterEmitContinuesAfterListenerThrows = (): void => { const emitter = new EventEmitter(); const secondListener = vi.fn(); + // Spy on the console.error method + const consoleErrorSpy = vi.spyOn(console, 'error').mockImplementation(() => {}); + emitter.on('message', (): void => { throw new Error('listener failure'); }); @@ -110,6 +113,12 @@ const testEventEmitterEmitContinuesAfterListenerThrows = (): void => { expect(secondListener).toHaveBeenCalledOnce(); expect(secondListener).toHaveBeenCalledWith('hello'); + + expect(consoleErrorSpy).toHaveBeenCalledOnce(); + expect(consoleErrorSpy).toHaveBeenCalledWith(new Error('listener failure')); + + // Remove the mock on console.error + consoleErrorSpy.mockRestore(); }; /** diff --git a/test/misc.test.ts b/test/misc.test.ts new file mode 100644 index 0000000..9dbdc64 --- /dev/null +++ b/test/misc.test.ts @@ -0,0 +1,59 @@ +import { expect, test } from 'vitest'; +import { normalizeError } from '../source/misc.ts'; + +/** Test cases for the normalizeError function. */ +const normalizeErrorTestCases = [ + { + input: new Error('test error'), + expected: new Error('test error'), + }, + { + input: 'test error', + expected: new Error('test error'), + }, + { + input: 1, + expected: new Error('1'), + }, + { + input: undefined, + expected: new Error('undefined'), + }, + { + input: null, + expected: new Error('null'), + }, + { + input: false, + expected: new Error('false'), + }, + { + input: true, + expected: new Error('true'), + }, + { + input: {}, + expected: new Error('[object Object]'), + }, + { + input: [], + expected: new Error(), + }, +]; + +/** + * Tests that normalizeError converts non-Error throws to Error instances. + */ +const testNormalizeError = async (): Promise => { + for (const testCase of normalizeErrorTestCases) { + const result = normalizeError(testCase.input); + expect(result).toBeInstanceOf(Error); + expect(result.message).toBe(testCase.expected.message); + } +}; + +const runTests = async (): Promise => { + test('normalizeError: converts non-Error throws to Error instances', testNormalizeError); +}; + +await runTests(); diff --git a/test/sse-session/helpers/sse-stream.ts b/test/sse-session/helpers/sse-stream.ts new file mode 100644 index 0000000..43e372a --- /dev/null +++ b/test/sse-session/helpers/sse-stream.ts @@ -0,0 +1,65 @@ +export type SseTestStreamOptions = { + + /** Milliseconds to wait before enqueueing each chunk after the first. */ + chunkDelayMs?: number; + + /** + * When true, closes the body as soon as all initial chunks have been sent. + * Use this to simulate a server that sends events and then ends the stream. + */ + closeWhenDone?: boolean; +}; + +/** + * Test double for an SSE HTTP response body. + * + * Enqueues fixture chunks in order and stays open until {@link close} is called, + * matching real servers that keep the connection alive after each event's trailing + * `\n\n` frame boundary. + */ +export class SseTestStream { + readonly stream: ReadableStream; + + private controller: ReadableStreamDefaultController | null = null; + + private closed = false; + + /** + * @param chunks - Fixture `raw` strings, whole or split, to simulate chunk boundaries. + * @param options - Delivery timing and optional auto-close after the initial chunks. + */ + constructor(chunks: string[], options: SseTestStreamOptions = {}) { + const { chunkDelayMs = 0, closeWhenDone = false } = options; + const encoder = new TextEncoder(); + + this.stream = new ReadableStream({ + start: async (controller): Promise => { + this.controller = controller; + + for (let i = 0; i < chunks.length; i++) { + if (chunkDelayMs > 0 && i > 0) { + await new Promise((resolve) => setTimeout(resolve, chunkDelayMs)); + } + + if (this.closed) return; + + controller.enqueue(encoder.encode(chunks[i]!)); + } + + if (closeWhenDone) { + this.close(); + } + }, + }); + } + + /** + * Ends the HTTP body the way a server closing the SSE connection would. + */ + close(): void { + if (this.closed) return; + + this.closed = true; + this.controller?.close(); + } +} diff --git a/test/sse-session/sse-session.test.ts b/test/sse-session/sse-session.test.ts new file mode 100644 index 0000000..6e343ea --- /dev/null +++ b/test/sse-session/sse-session.test.ts @@ -0,0 +1,1185 @@ +import { afterEach, expect, test, vi, type Mock } from 'vitest'; + +import { SSESession } from '../../source/sse-session/sse-session.ts'; +import { ExponentialBackoff, ExponentialBackoffMaxRetriesHitError } from '../../source/exponential-backoff/index.ts'; +import type { SSESessionEventCallbacks, SSESessionOptions, SSEvent } from '../../source/sse-session/types.ts'; + +import { HTTPError, ResponseBodyNullError, SSESessionUnsubscribePluginNotAFunctionError } from '../../source/sse-session/errors.ts'; +import * as utils from '../../source/misc.ts'; + +import { SseTestStream, type SseTestStreamOptions } from './helpers/sse-stream.ts'; +import { priceOracleEvents, storageEvents } from './fixtures/events.fixtures.ts'; + +class DocumentStub extends EventTarget { + public visibilityState: DocumentVisibilityState = 'visible'; +} + +const documentStub = new DocumentStub(); + +vi.stubGlobal('document', documentStub); + +/** URL passed to every session under test. */ +const EVENTS_URL = '/events'; + +/** Headers required for a valid SSE response in these tests. */ +const SSE_HEADERS = { 'Content-Type': 'text/event-stream' }; + +type FetchFn = SSESessionOptions['fetch']; + +/** + * Builds a minimal ExponentialBackoff so reconnect and retry paths finish quickly in tests. + * Real production delays would make vi.waitFor-based assertions time out. + * + * @param maxAttempts - Maximum retry attempts; defaults to 1. + */ +const testRetry = (maxAttempts = 1): ExponentialBackoff => { + return new ExponentialBackoff({ + baseDelay: 1, + maxDelay: 1, + maxAttempts, + growthRate: 1, + jitter: 0, + }); +}; + +/** + * Wraps fixture chunks in a Response backed by {@link SseTestStream}. + * SseTestStream simulates a real HTTP body: chunks arrive over time and the stream + * can optionally close itself when all chunks are sent. + * + * @param chunks - Raw SSE payload strings to stream. + * @param options - Optional stream timing and close behavior. + */ +const sseFetchResponse = (chunks: string[], options: SseTestStreamOptions = {}): Response => { + return new Response(new SseTestStream(chunks, options).stream, { + status: 200, + headers: SSE_HEADERS, + }); +}; + +/** + * Creates a vitest mock fetch that delegates to the given responder. + * SSESession requires fetch injection so tests never hit the network. + * + * @param responder - Function that returns the Response for each fetch call. + */ +const createFetchMock = (responder: (url: string, init: RequestInit) => Response | Promise): Mock => { + return vi.fn(async (url: string, init: RequestInit) => responder(url, init)); +}; + +/** + * Session defaults that disable reconnect noise unless a test opts in. + * attemptReconnect: false — transport errors should not auto-retry by default. + * persistent: false — server closing the stream should close the message iterator. + */ +const defaultSessionOptions: Partial = { + attemptReconnect: false, + persistent: false, + retry: testRetry(), +}; + +/** + * Creates an SSESession wired to the injected fetch mock. + * SSESession.create immediately calls connect(), so fetch is invoked during creation. + * + * @param fetch - Mock fetch implementation. + * @param options - Per-test session overrides merged on top of defaults. + */ +const createSession = async (fetch: FetchFn, options: Partial = {}): Promise => { + return SSESession.create(EVENTS_URL, { + ...defaultSessionOptions, + onError: vi.fn(), + ...options, + fetch, + }); +}; + +/** + * Runs a callback against a session and always closes the session afterward. + * Most tests use this so session lifecycle (connect on create, disconnect on exit) + * is consistent and resources are not leaked between cases. + * + * @param fetch - Mock fetch for the session. + * @param options - Session options. + * @param run - Test body receiving the connected session. + */ +const withSession = async ( + fetch: FetchFn, + options: Partial, + run: (session: SSESession) => Promise, +): Promise => { + const session = await createSession(fetch, options); + + try { + return await run(session); + } finally { + session.close(); + } +}; + +/** + * Reads up to `count` messages from the session's async iterator. + * Breaking out of the for-await loop early leaves the underlying stream open, + * which is intentional for tests that inspect post-read session state. + * + * @param session - Connected SSE session. + * @param count - Number of events to collect before stopping. + */ +const readMessages = async (session: SSESession, count: number): Promise => { + const events: SSEvent[] = []; + + for await (const event of session.messages) { + events.push(event); + + if (events.length >= count) { + break; + } + } + + return events; +}; + +/** + * Builds a ReadableStream that emits one chunk then errors after a delay. + * Simulates a mid-stream network failure: the client receives partial data, + * then the connection drops before the server finishes sending. + * + * @param raw - SSE payload to enqueue before the error. + * @param delayMs - Milliseconds to wait before erroring (gives the parser time to process the chunk). + */ +const failingStreamAfter = (raw: string, delayMs = 50): ReadableStream => { + const encoder = new TextEncoder(); + + return new ReadableStream({ + async start(controller): Promise { + controller.enqueue(encoder.encode(raw)); + await new Promise((resolve) => setTimeout(resolve, delayMs)); + controller.error(new Error('network failure')); + }, + }); +}; + +/** + * Scenario: a client opens a standard GET SSE connection. + * Verifies the session passes the correct fetch options for a spec-compliant SSE request. + */ +const testSseSessionConnectCallsFetchWithExpectedOptions = async (): Promise => { + // Return one valid storage fixture event so connect succeeds and the stream stays open briefly. + const fetchMock = createFetchMock(() => sseFetchResponse([ storageEvents[0]!.raw ])); + + try { + // withSession → createSession → SSESession.create → connect → fetchMock is called once. + await withSession(fetchMock, {}, async () => { + expect(fetchMock).toHaveBeenCalledWith( + EVENTS_URL, + expect.objectContaining({ + method: 'GET', + cache: 'no-store', + signal: expect.any(AbortSignal), + headers: expect.objectContaining({ + Accept: 'text/event-stream', + 'Cache-Control': 'no-cache', + }), + }), + ); + + // While the stream is active the abort signal must not yet be triggered. + // Grab the first call to fetchMock + const firstCall = fetchMock.mock.calls[0]; + + // Grab the second parameter of the first call to fetchMock (the request init) + // - `fetch(url, options)` - we are grabbing the options object. + const firstCallParams = firstCall?.[1]; + + // Assert the abort signal is not yet triggered. If it is, the stream is closed. + expect(firstCallParams?.signal?.aborted).toBe(false); + }); + } finally { + vi.restoreAllMocks(); + } +}; + +const testSseSessionUsesDefaultFetchImplementation = async (): Promise => { + const fetchMock = createFetchMock(() => sseFetchResponse([ storageEvents[0]!.raw ])); + vi.stubGlobal('fetch', fetchMock); + + const session = new SSESession('http://test.com'); + await session.connect(); + + expect(fetchMock).toHaveBeenCalledWith( + 'http://test.com', + expect.objectContaining({ + method: 'GET', + cache: 'no-store', + signal: expect.any(AbortSignal), + headers: { + Accept: 'text/event-stream', + 'Cache-Control': 'no-cache', + }, + }), + ); +}; + +/** + * Scenario: a client connects and the application wants a lifecycle callback when the stream is ready. + * Verifies onConnected fires after the transport is established. + */ +const testSseSessionConnectInvokesOnConnected = async (): Promise => { + const onConnected = vi.fn(); + const fetchMock = createFetchMock(() => sseFetchResponse([ storageEvents[0]!.raw ])); + + try { + await withSession(fetchMock, { onConnected }, async () => { + expect(onConnected).toHaveBeenCalledOnce(); + }); + } finally { + vi.restoreAllMocks(); + } +}; + +/** + * Scenario: a client is already connected and something calls connect() again (e.g. a duplicate init). + * Verifies the session does not open a second HTTP transport. + */ +const testSseSessionConnectDoesNotOpenSecondTransport = async (): Promise => { + const fetchMock = createFetchMock(() => sseFetchResponse([ storageEvents[0]!.raw ])); + + try { + await withSession(fetchMock, {}, async (session) => { + // Pull one event so the first transport is fully established and reading. + await readMessages(session, 1); + + // The session has already been connected, so we expect the fetchMock to have been called once. + expect(fetchMock).toHaveBeenCalledOnce(); + + // Idempotent connect — should be a no-op at the fetch layer. + // If this were to create a second transport, it would increment the fetchMock call count by one. + await session.connect(); + + // Since we are testing that a second transport is not opened, we expect the fetchMock to still have only been called once. + expect(fetchMock).toHaveBeenCalledOnce(); + }); + } finally { + vi.restoreAllMocks(); + } +}; + +/** + * Scenario: the application attaches auth or other headers via onRequest before each fetch. + * Verifies mutations from onRequest reach the actual fetch call. + */ +const testSseSessionConnectPassesOnRequestMutations = async (): Promise => { + const fetchMock = createFetchMock(() => sseFetchResponse([ storageEvents[0]!.raw ])); + + try { + await withSession( + fetchMock, + { + // onRequest runs during connect and can rewrite headers/body before fetch sees them. + onRequest: async (request) => ({ + ...request, + headers: { ...request.headers, Authorization: 'Bearer test-token' }, + }), + }, + async () => { + expect(fetchMock).toHaveBeenCalledWith( + EVENTS_URL, + expect.objectContaining({ + headers: expect.objectContaining({ + Authorization: 'Bearer test-token', + }), + }), + ); + }, + ); + } finally { + vi.restoreAllMocks(); + } +}; + +/** + * Scenario: some SSE endpoints require POST with a form body instead of a plain GET. + * Verifies method and body are forwarded to fetch. + */ +const testSseSessionConnectSendsPostBody = async (): Promise => { + const body = new FormData(); + body.set('topic', 'prices'); + + const fetchMock = createFetchMock(() => sseFetchResponse([ storageEvents[0]!.raw ])); + + try { + await withSession( + fetchMock, + { + method: 'POST', + body, + }, + async () => { + expect(fetchMock).toHaveBeenCalledWith( + EVENTS_URL, + expect.objectContaining({ + method: 'POST', + body, + }), + ); + }, + ); + } finally { + vi.restoreAllMocks(); + } +}; + +/** + * Scenario: a single long-lived connection delivers many events from different domains (LLM, oracle, storage). + * Verifies the parser and session deliver every fixture event in order through one stream. + */ +const testSseSessionDeliversMultipleFixtureEvents = async (): Promise => { + const fixtures = [ ...priceOracleEvents, ...storageEvents ]; + const expected = fixtures.flatMap(({ parsed }) => parsed ?? []); + + // All raw payloads are concatenated into one SseTestStream response. + const fetchMock = createFetchMock(() => sseFetchResponse(fixtures.map(({ raw }) => raw))); + + try { + await withSession(fetchMock, {}, async (session) => { + const received = await readMessages(session, expected.length); + + expect(received).toEqual(expected); + }); + } finally { + vi.restoreAllMocks(); + } +}; + +/** + * Scenario: application code listens via session.on('message') instead of the async iterator. + * Verifies the EventEmitter path receives the same parsed events as the iterator. + */ +const testSseSessionEmitsMessageEvents = async (): Promise => { + const { raw, parsed } = storageEvents[0]!; + const emitted: SSEvent[] = []; + + // ':\n\n' is an SSE comment/heartbeat; chunkDelayMs forces it to arrive as a separate chunk + // so incremental parsing is exercised before the real data frame. + const fetchMock = createFetchMock(() => sseFetchResponse([ ':\n\n', raw ], { chunkDelayMs: 50 })); + + // Use createSession directly (not withSession) so we control disconnect timing in finally. + const session = await createSession(fetchMock, {}); + + try { + session.on('message', (event) => emitted.push(event)); + + await readMessages(session, 1); + + expect(emitted).toEqual([ parsed![0] ]); + } finally { + session.close(); + vi.restoreAllMocks(); + } +}; + +/** + * Scenario: network chunks split an SSE frame at an arbitrary byte boundary. + * Verifies the internal parser buffers partial data and still emits a complete event. + */ +const testSseSessionParsesEventSplitAcrossChunks = async (): Promise => { + const { raw, parsed } = storageEvents[0]!; + const mid = Math.floor(raw.length / 2); + + // SseTestStream sends each array element as a separate chunk — the event is cut in half. + const fetchMock = createFetchMock(() => sseFetchResponse([ raw.slice(0, mid), raw.slice(mid) ])); + + try { + await withSession(fetchMock, {}, async (session) => { + const [ event ] = await readMessages(session, 1); + + expect(event).toEqual(parsed![0]); + }); + } finally { + vi.restoreAllMocks(); + } +}; + +/** + * Scenario: server closes the stream after one event and the client is not in persistent mode. + * Verifies the message iterator closes and onDisconnected fires. + */ +const testSseSessionClosesOnServerCloseWhenNotPersistent = async (): Promise => { + // closeWhenDone: true makes SseTestStream end the body after sending the chunk. + const fetchMock = createFetchMock(() => sseFetchResponse([ storageEvents[0]!.raw ], { closeWhenDone: true })); + const disconnected = vi.fn(); + + try { + // persistent defaults to false — terminal server close should shut down messages. + await withSession(fetchMock, { onDisconnected: disconnected }, async (session) => { + await readMessages(session, 1); + + await vi.waitFor(() => expect(session.messages.closed).toBe(true)); + expect(disconnected).toHaveBeenCalled(); + }); + } finally { + vi.restoreAllMocks(); + } +}; + +/** + * Scenario: server closes the stream, the message iterator closes, then the app manually reconnects. + * Verifies a second fetch opens and events flow again (non-persistent manual reconnect path). + */ +const testSseSessionReconnectsAfterTerminalClose = async (): Promise => { + const fetchMock = vi.fn(); + + // First fetch closes after one event, second fetch stays open. + fetchMock + .mockResolvedValueOnce(sseFetchResponse([ priceOracleEvents[0]!.raw ], { closeWhenDone: true })) + .mockResolvedValueOnce(sseFetchResponse([ priceOracleEvents[1]!.raw ])); + + try { + await withSession(fetchMock, {}, async (session) => { + // Read the first streams message + const [ event0 ] = await readMessages(session, 1); + expect(event0).toEqual(priceOracleEvents[0]!.parsed![0]); + + // Wait for the stream to close + await vi.waitFor(() => expect(session.messages.closed).toBe(true)); + + // App-initiated reconnect after the first stream ended. + await session.connect(); + + // Read the next message from the stream + // It should be different because we should be using a different fetch call instance) + const [ event1 ] = await readMessages(session, 1); + expect(event1).toEqual(priceOracleEvents[1]!.parsed![0]); + + // Assert that we have made two fetch calls + expect(fetchMock).toHaveBeenCalledTimes(2); + }); + } finally { + vi.restoreAllMocks(); + } +}; + +/** + * Scenario: persistent client — server closes the stream but the session should auto-reconnect. + * Verifies two fetch calls and events from both streams arrive on the same message iterator. + */ +const testSseSessionPersistentReconnectsOnServerClose = async (): Promise => { + const fetchMock = vi.fn(); + + // First fetch closes after one event, second fetch stays open. + fetchMock + .mockResolvedValueOnce(sseFetchResponse([ priceOracleEvents[0]!.raw ], { closeWhenDone: true })) + .mockResolvedValueOnce(sseFetchResponse([ priceOracleEvents[1]!.raw ])); + + try { + await withSession( + fetchMock, + { + persistent: true, + retry: testRetry(2), + }, + async (session) => { + const [ first, second ] = await readMessages(session, 2); + + // Assert that we have received the first event + expect(first).toEqual(priceOracleEvents[0]!.parsed![0]); + + // Assert that we have received the second event + expect(second).toEqual(priceOracleEvents[1]!.parsed![0]); + + // Assert that we have made two fetch calls + await vi.waitFor(() => expect(fetchMock).toHaveBeenCalledTimes(2)); + }, + ); + } finally { + vi.restoreAllMocks(); + } +}; + +/** + * Scenario: mid-stream network error with attemptReconnect enabled. + * Verifies the session fetches again and the iterator continues delivering events. + */ +const testSseSessionReconnectsOnTransportError = async (): Promise => { + const fetchMock = vi.fn(); + + fetchMock + // First connection: delivers one event then the stream errors (failingStreamAfter). + .mockResolvedValueOnce(new Response(failingStreamAfter(priceOracleEvents[0]!.raw), { status: 200, headers: SSE_HEADERS })) + // Second connection: clean stream with the next fixture event. + .mockResolvedValueOnce(sseFetchResponse([ priceOracleEvents[1]!.raw ])); + + try { + await withSession( + fetchMock, + { + attemptReconnect: true, + retry: testRetry(2), + }, + async (session) => { + const [ beforeError, recovered ] = await readMessages(session, 2); + + expect(beforeError).toEqual(priceOracleEvents[0]!.parsed![0]); + expect(recovered).toEqual(priceOracleEvents[1]!.parsed![0]); + await vi.waitFor(() => expect(fetchMock).toHaveBeenCalledTimes(2)); + }, + ); + } finally { + vi.restoreAllMocks(); + } +}; + +/** + * Scenario: mid-stream network error with attemptReconnect disabled (default). + * Verifies the session terminates: messages close, onError runs, and an error event is emitted. + */ +const testSseSessionClosesOnTransportErrorWhenNotReconnecting = async (): Promise => { + const onError = vi.fn(); + const response = new Response(failingStreamAfter(priceOracleEvents[0]!.raw), { status: 200, headers: SSE_HEADERS }); + const fetchMock = createFetchMock(() => response); + + try { + await withSession(fetchMock, { onError }, async (session) => { + const errorEvent = new Promise((resolve) => session.once('error', resolve)); + + await vi.waitFor(() => expect(session.messages.closed).toBe(true)); + expect(onError).toHaveBeenCalled(); + await expect(errorEvent).resolves.toBeInstanceOf(Error); + }); + } finally { + vi.restoreAllMocks(); + } +}; + +/** + * Scenario: the user aborts an active connection (e.g. navigation away) but may reconnect later. + * Verifies onDisconnected fires, fetch is aborted, but the message iterator stays open. + */ +const testSseSessionDisconnectEmitsDisconnectedAndKeepsMessagesOpen = async (): Promise => { + const onDisconnected = vi.fn(); + const fetchMock = createFetchMock(() => sseFetchResponse([ storageEvents[0]!.raw ])); + + try { + await withSession(fetchMock, { onDisconnected }, async (session) => { + await readMessages(session, 1); + session.disconnect(); + + expect(onDisconnected).toHaveBeenCalled(); + expect(session.messages.closed).toBe(false); + }); + } finally { + vi.restoreAllMocks(); + } +}; + +/** + * Scenario: the application fully tears down the session (logout, component unmount, etc.). + * Verifies close closes the message iterator and emits the closed event. + */ +const testSseSessionCloseClosesMessagesAndEmitsClosed = async (): Promise => { + const fetchMock = createFetchMock(() => sseFetchResponse([ storageEvents[0]!.raw ])); + + const session = await createSession(fetchMock, {}); + const closed = new Promise((resolve) => session.once('closed', () => resolve())); + + try { + session.close(); + + expect(session.messages.closed).toBe(true); + await expect(closed).resolves.toBeUndefined(); + } finally { + vi.restoreAllMocks(); + } +}; + +/** + * Scenario: the server returns a non-2xx HTTP status (500). + * Verifies create rejects, onError is invoked, and the error is a real Error instance. + */ +const testSseSessionHttpErrorCallsOnErrorAndThrows = async (): Promise => { + const onError = vi.fn(); + const fetchMock = createFetchMock(() => + new Response('nope', { + status: 500, + statusText: 'Internal Server Error', + })); + + const expectedError = new ExponentialBackoffMaxRetriesHitError([ new HTTPError(500, 'nope') ]); + + try { + await expect(createSession(fetchMock, { onError })).rejects.toThrow(expectedError); + + expect(onError).toHaveBeenCalled(); + + // Grab the parameters of the first call to onError + const firstCall = onError.mock.calls[0]; + + // Grab the First Parameter of the first call to onError + // - `onError(error)` - we are grabbing the error object. + const firstCallParams = firstCall?.[0]; + + // Assert the error is an instance of Error that matches the expected error. + expect(firstCallParams).toBeInstanceOf(Error); + } finally { + vi.restoreAllMocks(); + } +}; + +/** + * Scenario: The connectionController changes between the reader being created and #readStream being called. + * Verifies the session closes the reader and doesn't continue reading the stream. + */ +const testSseSessionClosesReaderIfConnectionChanges = async (): Promise => { + const fetchMock = createFetchMock(() => sseFetchResponse([ storageEvents[0]!.raw ])); + const onConnected = vi.fn(); + + const abortController = new AbortController(); + const onRequest = async (requestInit: RequestInit): Promise => { + // Swap out the signal (so it doesn't close the connection) + requestInit.signal = abortController.signal; + + return requestInit; + }; + + const session = new SSESession('http://test.com', { fetch: fetchMock, onRequest, onConnected, persistent: true, retry: testRetry(2) }); + + // Provide a mock retry which will swap out the connection controller after the session connects + const retry = { + run: vi.fn().mockImplementation(async (fn) => { + const result = await fn(); + + // "Disconnect" the session + // This won't disconnect the fetch since we replaced the signal with a different one + session.disconnect(); + + return result; + }), + }; + + // Replace the retry with our mock retry + session.options.retry = retry; + + // Connect the session + await session.connect(); + + // Expect the session to not have connected (it should have hit the exceptional behavior block) + expect(onConnected).not.toHaveBeenCalled(); +}; + +/** + * Scenario: The #readStream throws an unhandled error + * Verifies the error event is emitted and the session is ready for a new connection. + */ +const testSseSessionEmitsErrorEventOnReadStreamError = async (): Promise => { + // The mock fetch will return a "reader" that throws from .read() so that we can hit the catch block in #readStream. + // We will also mock the `normalizeError` method to throw, this should cascade up to the .catch handler in `connect` + const fetchMock = createFetchMock(() => { + return { + body: { + getReader: vi.fn().mockReturnValue({ + read: vi.fn().mockRejectedValue(new Error('Network failure')), + }), + }, + status: 200, + statusText: 'OK', + headers: new Headers(), + ok: true, + } as unknown as Response; + }); + + // Mock the normalizeError method to throw, this should cascade up to the .catch handler in `connect` + const normalizeErrorSpy = vi.spyOn(utils, 'normalizeError').mockImplementation(() => { + throw new Error('Normalize Error failed'); + }); + + // Create listeners for the error and disconnected events + const onError = vi.fn(); + const onDisconnected = vi.fn(); + + // Create a session + const session = new SSESession('http://test.com', { fetch: fetchMock, onError, onDisconnected, persistent: true, retry: testRetry(2) }); + + // Connect the session + await session.connect(); + + // Wait for the normalizeError method to be called (since #readStream is run as "fire and forget", connect resolved before our error is thrown) + await vi.waitFor(() => expect(normalizeErrorSpy).toHaveBeenCalled()); + + // Expect the error event to be emitted with the error from the normalizeError method + expect(onError).toHaveBeenCalledWith(new Error('Normalize Error failed')); + + // OnDisconnected should not have been called because the error should have thrown earlier than this emitter + expect(onDisconnected).not.toHaveBeenCalled(); +}; + +/** + * Scenario: fetch returns 200 but with a null body (misconfigured proxy or server bug). + * Verifies create rejects because SSE requires a readable stream body. + */ +const testSseSessionRejectsWhenResponseBodyIsNull = async (): Promise => { + const onError = vi.fn(); + const fetchMock = createFetchMock(() => new Response(null, { status: 200, headers: SSE_HEADERS })); + + const expectedError = new ExponentialBackoffMaxRetriesHitError([ new ResponseBodyNullError() ]); + + try { + await expect(createSession(fetchMock, { onError, retry: testRetry() })).rejects.toThrow(expectedError); + + expect(onError).toHaveBeenCalled(); + } finally { + vi.restoreAllMocks(); + } +}; + +/** + * Scenario: SSE spec resume — after receiving an event with an id, reconnect should send Last-Event-ID. + * Verifies the second fetch includes the id from the first event (1234 in storageEvents[0]). + */ +const testSseSessionSendsLastEventIdOnReconnect = async (): Promise => { + const { raw } = storageEvents[0]!; + const fetchMock = vi.fn(); + + fetchMock + // First connection: heartbeat chunk, then the event (with id: 1234), then server closes. + .mockResolvedValueOnce(sseFetchResponse([ ':\n\n', raw ], { closeWhenDone: true })) + // Second connection: just connects, we don't really care + .mockResolvedValueOnce(sseFetchResponse([])); + + const session = new SSESession('http://test.com', { fetch: fetchMock, persistent: true, retry: testRetry(2) }); + + try { + // Registers an onRequest hook that copies the last seen event id into reconnect headers. + SSESession.addLastEventIdReconnect(session); + + // Connect the session + await session.connect(); + + // Wait for the first message to be read, then we will disable the listener + await readMessages(session, 1); + + // Wait for us to fetch 2 times + await vi.waitFor(() => expect(fetchMock).toHaveBeenCalledTimes(2)); + + // Expect that the second fetch (so after our first message) includes the Last-Event-ID header with the id of the first message + expect(fetchMock).toHaveBeenNthCalledWith( + 2, + expect.anything(), + expect.objectContaining({ + headers: expect.objectContaining({ + 'Last-Event-ID': '1234', + }), + }), + ); + } finally { + session.close(); + vi.restoreAllMocks(); + } +}; + +/** + * Scenario: Attempt to add 2 `lastEventIdReconnect` plugins to the same session. + * Verifies the second plugin is not added, but no error is thrown + */ +const testSseSessionIdempotentlyHandlesAddingTheSamePluginTwice = async (): Promise => { + // Create a new session + const session = new SSESession('http://test.com', { persistent: true, retry: testRetry(2) }); + + // Add the listener the first time + SSESession.addLastEventIdReconnect(session); + + // Expect it to be added and that there is only a single plugin installed + expect(session.plugins.has(SSESession.addLastEventIdReconnect)).toBe(true); + expect(session.plugins.size).toBe(1); + + // Add the listener the second time + SSESession.addLastEventIdReconnect(session); + + // Expect it to be added and that there is only a single plugin installed + expect(session.plugins.has(SSESession.addLastEventIdReconnect)).toBe(true); + expect(session.plugins.size).toBe(1); +}; + +/** + * Scenario: A non-function was provided as the unsubscribe function for a plugin. + * Verifies an error is thrown and the plugin is not added. + */ +const testSseSessionThrowsWhenProvidingANonFunctionAsTheUnsubscribeFunction = async (): Promise => { + // Create a new session + const session = new SSESession('http://test.com', { persistent: true, retry: testRetry(2) }); + + // Set the plugin to a non-function + session.plugins.set(SSESession.addLastEventIdReconnect, 'not a function'); + + // Add the listener with a non-function + expect(() => SSESession.addLastEventIdReconnect(session)).toThrow(SSESessionUnsubscribePluginNotAFunctionError); +}; + +/** + * Scenario: The lastEventIdReconnect plugin can be removed when the unsubscribe method is called. + * Verifies the plugin is removed and the session no longer sends the Last-Event-ID header on reconnect. + */ +const testSseSessionLastEventIdReconnectPluginCanBeRemoved = async (): Promise => { + const events = [{ raw: 'id: 1\ndata: 1\n\n' }, { raw: 'id: 2\ndata: 2\n\n' }, { raw: 'data: 3\n\n' }]; + const fetchMock = vi.fn(); + + // Create a mock fetch that will be used to create the session + fetchMock + // First connection: Send event with id: 1 + .mockResolvedValueOnce(sseFetchResponse([ events[0]!.raw ], { closeWhenDone: true })) + // Second connection: Send event with id: 2 + .mockResolvedValueOnce(sseFetchResponse([ events[1]!.raw ], { closeWhenDone: true })) + // Third connection: Send event with id: 3, we don't expect to see the header with id 2 here + .mockResolvedValueOnce(sseFetchResponse([ events[2]!.raw ], { closeWhenDone: false })); + + // Create a new session + const session = new SSESession('http://test.com', { fetch: fetchMock, persistent: true, retry: testRetry(2) }); + + // Add the plugin to the session + const { removeListener } = SSESession.addLastEventIdReconnect(session); + + // Expect the plugin to be added + expect(session.plugins.has(SSESession.addLastEventIdReconnect)).toBe(true); + + // Connect the session + await session.connect(); + + // Wait for the first message to be read, then we will disable the listener + await readMessages(session, 1); + + // Call the unsubscribe method + removeListener?.(); + + // Expect the plugin to be removed + expect(session.plugins.has(SSESession.addLastEventIdReconnect)).toBe(false); + + // Wait for the rest of the messages to be read + await readMessages(session, 2); + + // Wait for us to fetch 3 times + await vi.waitFor(() => expect(fetchMock).toHaveBeenCalledTimes(3)); + + // Expect that the second fetch (so after our first message) includes the Last-Event-ID header with the id of the first message + expect(fetchMock).toHaveBeenNthCalledWith( + 2, + expect.anything(), + expect.objectContaining({ + headers: expect.objectContaining({ + 'Last-Event-ID': '1', + }), + }), + ); + + // Our second reconnect, the listener was disabled, so we don't expect to see the Last-Event-ID header with the id of the second message + expect(fetchMock).toHaveBeenNthCalledWith( + 3, + expect.anything(), + expect.objectContaining({ + headers: expect.not.objectContaining({ + 'Last-Event-ID': expect.any(String), + }), + }), + ); +}; + +/** + * Scenario: The browser visibility change handler is supported + * Verifies the session is connected and disconnected when the document visibility state changes. + */ +const testSseSessionBrowserVisibilityChangeIsSupported = async (): Promise => { + // Create a fetch mock + const response = new Response(failingStreamAfter(priceOracleEvents[0]!.raw), { status: 200, headers: SSE_HEADERS }); + const fetchMock = createFetchMock(() => response); + + // Create mock callbacks for connection to test connection state changes + const onConnected = vi.fn(); + const onDisconnected = vi.fn(); + + // Create, and connect, a session + const session = await createSession(fetchMock, { persistent: true, retry: testRetry(2), onConnected, onDisconnected }); + + // Expect the session to be connected + await vi.waitFor(() => expect(onConnected).toHaveBeenCalled()); + expect(session.active).toBe(true); + + // Add browser visibility change listener + SSESession.addBrowserVisibilityHandler(session); + + // Expect the session to be connected + expect(session.active).toBe(true); + + // Set the document visibility state to hidden and dispatch a visibilitychange event + documentStub.visibilityState = 'hidden'; + documentStub.dispatchEvent(new Event('visibilitychange')); + + // Expect the session to be disconnected + expect(session.active).toBe(false); + await vi.waitFor(() => expect(onDisconnected).toHaveBeenCalledWith({ reason: 'disconnect', source: SSESession.addBrowserVisibilityHandler })); + + // Wait for the session to be disconnected + documentStub.visibilityState = 'visible'; + documentStub.dispatchEvent(new Event('visibilitychange')); + + // Expect the session to be connected again + expect(session.active).toBe(true); +}; + +/** + * Scenario: The browser visibility change handler is only added once + * Verifies the handler is only added once and that the session is connected and disconnected when the document visibility state changes. + */ +const testSseSessionBrowserVisibilityOnlyAddsSingleHandler = async (): Promise => { + // Create a fetch mock + const response = new Response(failingStreamAfter(priceOracleEvents[0]!.raw), { status: 200, headers: SSE_HEADERS }); + const fetchMock = createFetchMock(() => response); + + // Create a spy on the document.addEventListener method + const addEventListenerSpy = vi.spyOn(documentStub, 'addEventListener'); + + // Create, and connect, a session + const session = await createSession(fetchMock, { persistent: true, retry: testRetry(2) }); + + // Expect the session to be connected + expect(session.active).toBe(true); + + // Add browser visibility change listener + SSESession.addBrowserVisibilityHandler(session); + + // Expect the plugin to be added and that there is only a single plugin installed + expect(session.plugins.size).toBe(1); + expect(addEventListenerSpy).toHaveBeenCalledWith('visibilitychange', expect.any(Function)); + expect(addEventListenerSpy).toHaveBeenCalledTimes(1); + + // Add the listener the second time + SSESession.addBrowserVisibilityHandler(session); + + // Expect the plugin to be added and that there is only a single plugin installed + expect(session.plugins.size).toBe(1); + expect(addEventListenerSpy).toHaveBeenCalledTimes(1); +}; + +/** + * Scenario: The browser visibility change handler throws an error if the existing plugin is not a function + * Verifies the handler is not added and an error is thrown. + */ +const testSseSessionBrowserVisibilityThrowsErrorIfExistingPluginIsNotAFunction = async (): Promise => { + // Create a fetch mock + const response = new Response(failingStreamAfter(priceOracleEvents[0]!.raw), { status: 200, headers: SSE_HEADERS }); + const fetchMock = createFetchMock(() => response); + + // Spy on the document.addEventListener method + const addEventListenerSpy = vi.spyOn(documentStub, 'addEventListener'); + + // Create a new session + const session = await createSession(fetchMock, { persistent: true, retry: testRetry(2) }); + + // Set the plugin to a non-function + session.plugins.set(SSESession.addBrowserVisibilityHandler, 'not a function'); + + // Expect an error to be thrown + expect(() => SSESession.addBrowserVisibilityHandler(session)).toThrow(SSESessionUnsubscribePluginNotAFunctionError); + + // Expect the document.addEventListener method to not have been called + expect(addEventListenerSpy).toHaveBeenCalledTimes(0); +}; + +/** + * Scenario: The browser visibility change handler is removed if the session is disconnected by `disconnect`. + * Verifies the handler is removed and the session is no longer connected when the document visibility state changes. + */ +const testSseSessionBrowserVisibilityRemovesVisibleHandlerIfDisconnectIsCalled = async (): Promise => { + // Create a fetch mock + const response = new Response(failingStreamAfter(priceOracleEvents[0]!.raw), { status: 200, headers: SSE_HEADERS }); + const fetchMock = createFetchMock(() => response); + + // Create a mock callbacks for connection to test connection state changes + const onConnected = vi.fn(); + const onDisconnected = vi.fn(); + const onBeforeDisconnect = vi.fn(); + + // Create a spy on the document.removeEventListener method + const removeEventListenerSpy = vi.spyOn(documentStub, 'removeEventListener'); + + // Create, and connect, a session + const session = await createSession(fetchMock, { persistent: true, retry: testRetry(2), onConnected, onDisconnected }); + session.on('beforeDisconnect', onBeforeDisconnect); + + // Expect the session to be connected + await vi.waitFor(() => expect(onConnected).toHaveBeenCalled()); + expect(session.active).toBe(true); + + // Add browser visibility change listener + SSESession.addBrowserVisibilityHandler(session); + + // Expect the session to be connected + expect(session.active).toBe(true); + + // Set the document visibility state to hidden and dispatch a visibilitychange event + documentStub.visibilityState = 'hidden'; + documentStub.dispatchEvent(new Event('visibilitychange')); + + expect(session.active).toBe(false); + await vi.waitFor(() => expect(onBeforeDisconnect).toHaveBeenCalledWith({ source: SSESession.addBrowserVisibilityHandler })); + await vi.waitFor(() => expect(onDisconnected).toHaveBeenCalledWith({ reason: 'disconnect', source: SSESession.addBrowserVisibilityHandler })); + expect(removeEventListenerSpy).toHaveBeenCalledTimes(0); + + // Disconnect the session + session.disconnect({ source: 'force-disconnect' }); + + // Should still be disconnected + expect(session.active).toBe(false); + await vi.waitFor(() => expect(onBeforeDisconnect).toHaveBeenLastCalledWith({ source: 'force-disconnect' })); + expect(removeEventListenerSpy).toHaveBeenCalledWith('visibilitychange', expect.any(Function)); + expect(removeEventListenerSpy).toHaveBeenCalledTimes(1); + + // Set the document visibility state to visible and dispatch a visibilitychange event + documentStub.visibilityState = 'visible'; + documentStub.dispatchEvent(new Event('visibilitychange')); + + // Should still be disconnected + expect(session.active).toBe(false); +}; + +/** + * Scenario: The browser visibility change handler is removed if the `removeListener` method is called. + * Verifies the handler is removed and the session is no longer connected when the document visibility state changes. + */ +const testSseSessionBrowserVisibilityRemovesVisibleHandlerWhenUnsubscribed = async (): Promise => { + // Create a fetch mock + const response = new Response(failingStreamAfter(priceOracleEvents[0]!.raw), { status: 200, headers: SSE_HEADERS }); + const fetchMock = createFetchMock(() => response); + + const onConnected = vi.fn(); + const onDisconnected = vi.fn(); + + // Create a new session + const session = await createSession(fetchMock, { persistent: true, retry: testRetry(2), onConnected, onDisconnected }); + + expect(onConnected).toHaveBeenCalled(); + expect(session.active).toBe(true); + + // Add browser visibility change listener + const { removeListener } = SSESession.addBrowserVisibilityHandler(session); + + // Expect the plugin to be added and that there is only a single plugin installed + expect(session.plugins.size).toBe(1); + + // Call the unsubscribe method + removeListener?.(); + + // Expect the plugin to be removed + expect(session.plugins.size).toBe(0); + + // Test the disconnect by hiding the document and dispatching a visibilitychange event + documentStub.visibilityState = 'hidden'; + documentStub.dispatchEvent(new Event('visibilitychange')); + + // Expect the session to be disconnected + expect(session.active).toBe(true); + expect(onDisconnected).not.toHaveBeenCalled(); +}; + +/** + * Scenario: The browser visibility change handler is not removed if the method to remove the handler is not the same + * Verifies the handler is not removed and the session is still connected when the document visibility state changes. + */ +const testSseSessionBrowserVisibilityDoesNotRemoveHandlerIfPluginIsNotTheSame = async (): Promise => { + // Create a fetch mock + const response = new Response(failingStreamAfter(priceOracleEvents[0]!.raw), { status: 200, headers: SSE_HEADERS }); + const fetchMock = createFetchMock(() => response); + + const onConnected = vi.fn(); + const onDisconnected = vi.fn(); + + // Create a new session + const session = await createSession(fetchMock, { persistent: true, retry: testRetry(2), onConnected, onDisconnected }); + + // Assert that the session connected + expect(onConnected).toHaveBeenCalled(); + expect(session.active).toBe(true); + + // Add browser visibility change listener + const { removeListener } = SSESession.addBrowserVisibilityHandler(session); + + // Overwrite the plugin with a different function + session.plugins.set(SSESession.addBrowserVisibilityHandler, () => 1); + + // Call the unsubscribe method + removeListener?.(); + + // Expect the plugin to be removed + expect(session.plugins.size).toBe(1); + + // Test the disconnect by hiding the document and dispatching a visibilitychange event + documentStub.visibilityState = 'hidden'; + documentStub.dispatchEvent(new Event('visibilitychange')); + + // Expect the session to be disconnected + expect(session.active).toBe(true); + await vi.waitFor(() => expect(onDisconnected).not.toHaveBeenCalled()); +}; + +const runTests = async (): Promise => { + afterEach(() => { + vi.restoreAllMocks(); + vi.resetAllMocks(); + + documentStub.visibilityState = 'visible'; + }); + + test('SSESession.connect: calls injected fetch with method, headers, and abort signal', testSseSessionConnectCallsFetchWithExpectedOptions); + test('SSESession.connect: uses default fetch implementation if no fetch is provided', testSseSessionUsesDefaultFetchImplementation); + test('SSESession.connect: invokes onConnected when the stream is established', testSseSessionConnectInvokesOnConnected); + test('SSESession.connect: does not open a second transport when connect is called again', testSseSessionConnectDoesNotOpenSecondTransport); + test('SSESession.connect: passes request mutations from onRequest to fetch', testSseSessionConnectPassesOnRequestMutations); + test('SSESession.connect: sends POST bodies for POST-based SSE endpoints', testSseSessionConnectSendsPostBody); + test('SSESession: delivers multiple fixture events through a single session', testSseSessionDeliversMultipleFixtureEvents); + test('SSESession: emits message events for incoming SSE frames', testSseSessionEmitsMessageEvents); + test('SSESession: parses a fixture event split across chunk boundaries', testSseSessionParsesEventSplitAcrossChunks); + test('SSESession: closes messages and emits disconnected when persistent is false', testSseSessionClosesOnServerCloseWhenNotPersistent); + test('SSESession: opens a new message stream after reconnecting following a terminal close', testSseSessionReconnectsAfterTerminalClose); + test('SSESession: reconnects when the server closes the stream and persistent is true', testSseSessionPersistentReconnectsOnServerClose); + test('SSESession: reconnects when attemptReconnect is true', testSseSessionReconnectsOnTransportError); + test('SSESession: closes messages and emits error when attemptReconnect is false', testSseSessionClosesOnTransportErrorWhenNotReconnecting); + test( + 'SSESession.disconnect: disconnects fetch, emits disconnected, and keeps messages open', + testSseSessionDisconnectEmitsDisconnectedAndKeepsMessagesOpen, + ); + test('SSESession.close: closes messages and emits closed', testSseSessionCloseClosesMessagesAndEmitsClosed); + test('SSESession: calls onError, emits error, and throws from create on HTTP error', testSseSessionHttpErrorCallsOnErrorAndThrows); + test('SSESession: closes reader if connection changes', testSseSessionClosesReaderIfConnectionChanges); + test('SSESession: emits error event on read stream error', testSseSessionEmitsErrorEventOnReadStreamError); + test('SSESession: rejects when the response body is null', testSseSessionRejectsWhenResponseBodyIsNull); + test( + 'SSESession: lastEventIdReconnect sends Last-Event-ID on reconnect after receiving an event with an id', + testSseSessionSendsLastEventIdOnReconnect, + ); + test( + 'SSESession: lastEventIdReconnect idempotently handles adding the same plugin twice', + testSseSessionIdempotentlyHandlesAddingTheSamePluginTwice, + ); + test( + 'SSESession: lastEventIdReconnect throws when providing a non-function as the unsubscribe function', + testSseSessionThrowsWhenProvidingANonFunctionAsTheUnsubscribeFunction, + ); + test('SSESession: lastEventIdReconnect plugin can be removed', testSseSessionLastEventIdReconnectPluginCanBeRemoved); + test('SSESession: browser visibility change is supported', testSseSessionBrowserVisibilityChangeIsSupported); + test('SSESession: browser visibility only adds a single handler', testSseSessionBrowserVisibilityOnlyAddsSingleHandler); + test( + 'SSESession: browser visibility throws error if existing plugin is not a function', + testSseSessionBrowserVisibilityThrowsErrorIfExistingPluginIsNotAFunction, + ); + test( + 'SSESession: browser visibility removes visible handler if disconnect is called', + testSseSessionBrowserVisibilityRemovesVisibleHandlerIfDisconnectIsCalled, + ); + test( + 'SSESession: browser visibility removes visible handler when unsubscribed', + testSseSessionBrowserVisibilityRemovesVisibleHandlerWhenUnsubscribed, + ); + test( + 'SSESession: browser visibility does not remove handler if plugin is not the same', + testSseSessionBrowserVisibilityDoesNotRemoveHandlerIfPluginIsNotTheSame, + ); +}; + +await runTests();