From 4fd99c5eadff23fdf71e9dbcb35a35f8e6cf3113 Mon Sep 17 00:00:00 2001 From: Eric Andrechek Date: Thu, 13 Aug 2026 03:36:27 -0400 Subject: [PATCH 01/49] feat(sdk)!: stream over fetch so SSE authenticates by header MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replaces the EventSource-based SSE transport with one built on fetch, so the JWT travels as `Authorization: Bearer` instead of `?token=` in the request URI, where every proxy and CDN in front of WaveHouse could log it. The server already preferred the header and already allow-listed it (plus Last-Event-ID) in the CORS preflight, so nothing changes server-side; `?token=` stays accepted for clients that cannot set headers. Owning the transport means owning what EventSource provided: framing (rented from eventsource-parser, exact-pinned), jittered reconnect backoff, and Last-Event-ID resumption. Three behavioral consequences: - `auth` is re-read per connection attempt, so a stream outliving its token no longer reconnects forever with the expired one and silently degrades to default_role. Prerequisite for #239. - The last *non-empty* event id is retained; the hub emits a blank `id:` for passthrough payloads, which per spec would clear resumption state. - The Node EventSource polyfill requirement is gone. `options.fetch` / `headers` / `fetchOptions` now reach streams, closing the carve-out documented in #456, and FetchLike's URL parameter narrows to `string` so hand-written middleware assigns — what the type intended when it shipped. Streaming needs a readable response body, which the type cannot express, so the transport fails with SSE_NO_STREAM_BODY instead of hanging on a fetch wrapper that buffers. Closes #203. --- CHANGELOG.md | 6 +- clients/ts/package.json | 3 + clients/ts/src/client.ts | 11 +- clients/ts/src/http.ts | 7 +- clients/ts/src/stream/sse.test.ts | 521 +++++++++++++++++++++--- clients/ts/src/stream/sse.ts | 384 ++++++++++++++--- clients/ts/src/types.ts | 65 ++- docs/src/content/docs/api.md | 4 +- docs/src/content/docs/reverse-proxy.mdx | 8 +- docs/src/content/docs/sdk/index.mdx | 74 ++-- docs/src/content/docs/sdk/reference.md | 17 +- docs/src/content/docs/sdk/streaming.md | 10 +- pnpm-lock.yaml | 15 +- tests/e2e/sdk/package.json | 1 - tests/e2e/sdk/polyfills.ts | 6 - tests/e2e/sdk/vitest.config.ts | 5 +- 16 files changed, 923 insertions(+), 214 deletions(-) delete mode 100644 tests/e2e/sdk/polyfills.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 82199699..69d0e5b1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,7 +11,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added -- **HTTP customization for the SDK — `options.headers`, `options.fetchOptions`, and `options.fetch`** (`clients/ts/src/types.ts`, `clients/ts/src/client.ts`, `clients/ts/src/http.ts`, `clients/ts/src/index.ts`, `clients/ts/src/client.test.ts`, `docs/src/content/docs/sdk/index.mdx`): closes #269. `ClientOptions` exposed only `maxRetries`, so a WaveHouse behind a header-gated proxy (Cloudflare Access, an mTLS sidecar, an auth gateway) or a cookie-authenticated origin was unreachable from the SDK — a defense-in-depth gate forced consumers off the client entirely. Three knobs land together, shaped after the conventions in Supabase's, OpenAI's, and Anthropic's clients rather than invented here. **`headers`** adds static headers to every REST request; names match case-insensitively as HTTP requires, and they apply *underneath* the SDK's own — `auth` keeps `Authorization`, and a request's `Content-Type`/`Accept` can't be displaced by a global one, since a header joined rather than replaced is how you ship `Content-Type: application/json, image/png`. **`fetchOptions`** merges extra `RequestInit` fields (`credentials: "include"` for the cookie case, `mode`, `cache`, or a runtime extension like Next.js's `next: { tags }`); the fields the SDK controls — `method`, `headers`, `body`, `signal` — always win, so it cannot corrupt the request. **`fetch`** replaces the HTTP implementation outright. All three are REST-only: `.stream()` and `.liveQuery()`'s live connection go through `EventSource`, which accepts neither headers nor a `fetch`, so a header-gated deployment can query but not stream until #203 changes that transport — stated plainly in the docs rather than left implicit, since the equivalent gap in Supabase's realtime client was found the hard way by a user whose RLS policies silently stopped matching. `.liveQuery()`'s initial backfill is an ordinary REST call and *is* covered. Per-call overrides and dynamic header callbacks are deliberately deferred to #459; the per-call slot (`.fetch(opts)`) already exists, so adding them later is additive. The exported `FetchLike` matches the standard `fetch` signature, written out rather than as `typeof fetch` because that resolves differently depending on whether the consumer's TypeScript `lib` includes DOM — the same fragility behind Supabase's long tail of `node-fetch` resolution issues. `options.fetch` accepts any `fetch`-compatible function (exported as the `FetchLike` type) and is used for every request, retries included, so middleware sees each attempt. The motivating case is a runtime bug consumers can't fix themselves: undici 8.8.0–8.9.0 stalls a request before it goes out when a keep-alive socket is reused while the event loop is idle ([nodejs/undici#5600](https://github.com/nodejs/undici/issues/5600), fixed in 8.10.0), and Node 26 bundles 8.9.0 — so calls that should take milliseconds don't, with no recourse inside the SDK. Severity varies with the runtime and the idle gap: the upstream report measured ~450–465 ms, and our own runs against an instant-answering server have ranged from ~100 ms to tens of seconds. Upgrading undici is the real fix, and `options.fetch` is how you get it without waiting for a new runtime: install undici yourself and route through it, passing its dispatcher **explicitly**. That last part is load-bearing and not obvious — undici keeps its connection pool on a shared `globalThis` symbol claimed by whichever copy loads first (Node claims it for the bundled copy on the first built-in `fetch` call, not at startup), so calling an installed 8.10.0's `fetch` without a `dispatcher` resolves whatever is on that symbol and can still stall through the bundled 8.9.0's pool. Measured with both copies loaded, 1.5 s idle gaps against a 10 ms server: `21, 1514, 1495, 583 ms` with an implied dispatcher versus `17, 14, 12, 13 ms` with an explicit `new Agent()`. (Tuning `keepAliveTimeout` is *not* a workaround — measured, it changes nothing, because the retirement timer is starved by the same idle event loop; `new Agent({ pipelining: 0 })` does work for anyone pinned to an affected version, at a connection per request.) The same hook covers the ordinary reasons an SDK grows one: proxies, client certificates, tracing or circuit-breaker middleware, and mocking HTTP in a consumer's own tests without monkey-patching a global. `fetch` stays optional all the way through to the internal `HttpContext` rather than being defaulted at construction, so the default path still calls the global directly — that keeps it late-bound (replacing `globalThis.fetch` after a client exists still works, which is what `vi.stubGlobal` does) and avoids invoking a detached `fetch` reference, which is not universally safe; both properties are pinned by tests. Only the SSE transport is exempt: the live connection behind `.stream()`/`.liveQuery()` goes through `EventSource`, which this does not replace — but `.liveQuery()`'s initial backfill is an ordinary request and does go through the supplied function. Implementations shipping their own request/response declarations (undici, `node-fetch`) need casts on the URL argument, the init, and the return value, since those types are separate from the ones behind the global `fetch`; the narrow documented runtime contract is what makes them safe — a string URL and plain `RequestInit` in, and only `.ok`/`.headers` plus `.text()` on success and `.status`/`.statusText`/`.json()` on a non-`ok` response read back. Abort handling is part of that contract: `ABORTED` requires a `DOMException` named `AbortError` (platform `fetch` and undici), and any other rejection is retried as `NETWORK_ERROR`. +- **HTTP customization for the SDK — `options.headers`, `options.fetchOptions`, and `options.fetch`** (`clients/ts/src/types.ts`, `clients/ts/src/client.ts`, `clients/ts/src/http.ts`, `clients/ts/src/index.ts`, `clients/ts/src/client.test.ts`, `docs/src/content/docs/sdk/index.mdx`): closes #269. `ClientOptions` exposed only `maxRetries`, so a WaveHouse behind a header-gated proxy (Cloudflare Access, an mTLS sidecar, an auth gateway) or a cookie-authenticated origin was unreachable from the SDK — a defense-in-depth gate forced consumers off the client entirely. Three knobs land together, shaped after the conventions in Supabase's, OpenAI's, and Anthropic's clients rather than invented here. **`headers`** adds static headers to every REST request; names match case-insensitively as HTTP requires, and they apply *underneath* the SDK's own — `auth` keeps `Authorization`, and a request's `Content-Type`/`Accept` can't be displaced by a global one, since a header joined rather than replaced is how you ship `Content-Type: application/json, image/png`. **`fetchOptions`** merges extra `RequestInit` fields (`credentials: "include"` for the cookie case, `mode`, `cache`, or a runtime extension like Next.js's `next: { tags }`); the fields the SDK controls — `method`, `headers`, `body`, `signal` — always win, so it cannot corrupt the request. **`fetch`** replaces the HTTP implementation outright. All three shipped REST-only, because `.stream()` went through `EventSource`, which accepts neither headers nor a `fetch`; **#203 closed that gap within this same unreleased cycle**, so as released they apply to streaming too — see the entry above for the streaming contract they carry. `.liveQuery()`'s initial backfill is an ordinary REST call and *is* covered. Per-call overrides and dynamic header callbacks are deliberately deferred to #459; the per-call slot (`.fetch(opts)`) already exists, so adding them later is additive. The exported `FetchLike` matches the standard `fetch` signature, written out rather than as `typeof fetch` because that resolves differently depending on whether the consumer's TypeScript `lib` includes DOM — the same fragility behind Supabase's long tail of `node-fetch` resolution issues. `options.fetch` accepts any `fetch`-compatible function (exported as the `FetchLike` type) and is used for every request, retries included, so middleware sees each attempt. The motivating case is a runtime bug consumers can't fix themselves: undici 8.8.0–8.9.0 stalls a request before it goes out when a keep-alive socket is reused while the event loop is idle ([nodejs/undici#5600](https://github.com/nodejs/undici/issues/5600), fixed in 8.10.0), and Node 26 bundles 8.9.0 — so calls that should take milliseconds don't, with no recourse inside the SDK. Severity varies with the runtime and the idle gap: the upstream report measured ~450–465 ms, and our own runs against an instant-answering server have ranged from ~100 ms to tens of seconds. Upgrading undici is the real fix, and `options.fetch` is how you get it without waiting for a new runtime: install undici yourself and route through it, passing its dispatcher **explicitly**. That last part is load-bearing and not obvious — undici keeps its connection pool on a shared `globalThis` symbol claimed by whichever copy loads first (Node claims it for the bundled copy on the first built-in `fetch` call, not at startup), so calling an installed 8.10.0's `fetch` without a `dispatcher` resolves whatever is on that symbol and can still stall through the bundled 8.9.0's pool. Measured with both copies loaded, 1.5 s idle gaps against a 10 ms server: `21, 1514, 1495, 583 ms` with an implied dispatcher versus `17, 14, 12, 13 ms` with an explicit `new Agent()`. (Tuning `keepAliveTimeout` is *not* a workaround — measured, it changes nothing, because the retirement timer is starved by the same idle event loop; `new Agent({ pipelining: 0 })` does work for anyone pinned to an affected version, at a connection per request.) The same hook covers the ordinary reasons an SDK grows one: proxies, client certificates, tracing or circuit-breaker middleware, and mocking HTTP in a consumer's own tests without monkey-patching a global. `fetch` stays optional all the way through to the internal `HttpContext` rather than being defaulted at construction, so the default path still calls the global directly — that keeps it late-bound (replacing `globalThis.fetch` after a client exists still works, which is what `vi.stubGlobal` does) and avoids invoking a detached `fetch` reference, which is not universally safe; both properties are pinned by tests. Implementations shipping their own request/response declarations (undici, `node-fetch`) need casts on the init and the return value, since those types are separate from the ones behind the global `fetch`; the narrow documented runtime contract is what makes them safe — a string URL and plain `RequestInit` in, and only `.ok`/`.headers` plus `.text()` on success and `.status`/`.statusText`/`.json()` on a non-`ok` response read back. Abort handling is part of that contract: `ABORTED` requires a `DOMException` named `AbortError` (platform `fetch` and undici), and any other rejection is retried as `NETWORK_ERROR`. - **Non-JWT operator key for full-access admin + break-glass recovery** (`internal/auth/auth.go`, `internal/auth/context.go`, `internal/auth/auth_test.go`, `internal/api/router.go`, `internal/api/router_test.go`, `internal/config/config.go`, `internal/config/config_test.go`, `cmd/wavehouse/main.go`, `config.yaml`, `deployments/compose/standalone.yaml`, `tests/e2e/fixtures/config.yaml`, `docs/src/content/docs/configuration.mdx`, `docs/src/content/docs/access-control.mdx`, `docs/src/content/docs/api.md`, `docs/src/content/docs/deployment.md`, `docs/src/content/docs/development.md`, `docs/src/content/docs/architecture.md`, `docs/src/content/docs/reverse-proxy.mdx`, `AGENTS.md`, `SECURITY.md`): closes #240; partially advances the auth-hardening epic [#228](https://github.com/Wave-RF/WaveHouse/issues/228) and the management/data-plane split [#359](https://github.com/Wave-RF/WaveHouse/issues/359). Adds an optional `auth.operator_key` (`WH_AUTH_OPERATOR_KEY`): a request presenting it — via an `Authorization: Operator ` header (forwarded verbatim by proxies, no collision with Bearer JWTs) or the `X-Operator-Key` alias — is authorized as a full-access platform operator — the whole data plane *and* the `/v1/admin/*` management surface — without minting a JWT and independently of the token verifier, giving the person running the deployment a role-free credential for bootstrap and break-glass. The middleware checks it before the Bearer token with a constant-time comparison (`crypto/subtle`) and stamps two things into the request context: the live `admin_role` (so the policy evaluator's admin bypass grants unrestricted data-plane access while a policy exists) and a platform-operator bit that `RequireAdmin` honors **even when the policy is `nil`/deleted** — the one HTTP path that can restore a wiped policy, which previously required SSH access and a reboot. Empty (the default) disables it, so existing deployments are unchanged; treat it as an admin secret (load from a secret store, serve only over TLS). A successful operator authentication is audit-logged at `INFO`; a request presenting a *non-matching* operator key is logged at `WARN` and counted by a new `wavehouse_auth_operator_key_failures_total` counter (a probing/brute-force signal on the most privileged credential in the system) before falling through to the normal token/default path — the middleware still never rejects. The `Authorization` auth-scheme is matched case-insensitively (RFC 7235) via a shared `authScheme` helper, which also makes the existing `Bearer` JWT scheme case-insensitive (previously it required the canonical `Bearer` casing). Explicitly out of scope, tracked in #359: scoping the operator credential to the management surface only, and capability-scoped admin permissions. - **Missing-dedupe-id observability + optional strict mode** (`internal/api/ingest.go`, `internal/api/ingest_test.go`, `internal/config/config.go`, `cmd/wavehouse/main.go`, `config.yaml`, `deployments/compose/standalone.yaml`, `docs/src/content/docs/configuration.mdx`, `docs/src/content/docs/api.md`, `docs/src/content/docs/architecture.md`): closes #219. With dedupe enabled, a row missing the configured `id_field` can't be deduped — previously it was published with idempotency silently disabled and *no* log or metric, so a producer bug that dropped the id turned off the guarantee for those rows unnoticed. Now every such row is logged at `WARN` and counted by a new `wavehouse_ingest_dedupe_missing_id_total` counter (labeled by `table`), making the loss observable server-side. A new opt-in `dedupe.require_id` (`WH_DEDUPE_REQUIRE_ID`, default `false`) turns that signal into enforcement: a row missing the id is rejected (`400` for a single insert; a per-record failure in a batch) instead of published — a tripwire for producers that must guarantee the id (complements the client-side [#202](https://github.com/Wave-RF/WaveHouse/issues/202)). Default behavior is unchanged. @@ -25,6 +25,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- **BREAKING (SDK): streaming moves from `EventSource` to `fetch`, so the JWT rides in a header instead of `?token=`** (`clients/ts/src/stream/sse.ts`, `clients/ts/src/stream/sse.test.ts`, `clients/ts/src/client.ts`, `clients/ts/src/http.ts`, `clients/ts/src/types.ts`, `clients/ts/package.json`, `tests/e2e/sdk/vitest.config.ts`, `tests/e2e/sdk/package.json`, `tests/e2e/sdk/polyfills.ts` (deleted), `docs/src/content/docs/sdk/index.mdx`, `docs/src/content/docs/sdk/streaming.md`, `docs/src/content/docs/sdk/reference.md`, `docs/src/content/docs/api.md`, `docs/src/content/docs/reverse-proxy.mdx`): closes #203. `.stream()` and `.liveQuery()`'s live connection used the browser `EventSource` API, which cannot set request headers — so the JWT travelled as `?token=` in the request URI, past every proxy, CDN, and load balancer in front of WaveHouse, each free to log it. The server has always preferred an `Authorization` header (`internal/auth/auth.go`) and its CORS preflight has allow-listed both `Authorization` and `Last-Event-ID` since #215, so nothing changes server-side: `?token=` remains accepted for clients that genuinely can't set headers, and its deprecation is tracked separately. The transport now issues an ordinary `fetch` and parses `text/event-stream` off the response body, which means it owns what `EventSource` used to provide — framing, reconnection, and `Last-Event-ID` resumption. Reconnect uses **jittered** exponential backoff capped at 30s and reset once a connection goes live; the jitter is the deliberate difference from `http.ts`'s REST backoff, because every subscriber to a dropped stream re-dials at the same instant and a fixed schedule synchronizes the fleet into a thundering herd against a server that is already unwell. A `4xx` is terminal — retrying cannot fix a rejected token or a missing table — and errors now carry the server's real status and message instead of `EventSource`'s status-free `onerror`, so an expired token surfaces as `HTTP_401` rather than an opaque retry loop. Three consequences worth calling out. **(1) `auth` is now invoked per connection attempt** rather than once and baked into a URL: previously a stream that outlived its token reconnected forever with the expired one, and because `/v1/stream` never rejects a caller — it resolves them to `default_role` — the stream stayed open and silently served a reduced view instead of failing. That also makes this a prerequisite for #239 (enforcing token expiry on long-lived streams), which would turn the old behavior into a hard break. A token provider that *throws* is treated as transient and retried (`SSE_AUTH_ERROR`) rather than terminating an otherwise healthy stream — the tradeoff that per-attempt auth forces. **(2) The last non-empty event id is retained** for resumption: the hub emits a blank `id:` line for passthrough payloads (`internal/stream/hub.go`), and per the SSE spec an empty `id` field *clears* the last-event-id, so a spec-faithful client would silently discard its resumption point mid-stream and re-open from the beginning. **(3) The Node `EventSource` polyfill requirement is gone** — streaming uses the same `fetch`/`ReadableStream` globals as the rest of the SDK, so `globalThis.EventSource = …` is no longer needed, and the e2e suite's `polyfills.ts` and `eventsource` devDependency are deleted. Framing is parsed by `eventsource-parser@3.1.0` (exact-pinned; MIT, zero transitive dependencies, npm provenance-attested, and dual CJS/ESM so it composes with the SDK's `dist/index.cjs` — 4.0.0 is ESM-only and, at 3 days old, blocked by the workspace's 7-day `minimumReleaseAge` cooldown regardless), the SDK's first runtime dependency, costing ~5 KB raw / ~2 KB gzipped in the CDN IIFE bundle. Renting it rather than hand-rolling is a deliberate call: the correct parser is ~440 lines, and the parts that matter — reassembling a frame split across chunk boundaries without quadratic recopying, disambiguating a trailing `\r` at a chunk boundary between a bare-CR terminator and half a split `\r\n`, and capping buffered input against a malformed or hostile stream — are exactly what a naive implementation gets wrong, and are unavoidable even though we control the server, since HTTP/2 and intermediaries re-chunk freely. The buffer cap is set explicitly to 16 MiB (the parser defaults to unbounded). The transport also keeps ownership of three `RequestInit` fields that `fetchOptions` controls on the REST path, because on a stream they are correctness rather than preference: `cache: "no-store"` **as an init field, never a `Cache-Control` header** (the header form isn't in the server's fixed `Access-Control-Allow-Headers`, so it would fail every cross-origin preflight); `redirect: "error"`, because platforms strip `Authorization` on a cross-origin redirect ([whatwg/fetch#1544](https://github.com/whatwg/fetch/pull/1544), the mitigation for the class of bug behind [CVE-2022-1650](https://github.com/advisories/GHSA-6h5x-7c5m-7cr7)) and this endpoint answers a credential-less caller with a reduced view rather than an error, so following one would silently downgrade the stream instead of failing it; and `credentials`, honored in browsers and dropped elsewhere because some runtimes (Cloudflare Workers) throw if it is set at all. **Migration:** none for SDK consumers on the default transport — `.stream()` and `.liveQuery()` are unchanged. Remove any `globalThis.EventSource` polyfill and the `eventsource` dependency. Deployments whose proxy strips `Authorization` on `/v1/stream`, or redirects that path, must fix the proxy; both previously "worked" by accident. + +- **BREAKING (SDK): `options.fetch`, `options.headers`, and `options.fetchOptions` now reach the streaming transport, and `FetchLike` narrows its URL parameter** (`clients/ts/src/types.ts`, `clients/ts/src/http.ts`, `clients/ts/src/stream/sse.ts`, `docs/src/content/docs/sdk/index.mdx`): the carve-out documented in #456 is closed — all three knobs apply to `.stream()` and `.liveQuery()`, not just REST, which is what makes a header-gated origin streamable and completes #269's story to the extent CORS allows (WaveHouse's own preflight advertises a fixed header set, so custom headers reach it cross-origin only behind a proxy that terminates the preflight; server-side callers never preflight). `mergeHeaders` is shared between the two paths rather than duplicated, so header precedence has one definition: the SDK's own headers outrank configured ones, and `auth` keeps `Authorization`. Two breaking details. **`FetchLike`'s first parameter narrows from `string | URL | Request` to `string`.** Because function parameters are contravariant, narrowing it *widens* the set of assignable implementations — the global `fetch` still fits, and so now does hand-written `(url: string, init?: RequestInit) => Promise` middleware, which the wider spelling rejected at compile time. That was the stated intent when the type shipped; the published signature didn't match it. Only code that *imported* `FetchLike` and called through it with a `URL` or `Request` breaks. **And the runtime contract widens:** streaming reads the response as it arrives via `.body.getReader()`, where REST only ever calls `.text()`. An `options.fetch` that buffers the response — or clones it to log the body, the most common wrapper anyone writes — satisfies every REST call and then hangs forever on a stream that never ends. This cannot be expressed in the type, since `Response.body` is legitimately nullable, so the transport checks for a readable body and fails with `SSE_NO_STREAM_BODY` instead of stalling. Tightened now, nine days after `options.fetch` shipped, precisely because nobody has yet written a non-streaming implementation against the narrower promise. + - **BREAKING (SDK): `PipeRef.fetch` no longer accepts a `limit` it silently ignored** (`clients/ts/src/pipes.ts`, `clients/ts/src/client.test.ts`, `docs/src/content/docs/sdk/pipes.md`, `docs/src/content/docs/sdk/reference.md`): closes #464, raised by CodeRabbit on #456. It took the same per-call options type as the query builder — which carries `limit` — but forwarded only `signal`, so `wh.pipe('top_pages').fetch({ limit: 10 })` type-checked, ran, and quietly returned whatever the pipe's SQL returned. `QueryBuilder.fetch` and `TableRef.fetch` both honour `limit`, so the inconsistency sat inside one shared type. There is nothing to forward: the endpoint binds the request body as the pipe's *parameters* (`internal/api/pipes.go` → `pipes.BindParams`), and a key the SQL doesn't declare is ignored, so a client-side row cap is not something the pipes surface offers. The parameter is now a dedicated `PipeRequestOptions` (exported) declaring `signal?: AbortSignal` and `limit?: never`, making the dead option a compile error rather than a silent no-op. `never` rather than simply omitting `limit`, because omitting it only rejects fresh object literals — TypeScript's excess-property check doesn't apply to a *variable*, so a shared `const opts: RequestOptions` carrying a limit would still have passed and still been dropped, which is the defect rather than a narrower version of it. Both cases are pinned by `@ts-expect-error` tests. **Note the collateral effect**, which is the half most consumers will actually meet: a value *declared* `RequestOptions` no longer assigns to a pipe `.fetch()` at all, even when it carries no limit at runtime, because the declared type permits one and assignability is decided on the type. Type a shared options object as `PipeRequestOptions` — the table and query-builder `.fetch()` accept it too, so it works everywhere — or inline `{ signal }` at the pipe call. Structural wrappers are unaffected: method parameters compare bivariantly, so an `interface Fetchable { fetch(opts?: RequestOptions): … }` is still satisfied by `PipeRef`. **Migration:** declare a `{{limit}}` parameter in the pipe's SQL and pass it as a pipe parameter — `wh.pipe(name, { limit })` — which is what the docs already showed. Pre-existing rather than introduced by #456, folded in there because that PR renames the type in question. - **BREAKING (SDK): `FetchOptions` is renamed `RequestOptions`** (`clients/ts/src/types.ts`, `clients/ts/src/index.ts`, `clients/ts/src/query-builder.ts`, `clients/ts/src/table.ts`, `clients/ts/src/pipes.ts`): the per-call options type accepted by `.fetch()`. The old name collided conceptually with the new `options.fetchOptions` — which, following OpenAI, Anthropic, and the wider ecosystem, means "extra `RequestInit` fields", not "options for our `.fetch()` method". Shipping both would have left `FetchOptions` and `fetchOptions` in the same SDK one capital letter apart, meaning unrelated things. `RequestOptions` is what Anthropic's SDK calls the identical concept. No deprecated alias: the type is unreferenced by anything consuming the pre-1.0 package, and keeping it would preserve exactly the ambiguity the rename removes. Renaming the import is the whole migration for this entry — note the separate `PipeRef.fetch` narrowing above, which is a behavioural break in the same file. The module-private `RequestOptions` in `http.ts` — the internal request descriptor — becomes `RequestSpec` to free the name. diff --git a/clients/ts/package.json b/clients/ts/package.json index b11bbd13..5142c460 100644 --- a/clients/ts/package.json +++ b/clients/ts/package.json @@ -40,6 +40,9 @@ "prepublishOnly": "pnpm run typecheck && pnpm run build", "codegen": "tsx src/cli/codegen.ts" }, + "dependencies": { + "eventsource-parser": "3.1.0" + }, "devDependencies": { "@types/node": "catalog:", "@vitest/coverage-v8": "^4.1.10", diff --git a/clients/ts/src/client.ts b/clients/ts/src/client.ts index bb3987a9..04aa24b9 100644 --- a/clients/ts/src/client.ts +++ b/clients/ts/src/client.ts @@ -95,19 +95,14 @@ export class WaveHouseClient { table: string, opts?: StreamOptions, ): StreamController { - if (typeof EventSource === "undefined") { - // TODO: fallback method? polling? - throw new Error( - "[WaveHouse SDK] Native EventSource is not available in this environment. " + - "Please provide a global polyfill (e.g., `globalThis.EventSource = require('eventsource')`).", - ); - } - const transport = new SSETransport({ baseURL: this._ctx.baseURL, table, since: opts?.since, auth: this._ctx.auth, + fetch: this._ctx.options.fetch, + headers: this._ctx.options.headers, + fetchOptions: this._ctx.options.fetchOptions, }); const controller = new StreamController(transport); if (opts?.signal) controller.attachSignal(opts.signal); diff --git a/clients/ts/src/http.ts b/clients/ts/src/http.ts index 905a8556..141eb6ea 100644 --- a/clients/ts/src/http.ts +++ b/clients/ts/src/http.ts @@ -31,8 +31,13 @@ export interface HttpResult { * `Authorization` and let the server pick. Canonical casing wins, and values * replace rather than append — a header joined instead of replaced is a known * way to produce `Content-Type: application/json, image/png`. + * + * Shared with the SSE transport so both paths resolve `headers` the same way, + * rather than growing a second precedence story. + * + * @internal */ -function mergeHeaders( +export function mergeHeaders( base: Record, extra: Record | undefined, ): Record { diff --git a/clients/ts/src/stream/sse.test.ts b/clients/ts/src/stream/sse.test.ts index 4a51e8c7..01890d37 100644 --- a/clients/ts/src/stream/sse.test.ts +++ b/clients/ts/src/stream/sse.test.ts @@ -1,98 +1,495 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import type { FetchLike, StreamEvent, StreamStatus, WaveHouseError } from "../types.js"; import { SSETransport } from "./sse.js"; -/** URLs handed to `new EventSource(...)`, newest last. */ -let opened: string[] = []; +/** One recorded call into the injected fetch. */ +interface Attempt { + url: string; + init: RequestInit; +} -class FakeEventSource { - static readonly CONNECTING = 0; - static readonly OPEN = 1; - static readonly CLOSED = 2; +/** Headers are always handed to fetch as a plain object by the transport. */ +function headersOf(attempt: Attempt): Record { + return (attempt.init.headers ?? {}) as Record; +} - readyState = FakeEventSource.CONNECTING; - onopen: (() => void) | null = null; - onmessage: ((e: MessageEvent) => void) | null = null; - onerror: (() => void) | null = null; +/** + * A response whose body is a stream the test drives frame by frame, standing in + * for a connection the server holds open. + */ +function streamingResponse(): { + res: Response; + push: (chunk: string) => void; + close: () => void; +} { + const encoder = new TextEncoder(); + let ctrl!: ReadableStreamDefaultController; + const body = new ReadableStream({ + start(c) { + ctrl = c; + }, + }); + return { + res: new Response(body, { + status: 200, + headers: { "Content-Type": "text/event-stream" }, + }), + push: (chunk) => ctrl.enqueue(encoder.encode(chunk)), + close: () => ctrl.close(), + }; +} - constructor(url: string) { - opened.push(url); - } +/** A scripted fetch: each queued entry answers one connection attempt. */ +function makeFetch() { + const attempts: Attempt[] = []; + const queue: Array<() => Response> = []; + const impl: FetchLike = async (url, init) => { + attempts.push({ url, init: init ?? {} }); + const next = queue.shift(); + if (!next) return streamingResponse().res; // idle, never-ending + return next(); + }; + return { impl, attempts, queue }; +} - close(): void { - this.readyState = FakeEventSource.CLOSED; - } +/** Collector wired to a transport's three callbacks. */ +function collect(t: SSETransport) { + const events: StreamEvent[] = []; + const statuses: StreamStatus[] = []; + const errors: WaveHouseError[] = []; + t.onEvent = (e) => events.push(e); + t.onStatus = (s) => statuses.push(s); + t.onError = (e) => errors.push(e); + return { events, statuses, errors }; } -/** Let `connect()`'s async `_doConnect()` reach `new EventSource(...)`. */ +const BASE = "http://localhost:8080"; + +/** Let the transport's async connect path reach its next await. */ const flush = () => new Promise((r) => setTimeout(r, 0)); -describe("SSETransport URL construction", () => { - beforeEach(() => { - opened = []; - vi.stubGlobal("EventSource", FakeEventSource); - }); +const frame = (id: string, payload: unknown) => `id: ${id}\ndata: ${JSON.stringify(payload)}\n\n`; - afterEach(() => { - vi.unstubAllGlobals(); - }); - - it("connects at the origin root for a root-hosted base", async () => { - new SSETransport({ baseURL: "http://localhost:8080", table: "clicks" }).connect(); - await flush(); +describe("SSETransport request construction", () => { + it("authenticates with a Bearer header and keeps the token out of the URL", async () => { + const f = makeFetch(); + const t = new SSETransport({ + baseURL: BASE, + table: "clicks", + auth: () => "jwt-abc", + fetch: f.impl, + }); + t.connect(); + await vi.waitFor(() => expect(f.attempts).toHaveLength(1)); - expect(opened).toHaveLength(1); - const url = new URL(opened[0]); + const url = new URL(f.attempts[0].url); expect(url.pathname).toBe("/v1/stream"); expect(url.searchParams.get("table")).toBe("clicks"); + expect(url.searchParams.get("token")).toBeNull(); + expect(url.search).not.toContain("jwt-abc"); + expect(headersOf(f.attempts[0]).Authorization).toBe("Bearer jwt-abc"); + + t.disconnect(); }); it("preserves a base path prefix", async () => { - new SSETransport({ + const f = makeFetch(); + const t = new SSETransport({ baseURL: "https://app.example.com/api/warehouse", table: "clicks", - }).connect(); - await flush(); + fetch: f.impl, + }); + t.connect(); + await vi.waitFor(() => expect(f.attempts).toHaveLength(1)); - const url = new URL(opened[0]); - expect(url.pathname).toBe("/api/warehouse/v1/stream"); - expect(url.searchParams.get("table")).toBe("clicks"); + expect(new URL(f.attempts[0].url).pathname).toBe("/api/warehouse/v1/stream"); + t.disconnect(); + }); + + it("sets the stream-critical init fields the SDK owns", async () => { + const f = makeFetch(); + const t = new SSETransport({ baseURL: BASE, table: "clicks", fetch: f.impl }); + t.connect(); + await vi.waitFor(() => expect(f.attempts).toHaveLength(1)); + + const { init } = f.attempts[0]; + expect(headersOf(f.attempts[0]).Accept).toBe("text/event-stream"); + // As an init field, never a Cache-Control header — the header form is not + // in the server's Access-Control-Allow-Headers and would fail preflight. + expect(init.cache).toBe("no-store"); + expect(headersOf(f.attempts[0])["Cache-Control"]).toBeUndefined(); + // Fail closed rather than follow a redirect that strips Authorization. + expect(init.redirect).toBe("error"); + + t.disconnect(); }); - it("preserves the prefix alongside since and token params", async () => { - new SSETransport({ - baseURL: "https://app.example.com/api/warehouse/", + it("does not set credentials outside a browser, even when configured", async () => { + const f = makeFetch(); + const t = new SSETransport({ + baseURL: BASE, table: "clicks", - since: "2024-01-01T00:00:00Z", - auth: async () => "my-token", - }).connect(); - await flush(); + fetch: f.impl, + fetchOptions: { credentials: "include" }, + }); + t.connect(); + await vi.waitFor(() => expect(f.attempts).toHaveLength(1)); - const url = new URL(opened[0]); - expect(url.pathname).toBe("/api/warehouse/v1/stream"); - expect(url.searchParams.get("since")).toBe("2024-01-01T00:00:00Z"); - expect(url.searchParams.get("token")).toBe("my-token"); + expect(f.attempts[0].init.credentials).toBeUndefined(); + t.disconnect(); }); - it("omits the token param when auth resolves empty", async () => { - new SSETransport({ - baseURL: "https://app.example.com/api/warehouse", + it("merges configured headers but keeps Authorization for auth", async () => { + const f = makeFetch(); + const t = new SSETransport({ + baseURL: BASE, table: "clicks", - auth: async () => "", - }).connect(); - await flush(); + auth: () => "real-token", + headers: { "CF-Access-Client-Id": "svc", authorization: "Bearer smuggled" }, + fetch: f.impl, + }); + t.connect(); + await vi.waitFor(() => expect(f.attempts).toHaveLength(1)); - expect(new URL(opened[0]).searchParams.has("token")).toBe(false); + const headers = headersOf(f.attempts[0]); + expect(headers["CF-Access-Client-Id"]).toBe("svc"); + expect(headers.Authorization).toBe("Bearer real-token"); + expect(headers.authorization).toBeUndefined(); + + t.disconnect(); }); - it("reports a bad baseURL through onError instead of throwing", async () => { - const onError = vi.fn(); - const t = new SSETransport({ baseURL: "not-a-url", table: "clicks" }); - t.onError = onError; + it("passes `since` on the first connect and no Last-Event-ID", async () => { + const f = makeFetch(); + const t = new SSETransport({ + baseURL: BASE, + table: "clicks", + since: "2026-08-01T00:00:00Z", + fetch: f.impl, + }); t.connect(); + await vi.waitFor(() => expect(f.attempts).toHaveLength(1)); + + expect(new URL(f.attempts[0].url).searchParams.get("since")).toBe("2026-08-01T00:00:00Z"); + expect(headersOf(f.attempts[0])["Last-Event-ID"]).toBeUndefined(); + t.disconnect(); + }); + + it("uses the global fetch when none is configured", async () => { + const spy = vi.fn(async () => streamingResponse().res); + vi.stubGlobal("fetch", spy); + const t = new SSETransport({ baseURL: BASE, table: "clicks" }); + t.connect(); + await vi.waitFor(() => expect(spy).toHaveBeenCalledTimes(1)); + t.disconnect(); + vi.unstubAllGlobals(); + }); +}); + +describe("SSETransport framing", () => { + it("emits an event per frame", async () => { + const f = makeFetch(); + const conn = streamingResponse(); + f.queue.push(() => conn.res); + + const t = new SSETransport({ baseURL: BASE, table: "clicks", fetch: f.impl }); + const seen = collect(t); + t.connect(); + await vi.waitFor(() => expect(seen.statuses).toContain("live")); + + conn.push( + frame("2026-08-01T00:00:01Z", { + table_name: "clicks", + received_timestamp: "2026-08-01T00:00:01Z", + data: { a: 1 }, + }), + ); + await vi.waitFor(() => expect(seen.events).toHaveLength(1)); + + expect(seen.events[0]).toEqual({ + table: "clicks", + timestamp: "2026-08-01T00:00:01Z", + data: { a: 1 }, + }); + t.disconnect(); + }); + + it("ignores comment frames — the connect preamble and keepalives", async () => { + const f = makeFetch(); + const conn = streamingResponse(); + f.queue.push(() => conn.res); + + const t = new SSETransport({ baseURL: BASE, table: "clicks", fetch: f.impl }); + const seen = collect(t); + t.connect(); + await vi.waitFor(() => expect(seen.statuses).toContain("live")); + + conn.push(": connected\n\n"); + conn.push(": keepalive\n\n"); await flush(); - expect(opened).toHaveLength(0); - expect(onError).toHaveBeenCalledOnce(); - expect(onError.mock.calls[0][0].code).toBe("SSE_CONNECT_ERROR"); + expect(seen.events).toHaveLength(0); + expect(seen.errors).toHaveLength(0); + t.disconnect(); + }); + + it("reassembles a frame split across chunk boundaries", async () => { + const f = makeFetch(); + const conn = streamingResponse(); + f.queue.push(() => conn.res); + + const t = new SSETransport({ baseURL: BASE, table: "clicks", fetch: f.impl }); + const seen = collect(t); + t.connect(); + await vi.waitFor(() => expect(seen.statuses).toContain("live")); + + // A single frame delivered one character at a time — the boundary case a + // naive `prefix + chunk` parser corrupts. + const whole = frame("id-1", { + table_name: "clicks", + received_timestamp: "2026-08-01T00:00:02Z", + data: { split: true }, + }); + for (const ch of whole) conn.push(ch); + await vi.waitFor(() => expect(seen.events).toHaveLength(1)); + + expect(seen.events[0].data).toEqual({ split: true }); + t.disconnect(); + }); + + it("warns on a malformed payload without emitting or erroring", async () => { + const warn = vi.spyOn(console, "warn").mockImplementation(() => {}); + const f = makeFetch(); + const conn = streamingResponse(); + f.queue.push(() => conn.res); + + const t = new SSETransport({ baseURL: BASE, table: "clicks", fetch: f.impl }); + const seen = collect(t); + t.connect(); + await vi.waitFor(() => expect(seen.statuses).toContain("live")); + + conn.push("id: x\ndata: {not json\n\n"); + await vi.waitFor(() => expect(warn).toHaveBeenCalled()); + + expect(seen.events).toHaveLength(0); + expect(seen.errors).toHaveLength(0); + t.disconnect(); + warn.mockRestore(); + }); +}); + +describe("SSETransport reconnect and resumption", () => { + beforeEach(() => { + vi.useFakeTimers(); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + it("resumes from the last non-empty id and re-mints the token", async () => { + const f = makeFetch(); + const first = streamingResponse(); + const second = streamingResponse(); + f.queue.push( + () => first.res, + () => second.res, + ); + + let issued = 0; + const t = new SSETransport({ + baseURL: BASE, + table: "clicks", + auth: () => `token-${++issued}`, + fetch: f.impl, + }); + const seen = collect(t); + t.connect(); + await vi.waitFor(() => expect(f.attempts).toHaveLength(1)); + + first.push( + frame("2026-08-01T00:00:03Z", { + table_name: "clicks", + received_timestamp: "2026-08-01T00:00:03Z", + data: { n: 1 }, + }), + ); + await vi.waitFor(() => expect(seen.events).toHaveLength(1)); + + // A passthrough payload carries a blank id. Per the SSE spec that clears + // the last-event-id, which would silently lose the resumption point. + first.push("id: \ndata: {}\n\n"); + await vi.waitFor(() => expect(seen.events).toHaveLength(2)); + + first.close(); + await vi.advanceTimersByTimeAsync(2000); + await vi.waitFor(() => expect(f.attempts).toHaveLength(2)); + + expect(headersOf(f.attempts[1])["Last-Event-ID"]).toBe("2026-08-01T00:00:03Z"); + expect(headersOf(f.attempts[1]).Authorization).toBe("Bearer token-2"); + expect(seen.statuses).toContain("reconnecting"); + + t.disconnect(); + }); + + it("stops for good on a 401 rather than retrying a rejected token", async () => { + const f = makeFetch(); + f.queue.push(() => new Response(JSON.stringify({ error: "invalid token" }), { status: 401 })); + + const t = new SSETransport({ baseURL: BASE, table: "clicks", fetch: f.impl }); + const seen = collect(t); + t.connect(); + await vi.waitFor(() => expect(seen.errors).toHaveLength(1)); + + expect(seen.errors[0].status).toBe(401); + expect(seen.errors[0].message).toBe("invalid token"); + expect(seen.statuses).toContain("closed"); + + await vi.advanceTimersByTimeAsync(60_000); + expect(f.attempts).toHaveLength(1); + }); + + it("retries a 503", async () => { + const f = makeFetch(); + f.queue.push(() => new Response("{}", { status: 503 })); + + const t = new SSETransport({ baseURL: BASE, table: "clicks", fetch: f.impl }); + const seen = collect(t); + t.connect(); + await vi.waitFor(() => expect(seen.errors).toHaveLength(1)); + + expect(seen.errors[0].retryable).toBe(true); + await vi.advanceTimersByTimeAsync(2000); + await vi.waitFor(() => expect(f.attempts.length).toBeGreaterThan(1)); + + t.disconnect(); + }); + + it("retries when the connection itself fails", async () => { + const f = makeFetch(); + f.queue.push(() => { + throw new TypeError("connect ECONNREFUSED"); + }); + + const t = new SSETransport({ baseURL: BASE, table: "clicks", fetch: f.impl }); + const seen = collect(t); + t.connect(); + await vi.waitFor(() => expect(seen.errors).toHaveLength(1)); + + expect(seen.errors[0].code).toBe("SSE_NETWORK_ERROR"); + await vi.advanceTimersByTimeAsync(2000); + await vi.waitFor(() => expect(f.attempts.length).toBeGreaterThan(1)); + + t.disconnect(); + }); +}); + +describe("SSETransport lifecycle", () => { + it("rejects a fetch that cannot stream instead of hanging", async () => { + const f = makeFetch(); + // Exactly what a wrapper that buffers or logs the body hands back. + f.queue.push(() => new Response(null, { status: 200 })); + + const t = new SSETransport({ baseURL: BASE, table: "clicks", fetch: f.impl }); + const seen = collect(t); + t.connect(); + await vi.waitFor(() => expect(seen.errors).toHaveLength(1)); + + expect(seen.errors[0].code).toBe("SSE_NO_STREAM_BODY"); + expect(seen.errors[0].retryable).toBe(false); + expect(seen.statuses).toContain("closed"); + }); + + it("ends the stream on a baseURL that cannot resolve", async () => { + vi.useFakeTimers(); + const f = makeFetch(); + const t = new SSETransport({ baseURL: "not-a-url", table: "clicks", fetch: f.impl }); + const seen = collect(t); + t.connect(); + await vi.waitFor(() => expect(seen.errors).toHaveLength(1)); + + expect(seen.errors[0].code).toBe("SSE_CONNECT_ERROR"); + expect(seen.errors[0].retryable).toBe(false); + expect(seen.statuses).toContain("closed"); + + // Deterministic failure — retrying would only reproduce it. + await vi.advanceTimersByTimeAsync(60_000); + expect(f.attempts).toHaveLength(0); + vi.useRealTimers(); + }); + + it("keeps retrying when the token provider throws", async () => { + vi.useFakeTimers(); + const f = makeFetch(); + let calls = 0; + const t = new SSETransport({ + baseURL: BASE, + table: "clicks", + auth: () => { + calls++; + if (calls === 1) throw new Error("refresh endpoint down"); + return "recovered"; + }, + fetch: f.impl, + }); + const seen = collect(t); + t.connect(); + await vi.waitFor(() => expect(seen.errors).toHaveLength(1)); + + // Transient by assumption: `auth` runs per attempt now, so one bad minute + // at the token endpoint must not tear down a long-lived stream. + expect(seen.errors[0].code).toBe("SSE_AUTH_ERROR"); + expect(seen.errors[0].retryable).toBe(true); + expect(f.attempts).toHaveLength(0); + + await vi.advanceTimersByTimeAsync(2000); + await vi.waitFor(() => expect(f.attempts).toHaveLength(1)); + expect(headersOf(f.attempts[0]).Authorization).toBe("Bearer recovered"); + + t.disconnect(); + vi.useRealTimers(); + }); + + it("aborts the in-flight request on disconnect", async () => { + const f = makeFetch(); + const t = new SSETransport({ baseURL: BASE, table: "clicks", fetch: f.impl }); + const seen = collect(t); + t.connect(); + await vi.waitFor(() => expect(f.attempts).toHaveLength(1)); + + const signal = f.attempts[0].init.signal as AbortSignal; + expect(signal.aborted).toBe(false); + + t.disconnect(); + expect(signal.aborted).toBe(true); + expect(seen.statuses).toContain("closed"); + }); + + it("does not reconnect after disconnect", async () => { + vi.useFakeTimers(); + const f = makeFetch(); + const conn = streamingResponse(); + f.queue.push(() => conn.res); + + const t = new SSETransport({ baseURL: BASE, table: "clicks", fetch: f.impl }); + collect(t); + t.connect(); + await vi.waitFor(() => expect(f.attempts).toHaveLength(1)); + + t.disconnect(); + conn.close(); + await vi.advanceTimersByTimeAsync(60_000); + + expect(f.attempts).toHaveLength(1); + vi.useRealTimers(); + }); + + it("connect() after disconnect() is inert", async () => { + const f = makeFetch(); + const t = new SSETransport({ baseURL: BASE, table: "clicks", fetch: f.impl }); + t.disconnect(); + t.connect(); + await flush(); + expect(f.attempts).toHaveLength(0); }); }); diff --git a/clients/ts/src/stream/sse.ts b/clients/ts/src/stream/sse.ts index 726873f4..27d37ba6 100644 --- a/clients/ts/src/stream/sse.ts +++ b/clients/ts/src/stream/sse.ts @@ -1,4 +1,7 @@ -import type { StreamEvent, StreamStatus, WaveHouseError } from "../types.js"; +import { createParser } from "eventsource-parser"; +import { parseErrorResponse } from "../errors.js"; +import { mergeHeaders } from "../http.js"; +import type { FetchLike, StreamEvent, StreamStatus, WaveHouseError } from "../types.js"; import { resolveURL } from "../url.js"; import type { StreamTransport } from "./controller.js"; @@ -7,16 +10,74 @@ export interface SSEOptions { table: string; since?: string; auth?: () => Promise | string; + fetch?: FetchLike; + headers?: Record; + fetchOptions?: RequestInit; } /** Module-level active SSE connection counter. */ let activeSSEConnections = 0; const SSE_WARN_THRESHOLD = 5; -/** SSE transport. Native EventSource auto-reconnects. */ +/** + * Cap on characters the frame parser will buffer before giving up, guarding + * against unbounded growth from a malformed or hostile stream. The parser + * defaults to unbounded, so this has to be set explicitly. 16 MiB is generous + * headroom over the ~1 MiB NATS payload ceiling a single event can carry. + */ +const MAX_BUFFER_CHARS = 16 * 1024 * 1024; + +/** Ceiling for reconnect backoff. */ +const MAX_BACKOFF_MS = 30_000; + +/** + * Delay before reconnect attempt `attempt` (0-based), jittered. + * + * Unlike the REST backoff in `http.ts`, this one is randomized: REST retries + * are spread across independent calls, but every subscriber to a stream that + * drops reconnects at the same instant, so a fixed schedule synchronizes the + * whole fleet into a thundering herd against a server that is probably already + * unwell. Each attempt waits between 50% and 100% of its nominal delay. + */ +function backoff(attempt: number, retryFloorMs: number): number { + const nominal = Math.max(retryFloorMs, Math.min(1000 * 2 ** attempt, MAX_BACKOFF_MS)); + return nominal / 2 + Math.random() * (nominal / 2); +} + +/** Outcome of one connection attempt, driving the reconnect decision. */ +interface AttemptResult { + /** The attempt reached a readable stream — resets the backoff. */ + live: boolean; + /** Stop for good: a bad token or missing table can't be fixed by retrying. */ + terminal: boolean; +} + +/** + * SSE transport over `fetch`. + * + * Replaces `EventSource`, which cannot set request headers — the reason the JWT + * used to ride in `?token=`, where it crossed every intermediary in the request + * URI. In exchange for the `Authorization` header this owns what `EventSource` + * used to provide: framing, reconnect, and `Last-Event-ID` resumption. + */ export class SSETransport> implements StreamTransport { private _opts: SSEOptions; - private _es: EventSource | null = null; + private _abort: AbortController | null = null; + private _closed = false; + private _counted = false; + /** + * Most recent non-empty event id, replayed as `Last-Event-ID` on reconnect. + * + * Deliberately ignores empty ids. The server emits a blank `id:` line for + * passthrough payloads, and per the SSE spec an empty `id` field *clears* the + * last-event-id — so tracking it faithfully would silently discard the + * resumption point mid-stream and re-open from the beginning. + */ + private _lastEventId: string | null = null; + /** Reconnect floor most recently requested by the server via `retry:`. */ + private _retryFloorMs = 0; + /** Cuts a reconnect gap short when set — see `_sleep`. */ + private _wake: (() => void) | null = null; onEvent: ((event: StreamEvent) => void) | null = null; onStatus: ((status: StreamStatus) => void) | null = null; @@ -27,96 +88,305 @@ export class SSETransport> implements StreamTranspor } connect(): void { - if (typeof EventSource === "undefined") { + if (this._closed) return; + + if (!this._opts.fetch && typeof fetch === "undefined") { throw new Error( - "[wavehouse] EventSource is not available in this environment. " + - "Please provide a global polyfill (e.g., `globalThis.EventSource = require('eventsource')`).", + "[wavehouse] global fetch is not available in this environment. " + + "Upgrade to Node 22+ or supply one via `options.fetch`.", ); } - this._doConnect().catch((err) => { - this.onError?.({ + activeSSEConnections++; + this._counted = true; + if (activeSSEConnections > SSE_WARN_THRESHOLD) { + console.warn( + `[wavehouse] ${activeSSEConnections} SSE connections open. ` + + `Browsers limit HTTP/1.1 to 6 connections per domain.`, + ); + } + + // `_run` resolves rather than rejects on stream failure; this catch is for + // a programming error inside the loop itself. + this._run().catch((err) => { + this._emitError({ status: 0, code: "SSE_CONNECT_ERROR", message: err instanceof Error ? err.message : String(err), - retryable: true, + retryable: false, }); + this.disconnect(); }); } disconnect(): void { - if (this._es) { - this._es.close(); - this._es = null; - activeSSEConnections = Math.max(0, activeSSEConnections - 1); + if (!this._closed) { + this._closed = true; + this._abort?.abort(); + this._abort = null; + this._wake?.(); + if (this._counted) { + activeSSEConnections = Math.max(0, activeSSEConnections - 1); + this._counted = false; + } } this.onStatus?.("closed"); } - private async _doConnect(): Promise { + /** Connect/read/reconnect until the stream is closed or hits a terminal error. */ + private async _run(): Promise { + let attempt = 0; + + while (!this._closed) { + const { live, terminal } = await this._attempt(); + + if (this._closed) return; + if (terminal) { + this.disconnect(); + return; + } + + // A connection that produced a readable stream resets the schedule, so a + // long-lived subscription doesn't inherit a maxed-out delay on its first + // drop after hours of health. + if (live) attempt = 0; + + this.onStatus?.("reconnecting"); + await this._sleep(backoff(attempt, this._retryFloorMs)); + attempt++; + } + } + + /** One connection attempt: request, validate, then pump frames until it ends. */ + private async _attempt(): Promise { + const ac = new AbortController(); + this._abort = ac; + + // A URL that won't build is deterministic — every retry produces the same + // failure — so it ends the stream rather than looping. + let target: string; + try { + target = this._url(); + } catch (e) { + this._emitError({ + status: 0, + code: "SSE_CONNECT_ERROR", + message: e instanceof Error ? e.message : String(e), + retryable: false, + }); + return { live: false, terminal: true }; + } + + // A token provider that throws is treated as transient, unlike the URL + // above: `auth` now runs on every attempt, so a refresh endpoint having a + // bad minute would otherwise tear down a healthy long-lived stream. + let init: RequestInit; + try { + init = await this._init(ac.signal); + } catch (e) { + if (this._isAbort(e)) return { live: false, terminal: true }; + this._emitError({ + status: 0, + code: "SSE_AUTH_ERROR", + message: e instanceof Error ? e.message : String(e), + retryable: true, + }); + return { live: false, terminal: false }; + } + + let res: Response; + try { + const doFetch = this._opts.fetch; + res = doFetch ? await doFetch(target, init) : await fetch(target, init); + } catch (e) { + if (this._isAbort(e)) return { live: false, terminal: true }; + this._emitError({ + status: 0, + code: "SSE_NETWORK_ERROR", + message: e instanceof Error ? e.message : String(e), + retryable: true, + }); + return { live: false, terminal: false }; + } + + if (!res.ok) { + const error = await parseErrorResponse(res); + this._emitError(error); + // 4xx is terminal: reconnecting can't fix a rejected token or a table + // that doesn't exist, and native EventSource likewise stops on non-200. + return { live: false, terminal: !error.retryable }; + } + + // `options.fetch` only has to stream on this path — every REST call reads + // the body with `.text()`. A wrapper that buffers, or clones and logs the + // body, satisfies REST and then hangs here forever, so say so instead. + const body = res.body; + if (!body || typeof body.getReader !== "function") { + this._emitError({ + status: res.status, + code: "SSE_NO_STREAM_BODY", + message: + "Streaming requires a response with a readable body. The configured `options.fetch` " + + "returned one without `body` — an implementation that buffers or reads the response " + + "cannot be used for `.stream()` or `.liveQuery()`.", + retryable: false, + }); + return { live: false, terminal: true }; + } + + this.onStatus?.("live"); + await this._pump(body); + return { live: true, terminal: false }; + } + + /** Resolve the stream URL. Throws on a `baseURL` that isn't absolute. */ + private _url(): string { const url = resolveURL(this._opts.baseURL, "/v1/stream"); url.searchParams.set("table", this._opts.table); if (this._opts.since) { url.searchParams.set("since", this._opts.since); } + return url.toString(); + } + + /** Build the request init, minting a token for this attempt. */ + private async _init(signal: AbortSignal): Promise { + const base: Record = { Accept: "text/event-stream" }; - // EventSource can't set request headers, so the JWT goes in ?token= - // (the server also accepts an Authorization header for non-browser clients) + // Resolved per attempt, not once per stream: a stream outlives its token, + // and a reconnect replaying an expired JWT authenticates as no one — which + // this endpoint answers with a silently reduced view, not an error. if (this._opts.auth) { const token = await this._opts.auth(); if (token) { - url.searchParams.set("token", token); + base.Authorization = `Bearer ${token}`; } } - activeSSEConnections++; - if (activeSSEConnections > SSE_WARN_THRESHOLD) { - console.warn( - `[wavehouse] ${activeSSEConnections} SSE connections open. ` + - `Browsers limit HTTP/1.1 to 6 connections per domain.`, - ); + // Takes precedence over `since` server-side, so the initial window stays on + // the URL and resumption rides the header once there's something to resume. + if (this._lastEventId) { + base["Last-Event-ID"] = this._lastEventId; } - this._es = new EventSource(url.toString()); - - this._es.onopen = () => { - this.onStatus?.("live"); + const init: RequestInit = { + ...this._opts.fetchOptions, + method: "GET", + headers: mergeHeaders(base, this._opts.headers), + signal, + // Owned by the SDK rather than left to `fetchOptions`, unlike the REST + // path, because on a stream these are correctness rather than preference: + // + // cache — as an init field, not a `Cache-Control` header. The header + // form is not in the server's Access-Control-Allow-Headers, + // so it would fail every cross-origin preflight. + // redirect — `error` over `follow`. Platforms strip `Authorization` on a + // cross-origin redirect (whatwg/fetch#1544), and this endpoint + // never rejects an unauthenticated caller — it serves them the + // default role. Following a redirect would therefore turn a + // misdirected stream into a silently downgraded one. + cache: "no-store", + redirect: "error", }; - this._es.onmessage = (e) => { - try { - const msg = JSON.parse(e.data as string) as { - table_name: string; - received_timestamp: string; - data: T; - }; - const event: StreamEvent = { - table: msg.table_name, - timestamp: msg.received_timestamp, - data: msg.data, - }; - this.onEvent?.(event); - } catch { - console.warn("[wavehouse] SSE received malformed message:", e.data); - // ignore malformed messages - } - }; + // Only browsers get `credentials`: some runtimes (Cloudflare Workers) throw + // outright if it is present. Inside a browser an explicit value from + // `fetchOptions` survives, which is how a cookie-authenticated origin opts + // into `include`. + if (!("window" in globalThis)) { + delete init.credentials; + } + + return init; + } - this._es.onerror = () => { - if (this._es?.readyState === EventSource.CONNECTING) { - this.onStatus?.("reconnecting"); - } else if (this._es?.readyState === EventSource.CLOSED) { - this.onStatus?.("closed"); - } else if (this._es?.readyState === EventSource.OPEN) { - this.onStatus?.("live"); - } else { - this.onError?.({ + /** Decode and parse frames until the stream ends, aborts, or errors. */ + private async _pump(body: ReadableStream): Promise { + const parser = createParser({ + onEvent: (msg) => { + if (msg.id) this._lastEventId = msg.id; + if (!msg.data) return; + this._dispatch(msg.data); + }, + onRetry: (ms) => { + this._retryFloorMs = Math.min(Math.max(ms, 0), MAX_BACKOFF_MS); + }, + onError: (err) => { + this._emitError({ status: 0, - code: "SSE_ERROR", - message: "SSE connection error", + code: "SSE_PARSE_ERROR", + message: err.message, retryable: true, }); + }, + maxBufferSize: MAX_BUFFER_CHARS, + }); + + const reader = body.getReader(); + const decoder = new TextDecoder(); + + try { + for (;;) { + const { done, value } = await reader.read(); + if (done) return; + parser.feed(decoder.decode(value, { stream: true })); } - }; + } catch (e) { + if (this._isAbort(e)) return; + this._emitError({ + status: 0, + code: "SSE_READ_ERROR", + message: e instanceof Error ? e.message : String(e), + retryable: true, + }); + } finally { + reader.cancel().catch(() => { + // Already torn down — nothing left to release. + }); + } + } + + /** Turn one frame's `data` into a StreamEvent. */ + private _dispatch(data: string): void { + try { + const msg = JSON.parse(data) as { + table_name: string; + received_timestamp: string; + data: T; + }; + this.onEvent?.({ + table: msg.table_name, + timestamp: msg.received_timestamp, + data: msg.data, + }); + } catch { + console.warn("[wavehouse] SSE received malformed message:", data); + // ignore malformed messages + } + } + + /** Suppress callbacks that race a `disconnect()`. */ + private _emitError(error: WaveHouseError): void { + if (this._closed) return; + this.onError?.(error); + } + + private _isAbort(e: unknown): boolean { + return (e instanceof DOMException && e.name === "AbortError") || this._closed; + } + + /** Wait out a reconnect gap, cut short by `disconnect()`. */ + private _sleep(ms: number): Promise { + return new Promise((resolve) => { + const timer = setTimeout(() => { + this._wake = null; + resolve(); + }, ms); + this._wake = () => { + clearTimeout(timer); + this._wake = null; + resolve(); + }; + }); } } diff --git a/clients/ts/src/types.ts b/clients/ts/src/types.ts index a77b9685..b76f0563 100644 --- a/clients/ts/src/types.ts +++ b/clients/ts/src/types.ts @@ -87,26 +87,41 @@ export interface ClientConfig<_DB extends Database = Database> { * depending on whether the consumer's TypeScript `lib` includes DOM — the same * signature either way, but stable across configurations. * - * In practice the SDK only ever calls it with a string URL, and reads `.ok` and - * `.headers` always, `.text()` on success, and `.status`, `.statusText` plus - * `.json()` when the response is not `ok`. A rejection becomes a - * `NETWORK_ERROR` result and is retried with backoff — except a `DOMException` - * named `AbortError` (what the platform `fetch` and undici throw on abort), - * which becomes `ABORTED` and is not retried. Implementations that reject with - * some other abort error, such as `node-fetch`, are retried instead. + * The parameter is `string` rather than `typeof fetch`'s wider union because a + * string URL is all the SDK ever passes. Since parameters are contravariant, + * narrowing it *widens* what can be assigned: the global `fetch` still fits, and + * so does hand-written `(url: string, init?: RequestInit) => Promise` + * middleware, which the wider spelling rejects at compile time. + * + * On REST the SDK reads `.ok` and `.headers` always, `.text()` on success, and + * `.status`, `.statusText` plus `.json()` when the response is not `ok`. A + * rejection becomes a `NETWORK_ERROR` result and is retried with backoff — + * except a `DOMException` named `AbortError` (what the platform `fetch` and + * undici throw on abort), which becomes `ABORTED` and is not retried. + * Implementations that reject with some other abort error, such as + * `node-fetch`, are retried instead. + * + * **Streaming needs more.** `.stream()` and `.liveQuery()` read the response as + * it arrives, via `.body.getReader()`, so an implementation used with them must + * return a response carrying a live `ReadableStream` — something the type cannot + * enforce, since `Response.body` is legitimately nullable. An implementation + * that buffers the response, or clones it to log the body, satisfies every REST + * call and then hangs forever on a stream that never ends. The transport checks + * for a readable body and fails with `SSE_NO_STREAM_BODY` rather than stalling, + * but the requirement is on you to meet. * * Implementations shipping their own request/response declarations (undici, - * `node-fetch`) need casts on the URL, the init and the return value, since - * those types are separate from the ones behind your global `fetch`; the - * narrow runtime contract above is what makes them safe. + * `node-fetch`) need casts on the init and the return value, since those types + * are separate from the ones behind your global `fetch`; the narrow runtime + * contract above is what makes them safe. */ -export type FetchLike = (input: string | URL | Request, init?: RequestInit) => Promise; +export type FetchLike = (input: string, init?: RequestInit) => Promise; export interface ClientOptions { /** Maximum retry attempts for failed requests. Default: 2. */ maxRetries?: number; /** - * HTTP implementation used for every REST request (SSE streams excluded). + * HTTP implementation used for every request, REST and streaming alike. * Defaults to the global `fetch`. * * Provide one to route through a proxy, attach client certificates, add @@ -130,14 +145,13 @@ export interface ClientOptions { * }); * ``` * - * Only the SSE transport is exempt: the live connection behind `.stream()` - * and `.liveQuery()` uses `EventSource`, which this does not replace. - * `.liveQuery()`'s initial backfill is an ordinary request and does go - * through your function. + * `.stream()` and `.liveQuery()` go through it too, which imposes the extra + * streaming requirement described on {@link FetchLike} — read that before + * supplying a wrapper that touches the response body. */ fetch?: FetchLike; /** - * Headers added to every REST request (SSE streams excluded) — for a + * Headers added to every request, REST and streaming alike — for a * header-gated proxy in front of WaveHouse, such as a Cloudflare Access * service token. * @@ -146,16 +160,29 @@ export interface ClientOptions { * `Content-Type` and `Accept` a given request needs are not overridable * here — a global `Content-Type` that outranked the request's own is a * documented way to break uploads. + * + * In a browser these are subject to CORS: a header outside the safelist adds + * it to the preflight, which the origin must allow. WaveHouse's own CORS + * advertises a fixed set, so custom headers reach it cross-origin only when + * a proxy in front terminates the preflight — which is the deployment they + * exist for. Server-side callers never preflight. */ headers?: Record; /** - * Extra `RequestInit` fields merged into every REST request — `credentials` - * for a cookie-authenticated origin, `mode`, `cache`, `keepalive`, or a + * Extra `RequestInit` fields merged into every request — `credentials` for a + * cookie-authenticated origin, `mode`, `cache`, `keepalive`, or a * runtime-specific extension such as Next.js's `next: { tags }`. * * Fields the SDK controls (`method`, `headers`, `body`, `signal`) always * win, so this cannot corrupt the request itself. Non-standard fields may * need a cast, since `RequestInit` only declares the standard ones. + * + * The streaming transport additionally owns `cache` and `redirect`, where the + * values are load-bearing rather than preference: a `Cache-Control` header + * would fail cross-origin preflight, and following a redirect would strip the + * `Authorization` header on a cross-origin hop and silently downgrade the + * stream to the default role instead of failing. `credentials` is honored in + * browsers and dropped elsewhere, since some runtimes throw if it is set. */ fetchOptions?: RequestInit; } diff --git a/docs/src/content/docs/api.md b/docs/src/content/docs/api.md index ee42e710..09af8b98 100644 --- a/docs/src/content/docs/api.md +++ b/docs/src/content/docs/api.md @@ -23,7 +23,9 @@ For SSE connections where custom headers are not possible, you can pass the toke GET /v1/stream?token= ``` -The `Authorization` header takes precedence when both are provided: the `?token=` query parameter is only a fallback for clients that can't set headers (browser `EventSource`), so a token in the more log-leakable URL never overrides an explicit header credential. A `?token=` is stripped from the URL after extraction whichever credential wins, so it stays out of WaveHouse's own logs — but it has already crossed the wire in the request URI, so redact query strings at any proxy, CDN, or load balancer in front. +The `Authorization` header takes precedence when both are provided: the `?token=` query parameter is only a fallback for clients that can't set headers — a hand-rolled browser `EventSource`, for instance — so a token in the more log-leakable URL never overrides an explicit header credential. A `?token=` is stripped from the URL after extraction whichever credential wins, so it stays out of WaveHouse's own logs — but it has already crossed the wire in the request URI, so redact query strings at any proxy, CDN, or load balancer in front. + +Prefer the header wherever you can. The TypeScript SDK streams over `fetch` and always uses `Authorization`, on browsers and servers alike; the query parameter exists for clients that have no other option. **Authentication is decoupled from authorization.** A request with **no token**, or an **invalid/expired/malformed** one, is *not* rejected outright — it falls back to an empty role that resolves to the policy `default_role`, and authorization is decided downstream. Because the bad-token reason is remembered, a request that is then denied for lacking permission fails loud (`401` "invalid/expired token") instead of a bare `403`. Elevated access requires a valid token whose role is granted (or equals the `admin_role`). A `403` body has two forms: a request that resolves to **no role at all** (no token and no `default_role` configured) returns `{"error":"forbidden: request has no role and no public default_role is configured"}`, while a request carrying a concrete-but-unauthorized role returns the bare `{"error":"forbidden"}` shown in the tables below. diff --git a/docs/src/content/docs/reverse-proxy.mdx b/docs/src/content/docs/reverse-proxy.mdx index ebd64903..9cb86433 100644 --- a/docs/src/content/docs/reverse-proxy.mdx +++ b/docs/src/content/docs/reverse-proxy.mdx @@ -134,11 +134,13 @@ Raising the proxy's read/idle timeout for `/v1/stream` is therefore no longer re ### 3. Forward the token, `since`, and `Last-Event-ID` -A browser `EventSource` can't set request headers, so the JWT can be passed as a query parameter: `GET /v1/stream?token=` (it's stripped after extraction — whichever credential wins — so it stays out of WaveHouse's own logs, but it has already crossed your proxy in the URI, hence the third bullet below; an `Authorization` header still takes precedence). Make sure your proxy: +The SDK streams over `fetch` and sends the JWT as an `Authorization` header, the same as on every other request. A client that can't set headers — a hand-rolled browser `EventSource`, say — can still pass it as `GET /v1/stream?token=` (it's stripped after extraction, whichever credential wins, so it stays out of WaveHouse's own logs; an `Authorization` header always takes precedence). Make sure your proxy: -- forwards the query string (`token`, `since`) — most do by default; +- forwards the `Authorization` request header — check this first if streams authenticate from `curl` but not through the proxy; +- forwards the query string (`since`, and `token` if any client still uses it) — most do by default; - forwards the `Last-Event-ID` request header so reconnects resume from the right point (it's already allow-listed in WaveHouse's CORS preflight); -- does not log the full URL with `token` in it. +- does not redirect `/v1/stream`. The SDK's streaming transport refuses redirects rather than following them: a cross-origin hop strips `Authorization`, and this endpoint answers an unauthenticated caller with a reduced view rather than an error, so following one would silently downgrade the stream instead of failing; +- does not log the full URL, for any client still sending `token` in it. ## Idle timeouts by provider diff --git a/docs/src/content/docs/sdk/index.mdx b/docs/src/content/docs/sdk/index.mdx index 78ccd1d8..1573f508 100644 --- a/docs/src/content/docs/sdk/index.mdx +++ b/docs/src/content/docs/sdk/index.mdx @@ -54,8 +54,8 @@ import { Tabs, TabItem, LinkCard, CardGrid } from "@astrojs/starlight/components pages that can't use ES modules, the bundled IIFE build at `https://cdn.jsdelivr.net/npm/@wavehouse/sdk` exposes a `WaveHouse` global (`WaveHouse.createClient({ … })`) for a classic `` then `WaveHouse.createClient({ … })` works on a no-build, FTP-deployed page — and the SDK README + `sdk.md` gain a "No build step (CDN)" section covering both the ESM-CDN (`` then `WaveHouse.createClient({ … })` works on a no-build, FTP-deployed page — and the SDK README + `sdk.md` gain a "No build step (CDN)" section covering both the ESM-CDN (`` then `WaveHouse.createClient({ … })` works on a no-build, FTP-deployed page — and the SDK README + `sdk.md` gain a "No build step (CDN)" section covering both the ESM-CDN (`` then `WaveHouse.createClient({ … })` works on a no-build, FTP-deployed page — and the SDK README + `sdk.md` gain a "No build step (CDN)" section covering both the ESM-CDN (`