@aiquants/sse

Host-agnostic Server-Sent Events toolkit for TypeScript:

• a reopening EventSource client that survives non-200 responses, stops for good on caller-declared in-band terminal events, and resets its full-jitter backoff only after a healthy open;
• a React hook that keeps one connection per key and never reconnects because a handler changed identity;
• WHATWG Response helpers for the server (framing, streaming response with single-fire close reasons, terminal response, heartbeat, resume cursor) that run anywhere Response and ReadableStream exist;
• a ref-counted shared source that runs one upstream per process and fans it out to every subscriber;
• a multiplexer that carries many topics of one endpoint over one reopening connection and rebuilds the URL from every topic's latest cursor on each open;
• testing helpers (fake EventSource, manual clock, WHATWG parser, response reader).

Zero runtime dependencies. react is an optional peer used only by @aiquants/sse/react. No module-level state, no defaults: every option is passed explicitly.

Contents

• What it is and what it is not
• Install
• Quick start
• Wire contract
• Failure semantics
• API
• Recommended values (not defaults)
• Testing helpers
• Security notes
• Deployment
• Adopting in another project
• Out of scope
• License

What it is and what it is not

It is a small, strict implementation of one wire contract for text/event-stream, shared by the server and the client, plus the reconnection rules that make native EventSource safe to run for hours:

| Defect class | How the package closes it | | --- | --- | | A native EventSource that receives 401 / 429 / 502 / 503 becomes CLOSED and never reconnects | The client treats every non-200 as retryable, closes the dead source and reopens with its own full-jitter backoff | | An expired session makes the browser reconnect forever | The server sends a 200 response that carries retry: <large> and then a named terminal event; the client stops for good. Old clients without a listener sleep on the large retry: | | A stream that opens and closes at once resets the backoff on every open (a reconnect loop) | The failure count returns to 0 only after the stream stayed open for healthyOpenMs | | Connection: keep-alive on an HTTP/2 response is a protocol error | SSE_RESPONSE_HEADERS carries no Connection, and hop-by-hop headers passed by the caller are rejected with TypeError | | A React effect that depends on handler identity reconnects and loses its cursor | The hook reads handlers through a ref; only a connectionKey change reopens | | Every subscriber opens its own upstream | createSharedSource starts one upstream per process and serializes start and stop on one promise lane | | Every topic on a page (one panel per channel, for example) opens its own EventSource and uses up the browser's six HTTP/1.1 connections per origin; a hand-written hub stops on the first non-200, drops topics beyond the server's limit from the URL and lets a second subscriber replace the first | createMultiplexedEventSource carries every topic over one reopening connection, rebuilds the URL from each topic's cursor on every open, and throws on a duplicate topic or beyond maxTopics |

It is not an NDJSON or WebSocket library, not a cross-tab hub (no BroadcastChannel / SharedWorker), and it does not pause on visibilitychange or apply backpressure. Application-specific terminal event names, payload schemas and feeds stay in the application.

Install

pnpm add @aiquants/sse
# only when you use @aiquants/sse/react
pnpm add react

In a pnpm workspace, depend on it with "@aiquants/sse": "workspace:*".

Subpath exports:

| Import path | Contents | Environment | | --- | --- | --- | | @aiquants/sse | Wire contract (heartbeat codec, name / id rules, reserved names) and the SseRuntime port | Any | | @aiquants/sse/client | openReopeningEventSource, createBrowserEventSource, ready-state constants | Browser (or any EventSource implementation) | | @aiquants/sse/react | useReopeningEventSource | React >= 18 | | @aiquants/sse/server | Framing, createSseResponse, terminalStreamResponse, startSseHeartbeat, readLastEventId | Any runtime with WHATWG Response / ReadableStream / TextEncoder / Headers / AbortController / queueMicrotask | | @aiquants/sse/shared-source | createSharedSource | Any | | @aiquants/sse/multiplex | createMultiplexedEventSource | Browser (or any EventSource implementation) | | @aiquants/sse/testing | FakeEventSource, createManualRuntime, parseSseText, readSseResponse | Tests |

The build targets ES2020 and ships CJS, ESM and type declarations without source maps. The server half never imports node: modules.

Quick start

Declare the terminal events once in the application and import the type from both halves:

// sse-contract.ts (application code)
export type FeedTerminalEvent = "auth-required" | "forbidden"
export const FEED_TERMINAL_EVENTS = ["auth-required", "forbidden"] as const satisfies readonly FeedTerminalEvent[]
export const FEED_CURSOR_PARAM = "cursor"

Server route (any framework whose handler receives a WHATWG Request and returns a Response):

import { systemSseRuntime } from "@aiquants/sse"
import { createSseResponse, readLastEventId, startSseHeartbeat, terminalStreamResponse } from "@aiquants/sse/server"
import { FEED_CURSOR_PARAM, type FeedTerminalEvent } from "./sse-contract"

export const handleFeedStream = async (request: Request): Promise<Response> => {
    const session = await authenticate(request) // your authentication
    if (session === null) {
        // 200 + retry first + named terminal event: the client stops, old clients sleep
        return terminalStreamResponse("auth-required" satisfies FeedTerminalEvent, { data: "", retryMs: 86_400_000, headers: null })
    }
    const cursor = readLastEventId(request, FEED_CURSOR_PARAM)
    return createSseResponse({
        requestSignal: request.signal,
        headers: null,
        onError: (error, origin) => console.error(`feed stream ${origin} failed`, error),
        produce: (writer) => {
            startSseHeartbeat(writer, { intervalMs: 15_000, retry: { baseMs: 2_000, jitterMs: 8_000 }, runtime: systemSseRuntime })
            const unsubscribe = feed.subscribe(session.userId, cursor, (item) => {
                writer.writeEvent({ event: "item", id: item.cursor, data: JSON.stringify(item) })
            })
            writer.onClose(unsubscribe)
        },
    })
}

React client:

import { systemSseRuntime } from "@aiquants/sse"
import { createBrowserEventSource } from "@aiquants/sse/client"
import { useReopeningEventSource } from "@aiquants/sse/react"
import { useEffect, useState } from "react"
import { FEED_CURSOR_PARAM, FEED_TERMINAL_EVENTS } from "./sse-contract"

const FEED_ORIGIN = "https://example.com"

export const FeedPanel = ({ feedId }: { readonly feedId: string }) => {
    const [items, setItems] = useState<readonly string[]>([])
    const { status } = useReopeningEventSource(feedId, {
        resolveUrl: ({ lastEventId }) => {
            const url = new URL(`/sse/feed/${encodeURIComponent(feedId)}`, FEED_ORIGIN)
            if (lastEventId !== null) {
                url.searchParams.set(FEED_CURSOR_PARAM, lastEventId)
            }
            return url.toString()
        },
        initialLastEventId: null,
        withCredentials: true,
        createEventSource: createBrowserEventSource,
        runtime: systemSseRuntime,
        backoff: { initialDelayMs: 1_000, maxDelayMs: 30_000, multiplier: 2 },
        healthyOpenMs: 60_000,
        nativeRetryLimit: 1,
        giveUpAfterFailures: "never",
        terminalEvents: FEED_TERMINAL_EVENTS,
        events: { item: (event) => setItems((previous) => [...previous, event.data]) },
        heartbeat: { staleAfterMs: 45_000, onHeartbeat: null },
    })

    useEffect(() => {
        // side effects of a status belong here, not in the hook
        if (status.kind === "terminated" && status.terminalEvent === "auth-required") {
            window.location.reload()
        }
    }, [status])

    return <FeedList items={items} status={status.kind} />
}

Wire contract

| Element | Rule | | --- | --- | | Response headers | Exactly Content-Type: text/event-stream; charset=utf-8, Cache-Control: no-store, no-transform, X-Accel-Buffering: no, plus the caller's extra headers. A response created with compressible: true drops no-transform and changes nothing else. Never Connection, Keep-Alive, Proxy-Connection, Transfer-Encoding, Upgrade, TE or Content-Length | | Status | Always 200, also for the terminal response. A non-200 is reserved for failures the client should retry | | Data event | event: <name> (omitted for unnamed events, which arrive as message), optional id: <cursor>, then one data: line per line of the value. CR and CRLF inside data arrive as LF | | Heartbeat | Reserved event heartbeat with data: {"serverTime":<epoch ms>} and no id: line, so the cursor never moves. startSseHeartbeat writes one retry: base + jitter first | | Terminal event | A caller-declared name, written in a 200 body as retry: <large> before event: <name>, then the body ends. The client stops for good on the first one | | Cursor | Every non-empty id: becomes the client's lastEventId. The browser's own reconnect sends it as the Last-Event-ID header on the original URL; a reopen by the utility puts it in the URL through resolveUrl. The server reads the header first, then the query parameter (readLastEventId) | | Reserved names | open, error, message and heartbeat can be neither data nor terminal event names (message stays usable as an events key on the client, where it receives unnamed events) | | Names and ids | Names: non-empty, without CR, LF or NUL. Ids: non-empty printable ASCII (U+0020–U+007E) without leading or trailing spaces, because the browser's own reconnect sends the id back in the Last-Event-ID header, which a server reads as latin1 bytes with the surrounding whitespace trimmed. An empty id: would erase the browser's cursor, so it is rejected |

Example terminal body (terminalStreamResponse("auth-required", { data: "", retryMs: 86_400_000, headers: null })):

retry: 86400000

event: auth-required
data: 

Failure semantics

What the client sees for each server outcome, measured with undici 7.28 and with Chromium 148, Firefox 150 and WebKit 26.4 behind nginx over HTTP/2:

| Server outcome | readyState at error | Browser's own retry | What the utility does | | --- | --- | --- | --- | | Any non-200 (401, 429, 500, 502, 503, 204) or a wrong Content-Type | 2 (CLOSED) in every engine | Never | Closes the source, waits a full-jitter delay and reopens (waiting), or stops as exhausted | | 200, then the stream ends | 0 (CONNECTING) | After the last retry: value (default 3000 ms in Chromium and WebKit, 5000 ms in Firefox) | Leaves up to nativeRetryLimit consecutive failures to the browser (retrying, stale-watched), then closes and uses its own backoff | | The browser's own reconnect gets no response (its request landed on a dead HTTP/2 connection) | none (stays CONNECTING) | none | The stale watch, re-armed at the error that started retrying, closes the source after staleAfterMs, counts a failure and reopens with its own backoff (waiting) | | 200 + named terminal event + end, with a listener | open, then the event | Not reached | Adopts the event's id, closes, reports terminated | | 200 + named terminal event + end, without a listener (old bundle) | open, error(0), repeated | Every retry: interval | With retry: 86400000 written first: 1 request in a 30 s window in all three engines. Without it: about 1,200 requests per hour per tab (Chromium, WebKit) or 720 (Firefox) | | No event for staleAfterMs (half-open TCP, a silently dead HTTP/2 stream) | none | none | The stale watch closes the source and reopens like a CLOSED error |

Three HTTP/2 cases make the heartbeat watch mandatory: with WebKit's Linux network stack, a stream reset mid-body and a connection closed by an nginx reload (worker_shutdown_timeout) left the EventSource OPEN and silent with no error event, and a browser reconnect whose request landed on a dead connection stayed CONNECTING with neither error nor open. Closing and reopening it recovered at once.

Real Safari on iOS / macOS uses a different network stack, and iOS suspends Safari's connections in the background without always firing error, so run the stale watch everywhere (heartbeat: { staleAfterMs, onHeartbeat }).

API

Every function validates its input synchronously and throws TypeError (names, functions, headers) or RangeError (numbers, cursors) instead of substituting a value. No option has a default.

Timer limit. Browsers and Node store timer delays in a signed 32-bit integer and run anything larger after about 1 ms. Every delay option (backoff.maxDelayMs, healthyOpenMs, heartbeat.staleAfterMs, the heartbeat intervalMs, lingerMs, readSseResponse timeoutMs) is therefore capped at 2,147,483,647 ms and rejected above it, and systemSseRuntime.setTimer throws RangeError for NaN, negative, infinite or larger delays.

Wire retry: limit. Every wire retry: value (formatSseRetry, writer.writeRetry, the terminal retryMs, and the heartbeat's baseMs + jitterMs) has the same cap, because it drives the browser's own reconnect timer: Firefox reads it into an unsigned 32-bit integer without an overflow check, so 4,294,967,296 wraps into a 500 ms reconnect loop, and Chromium and WebKit ignore values of 2^64 or more.

@aiquants/sse

| Export | Kind | Contract | | --- | --- | --- | | SseRuntime | type | { now(), setTimer(handler, delayMs), clearTimer(handle), random() }. Every stateful API takes one; nothing reads global timers or randomness except through it | | systemSseRuntime | value | Reads globalThis.Date / setTimeout / clearTimeout / Math at call time (SSR-safe; fake timers installed later still apply). Frozen | | SSE_HEARTBEAT_EVENT | value | "heartbeat" | | SseHeartbeat | type | { readonly serverTime: number } | | formatSseHeartbeatData(serverTime) | function | {"serverTime":<ms>}; RangeError for a non-finite time | | parseSseHeartbeatData(data) | function | The decoded beat, or null for anything malformed (a rejection, not a substitute) | | isValidSseEventName(name) / isValidSseEventId(id) | function | Name: true for a non-empty string without CR, LF or NUL. Id: true for a non-empty string of printable ASCII (U+0020–U+007E) without leading or trailing spaces. false for any non-string value | | SSE_RESERVED_EVENT_NAMES / SseReservedEventName | value / type | ["open", "error", "message", "heartbeat"] (frozen) |

import { formatSseHeartbeatData, isValidSseEventId, parseSseHeartbeatData } from "@aiquants/sse"

formatSseHeartbeatData(1_700_000_000_000) // '{"serverTime":1700000000000}'
parseSseHeartbeatData('{"serverTime":1700000000000}') // { serverTime: 1700000000000 }
parseSseHeartbeatData('{"time":1}') // null
isValidSseEventId("") // false: an empty id would erase the browser's cursor

@aiquants/sse/client

openReopeningEventSource(options) opens one connection and returns a handle. All options are required:

| Option | Rule | | --- | --- | | resolveUrl({ lastEventId }) | Called on every open (first open and every reopen) with the latest cursor | | initialLastEventId | null, or an id isValidSseEventId accepts; anything else ("", CR / LF / NUL, non-ASCII, leading or trailing spaces, undefined) is RangeError | | withCredentials | A boolean, forwarded to every createEventSource call | | createEventSource(url, { withCredentials }) | One EventSource per open; createBrowserEventSource in a browser | | runtime | SseRuntime: an object whose now, setTimer, clearTimer and random are functions | | backoff | { initialDelayMs, maxDelayMs, multiplier }: all positive finite, maxDelayMs >= initialDelayMs, maxDelayMs <= 2,147,483,647, multiplier >= 1. Delay = floor(random × min(max, initial × multiplier^(failures − 1))) | | healthyOpenMs | Positive, at most 2,147,483,647. The failure count returns to 0 only after the stream stayed open this long. With the stale watch on, keep it at least staleAfterMs + the heartbeat interval (see Recommended values) | | nativeRetryLimit | Integer >= 0. How many consecutive failures in CONNECTING are left to the browser's own reconnect. 0 means never | | giveUpAfterFailures | Integer >= 1 and greater than nativeRetryLimit, or "never" | | terminalEvents | Non-empty tuple of valid, non-reserved, unique names that are not events keys (an empty tuple is a compile error and a TypeError) | | events | Record<eventType, (event: MessageEvent<string>) => void>; the message key receives unnamed events. open, error, heartbeat and invalid names are TypeError | | heartbeat | { staleAfterMs, onHeartbeat } (staleAfterMs positive, at most 2,147,483,647; onHeartbeat a function or null) or "ignore" (no heartbeat listener, no stale watch). The stale watch runs while a source the utility opened waits for open, while the browser's own reconnect is in progress (retrying) and while the stream is open | | onStatus(status) | Receives every status change; the first connecting synchronously |

resolveUrl, createEventSource, onStatus and every handler must be functions, withCredentials a boolean, runtime an object with the four functions and terminalEvents an array; anything else is TypeError. All of it is checked before any socket opens and before onStatus is called.

Statuses (ReopeningEventSourceStatus<TTerminal>); terminated, faulted and closed are final:

| Kind | Meaning | | --- | --- | | connecting | A source is being opened (consecutiveFailures) | | open | The stream is open | | retrying | The browser's own reconnect is in progress (within nativeRetryLimit); stale-watched from the error on | | waiting | The source is closed and a reopen is scheduled after delayMs | | exhausted | giveUpAfterFailures was reached; reconnectNow() can revive it | | terminated | A declared terminal event arrived (terminalEvent is narrowed to TTerminal, data is its payload) | | faulted | A reopen threw (resolveUrl, createEventSource or the runtime); error holds the exception | | closed | close() was called |

Handle (ReopeningEventSourceHandle): status(), lastEventId(), reconnectNow() (only from waiting / exhausted: resets the failure count and opens at once; returns false elsewhere), close() (idempotent).

Behaviour worth knowing:

• An error while the source is CLOSED or in an unknown state (every non-200) closes it before the reopen is scheduled, so at most one EventSource is ever open. A generation counter drops events still queued on a retired source.
• A stale watch expiry is handled like a CLOSED error. Data, terminal and heartbeat events re-arm it.
• The stale watch starts as soon as the utility opens a source, so a connect that never answers (for example a stream multiplexed onto a dead HTTP/2 connection) is closed and reopened too.
• The stale watch also runs in retrying, re-armed with the full staleAfterMs at the error that started it: a browser reconnect that never answers (WebKit leaves one that lands on a dead HTTP/2 connection CONNECTING without error or open) is closed, counted as a failure and reopened through waiting with the utility's own backoff. A native open re-arms it as usual.
• In retrying the watch therefore also caps the wait for the server's retry: value at staleAfterMs: keep staleAfterMs above the largest retry: a live stream writes (the heartbeat's baseMs + jitterMs), and pause a client for longer with a terminal event (the utility stops before the browser waits) or a non-200 (the utility backs off itself), not with a long retry:.
• When onStatus throws on the connecting of a reopen, the source is still opened first; the exception then propagates to the host (from the timer) or to the reconnectNow() caller, and the connection never stays connecting with nothing open.
• A malformed heartbeat counts as liveness but does not reach onHeartbeat.
• Only non-empty event ids are adopted; an event without an id never erases the cursor.
• A handler's exception propagates to the host's error reporting after the cursor and the stale watch were updated.
• Invalid options throw before any socket opens. An exception from the first resolveUrl / createEventSource propagates to the caller and leaves nothing open. The utility itself has no side effects such as navigation.

import { systemSseRuntime } from "@aiquants/sse"
import { createBrowserEventSource, openReopeningEventSource } from "@aiquants/sse/client"

const handle = openReopeningEventSource({
    resolveUrl: ({ lastEventId }) => `https://example.com/sse/feed${lastEventId === null ? "" : `?cursor=${encodeURIComponent(lastEventId)}`}`,
    initialLastEventId: null,
    withCredentials: false,
    createEventSource: createBrowserEventSource,
    runtime: systemSseRuntime,
    backoff: { initialDelayMs: 1_000, maxDelayMs: 30_000, multiplier: 2 },
    healthyOpenMs: 60_000,
    nativeRetryLimit: 0,
    giveUpAfterFailures: 20,
    terminalEvents: ["auth-required"],
    events: { message: (event) => console.log(event.data) },
    heartbeat: { staleAfterMs: 45_000, onHeartbeat: ({ serverTime, receivedAt }) => console.log(receivedAt - serverTime) },
    onStatus: (status) => console.log(status.kind),
})
// later
handle.close()

createBrowserEventSource(url, { withCredentials }) constructs the native EventSource resolved from globalThis at call time and throws TypeError when it is absent (for example during server rendering). EVENT_SOURCE_READY_STATE is { CONNECTING: 0, OPEN: 1, CLOSED: 2 } without touching the global. EventSourceLike / EventSourceFactory describe the minimal surface; undici's EventSource satisfies it too.

@aiquants/sse/react

useReopeningEventSource(connectionKey, options) takes every client option except onStatus and returns { status, getLastEventId, reconnectNow }.

• connectionKey === null keeps the hook idle; anything but null or a non-empty string throws TypeError during render. Only a key change reopens; a key change and unmount call close(), and under StrictMode exactly one instance stays open.
• While the key is not null, every render checks that each events value is a function and that heartbeat.onHeartbeat is a function or null (TypeError before anything opens); the relays would otherwise hide them from openReopeningEventSource.
• options is copied into a ref on every commit. Handlers, resolveUrl, createEventSource and onHeartbeat are called through relays, so the latest closure always runs and a new identity never reconnects.
• initialLastEventId, withCredentials, runtime, the numeric options and staleAfterMs are read when the connection of a key opens.
• The events key set, the terminalEvents set and whether the heartbeat is watched are the connection's structure. Changing them without changing the key throws TypeError (order does not matter).
• status is idle while the key is null, during server rendering and until the effect of a new key ran. getLastEventId and reconnectNow keep their identity across renders.
• Put side effects (reload, banners) in your own useEffect([status]); the hook has no second callback path.
• Where document is absent (server rendering), the ref sync is registered with useEffect instead of useLayoutEffect; neither runs on the server, and React 18 prints no server-rendering useLayoutEffect warning.

@aiquants/sse/server

| Export | Contract | | --- | --- | | SSE_RESPONSE_HEADERS | The three owned headers (frozen) | | SSE_COMPRESSIBLE_RESPONSE_HEADERS | The same three with no-transform dropped (frozen), which compressible: true returns | | formatSseEvent({ event, id, data }) | One event frame ending with a blank line; TypeError for an invalid or reserved name (use event: null for unnamed events), an id that isValidSseEventId rejects, or non-string data (an omitted event or id is rejected, never written as undefined) | | formatSseComment(text) | :<text> plus a blank line; TypeError for a non-string or for CR or LF | | formatSseRetry(delayMs) | retry: <n> plus a blank line; RangeError unless an integer from 1 to 2,147,483,647 | | createSseResponse({ requestSignal, headers, produce, onError, compressible? }) | A 200 streaming Response whose close reason is decided exactly once. compressible defaults to false: pass true only where the host flushes after every chunk and no part of the body is chosen by the reader | | terminalStreamResponse(name, { data, retryMs, headers }) | A 200 Response with retry: first, then one named event, then the end | | startSseHeartbeat(writer, { intervalMs, retry: { baseMs, jitterMs }, runtime }) | Jittered retry:, an immediate beat, then one per interval; returns an idempotent stop function | | readLastEventId(request, queryParam) | Non-empty Last-Event-ID header, else the non-empty query parameter, else null; the value is not validated. Both paths return the same value for every id isValidSseEventId accepts |

createSseResponse:

• Extra headers are merged with Headers.append, so several Set-Cookie values survive (this needs a Headers iterator that yields each Set-Cookie separately, as current runtimes do; verified on Node 24). Hop-by-hop / framing headers and overrides of the owned headers are TypeError; so is headers: undefined (pass null for none).
• An already aborted requestSignal ends the body at once and produce is never called.
• produce(writer) runs synchronously; a throw or rejection goes to onError(error, "producer") and closes with producer-failed. Resolving does not close the stream.
• The first close reason wins; onClose listeners run in registration order exactly once (a listener registered after close runs immediately). A throwing listener goes to onError(error, "close-listener") and the others still run.
• onError should not throw. If it does, its exception is rethrown from a microtask (the host's uncaught-error reporting) after the remaining listeners ran and the stream closed, so it never escapes writer.close() / createSseResponse and never becomes an unhandled rejection.
• writer.signal aborts with the default AbortError when the stream closes for any reason, so fetch(url, { signal: writer.signal }) cancels upstream work.

| Close reason | Trigger | Body closed by the package | | --- | --- | --- | | closed-by-server | writer.close() | yes | | client-aborted | requestSignal aborted | yes | | stream-cancelled | The response body was cancelled by its reader | no (already closed) | | enqueue-failed | The body stream refused a chunk | no (errored) | | producer-failed | produce threw or rejected | yes |

startSseHeartbeat needs a writer created by createSseResponse (TypeError otherwise) and a runtime with now, setTimer, clearTimer and random (TypeError). intervalMs, baseMs and jitterMs are positive safe integers, with intervalMs <= 2,147,483,647 and baseMs + jitterMs <= 2,147,483,647 (RangeError, before anything is written). It stops when a write returns false or the writer closes, and writes nothing to an already closed writer.

import { formatSseComment, formatSseEvent, formatSseRetry, readLastEventId } from "@aiquants/sse/server"

formatSseEvent({ event: "item", id: "c-42", data: "line 1\nline 2" })
// "event: item\nid: c-42\ndata: line 1\ndata: line 2\n\n"
formatSseComment("ping") // ":ping\n\n"
formatSseRetry(5_000) // "retry: 5000\n\n"
readLastEventId(new Request("https://example.com/sse/feed?cursor=c-1", { headers: { "Last-Event-ID": "c-2" } }), "cursor") // "c-2"

@aiquants/sse/shared-source

createSharedSource<TValue>({ start, lingerMs, replay, runtime, onListenerError, onStopError }) returns { subscribe, subscriberCount, status, dispose }.

• The first subscribe calls start(sink) once; every subscriber shares that run. After the last unsubscribe the run stops at once (lingerMs: 0) or after lingerMs; a subscribe during the linger keeps the same run.
• start and every stop run one at a time on a single promise lane, so a new run never starts before the previous stop finished. start must resolve to a stop function.
• sink.next(value) delivers synchronously, in subscription order, to a snapshot of the subscribers; a listener's exception goes to onListenerError. Calls from a stopped or failed run are ignored.
• replay: "latest" hands the run's latest value (including null / undefined) to a new subscriber before subscribe returns; the cache never crosses runs. replay: "none" caches nothing. Any other value is TypeError.
• sink.fail(error), a throwing / rejecting start, or a start that does not resolve to a function is fail closed: every subscriber is detached and receives error.
• dispose() drops subscribers without notifying them, stops the run and resolves when the stop finished (idempotent). subscribe throws after it.
• lingerMs must be an integer in [0, 2_147_483_647] (RangeError). start, onListenerError and onStopError must be functions and runtime must carry setTimer / clearTimer functions (TypeError at creation, before start could run). A reporter (onListenerError / onStopError) that throws is rethrown from a zero-delay timer, never swallowed. A start or stop that never settles blocks the lane: bound it inside start.

import { systemSseRuntime } from "@aiquants/sse"
import { createSharedSource } from "@aiquants/sse/shared-source"

type Reading = { readonly sensorId: string; readonly value: number }

// one upstream per process, shared by every open SSE response
export const readings = createSharedSource<Reading>({
    start: async (sink) => {
        const upstream = await connectReadings("wss://feed.example.com/readings") // your transport
        upstream.onReading = (reading) => sink.next(reading)
        upstream.onFailure = (error) => sink.fail(error)
        return () => upstream.close()
    },
    lingerMs: 5_000,
    replay: "latest",
    runtime: systemSseRuntime,
    onListenerError: (error) => console.error("reading subscriber failed", error),
    onStopError: (error) => console.error("reading upstream failed to stop", error),
})

// inside produce(writer):
const unsubscribe = readings.subscribe({
    next: (reading) => writer.writeEvent({ event: "reading", id: null, data: JSON.stringify(reading) }),
    error: () => writer.close(),
})
writer.onClose(unsubscribe)

@aiquants/sse/multiplex

createMultiplexedEventSource(options) carries many topics of one endpoint over one reopening connection (for example every panel of a page following its own channel on one stream endpoint) and returns { subscribe, topics, status, reopenNow, dispose }. The connection is openReopeningEventSource; the multiplexer only builds the URL, keeps the subscription set, routes frames and fans out the status. All options are required:

| Option | Rule | | --- | --- | | buildUrl(subscriptions) | Called on every open (each rebuild and every reopen of the connection) with { topic, cursor } for each subscription in subscription order, cursor being that subscriber's getCursor() at that moment; never with an empty list. Must return a non-empty string | | routeFrame(event) | Decode a frame once and return { topic, frame }, or null when it cannot be routed | | onUnroutedFrame(event, topic) | Receives every frame that was not delivered: topic is null when routeFrame returned null, otherwise the topic without a subscriber (for example one unsubscribed during the debounce). Nothing is dropped silently | | frameEvents | Non-empty tuple of unique event types that carry frames (message receives unnamed events); open, error, heartbeat, terminal names, duplicates and invalid names are TypeError | | maxTopics | Integer >= 1. A subscribe beyond it throws RangeError; the URL is never truncated | | rebuildDebounceMs | Integer in [0, 2_147_483_647]. Every subscription change restarts it, so a burst of changes costs one rebuild; 0 rebuilds on the next timer turn. It is a delay after the last change, not a bound: changes that keep arriving faster than this postpone the rebuild (the first open included) until they pause | | connection | Every openReopeningEventSource option except resolveUrl, events, onStatus, initialLastEventId and nativeRetryLimit, which the multiplexer owns (passing one is TypeError). Validated at creation with the client's rules, then copied down to backoff, the heartbeat settings and terminalEvents, so mutating the passed objects later changes nothing (runtime, createEventSource and onHeartbeat are used as passed) |

A subscriber (MultiplexSubscriber) is { getCursor, onFrame(frame, lastEventId), onStatus(status) }. status is a MultiplexStatus<TTerminal>: a client status, or { kind: "idle" } while nothing is subscribed and before the first open.

Behaviour worth knowing:

• nativeRetryLimit is fixed at 0: the browser's own reconnect would resend the old URL with stale per-topic cursors, and a Last-Event-ID header cannot carry one cursor per topic. Every failure (a CONNECTING error included) closes the source and reopens through the backoff with a URL rebuilt from the latest cursors. Backoff, healthy-open gate, giveUpAfterFailures, terminal events and the stale watch all come from connection, in one place for every topic.
• A subscription change closes the current source before the rebuild opens the next one (after the debounce), so at most one EventSource is ever open. A rebuild opens a fresh connection (failure count 0).
• While the connection is waiting (closed, with a reopen scheduled by the backoff), a subscription change does not rebuild: the pending reopen calls buildUrl with the subscriptions of that moment, so topics joining or leaving during an outage keep the backoff delay, the failure count and giveUpAfterFailures. Entering waiting also cancels a rebuild that is still pending.
• The last unsubscribe cancels a pending rebuild, closes the source at once and returns to idle without notifying anyone.
• Every status reaches every subscriber in subscription order; the closed of a retired connection never does. A subscriber that joins while the status is not idle receives it synchronously inside subscribe. When a subscriber's onStatus causes a newer status (reopenNow, an unsubscribe, dispose), the older one is not delivered to the rest.
• A terminal event is final for every topic while at least one topic stays subscribed: a pending rebuild is cancelled and later subscription changes do not reopen. reopenNow() rebuilds at once whatever the status (for example after a re-login in another tab) and returns false only when nothing is subscribed. Once the last topic left, status() is idle again and the next subscribe starts a new connection; a host that must keep the stop past that point keeps its own flag and calls reopenNow() when it resumes.
• A rebuild whose open throws (buildUrl, getCursor, createEventSource) reports faulted; the next subscription change or reopenNow() tries again. exhausted behaves the same way.
• subscribe throws Error for a topic that is already subscribed (a second subscriber never silently replaces the first) and after dispose(), RangeError beyond maxTopics, and TypeError for a non-string topic or callbacks that are not functions. The unsubscribe function is idempotent.
• An exception from a subscriber's onStatus is rethrown from a zero-delay runtime.setTimer handler (the host's error reporting), so the other subscribers are still notified. Exceptions from onFrame, routeFrame and onUnroutedFrame propagate to the event dispatcher, like client handlers. buildUrl, getCursor and routeFrame must not call back into the multiplexer.
• dispose() closes the connection and drops every subscriber without notifying it (idempotent); subscribe throws afterwards.

import { systemSseRuntime } from "@aiquants/sse"
import { createBrowserEventSource } from "@aiquants/sse/client"
import { createMultiplexedEventSource } from "@aiquants/sse/multiplex"

type ChannelMessage = { readonly channelId: string; readonly seq: number; readonly text: string }

// one page-wide connection shared by every open panel
export const channelStream = createMultiplexedEventSource<string, number, ChannelMessage, "auth-required">({
    buildUrl: (subscriptions) => `https://example.com/sse/channels?channels=${subscriptions.map(({ topic, cursor }) => `${encodeURIComponent(topic)}:${cursor}`).join(",")}`,
    routeFrame: (event) => {
        let frame: ChannelMessage
        try {
            frame = JSON.parse(event.data)
        } catch {
            return null
        }
        return typeof frame?.channelId === "string" ? { topic: frame.channelId, frame } : null
    },
    onUnroutedFrame: (event, topic) => console.warn("undelivered frame", topic, event.lastEventId),
    frameEvents: ["message"],
    maxTopics: 8, // the server's own limit of channels per request
    rebuildDebounceMs: 300,
    connection: {
        withCredentials: true,
        createEventSource: createBrowserEventSource,
        runtime: systemSseRuntime,
        backoff: { initialDelayMs: 1_000, maxDelayMs: 30_000, multiplier: 2 },
        healthyOpenMs: 60_000,
        giveUpAfterFailures: 5,
        terminalEvents: ["auth-required"],
        heartbeat: { staleAfterMs: 45_000, onHeartbeat: null },
    },
})

// in one panel
let cursor = 0
const unsubscribe = channelStream.subscribe(channelId, {
    getCursor: () => cursor,
    onFrame: (frame) => {
        cursor = frame.seq
        render(frame.text)
    },
    onStatus: (status) => showConnectionState(status.kind),
})
// on "auth-required": show a banner; after the user signed in again, call channelStream.reopenNow()

Recommended values (not defaults)

The package ships no defaults. These are the values the measurements behind this package support; pass them explicitly and adjust them to your edge.

| Option | Recommended | Why | | --- | --- | --- | | Heartbeat intervalMs | 15000 | Well below common proxy idle timeouts (around 60 s) | | heartbeat.staleAfterMs | 45000 (3 × interval) | Two missing beats in a row are not yet a false positive, and it stays above the heartbeat's largest retry: (10 s), so a normal browser reconnect is never cut short | | Heartbeat retry | { baseMs: 2000, jitterMs: 8000 } | Spreads reconnects when one HTTP/2 connection drops every tab's streams at once | | backoff | { initialDelayMs: 1000, maxDelayMs: 30000, multiplier: 2 } | Retries within 30 s during long outages while keeping the request count low | | healthyOpenMs | At least heartbeat.staleAfterMs + heartbeat intervalMs: 60000 with the values above | Turns every open-then-close loop into a backoff. A stream is forgiven its failures only after it survived a whole stale window and kept delivering past it: a value at or below staleAfterMs lets a source that opens and then stays silent reset the count before the stale watch fires, so every expiry is the first failure again, the backoff never grows and giveUpAfterFailures is never reached | | nativeRetryLimit | 1 for one stream whose cursor travels in the Last-Event-ID header; 0 when the cursor must travel in the URL | The browser's own reconnect reuses the original URL. 1 is safe with the stale watch on: retrying is watched, so a native reconnect that hangs on a dead HTTP/2 connection is closed after staleAfterMs and reopened by the utility | | rebuildDebounceMs (multiplex) | 300 | Not measured. Coalesces a burst of subscription changes (a page restoring several panels) into one rebuild. The first open waits this long after the last change of the burst; changes arriving faster than that keep postponing it | | maxTopics (multiplex) | The server's per-request limit | A subscribe beyond it throws instead of building a URL the server rejects for every topic at once | | Terminal retryMs | 86400000 | Honoured by Chromium, Firefox and WebKit (1 request in 30 s). Values above 2,147,483,647 are rejected with RangeError: Firefox parses retry: into a 32-bit integer, so 4,294,967,296 would wrap into a 500 ms reconnect loop, and Chromium / WebKit ignore values of 2^64 or more. Firefox also never waits less than 500 ms |

Testing helpers

| Export | Contract | | --- | --- | | FakeEventSource | EventTarget implementing EventSourceLike. simulateOpen(), simulateEvent(type, { data, lastEventId }), simulateNetworkError() (to CONNECTING), simulateFailure() (to CLOSED, like a non-200), close(), closeCallCount(). After close() a simulate* call still dispatches (a task queued before close) but the state stays CLOSED | | createFakeEventSourceFactory() | { create, instances, latest(), liveCount() }; liveCount() counts instances whose close() was never called | | createManualRuntime({ now, randomSequence }) | SseRuntime whose clock moves only through advanceBy(ms); timers fire in due-time order; random() throws once the sequence runs out; pendingTimerCount(). Rejects invalid delays like systemSseRuntime | | parseSseText(text) | WHATWG HTML interpretation of a complete body into ordered SseRecords (event, retry, comment); one leading BOM stripped, an unterminated trailing event discarded | | readSseResponse(response, { maxRecords, timeoutMs, runtime }) | Reads records until maxRecords (then cancels the body, ended: false) or the end (ended: true); rejects on timeout after cancelling the body |

import { createFakeEventSourceFactory, createManualRuntime } from "@aiquants/sse/testing"
import { openReopeningEventSource } from "@aiquants/sse/client"
import { expect, it } from "vitest"

it("reopens after a non-200 response", () => {
    const factory = createFakeEventSourceFactory()
    const runtime = createManualRuntime({ now: 0, randomSequence: [0.5] })
    openReopeningEventSource({
        resolveUrl: () => "https://example.com/sse/feed",
        initialLastEventId: null,
        withCredentials: false,
        createEventSource: factory.create,
        runtime,
        backoff: { initialDelayMs: 1_000, maxDelayMs: 30_000, multiplier: 2 },
        healthyOpenMs: 60_000,
        nativeRetryLimit: 0,
        giveUpAfterFailures: "never",
        terminalEvents: ["auth-required"],
        events: { message: () => undefined },
        heartbeat: "ignore",
        onStatus: () => undefined,
    })
    factory.latest().simulateFailure()
    expect(factory.liveCount()).toBe(0)
    runtime.advanceBy(500) // floor(0.5 × 1000)
    expect(factory.instances).toHaveLength(2)
})

import { systemSseRuntime } from "@aiquants/sse"
import { terminalStreamResponse } from "@aiquants/sse/server"
import { parseSseText, readSseResponse } from "@aiquants/sse/testing"

parseSseText("event: item\nid: c-1\ndata: a\ndata: b\n\n")
// [{ kind: "event", type: "item", data: "a\nb", lastEventId: "c-1" }]

const { records, ended } = await readSseResponse(terminalStreamResponse("auth-required", { data: "", retryMs: 86_400_000, headers: null }), {
    maxRecords: 10,
    timeoutMs: 1_000,
    runtime: systemSseRuntime,
})
// records: [{ kind: "retry", delayMs: 86400000 }, { kind: "event", type: "auth-required", data: "", lastEventId: "" }], ended: true

Security notes

• No hop-by-hop headers. The package never writes Connection and rejects hop-by-hop and framing headers from callers; an HTTP/2 client treats them as a protocol error. Upgrade / Proxy-Connection leaking from a backend through an HTTP/2 edge make strict clients fail (curl exit code 16) and made WebKit hang every stream on the connection; hide them at the edge on the SSE location (see Deployment).
• Authentication failures are in-band. Answer an unauthenticated or forbidden stream request with terminalStreamResponse, not with 401 / 403: a non-200 makes the browser close the stream silently, and the reopening client would keep retrying it. Do not put secrets in the terminal event's data.
• Cursors are client-controlled. readLastEventId returns the raw header or query value; validate its format (for example with isValidSseEventId) and authorize access to what it points at on the server. Event ids and cursors appear in URLs (resolveUrl) and therefore in access logs: use opaque cursors, never tokens or personal data.
• Event data is untrusted text on the client. Parse it (for example with a schema) and never inject it as HTML.
• Credentials. withCredentials: true sends cookies cross-origin; the server must answer with a matching Access-Control-Allow-Origin (not *) and Access-Control-Allow-Credentials: true.
• Caching and buffering. Cache-Control: no-store, no-transform keeps private streams out of shared caches and away from compression middleware; X-Accel-Buffering: no turns off nginx response buffering for the response.
• Compression is opt-in per response. compressible: true drops no-transform so the host may compress; omitting it keeps the refusal. It is worth doing only where the host flushes after every chunk, because one deflate stream carried across events is where the gain is: 149 bytes per event against 2,539 when each event is compressed alone, measured on a 15 KB payload repeated every three seconds. A compressor that batches instead would hold events, which is the one thing an SSE stream must not do.
• A compressible body must not contain what its reader chose. Compression leaks the size of what it compressed, so a body any part of which the reader can choose would tell that reader about the rest of it.
• Edge rate limits keyed by session. When a limit is keyed by a session cookie at the edge, key it by the cookie's signature only, never by the whole cookie: nginx caps limit_conn keys at 255 bytes, silently skips the limit above that and writes the full key (a replayable session) to the error log.
• Supply chain. No runtime dependencies, no source maps, no comments in the type declarations, and no repository or e-mail metadata in the manifest.

Deployment

Rules for serving SSE through a reverse proxy over HTTP/2, written for nginx (the same ideas apply to other proxies):

• Terminate TLS and HTTP/2 at the proxy and speak HTTP/1.1 to the application. A Node HTTP/2 server fails a response that carries a single Connection header.
• Enable HTTP/2 per server block (http2 on;, nginx 1.25.1 or later) rather than with listen ... http2, which applies to every server on that listener. Keep keepalive_timeout at 30 s or more: Safari does not reuse an HTTP/1 connection that stayed idle for more than about 29 s.
• SSE location: proxy_buffering off;, proxy_cache off;, proxy_http_version 1.1;, proxy_set_header Connection "";, a proxy_read_timeout of at least three heartbeat intervals, and proxy_hide_header Upgrade; / proxy_hide_header Proxy-Connection;. Hide those two headers on the SSE location only: on a location that proxies WebSocket they would strip Upgrade from the 101 response.
• No compression for text/event-stream (keep it out of gzip_types); compression holds events back.
• Rate limits before HTTP/2. Over HTTP/2 one connection delivers all of a page's asset requests at once, and limit_conn counts streams instead of TCP connections. Retune the limits (burst sizes, nodelay, 429 instead of 503), let them run, and only then enable HTTP/2; a native EventSource never reconnects after a non-200.
• Shared egress IPs. When users share one egress IP (for example behind an office NAT), a per-IP limit throttles all of them as one client. Key a per-browser ceiling by the session cookie's signature only (see Security notes), and map a missing or malformed cookie to the per-IP key, never to an empty key (nginx exempts empty keys from the limit).
• Reload instead of restart on certificate renewal: a restart drops every stream at once, and over HTTP/2 every tab of a browser reconnects at the same moment. Bound worker_shutdown_timeout so old workers do not keep streams on the old configuration forever, and call the proxy binary by absolute path from scheduled jobs.
• Keep the stale watch on (heartbeat: { staleAfterMs, onHeartbeat }) for every client served over HTTP/2: a stream reset mid-body or a connection closed after a reload can leave an EventSource OPEN and silent without an error event, and a browser reconnect that lands on a dead connection can stay CONNECTING without error or open (both observed with WebKit's Linux network stack).
• Check after each step: ALPN offers h2 on the application host; curl --http2 against the SSE path, a page and an API path exits 0 (or 28 when the stream is still open at --max-time) and never 16 (an HTTP/2 framing or header violation); rejected requests answer 429.

Adopting in another project

1. Run pnpm add @aiquants/sse (and react >= 18 if you use the hook).
2. Define one type for the application's terminal event names (for example type AppSseTerminalEvent = "auth-required" | "forbidden") and import it from both the server route and the client hook. Do not add names to the package.
3. Server: return createSseResponse({ requestSignal: request.signal, headers, produce, onError }) from the route and use startSseHeartbeat and readLastEventId inside produce. Answer expired sessions, missing permissions and invalid requests with terminalStreamResponse(kind, { data, retryMs, headers }) instead of a non-200. Never set Connection by hand.
4. Client: use useReopeningEventSource(key, { ... }) and pass every option explicitly (see Recommended values). Put the side effect of a terminal event in useEffect([status]). On screens that hold state in memory (an unsaved form, a recording), show a banner instead of reloading.
5. When several clients read the same upstream, put it behind one createSharedSource per process. When one page follows several topics of one endpoint, carry them over one connection with createMultiplexedEventSource instead of one EventSource per topic.
6. Align the edge with Deployment. Let compression middleware decide from no-transform rather than from the request's Accept, so that the per-response choice is the only rule; an edge that ignores no-transform (nginx's gzip does) has to exclude text/event-stream instead.
7. Write tests with FakeEventSource, createManualRuntime and readSseResponse from @aiquants/sse/testing.

Out of scope

• Pausing on visibilitychange: some screens must keep alerting while hidden, so the caller decides.
• Backpressure: the package does not watch desiredSize. Some Node adapters for WHATWG responses read the body eagerly, and there it would not reflect the client.
• NDJSON, WebSocket, cross-tab sharing and application payload schemas.

License

MIT