From ab8617cc0873097b283dad184b5ef1423d707302 Mon Sep 17 00:00:00 2001 From: Steven Zhang Date: Fri, 18 Sep 2026 15:08:45 -0400 Subject: [PATCH 1/5] chore: Scaffold @launchdarkly/eventsource with transport-independent core modules --- .github/workflows/eventsource.yml | 27 +++ package.json | 1 + packages/shared/eventsource/CHANGELOG.md | 4 + packages/shared/eventsource/LICENSE | 43 +++++ .../eventsource/__tests__/capacity.test.ts | 35 ++++ .../__tests__/listenerRegistry.test.ts | 80 ++++++++ .../eventsource/__tests__/retryDelay.test.ts | 97 ++++++++++ packages/shared/eventsource/jest.config.cjs | 11 ++ packages/shared/eventsource/package.json | 58 ++++++ packages/shared/eventsource/src/Event.ts | 100 ++++++++++ packages/shared/eventsource/src/capacity.ts | 53 ++++++ .../shared/eventsource/src/httpHelpers.ts | 81 +++++++++ packages/shared/eventsource/src/index.ts | 3 + .../eventsource/src/listenerRegistry.ts | 49 +++++ packages/shared/eventsource/src/parsing.ts | 41 +++++ packages/shared/eventsource/src/retryDelay.ts | 78 ++++++++ packages/shared/eventsource/src/types.ts | 171 ++++++++++++++++++ packages/shared/eventsource/tsconfig.json | 18 ++ packages/shared/eventsource/tsconfig.ref.json | 7 + packages/shared/eventsource/tsup.config.js | 15 ++ packages/shared/eventsource/typedoc.json | 5 + tsconfig.json | 3 + 22 files changed, 980 insertions(+) create mode 100644 .github/workflows/eventsource.yml create mode 100644 packages/shared/eventsource/CHANGELOG.md create mode 100644 packages/shared/eventsource/LICENSE create mode 100644 packages/shared/eventsource/__tests__/capacity.test.ts create mode 100644 packages/shared/eventsource/__tests__/listenerRegistry.test.ts create mode 100644 packages/shared/eventsource/__tests__/retryDelay.test.ts create mode 100644 packages/shared/eventsource/jest.config.cjs create mode 100644 packages/shared/eventsource/package.json create mode 100644 packages/shared/eventsource/src/Event.ts create mode 100644 packages/shared/eventsource/src/capacity.ts create mode 100644 packages/shared/eventsource/src/httpHelpers.ts create mode 100644 packages/shared/eventsource/src/index.ts create mode 100644 packages/shared/eventsource/src/listenerRegistry.ts create mode 100644 packages/shared/eventsource/src/parsing.ts create mode 100644 packages/shared/eventsource/src/retryDelay.ts create mode 100644 packages/shared/eventsource/src/types.ts create mode 100644 packages/shared/eventsource/tsconfig.json create mode 100644 packages/shared/eventsource/tsconfig.ref.json create mode 100644 packages/shared/eventsource/tsup.config.js create mode 100644 packages/shared/eventsource/typedoc.json diff --git a/.github/workflows/eventsource.yml b/.github/workflows/eventsource.yml new file mode 100644 index 0000000000..72aaef1172 --- /dev/null +++ b/.github/workflows/eventsource.yml @@ -0,0 +1,27 @@ +name: shared/eventsource + +on: + push: + branches: [main, 'feat/**'] + paths-ignore: + - '**.md' #Do not need to run CI for markdown changes. + pull_request: + branches: [main, 'feat/**'] + paths-ignore: + - '**.md' + +jobs: + build-test-eventsource: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + - uses: ./actions/setup-yarn + with: + node-version: 24.x + registry-url: 'https://registry.npmjs.org' + - id: shared + name: Shared CI Steps + uses: ./actions/ci + with: + workspace_name: '@launchdarkly/eventsource' + workspace_path: packages/shared/eventsource diff --git a/package.json b/package.json index 81649a3772..8e9bd8fd98 100644 --- a/package.json +++ b/package.json @@ -10,6 +10,7 @@ "packages/shared/sdk-server-edge", "packages/shared/akamai-edgeworker-sdk", "packages/shared/openfeature-server-common", + "packages/shared/eventsource", "packages/sdk/server-node", "packages/sdk/server-node/contract-tests", "packages/sdk/server-node/examples/features/custom-socks-proxy", diff --git a/packages/shared/eventsource/CHANGELOG.md b/packages/shared/eventsource/CHANGELOG.md new file mode 100644 index 0000000000..cd625c5ff5 --- /dev/null +++ b/packages/shared/eventsource/CHANGELOG.md @@ -0,0 +1,4 @@ +# Changelog + +All notable changes to `@launchdarkly/eventsource` will be documented in this file. This +project adheres to [Semantic Versioning](http://semver.org). diff --git a/packages/shared/eventsource/LICENSE b/packages/shared/eventsource/LICENSE new file mode 100644 index 0000000000..2964cd1c62 --- /dev/null +++ b/packages/shared/eventsource/LICENSE @@ -0,0 +1,43 @@ +Copyright 2026 Catamorphic, Co. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. + +------------------------------------------------------------------------------- + +Third-party notices + +Portions of this package are derived from the "eventsource" npm package, +used under the following license: + +The MIT License + +Copyright (c) EventSource GitHub organisation + +Permission is hereby granted, free of charge, to any person obtaining +a copy of this software and associated documentation files (the +"Software"), to deal in the Software without restriction, including +without limitation the rights to use, copy, modify, merge, publish, +distribute, sublicense, and/or sell copies of the Software, and to +permit persons to whom the Software is furnished to do so, subject to +the following conditions: + +The above copyright notice and this permission notice shall be +included in all copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, +EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF +MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND +NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE +LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION +OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION +WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. diff --git a/packages/shared/eventsource/__tests__/capacity.test.ts b/packages/shared/eventsource/__tests__/capacity.test.ts new file mode 100644 index 0000000000..6163e26f1b --- /dev/null +++ b/packages/shared/eventsource/__tests__/capacity.test.ts @@ -0,0 +1,35 @@ +import calculateCapacity from '../src/capacity'; + +const maxOverAllocation = 1024 * 1024; + +it('uses the minimum capacity', () => { + const [resize, newCapacity] = calculateCapacity(0, 1, maxOverAllocation); + expect(resize).toBe(true); + // The minimum is a fixed design constant. It must not track the host's Buffer.poolSize, + // which changed between Node versions and does not exist in a browser. + expect(newCapacity).toEqual(8192); +}); + +it('does not increase capacity when there is sufficient capacity', () => { + const [resize, newCapacity] = calculateCapacity(1024, 1023, maxOverAllocation); + expect(resize).toBe(false); + expect(newCapacity).toEqual(0); +}); + +it('uses exponential doubling capacity scaling', () => { + expect(calculateCapacity(8192, 8193, maxOverAllocation)).toEqual([true, 16384]); + expect(calculateCapacity(16384, 16385, maxOverAllocation)).toEqual([true, 32768]); +}); + +it('uses required capacity when it exceeds doubling', () => { + const [resize, newCapacity] = calculateCapacity(1024, 16384, maxOverAllocation); + expect(resize).toBe(true); + expect(newCapacity).toEqual(16384); +}); + +it('does not exceed max over allocation', () => { + const capacity = 1024 * 1024 * 3; + const [resize, newCapacity] = calculateCapacity(capacity, capacity + 1, maxOverAllocation); + expect(resize).toBe(true); + expect(newCapacity).toEqual(capacity + maxOverAllocation + 1); +}); diff --git a/packages/shared/eventsource/__tests__/listenerRegistry.test.ts b/packages/shared/eventsource/__tests__/listenerRegistry.test.ts new file mode 100644 index 0000000000..a56c3ad198 --- /dev/null +++ b/packages/shared/eventsource/__tests__/listenerRegistry.test.ts @@ -0,0 +1,80 @@ +import { createDefaultEventRegistry } from '../src/listenerRegistry'; + +it('dispatches an event to each listener registered for its type', () => { + const registry = createDefaultEventRegistry(); + const received: unknown[] = []; + registry.addEventListener('greeting', (e) => received.push(e)); + registry.addEventListener('greeting', (e) => received.push(e)); + const event = { type: 'greeting', data: 'hello' }; + registry.dispatch('greeting', event); + expect(received).toEqual([event, event]); +}); + +it('does not dispatch to listeners registered for another type', () => { + const registry = createDefaultEventRegistry(); + const received: unknown[] = []; + registry.addEventListener('other', (e) => received.push(e)); + registry.dispatch('greeting', { type: 'greeting' }); + expect(received).toEqual([]); +}); + +it('does nothing when dispatching a type with no listeners', () => { + const registry = createDefaultEventRegistry(); + expect(() => registry.dispatch('greeting', { type: 'greeting' })).not.toThrow(); +}); + +it('removes a registered listener', () => { + const registry = createDefaultEventRegistry(); + const received: unknown[] = []; + const listener = (e: unknown) => received.push(e); + registry.addEventListener('greeting', listener); + registry.removeEventListener('greeting', listener); + registry.dispatch('greeting', { type: 'greeting' }); + expect(received).toEqual([]); +}); + +it('removes at most one registration of a listener that registered twice', () => { + const registry = createDefaultEventRegistry(); + let count = 0; + const listener = () => { + count += 1; + }; + registry.addEventListener('greeting', listener); + registry.addEventListener('greeting', listener); + registry.removeEventListener('greeting', listener); + registry.dispatch('greeting', {}); + expect(count).toEqual(1); +}); + +it('does nothing when removing a listener that never registered', () => { + const registry = createDefaultEventRegistry(); + expect(() => registry.removeEventListener('greeting', () => {})).not.toThrow(); +}); + +it('dispatches to a snapshot of the listener list', () => { + const registry = createDefaultEventRegistry(); + const calls: string[] = []; + const second = () => calls.push('second'); + registry.addEventListener('greeting', () => { + calls.push('first'); + // Removed during this dispatch: still runs for this dispatch. + registry.removeEventListener('greeting', second); + // Added during this dispatch: does not run until the next dispatch. + registry.addEventListener('greeting', () => calls.push('third')); + }); + registry.addEventListener('greeting', second); + registry.dispatch('greeting', {}); + expect(calls).toEqual(['first', 'second']); +}); + +it('treats prototype-chain names as ordinary event types', () => { + const registry = createDefaultEventRegistry(); + // The server controls event type names through the SSE `event:` field, so names that collide + // with Object.prototype members must behave like any other type. + expect(() => registry.dispatch('__proto__', {})).not.toThrow(); + expect(() => registry.dispatch('constructor', {})).not.toThrow(); + const received: unknown[] = []; + registry.addEventListener('__proto__', (e) => received.push(e)); + registry.dispatch('__proto__', { type: '__proto__' }); + expect(received).toEqual([{ type: '__proto__' }]); +}); diff --git a/packages/shared/eventsource/__tests__/retryDelay.test.ts b/packages/shared/eventsource/__tests__/retryDelay.test.ts new file mode 100644 index 0000000000..07f1b3762d --- /dev/null +++ b/packages/shared/eventsource/__tests__/retryDelay.test.ts @@ -0,0 +1,97 @@ +import { RetryDelayStrategy, defaultBackoff, defaultJitter } from '../src/retryDelay'; + +function expectInRange(value: number, min: number, max: number) { + expect(value).toBeGreaterThanOrEqual(min); + expect(value).toBeLessThanOrEqual(max); +} + +it('can return fixed delay with no backoff or jitter', () => { + const d0 = 1000; + const r = RetryDelayStrategy(d0, 0); + const t0 = new Date().getTime() - 10000; + expect(r.nextRetryDelay(t0)).toEqual(d0); + expect(r.nextRetryDelay(t0 + 1000)).toEqual(d0); + expect(r.nextRetryDelay(t0 + 2000)).toEqual(d0); +}); + +it('can use backoff without jitter', () => { + const d0 = 10000; + const max = 60000; + const r = RetryDelayStrategy(d0, 0, defaultBackoff(max)); + const t0 = new Date().getTime() - 10000; + expect(r.nextRetryDelay(t0)).toEqual(d0); + expect(r.nextRetryDelay(t0 + 1000)).toEqual(d0 * 2); + expect(r.nextRetryDelay(t0 + 2000)).toEqual(d0 * 4); + expect(r.nextRetryDelay(t0 + 3000)).toEqual(max); +}); + +it('can use jitter without backoff', () => { + const d0 = 1000; + const r = RetryDelayStrategy(d0, 0, null, defaultJitter(0.5)); + const t0 = new Date().getTime() - 10000; + expectInRange(r.nextRetryDelay(t0), d0 / 2, d0); + expectInRange(r.nextRetryDelay(t0 + 1000), d0 / 2, d0); + expectInRange(r.nextRetryDelay(t0 + 2000), d0 / 2, d0); +}); + +it('can use jitter with backoff', () => { + const d0 = 10000; + const max = 60000; + const r = RetryDelayStrategy(d0, 0, defaultBackoff(max), defaultJitter(0.5)); + const t0 = new Date().getTime() - 10000; + expectInRange(r.nextRetryDelay(t0), d0 / 2, d0); + expectInRange(r.nextRetryDelay(t0 + 1000), d0, d0 * 2); + expectInRange(r.nextRetryDelay(t0 + 2000), d0 * 2, d0 * 4); + expectInRange(r.nextRetryDelay(t0 + 3000), max / 2, max); +}); + +it('can reset backoff based on reset interval', () => { + const d0 = 10000; + const max = 60000; + const resetInterval = 45000; + const r = RetryDelayStrategy(d0, resetInterval, defaultBackoff(max)); + const t0 = new Date().getTime() - 10000; + r.setGoodSince(t0); + + const t1 = t0 + 1000; + const d1 = r.nextRetryDelay(t1); + expect(d1).toEqual(d0); + + const t2 = t1 + d1; + r.setGoodSince(t2); + + const t3 = t2 + 10000; + const d2 = r.nextRetryDelay(t3); + expect(d2).toEqual(d0 * 2); + + const t4 = t3 + d2; + r.setGoodSince(t4); + + const t5 = t4 + resetInterval; + expect(r.nextRetryDelay(t5)).toEqual(d0); +}); + +it('resets the retry count when the base delay is changed', () => { + const d0 = 1000; + const r = RetryDelayStrategy(d0, 0, defaultBackoff(60000)); + const t0 = new Date().getTime() - 10000; + expect(r.nextRetryDelay(t0)).toEqual(d0); + expect(r.nextRetryDelay(t0)).toEqual(d0 * 2); + r.setBaseDelay(500); + expect(r.nextRetryDelay(t0)).toEqual(500); +}); + +it('correctly handles backoff and jitter with high retry count', () => { + // Verifies that we don't get numeric overflow errors due to using a very high exponential + // backoff that should not be a concern in floating point. + const d0 = 1000; + const max = 1000 * 60 * 60 * 24 * 365 * 200; // 200 years + const retryCount = 35; // 2^35 seconds + const r = RetryDelayStrategy(d0, 0, defaultBackoff(max), defaultJitter(0.5)); + + let d = 0; + for (let i = 0; i < retryCount; i += 1) { + d = r.nextRetryDelay(new Date().getTime()); + } + expectInRange(d, max / 2, max); +}); diff --git a/packages/shared/eventsource/jest.config.cjs b/packages/shared/eventsource/jest.config.cjs new file mode 100644 index 0000000000..022055c1ee --- /dev/null +++ b/packages/shared/eventsource/jest.config.cjs @@ -0,0 +1,11 @@ +module.exports = { + transform: { '^.+\\.ts?$': 'ts-jest' }, + testMatch: ['**/__tests__/**/*test.ts?(x)'], + testEnvironment: 'node', + moduleFileExtensions: ['ts', 'tsx', 'js', 'jsx', 'json', 'node'], + collectCoverageFrom: ['src/**/*.ts'], + // launchdarkly-js-test-helpers hands out ports from a per-process static counter, so running + // this package's test files across multiple jest worker processes lets two files allocate the + // same port and cross-talk. Run in a single worker so the counter stays globally unique. + maxWorkers: 1, +}; diff --git a/packages/shared/eventsource/package.json b/packages/shared/eventsource/package.json new file mode 100644 index 0000000000..92f6089eb5 --- /dev/null +++ b/packages/shared/eventsource/package.json @@ -0,0 +1,58 @@ +{ + "name": "@launchdarkly/eventsource", + "version": "0.0.1", + "type": "module", + "main": "./dist/index.cjs", + "module": "./dist/index.js", + "types": "./dist/index.d.ts", + "exports": { + ".": { + "import": { + "types": "./dist/index.d.ts", + "default": "./dist/index.js" + }, + "require": { + "types": "./dist/index.d.cts", + "default": "./dist/index.cjs" + } + } + }, + "homepage": "https://github.com/launchdarkly/js-core/tree/main/packages/shared/eventsource", + "repository": { + "type": "git", + "url": "https://github.com/launchdarkly/js-core.git" + }, + "description": "W3C compliant EventSource (server-sent events) client built on fetch, used by LaunchDarkly SDKs", + "files": [ + "dist" + ], + "keywords": [ + "launchdarkly", + "eventsource", + "sse", + "server-sent-events", + "streaming" + ], + "scripts": { + "build": "tsup", + "test": "npx jest --ci", + "lint": "oxlint .", + "lint:fix": "yarn run lint --fix", + "format": "oxfmt .", + "format:check": "oxfmt --check ." + }, + "license": "Apache-2.0", + "devDependencies": { + "@launchdarkly/js-sdk-common": "2.26.0", + "@types/jest": "^29.4.0", + "@types/node": "^25.9.1", + "jest": "^29.5.0", + "launchdarkly-js-test-helpers": "^2.2.0", + "oxfmt": "0.63.0", + "oxlint": "1.78.0", + "ts-jest": "^29.0.5", + "tsup": "^8.5.1", + "typedoc": "0.25.0", + "typescript": "5.1.6" + } +} diff --git a/packages/shared/eventsource/src/Event.ts b/packages/shared/eventsource/src/Event.ts new file mode 100644 index 0000000000..eedce7fe7d --- /dev/null +++ b/packages/shared/eventsource/src/Event.ts @@ -0,0 +1,100 @@ +/** The raw properties of an `error` or `end` event. */ +export type RawErrorPayload = { + status?: number; + headers?: Record; + message?: string; +}; + +/** + * The payload of the `error` and `end` events. + * + * @remark + * This shape mirrors the platform's `HttpErrorResponse`. It is the argument of `onerror` and the + * argument of the SDK's `errorFilter` callback. For this reason, it must stay interchangeable with + * `HttpErrorResponse` in both directions. + */ +export interface ErrorEvent { + readonly type?: string; + readonly message: string; + readonly status?: number; + readonly headers?: Record; +} + +/** + * The payload of the `open` event. + */ +export interface OpenEvent { + readonly type?: string; + readonly headers?: Record; +} + +/** + * The payload of the `retrying` event. + */ +export interface RetryingEvent { + readonly type?: string; + readonly delayMillis: number; +} + +/** + * The payload of the `closed` event. + */ +export interface ClosedEvent { + readonly type?: string; +} + +/** + * The payload of the `message` event, and of each event for a named SSE `event:` type. + */ +export interface MessageEvent { + readonly type: string; + readonly data: string; + readonly lastEventId: string; + readonly origin: string; +} + +/** + * The payload type for each event type that this library dispatches, keyed by event type name. + * The typed `addEventListener`/`removeEventListener` overloads on `EventSource` use this map. + */ +export interface EventSourceEventMap { + closed: ClosedEvent; + end: ErrorEvent; + error: ErrorEvent | MessageEvent; + message: MessageEvent; + open: OpenEvent; + retrying: RetryingEvent; +} + +/** + * W3C Event factory, and W3C MessageEvent factory for a server-named SSE frame. + * + * For a literal event type from {@link EventSourceEventMap}, the return type matches that map. + * For a non-literal string type, the return type is {@link MessageEvent}: a server-chosen SSE + * `event:` name always carries a `MessageEvent` payload. + * + * @see http://www.w3.org/TR/DOM-Level-3-Events/#interface-Event + * @see http://www.w3.org/TR/webmessaging/#event-definitions + */ +export function makeEvent( + type: K, + init?: Omit, +): EventSourceEventMap[K]; +export function makeEvent(type: string, init: Omit): MessageEvent; +export function makeEvent( + type: string, + optionalProperties?: Record, +): { type: string } { + const event: Record = {}; + Object.defineProperty(event, 'type', { writable: false, value: type, enumerable: true }); + if (optionalProperties) { + Object.keys(optionalProperties).forEach((f) => { + Object.defineProperty(event, f, { + writable: false, + value: optionalProperties[f], + enumerable: true, + }); + }); + } + return event as { type: string }; +} diff --git a/packages/shared/eventsource/src/capacity.ts b/packages/shared/eventsource/src/capacity.ts new file mode 100644 index 0000000000..a20d2faf10 --- /dev/null +++ b/packages/shared/eventsource/src/capacity.ts @@ -0,0 +1,53 @@ +/** + * The smallest capacity that this module requests. + * + * @remark + * The value keeps the allocation behavior of the original node implementation, which used the 8 KiB + * `Buffer.poolSize` default of the Node versions it targeted. + */ +const MIN_CAPACITY = 8192; + +/** + * Determine if a new capacity is needed for a buffer, and if it is, then what + * that new capacity should be. + * + * @param currentCapacity The current capacity of the buffer. + * @param requiredCapacity The required capacity from the buffer. + * @param maxOverAllocation The maximum extra capacity to allocate. This is how much the capacity + * can exceed the required capacity. If the over allocation exceeds this amount (from doubling), + * then instead the amount over allocated will be equal to maxOverAllocation. + * + * @returns Either `[false, 0]` if no allocation is needed, or `[true, capacity]` if an allocation + * is needed. + */ +export default function calculateCapacity( + currentCapacity: number, + requiredCapacity: number, + maxOverAllocation: number, +): [boolean, number] { + if (requiredCapacity > currentCapacity) { + let newCapacity = requiredCapacity; + + // Might as well start with a buffer of the maximum size that can be pooled. + // It is unlikely the initial buffer would be small enough to encounter + // this case. + if (newCapacity < MIN_CAPACITY) { + newCapacity = MIN_CAPACITY; + } + + // Exponential doubling capacity scaling. + const doubleCapacity = currentCapacity * 2; + if (newCapacity < doubleCapacity) { + newCapacity = doubleCapacity; + } + + // Doubling could become problematic over a certain size, so limit the total + // amount of extra capacity allocated. + const overAllocation = newCapacity - requiredCapacity; + if (overAllocation > maxOverAllocation) { + newCapacity = requiredCapacity + maxOverAllocation; + } + return [true, newCapacity]; + } + return [false, 0]; +} diff --git a/packages/shared/eventsource/src/httpHelpers.ts b/packages/shared/eventsource/src/httpHelpers.ts new file mode 100644 index 0000000000..c2cdb57fa5 --- /dev/null +++ b/packages/shared/eventsource/src/httpHelpers.ts @@ -0,0 +1,81 @@ +/** + * The structural fetch types below describe the exact subset of the WHATWG fetch API that this + * package uses. + * + * @see https://fetch.spec.whatwg.org/ + * + * @remark + * Because they are a structural subset, any standard `fetch` implementation should be able to + * satisfy `FetchFn` without casts. A platform that cannot use a global `fetch` (for example, + * one that needs an agent or TLS configuration) supplies its own function with the same shape. + **/ + +/** + * The response headers. `forEach` is the only member this package reads. + */ +export interface FetchHeaders { + forEach(callback: (value: string, key: string) => void): void; +} + +/** + * A reader over the response body stream. `read` resolves with the next chunk, or with + * `done: true` when the server ends the stream, and rejects when the connection drops. + */ +export interface FetchBodyReader { + read(): Promise<{ done: boolean; value?: Uint8Array }>; +} + +/** + * The streamable response body. + */ +export interface FetchResponseBody { + getReader(): FetchBodyReader; +} + +/** + * The subset of a fetch `Response` that this package uses. + */ +export interface FetchResponse { + readonly status: number; + readonly statusText: string; + readonly headers: FetchHeaders; + readonly body?: FetchResponseBody | null; +} + +/** + * The subset of a fetch `RequestInit` that this package produces. + * + * A transport can ignore the fields it has no equivalent for. In particular, a transport over a + * raw socket API can ignore `credentials`, and it does not need to follow redirects; the client + * treats a redirect status from such a transport as an ordinary non-200 failure. + */ +export interface FetchRequestOptions { + method?: string; + headers?: Record; + body?: string; + signal?: AbortSignal; + credentials?: 'include' | 'same-origin' | 'omit'; +} + +/** + * The fetch-shaped transport function. The global `fetch` satisfies this type. + */ +export type FetchFn = (url: string, init: FetchRequestOptions) => Promise; + +/** + * The default transport. It reads the global at call time, so a test can replace + * `globalThis.fetch` after this module loads. The wrapper also avoids an unbound reference to + * the global function, which some platforms reject. + */ +export const defaultFetch: FetchFn = (url, init) => globalThis.fetch(url, init); + +/** The methods for which a `fetch()` request cannot have a body. */ +export const bodylessMethods = ['GET', 'HEAD']; + +export function headersToObject(headers: FetchHeaders): Record { + const result: Record = {}; + headers.forEach((value, key) => { + result[key] = value; + }); + return result; +} diff --git a/packages/shared/eventsource/src/index.ts b/packages/shared/eventsource/src/index.ts new file mode 100644 index 0000000000..e03dc059b4 --- /dev/null +++ b/packages/shared/eventsource/src/index.ts @@ -0,0 +1,3 @@ +// TODO(scaffold): the EventSource client exports arrive with the client implementation. +export { createDefaultEventRegistry } from './listenerRegistry'; +export * from './types'; diff --git a/packages/shared/eventsource/src/listenerRegistry.ts b/packages/shared/eventsource/src/listenerRegistry.ts new file mode 100644 index 0000000000..748575147e --- /dev/null +++ b/packages/shared/eventsource/src/listenerRegistry.ts @@ -0,0 +1,49 @@ +import { EventListenerRegistry, RegisteredEventListener } from './types'; + +/** + * Creates the default `EventListenerRegistry`. + */ +export function createDefaultEventRegistry(): EventListenerRegistry { + const listeners = new Map(); + + return { + addEventListener(type: string, listener: RegisteredEventListener): void { + const forType = listeners.get(type); + if (forType) { + forType.push(listener); + } else { + listeners.set(type, [listener]); + } + }, + + removeEventListener(type: string, listener: RegisteredEventListener): void { + const forType = listeners.get(type); + if (!forType) { + return; + } + // At most one registration is removed, the most recent one, as `removeListener` in + // `EventEmitter` does. This keeps the default registry interchangeable with an + // `EventEmitter`-backed substitute. + for (let i = forType.length - 1; i >= 0; i -= 1) { + if (forType[i] === listener) { + forType.splice(i, 1); + break; + } + } + if (forType.length === 0) { + listeners.delete(type); + } + }, + + dispatch(type: string, event: unknown): void { + // The loop iterates over a copy of the list, as `emit` in `EventEmitter` does. A listener + // removed during this dispatch still runs. A listener added during this dispatch does not + // run until the next dispatch. An exception from a listener stops the dispatch and reaches + // the caller, as it does with NodeJS `EventEmitter`. + listeners + .get(type) + ?.slice() + .forEach((listener) => listener(event)); + }, + }; +} diff --git a/packages/shared/eventsource/src/parsing.ts b/packages/shared/eventsource/src/parsing.ts new file mode 100644 index 0000000000..9d14adbb34 --- /dev/null +++ b/packages/shared/eventsource/src/parsing.ts @@ -0,0 +1,41 @@ +export const bom = [239, 187, 191]; +export const colon = 58; +export const space = 32; +export const lineFeed = 10; +export const carriageReturn = 13; + +export const MAX_OVER_ALLOCATION = 1024 * 1024; // 1 MiB + +export function hasBom(buf: Uint8Array): boolean { + return bom.every((charCode, index) => buf[index] === charCode); +} + +/* + * Helpers for the fetch transport. + */ + +/** + * + * The parser decodes each field name and each field value as its own byte range. A `decode()` + * call without `{ stream: true }` keeps no state between calls. One shared decoder is therefore + * safe for unrelated ranges. + * + * The `ignoreBOM: true` flag keeps a leading U+FEFF in the output. Without the flag, the decoder + * removes a leading U+FEFF from every decoded range. That removal corrupts a value that starts + * with U+FEFF. The flag does not replace `hasBom()`. `hasBom()` removes the one byte order mark + * that the SSE specification ignores at the start of the stream, before the parser reads any + * bytes. The flag and `hasBom()` are complementary. + */ +export const utf8Decoder = new TextDecoder('utf-8', { ignoreBOM: true }); + +/** + * The `Headers` class of `fetch()` rejects some values through its ByteString conversion: a C0 + * control character other than tab, DEL, and each code point above U+00FF. A value outside the + * permitted set would make the Last-Event-ID header assignment throw on reconnect. That failure + * would block the connection permanently. + * + * @remark + * The original `launchdarkly-eventsource` implementation rejects only a NUL character; this stricter + * rule is what a `fetch()` transport requires. + */ +export const INVALID_HEADER_VALUE_CHAR = /[^\t\x20-\x7e\x80-\xff]/; diff --git a/packages/shared/eventsource/src/retryDelay.ts b/packages/shared/eventsource/src/retryDelay.ts new file mode 100644 index 0000000000..9355b89c3e --- /dev/null +++ b/packages/shared/eventsource/src/retryDelay.ts @@ -0,0 +1,78 @@ +/** + * Computes a backoff delay from the current base delay and the number of consecutive retries. + */ +export type BackoffStrategy = (baseDelayMillis: number, retryCount: number) => number; + +/** + * Applies jitter to a computed delay. + */ +export type JitterStrategy = (computedDelayMillis: number) => number; + +/** + * The object that the `RetryDelayStrategy` factory returns. + */ +export interface RetryDelayStrategy { + nextRetryDelay(currentTimeMillis: number): number; + setGoodSince(goodSinceTimeMillis: number): void; + setBaseDelay(baseDelay: number): void; +} + +// Encapsulation of configurable backoff/jitter behavior. +// +// - The system can either be in a "good" state or a "bad" state. The initial state is "bad"; the +// caller is responsible for indicating when it transitions to "good". When we ask for a new retry +// delay, that implies the state is now transitioning to "bad". +// +// - There is a configurable base delay, which can be changed at any time (if the SSE server sends +// us a "retry:" directive). +// +// - There are optional strategies for applying backoff and jitter to the delay. +// +// The factory keeps the PascalCase name of the original function. +// eslint-disable-next-line @typescript-eslint/naming-convention +export function RetryDelayStrategy( + baseDelayMillis: number, + resetIntervalMillis?: number, + backoff?: BackoffStrategy | null, + jitter?: JitterStrategy | null, +): RetryDelayStrategy { + let currentBaseDelay = baseDelayMillis; + let retryCount = 0; + let goodSince: number | null | undefined; + return { + nextRetryDelay(currentTimeMillis: number): number { + if ( + goodSince && + resetIntervalMillis && + currentTimeMillis - goodSince >= resetIntervalMillis + ) { + retryCount = 0; + } + goodSince = null; + const delay = backoff ? backoff(currentBaseDelay, retryCount) : currentBaseDelay; + retryCount += 1; + return jitter ? jitter(delay) : delay; + }, + + setGoodSince(goodSinceTimeMillis: number): void { + goodSince = goodSinceTimeMillis; + }, + + setBaseDelay(baseDelay: number): void { + currentBaseDelay = baseDelay; + retryCount = 0; + }, + }; +} + +export function defaultBackoff(maxDelayMillis: number): BackoffStrategy { + return (baseDelayMillis: number, retryCount: number) => { + const d = baseDelayMillis * 2 ** retryCount; + return d > maxDelayMillis ? maxDelayMillis : d; + }; +} + +export function defaultJitter(ratio: number): JitterStrategy { + return (computedDelayMillis: number) => + computedDelayMillis - Math.trunc(Math.random() * ratio * computedDelayMillis); +} diff --git a/packages/shared/eventsource/src/types.ts b/packages/shared/eventsource/src/types.ts new file mode 100644 index 0000000000..33ae0b4380 --- /dev/null +++ b/packages/shared/eventsource/src/types.ts @@ -0,0 +1,171 @@ +/* + * Several types in this file mirror shapes from `@launchdarkly/js-sdk-common` without importing + * them. The `typeCompatibility` test suite enforces that they stay compatible with that platform + * contract. + */ + +export type { + ClosedEvent, + ErrorEvent, + EventSourceEventMap, + MessageEvent, + OpenEvent, + RetryingEvent, +} from './Event'; + +import type { ErrorEvent } from './Event'; +import type { FetchFn } from './httpHelpers'; + +/** + * A listener as the registry stores it. The registry stores listeners for every event type, + * including SSE event names the server chooses at runtime, so the stored shape is untyped. The + * typed `addEventListener` overloads on `EventSource` narrow the payload for callers. + */ +export type RegisteredEventListener = (event: any) => void; + +/** + * Stores and dispatches the listeners that an `EventSource` instance registers through + * `addEventListener`. The `on*` slots are plain members of the instance. The registry never + * sees them. + * + * The default implementation uses a `Map` (`createDefaultEventRegistry`). A caller can supply a + * different implementation through the `createEventRegistry` init option, for example one that + * uses Node's `EventEmitter`. Note that `EventEmitter` throws from `emit('error')` when `error` + * has no listener. A substitute must not do this. `dispatch` must stay silent when a type has + * no listener. + */ +export interface EventListenerRegistry { + /** + * Adds `listener` at the end of the list for `type`. The same listener can register more than + * once. + */ + addEventListener(type: string, listener: RegisteredEventListener): void; + + /** + * Removes at most one registration of `listener` for `type`. The most recent registration is + * the one removed, as `removeListener` in `EventEmitter` does. + */ + removeEventListener(type: string, listener: RegisteredEventListener): void; + + /** + * Calls each listener registered for `type` with `event`. Does nothing when `type` has no + * registered listener. Unlike `emit('error')` in `EventEmitter`, this never throws when a + * type has no listener. + * + * A substitute implementation must copy the listener list before it iterates. A listener + * removed during a dispatch still runs in that dispatch. A listener added during a dispatch + * waits for the next dispatch. An exception from a listener propagates to the caller of + * `dispatch`. + */ + dispatch(type: string, event: unknown): void; +} + +/** + * Creates the registry for one `EventSource` instance. `createEventSource` calls the factory + * once, so each instance gets its own registry. + */ +export type EventListenerRegistryFactory = () => EventListenerRegistry; + +export type { + FetchBodyReader, + FetchFn, + FetchHeaders, + FetchRequestOptions, + FetchResponse, + FetchResponseBody, +} from './httpHelpers'; + +/** + * The options that `createEventSource` understands. + * + * The first group of fields mirrors the platform's `EventSourceInitDict`. Each field that the SDK + * supplies stays required here, to match the platform contract. A caller that calls + * `createEventSource` directly can omit each field, because that function's parameter is + * `Partial<...>`. A missing field falls back to a default (`errorFilter`, + * `initialRetryDelayMillis`), or the feature stays off (`headers`, `readTimeoutMillis`, + * `retryResetIntervalMillis`). + */ +export interface EventSourceInitDict { + method?: string; + + headers: { [key: string]: string | string[] }; + + body?: string; + + /** + * Decides if the client retries a given error. Return false to stop the retries and close the + * stream. When this field is not set, the client retries I/O errors and HTTP 500, 502, 503, + * and 504. + */ + errorFilter: (err: ErrorEvent) => boolean; + + initialRetryDelayMillis: number; + + readTimeoutMillis: number; + + retryResetIntervalMillis: number; + + /** + * If set, the client decreases each computed retry delay by a random amount. The maximum + * decrease is this fraction of the delay. + */ + jitterRatio?: number; + + /** + * If set, retry delays grow exponentially up to this limit. + */ + maxBackoffMillis?: number; + + /** + * When true, the request omits the default `Cache-Control: no-cache` and + * `Accept: text/event-stream` headers. + */ + skipDefaultHeaders?: boolean; + + /** + * When true, the client sends credentials (cookies, HTTP authentication) with the request, also + * cross-origin. This flag maps to `credentials: 'include'` in `fetch()`. A transport without a + * credentials concept ignores it. + */ + withCredentials?: boolean; + + /** + * Returns the URL for the next connection attempt. When set, the client calls this function + * immediately before each request, including the first. This keeps a query parameter current + * when the parameter changes between attempts. + */ + urlBuilder?: () => string; + + /** + * The transport used to open the stream. When absent, the client uses the global `fetch`. + * + * The client only needs the structural subset declared by {@link FetchFn}, so a real `fetch` + * implementation satisfies this option without casts. A platform without a suitable global + * `fetch`, or one that needs agents, proxies, or TLS configuration, supplies a function with + * the same shape over its own transport. + */ + fetch?: FetchFn; + + /** + * Creates the listener registry for this instance. The registry stores the listeners + * registered through `addEventListener`. When this option is absent, the instance uses the + * internal `Map`-backed registry (`createDefaultEventRegistry`). + * + * A substitute must satisfy {@link EventListenerRegistry}. An SDK can supply one that uses a + * native eventing primitive, such as Node's `EventEmitter`. A raw `EventEmitter` throws from + * `emit('error')` when `error` has no listener. An `EventEmitter`-backed substitute must + * guard against that case. + * + * The registry is trusted code. It sees every event this instance dispatches, and it controls + * what its listeners receive. + * + * `createEventSource` throws a `TypeError` that names `createEventRegistry` when this option + * is not a function, or when the returned value is missing a required `EventListenerRegistry` + * method. + * + * This function must return a new registry on each call. `createEventSource` calls it once + * per instance. When a factory returns one shared registry for more than one instance, the + * separate event sources deliver each other's events. + */ + createEventRegistry?: EventListenerRegistryFactory; +} diff --git a/packages/shared/eventsource/tsconfig.json b/packages/shared/eventsource/tsconfig.json new file mode 100644 index 0000000000..0b57072a5a --- /dev/null +++ b/packages/shared/eventsource/tsconfig.json @@ -0,0 +1,18 @@ +{ + "compilerOptions": { + "rootDir": "src", + "outDir": "dist", + "target": "ES2017", + "lib": ["es2020", "dom"], + "module": "commonjs", + "strict": true, + "noImplicitOverride": true, + // Needed for CommonJS modules. + "allowSyntheticDefaultImports": true, + "sourceMap": true, + "declaration": true, + "declarationMap": true, // enables importers to jump to source + "stripInternal": true + }, + "exclude": ["**/*.test.ts", "dist", "node_modules", "__tests__", "contract-tests"] +} diff --git a/packages/shared/eventsource/tsconfig.ref.json b/packages/shared/eventsource/tsconfig.ref.json new file mode 100644 index 0000000000..0c86b2c554 --- /dev/null +++ b/packages/shared/eventsource/tsconfig.ref.json @@ -0,0 +1,7 @@ +{ + "extends": "./tsconfig.json", + "include": ["src/**/*"], + "compilerOptions": { + "composite": true + } +} diff --git a/packages/shared/eventsource/tsup.config.js b/packages/shared/eventsource/tsup.config.js new file mode 100644 index 0000000000..2174bc1f74 --- /dev/null +++ b/packages/shared/eventsource/tsup.config.js @@ -0,0 +1,15 @@ +import { defineConfig } from 'tsup'; + +export default defineConfig([ + { + entry: { + index: 'src/index.ts', + }, + minify: true, + format: ['esm', 'cjs'], + sourcemap: true, + clean: true, + dts: true, + metafile: true, + }, +]); diff --git a/packages/shared/eventsource/typedoc.json b/packages/shared/eventsource/typedoc.json new file mode 100644 index 0000000000..7ac616b544 --- /dev/null +++ b/packages/shared/eventsource/typedoc.json @@ -0,0 +1,5 @@ +{ + "extends": ["../../../typedoc.base.json"], + "entryPoints": ["src/index.ts"], + "out": "docs" +} diff --git a/tsconfig.json b/tsconfig.json index 1ebba5d3f9..ad036a527f 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -22,6 +22,9 @@ { "path": "./packages/shared/openfeature-server-common/tsconfig.ref.json" }, + { + "path": "./packages/shared/eventsource/tsconfig.ref.json" + }, { "path": "./packages/sdk/server-node/tsconfig.ref.json" }, From e4bfaeb3166b44d00976e902cdf60d01f2342ca9 Mon Sep 17 00:00:00 2001 From: Steven Zhang Date: Wed, 30 Sep 2026 13:58:32 -0400 Subject: [PATCH 2/5] feat: Add a forked eventsource-parser as the transport-independent SSE parser --- packages/shared/eventsource/LICENSE | 27 + .../__tests__/parser/parse.test.ts | 205 +++++++ .../shared/eventsource/src/parser/errors.ts | 51 ++ .../shared/eventsource/src/parser/index.ts | 12 + .../shared/eventsource/src/parser/parse.ts | 580 ++++++++++++++++++ .../shared/eventsource/src/parser/types.ts | 145 +++++ 6 files changed, 1020 insertions(+) create mode 100644 packages/shared/eventsource/__tests__/parser/parse.test.ts create mode 100644 packages/shared/eventsource/src/parser/errors.ts create mode 100644 packages/shared/eventsource/src/parser/index.ts create mode 100644 packages/shared/eventsource/src/parser/parse.ts create mode 100644 packages/shared/eventsource/src/parser/types.ts diff --git a/packages/shared/eventsource/LICENSE b/packages/shared/eventsource/LICENSE index 2964cd1c62..7bd89472bd 100644 --- a/packages/shared/eventsource/LICENSE +++ b/packages/shared/eventsource/LICENSE @@ -41,3 +41,30 @@ NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + +------------------------------------------------------------------------------- + +Portions of this package (the stream parser at src/parser) are derived from +the "eventsource-parser" npm package, used under the following license: + +MIT License + +Copyright (c) 2026 Espen Hovlandsdal + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/packages/shared/eventsource/__tests__/parser/parse.test.ts b/packages/shared/eventsource/__tests__/parser/parse.test.ts new file mode 100644 index 0000000000..14939d83e7 --- /dev/null +++ b/packages/shared/eventsource/__tests__/parser/parse.test.ts @@ -0,0 +1,205 @@ +import { createParser, EventSourceMessage, EventSourceParser, ParseError } from '../../src/parser'; + +interface RecordingParser { + parser: EventSourceParser; + events: EventSourceMessage[]; + ids: string[]; + retries: number[]; + comments: string[]; + errors: ParseError[]; +} + +function recordingParser(): RecordingParser { + const events: EventSourceMessage[] = []; + const ids: string[] = []; + const retries: number[] = []; + const comments: string[] = []; + const errors: ParseError[] = []; + const parser = createParser({ + onEvent: (event) => events.push(event), + onId: (id) => ids.push(id), + onRetry: (retry) => retries.push(retry), + onComment: (comment) => comments.push(comment), + onError: (error) => errors.push(error), + }); + return { parser, events, ids, retries, comments, errors }; +} + +it('parses a single-line event', () => { + const { parser, events } = recordingParser(); + parser.feed('data: hello\n\n'); + expect(events).toHaveLength(1); + expect(events[0].data).toEqual('hello'); + expect(events[0].event).toBeUndefined(); +}); + +it('parses two events from one chunk', () => { + const { parser, events } = recordingParser(); + parser.feed('data: a\n\ndata: b\n\n'); + expect(events.map((e) => e.data)).toEqual(['a', 'b']); +}); + +it('accepts CRLF as the line terminator', () => { + const { parser, events } = recordingParser(); + parser.feed('data: hello\r\n\r\n'); + expect(events.map((e) => e.data)).toEqual(['hello']); +}); + +it('accepts a bare CR as the line terminator', () => { + const { parser, events } = recordingParser(); + parser.feed('data: hello\r\r'); + expect(events.map((e) => e.data)).toEqual(['hello']); +}); + +it('treats a CRLF split across two feeds as one terminator', () => { + const { parser, events } = recordingParser(); + parser.feed('data: hello\r'); + parser.feed('\n\r\n'); + expect(events.map((e) => e.data)).toEqual(['hello']); +}); + +it('parses a line terminated by a CR at a chunk boundary when the next chunk is not a line feed', () => { + const { parser, events } = recordingParser(); + parser.feed('data: a\r'); + parser.feed('data: b\r\r'); + expect(events.map((e) => e.data)).toEqual(['a\nb']); +}); + +it('strips a leading byte order mark', () => { + const { parser, events } = recordingParser(); + parser.feed('\uFEFFdata: x\n\n'); + expect(events.map((e) => e.data)).toEqual(['x']); +}); + +it('strips a raw byte order mark split across feeds', () => { + const { parser, events } = recordingParser(); + // The three UTF-8 BOM bytes arrive as separate byte-valued code points across chunks. + parser.feed('\xEF'); + parser.feed('\xBB\xBF'); + parser.feed('data: x\n\n'); + expect(events.map((e) => e.data)).toEqual(['x']); +}); + +it('does not strip a byte order mark after the start of the stream', () => { + const { parser, events, errors } = recordingParser(); + parser.feed('data: x\n\uFEFFdata: y\n\n'); + expect(events.map((e) => e.data)).toEqual(['x']); + expect(errors).toHaveLength(1); + expect(errors[0].type).toEqual('unknown-field'); +}); + +it('joins multi-line data with newlines', () => { + const { parser, events } = recordingParser(); + parser.feed('data: one\ndata:two\n\n'); + expect(events.map((e) => e.data)).toEqual(['one\ntwo']); +}); + +it('reports the id before the event when a block has both', () => { + const order: string[] = []; + const parser = createParser({ + onId: (id) => order.push(`id:${id}`), + onEvent: (event) => order.push(`event:${event.data}`), + }); + parser.feed('id: 1\ndata: x\n\n'); + expect(order).toEqual(['id:1', 'event:x']); +}); + +it('reports an id for a block that has no data', () => { + const { parser, events, ids } = recordingParser(); + parser.feed('id: 5\n\n'); + expect(ids).toEqual(['5']); + expect(events).toHaveLength(0); +}); + +it('ignores an id that contains a null character', () => { + const { parser, events, ids } = recordingParser(); + parser.feed('id: a\u0000b\ndata: x\n\n'); + expect(ids).toEqual([]); + expect(events.map((e) => e.data)).toEqual(['x']); +}); + +it('reports an empty id field as an empty string', () => { + const { parser, ids } = recordingParser(); + parser.feed('id:\ndata: x\n\n'); + expect(ids).toEqual(['']); +}); + +it('forwards an all-digit retry value', () => { + const { parser, retries } = recordingParser(); + parser.feed('retry: 2500\n\n'); + expect(retries).toEqual([2500]); +}); + +it('reports a retry value that is not all digits through onError and does not forward it', () => { + const { parser, retries, errors } = recordingParser(); + parser.feed('retry: 12abc\n\n'); + expect(retries).toEqual([]); + expect(errors).toHaveLength(1); + expect(errors[0].type).toEqual('invalid-retry'); +}); + +it('treats a field name without a colon as a field with an empty value', () => { + const { parser, events } = recordingParser(); + parser.feed('data\n\n'); + expect(events.map((e) => e.data)).toEqual(['']); +}); + +it('reports comment lines through onComment', () => { + const { parser, events, comments } = recordingParser(); + parser.feed(': keep-alive\n\n:x\n\n'); + expect(comments).toEqual(['keep-alive', 'x']); + expect(events).toHaveLength(0); +}); + +it('reports an unknown field through onError and dispatches nothing', () => { + const { parser, events, errors } = recordingParser(); + parser.feed('foo: bar\n\n'); + expect(events).toHaveLength(0); + expect(errors).toHaveLength(1); + expect(errors[0].type).toEqual('unknown-field'); + expect(errors[0].field).toEqual('foo'); +}); + +it('leaves the event type undefined when the event field is empty', () => { + const { parser, events } = recordingParser(); + parser.feed('event:\ndata: x\n\n'); + expect(events).toHaveLength(1); + expect(events[0].event).toBeUndefined(); +}); + +it('carries an explicit event type on the parsed event', () => { + const { parser, events } = recordingParser(); + parser.feed('event: greeting\ndata: hi\n\n'); + expect(events[0].event).toEqual('greeting'); + expect(events[0].data).toEqual('hi'); +}); + +it('buffers a field split across many small feeds', () => { + const { parser, events } = recordingParser(); + parser.feed('da'); + parser.feed('ta: Hel'); + parser.feed('lo\n\n'); + expect(events.map((e) => e.data)).toEqual(['Hello']); +}); + +it('discards a line that cannot become a field and resumes at the next line', () => { + const { parser, events } = recordingParser(); + parser.feed('xyzzy'); + parser.feed('zzz\ndata: x\n\n'); + expect(events.map((e) => e.data)).toEqual(['x']); +}); + +it('drops a buffered partial line on reset', () => { + const { parser, events } = recordingParser(); + parser.feed('data: par'); + parser.reset(); + parser.feed('data: x\n\n'); + expect(events.map((e) => e.data)).toEqual(['x']); +}); + +it('consumes a pending partial line when reset is told to consume', () => { + const { parser, retries } = recordingParser(); + parser.feed('retry: 99'); + parser.reset({ consume: true }); + expect(retries).toEqual([99]); +}); diff --git a/packages/shared/eventsource/src/parser/errors.ts b/packages/shared/eventsource/src/parser/errors.ts new file mode 100644 index 0000000000..e3cb95542f --- /dev/null +++ b/packages/shared/eventsource/src/parser/errors.ts @@ -0,0 +1,51 @@ +/** + * @remark + * This implementation is derived from `errors.ts` of the `eventsource-parser` package, + * version 4.1.1 (https://github.com/rexxars/eventsource-parser). See the LICENSE file at the + * package root for the original license terms. + */ + +/** + * The type of error that occurred. + * @public + */ +export type ErrorType = 'invalid-retry' | 'unknown-field' | 'max-buffer-size-exceeded'; + +/** + * Error thrown when encountering an issue during parsing. + * + * @public + */ +export class ParseError extends Error { + /** + * The type of error that occurred. + */ + type: ErrorType; + + /** + * In the case of a completed unknown field encountered in the stream, this will be the field name. + */ + field?: string | undefined; + + /** + * In the case of a completed unknown field encountered in the stream, this will be the value of the field. + */ + value?: string | undefined; + + /** + * The line that caused the error, if available. + */ + line?: string | undefined; + + constructor( + message: string, + options: { type: ErrorType; field?: string; value?: string; line?: string }, + ) { + super(message); + this.name = 'ParseError'; + this.type = options.type; + this.field = options.field; + this.value = options.value; + this.line = options.line; + } +} diff --git a/packages/shared/eventsource/src/parser/index.ts b/packages/shared/eventsource/src/parser/index.ts new file mode 100644 index 0000000000..1ea09e522d --- /dev/null +++ b/packages/shared/eventsource/src/parser/index.ts @@ -0,0 +1,12 @@ +/** + * @remark + * This implementation is derived from `index.ts` of the `eventsource-parser` package, + * version 4.1.1 (https://github.com/rexxars/eventsource-parser). See the LICENSE file at the + * package root for the original license terms. The upstream `stream.ts` module (a TransformStream + * wrapper) is not ported because nothing in this package uses it. + */ + +export { ParseError } from './errors'; +export type { ErrorType } from './errors'; +export { createParser } from './parse'; +export type { EventSourceMessage, EventSourceParser, ParserCallbacks, ParserConfig } from './types'; diff --git a/packages/shared/eventsource/src/parser/parse.ts b/packages/shared/eventsource/src/parser/parse.ts new file mode 100644 index 0000000000..78e702ac35 --- /dev/null +++ b/packages/shared/eventsource/src/parser/parse.ts @@ -0,0 +1,580 @@ +/** + * @remark + * This implementation is derived from `parse.ts` of the `eventsource-parser` package, + * version 4.1.1 (https://github.com/rexxars/eventsource-parser). See the LICENSE file at the + * package root for the original license terms. The logic is unchanged. The file only reorders + * function declarations and avoids parameter reassignment to satisfy this repository's lint + * rules. + */ +/** + * EventSource/Server-Sent Events parser + * @see https://html.spec.whatwg.org/multipage/server-sent-events.html + */ +import { ParseError } from './errors'; +import type { EventSourceParser, ParserConfig } from './types'; + +// ASCII codes used in the hot parsing paths. +const LF = 10; +const CR = 13; +const SPACE = 32; +const MAX_FIELD_PREFIX_LENGTH = 6; + +/** + * Checks if `chunk` starts with the literal `data:` at index `i`. + * + * Equivalent to `chunk.startsWith('data:', i)`, but benchmarks show this + * hand-unrolled char-code comparison is ~20% faster on common event types. + * The caller passes `firstCharCode` (the code at `i`) so it can be reused + * across prefix checks. + * + * ASCII: 'd' = 100, 'a' = 97, 't' = 116, 'a' = 97, ':' = 58 + */ +function isDataPrefix(chunk: string, i: number, firstCharCode: number): boolean { + return ( + firstCharCode === 100 && + chunk.charCodeAt(i + 1) === 97 && + chunk.charCodeAt(i + 2) === 116 && + chunk.charCodeAt(i + 3) === 97 && + chunk.charCodeAt(i + 4) === 58 + ); +} + +/** + * Checks if `chunk` starts with the literal `event:` at index `i`. + * + * See {@link isDataPrefix} for why this is hand-unrolled rather than using + * `String.prototype.startsWith`. + * + * ASCII: 'e' = 101, 'v' = 118, 'e' = 101, 'n' = 110, 't' = 116, ':' = 58 + */ +function isEventPrefix(chunk: string, i: number, firstCharCode: number): boolean { + return ( + firstCharCode === 101 && + chunk.charCodeAt(i + 1) === 118 && + chunk.charCodeAt(i + 2) === 101 && + chunk.charCodeAt(i + 3) === 110 && + chunk.charCodeAt(i + 4) === 116 && + chunk.charCodeAt(i + 5) === 58 + ); +} + +function isPotentialField(line: string, field: string): boolean { + let i = 1; + while (i < line.length && i < field.length) { + if (line.charCodeAt(i) !== field.charCodeAt(i)) { + return false; + } + i++; + } + + return line.length <= field.length || line.charCodeAt(field.length) === 58; +} + +/** + * Creates a new EventSource parser. + * + * @param config - Parser configuration. Accepts callbacks (see {@link ParserCallbacks}) + * and options like `maxBufferSize` (see {@link ParserConfig}). + * + * @returns A new EventSource parser, with `feed` and `reset` methods. + * @public + */ +export function createParser(config: ParserConfig): EventSourceParser { + if (typeof config === 'function') { + throw new TypeError( + '`config` must be an object, got a function instead. Did you mean `createParser({onEvent: fn})`?', + ); + } + + const { maxBufferSize, onComment, onError, onEvent, onId, onRetry } = config; + + // Trailing bytes from prior `feed()` calls that did not yet form a complete line. + // Stored as an array of fragments and only joined when a line terminator arrives. + // Concatenating per-feed (`prefix + chunk`) is O(N²) when a single SSE line spans + // many chunks (e.g. a large `data:` payload streamed in tiny slices, or an MCP-style + // server that emits one giant content block). Buffering as fragments + joining once + // makes the same workload linear. + const pendingFragments: string[] = []; + + // Running total of `pendingFragments` lengths, kept in sync with the array so the + // `maxBufferSize` check doesn't have to walk the fragment list on every feed. + let pendingFragmentsLength = 0; + + // Empty or partial leading BOM; undefined once the BOM check is complete. + let bomPrefix: string | undefined = ''; + let id: string | undefined; + let data = ''; + let dataLines = 0; + let eventType: string | undefined; + + // Set after a `maxBufferSize` overflow. Once tripped, `feed()` throws until + // `reset()` is called — see the comment on `maxBufferSize` in `ParserConfig`. + let terminated = false; + let skippingLine = false; + + // Set when a line (parsed or discarded) was terminated by a trailing `\r` at the very + // end of a chunk. That `\r` is ambiguous: it may be a bare-CR terminator or the first + // half of a `\r\n` whose `\n` lands in the next chunk. The line itself is complete + // either way, but we must remember to swallow a single leading `\n` from the next + // chunk so the pair is treated as one terminator rather than a blank line (which + // would dispatch an event prematurely). + let skipNextLineFeed = false; + + function dispatchEvent() { + if (id !== undefined) { + onId?.(id); + } + + if (dataLines > 0) { + onEvent?.({ + id, + event: eventType, + data, + }); + } + + id = undefined; + data = ''; + dataLines = 0; + eventType = undefined; + } + + function processField(field: string, value: string, line: string) { + // Field names must be compared literally, with no case folding performed. + switch (field) { + case 'event': + // Set the `event type` buffer to field value + eventType = value || undefined; + break; + case 'data': + data = dataLines === 0 ? value : `${data}\n${value}`; + dataLines++; + break; + case 'id': + // If the field value does not contain U+0000 NULL, then set the `ID` buffer to + // the field value. Otherwise, ignore the field. + if (!value.includes('\0')) id = value; + break; + case 'retry': + // If the field value consists of only ASCII digits, then interpret the field value as an + // integer in base ten, and set the event stream's reconnection time to that integer. + // Otherwise, ignore the field. + if (/^\d+$/.test(value)) { + onRetry?.(parseInt(value, 10)); + } else { + onError?.( + new ParseError(`Invalid \`retry\` value: "${value}"`, { + type: 'invalid-retry', + value, + line, + }), + ); + } + break; + default: + // Otherwise, the field is ignored. + onError?.( + new ParseError( + `Unknown field "${field.length > 20 ? `${field.slice(0, 20)}…` : field}"`, + { + type: 'unknown-field', + field, + value, + line, + }, + ), + ); + break; + } + } + + function parseLine(chunk: string, start: number, end: number) { + if (start === end) { + dispatchEvent(); + return; + } + + const firstCharCode = chunk.charCodeAt(start); + + if (isDataPrefix(chunk, start, firstCharCode)) { + // 'data:'.length === 5, 'data: '.length === 6 + const valueStart = chunk.charCodeAt(start + 5) === SPACE ? start + 6 : start + 5; + const value = chunk.slice(valueStart, end); + data = dataLines === 0 ? value : `${data}\n${value}`; + dataLines++; + return; + } + + if (isEventPrefix(chunk, start, firstCharCode)) { + // 'event:'.length === 6, 'event: '.length === 7 + eventType = + chunk.slice(chunk.charCodeAt(start + 6) === SPACE ? start + 7 : start + 6, end) || + undefined; + return; + } + + // Fast path for "id:" — 'i' = 105, 'd' = 100, ':' = 58 + if ( + firstCharCode === 105 && + chunk.charCodeAt(start + 1) === 100 && + chunk.charCodeAt(start + 2) === 58 + ) { + // 'id:'.length === 3, 'id: '.length === 4 + const value = chunk.slice(chunk.charCodeAt(start + 3) === SPACE ? start + 4 : start + 3, end); + // If the field value does not contain U+0000 NULL, then set the `ID` buffer to + // the field value. Otherwise, ignore the field. + if (!value.includes('\0')) id = value; + return; + } + + // Comment line — ':' = 58 + if (firstCharCode === 58) { + if (onComment) { + const line = chunk.slice(start, end); + // skip ':' (+1), or ': ' (+2) when a space follows + onComment(line.slice(chunk.charCodeAt(start + 1) === SPACE ? 2 : 1)); + } + return; + } + + const line = chunk.slice(start, end); + const fieldSeparatorIndex = line.indexOf(':'); + if (fieldSeparatorIndex === -1) { + processField(line, '', line); + return; + } + + const field = line.slice(0, fieldSeparatorIndex); + // skip ':' (+1), or ': ' (+2) when a space follows + const offset = line.charCodeAt(fieldSeparatorIndex + 1) === SPACE ? 2 : 1; + const value = line.slice(fieldSeparatorIndex + offset); + processField(field, value, line); + } + + /** + * Splits `chunk` into SSE lines and dispatches each to the appropriate handler. + * Returns any trailing bytes that did not terminate with a line break, so the + * caller can prepend them to the next chunk. + * + * The SSE spec permits three line terminators: `\n`, `\r`, and `\r\n`. Real-world + * streams almost always use plain `\n`, so we take a fast path when no `\r` is + * present in the chunk. The slow path is spec-correct but does more work per line. + */ + function processLines(chunk: string): string { + let searchIndex = 0; + + // Fast path: LF-only chunk (the common case for typical SSE servers). + // We can scan forward with a single `indexOf('\n')` per line and inline + // the hot-path branches for `data:` and `event:` without the CR bookkeeping + // the slow path needs. + if (chunk.indexOf('\r') === -1) { + let lfIndex = chunk.indexOf('\n', searchIndex); + while (lfIndex !== -1) { + // Blank line: end-of-event marker. Dispatch the accumulated event (if any) + // and reset the buffered fields. This is hoisted out of `parseLine` because + // it's the single most common line shape after `data:` lines. + if (searchIndex === lfIndex) { + if (id !== undefined) { + onId?.(id); + } + if (dataLines > 0) { + onEvent?.({ id, event: eventType, data }); + } + id = undefined; + data = ''; + dataLines = 0; + eventType = undefined; + searchIndex = lfIndex + 1; + lfIndex = chunk.indexOf('\n', searchIndex); + continue; + } + const firstCharCode = chunk.charCodeAt(searchIndex); + if (isDataPrefix(chunk, searchIndex, firstCharCode)) { + // `data:` line — append the value to the event's data buffer. + // 'data:'.length === 5, 'data: '.length === 6 + const valueStart = + chunk.charCodeAt(searchIndex + 5) === SPACE ? searchIndex + 6 : searchIndex + 5; + const value = chunk.slice(valueStart, lfIndex); + // Fast path within a fast path: if this is the first data line AND the + // next char is another LF (i.e. `data:foo\n\n`), dispatch immediately + // without ever writing to the `data` buffer. This is the shape of a + // typical single-line SSE event (ChatGPT-style streams, etc.) and is + // hot enough to be worth the duplication. + if (dataLines === 0 && chunk.charCodeAt(lfIndex + 1) === LF) { + if (id !== undefined) { + onId?.(id); + } + onEvent?.({ id, event: eventType, data: value }); + id = undefined; + data = ''; + eventType = undefined; + searchIndex = lfIndex + 2; + lfIndex = chunk.indexOf('\n', searchIndex); + continue; + } + // Multi-line data: concatenate with newline separator per spec. + data = dataLines === 0 ? value : `${data}\n${value}`; + dataLines++; + } else if (isEventPrefix(chunk, searchIndex, firstCharCode)) { + // `event:` line — set the event type for the next dispatch. Per spec, + // an empty value resets `event type` to its default (undefined here). + // 'event:'.length === 6, 'event: '.length === 7 + eventType = + chunk.slice( + chunk.charCodeAt(searchIndex + 6) === SPACE ? searchIndex + 7 : searchIndex + 6, + lfIndex, + ) || undefined; + } else { + // Everything else: `id:`, `retry:`, comment lines (`:` prefix), unknown + // fields, or malformed lines. These are rarer and go through the full + // per-line parser, which handles the SSE field grammar in detail. + parseLine(chunk, searchIndex, lfIndex); + } + searchIndex = lfIndex + 1; + lfIndex = chunk.indexOf('\n', searchIndex); + } + return chunk.slice(searchIndex); + } + + // Slow path: the chunk contains at least one `\r`, so lines may be terminated + // by `\r`, `\n`, or `\r\n`. We locate the next terminator by looking at both + // the nearest `\r` and `\n` and picking whichever comes first. + while (searchIndex < chunk.length) { + const crIndex = chunk.indexOf('\r', searchIndex); + const lfIndex = chunk.indexOf('\n', searchIndex); + + let lineEnd = -1; + if (crIndex !== -1 && lfIndex !== -1) { + lineEnd = crIndex < lfIndex ? crIndex : lfIndex; + } else if (crIndex !== -1) { + // A trailing `\r` at the very end of the chunk is ambiguous: it could be + // a bare-CR terminator, or the first half of a `\r\n` whose `\n` arrives + // in the next chunk. Defer until we see more input. + if (crIndex === chunk.length - 1) { + lineEnd = -1; + } else { + lineEnd = crIndex; + } + } else if (lfIndex !== -1) { + lineEnd = lfIndex; + } + + if (lineEnd === -1) { + break; + } + + parseLine(chunk, searchIndex, lineEnd); + searchIndex = lineEnd + 1; + // If we just consumed a `\r` and the next char is `\n`, skip it so the + // pair is treated as a single terminator rather than an empty line. + if (chunk.charCodeAt(searchIndex - 1) === CR && chunk.charCodeAt(searchIndex) === LF) { + searchIndex++; + } + } + + return chunk.slice(searchIndex); + } + + function shouldBufferTrailing(trailing: string) { + const firstCharCode = trailing.charCodeAt(0); + return ( + (firstCharCode === 58 && !!onComment) || + (firstCharCode === 100 && isPotentialField(trailing, 'data')) || + (firstCharCode === 101 && isPotentialField(trailing, 'event')) || + (firstCharCode === 105 && isPotentialField(trailing, 'id')) || + (firstCharCode === 114 && isPotentialField(trailing, 'retry')) + ); + } + + function storeTrailing(trailing: string) { + if (!trailing) return; + + // A trailing that ends with `\r` is not a partial line: the `\r` terminates it, + // and only the CRLF-vs-CR ambiguity remains (the matching `\n` may start the next + // chunk). Parse the completed line now — buffering it could grow unbounded through + // the no-terminator append path, and discarding it would eat the terminator and + // make skip mode swallow the next line. + if (trailing.charCodeAt(trailing.length - 1) === CR) { + parseLine(trailing, 0, trailing.length - 1); + skipNextLineFeed = true; + return; + } + + if (shouldBufferTrailing(trailing)) { + pendingFragments.push(trailing); + pendingFragmentsLength = trailing.length; + return; + } + + skippingLine = true; + } + + function checkBufferSize() { + if (maxBufferSize === undefined) return; + if (pendingFragmentsLength + data.length <= maxBufferSize) return; + + terminated = true; + pendingFragments.length = 0; + pendingFragmentsLength = 0; + id = undefined; + data = ''; + dataLines = 0; + eventType = undefined; + skippingLine = false; + skipNextLineFeed = false; + onError?.( + new ParseError(`Buffered data exceeded max buffer size of ${maxBufferSize} characters`, { + type: 'max-buffer-size-exceeded', + }), + ); + } + + // Out-of-line handler for the rare post-boundary states. `skipNextLineFeed` means a + // line (parsed or discarded) ended with a trailing `\r` at the previous chunk boundary, + // so a single leading `\n` must be swallowed to consume a split `\r\n` as one terminator. + // `skippingLine` means we are discarding an invalid line until its terminator arrives. + // The two states are mutually exclusive. Returns the remaining chunk to parse ('' if + // this chunk was fully consumed). + function resumeAfterSkip(chunk: string): string { + if (chunk.length === 0) { + return chunk; + } + + if (skipNextLineFeed) { + skipNextLineFeed = false; + return chunk.charCodeAt(0) === LF ? chunk.slice(1) : chunk; + } + + const crIndex = chunk.indexOf('\r'); + const lfIndex = chunk.indexOf('\n'); + const lineEnd = + crIndex === -1 ? lfIndex : lfIndex === -1 ? crIndex : crIndex < lfIndex ? crIndex : lfIndex; + + if (lineEnd === -1) { + return ''; + } + + // Trailing `\r` at the very end of the chunk: defer the CRLF/CR decision to the + // next chunk (see `skipNextLineFeed`) rather than greedily consuming just the `\r`. + if (lineEnd === chunk.length - 1 && chunk.charCodeAt(lineEnd) === CR) { + skippingLine = false; + skipNextLineFeed = true; + return ''; + } + + skippingLine = false; + return chunk.slice( + lineEnd + (chunk.charCodeAt(lineEnd) === CR && chunk.charCodeAt(lineEnd + 1) === LF ? 2 : 1), + ); + } + + /** + * Feeds a chunk of the SSE stream to the parser. Any trailing bytes that do + * not yet form a complete line are held back and prepended to the next chunk, + * so callers can pass arbitrary slices of the stream without worrying about + * line boundaries. + * + * Per the SSE spec, one leading UTF-8 BOM is stripped before parsing, + * even when split across chunks. This handles both the raw 3-byte form (0xEF 0xBB + * 0xBF) and a single decoded U+FEFF, so a leading BOM is ignored regardless of + * how the caller decoded the bytes. + * + * @see https://html.spec.whatwg.org/multipage/server-sent-events.html#parsing-an-event-stream + */ + function feed(chunk: string) { + if (terminated) { + throw new Error( + 'Cannot feed parser: it was terminated after exceeding the configured max buffer size. Call `reset()` to resume parsing.', + ); + } + + // The repository lint rules do not allow assignment to a parameter. The mutations below use a + // local copy. + let input = chunk; + + if (bomPrefix !== undefined) { + input = bomPrefix + input; + // Wait only while the input could still be a raw BOM. Empty chunks must + // not finish the check, and a mismatched prefix must remain ordinary input. + if (input === '' || input === '\xEF' || input === '\xEF\xBB') { + bomPrefix = input; + return; + } + bomPrefix = undefined; + // Strip exactly one leading BOM, in decoded or raw byte-valued form. + input = input.replace(/^(?:\uFEFF|\xEF\xBB\xBF)/, ''); + } + + // Rare resume states from a prior chunk boundary: a pending `\r\n` split or a + // line being discarded. Single combined check keeps the hot path to one branch; + // the actual handling lives out-of-line in `resumeAfterSkip`. + if (skippingLine || skipNextLineFeed) { + input = resumeAfterSkip(input); + if (!input) { + return; + } + } + + // Hot path: no buffered prefix from a prior partial line. Hand the chunk + // straight to `processLines`, exactly like the original implementation. + // Zero new work in the common case (every chunk ends with `\n\n`). + if (!pendingFragments.length) { + const trailing = processLines(input); + if (trailing !== '') { + storeTrailing(trailing); + } + checkBufferSize(); + return; + } + + // We have a buffered prefix. If this chunk also has no terminator, append + // to the buffer without concatenating — that's the O(N²) trap we're + // avoiding (large single `data:` payload split across many tiny chunks). + if (input.indexOf('\n') === -1 && input.indexOf('\r') === -1) { + if (pendingFragmentsLength < MAX_FIELD_PREFIX_LENGTH) { + const head = + pendingFragments.join('') + + input.slice(0, MAX_FIELD_PREFIX_LENGTH - pendingFragmentsLength); + if (!shouldBufferTrailing(head)) { + pendingFragments.length = 0; + pendingFragmentsLength = 0; + skippingLine = true; + return; + } + } + pendingFragments.push(input); + pendingFragmentsLength += input.length; + checkBufferSize(); + return; + } + + // Terminator arrived. Join the accumulated fragments + this chunk once, + // process, and buffer any new trailing partial line. + pendingFragments.push(input); + const joined = pendingFragments.join(''); + pendingFragments.length = 0; + pendingFragmentsLength = 0; + storeTrailing(processLines(joined)); + checkBufferSize(); + } + + function reset(options: { consume?: boolean } = {}) { + if (options.consume && pendingFragments.length > 0) { + const incompleteLine = pendingFragments.join(''); + parseLine(incompleteLine, 0, incompleteLine.length); + } + + bomPrefix = ''; + id = undefined; + data = ''; + dataLines = 0; + eventType = undefined; + pendingFragments.length = 0; + pendingFragmentsLength = 0; + terminated = false; + skippingLine = false; + skipNextLineFeed = false; + } + + return { feed, reset }; +} diff --git a/packages/shared/eventsource/src/parser/types.ts b/packages/shared/eventsource/src/parser/types.ts new file mode 100644 index 0000000000..71ad713ec9 --- /dev/null +++ b/packages/shared/eventsource/src/parser/types.ts @@ -0,0 +1,145 @@ +/** + * @remark + * This implementation is derived from `types.ts` of the `eventsource-parser` package, + * version 4.1.1 (https://github.com/rexxars/eventsource-parser). See the LICENSE file at the + * package root for the original license terms. + */ + +import type { ParseError } from './errors'; + +/** + * EventSource parser instance. + * + * Needs to be reset between reconnections/when switching data source, using the `reset()` method. + * + * @public + */ +export interface EventSourceParser { + /** + * Feeds the parser another chunk. The method _does not_ return a parsed message. + * Instead, callbacks passed when creating the parser will be triggered once we see enough data + * for a valid/invalid parsing step (see {@link ParserCallbacks}). + * + * @param chunk - The chunk to parse. Can be a partial, eg in the case of streaming messages. + * @public + */ + feed(chunk: string): void; + + /** + * Resets the parser state. This is required when you have a new stream of messages - + * for instance in the case of a client being disconnected and reconnecting. + * + * Previously received, incomplete data will NOT be parsed unless you pass `consume: true`, + * which tells the parser to attempt to consume any incomplete data as if it ended with a newline + * character. This is useful for cases when a server sends a non-EventSource message that you + * want to be able to react to in an `onError` callback. + * + * @public + */ + reset(options?: { consume?: boolean }): void; +} + +/** + * A parsed EventSource message event + * + * @public + */ +export interface EventSourceMessage { + /** + * The event type sent from the server. Note that this differs from the browser `EventSource` + * implementation in that browsers will default this to `message`, whereas this parser will + * leave this as `undefined` if not explicitly declared. + */ + event?: string | undefined; + + /** + * ID of the message, if any was provided by the server. Can be used by clients to keep the + * last received message ID in sync when reconnecting. + */ + id?: string | undefined; + + /** + * The data received for this message + */ + data: string; +} + +/** + * Callbacks that can be passed to the parser to handle different types of parsed messages + * and errors. + * + * @public + */ +export interface ParserCallbacks { + /** + * Callback for an `id` field when a blank line ends an event block. + * + * Called once with the last valid `id` value in the block, whether or not the block contains a + * `data` field. When the block produces an event, `onId` is called before `onEvent`. An empty + * `id` field is reported as an empty string. An `id` field containing U+0000 is ignored. + * + * @param id - The value of the last valid `id` field in the block + */ + onId?: ((id: string) => void) | undefined; + + /** + * Callback for when a new event/message is parsed from the stream. + * This is the main callback that clients will use to handle incoming messages. + * + * @param event - The parsed event/message + */ + onEvent?: ((event: EventSourceMessage) => void) | undefined; + + /** + * Callback for when the server sends a new reconnection interval through the `retry` field. + * + * @param retry - The number of milliseconds to wait before reconnecting. + */ + onRetry?: ((retry: number) => void) | undefined; + + /** + * Callback for when a comment is encountered in the stream. + * + * @param comment - The comment encountered in the stream. + */ + onComment?: ((comment: string) => void) | undefined; + + /** + * Callback for when an error occurs during parsing. This is a catch-all for any errors + * that occur during parsing, and can be used to handle them in a custom way. Most clients + * tend to silently ignore any errors and instead retry, but it can be helpful to log/debug. + * + * Incomplete lines that cannot become valid SSE fields may be discarded before completion + * to avoid unbounded buffering. Those discarded lines do not trigger `onError`. + * + * @param error - The error that occurred during parsing + */ + onError?: ((error: ParseError) => void) | undefined; +} + +/** + * Configuration accepted by {@link createParser}. Extends {@link ParserCallbacks} with + * additional options that control parser behavior. + * + * @public + */ +export interface ParserConfig extends ParserCallbacks { + /** + * Maximum number of characters the parser is allowed to buffer across calls to `feed()`. + * + * Two buffered surfaces exist in a streaming SSE parser: + * - A valid partial line that has not yet been terminated by `\n`, `\r`, or `\r\n`. + * - A multi-line event whose terminating blank line has not yet arrived (each `data:` + * field gets appended to the buffered event). + * + * Lines that cannot become valid SSE fields are discarded without buffering. + * + * When the combined size of these buffers exceeds `maxBufferSize`, the parser emits a + * `ParseError` with `type: 'max-buffer-size-exceeded'` to `onError` and becomes + * terminated — subsequent calls to `feed()` will throw until `reset()` is called. + * This protects against unbounded memory growth from malformed or malicious streams. + * + * @defaultValue `undefined` (unbounded) + */ + maxBufferSize?: number | undefined; +} From f1ea78ca101c8654b18c2fb63f1d8c2a518241ae Mon Sep 17 00:00:00 2001 From: joker23 <2494686+joker23@users.noreply.github.com> Date: Thu, 1 Oct 2026 11:41:07 -0400 Subject: [PATCH 3/5] Update packages/shared/eventsource/src/Event.ts Co-authored-by: Todd Anderson <127344469+tanderson-ld@users.noreply.github.com> --- packages/shared/eventsource/src/Event.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/shared/eventsource/src/Event.ts b/packages/shared/eventsource/src/Event.ts index eedce7fe7d..f59e5c43f7 100644 --- a/packages/shared/eventsource/src/Event.ts +++ b/packages/shared/eventsource/src/Event.ts @@ -31,7 +31,7 @@ export interface OpenEvent { /** * The payload of the `retrying` event. */ -export interface RetryingEvent { +export interface RetryEvent { readonly type?: string; readonly delayMillis: number; } From 20a7e07337cb9e1293e9db779418a62955eb8add Mon Sep 17 00:00:00 2001 From: joker23 <2494686+joker23@users.noreply.github.com> Date: Thu, 1 Oct 2026 11:41:17 -0400 Subject: [PATCH 4/5] Update packages/shared/eventsource/src/Event.ts Co-authored-by: Todd Anderson <127344469+tanderson-ld@users.noreply.github.com> --- packages/shared/eventsource/src/Event.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/shared/eventsource/src/Event.ts b/packages/shared/eventsource/src/Event.ts index f59e5c43f7..9091f54f6e 100644 --- a/packages/shared/eventsource/src/Event.ts +++ b/packages/shared/eventsource/src/Event.ts @@ -63,7 +63,7 @@ export interface EventSourceEventMap { error: ErrorEvent | MessageEvent; message: MessageEvent; open: OpenEvent; - retrying: RetryingEvent; + retry: RetryEvent; } /** From abf8a883028b50b493a7866b898667d18b062598 Mon Sep 17 00:00:00 2001 From: Steven Zhang Date: Thu, 1 Oct 2026 12:07:34 -0400 Subject: [PATCH 5/5] cap the server-directed retry values to one hour --- .../eventsource/__tests__/parser/parse.test.ts | 12 ++++++++++++ packages/shared/eventsource/src/Event.ts | 2 +- packages/shared/eventsource/src/parser/parse.ts | 15 ++++++++++----- packages/shared/eventsource/src/types.ts | 2 +- 4 files changed, 24 insertions(+), 7 deletions(-) diff --git a/packages/shared/eventsource/__tests__/parser/parse.test.ts b/packages/shared/eventsource/__tests__/parser/parse.test.ts index 14939d83e7..a13d293185 100644 --- a/packages/shared/eventsource/__tests__/parser/parse.test.ts +++ b/packages/shared/eventsource/__tests__/parser/parse.test.ts @@ -130,6 +130,18 @@ it('forwards an all-digit retry value', () => { expect(retries).toEqual([2500]); }); +it('caps a retry value above one hour at one hour', () => { + const { parser, retries } = recordingParser(); + parser.feed('retry: 3600001\n\n'); + expect(retries).toEqual([3600000]); +}); + +it('caps an overlong retry value that parses to Infinity at one hour', () => { + const { parser, retries } = recordingParser(); + parser.feed(`retry: ${'9'.repeat(400)}\n\n`); + expect(retries).toEqual([3600000]); +}); + it('reports a retry value that is not all digits through onError and does not forward it', () => { const { parser, retries, errors } = recordingParser(); parser.feed('retry: 12abc\n\n'); diff --git a/packages/shared/eventsource/src/Event.ts b/packages/shared/eventsource/src/Event.ts index 9091f54f6e..b63a05f8ae 100644 --- a/packages/shared/eventsource/src/Event.ts +++ b/packages/shared/eventsource/src/Event.ts @@ -63,7 +63,7 @@ export interface EventSourceEventMap { error: ErrorEvent | MessageEvent; message: MessageEvent; open: OpenEvent; - retry: RetryEvent; + retrying: RetryEvent; } /** diff --git a/packages/shared/eventsource/src/parser/parse.ts b/packages/shared/eventsource/src/parser/parse.ts index 78e702ac35..b0a363d6e8 100644 --- a/packages/shared/eventsource/src/parser/parse.ts +++ b/packages/shared/eventsource/src/parser/parse.ts @@ -2,9 +2,9 @@ * @remark * This implementation is derived from `parse.ts` of the `eventsource-parser` package, * version 4.1.1 (https://github.com/rexxars/eventsource-parser). See the LICENSE file at the - * package root for the original license terms. The logic is unchanged. The file only reorders - * function declarations and avoids parameter reassignment to satisfy this repository's lint - * rules. + * package root for the original license terms. The logic is unchanged, except that a `retry:` + * value is capped at one hour. The file also reorders function declarations and avoids parameter + * reassignment to satisfy this repository's lint rules. */ /** * EventSource/Server-Sent Events parser @@ -19,6 +19,11 @@ const CR = 13; const SPACE = 32; const MAX_FIELD_PREFIX_LENGTH = 6; +// Upper bound on a server-directed `retry:` value. The cap keeps a server from pushing the +// reconnection delay past one hour. It also keeps an overlong digit string, which parses to +// Infinity, from reaching the callback. +const MAX_RETRY_DELAY_MILLIS = 3_600_000; + /** * Checks if `chunk` starts with the literal `data:` at index `i`. * @@ -158,9 +163,9 @@ export function createParser(config: ParserConfig): EventSourceParser { case 'retry': // If the field value consists of only ASCII digits, then interpret the field value as an // integer in base ten, and set the event stream's reconnection time to that integer. - // Otherwise, ignore the field. + // Otherwise, ignore the field. The value is capped at MAX_RETRY_DELAY_MILLIS. if (/^\d+$/.test(value)) { - onRetry?.(parseInt(value, 10)); + onRetry?.(Math.min(parseInt(value, 10), MAX_RETRY_DELAY_MILLIS)); } else { onError?.( new ParseError(`Invalid \`retry\` value: "${value}"`, { diff --git a/packages/shared/eventsource/src/types.ts b/packages/shared/eventsource/src/types.ts index 33ae0b4380..d70815325f 100644 --- a/packages/shared/eventsource/src/types.ts +++ b/packages/shared/eventsource/src/types.ts @@ -10,7 +10,7 @@ export type { EventSourceEventMap, MessageEvent, OpenEvent, - RetryingEvent, + RetryEvent, } from './Event'; import type { ErrorEvent } from './Event';