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

clients/ts/src/client.test.ts

@@ -6,11 +6,17 @@ import { PolicyNamespace } from "./policy.js";
 import { SchemaNamespace } from "./schema.js";
 import { SysNamespace } from "./sys.js";
 import { TableRef } from "./table.js";
+import type { FetchLike, PipeRequestOptions, RequestOptions } from "./types.js";
 
 let fetchSpy: ReturnType<typeof vi.fn>;
 
 beforeEach(() => {
-  fetchSpy = vi.fn().mockResolvedValue(new Response(JSON.stringify([]), { status: 200 }));
+  // A fresh Response per call: a Response body can only be read once, so a
+  // shared instance makes the second request fail and retry, and a test
+  // asserting on `mock.calls[1]` would silently be reading that retry.
+  fetchSpy = vi
+    .fn()
+    .mockImplementation(() => Promise.resolve(new Response(JSON.stringify([]), { status: 200 })));
   vi.stubGlobal("fetch", fetchSpy);
 });
 
@@ -113,6 +119,28 @@ describe("WaveHouseClient.pipe()", () => {
     const pipe = client.pipe("top_pages", { limit: 10 });
     expect(typeof pipe.then).toBe("function");
   });
+
+  it("rejects a per-call limit, which the pipes endpoint cannot honour", async () => {
+    const client = createClient({ baseURL: "http://localhost:8080" });
+    // Compile-time assertions: accepting `limit` here would silently drop it,
+    // since the endpoint binds the body as the pipe's parameters. A row cap
+    // belongs in the pipe's own SQL, supplied via `wh.pipe(name, { limit })`.
+
+    // @ts-expect-error — as a fresh literal
+    await client.pipe("top_pages").fetch({ limit: 10 });
+
+    // ...and as a named value. This is the case a plain `Pick<>` would let
+    // through, since excess-property checking only rejects literals — so the
+    // limit would reach the endpoint, be ignored, and never be flagged.
+    const shared: RequestOptions = { signal: AbortSignal.timeout(1000), limit: 10 };
+    // @ts-expect-error — `limit` is not part of PipeRef.fetch's options
+    await client.pipe("top_pages").fetch(shared);
+
+    // `signal` alone is accepted, literal or named.
+    await client.pipe("top_pages").fetch({ signal: AbortSignal.timeout(1000) });
+    const signalOnly: PipeRequestOptions = { signal: AbortSignal.timeout(1000) };
+    await client.pipe("top_pages").fetch(signalOnly);
+  });
 });
 
 describe("WaveHouseClient.sql()", () => {
@@ -143,3 +171,190 @@ 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();
+  });
+});
+
+describe("options.headers", () => {
+  const headersOf = (call: number) =>
+    fetchSpy.mock.calls[call][1].headers as Record<string, string>;
+
+  it("adds configured headers to every request", async () => {
+    const client = createClient({
+      baseURL: "http://localhost:8080",
+      options: { headers: { "CF-Access-Client-Id": "abc.access", "X-Tenant": "acme" } },
+    });
+
+    await client.from("clicks").select("*").limit(1);
+    await client.from("clicks").select("*").limit(1);
+
+    // Exactly two — proves call 1 is a second request, not a retry of the first.
+    expect(fetchSpy).toHaveBeenCalledTimes(2);
+    for (const call of [0, 1]) {
+      expect(headersOf(call)).toMatchObject({
+        "CF-Access-Client-Id": "abc.access",
+        "X-Tenant": "acme",
+      });
+    }
+  });
+
+  it("cannot override the Content-Type a request needs", async () => {
+    // A global Content-Type outranking the request's own is a documented way to
+    // break uploads — it must lose, not merge.
+    const client = createClient({
+      baseURL: "http://localhost:8080",
+      options: { headers: { "Content-Type": "text/plain" } },
+    });
+
+    await client.from("clicks").select("*").limit(1);
+
+    expect(headersOf(0)["Content-Type"]).toBe("application/json");
+    expect(Object.values(headersOf(0))).not.toContain("text/plain");
+  });
+
+  it("cannot displace the auth token, even under different casing", async () => {
+    const client = createClient({
+      baseURL: "http://localhost:8080",
+      auth: () => "real-token",
+      options: { headers: { authorization: "Bearer impostor" } },
+    });
+
+    await client.from("clicks").select("*").limit(1);
+
+    const sent = headersOf(0);
+    expect(sent.Authorization).toBe("Bearer real-token");
+    // The lowercase spelling must not ride along beside the canonical one.
+    expect(sent.authorization).toBeUndefined();
+  });
+
+  it("collapses two configured spellings of one header instead of sending both", async () => {
+    // Left as separate keys, the Headers constructor comma-joins them at fetch
+    // time — `x-tenant: acme, beta` — which is the corruption this guards.
+    const client = createClient({
+      baseURL: "http://localhost:8080",
+      options: { headers: { "x-tenant": "acme", "X-Tenant": "beta" } },
+    });
+
+    await client.from("clicks").select("*").limit(1);
+
+    const sent = headersOf(0);
+    const spellings = Object.keys(sent).filter((k) => k.toLowerCase() === "x-tenant");
+    expect(spellings).toHaveLength(1);
+    expect(new Headers(sent).get("x-tenant")).toBe("beta");
+  });
+
+  it("still applies when no auth is configured", async () => {
+    const client = createClient({
+      baseURL: "http://localhost:8080",
+      options: { headers: { "X-Tenant": "acme" } },
+    });
+
+    await client.from("clicks").select("*").limit(1);
+
+    expect(headersOf(0)["X-Tenant"]).toBe("acme");
+    expect(headersOf(0).Authorization).toBeUndefined();
+  });
+});
+
+describe("options.fetchOptions", () => {
+  it("merges configured RequestInit fields into every request", async () => {
+    const client = createClient({
+      baseURL: "http://localhost:8080",
+      options: { fetchOptions: { credentials: "include", cache: "no-store" } },
+    });
+
+    await client.from("clicks").select("*").limit(1);
+
+    const init = fetchSpy.mock.calls[0][1];
+    expect(init.credentials).toBe("include");
+    expect(init.cache).toBe("no-store");
+  });
+
+  it("cannot corrupt the fields the SDK controls", async () => {
+    const client = createClient({
+      baseURL: "http://localhost:8080",
+      auth: () => "test-token",
+      options: {
+        headers: { "X-Tenant": "acme" },
+        fetchOptions: {
+          method: "DELETE",
+          body: "hijacked",
+          headers: { "X-Tenant": "overridden", Authorization: "Bearer impostor" },
+        } as RequestInit,
+      },
+    });
+
+    await client.from("clicks").select("*").limit(1);
+
+    const init = fetchSpy.mock.calls[0][1];
+    expect(init.method).toBe("POST");
+    expect(init.body).not.toBe("hijacked");
+    // options.headers is the header channel; fetchOptions.headers is discarded
+    // wholesale rather than merged, so it can't smuggle an Authorization past auth.
+    expect(init.headers).toMatchObject({
+      "X-Tenant": "acme",
+      Authorization: "Bearer test-token",
+    });
+  });
+});

clients/ts/src/client.ts

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

clients/ts/src/http.ts

@@ -2,7 +2,7 @@ import { networkError, parseErrorResponse } from "./errors.js";
 import type { HttpContext, WaveHouseError } from "./types.js";
 import { resolveURL } from "./url.js";
 
-interface RequestOptions {
+interface RequestSpec {
   method: string;
   path: string;
   body?: unknown;
@@ -25,11 +25,41 @@ export interface HttpResult<T> {
   headers: Headers;
 }
 
+/**
+ * Merge configured headers underneath the SDK's own, matching names
+ * case-insensitively so a caller's `authorization` can't sit alongside our
+ * `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`.
+ */
+function mergeHeaders(
+  base: Record<string, string>,
+  extra: Record<string, string> | undefined,
+): Record<string, string> {
+  if (!extra) return base;
+  const sdkNames = new Set(Object.keys(base).map((k) => k.toLowerCase()));
+  const merged = { ...base };
+  // Spelling each configured name was last stored under, so two entries
+  // differing only in case replace each other here rather than surviving as
+  // separate keys for `Headers` to comma-join at fetch time.
+  const configured = new Map<string, string>();
+  for (const [name, value] of Object.entries(extra)) {
+    const lower = name.toLowerCase();
+    // Skip rather than overwrite: `base` is the SDK's own, which outranks.
+    if (sdkNames.has(lower)) continue;
+    const prior = configured.get(lower);
+    if (prior !== undefined) delete merged[prior];
+    merged[name] = value;
+    configured.set(lower, name);
+  }
+  return merged;
+}
+
 /**
  * Internal fetch wrapper with auth injection, retry, backoff, and Retry-After.
  * @internal
  */
-export async function request<T>(ctx: HttpContext, opts: RequestOptions): Promise<HttpResult<T>> {
+export async function request<T>(ctx: HttpContext, opts: RequestSpec): Promise<HttpResult<T>> {
   const url = resolveURL(ctx.baseURL, opts.path, opts.params).toString();
   const headers: Record<string, string> = {
     "Content-Type": opts.contentType ?? "application/json",
@@ -51,17 +81,29 @@ export async function request<T>(ctx: HttpContext, opts: RequestOptions): Promis
     }
   }
 
+  // After auth, so `auth` keeps ownership of Authorization, and after the
+  // Content-Type/Accept this request needs.
+  const finalHeaders = mergeHeaders(headers, ctx.options.headers);
+
   let lastError: WaveHouseError | null = null;
   const maxAttempts = ctx.options.maxRetries + 1;
 
   for (let attempt = 0; attempt < maxAttempts; attempt++) {
     try {
-      const res = await fetch(url, {
+      // Configured RequestInit first, so the fields the SDK controls overwrite
+      // it — a supplied `method` or `body` would otherwise corrupt the request.
+      const init: RequestInit = {
+        ...ctx.options.fetchOptions,
         method: opts.method,
-        headers,
+        headers: finalHeaders,
         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,7 +28,8 @@ export type {
   Database,
   // DLQ
   DLQStats,
-  FetchOptions,
+  // HTTP
+  FetchLike,
   // Query
   FilterOp,
   // Ingest
@@ -38,10 +39,12 @@ export type {
   ParamDef,
   // Pipes
   Pipe,
+  PipeRequestOptions,
   // Policy
   Policy,
   PolicyFilter,
   QueryFilter,
+  RequestOptions,
   Result,
   RolePermissions,
   Schemas,

clients/ts/src/pipes.ts

@@ -1,7 +1,7 @@
 import { err, ok } from "./errors.js";
 import { request } from "./http.js";
 import type { StreamController } from "./stream/controller.js";
-import type { FetchOptions, HttpContext, Pipe, Result, StreamOptions } from "./types.js";
+import type { HttpContext, Pipe, PipeRequestOptions, Result, StreamOptions } from "./types.js";
 
 type CreateStreamFn<Row> = (table: string, opts?: StreamOptions) => StreamController<Row>;
 
@@ -24,8 +24,15 @@ export class PipeRef<Row = Record<string, unknown>> implements PromiseLike<Resul
     this._createStream = createStream;
   }
 
-  /** Execute the pipe and return results. */
-  async fetch(opts?: FetchOptions): Promise<Result<Row[]>> {
+  /**
+   * Execute the pipe and return results.
+   *
+   * Takes only `signal` — deliberately narrower than the `RequestOptions` the
+   * query builder accepts. The pipes endpoint binds the body as the pipe's
+   * parameters, so there is no row cap to forward; give the pipe a `{{limit}}`
+   * parameter in its SQL and pass it via `wh.pipe(name, { limit })`.
+   */
+  async fetch(opts?: PipeRequestOptions): Promise<Result<Row[]>> {
     const { data, error } = await request<Row[]>(this._ctx, {
       method: "POST",
       path: `/v1/pipes/${encodeURIComponent(this._name)}`,