sse-session/README.md

# sse-session

A fetch-based Server-Sent Events client for modern browsers and Node.js runtimes that provide the standard Fetch and Streams APIs.

`sse-session` supports GET and POST streams, custom request hooks, connection retries, automatic reconnects, async iteration, lifecycle events, optional Page Visibility handling, and opt-in `Last-Event-ID` headers.

## Installation

```sh
npm install sse-session
```

The package is authored as an ES module.

```ts
import {
  SSESession,
  SSEEventParser,
  ExponentialBackoff,
  type SSEvent,
  type SSESessionOptions,
} from "sse-session";
```

The runtime must provide `fetch`, `Response`, `ReadableStream`, `AbortController`, `FormData`, and `TextDecoder`. A custom fetch-compatible function can be supplied through the `fetch` option.

## Quick start

`SSESession.create()` constructs a session and waits until its HTTP response body is available. Stream reading then continues in the background.

```ts
import { SSESession } from "sse-session";

const session = await SSESession.create("https://api.example.com/events");

try {
  for await (const event of session.messages) {
    console.log(event.event, event.data);
  }
} finally {
  await session.disconnect();
}
```

Messages are also emitted as events. Construct the session directly when listeners must be installed before the connection starts:

```ts
const session = new SSESession("https://api.example.com/events");

session.on("message", (event) => {
  console.log(event.data);
});

await session.connect();
```

## Session API

### Creating and connecting

```ts
const session = new SSESession(url, options);
await session.connect();
```

or, as a convenience:

```ts
const session = await SSESession.create(url, options);
```

`connect()` is idempotent while a transport is active. It resolves after the fetch succeeds, the response has a non-null body, `onConnected` runs, and the `"connected"` event is emitted. It does not wait for the stream to finish.

Non-2xx responses and responses without a body are treated as connection failures. The response `Content-Type` is not validated.

### Receiving messages

Each parsed event is delivered through both of these interfaces:

```ts
const unsubscribe = session.on("message", (event) => {
  handle(event);
});

for await (const event of session.messages) {
  handle(event);
}

unsubscribe();
```

`messages` is an `AsyncPushIterator<SSEvent>`. It buffers values until they are read, and breaking out of a loop does not cancel the underlying queue. Only one async iterator may consume a `messages` instance at a time; use `"message"` listeners when multiple concurrent consumers are needed.

Do not cache `session.messages` across a terminal close. A later `connect()` replaces a closed iterator with a new one.

### Stopping a session

`abort()` stops the active transport without ending the session:

- the active fetch is aborted;
- partial parser state is reset;
- `onDisconnected` and `"disconnected"` run;
- `messages` stays open for a later reconnect;
- `"closed"` and `"error"` are not emitted.

Calling `abort()` with no active transport is a no-op.

`disconnect()` is the terminal operation for the current session lifecycle:

- `messages` is closed;
- `"closed"` is emitted;
- an active transport is then aborted, which also produces `onDisconnected` and `"disconnected"`.

The same `SSESession` object can be connected again. When that happens, it creates a fresh `messages` iterator.

## Lifecycle and reconnect behavior

There are two separate reconnect controls:

| Situation                                  | Option             | Default | Behavior                                                              |
| ------------------------------------------ | ------------------ | ------- | --------------------------------------------------------------------- |
| The server ends the response body normally | `persistent`       | `false` | Reconnect when `true`; otherwise close `messages`.                    |
| Reading the response body throws           | `attemptReconnect` | `true`  | Reconnect when `true`; otherwise emit the error and close `messages`. |

An intentional `abort()` does not trigger `attemptReconnect`. Automatic reconnects keep the existing `messages` iterator open.

Every call to `connect()`, including automatic reconnects, uses the configured retry policy while establishing the HTTP response. The session default is:

```ts
new ExponentialBackoff({
  baseDelay: 1_000,
  maxDelay: 10_000,
  maxAttempts: 0, // unlimited
  growthRate: 1.3,
  jitter: 0.3,
});
```

With the default unlimited policy, `create()` or `connect()` can remain pending indefinitely when the endpoint cannot be reached. Supply a finite policy when the caller needs connection failure to reject:

```ts
const session = await SSESession.create(url, {
  retry: ExponentialBackoff.from({
    maxAttempts: 5,
    baseDelay: 500,
  }),
});
```

When a finite policy is exhausted, `connect()` closes `messages`, invokes the disconnection and error notifications, and rejects with the retry policy's error (`AggregateError` for `ExponentialBackoff`).

## Events and callbacks

`SSESession` extends the exported typed `EventEmitter`.

| Event          | Payload   | Emitted when                                                                                                       |
| -------------- | --------- | ------------------------------------------------------------------------------------------------------------------ |
| `connected`    | `void`    | The response stream has been opened.                                                                               |
| `message`      | `SSEvent` | A complete event has been parsed.                                                                                  |
| `disconnected` | `void`    | An established transport ends or `abort()` is called. It is also emitted after a connection attempt finally fails. |
| `error`        | `Error`   | A request hook, fetch, response, stream read, or `onConnected` callback fails. Intentional aborts are excluded.    |
| `closed`       | `void`    | `disconnect()` is called.                                                                                          |

The emitter provides `on`, `once`, `off`, `emit`, `removeAllListeners`, and `waitFor`. `on` and `once` return an unsubscribe function and accept an optional debounce duration as their third argument.

The corresponding option callbacks are invoked before their lifecycle event:

| Callback             | Purpose                                                                                                |
| -------------------- | ------------------------------------------------------------------------------------------------------ |
| `onRequest(request)` | Returns the `RequestInit` used for each fetch. Use it for refreshed auth or other per-attempt changes. |
| `onConnected()`      | Runs after the response body is available and before `"connected"`.                                    |
| `onDisconnected()`   | Runs before `"disconnected"`.                                                                          |
| `onError(error)`     | Runs before `"error"`. Defaults to logging the error.                                                  |

An error thrown by `onConnected` is reported through `onError` and `"error"`, after which the session still emits `"connected"` and starts reading. An error thrown by `onDisconnected` is passed to `onError`. If `onError` itself throws, that callback error is logged.

## Options

The constructor and `create()` accept `Partial<SSESessionOptions>`.

| Option             | Type                                                    | Default                                                     |
| ------------------ | ------------------------------------------------------- | ----------------------------------------------------------- |
| `method`           | `"GET" \| "POST"`                                       | `"GET"`                                                     |
| `headers`          | `Record<string, string>`                                | `Accept: text/event-stream` and `Cache-Control: no-cache`   |
| `body`             | `string \| FormData`                                    | Empty `FormData`; sent only for POST requests               |
| `fetch`            | `(url: string, init: RequestInit) => Promise<Response>` | Global `fetch`                                              |
| `onRequest`        | `(init: RequestInit) => Promise<RequestInit>`           | Returns the input unchanged                                 |
| `onConnected`      | `() => void`                                            | No-op                                                       |
| `onDisconnected`   | `() => void`                                            | No-op                                                       |
| `onError`          | `(error: Error) => void`                                | Logs to `console.error`                                     |
| `retry`            | Object with a `run()` method                            | Unlimited session-specific `ExponentialBackoff` shown above |
| `attemptReconnect` | `boolean`                                               | `true`                                                      |
| `persistent`       | `boolean`                                               | `false`                                                     |
| `eventParser`      | Object with `parseEvents()` and `reset()`               | A new `SSEEventParser` per session                          |

Custom headers are merged over the two default headers. Each fetch also receives `cache: "no-store"`, the session's abort signal, and a `null` body for GET requests.

The resolved configuration is publicly available as `session.options`. It can be changed between connections, although passing options to the constructor is preferable when runtime mutation is not required.

Example POST stream with refreshed authentication:

```ts
const session = await SSESession.create("https://api.example.com/events", {
  method: "POST",
  headers: { "Content-Type": "application/json" },
  body: JSON.stringify({ topic: "prices" }),
  onRequest: async (request) => ({
    ...request,
    headers: {
      ...request.headers,
      Authorization: `Bearer ${await getAccessToken()}`,
    },
  }),
});
```

The custom fetch signature deliberately accepts any string URL, which allows adapters for non-standard transports:

```ts
const session = await SSESession.create("libp2p://peer/events", {
  fetch: (url, init) => libp2pFetch(url, init),
});
```

## Page Visibility integration

`addBrowserVisibilityHandler()` attaches a `document.visibilitychange` listener to an existing session:

```ts
const session = new SSESession("/events");
SSESession.addBrowserVisibilityHandler(session);
await session.connect();
```

When the document becomes hidden, the handler calls `abort()`. When it becomes visible, it calls `connect()`. `disconnect()` removes the listener; connecting the same session later reattaches it after the connection succeeds. In runtimes without `document`, this method is a no-op.

The helper does not inspect the document's visibility when it is attached. To avoid the initial connection while the page is already hidden, gate it explicitly:

```ts
const session = new SSESession("/events");
SSESession.addBrowserVisibilityHandler(session);

if (document.visibilityState !== "hidden") {
  await session.connect();
}
```

There is currently no `withBrowserVisibility()` factory.

## Last-Event-ID reconnects

Resume headers are opt-in:

```ts
const session = new SSESession("/events", { persistent: true });

await SSESession.addLastEventIdReconnect(session);
await session.connect();
```

The helper remembers the most recent non-`undefined` `id` received after it is attached and adds it as `Last-Event-ID` on subsequent fetches. The original `onRequest` hook runs afterward and can further modify the request. No header is sent until a non-empty ID has been observed.

Attach the helper before `connect()` if the first event must always be observed for later resume. The helper changes request headers only; it does not itself enable `persistent` or `attemptReconnect`.

## Parsed events

```ts
interface SSEvent {
  data: string;
  event?: string;
  id?: string;
  retry?: number;
}
```

`data` is required because frames without any `data` field are not emitted. Multiple `data` lines are joined with `\n`. The optional `event` field remains `undefined` when the server omits it; the parser does not fill in `"message"`.

The parser exposes a numeric `retry` field on the returned event, but `SSESession` does not apply it to its retry policy.

## SSEEventParser

`SSEEventParser` is the incremental parser used by the session and is also exported for direct use:

```ts
const parser = new SSEEventParser();

for (const chunk of chunks) {
  for (const event of parser.parseEvents(chunk)) {
    console.log(event);
  }
}

parser.reset();
```

It buffers incomplete input until a blank line completes a frame, preserves multi-byte UTF-8 characters split across chunks, ignores comments and unknown fields, removes one optional space after a field's colon, and recognizes `data`, `event`, `id`, and numeric `retry` fields. It accepts `\n`, `\r`, and `\r\n` line endings.

Current parser limitation: a `\r\n` pair split between two byte chunks is not handled correctly. Keep those two bytes in the same chunk, or use `\n`-delimited streams, until that edge case is fixed.

## Other exports

The package currently exports all source utilities as part of its public entry point:

- `AsyncPushIterator`
- `EventEmitter`
- `ExponentialBackoff`
- `tryAsync`
- the option, event-map, parser, retry, and backoff TypeScript types

These utilities are used internally by `SSESession` but can also be imported directly from `sse-session`.

## Current caveat for in-progress connection retries

`abort()` aborts the fetch signal, but it does not cancel the configured retry runner itself. If `connect()` is currently retrying failed fetches under the default unlimited policy, aborting or disconnecting the session does not guarantee that the pending `connect()` promise will settle. Use a finite retry policy when connection attempts need a bounded lifetime.