diff --git a/.wiki/Architecture.md b/.wiki/Architecture.md index e3765733..a911dffd 100644 --- a/.wiki/Architecture.md +++ b/.wiki/Architecture.md @@ -51,7 +51,11 @@ module mocking. ## Query path 1. The editor/controller prepares SQL and typed parameters. -2. `src/net/ch-client.js` sends the HTTP request with injected auth/fetch context. +2. `src/net/ch-client.js` sends the HTTP request with injected auth/fetch context, + delegating generic request construction and stream mechanics through a narrow + transport contract (`src/net/clickhouse-transport.types.js` + + `src/net/clickhouse-http-transport.js`, #585 Phase 1) — auth/epoch/retry policy + stays in `ch-client.js`. 3. `JSONStringsEachRowWithProgress` is folded line by line by pure stream logic. 4. Results resolve through the panel registry to table, chart, logs, KPI, filter, text, or graph-oriented renderers. diff --git a/.wiki/Decisions-and-Roadmap.md b/.wiki/Decisions-and-Roadmap.md index cbfa9263..ced6db90 100644 --- a/.wiki/Decisions-and-Roadmap.md +++ b/.wiki/Decisions-and-Roadmap.md @@ -130,9 +130,13 @@ Two roadmap tracks are current: 0 completed as a fully valid outcome — **no production cutover occurred**, and `src/net/ch-client.ts` (the current custom transport) **remains authoritative**. Phase 1 (separating application policy from the concrete - transport implementation) remains useful and unblocked on its own merits; - Phases 2–4 (production adoption, cutover, and custom-transport deletion) do - not proceed without a new decision. + transport implementation) — **landed** on `wip/585-phase1-transport-seam`: + `src/net/clickhouse-transport.types.ts` (the `ClickHouseTransport` contract) + + `src/net/clickhouse-http-transport.ts` (`createHttpTransport`, the current + implementation behind it) put the existing transport behind a narrow seam + with zero behavior change; `ch-client.ts` keeps every auth/epoch/retry + policy and product operation. Phases 2–4 (production adoption, cutover, and + custom-transport deletion) do not proceed without a new decision. Re-read GitHub before acting because issue state can change; a MERGED PR is not proof its code is on `main` (see the reset above). diff --git a/.wiki/Source-Map.md b/.wiki/Source-Map.md index 1c231812..23845600 100644 --- a/.wiki/Source-Map.md +++ b/.wiki/Source-Map.md @@ -16,7 +16,9 @@ Back to [[Home]]. Related: [[Architecture]], [[Product-and-Features]]. | `src/dashboard/application/dashboard-repaint-plan.js` | pure repaint-decision arbitration extracted from `ui/dashboard.js`'s `renderDashboard` effect (#589) | | `src/ui/dashboard-tile-gestures.js` | Dashboard corner-drag resize, Command/Ctrl-drag reorder, and modifier-cue controller, extracted from `ui/dashboard.js` behind an injected `TileGestureDeps` seam (#589) | | `src/state.js` | signals-backed state model and persistence operations | -| `src/net/ch-client.js` | ClickHouse HTTP execution and schema calls | +| `src/net/ch-client.js` | ClickHouse HTTP execution and schema calls; auth/epoch/retry policy, product operations, `ChCtx` (#585 Phase 1: generic request/stream mechanics delegate through the transport seam below) | +| `src/net/clickhouse-transport.types.js` | Type-only `ClickHouseTransport` contract (`send`/`streamLines`, `TransportDeps`, `TransportRequest`) (#585 Phase 1) | +| `src/net/clickhouse-http-transport.js` | `createHttpTransport` — the current custom HTTP implementation of that contract, plus `chUrl`/`ChUrlOpts` (#585 Phase 1) | | `src/net/oauth.js` | OAuth flow/token exchange | | `src/editor/editor-port.js` | SQL editor contract and safe no-op port | | `src/editor/codemirror-adapter.js` | SQL CodeMirror 6 adapter | diff --git a/CHANGELOG.md b/CHANGELOG.md index ae7b31bd..b56afa6d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -25,6 +25,29 @@ auto-generated per-PR notes; this file is the curated, human-readable history. consistent comment/blank-stripped metric). Production ClickHouse transport behavior is unchanged — `src/net/ch-client.ts` remains authoritative, and no production cutover occurred. +- **#585 Phase 1: internal transport seam, no behavior change.** Defined a + narrow SQL Browser ClickHouse-transport contract + (`src/net/clickhouse-transport.types.ts`: `ClickHouseTransport`, + `TransportDeps`, `TransportRequest`, `StreamCallbacks`) and put the current + custom HTTP implementation behind it + (`src/net/clickhouse-http-transport.ts`: `createHttpTransport`, plus + `chUrl`/`ChUrlOpts` moved verbatim from `ch-client.ts` and re-exported + unchanged). `ch-client.ts`'s auth/epoch/retry policy, product operations, + and `ChCtx` are otherwise untouched — no new `ChCtx` field, no runtime + transport switch. `authedFetch` is the one exported-signature change (its + only importers are `ch-client.ts` internals and its own unit test): it now + takes a structured request instead of a prebuilt URL string, and snapshots + the caller's `settings`/`params` synchronously at entry (before its first + await) as a single centralized aliasing defense. Added a reusable + contract-test-suite factory (`tests/unit/clickhouse-transport-contract.ts`), + registered against the current implementation only — Phases 2–4 (an + official-client cutover) remain out of scope and do not proceed without a + new decision, per ADR-0005 (Rejected). `build/check-boundaries.mjs` gains a + bare-specifier ban on `@clickhouse/client-web` anywhere under `src/**` + (mirrored as a coverage-gated unit test) and twin rules keeping the + transport implementation and its type-only contract from reaching + auth/application policy or UI, even type-only. No user-visible or + production-behavior change; bundle size delta ≈ 0 (pure code movement). - **Opt-in ChatGPT-authored `/ship` planning.** `/ship --planner chatgpt` keeps the existing Fable-authored workflow as the default, but lets ChatGPT own complete plan drafts and revisions while Fable/high performs repository-grounded diff --git a/build/check-boundaries.mjs b/build/check-boundaries.mjs index 18df03df..fd048eb0 100644 --- a/build/check-boundaries.mjs +++ b/build/check-boundaries.mjs @@ -111,6 +111,26 @@ const RULES = [ except: ['src/ui/dnd-mime.js', 'src/ui/dnd-mime.ts', 'src/ui/dom.js', 'src/ui/dom.ts'], why: 'issue #60/#313: the editor layer is a leaf — UI actions are injected via app callbacks, not imported', }, + // Issue #585 Phase 1: the generic ClickHouse transport (and its type-only + // contract) is a low-level leaf that must not reach auth/application + // policy or UI, even type-only (the checker's own header comment: `import + // type` counts too). Two entries, not one, because collectFiles(ruleDir) + // matches per-`dir` — a single-file rule naming only the implementation + // file would leave the sibling contract file unguarded. `forbidden` targets + // are resolved repo-relative paths (never raw specifier strings like + // './ch-client.ts'), matching how the checker resolves and compares them. + { + dir: 'src/net/clickhouse-http-transport.ts', + forbidden: ['src/net/ch-client.ts', 'src/net/oauth.ts', + 'src/net/oauth-config.ts', 'src/application', 'src/ui'], + why: 'issue #585 Phase 1: the generic transport cannot reach auth/application policy or UI', + }, + { + dir: 'src/net/clickhouse-transport.types.ts', + forbidden: ['src/net/ch-client.ts', 'src/net/oauth.ts', + 'src/net/oauth-config.ts', 'src/application', 'src/ui'], + why: 'issue #585 Phase 1: the transport contract must not couple to auth/application policy or UI, even type-only', + }, ]; function collectFiles(target) { @@ -225,6 +245,28 @@ for (const file of collectFiles(path.join(repoRoot, 'src'))) { } } +// Issue #585 Phase 1: no file under src/** may import the official +// `@clickhouse/client-web` package (a bare specifier — the `RULES` loop above +// skips those, `if (!spec.startsWith('.')) continue;`, hence this separate +// block). ADR-0005 (docs/ADR-0005-clickhouse-web-client.md) is Rejected, so +// Phases 2-4 (the official-client cutover) do not proceed without a new +// decision — today this bans the import everywhere in `src/`. The single +// allowlist entry names the FUTURE official transport file (does not exist +// yet); the rule is written so it activates correctly the moment that file is +// born, rather than needing a second edit here. +const CLIENT_WEB_SPECIFIER = '@clickhouse/client-web'; +const CLIENT_WEB_ALLOWLIST = new Set(['src/net/clickhouse-web-transport.ts']); +for (const file of collectFiles(path.join(repoRoot, 'src'))) { + const relFile = path.relative(repoRoot, file).split(path.sep).join('/'); + checkedFiles += 1; + const source = fs.readFileSync(file, 'utf8'); + for (const spec of extractSpecifiers(source)) { + if (spec !== CLIENT_WEB_SPECIFIER && !spec.startsWith(`${CLIENT_WEB_SPECIFIER}/`)) continue; + if (CLIENT_WEB_ALLOWLIST.has(relFile)) continue; + violations.push(`${relFile} → ${spec} (issue #585 Phase 1: only the future official transport file may import @clickhouse/client-web — ADR-0005 is Rejected, Phases 2-4 do not proceed without a new decision)`); + } +} + if (violations.length) { console.error('check-boundaries: architecture violations:'); for (const line of violations) console.error(` ${line}`); diff --git a/docs/ADR-0005-clickhouse-web-client.md b/docs/ADR-0005-clickhouse-web-client.md index f7ea23cf..1e4760f9 100644 --- a/docs/ADR-0005-clickhouse-web-client.md +++ b/docs/ADR-0005-clickhouse-web-client.md @@ -630,6 +630,30 @@ teardown; permanent live-ClickHouse CI integration; general supported-browser documentation (tracked separately by #71); a permanent second transport path. +### Phase 1 addendum (landed) + +Phase 1 — "establish the transport seam without behavior change" — landed on +`wip/585-phase1-transport-seam`, per this ADR's own prescription for a +Rejected outcome ("retain the current transport with the layer separation +from Phase 1 remains useful independent of this decision"). It defines +`src/net/clickhouse-transport.types.ts` (the `ClickHouseTransport` contract: +`send`/`streamLines`, `TransportDeps`, `TransportRequest`, `StreamCallbacks`) +and `src/net/clickhouse-http-transport.ts` (`createHttpTransport` — the +current custom HTTP implementation behind that contract, plus `chUrl`/ +`ChUrlOpts` moved verbatim from `ch-client.ts`). `ch-client.ts`'s auth/epoch/ +retry policy, product operations, and `ChCtx` are otherwise unchanged — no +new `ChCtx` field, no runtime transport switch; `authedFetch` is the one +exported-signature change, scoped to its own unit test (no production +importer exists outside `ch-client.ts` itself). A reusable contract-test- +suite factory (`tests/unit/clickhouse-transport-contract.ts`) registers +against the current implementation only — Phases 2–4 remain exactly as +scoped above and do not proceed without a new decision. `build/ +check-boundaries.mjs` gained a bare-specifier ban on `@clickhouse/client-web` +under `src/**` (mirrored as a coverage-gated unit test), naming +`src/net/clickhouse-web-transport.ts` as the single allowlisted (not yet +existing) future official-transport file. No user-visible or production- +behavior change; bundle size delta ≈ 0. + ## Reproduction commands ```sh diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index d492572d..90a8a2a1 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -199,6 +199,28 @@ folded via the pure `applyStreamLine`; a single automatic token refresh on 401/403/`token_verification_exception` (before `authConfirmed` flips, an auth failure signs out; after, it is a query error). +### Transport seam (#585 Phase 1) + +Generic request construction and stream mechanics are split out behind a +narrow contract: `net/clickhouse-transport.types.ts` declares +`ClickHouseTransport` (`send`/`streamLines`), `TransportDeps` (`fetch`/ +`origin` accessors — read live per request, never snapshotted, since the +live `chCtx.origin` is mutated in place on sign-in), and `TransportRequest`. +`net/clickhouse-http-transport.ts`'s `createHttpTransport` is the current +custom HTTP implementation of that contract — `chUrl`/`ChUrlOpts` live there +now, re-exported unchanged from `ch-client.ts`. `ch-client.ts` keeps every +auth/epoch/retry/lifecycle policy (`authedFetch`), product operation, and +`ChCtx` exactly as before; a module-private `transportFor(ctx)` delegates +unconditionally to `createHttpTransport` — `ChCtx` gained no field and there +is no runtime transport switch. `authedFetch` snapshots the caller's +`settings`/`params` synchronously at entry, before its first await, as one +centralized defense against a caller mutating those objects while a +token/refresh await is pending. A reusable contract-test-suite factory +(`tests/unit/clickhouse-transport-contract.ts`) registers against this one +implementation; a future official-client implementation (ADR-0005 is +Rejected; Phases 2–4 do not proceed without a new decision) would satisfy the +same contract and reuse the same suite. + ## Build `build/build.mjs` runs esbuild (bundle + minify, IIFE), inlines the script and diff --git a/src/net/ch-client.ts b/src/net/ch-client.ts index 08add363..61f27e97 100644 --- a/src/net/ch-client.ts +++ b/src/net/ch-client.ts @@ -12,6 +12,19 @@ import type { StreamLine } from '../core/stream.js'; import { parseAstTables, buildSchemaGraph, externalDbs } from '../core/schema-graph.js'; import type { SchemaGraphTableRow, SchemaGraphDictRow } from '../core/schema-graph.js'; import { sqlString } from '../core/format.js'; +// Issue #585 Phase 1 — the transport seam. `chUrl` moved verbatim to +// `clickhouse-http-transport.ts`; re-exported here (with its `ChUrlOpts` +// parameter type) so every existing importer — including +// `tests/spike/clickhouse-client/current-adapter.ts` — keeps resolving. The +// generic request-construction/fetch/stream mechanics live in +// `createHttpTransport`; this module keeps every auth/epoch/retry policy, +// product operation, and `ChCtx` exactly as before, delegating through the +// transport instead of calling `chUrl`/`ctx.fetch` directly. +import { chUrl, createHttpTransport } from './clickhouse-http-transport.js'; +import type { TransportRequest } from './clickhouse-transport.types.js'; +export { chUrl }; +export type { ChUrlOpts } from './clickhouse-http-transport.js'; +export type { ClickHouseTransport, StreamCallbacks, TransportDeps, TransportRequest } from './clickhouse-transport.types.js'; // ── Injected ctx seam ──────────────────────────────────────────────────────── @@ -101,33 +114,56 @@ export interface ChJsonResult> { data?: T[]; } -/** `chUrl`'s query-string options. */ -export interface ChUrlOpts { - format?: string; - extra?: Record; - params?: Record; -} - -/** Build a ClickHouse HTTP URL with query-string options. Pure. */ -export function chUrl(origin: string, opts: ChUrlOpts = {}): string { - const format = opts.format || 'JSONStringsEachRowWithProgress'; - let url = origin + '?default_format=' + format + '&enable_http_compression=1'; - for (const [k, v] of Object.entries(opts.extra || {})) { - url += '&' + k + '=' + encodeURIComponent(v); - } - for (const [k, v] of Object.entries(opts.params || {})) { - url += '&' + k + '=' + encodeURIComponent(v); - } - return url; +/** Delegates unconditionally to the single current transport implementation. + * `deps.fetch`/`deps.origin` are accessors reading the LIVE mutable `ctx` + * fields per request (never a snapshot) — `ctx.origin` is mutated in place on + * sign-in (`connection-session.ts`), so a request issued after that mutation + * must observe the new value. `ChCtx` itself gains no new field: there is no + * production runtime switch, only this one unconditional wiring (an + * injectable composition seam is introduced only when a second + * implementation actually exists — Phase 2, which requires a new decision). */ +function transportFor(ctx: ChCtx) { + return createHttpTransport({ fetch: () => ctx.fetch, origin: () => ctx.origin }); } /** - * POST `sql` to ClickHouse with one automatic token-refresh retry. Resolves to - * the raw Response. Throws Error('signed out') after calling ctx.onSignedOut() - * when authentication cannot be recovered. + * POST `request.sql` to ClickHouse with one automatic token-refresh retry. + * Resolves to the raw Response. Throws Error('signed out') after calling + * ctx.onSignedOut() when authentication cannot be recovered. `request` omits + * `authorization` — this function resolves the credential for THIS request + * (and its retry) itself; every other `TransportRequest` field is the + * caller's request, unchanged. */ -export async function authedFetch(ctx: ChCtx, url: string, sql: string, signal?: AbortSignal): Promise { +export async function authedFetch(ctx: ChCtx, request: Omit): Promise { const requestEpoch = ctx.currentEpoch?.(); + // Centralized aliasing defense (review finding folded in, pass-5 revision): + // snapshot the incoming request's settings/params synchronously HERE, at + // entry, before the first await (`ctx.getToken()`) — one mechanism for + // every present and future caller, rather than per-call-site defensive + // spreads. This preserves today's invocation-time capture (today `chUrl` + // serializes both records into the URL string synchronously, before this + // function's first await), so a caller that retains and mutates either + // record while a token/refresh await is pending cannot change the request + // this function already committed to sending — on the initial attempt AND + // the one-refresh retry alike. + const settings = request.settings ? { ...request.settings } : undefined; + const params = request.params ? { ...request.params } : undefined; + const { sql, defaultFormat, signal } = request; + // Request-preparation failures are not transport failures (ChatGPT review + // pass 2, P1). Pre-refactor, every caller built `chUrl(...)` as a plain + // argument BEFORE invoking authedFetch, so a URIError from malformed + // settings/params (e.g. `encodeURIComponent` on a lone UTF-16 surrogate) + // propagated as a synchronous throw with no token read, no fetch, and no + // `onTransportOffline` call. `transport.send` now builds that same URL + // internally, inside the try/catch that classifies fetch rejections as + // transport-offline — so without this eager, discarded validation call, the + // identical throw would surface only after `ctx.getToken()` and would be + // misclassified as a network failure. Calling `chUrl` here, before the first + // await, reproduces the exact original failure shape; `transport.send` still + // calls it again at actual send time against the (possibly since-mutated) + // live `ctx.origin` — origin is only concatenated, never `encodeURIComponent`- + // encoded, so it cannot itself throw and re-validating it changes nothing. + chUrl(ctx.origin, { format: defaultFormat, extra: settings, params }); const token = await ctx.getToken(); // getToken may have awaited a sign-in/sign-out replacement. Its credential // belongs to that replacement and this request must not send it. @@ -141,19 +177,19 @@ export async function authedFetch(ctx: ChCtx, url: string, sql: string, signal?: // ctx.authHeader(token) lets the app pick the scheme (Bearer vs Basic); // default to Bearer so the seam stays optional. const authHeader = ctx.authHeader || ((t: string) => 'Bearer ' + t); + const transport = transportFor(ctx); for (;;) { let resp: Response; try { // Fence every attempt immediately before the injected side effect. A // retry must never send a replacement session's newly-read credential. + // (Precision: `transport.send` internally evaluates the REQUIRED-PURE + // `deps.origin()`/`deps.fetch()` accessors and builds the URL AFTER + // this fence, immediately before the fetch itself — see + // `clickhouse-transport.types.ts`'s `TransportDeps` doc comment.) const authorization = authHeader(bearer); if (!isCurrentEpoch(ctx, requestEpoch)) throw staleEpochAbort(); - resp = await ctx.fetch(url, { - method: 'POST', - body: sql, - headers: { Authorization: authorization }, - signal, - }); + resp = await transport.send({ sql, defaultFormat, settings, params, authorization, signal }); } catch (e) { // Only a rejected fetch is a transport failure. HTTP failures are normal // responses and caller cancellation is deliberately invisible here. @@ -222,7 +258,7 @@ export async function queryJson>( extra?: Record, params?: Record, ): Promise> { - const resp = await authedFetch(ctx, chUrl(ctx.origin, { format: 'JSON', extra, params }), sql, signal); + const resp = await authedFetch(ctx, { sql, defaultFormat: 'JSON', settings: extra, params, signal }); if (!resp.ok) throw new Error(parseExceptionText(await resp.text())); return resp.json(); } @@ -329,10 +365,14 @@ export async function killQueryWithLease( ): Promise { if (!queryId) return; try { - await lease.fetch(chUrl(lease.origin, { format: 'JSON' }), { - method: 'POST', - body: 'KILL QUERY WHERE query_id = ' + sqlString(queryId) + ' ASYNC', - headers: { Authorization: lease.authorization }, + // A one-shot transport built directly from the frozen lease — never + // `transportFor(ctx)` / `authedFetch` — so cleanup reads no mutable auth, + // token, or refresh state (hard invariant 8). + const transport = createHttpTransport({ fetch: () => lease.fetch, origin: () => lease.origin }); + await transport.send({ + sql: 'KILL QUERY WHERE query_id = ' + sqlString(queryId) + ' ASYNC', + defaultFormat: 'JSON', + authorization: lease.authorization, }); } catch { /* best-effort */ } } @@ -966,11 +1006,12 @@ export interface ExportQueryOptions { */ export async function exportQuery(ctx: ChCtx, sql: string, opts: ExportQueryOptions = {}): Promise { const { queryId, signal, format, params } = opts; - const url = chUrl(ctx.origin, { - format: format || 'TabSeparatedWithNames', + const resp = await authedFetch(ctx, { + sql, + defaultFormat: format || 'TabSeparatedWithNames', params: { ...(queryId ? { query_id: queryId } : {}), ...(params || {}) }, + signal, }); - const resp = await authedFetch(ctx, url, sql, signal); if (!resp.ok) throw new Error(parseExceptionText(await resp.text())); return resp; } @@ -1039,20 +1080,21 @@ export async function runQuery(ctx: ChCtx, sql: string, o: RunQueryOptions = {}) const cap: Record = (o.resultRowLimit ?? 0) > 0 ? { max_result_rows: o.resultRowLimit!, result_overflow_mode: 'break' } : {}; - const url = chUrl(ctx.origin, { - format: fmtParam, + const resp = await authedFetch(ctx, { + sql, + defaultFormat: fmtParam, // wait_end_of_query buffers the whole response server-side so the HTTP // status reflects errors — but it defeats progressive streaming (first rows // wait for the query to finish: ~16s vs ~0.5s on a 1.3M-row scan). Keep it // only for raw modes (read whole anyway); the streaming Table path drops it // and surfaces mid-stream errors via the in-band `exception` line instead. - extra: { ...(isStreaming ? {} : { wait_end_of_query: 1 }), ...cap, add_http_cors_header: 1 }, + settings: { ...(isStreaming ? {} : { wait_end_of_query: 1 }), ...cap, add_http_cors_header: 1 }, // Tagging the request with a query_id lets Cancel issue KILL QUERY for it. // Caller-supplied params (o.params) ride alongside — e.g. multiquery SELECTs // add max_result_rows / result_overflow_mode to cap the result server-side. params: { ...(o.queryId ? { query_id: o.queryId } : {}), ...(o.params || {}) }, + signal: o.signal, }); - const resp = await authedFetch(ctx, url, sql, o.signal); if (!resp.ok) { return { error: parseExceptionText(await resp.text()) }; @@ -1060,33 +1102,6 @@ export async function runQuery(ctx: ChCtx, sql: string, o: RunQueryOptions = {}) if (!isStreaming) { return { raw: await resp.text() }; } - const reader = resp.body!.getReader(); - const decoder = new TextDecoder(); - let buffer = ''; - for (;;) { - const { done, value } = await reader.read(); - if (done) break; - buffer += decoder.decode(value, { stream: true }); - const lines = buffer.split('\n'); - buffer = lines[lines.length - 1]; - for (const line of lines.slice(0, -1)) { - if (!line) continue; - let json: StreamLine; - try { - json = JSON.parse(line); - } catch { - continue; - } - o.onLine && o.onLine(json); - } - o.onChunk && o.onChunk(); - } - if (buffer.trim()) { - try { - o.onLine && o.onLine(JSON.parse(buffer)); - } catch { - /* trailing partial line */ - } - } + await transportFor(ctx).streamLines(resp.body!, { onLine: o.onLine, onChunk: o.onChunk }); return { streamed: true }; } diff --git a/src/net/clickhouse-http-transport.ts b/src/net/clickhouse-http-transport.ts new file mode 100644 index 00000000..475719d0 --- /dev/null +++ b/src/net/clickhouse-http-transport.ts @@ -0,0 +1,96 @@ +// Issue #585 Phase 1 — the current custom ClickHouse HTTP transport, +// re-seated behind the `ClickHouseTransport` contract (`clickhouse-transport.types.ts`). +// This is a pure move: `chUrl` (+ its `ChUrlOpts` parameter type) and the +// progress-line stream-read loop are relocated here verbatim from +// `ch-client.ts`, which re-imports/re-exports both so every existing importer +// keeps resolving. No behavior change; no product SQL; no auth/lifecycle +// policy (that stays app-side in `ch-client.ts`'s `authedFetch`). +// +// Ownership boundary: this file may depend only on `src/core` — never on +// `ch-client.ts`, `oauth.ts`, `oauth-config.ts`, `src/application/`, or +// `src/ui/`. `build/check-boundaries.mjs` enforces this mechanically. + +import type { ClickHouseTransport, StreamCallbacks, TransportDeps, TransportRequest } from './clickhouse-transport.types.js'; +import type { StreamLine } from '../core/stream.js'; + +/** `chUrl`'s query-string options. */ +export interface ChUrlOpts { + format?: string; + extra?: Record; + params?: Record; +} + +/** Build a ClickHouse HTTP URL with query-string options. Pure. */ +export function chUrl(origin: string, opts: ChUrlOpts = {}): string { + const format = opts.format || 'JSONStringsEachRowWithProgress'; + let url = origin + '?default_format=' + format + '&enable_http_compression=1'; + for (const [k, v] of Object.entries(opts.extra || {})) { + url += '&' + k + '=' + encodeURIComponent(v); + } + for (const [k, v] of Object.entries(opts.params || {})) { + url += '&' + k + '=' + encodeURIComponent(v); + } + return url; +} + +/** Drives the progress-bearing JSON-lines read loop: decode, line split, + * `JSON.parse` per line, trailing-buffer flush, malformed-line skip — + * byte-for-byte the loop formerly inlined in `runQuery`. `onLine` fires per + * parsed object, `onChunk` once per network chunk. A single `TextDecoder` + * used with `{ stream: true }` for the whole body (not per-chunk) so a + * multi-byte UTF-8 character split across two byte chunks still decodes + * correctly. */ +async function streamLines(body: ReadableStream, cbs: StreamCallbacks): Promise { + const reader = body.getReader(); + const decoder = new TextDecoder(); + let buffer = ''; + for (;;) { + const { done, value } = await reader.read(); + if (done) break; + buffer += decoder.decode(value, { stream: true }); + const lines = buffer.split('\n'); + buffer = lines[lines.length - 1]; + for (const line of lines.slice(0, -1)) { + if (!line) continue; + let json: StreamLine; + try { + json = JSON.parse(line); + } catch { + continue; + } + cbs.onLine && cbs.onLine(json); + } + cbs.onChunk && cbs.onChunk(); + } + if (buffer.trim()) { + try { + cbs.onLine && cbs.onLine(JSON.parse(buffer)); + } catch { + /* trailing partial line */ + } + } +} + +/** The current custom HTTP implementation of `ClickHouseTransport`. `deps`' + * accessors are read per-request (REQUIRED-PURE — see the contract's doc + * comment) so a live, mutable `origin`/`fetch` (e.g. `ConnectionSession`'s + * `chCtx`, mutated in place on sign-in) is always observed at its current + * value, never pinned to a stale snapshot (Adaptation A5). */ +export function createHttpTransport(deps: TransportDeps): ClickHouseTransport { + return { + async send(request: TransportRequest): Promise { + const url = chUrl(deps.origin(), { + format: request.defaultFormat, + extra: request.settings, + params: request.params, + }); + return deps.fetch()(url, { + method: 'POST', + body: request.sql, + headers: { Authorization: request.authorization }, + signal: request.signal, + }); + }, + streamLines, + }; +} diff --git a/src/net/clickhouse-transport.types.ts b/src/net/clickhouse-transport.types.ts new file mode 100644 index 00000000..09827020 --- /dev/null +++ b/src/net/clickhouse-transport.types.ts @@ -0,0 +1,97 @@ +// Issue #585 Phase 1 — the narrow SQL Browser ClickHouse-transport contract. +// Type-only (ADR-0002 phase-0 convention for seam contracts, hence the +// `.types.ts` suffix rather than the issue's suggested `clickhouse-transport.ts` +// — "exact names may follow repository conventions" per the issue). Puts the +// CURRENT custom HTTP implementation behind a contract a future official +// transport (Phase 2, gated on a new decision — ADR-0005 is Rejected) could +// also satisfy, without moving any product SQL or auth/lifecycle policy here. +// +// Ownership boundary: this file (and its implementation, +// `clickhouse-http-transport.ts`) may depend only on `src/core` (the narrow +// `StreamLine` type) — never on `ch-client.ts`, `oauth.ts`, `oauth-config.ts`, +// `src/application/`, or `src/ui/`, even type-only. `build/check-boundaries.mjs` +// enforces this mechanically (twin `RULES` entries for this file and the +// implementation file). + +import type { StreamLine } from '../core/stream.js'; + +/** What the transport is allowed to see of the environment. Deliberately + * excludes tokens, refresh, epochs, lifecycle callbacks: a transport + * implementation is compile-time incapable of ACQUIRING credentials or + * signaling lifecycle. (It still receives the resolved Authorization header + * per request — Adaptation A6 — so single-send/no-retry/no-caching discipline + * is contract- and test-enforced, not compiler-enforced.) Accessors, not + * snapshots — the live chCtx's origin is mutated in place on sign-in and the + * transport must observe the current value per request. + * + * REQUIRED-PURE: both accessors must be synchronous, side-effect-free plain + * property reads (production: `() => ctx.fetch` / `() => ctx.origin`). This + * matters because `send` evaluates them AFTER `authedFetch`'s final epoch + * fence and before the fetch itself; the type system cannot express purity, + * so — exactly like A6's single-send discipline — this rule is enforced by + * this doc comment and review, not by the compiler or the existing epoch + * race test (whose proof stops at the `send` invocation boundary). */ +export interface TransportDeps { + fetch(): typeof fetch; + origin(): string; +} + +/** One ClickHouse HTTP request, fully specified. No client-level defaults + * exist: `authorization` is the complete header value (scheme + credential), + * resolved by the caller (SQL Browser auth policy) for THIS request. */ +export interface TransportRequest { + /** Opaque SQL text. The transport never parses, rewrites, or appends to it + * (hard invariant 16: an authored FORMAT clause always wins over + * `defaultFormat` server-side, exactly as today). */ + sql: string; + /** Exact ClickHouse format name sent as `default_format`. */ + defaultFormat: string; + /** HTTP query-string settings (wait_end_of_query, max_result_rows, + * result_overflow_mode, add_http_cors_header, readonly, …) — the caller's + * policy decides which; the transport only serializes. */ + settings?: Record; + /** Query-string params riding alongside: native `param_*` parameters, + * `query_id`, `session_id`, `role` — today's exact wire vocabulary, + * unchanged (Adaptation A2). */ + params?: Record; + /** Complete Authorization header value. Never optional, never defaulted. */ + authorization: string; + signal?: AbortSignal; +} + +// No TransportResponse type in Phase 1 (Adaptation A3): `send` resolves with +// the NATIVE fetch `Response`. A structural subset would be assignable only in +// the direction Response -> subset, so authedFetch/exportQuery could not keep +// their `Promise` signatures without an unsafe cast. Native Response +// gives raw bytes (`body`, hard invariant 17) and `clone()` for authedFetch's +// non-destructive error-body peek for free. + +/** Callbacks driving `streamLines`' progress-bearing JSON-lines read loop. */ +export interface StreamCallbacks { + onLine?: (line: StreamLine) => void; + onChunk?: () => void; +} + +/** The SQL Browser transport contract. In Phase 1 exactly one implementation + * exists (`createHttpTransport`, `clickhouse-http-transport.ts`); a Phase 2 + * official-client implementation (does not proceed without a new decision) + * would satisfy the same contract. */ +export interface ClickHouseTransport { + /** POST one query; resolves at HTTP settlement (headers received) with the + * NATIVE fetch `Response` — Phase 1 defines no adapter-owned response type + * (Adaptation A3), which is what preserves `authedFetch`/`exportQuery`'s + * `Promise` signatures and `export-service.ts`'s + * `streamToFile(resp: Response, …)` consumer without casts. Exactly one + * fetch invocation (contract-suite-asserted, incl. on non-2xx — A6); no + * retry, no token read, no lifecycle callback, no error classification, no + * body consumption. HTTP error statuses resolve (they are responses); only + * network I/O failure / abort rejects. */ + send(request: TransportRequest): Promise; + /** Supported-stream mechanics for the progress-bearing JSON-lines formats: + * drives the read loop (decode, line split, JSON.parse, trailing-buffer + * flush, malformed-line skip), invoking onLine per parsed object and + * onChunk per network chunk — byte-for-byte the loop currently inlined in + * runQuery. Consuming a body is a caller decision made AFTER policy has + * classified the settled response. */ + streamLines(body: ReadableStream, cbs: StreamCallbacks): Promise; +} diff --git a/tests/unit/ch-client.test.ts b/tests/unit/ch-client.test.ts index 721551f3..15c95541 100644 --- a/tests/unit/ch-client.test.ts +++ b/tests/unit/ch-client.test.ts @@ -113,14 +113,14 @@ describe('chUrl', () => { describe('authedFetch', () => { it('throws + signals out when no token', async () => { const ctx = ctxWith(() => jsonResp({}), { getToken: async () => null }); - await expect(authedFetch(ctx, 'u', 'sql')).rejects.toThrow('not signed in'); + await expect(authedFetch(ctx, { sql: 'sql', defaultFormat: 'JSONStringsEachRowWithProgress' })).rejects.toThrow('not signed in'); expect(ctx.onSignedOut).toHaveBeenCalled(); }); it('cancels without signaling auth loss when a missing-token request becomes stale', async () => { let epoch = 1; const token = deferred(); const ctx = ctxWith(() => jsonResp({}), { currentEpoch: () => epoch, getToken: () => token.promise }); - const pending = authedFetch(ctx, 'u', 'sql'); + const pending = authedFetch(ctx, { sql: 'sql', defaultFormat: 'JSONStringsEachRowWithProgress' }); epoch = 2; token.resolve(null); await expect(pending).rejects.toMatchObject({ name: 'AbortError' }); @@ -137,7 +137,7 @@ describe('authedFetch', () => { onTransportConnected, onTransportOffline, }); - const pending = authedFetch(ctx, 'u', 'sql'); + const pending = authedFetch(ctx, { sql: 'sql', defaultFormat: 'JSONStringsEachRowWithProgress' }); epoch = 2; token.resolve('replacement-token'); await expect(pending).rejects.toMatchObject({ name: 'AbortError' }); @@ -155,27 +155,27 @@ describe('authedFetch', () => { return 'Bearer replacement-' + token; }, }); - await expect(authedFetch(ctx, 'u', 'sql')).rejects.toMatchObject({ name: 'AbortError' }); + await expect(authedFetch(ctx, { sql: 'sql', defaultFormat: 'JSONStringsEachRowWithProgress' })).rejects.toMatchObject({ name: 'AbortError' }); expect(ctx.fetchMock).not.toHaveBeenCalled(); expect(ctx.onSignedOut).not.toHaveBeenCalled(); }); it('returns the response on success', async () => { const ctx = ctxWith(async () => jsonResp({ ok: 1 })); - const r = await authedFetch(ctx, 'u', 'sql'); + const r = await authedFetch(ctx, { sql: 'sql', defaultFormat: 'JSONStringsEachRowWithProgress' }); expect(r.ok).toBe(true); expect(ctx.fetchMock.mock.calls[0][1].headers.Authorization).toBe('Bearer tok'); }); it('reports a successful current transport connection when lifecycle hooks are supplied', async () => { const onTransportConnected = vi.fn(); const ctx = ctxWith(async () => jsonResp({ ok: 1 }), { currentEpoch: () => 4, onTransportConnected }); - await authedFetch(ctx, 'u', 'sql'); + await authedFetch(ctx, { sql: 'sql', defaultFormat: 'JSONStringsEachRowWithProgress' }); expect(onTransportConnected).toHaveBeenCalledTimes(1); }); it('reports a non-abort fetch rejection as transport-offline', async () => { const failure = new Error('network unavailable'); const onTransportOffline = vi.fn(); const ctx = ctxWith(async () => { throw failure; }, { currentEpoch: () => 4, onTransportOffline }); - await expect(authedFetch(ctx, 'u', 'sql')).rejects.toBe(failure); + await expect(authedFetch(ctx, { sql: 'sql', defaultFormat: 'JSONStringsEachRowWithProgress' })).rejects.toBe(failure); expect(onTransportOffline).toHaveBeenCalledWith(failure); }); it('does not report a stale fetch rejection as transport-offline', async () => { @@ -188,7 +188,7 @@ describe('authedFetch', () => { fetchStarted.resolve(); return rejectedFetch.promise; }, { currentEpoch: () => epoch, onTransportOffline }); - const pending = authedFetch(ctx, 'u', 'sql'); + const pending = authedFetch(ctx, { sql: 'sql', defaultFormat: 'JSONStringsEachRowWithProgress' }); await fetchStarted.promise; epoch = 2; rejectedFetch.reject(failure); @@ -198,18 +198,30 @@ describe('authedFetch', () => { it('does not report HTTP failures or caller cancellation as transport-offline', async () => { const onTransportOffline = vi.fn(); const httpCtx = ctxWith(async () => textResp('server error', false, 500), { onTransportOffline }); - await authedFetch(httpCtx, 'u', 'sql'); + await authedFetch(httpCtx, { sql: 'sql', defaultFormat: 'JSONStringsEachRowWithProgress' }); const controller = new AbortController(); controller.abort(); const abortCtx = ctxWith(async () => { throw new Error('cancelled request'); }, { onTransportOffline }); - await expect(authedFetch(abortCtx, 'u', 'sql', controller.signal)).rejects.toThrow('cancelled request'); + await expect(authedFetch(abortCtx, { sql: 'sql', defaultFormat: 'JSONStringsEachRowWithProgress', signal: controller.signal })).rejects.toThrow('cancelled request'); expect(onTransportOffline).not.toHaveBeenCalled(); }); it('does not report an AbortError rejection as transport-offline', async () => { const onTransportOffline = vi.fn(); const abortError = Object.assign(new Error('cancelled request'), { name: 'AbortError' }); const ctx = ctxWith(async () => { throw abortError; }, { onTransportOffline }); - await expect(authedFetch(ctx, 'u', 'sql')).rejects.toBe(abortError); + await expect(authedFetch(ctx, { sql: 'sql', defaultFormat: 'JSONStringsEachRowWithProgress' })).rejects.toBe(abortError); + expect(onTransportOffline).not.toHaveBeenCalled(); + }); + it('rejects a malformed URL param synchronously without a token read, a fetch, or an offline signal (review pass 2, P1)', async () => { + const onTransportOffline = vi.fn(); + const ctx = ctxWith(async () => jsonResp({ ok: 1 }), { onTransportOffline }); + // A lone UTF-16 surrogate makes `chUrl`'s `encodeURIComponent` throw a + // URIError — request-preparation failure, not a network failure. + await expect( + authedFetch(ctx, { sql: 'sql', defaultFormat: 'JSONStringsEachRowWithProgress', params: { x: '\ud800' } }), + ).rejects.toBeInstanceOf(URIError); + expect(ctx.getToken).not.toHaveBeenCalled(); + expect(ctx.fetchMock).not.toHaveBeenCalled(); expect(onTransportOffline).not.toHaveBeenCalled(); }); it('refreshes once on 401 then retries', async () => { @@ -218,7 +230,7 @@ describe('authedFetch', () => { refresh: vi.fn(async () => true), getToken: vi.fn(async () => (n === 0 ? 'old' : 'new')), }); - const r = await authedFetch(ctx, 'u', 'sql'); + const r = await authedFetch(ctx, { sql: 'sql', defaultFormat: 'JSONStringsEachRowWithProgress' }); expect(r.ok).toBe(true); expect(ctx.refresh).toHaveBeenCalledTimes(1); }); @@ -227,7 +239,7 @@ describe('authedFetch', () => { async () => textResp('Code: 516. DB::Exception: Authentication failed', false, 403), { refresh: async () => false }, ); - await expect(authedFetch(ctx, 'u', 'sql')).rejects.toThrow('signed out'); + await expect(authedFetch(ctx, { sql: 'sql', defaultFormat: 'JSONStringsEachRowWithProgress' })).rejects.toThrow('signed out'); expect(ctx.onSignedOut).toHaveBeenCalledTimes(1); // Not overridden by `over` in this test, so it's still ctxWith's default // `vi.fn()` — the cast just recovers `.mock` from the (over-widened, since @@ -239,7 +251,7 @@ describe('authedFetch', () => { }); it('marks the ctx authenticated on a successful response', async () => { const ctx = ctxWith(async () => jsonResp({ ok: 1 })); - await authedFetch(ctx, 'u', 'sql'); + await authedFetch(ctx, { sql: 'sql', defaultFormat: 'JSONStringsEachRowWithProgress' }); expect(ctx.authConfirmed).toBe(true); }); it('fences a stale successful response from lifecycle and authentication state', async () => { @@ -251,7 +263,7 @@ describe('authedFetch', () => { fetchStarted.resolve(); return response.promise; }, { currentEpoch: () => epoch, onTransportConnected }); - const pending = authedFetch(ctx, 'u', 'sql'); + const pending = authedFetch(ctx, { sql: 'sql', defaultFormat: 'JSONStringsEachRowWithProgress' }); await fetchStarted.promise; epoch = 2; response.resolve(jsonResp({ ok: 1 })); @@ -272,7 +284,7 @@ describe('authedFetch', () => { refresh: vi.fn(async () => true), onTransportOffline, }); - const pending = authedFetch(ctx, 'u', 'sql'); + const pending = authedFetch(ctx, { sql: 'sql', defaultFormat: 'JSONStringsEachRowWithProgress' }); await fetchStarted.promise; epoch = 2; response.resolve(jsonResp({}, false, 401)); @@ -298,7 +310,7 @@ describe('authedFetch', () => { currentEpoch: () => epoch, refresh: vi.fn(async () => true), }); - const pending = authedFetch(ctx, 'u', 'sql'); + const pending = authedFetch(ctx, { sql: 'sql', defaultFormat: 'JSONStringsEachRowWithProgress' }); await bodyStarted.promise; epoch = 2; body.resolve('jwt::token_verification_exception'); @@ -318,7 +330,7 @@ describe('authedFetch', () => { return refresh.promise; }), }); - const pending = authedFetch(ctx, 'u', 'sql'); + const pending = authedFetch(ctx, { sql: 'sql', defaultFormat: 'JSONStringsEachRowWithProgress' }); await refreshStarted.promise; expect(ctx.refresh).toHaveBeenCalledTimes(1); epoch = 2; @@ -342,7 +354,7 @@ describe('authedFetch', () => { return freshToken.promise; }), }); - const pending = authedFetch(ctx, 'u', 'sql'); + const pending = authedFetch(ctx, { sql: 'sql', defaultFormat: 'JSONStringsEachRowWithProgress' }); await freshTokenStarted.promise; epoch = 2; freshToken.resolve('new'); @@ -361,7 +373,7 @@ describe('authedFetch', () => { return refresh.promise; }), }); - const pending = authedFetch(ctx, 'u', 'sql'); + const pending = authedFetch(ctx, { sql: 'sql', defaultFormat: 'JSONStringsEachRowWithProgress' }); await refreshStarted.promise; epoch = 2; refresh.resolve(false); @@ -385,7 +397,7 @@ describe('authedFetch', () => { currentEpoch: () => epoch, refresh: vi.fn(async () => false), }); - const pending = authedFetch(ctx, 'u', 'sql'); + const pending = authedFetch(ctx, { sql: 'sql', defaultFormat: 'JSONStringsEachRowWithProgress' }); await textStarted.promise; epoch = 2; denialText.resolve('Code: 516. Authentication failed'); @@ -396,7 +408,7 @@ describe('authedFetch', () => { // e.g. SHOW CREATE USER → HTTP 403 / UNKNOWN_USER, mid-session. const ctx = ctxWith(async () => textResp('Code: 192. DB::Exception: There is no user x', false, 403), { authConfirmed: true }); - const resp = await authedFetch(ctx, 'u', 'sql'); + const resp = await authedFetch(ctx, { sql: 'sql', defaultFormat: 'JSONStringsEachRowWithProgress' }); expect(resp.status).toBe(403); expect(ctx.onSignedOut).not.toHaveBeenCalled(); expect(ctx.refresh).not.toHaveBeenCalled(); @@ -407,19 +419,19 @@ describe('authedFetch', () => { async () => (n++ === 0 ? textResp('jwt::token_verification_exception', false, 500) : jsonResp({ ok: 1 })), { refresh: vi.fn(async () => true) }, ); - const r = await authedFetch(ctx, 'u', 'sql'); + const r = await authedFetch(ctx, { sql: 'sql', defaultFormat: 'JSONStringsEachRowWithProgress' }); expect(r.ok).toBe(true); }); it('returns a non-auth error response unchanged', async () => { const ctx = ctxWith(async () => textResp('syntax error', false, 400)); - const r = await authedFetch(ctx, 'u', 'sql'); + const r = await authedFetch(ctx, { sql: 'sql', defaultFormat: 'JSONStringsEachRowWithProgress' }); expect(r.status).toBe(400); }); it('uses a provided authHeader (e.g. Basic) instead of Bearer', async () => { const ctx = ctxWith(async () => jsonResp({ ok: 1 }), { authHeader: (t) => 'Basic ' + t.toUpperCase(), }); - await authedFetch(ctx, 'u', 'sql'); + await authedFetch(ctx, { sql: 'sql', defaultFormat: 'JSONStringsEachRowWithProgress' }); expect(ctx.fetchMock.mock.calls[0][1].headers.Authorization).toBe('Basic TOK'); }); }); @@ -442,6 +454,103 @@ describe('queryJson', () => { }); }); +// Issue #585 Phase 1 — `authedFetch`'s centralized entry snapshot (pass-5 +// revision): `extra`/`params` are captured synchronously the instant +// `authedFetch` is entered, before its first await (`ctx.getToken()`), +// preserving today's pre-await `chUrl` serialization timing under ONE +// mechanism instead of per-call-site defensive spreads. `queryJson` is the +// caller that historically forwarded these by reference +// (`ch-client.ts:218-228` pre-refactor), so these cases exercise it directly. +describe('queryJson — invocation-time settings/params capture (authedFetch\'s entry snapshot)', () => { + it('captures extra/params synchronously at entry — mutating the caller\'s live objects while token acquisition is pending does not reach the initial request', async () => { + const token = deferred(); + const extra: Record = { readonly: 2 }; + const params: Record = { param_id: '1' }; + const ctx = ctxWith(async () => jsonResp({ data: [] }), { getToken: () => token.promise }); + const pending = queryJson(ctx, 'SELECT 1', undefined, extra, params); + // Mutate the caller's own objects while the token await is still pending. + extra.readonly = 999; + params.param_id = 'mutated'; + token.resolve('tok'); + await pending; + const url = ctx.fetchMock.mock.calls[0][0]; + expect(url).toContain('readonly=2'); + expect(url).toContain('param_id=1'); + expect(url).not.toContain('readonly=999'); + expect(url).not.toContain('mutated'); + }); + + it('keeps the invocation-time snapshot across the one-refresh retry — mutating the caller\'s objects during the pending refresh does not reach the retried request', async () => { + let n = 0; + const extra: Record = { readonly: 2 }; + const params: Record = { param_id: '1' }; + const refreshStarted = deferred(); + const refresh = deferred(); + const ctx = ctxWith(async () => (n++ === 0 ? jsonResp({}, false, 401) : jsonResp({ data: [] })), { + refresh: vi.fn(async () => { + refreshStarted.resolve(); + return refresh.promise; + }), + }); + const pending = queryJson(ctx, 'SELECT 1', undefined, extra, params); + await refreshStarted.promise; + // Mutate while the refresh (and the post-refresh getToken it gates) is pending. + extra.readonly = 999; + params.param_id = 'mutated'; + refresh.resolve(true); + await pending; + expect(ctx.fetchMock).toHaveBeenCalledTimes(2); + const secondUrl = ctx.fetchMock.mock.calls[1][0]; + expect(secondUrl).toContain('readonly=2'); + expect(secondUrl).toContain('param_id=1'); + expect(secondUrl).not.toContain('readonly=999'); + expect(secondUrl).not.toContain('mutated'); + }); +}); + +// Issue #585 Phase 1, Adaptation A5 — `TransportDeps.origin()` is an +// accessor read live per request, never snapshotted; the live, mutable +// `ctx.origin` (mutated in place on sign-in — `connection-session.ts`) stays +// authoritative through `ch-client.ts`'s own `transportFor` wiring, not just +// in `createHttpTransport`'s isolated unit tests. +describe('transportFor — live origin authority through production wiring (Adaptation A5)', () => { + it('reads ctx.origin live per request across two independent authedFetch calls', async () => { + const ctx = ctxWith(async () => jsonResp({ data: [] })); + await queryJson(ctx, 'SELECT 1'); + ctx.origin = 'https://new-cluster.example'; + await queryJson(ctx, 'SELECT 1'); + expect(ctx.fetchMock.mock.calls[0][0]).toContain('https://ch.example'); + expect(ctx.fetchMock.mock.calls[1][0]).toContain('https://new-cluster.example'); + }); + + // The case above reconstructs a transport per authedFetch call, so it can't + // by itself distinguish a live-read `deps.origin()` from one snapshotted at + // `createHttpTransport` construction time (sabotage case 2). This one uses + // the SAME transport instance across authedFetch's one-refresh retry loop + // (transportFor(ctx) is called once per authedFetch invocation, before the + // retry loop) to prove the origin read is live per SEND, not per + // construction. + it('reads ctx.origin live per send within one authedFetch retry cycle, not once at transport construction', async () => { + let n = 0; + const refreshStarted = deferred(); + const refresh = deferred(); + const ctx = ctxWith(async () => (n++ === 0 ? jsonResp({}, false, 401) : jsonResp({ data: [] })), { + refresh: vi.fn(async () => { + refreshStarted.resolve(); + return refresh.promise; + }), + }); + const pending = queryJson(ctx, 'SELECT 1'); + await refreshStarted.promise; + ctx.origin = 'https://new-cluster.example'; + refresh.resolve(true); + await pending; + expect(ctx.fetchMock).toHaveBeenCalledTimes(2); + expect(ctx.fetchMock.mock.calls[0][0]).toContain('https://ch.example'); + expect(ctx.fetchMock.mock.calls[1][0]).toContain('https://new-cluster.example'); + }); +}); + describe('loadServerVersion', () => { it('returns the version string', async () => { const ctx = ctxWith(async () => jsonResp({ data: [{ v: '26.3.1', u: 1 }] })); @@ -1001,6 +1110,20 @@ describe('killQueryWithLease', () => { await killQueryWithLease(frozen, null, sqlString); expect(fetchMock).not.toHaveBeenCalled(); }); + + it('never reads getToken/refresh even if the lease object happens to carry them (hard invariant 8, defense in depth)', async () => { + const { lease: frozen } = leaseFetch(async () => jsonResp({ data: [] })); + const getToken = vi.fn(); + const refresh = vi.fn(); + // `AuthenticatedCancellationLease` has exactly 4 fields; this widens the + // VALUE (not killQueryWithLease's declared parameter type) to prove the + // one-shot lease-scoped transport genuinely never calls these, rather + // than merely never being GIVEN them. + const leaseWithExtras = Object.freeze({ ...frozen, getToken, refresh }); + await killQueryWithLease(leaseWithExtras, 'scope-q', sqlString); + expect(getToken).not.toHaveBeenCalled(); + expect(refresh).not.toHaveBeenCalled(); + }); }); describe('exportQuery', () => { diff --git a/tests/unit/clickhouse-http-transport.test.ts b/tests/unit/clickhouse-http-transport.test.ts new file mode 100644 index 00000000..5fba628d --- /dev/null +++ b/tests/unit/clickhouse-http-transport.test.ts @@ -0,0 +1,171 @@ +import { describe, expect, it, vi } from 'vitest'; +import { chUrl, createHttpTransport } from '../../src/net/clickhouse-http-transport.js'; +import type { StreamLine } from '../../src/core/stream.js'; +import { runTransportContractSuite } from './clickhouse-transport-contract.js'; + +// Issue #585 Phase 1 — direct spec for the moved current-HTTP transport +// implementation. Registers the shared contract suite once (this is the +// ONLY implementation Phase 1 registers — see the suite factory's header +// comment), then covers the mechanics the shared suite deliberately leaves +// implementation-specific: `chUrl`'s exact URL shape and the moved +// progress-line stream loop, including two edge cases a manual move can +// silently alter (a split multi-byte UTF-8 character across byte chunks, +// and per-chunk onLine-before-onChunk ordering). + +runTransportContractSuite('createHttpTransport', createHttpTransport); + +function deps(fetchImpl: (url: string, init: RequestInit) => Response | Promise, origin = 'https://ch.example') { + const fetchMock = vi.fn(fetchImpl); + return { fetchMock, deps: { fetch: () => fetchMock as unknown as typeof fetch, origin: () => origin } }; +} + +// A stream that yields exactly the given byte chunks, in order — needed for +// the UTF-8-split case, which a whole-string-per-chunk helper is structurally +// incapable of producing. +function byteStream(chunks: Uint8Array[]): ReadableStream { + let i = 0; + return new ReadableStream({ + pull(controller) { + if (i < chunks.length) controller.enqueue(chunks[i++]); + else controller.close(); + }, + }); +} + +function stringStream(chunks: string[]): ReadableStream { + return byteStream(chunks.map((c) => new TextEncoder().encode(c))); +} + +describe('chUrl', () => { + it('uses default format and compression', () => { + expect(chUrl('https://o')).toBe('https://o?default_format=JSONStringsEachRowWithProgress&enable_http_compression=1'); + }); + it('applies format, extra and params', () => { + const url = chUrl('https://o', { format: 'JSON', extra: { wait_end_of_query: 1 }, params: { x: 'a b' } }); + expect(url).toContain('default_format=JSON'); + expect(url).toContain('wait_end_of_query=1'); + expect(url).toContain('x=a%20b'); + }); +}); + +describe('createHttpTransport().send — exact request shape', () => { + it('builds the exact literal URL from origin/format/settings/params, POSTs the SQL body, and sends the complete Authorization header', async () => { + const { fetchMock, deps: d } = deps(() => new Response('ok')); + const transport = createHttpTransport(d); + await transport.send({ + sql: 'SELECT 1', + defaultFormat: 'JSONCompact', + settings: { wait_end_of_query: 1 }, + params: { param_id: '5', query_id: 'q1', session_id: 's1', role: 'analyst' }, + authorization: 'Bearer tok', + }); + const [url, init] = fetchMock.mock.calls[0] as [string, RequestInit]; + expect(url).toBe( + 'https://ch.example?default_format=JSONCompact&enable_http_compression=1' + + '&wait_end_of_query=1¶m_id=5&query_id=q1&session_id=s1&role=analyst', + ); + expect(init.method).toBe('POST'); + expect(init.body).toBe('SELECT 1'); + expect((init.headers as Record).Authorization).toBe('Bearer tok'); + }); + + it('threads the abort signal through to fetch', async () => { + const controller = new AbortController(); + const { fetchMock, deps: d } = deps(() => new Response('ok')); + const transport = createHttpTransport(d); + await transport.send({ sql: 'x', defaultFormat: 'JSON', authorization: 'Bearer t', signal: controller.signal }); + expect((fetchMock.mock.calls[0][1] as RequestInit).signal).toBe(controller.signal); + }); +}); + +describe('createHttpTransport().streamLines — moved progress-line loop mechanics', () => { + it('reassembles a line split across multiple chunks', async () => { + const { deps: d } = deps(() => new Response('ok')); + const lines: StreamLine[] = []; + const stream = stringStream(['{"row":{"a"', ':"1"}}\n']); + await createHttpTransport(d).streamLines(stream, { onLine: (l) => lines.push(l) }); + expect(lines).toEqual([{ row: { a: '1' } }]); + }); + + it('skips empty lines between JSON objects', async () => { + const { deps: d } = deps(() => new Response('ok')); + const lines: StreamLine[] = []; + const stream = stringStream(['\n\n{"row":{"a":"1"}}\n\n']); + await createHttpTransport(d).streamLines(stream, { onLine: (l) => lines.push(l) }); + expect(lines).toEqual([{ row: { a: '1' } }]); + }); + + it('skips a malformed JSON line without throwing', async () => { + const { deps: d } = deps(() => new Response('ok')); + const lines: StreamLine[] = []; + const stream = stringStream(['not json\n', '{"row":{"a":"1"}}\n']); + await createHttpTransport(d).streamLines(stream, { onLine: (l) => lines.push(l) }); + expect(lines).toEqual([{ row: { a: '1' } }]); + }); + + it('flushes a valid trailing partial line (no terminating newline)', async () => { + const { deps: d } = deps(() => new Response('ok')); + const lines: StreamLine[] = []; + const stream = stringStream(['{"row":{"a":"1"}}']); + await createHttpTransport(d).streamLines(stream, { onLine: (l) => lines.push(l) }); + expect(lines).toEqual([{ row: { a: '1' } }]); + }); + + it('discards a malformed trailing partial line without throwing', async () => { + const { deps: d } = deps(() => new Response('ok')); + const lines: StreamLine[] = []; + const stream = stringStream(['{bad trailing']); + await expect(createHttpTransport(d).streamLines(stream, { onLine: (l) => lines.push(l) })).resolves.toBeUndefined(); + expect(lines).toEqual([]); + }); + + it('delivers an in-band exception line via onLine (hard invariant 10 feed-through)', async () => { + const { deps: d } = deps(() => new Response('ok')); + const lines: StreamLine[] = []; + const stream = stringStream(['{"exception":"DB::Exception: boom"}\n']); + await createHttpTransport(d).streamLines(stream, { onLine: (l) => lines.push(l) }); + expect(lines).toEqual([{ exception: 'DB::Exception: boom' }]); + }); + + it('calls onChunk once per network read', async () => { + const { deps: d } = deps(() => new Response('ok')); + let chunkCalls = 0; + const stream = stringStream(['{"row":{}}\n', '{"row":{}}\n', '{"row":{}}\n']); + await createHttpTransport(d).streamLines(stream, { onChunk: () => { chunkCalls++; } }); + expect(chunkCalls).toBe(3); + }); + + it('fires every onLine callback produced from a chunk before that chunk\'s onChunk (call-order)', async () => { + const { deps: d } = deps(() => new Response('ok')); + const order: string[] = []; + const stream = stringStream(['{"row":{"a":"1"}}\n{"row":{"a":"2"}}\n']); + await createHttpTransport(d).streamLines(stream, { + onLine: () => order.push('line'), + onChunk: () => order.push('chunk'), + }); + expect(order).toEqual(['line', 'line', 'chunk']); + }); + + it('tolerates entirely missing onLine/onChunk callbacks', async () => { + const { deps: d } = deps(() => new Response('ok')); + const stream = stringStream(['{"row":{}}\n', '{"row":{}}']); + await expect(createHttpTransport(d).streamLines(stream, {})).resolves.toBeUndefined(); + }); + + it('decodes a multi-byte UTF-8 character split across two byte chunks correctly (single TextDecoder with {stream:true})', async () => { + const { deps: d } = deps(() => new Response('ok')); + const line = '{"row":{"a":"€"}}\n'; // '€' — 3 UTF-8 bytes: 0xE2 0x82 0xAC + const fullBytes = new TextEncoder().encode(line); + let euroIdx = -1; + for (let i = 0; i < fullBytes.length - 2; i++) { + if (fullBytes[i] === 0xe2 && fullBytes[i + 1] === 0x82 && fullBytes[i + 2] === 0xac) { euroIdx = i; break; } + } + expect(euroIdx).toBeGreaterThan(-1); + const splitAt = euroIdx + 1; // split INSIDE the euro sign's 3-byte sequence + const chunk1 = fullBytes.slice(0, splitAt); + const chunk2 = fullBytes.slice(splitAt); + const lines: StreamLine[] = []; + await createHttpTransport(d).streamLines(byteStream([chunk1, chunk2]), { onLine: (l) => lines.push(l) }); + expect((lines[0].row as Record).a).toBe('€'); + }); +}); diff --git a/tests/unit/clickhouse-transport-contract.ts b/tests/unit/clickhouse-transport-contract.ts new file mode 100644 index 00000000..cf6e95f5 --- /dev/null +++ b/tests/unit/clickhouse-transport-contract.ts @@ -0,0 +1,163 @@ +// Issue #585 Phase 1 — reusable `ClickHouseTransport` contract-test-suite +// FACTORY. Deliberately NOT itself a `.test.ts` spec: vitest's `include` glob +// (`tests/vitest.config.ts`) is `tests/unit/**/*.test.{js,ts}`, so this file +// is never discovered directly — a real spec imports and calls +// `runTransportContractSuite(name, makeTransport)` to register it against one +// concrete implementation. +// +// Phase 1 registers this suite against `createHttpTransport` ONLY +// (`clickhouse-http-transport.test.ts`), because no official implementation +// exists yet and Phases 2-4 do not proceed without a new decision +// (docs/ADR-0005-clickhouse-web-client.md is Rejected). A future Phase 2 +// transport would import this same factory and re-run it unchanged against +// its own `makeTransport` — "structurally ready", not implementation-neutral: +// today's contract is current-HTTP-specific (native `Response` return, +// complete pre-resolved `authorization` header — Adaptations A3/A6), so this +// suite proves today's ONE implementation satisfies the contract, not that +// the contract already generalizes to a hypothetical second one. + +import { describe, expect, it, vi } from 'vitest'; +import type { ClickHouseTransport, TransportDeps, TransportRequest } from '../../src/net/clickhouse-transport.types.js'; + +type FetchImpl = (url: string, init: RequestInit) => Response | Promise; +type HeadersRecord = Record; + +/** A concrete `ClickHouseTransport` implementation under test, built from a + * `TransportDeps` this factory constructs and controls (so cases can flip + * `setOrigin` mid-test — Adaptation A5 / sabotage case 2 — and inspect every + * call the implementation made to the stub fetch). */ +export type MakeTransport = (deps: TransportDeps) => ClickHouseTransport; + +function baseRequest(overrides: Partial = {}): TransportRequest { + return { + sql: 'SELECT 1', + defaultFormat: 'JSON', + authorization: 'Bearer tok', + ...overrides, + }; +} + +export function runTransportContractSuite(name: string, makeTransport: MakeTransport): void { + describe(`ClickHouseTransport contract — ${name}`, () => { + function harness(fetchImpl: FetchImpl) { + const fetchMock = vi.fn(fetchImpl); + let origin = 'https://ch.example'; + const transport = makeTransport({ + fetch: () => fetchMock as unknown as typeof fetch, + origin: () => origin, + }); + return { transport, fetchMock, setOrigin: (o: string) => { origin = o; } }; + } + + it('serializes default_format, settings, params, and the complete Authorization header into one POST whose body is the SQL byte-identical', async () => { + const { transport, fetchMock } = harness(() => new Response('ok')); + await transport.send(baseRequest({ + sql: 'SELECT 1 FORMAT CSV', // an authored FORMAT clause — never appended to again + defaultFormat: 'JSONCompact', + settings: { wait_end_of_query: 1 }, + params: { param_id: '5', query_id: 'q1', session_id: 's1', role: 'analyst' }, + authorization: 'Bearer secret-token', + })); + expect(fetchMock).toHaveBeenCalledTimes(1); + const [url, init] = fetchMock.mock.calls[0] as [string, RequestInit]; + expect(url).toContain('default_format=JSONCompact'); + expect(url).toContain('enable_http_compression=1'); + expect(url).toContain('wait_end_of_query=1'); + expect(url).toContain('param_id=5'); + expect(url).toContain('query_id=q1'); + expect(url).toContain('session_id=s1'); + expect(url).toContain('role=analyst'); + expect(init.method).toBe('POST'); + expect(init.body).toBe('SELECT 1 FORMAT CSV'); + expect((init.headers as HeadersRecord).Authorization).toBe('Bearer secret-token'); + }); + + it('omits settings/params from the URL when absent (no client-level defaults)', async () => { + const { transport, fetchMock } = harness(() => new Response('ok')); + await transport.send(baseRequest()); + const url = fetchMock.mock.calls[0][0] as string; + expect(url).toBe('https://ch.example?default_format=JSON&enable_http_compression=1'); + }); + + it('carries each send\'s own authorization value with no state cached between sends', async () => { + const { transport, fetchMock } = harness(() => new Response('ok')); + await transport.send(baseRequest({ authorization: 'Bearer first' })); + await transport.send(baseRequest({ authorization: 'Bearer second' })); + expect((fetchMock.mock.calls[0][1] as RequestInit).headers as unknown as HeadersRecord).toEqual({ Authorization: 'Bearer first' }); + expect((fetchMock.mock.calls[1][1] as RequestInit).headers as unknown as HeadersRecord).toEqual({ Authorization: 'Bearer second' }); + }); + + it('invokes the stub fetch exactly once per send — no internal retry or header caching — on a 2xx response', async () => { + const { transport, fetchMock } = harness(() => new Response('ok', { status: 200 })); + await transport.send(baseRequest()); + expect(fetchMock).toHaveBeenCalledTimes(1); + }); + + it('invokes the stub fetch exactly once per send on a non-2xx response too', async () => { + const { transport, fetchMock } = harness(() => new Response('denied', { status: 403 })); + await transport.send(baseRequest()); + expect(fetchMock).toHaveBeenCalledTimes(1); + }); + + it('resolves (never throws) on a non-2xx response, with the body reaching the caller byte-identical', async () => { + const { transport } = harness(() => new Response('{"exception":"Code: 60. DB::Exception: table not found"}', { status: 500 })); + const resp = await transport.send(baseRequest()); + expect(resp.status).toBe(500); + expect(await resp.text()).toBe('{"exception":"Code: 60. DB::Exception: table not found"}'); + }); + + it('rejects with the network/abort failure rather than resolving, for an aborted signal', async () => { + const controller = new AbortController(); + controller.abort(); + const { transport } = harness((_url, init) => { + if ((init.signal as AbortSignal | undefined)?.aborted) { + const err = Object.assign(new Error('aborted'), { name: 'AbortError' }); + return Promise.reject(err); + } + return Promise.resolve(new Response('ok')); + }); + await expect(transport.send(baseRequest({ signal: controller.signal }))).rejects.toMatchObject({ name: 'AbortError' }); + }); + + it('surfaces a mid-stream abort from streamLines rather than swallowing it', async () => { + const { transport } = harness(() => new Response('ok')); + const abortError = Object.assign(new Error('aborted'), { name: 'AbortError' }); + const stream = new ReadableStream({ + start(controller) { controller.error(abortError); }, + }); + await expect(transport.streamLines(stream, {})).rejects.toBe(abortError); + }); + + it('preserves response status and arbitrary headers, including X-ClickHouse-Summary', async () => { + const { transport } = harness(() => new Response('ok', { + status: 200, + headers: { 'X-ClickHouse-Summary': '{"read_rows":"1"}' }, + })); + const resp = await transport.send(baseRequest()); + expect(resp.status).toBe(200); + expect(resp.headers.get('X-ClickHouse-Summary')).toBe('{"read_rows":"1"}'); + }); + + it('never consumes the response body inside send — bodyUsed stays false and the raw bytes are still readable', async () => { + const bytes = new TextEncoder().encode('raw-bytes'); + const stream = new ReadableStream({ + start(controller) { controller.enqueue(bytes); controller.close(); }, + }); + const { transport } = harness(() => new Response(stream, { status: 200 })); + const resp = await transport.send(baseRequest()); + expect(resp.bodyUsed).toBe(false); + const reader = resp.body!.getReader(); + const { value } = await reader.read(); + expect(value).toEqual(bytes); + }); + + it('reads deps.origin() live per request instead of snapshotting it at construction time (Adaptation A5)', async () => { + const { transport, fetchMock, setOrigin } = harness(() => new Response('ok')); + await transport.send(baseRequest()); + setOrigin('https://new-cluster.example'); + await transport.send(baseRequest()); + expect(fetchMock.mock.calls[0][0]).toContain('https://ch.example'); + expect(fetchMock.mock.calls[1][0]).toContain('https://new-cluster.example'); + }); + }); +} diff --git a/tests/unit/client-web-spike-policy.test.js b/tests/unit/client-web-spike-policy.test.js index f64fae7e..b612fc04 100644 --- a/tests/unit/client-web-spike-policy.test.js +++ b/tests/unit/client-web-spike-policy.test.js @@ -24,7 +24,7 @@ import { describe, expect, it } from 'vitest'; import { mkdtemp, readFile, readdir, rm, writeFile } from 'node:fs/promises'; import { tmpdir } from 'node:os'; -import { dirname, extname, join, relative, resolve } from 'node:path'; +import { dirname, extname, join, relative, resolve, sep } from 'node:path'; import { fileURLToPath } from 'node:url'; import { buildArtifact } from '../../build/build.mjs'; import { @@ -181,6 +181,63 @@ describe('no CDN or remote-URL import in spike sources', () => { }); }); +// Issue #585 Phase 1 — mirrors build/check-boundaries.mjs's bare-specifier +// ban on `@clickhouse/client-web` under `src/**`. This environment's npm +// config sets `ignore-scripts=true`, so a bare `npm test` never runs +// `pretest`/`check:arch` (CLAUDE.md hard rule 4) — mirroring the ban here +// means the coverage-gated suite enforces it too. This is an INDEPENDENTLY +// implemented scanner, not a shared one: `check-boundaries.mjs` exports +// nothing today, so unifying the two behind one exported scanner would be a +// real improvement but is a cross-cutting tooling change out of scope for +// this pure-refactor PR (tracked separately, `inbox`-labeled, per CLAUDE.md +// "Working discipline") — the two scanners can drift, the same accepted risk +// tests/unit/dashboard-boundaries.test.js already documents for its own +// mirror of the same script. Belt and braces: metafile (build-graph truth, +// tested above), source scan (this — fast, IDE-time), `check:arch` (gate/CI). +const SRC_DIR = resolve(projectRoot, 'src'); +const CLIENT_WEB_IMPORT_PATTERNS = [ + /\bimport\s+[\w*{}\s,]+\s+from\s*['"](@clickhouse\/client-web(?:\/[^'"]*)?)['"]/g, + /\bexport\s+[\w*{}\s,]+\s+from\s*['"](@clickhouse\/client-web(?:\/[^'"]*)?)['"]/g, + /\bimport\s*['"](@clickhouse\/client-web(?:\/[^'"]*)?)['"]/g, + /\bimport\s*\(\s*['"](@clickhouse\/client-web(?:\/[^'"]*)?)['"]/g, +]; +// Names the FUTURE official transport file (does not exist yet) — activates +// correctly the moment it is born, same as build/check-boundaries.mjs's own +// allowlist. +const CLIENT_WEB_ALLOWLIST = new Set(['src/net/clickhouse-web-transport.ts']); + +async function collectSrcSourceFiles(dir) { + const entries = await readdir(dir, { withFileTypes: true }); + const files = []; + for (const entry of entries) { + const full = join(dir, entry.name); + if (entry.isDirectory()) { + files.push(...await collectSrcSourceFiles(full)); + } else if (['.ts', '.tsx', '.js', '.mjs'].includes(extname(entry.name))) { + files.push(full); + } + } + return files; +} + +describe('no @clickhouse/client-web import anywhere under src/ (mirrors build/check-boundaries.mjs)', () => { + it('scans every src/** source file for a bare @clickhouse/client-web import, outside the not-yet-existing official-transport allowlist', async () => { + const files = await collectSrcSourceFiles(SRC_DIR); + expect(files.length).toBeGreaterThan(0); + const offenders = []; + for (const file of files) { + const relFile = relative(projectRoot, file).split(sep).join('/'); + if (CLIENT_WEB_ALLOWLIST.has(relFile)) continue; + const text = await readFile(file, 'utf8'); + for (const pattern of CLIENT_WEB_IMPORT_PATTERNS) { + pattern.lastIndex = 0; + if (pattern.test(text)) offenders.push(relFile); + } + } + expect(offenders).toEqual([]); + }); +}); + // Plan §31 "Wiki reconciliation": "The evidence validator and normal unit // test must reject a mismatch among: ADR Status; machine-readable decision; // wiki status text; wiki ADR link." This sub-task only owns the normal-