.ingest....`, `.hist..`, `.dlq..`).
- **nats_permprobe.go** — The verifier's permission probe: a pull and an unpin to each shard durable that the server rejects on their merits, so a refused one (reported only to the connection's error handler) names a shard the user may not pull from or unpin.
- **natstest/** — Test code that stands up NATS as an operator deploys it, from the shipped `deployments/nats` values and manifests: the config for a server, with the file store the Helm chart would give it (in process, or in the integration suite's container) and the operator's hand on it (applying the manifests, the lease bucket included; deleting a durable or the bucket; reading which process holds a lease). It lets `internal/app` and `tests/integration` run against a real server without importing NATS themselves.
- **embedded.go** — `EmbeddedNATS`, the in-process `Broker`: an in-process NATS server with JetStream, giving each tenant a queue of its own — stream `INGEST_` with subjects `ingest..>`, capped at the tenant's `mq.max_bytes_gb` (`DiscardNew`) and remembering idempotency keys for `EmbeddedDuplicateWindow` (two minutes, sized to `2 × the dedupe lease + 1s` — the in-flight `503` sends the full lease as `Retry-After`, so an obedient client's retry can land up to ~2×lease after the original `Reserve`, and the `+1s` covers a backend whose claim expiry itself rounds up by that much), and stream `DLQ_` (`dlq..>`, `DiscardOld`) at a tenth of it — with the durable consumers on the ingest one; nothing outside the package sees that layout. JetStream's own check of the streams' caps against the disk (75% of the free disk by default) is set out of reach, so a budget is a cap and never a reservation. Boot deletes the pair an earlier build kept for every tenant together — its subjects overlap every tenant's — and takes stock of the tenants' streams on disk with their budgets, so a consumer created later is held on every one, a tenant no longer served included. `SetMaxBytes` opens a tenant's queue the first time — its dead-letter stream first, so no row is queued that could not be parked — and every registered consumer joins it; a queue a consumer cannot join is not recorded open, `SetMaxBytes` returning the error and `MaxBytes` not reporting the budget, so the next reload tries the consumers that had not joined again. A publish or park that finds a stream missing reopens it at the budget last asked for the tenant, and so does a publish to a queue the broker has not recorded open — one a consumer could not join, or an open that timed out, which can leave a stream JetStream creates after all, which no consumer holds — either refused as a full queue with none asked yet. Publishes and parks that find the same queue not open share one attempt (`singleflight`), and after one fails the tenant's publishes and parks are refused at once for five seconds rather than each trying again under the broker's lock, which every tenant's open, resize and reload takes; a reload retries regardless. After that `SetMaxBytes` applies a reloaded budget to the tenant's two live streams as a pair: if the DLQ update fails after the ingest one succeeded, the ingest resize is undone so both stay on the previous budget — best effort, since if that undo also fails the ingest stream stays at the new limit and the DLQ at the previous, and the error says so. A dead-letter stream is never capped below the bytes it holds, which `DiscardOld` would delete to fit ([#532](https://github.com/Wave-RF/WaveHouse/issues/532)): it keeps what it holds, and that is logged. Its JetStream calls are bounded to ten seconds — plus ten more for the consumers joining a queue they do not hold yet, and five for the rollback of a failed resize, each a budget of its own rather than the one that just expired — since a reload holds the settings store's lock while its hooks run; `MaxBytes` reports the budget last applied in full, so a failed resize is retried by the next reload. The consumers `CreateConsumer` and `Subscribe` build hold one durable on each tenant's stream, looked up before anything is written so a boot over many queues writes nothing it need not, each delivering on a goroutine of its own into the one handler. `PurgeAcked` and `DeadLetterCounts` run per tenant stream. `Stats` reports connection and inbound-message counters for `observability.RegisterSystemMetrics`. Trace context rides in the message headers: `Publish` applies `observability.InjectHeaders`, and a message delivered through `Subscribe` (the hub bridge) carries `observability.ExtractHeaders` on its `Ctx`; the worker's `Consumer` path skips the extraction, since it batches across messages and reads no per-message context.
- **mqtest/** — The conformance suite for `Broker` (`mqtest.Run`): the behavior the rest of the process relies on — publish and consume round trips with names that need encoding, per-tenant order, redelivery, dead-lettering and its counts, replay bounds and isolation, the one `failed` report of a consumer whose delivery ends underneath it — checked through the interfaces alone, with no stream or subject name in sight. Each implementation runs it from a test of its own — the embedded one from `mqtest/embedded_test.go`, a test binary apart from `internal/mq`'s so the two share no 15s budget — handing it a fresh broker per case and flags (`mqtest.Caps`) for the few places where backends legitimately differ: whether a full queue refuses its own tenant alone, whether `PurgeAcked` removes anything, whether a tenant never given a budget has a dead-letter queue to report on, and whether `CreateConsumer` configures the durable or only finds one.
### `observability/` — OpenTelemetry Pipeline
- **provider.go** — `InitProvider(ctx, serviceName, ProviderConfig)` wires the OTel pipeline. Each output is independently gated; the W3C TraceContext + Baggage propagator is always installed (cheap, harmless when traces are off). Returns `(shutdown, promHandler http.Handler, err)` — `promHandler` is non-nil only when `PrometheusEnabled` is true and reads from a *private* `prometheus.Registry` to avoid leaking the process/Go collectors that `prometheus.DefaultRegisterer` auto-registers. OTLP-metrics push (`MetricsEnabled`) and Prometheus exposition (`PrometheusEnabled`) are independent: either, both, or neither may be set, and any combination produces a single MeterProvider feeding the active readers. The OTLP exporters (traces / metrics-OTLP / logs) take their endpoint from the `OTEL_EXPORTER_OTLP_*` env vars, not from `ProviderConfig`, and are only constructed when their signal is enabled, so Prometheus-only operation dials nothing even with those vars set. Provider init in `internal/app` runs whenever `otel.enabled` OR `prometheus.enabled` is true, so Prometheus-only operation (Alloy/scrape, no collector) is a first-class mode.
- **logger.go** — `NewLogger(component, level, isJSON, otlpSampleRate)` produces a slog logger that fans out to stdout (always 100%) and the OTLP log exporter (DEBUG/INFO sampled at `otlpSampleRate`, WARN/ERROR always 100% as a non-configurable safety floor). `TraceHandler` injects `trace_id`/`span_id` from the active span when one exists. `otlpSamplerFn` is exposed (lowercase) for unit testing the per-level rate logic without driving through the slogmulti middleware.
- **metrics.go** — `RegisterSystemMetrics(mqStats, pebbleStats)` registers observable gauges for the MQ's NATS connections and in-msgs (the embedded server's, or under `nats` this process's client connection), and Pebble dedupe storage stats. Both are functions read on every scrape — `mqStats` a `func() (MQStats, error)` (`mq.Broker.Stats` in production; nil skips the MQ gauges), `pebbleStats` a `func() map[string]int64` (`dedupe.Embedded.Stats`, the one instance's figures; nil, or a nil map while it is closed, skips the Pebble gauges) — this package never holds the NATS server or a store. Wired in `internal/app` after the providers are up.
- **tracer.go** — W3C TraceContext propagation over message headers (`InjectHeaders` / `ExtractHeaders` on a plain `map[string][]string`, the shape NATS and HTTP headers share) — `internal/mq` injects on every publish and extracts onto the delivered `Message.Ctx` on the `Subscribe` path; no consumer reads it yet (the SSE hub bridge forwards the bytes and starts no span), and the ingest worker's `Consumer` path skips extraction entirely (see `embedded.go` above), so nothing downstream of the queue is linked to the originating request span.
The package's design invariants — stdout always 100%, WARN+ERROR always export at 100%, gRPC exporters dial lazily so unreachable collectors never block startup, private Prometheus registry — are documented in AGENTS.md "Key Design Decisions" #15 and must be preserved by anything touching this package.
### `policy/` — Access Control
- **policy.go** — Hasura-style policy types, now **role-first**: `TablePolicy` is `map[string]RolePermissions`, and a role's grant carries a separate `SelectPermissions` and `InsertPermissions` — so a field only one side ever honored (`filter`, aggregations and the `max_*` limits on select; `check` on insert) does not exist on the other, and a document that puts one there fails the strict decode as an unknown key. `Evaluate()` resolves permissions against JWT claims (including `{{ jwt.claim.path }}` template resolution) for ONE operation, leaving the side it did not resolve **nil**. That is what the pointers buy: an *empty* side means "no restrictions" — what the admin return constructs on both sides — while a *nil* side means "you asked the wrong operation", and as value types the two were the same zero value. Every accessor fails closed on a nil side. The handful of bare field reads outside this package each sit past an accessor that denies an unresolved side first, so a nil `Select` never reaches one; if that ordering ever changed they would panic rather than silently widen. A nil guard that skips such a read must never be added, since an absent `WhereClause` is an unfiltered query. The per-column decision `IsColumnAllowed(col, insert)` takes the side it is being asked about, alongside its batch/projection forms `AllowedProjection()` and `RestrictsColumns()`, `IsAggregationAllowed()`, `CheckClauses()` — the write-side accessor for the one consumer that iterates a side's map instead of asking about a column, whose `ok=false` a caller must treat as *refuse the write*, never as *no checks to run* — `resolvePredicates()` — the one resolution both read surfaces render from, so the SQL `WHERE` and the in-memory row check cannot drift — and `Validate()`, split into `validateSelectPerms`/`validateInsertPerms` and run from `settings.Validate` on every adoption, which is where the rules in [Access Control](/access-control) are actually enforced.
- **rowfilter.go** — the in-memory row-visibility twin of the SQL `WHERE`: `HasRowFilter`, `RowVisible` (evaluates the resolved predicates against a decoded event, per subscriber), and `ColumnSpec` — the per-column comparison contract (`ColumnKind` `Numeric`/`Text`/`Time`/`Opaque`, plus each kind's parameters: the caller-supplied instant parser for `Time`, the `NumericSpec` storage model for `Numeric`) whose zero value is the fail-closed floor: numeric columns compare in the column's **storage domain** (operands rendered by canonical.go, compared by numeric.go — next two bullets), `String` bytewise, `DateTime`/`DateTime64` chronologically (both operands through the ingest grammar; either side unreadable ⇒ withheld), and everything else (including any column with no usable schema) admits byte-equality only, failing `!=`/`>`/`<` closed. Both `HasRowFilter` and `RowVisible` fail closed on a denied or unresolved grant: `HasRowFilter` is the gate in front of `RowVisible`, so it must answer *true* there or the whole-bucket fast path skips the check entirely.
- **canonical.go** — the one rendering layer for comparison operands: every value a `filter` or `check` compares — a JWT claim (`CanonicalScalar`), a policy-authored literal (`CanonicalNumericLiteral`), an ingested payload value (`numericCanonical`) — converges on one exact canonical decimal form (positional, digit-bounded, never a float64 round-trip), so what a read filter binds and what the stream compares can't drift; `scalarString` is the deliberate exception, the raw byte rendering that `Text`/`Opaque` equality compares.
- **numeric.go** — compares canonical forms the way the column that stores them would: `compareCanonicalDecimals` orders by exact digit-string arithmetic, and `NumericSpec` first narrows both operands the way ClickHouse narrows the stored value and the bound constant — `Float32`/`Float64` width rounding, `Decimal` scale truncation, integers exact at any width, with an operand outside the column's width or a `Decimal`'s precision budget refused rather than modeled; the `tests/integration` differential oracle holds in-range verdicts equal to a live ClickHouse's and asserts the never-admit-where-SQL-hides direction for the refused out-of-range operands.
- **source.go** — `Source`, a `func() *Policy` the `/v1/ops` gate (over a flat settings directory) reads per call, so a settings reload applies to the very next request; in production it is the default tenant's `settings.Store.Policy`, and `Static(p)` fixes one for tests. The tenant-aware surfaces take a keyed variant that resolves to the same `Store.Policy`: `api.PolicySource` (`func(*settings.Store) *policy.Policy`) for ingest, structured query and pipes, `stream.PolicySource` (`func(tenant.ID) *policy.Policy`) for the hub, and `auth.PolicySource` (same shape) for the operator key's admin role. A `nil` result is a deliberate lockout.
### `pipes/` — Named Query Pipes
- **pipes.go** — `NamedQuery` type with SQL template and parameter definitions, and `Source` (`Pipe(name)` / `Pipes()`), read per request — `settings.Store` in production (`pipes.json`), `Static(q...)` in tests. `BindParams()` resolves `{{param}}` / `{{param:default}}` placeholders by inlining escaped literal values into the SQL (strings single-quote-escaped; arrays rendered as escaped `(…)` `IN`-lists). A non-scalar value with no SQL form (a JSON object, or an empty array) is rejected rather than emitted raw.
### `query/` — Structured Query Engine
- **ast.go** — `StructuredQuery` AST types: columns, aggregations, filters, group by, order by, limit, time range.
- **builder.go** — `Build()` converts AST to parameterized SQL. It is the single chokepoint that validates every referenced identifier against the schema **and** authorizes every column reference — projection, aggregation args, filters, group_by, order_by, time_range — against the role's column allowlist (the [#223](https://github.com/Wave-RF/WaveHouse/issues/223) hard cap). A full-row read is requested with `select_all`, which expands to the role's allowed columns rather than emitting a raw `SELECT *`; an omitted projection selects nothing, and `*` in `columns` is a literal column name. Every identifier is backtick-quoted via `internal/chsql` (`QuoteIdent`) so any ClickHouse-legal name is accepted — a name containing `?` is refused fail-closed ([#279](https://github.com/Wave-RF/WaveHouse/issues/279)). The role's row-level-security predicate and `max_rows` cap are emitted by `Build()` itself, as part of the WHERE and LIMIT assembly — policy SQL is never spliced into rendered text ([#322](https://github.com/Wave-RF/WaveHouse/issues/322)). Timestamp bucketing for cache optimization.
### `settings/` — Settings Directory
The hot-reloadable half of configuration: a directory of four JSON files (`config.json`, `roles.json`, `policies.json`, `pipes.json`) — or, [nested](/deployment#the-nested-settings-directory), one folder of those four per tenant — that the server validates at boot and re-adopts while running. `Validate` is the single gate (strict decode, per-file shape rules, and the cross-file check that every role a policy grant or pipe allowlist names is declared in `roles.json`), a `Store` holds one tenant's adopted `Document` as one atomic snapshot, and the `Registry` maps tenant ids to stores and owns every reload. `Store` is also the runtime authority for access control and pipes: it implements `policy.Source` (`Store.Policy`, nil when `policies.json` is `{}` — fail closed) and `pipes.Source`, and there is no other copy — the files are the only write path. See [Settings Directory](/settings-directory).
- **validate.go** — `ValidateDir(dir)` reads, decodes, and checks one directory of the four files — a flat root, or one tenant's folder — in one pass (strict JSON — unknown fields and duplicate keys are errors; per-file shape rules; cross-file role references) and returns every `Finding` at once. Shared, by way of `Validate`, by `wavehouse validate`, boot, and every reload.
- **finding.go** — `Finding` / `Severity`: errors make the directory invalid, warnings don't block adoption. The JSON shape is part of the ops API (`POST /v1/ops/settings/reload` returns them).
- **tree.go** — `Validate(root)` reads the directory's shape off its entries and checks either one: a root holding any of the four file names is flat and goes through `ValidateDir` untouched; otherwise a root holding a folder is nested, each folder name going through `tenant.Parse` and each folder through `ValidateDir`, with the folder leading every finding's `File` (`acme/policies.json`). The `Tree` it returns is nil when the finding is about the root itself — it cannot be listed, or a nested root holds a loose file or an entry that cannot be stat'ed.
- **store.go** — `Store` is a passive holder: one tenant's adopted document behind an atomic pointer, swapped by the registry, which stamps it with the id of the tenant it created the store for (`Tenant()`, how a handler names its tenant to a per-tenant resource). Consumers read typed accessors per call (`ClickHouse()`, `Auth()`, `DedupeFor(table)`, `DLQFor(table)`, `Keepalive()`, …) rather than holding values.
- **registry.go** — `Registry` maps a tenant id to its `Store` and owns everything that changes one. `Open` validates and adopts at boot; `Reload` re-validates the whole directory and `ReloadTenant` one tenant's folder, serialized with each other; `AfterAdopt` hooks run after every reload the registry applied, with the tenants it adopted — none when it only rejected or removed one, which a consumer holding a resource per tenant needs to hear of too; `For(id)` and `All()` see only the tenants being served, `Known()` every tenant held, rejected ones included, and `Resolve(id)` tells a rejected tenant from an unknown one. The shape is fixed at `Open`. Flat: an invalid directory refuses boot, and a rejected reload keeps the previous snapshot. Nested: fail closed per tenant — a folder with an error finding stops being served (the store keeps its document for requests already admitted, and gets the next good one) while the rest carry on; a whole-directory reload mirrors the folders, down to none (an emptied directory is not a change of shape); and a finding about the directory itself refuses boot or rejects the reload whole, leaving every tenant as it was; at `Open`, a nested directory that would serve no tenant (every folder rejected, or none named by a tenant id) refuses boot too, while a reload to the same tree drops every tenant and keeps running. The tenant map is replaced whole by a reload, so a lookup is one lock-free load.
- **watch.go** — `Registry.Watch`, which `internal/app` starts for a flat directory only: fsnotify on the *directory* (not the files, so atomic-writer replaces and Kubernetes ConfigMap symlink swaps aren't lost), debounced into one reload; reloads once as soon as the watch exists so an edit between the boot read and the watch is never missed. `SIGHUP` and the reload endpoint funnel through the same serialized `Reload`.
- **seed.go** / **seed/** — The embedded (`go:embed`) starter directory with every key at its default. The binary carries no compiled defaults except that a missing `dedupe.retention` means `"0"`: `wavehouse bootstrap [dir]` writes this seed, and the compose stack and e2e fixture ship copies of it.
### `tenant/` — Tenant Identifier
- **tenant.go** — `ID`, a validated string (never a number: a 19-digit id already rounds as a float64), and `Parse`, the one grammar that makes an id safe both as a folder name and as a message-queue subject token: ASCII letters, digits, `_`, `-`, at most `MaxLen` (64) bytes. `Default` (`"0"`) is the tenant a request without the header resolves to; `Header` is `X-Tenant-ID`. The package imports nothing from the rest of the repository, so any package can name a tenant. HTTP handlers receive the tenant as its resolved `*settings.Store`, which knows its id (`Store.Tenant`) for the topics they publish and subscribe on; the stream hub and the ingest worker read each event's tenant off its `mq.Topic` — the leading subject token — and their settings getters take it as a parameter, which `internal/app` resolves through the registry; the sweeper hands the MQ each tenant's own gap window (`gapWindows`, a rejected tenant's included); each served tenant has a schema registry of its own, built with its id (story 6).
### `chconn/` — ClickHouse Connection Pools
- **chconn.go** — `Pools` holds one `Manager` per distinct connection tuple among the served tenants — `Identity{Addr, Database, Username, Password, TLS}`, a plain comparable value, the map key — reconciled from the settings registry's `AfterAdopt` hook after every reload ([#583](https://github.com/Wave-RF/WaveHouse/issues/583) story 6): tenants naming one tuple share its pool, sized to the largest `max_open_conns` and `max_idle_conns` among them (`Sizes`); a tenant whose tuple changed is repointed; a tuple no tenant names is released after the longest `query_timeout` among the tenants it had; a changed largest ask is a `Resize` with the same grace. The walk keeps the boot config's `clickhouse.max_total_conns` — the ceiling on the open pools' `max_open_conns` together — at every step, tenants no longer served leaving first and what was refused placed once more at the end: a refused resize keeps the pool's size, and a tuple that cannot be opened (the ceiling, a certificate file that cannot be read, or options the driver refuses — the pool opens as the walk places its first tenant, at the largest ask among the tenants naming it when that fits the ceiling and otherwise at that tenant's own, so each of these is undone in place) leaves its tenants on the pool they had — `Params` and all, so their `Target` stays whole — or on none; `NewPools` refuses boot on any refusal, `Reconcile` returns them joined for the wiring to log, and the next reload retries. `Manager` is a `driver.Conn` over one tuple's pool whose backing connection `Resize` swaps; like `clickhouse.Open` it never dials, so boot tolerates an unreachable ClickHouse (schema discovery retries) and a bad address surfaces where reachability is already handled (`/readyz`, query errors). The `tls` block is the tuple's, read once into one `tls.Config` handed to the driver when `tls.enabled` and carried on each tenant's `Target` for the https hop. Resolution is per tenant: `For` (the `driver.Conn`, nil for a tenant on no pool — the wiring returns an untyped nil), `Target` (the tenant's own `http_port`, `http_scheme` and `headers` over its pool's host, credentials, database and TLS config, from the `Params` last applied for it), `SharingTables` (the tenants on the same address and database, whatever their user — the cache fan-out's rule) and `Ping` (every pool at once, nil at the first answer). The HTTP-interface consumers (ingest INSERTs, raw-SQL proxy) take their `http.Client` from an `HTTPClients` cache, one client per TLS config ever handed to it, since the proxy serves tenants on different configs in alternation.
- **errclass.go** — `Classify`, what a failed ClickHouse request says about the request: `Unavailable` (connection refused/reset, timeouts, and the exception codes of a server that cannot take work — `TIMEOUT_EXCEEDED`, `TOO_MANY_SIMULTANEOUS_QUERIES`, `MEMORY_LIMIT_EXCEEDED`, `TOO_MANY_PARTS`, `READONLY`, `TABLE_IS_READ_ONLY`, `KEEPER_EXCEPTION`, …), `Denied` (`AUTHENTICATION_FAILED`, `ACCESS_DENIED`, …: the identity, not the request), `Rejected` (any other exception code — the server read the request and refused it), or `Unknown` (no exception code, and no failure recognizable as the way to ClickHouse). It reads the driver's `*clickhouse.Exception`/`*clickhouse.HTTPError` and the HTTP interface's `HTTPError` (`NewHTTPError`: the code from `X-ClickHouse-Exception-Code`, else the body's `Code: NNN.`), so it is one answer for the ingest worker and the query handlers (`api/ch_errors.go`); the worker retries every class but `Rejected`. `TableScoped` and `Splittable` refine a retried failure: the first names a failure of one table (read-only, too many parts or mutations, a missing grant), the second one a multi-row batch can earn by its size alone (`TOO_MANY_PARTS` for too many partitions in one INSERT, `MEMORY_LIMIT_EXCEEDED`), which the worker splits row by row before retrying.
### `chsql/` — ClickHouse SQL Helpers
- **chsql.go** — Dependency-free ClickHouse SQL helpers shared by `query/` and `policy/`, kept in their own package to break an import cycle. `QuoteIdent` is the single place every identifier — column, table, alias — becomes SQL text: always backtick-quoted and escaped, so any ClickHouse-legal name (dots, spaces, unicode, keywords) is safe. `BindUnsafe` reports whether a name contains a literal `?`, which would desync clickhouse-go's positional binder; such names are rejected fail-closed rather than silently mis-bound.
### `keyenc/` — Key Escaping
- **keyenc.go** — The one escaping composite keys are built from, so a name can never be mistaken for a separator: `Escape` keeps ASCII letters, digits, `_` and `-` — exactly the tenant-id grammar, so a tenant id is its own escaped form — and writes every other byte as `%XX` (uppercase hex); `Unescape` is `url.PathUnescape`, which decodes `%XX` in either case and takes any other byte as itself, so a `%2D` for `-` that an earlier build wrote still reads. `Join`/`AppendJoin` escape each field and put a separator between them, panicking on no fields and on a separator the escaping could write or one outside ASCII, and `Split` reverses them. The package that builds a key takes raw names and escapes them itself, so no caller has to: NATS subject tokens (`internal/mq`), the cache's keys — the version index and the shared backend's Redis keys (`internal/cache`) — and the dedupe keys (`internal/dedupe`, `/`-separated) use it. Keys built from it are stored, so changing what it keeps orphans them — and on the shared backend, whose keys every process builds for itself, it splits them for the length of a rolling upgrade: a bump one build makes does not reach the entries the other build filed, which are served until their TTL.
## Data Flows
### Ingest Path
```text wrap=false
Client POST /v1/ingest?table={table}
→ Tenant resolution: X-Tenant-ID → settings.Registry → the request's *settings.Store
(absent = tenant 0; 400 malformed / 404 unknown / 503 rejected, before auth)
→ JWT auth middleware (always runs; token optional)
→ refuseUnverifiable (503 + Retry-After: 30 for a token sent while the
tenant's JWKS has not been fetched yet; tokenless and operator-key
requests pass through)
→ Look up table schema from SchemaRegistry
→ Policy check: role allowed to insert into this table (before the body is parsed)
→ Resolve the declared Content-Type into the body's format (415 if absent,
unsupported, or if declarations disagree; before the body is read)
→ Read the whole body into a pooled buffer, bounded by the 16 MiB cap
(413 before any record is processed, so nothing is published)
→ Validate JSON body against schema (type checks, required columns)
→ Policy column rules + check clauses (disallowed columns rejected;
claim-derived values enforced or injected)
→ Canonicalize top-level DateTime/DateTime64 column values to RFC 3339 UTC
(rewrites the payload so every consumer shares one spelling; fail-open —
an unparseable value passes through verbatim for ClickHouse's parser to judge)
→ Optional dedupe: resolve the id (configurable ID field; a row missing it or
setting it to null is published un-deduped + logged/counted, or rejected
under require_id)
→ Encode the record; the steps below run per window of up to 256 records
→ Reserve the window's (tenant, table, id) keys in one call: a duplicate is
skipped, an id another request holds → 503 + Retry-After (dedupe.lease,
30s by default), a store that cannot answer → 503 + Retry-After: 5
→ Publish each record to NATS JetStream (ingest.{tenant}.{table}), a deduped
one under its idempotency key
→ Commit the published ids in one call; on a failed publish, commit the
records before it and release the rest (a publish whose outcome is unknown
keeps its claim until the lease lapses)
→ 200 OK returned immediately
→ (If the tenant's NATS stream is full, or not open: 503 + Retry-After header,
the id released)
Ingest worker pipeline (StartIngestWorker):
← JetStream pull consumer (buffer-consumer), one durable per tenant stream
(ingest.{tenant}.>), delivered into one handler
→ Parse the event envelope (an envelope the worker cannot read — malformed JSON,
an unknown or absent format, columns and row that don't pair — is parked on the
DLQ, or acked-and-dropped where the DLQ is off for the table; either way
counted by wavehouse_ingest_poison_total under its disposition)
→ Batch events per tenant table
→ At flush, a batch whose tenant has no ClickHouse connection is not inserted
(parkBatch takes it to the DLQ switch whole); otherwise bulk INSERT to ClickHouse
(INSERTs pin date_time_input_format=best_effort — the server default since
ClickHouse 26.5; see /ingest-pipeline for the basic-vs-best_effort divergence)
→ On success: DoubleAck messages
→ On failure ClickHouse could not take (down, overloaded, read-only, denied —
chconn.Classify): NakWithDelay the batch under the pool's backoff (the table's, for a
table-scoped code); never DLQ. A multi-row batch refused for its size
(chconn.Splittable) is split row by row first
→ On failure ClickHouse rejected: re-insert row by row; each row rejected again → DLQ output (dlq.{tenant}.{table}), then Ack to prevent infinite retry
(Insert-only pipeline. The wire format `EventMessage` carries only
{table_name, scope, received_timestamp, format, columns, row}; non-insert mutations
DELETE/UPDATE/TRUNCATE/DROP/etc. must go through POST /v1/ops/query (or an
operator-authored write pipe) — the /v1/ops/* RequireAdmin gate rejects
non-admin callers at the API layer, so
a no/invalid-token request (resolved to default_role, not admin in a
production config) cannot reach the proxy.)
Active Sweeper (async goroutine, every 60s, in the process holding the sweeper lease), on each tenant's stream:
→ Read buffer consumer's AckFloor (highest contiguous ACKed seq)
→ Binary search for first message within that tenant's own gap window
(a rejected tenant's as its folder last had it, unbounded for one rejected since boot; none for a removed tenant)
→ Purge target = MIN(ack_floor + 1, gap_window_seq)
→ Purge all messages below target from JetStream
```
### Query Path
```text
Client POST /v1/ops/query
→ JWT auth middleware (always runs, never rejects; a bad token yields an
empty role and stashes its verification error for the denying gate)
→ refuseUnverifiable (503 + Retry-After: 30 for a token sent while tenant 0's
JWKS has not been fetched yet — the ops tree resolves no tenant; tokenless
and operator-key requests pass through)
→ policy.ResolveRole (empty role → default_role — the one sanctioned
roleless exception)
→ /v1/ops RequireAdmin (resolved role == policy.admin_role, or the
operator-key bit) — single gate shared with the rest of /v1/ops/*
(pipes inspection, settings reload, schema discovery, DLQ stats). A denial is
401 when a stashed error shows the caller presented an invalid token,
else 403. Raw SQL has no per-statement scope check (a full SQL parser
would be needed to authorize predicates), so the role gate is the
entire authorization story. /v1/ops/query is the only surface for
ad-hoc non-SELECT statements (DELETE/UPDATE/TRUNCATE/DROP/ALTER/…);
non-admin callers use `POST /v1/ingest?table={table}` or a write pipe
that lists their role for writes, and the structured query endpoint or
named pipes for reads.
→ Decode {"sql": "..."} from the request body.
→ POST the SQL verbatim to ClickHouse's HTTP interface at
://:/?default_format=JSON
&date_time_output_format=iso&database=
Auth via X-ClickHouse-User / X-ClickHouse-Key headers.
Bound by a clickhouse.query_timeout context derived from the inbound request — client
disconnect cancels the upstream call.
→ ClickHouse parses the SQL natively and decides what to do:
→ Read: returns 200 + {"meta":[...], "data":[...], "rows":N,
"statistics":{...}} as JSON. The handler extracts `data` and
forwards just that array, preserving the [{...}, {...}] response
shape callers expect.
→ Mutation/DDL: returns 200 + empty body. The handler emits `[]` so
response shape stays "always an array."
→ Error: returns 4xx/5xx + plain-text error message + the
X-ClickHouse-Exception-Code header. The handler classes it by
that code (chconn.Classify), not by the HTTP status ClickHouse
uses for nearly everything: bad SQL → 400, a missing grant →
403, bad credentials → 502, an outage → 503 — with the trimmed
message, a `code` and `retryable` in the JSON error envelope.
→ Response carries Cache-Control: no-store so no downstream layer
(browser, CDN, corp proxy) caches the result.
```
The proxy-pattern wins are: zero classification logic on the WaveHouse side (no `IsMutation` heuristic to maintain), and any ClickHouse statement type — including verbs added in future versions and inline FORMAT overrides — works without WaveHouse code changes. Multi-statement input (`SELECT 1; TRUNCATE t`) is supported when the upstream ClickHouse has multi-query enabled, which is the default on recent versions; older or restrictively-configured servers will return a clear error from ClickHouse itself for the second statement. The proxy buffers the response in memory with a 64 MiB cap (502 with `clickhouse response exceeded N bytes` on overflow, to keep a runaway `SELECT *` from pinning RAM on the API server), and passes ClickHouse's `Content-Type` through when an inline `FORMAT` directive overrides the default JSON envelope. The structured query endpoint and pipes still go through `clickhouse-go`'s native driver (Query/Exec) for performance and to keep the cached row-array shape consistent.
### Streaming Path
```text
Client GET /v1/stream
→ Tenant resolution: X-Tenant-ID → settings.Registry → the request's *settings.Store
(absent = tenant 0; 400 malformed / 404 unknown / 503 rejected, before auth)
→ JWT auth middleware (always runs; token optional)
→ refuseUnverifiable (503 + Retry-After: 30 for a token sent while the
tenant's JWKS has not been fetched yet; tokenless and operator-key
requests pass through)
→ Announce the caller's projected column list as an `event: schema` frame
(no `id:`, so it never moves Last-Event-ID) BEFORE registering, so a client
on a quiet table learns its columns before any row arrives and a live event
during setup finds the announcement already recorded rather than repeating
it. This announcement is best-effort — skipped when the registry has no
schema for the table yet, or the role cannot read the table at all (a role
that CAN read it but whose column rules leave nothing readable is still
announced, with an empty list) — and what actually guarantees a row is
never sent without its names is the LIVE path's per-connection drift check
on every event
→ Register a Subscriber with the Stream Hub, keyed by (topic, role)
→ If ?since= / Last-Event-ID provided:
→ Create ephemeral NATS consumer with DeliverByStartTime
→ Send historical events (projected per-connection) first
→ Live events: MQ → Hub.Broadcast → projected & serialized ONCE per role
→ fan the finished frame to every Subscriber of that (topic, role);
a role carrying a row-level filter delivers per subscriber instead:
the shared frame goes only to subscribers whose JWT claims admit
the row (RowVisible)
→ a connection whose column list drifts (a schema change mid-stream) is
re-announced before the next row, per connection
→ Live and replay track that drift in SEPARATE state and do not reconcile
it (#543): replay keeps its own last-announced signature while the live
path recorded one at subscribe time, so a connection that gap-fills
ACROSS a column change can then receive live rows with no fresh
announcement. A client that checks arity drops the mismatched rows; a
same-length change (RENAME COLUMN, or a drop paired with an add) zips
under the wrong names until it reconnects
→ Handler drains keepalives + event frames from one byte-pump → client
→ Policy filtering (historical + live): denied tables skipped, denied
columns stripped, row filter evaluated per subscriber against claims.
Column projection runs once per role (Hub.Broadcast) — per-subscriber
work only where a row filter makes visibility per-connection; replay
shares the same column policy + row check but projects per-connection
```
## Technology Stack
| Component | Technology | Purpose |
| --------- | ---------- | ------- |
| Language | Go 1.26 | Core runtime |
| HTTP Router | Chi v5 | Request routing and middleware |
| Authentication | golang-jwt v5 + keyfunc v3 | JWT (HMAC + JWKS) parsing and validation |
| Analytics DB | ClickHouse | Primary data store + schema source of truth |
| Message Queue | NATS + JetStream | Durable event streaming |
| L1 Cache | Ristretto v2 | In-process memory cache |
| Shared Cache | [rueidis](https://github.com/redis/rueidis) | Redis-compatible client for the shared backend (`cache.backend: redis`) |
| Embedded KV | Pebble | Optional deduplication |
| Config | cleanenv | YAML + env var config loading |
| Release | GoReleaser | Cross-platform binary builds |
| Containers | Docker (distroless) | Minimal production images |