feat(sdk)!: add options.headers, fetchOptions, and fetch

CHANGELOG.md

@@ -11,6 +11,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- **`options.fetch` — supply your own HTTP implementation to the SDK** (`clients/ts/src/types.ts`, `clients/ts/src/client.ts`, `clients/ts/src/http.ts`, `clients/ts/src/client.test.ts`, `docs/src/content/docs/sdk/index.mdx`): the SDK called the global `fetch` directly and `ClientOptions` exposed only `maxRetries`, so a consumer had no way to influence how requests are made. `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, so the bundled copy never handles the request. (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. `FetchLike` is deliberately narrower than `typeof fetch` — it takes a `string` URL, which is all the SDK ever passes — so consumer middleware declared as `(url: string, init?: RequestInit) => Promise<Response>` assigns without a cast, where `typeof fetch` would reject it on parameter contravariance. Implementations shipping their own request/response declarations (undici, `node-fetch`) do still need a cast, since those types are separate from the ones behind the global `fetch`; the narrow documented contract is what makes it 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 <key>` 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.
 - **"Durability & Storage" operations guide** (`docs/src/content/docs/durability.md` (new), `docs/src/config/sidebar.ts`, `docs/src/content/docs/reverse-proxy.mdx`, `docs/src/content/docs/configuration.mdx`, `docs/src/content/docs/deployment.md`): documents #84. A new Operations page making the embedded-JetStream durability contract explicit before the docs site publishes: a `200` from `POST /v1/ingest` means the event has been `fsync`'d to disk on the node (the server runs with `SyncAlways: true` in `internal/mq/embedded.go`), which makes the storage substrate's `fsync` tail the ingest latency floor. Covers the contract (and how it differs from JetStream's default page-cache-then-periodic-sync mode), why a slow `fsync` tail manifests as `create stream: ... context deadline exceeded` and `503` backpressure, a where-it's-cheap-vs-expensive substrate table (managed cloud block storage and PLP NVMe vs. ZFS-without-SLOG / qcow2-on-`ext4` / spinning disks), an `fio` recipe + verdict bands to measure your own storage (with the macOS `F_FULLFSYNC` honesty caveat), and the symptom checklist. Forward-references the configurable group-commit interval (`mq.sync_interval`, [#139](https://github.com/Wave-RF/WaveHouse/issues/139)) and the planned `wavehouse storage-check` preflight ([#84](https://github.com/Wave-RF/WaveHouse/issues/84)) without claiming either exists yet. Cross-linked from Configuration (Message Queue), Deployment (Persistent Storage), and the Ingest Pipeline's worker-side ack section; no code changes.

clients/ts/src/client.test.ts

@@ -6,6 +6,7 @@ import { PolicyNamespace } from "./policy.js";
 import { SchemaNamespace } from "./schema.js";
 import { SysNamespace } from "./sys.js";
 import { TableRef } from "./table.js";
+import type { FetchLike } from "./types.js";
 
 let fetchSpy: ReturnType<typeof vi.fn>;
 
@@ -143,3 +144,67 @@ describe("WaveHouseClient.sql()", () => {
     expect(callWithLegacyParams).toThrow(/client\.sql\(sql, params\) was removed/);
   });
 });
+
+describe("options.fetch", () => {
+  it("routes requests through a supplied fetch instead of the global", async () => {
+    // Typed as FetchLike rather than cast: a consumer's override needs no cast,
+    // so the test shouldn't need one either.
+    const custom = vi.fn<FetchLike>(async () => new Response(JSON.stringify([]), { status: 200 }));
+    const client = createClient({
+      baseURL: "http://localhost:8080",
+      auth: () => "test-token",
+      options: { fetch: custom },
+    });
+
+    await client.from("clicks").select("*").limit(1);
+
+    expect(custom).toHaveBeenCalledTimes(1);
+    expect(fetchSpy).not.toHaveBeenCalled();
+    // The documented contract: a string URL and a complete RequestInit —
+    // proxy/middleware consumers rely on the whole request reaching them.
+    const [url, init] = custom.mock.calls[0];
+    expect(typeof url).toBe("string");
+    expect(url).toContain("http://localhost:8080");
+    expect(init?.method).toBe("POST");
+    expect(init?.headers).toMatchObject({
+      "Content-Type": "application/json",
+      Authorization: "Bearer test-token",
+    });
+    expect(typeof init?.body).toBe("string");
+  });
+
+  it("falls back to the global fetch when no override is given", async () => {
+    const client = createClient({ baseURL: "http://localhost:8080" });
+    await client.from("clicks").select("*").limit(1);
+    expect(fetchSpy).toHaveBeenCalledTimes(1);
+  });
+
+  it("keeps the global late-bound so it can be swapped after construction", async () => {
+    const client = createClient({ baseURL: "http://localhost:8080" });
+    const replacement = vi
+      .fn()
+      .mockResolvedValue(new Response(JSON.stringify([]), { status: 200 }));
+    vi.stubGlobal("fetch", replacement);
+
+    await client.from("clicks").select("*").limit(1);
+
+    expect(replacement).toHaveBeenCalledTimes(1);
+    expect(fetchSpy).not.toHaveBeenCalled();
+  });
+
+  it("applies the override to retries too, not just the first attempt", async () => {
+    const custom = vi
+      .fn<FetchLike>()
+      .mockResolvedValueOnce(new Response("boom", { status: 500 }))
+      .mockResolvedValueOnce(new Response(JSON.stringify([]), { status: 200 }));
+    const client = createClient({
+      baseURL: "http://localhost:8080",
+      options: { fetch: custom, maxRetries: 1 },
+    });
+
+    await client.from("clicks").select("*").limit(1);
+
+    expect(custom).toHaveBeenCalledTimes(2);
+    expect(fetchSpy).not.toHaveBeenCalled();
+  });
+});

clients/ts/src/client.ts

@@ -37,6 +37,7 @@ export class WaveHouseClient<DB extends Database = Database> {
       auth: config.auth,
       options: {
         maxRetries: config.options?.maxRetries ?? 2,
+        fetch: config.options?.fetch,
       },
     };
 

clients/ts/src/http.ts

@@ -56,12 +56,17 @@ export async function request<T>(ctx: HttpContext, opts: RequestOptions): Promis
 
   for (let attempt = 0; attempt < maxAttempts; attempt++) {
     try {
-      const res = await fetch(url, {
+      const init: RequestInit = {
         method: opts.method,
         headers,
         body: requestBody,
         signal: opts.signal,
-      });
+      };
+      // Call the global directly when no override is configured, rather than
+      // capturing it — a detached `fetch` reference is not universally safe to
+      // invoke, and late binding is what lets tests swap `globalThis.fetch`.
+      const doFetch = ctx.options.fetch;
+      const res = doFetch ? await doFetch(url, init) : await fetch(url, init);
 
       if (res.ok) {
         const text = await res.text();

clients/ts/src/index.ts

@@ -28,6 +28,8 @@ export type {
   Database,
   // DLQ
   DLQStats,
+  // HTTP
+  FetchLike,
   FetchOptions,
   // Query
   FilterOp,

clients/ts/src/types.ts

@@ -80,9 +80,58 @@ export interface ClientConfig<_DB extends Database = Database> {
   options?: ClientOptions;
 }
 
+/**
+ * A `fetch`-compatible function.
+ *
+ * Deliberately narrower than `typeof fetch`: the SDK only ever calls it with a
+ * string URL, so this is what it actually requires. The global `fetch` still
+ * assigns to it, and so does middleware you declare as
+ * `(url: string, init?: RequestInit) => Promise<Response>` — which `typeof
+ * fetch` would reject, since a parameter accepting only `string` is not
+ * assignable to one accepting `RequestInfo | URL`.
+ *
+ * The response is read for `.ok` and `.headers` always, `.text()` on success,
+ * and `.status`, `.statusText` plus `.json()` when it 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.
+ *
+ * Implementations shipping their own request/response declarations (undici,
+ * `node-fetch`) still need a cast, since those types are separate from the ones
+ * behind your global `fetch`; the narrow contract above is what makes it safe.
+ */
+export type FetchLike = (url: string, init?: RequestInit) => Promise<Response>;
+
 export interface ClientOptions {
   /** Maximum retry attempts for failed requests. Default: 2. */
   maxRetries?: number;
+  /**
+   * HTTP implementation used for every REST request (SSE streams excluded).
+   * Defaults to the global `fetch`.
+   *
+   * Provide one to route through a proxy, attach client certificates, add
+   * middleware (logging, tracing, circuit breaking), or bypass a transport bug
+   * in the runtime's bundled HTTP stack by routing through an implementation
+   * you install yourself — e.g. an undici new enough to be free of
+   * {@link https://github.com/nodejs/undici/issues/5600 | undici #5600}:
+   *
+   * ```ts
+   * import { fetch as undiciFetch } from "undici"; // npm install undici — 8.10.0+
+   * createClient({
+   *   baseURL,
+   *   options: {
+   *     fetch: (url, init) => undiciFetch(url, init as never) as unknown as Promise<Response>,
+   *   },
+   * });
+   * ```
+   *
+   * 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.
+   */
+  fetch?: FetchLike;
 }
 
 // --- Structured query AST (matches backend wire format) ---
@@ -276,5 +325,6 @@ export interface StreamOptions {
 export interface HttpContext {
   baseURL: string;
   auth?: () => Promise<string> | string;
-  options: { maxRetries: number };
+  // Stays optional rather than defaulted here, to keep the global late-bound.
+  options: { maxRetries: number; fetch?: FetchLike };
 }