diff --git a/.goreleaser.yaml b/.goreleaser.yaml index d809ce7d..c304f9c2 100644 --- a/.goreleaser.yaml +++ b/.goreleaser.yaml @@ -82,6 +82,10 @@ dockers_v2: - dockerfile: deployments/Dockerfile.goreleaser ids: - wavehouse + # The seed settings directory, baked into the image at /app/settings + # (see the Dockerfile). Keep in lockstep with deployments/Dockerfile. + extra_files: + - internal/settings/seed images: - "ghcr.io/wave-rf/wavehouse" # Every build gets an immutable reference plus one moving pointer. diff --git a/.testcoverage.yml b/.testcoverage.yml index 669d65be..76e52337 100644 --- a/.testcoverage.yml +++ b/.testcoverage.yml @@ -72,13 +72,15 @@ exclude: # that would shrink main.go enough to drop this exclude entirely. - ^cmd/wavehouse/main\.go$ e2e: - # internal/settings and the validate CLI entry point are currently - # reachable only via `wavehouse validate`, which the e2e suite - # (server-focused) never runs — they sat at 0% and dragged the e2e - # gate below its floor. The unit suite covers both thoroughly, and - # per-suite excludes don't touch the merged total, so nothing is - # hidden project-wide. Drop these excludes once the settings - # directory is wired into the running server and e2e can exercise - # them. + # internal/settings is mostly validation: the e2e suite (server- + # focused) only walks the happy path — boot adopts a valid fixture + # directory — which measures ~46% for the package and pulls the e2e + # gate to 58.9%, under its 60% floor. The remaining branches (every + # rejection path, the seed writer) are unit territory and the unit + # suite covers them thoroughly; per-suite excludes don't touch the + # merged total, so nothing is hidden project-wide. Same for the + # settings CLI entry points (`validate`, `init-settings`), which e2e + # never runs. Revisit if e2e grows a reload-rejection scenario. - ^internal/settings/ - ^cmd/wavehouse/validate\.go$ + - ^cmd/wavehouse/init_settings\.go$ diff --git a/AGENTS.md b/AGENTS.md index 9967534a..1210d510 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -54,7 +54,7 @@ The invariant index — what must stay true. Full narrative and rationale live i 5. **Per-table batching** — the worker groups events by table and bulk-INSERTs in schema column order; each table's batch is independent. 6. **Dead Letter Queue** — failed batch inserts publish to `WAVEHOUSE_DLQ` (`dlq.`), gated by `dlq.enabled`. No silent data loss. 7. **Auth: always on, fail-loud, decoupled from authz (security)** — the JWT middleware always runs (no `auth.enabled`/`dev_mode` flag); it verifies with HMAC **or** JWKS (not both), with accepted `alg` pinned to the active verifier and checked before any key is used (rejects `alg:none` and cross-family confusion). No/invalid/expired token → empty role → policy `default_role`, with the bad-token reason stashed so a denying gate returns a loud `401`, not a bare `403`. Elevated access needs a valid granted role. **Sanctioned exception:** a configured non-JWT operator key (`auth.operator_key`; presented via `Authorization: Operator ` or the `X-Operator-Key` alias) deliberately couples authN+authZ — a constant-time match authorizes a full-access platform operator (stamps the admin role plus an operator bit) independent of the verifier (see #11). Detail: architecture.md § `api/` + `internal/auth`; see also #11, §Security Considerations. -8. **Optional dedup** — opt-in via `dedupe.enabled`; `dedupe.id_field` selects the JSON key. +8. **Optional dedup** — opt-in via `dedupe.enabled` (boot config, owns Pebble's lifecycle); `dedupe.id_field` in the settings directory's `config.json` selects the JSON key, overridable per table. 9. **Singleflight** — `TieredCache` coalesces concurrent misses (`x/sync/singleflight`) to prevent cache stampede. 10. **Active Sweeper** — purges NATS messages that are both ACKed (written to CH) and older than the gap window; SSE gap-fill uses `DeliverByStartTime`, no in-process ring buffer. 11. **Hasura-style access control: fail-closed (security)** — `policy.IsAdmin` (role == `admin_role`, **exact case-sensitive**, default `"admin"`) is the single admin check, shared by `Evaluate`/`ResolveRole`/`Validate`/the `/v1/ops` gate/`RoleAllowed`. Empty/absent role matches nothing (no `"*"` wildcard); `Validate` rejects empty role keys; a `nil` policy (deleted) denies **everyone incl. admin** via a role — a total lockout for token-based callers, so bootstrap from the policy file, never an implicit admin grant (**exception:** the operator key's `auth.IsOperator` bit passes the `/v1/ops` gate even under a `nil` policy — a deliberate break-glass restore over HTTP, see #7). `default_role` is the one sanctioned roleless exception (`ResolveRole` maps empty → it pre-eval); `default_role == admin_role` is permitted but dev-only and loudly warned (`policy.DefaultRoleGrantsAdmin`). Preserve when touching `internal/policy` (policy twin of #13; see #159). Detail: architecture.md § `policy/`. @@ -62,7 +62,7 @@ The invariant index — what must stay true. Full narrative and rationale live i 13. **Named query pipes: fail-closed (security)** — pre-defined SQL templates (Tinybird-style) with param binding + caching; `GET/POST /v1/pipes/{name}` sit outside `RequireAdmin`, so per-pipe `allowed_roles` is the *only* execute-path gate, via `policy.RoleAllowed`: exact allowlist membership (no `"*"`), admin always passes, empty/absent role and empty-string entries authorize nobody, and no `allowed_roles` → admin-only. Preserve and exercise via `testutil.RunRoleMatrix` / `StandardRoleMatrix` (see #159). Detail: architecture.md § `pipes/`. 14. **TypeScript SDK** — `@wavehouse/sdk`: typed query builder, real-time SSE over `fetch`, live queries (incrementable/decomposable/poll aggregation), codegen CLI. Exactly one runtime dependency — `eventsource-parser` (SSE framing, itself dependency-free); adding a second needs the same scrutiny the first got. The canonical client (see §SDK Sync). 15. **Observability invariants** — stdout always 100% (sampling is OTLP-push-only); WARN+ERROR always export at 100% (a non-configurable floor — don't expose it); gRPC OTel exporters dial lazily so an unreachable collector never blocks startup; the OTel Prometheus exporter uses a **private** `prometheus.Registry`. The OTLP endpoint/TLS/custom-CA/mTLS/headers are delegated to the OpenTelemetry SDK's standard `OTEL_EXPORTER_OTLP_*` env vars — `InitProvider` passes **no** endpoint/header options. Known gap, intentionally not patched in WaveHouse app code: the pinned gRPC logs exporter (`otlploggrpc` v0.19/v0.20) ignores the env TLS-cert vars, so a custom/private CA and mutual TLS apply to traces/metrics but **not** the logs signal (public-CA/system-roots TLS and plaintext still work for logs) — upstream bug open-telemetry/opentelemetry-go#6661. A malformed `OTEL_EXPORTER_OTLP_HEADERS` is logged and skipped by the SDK (fail-soft), not fatal. Preserve when touching the logger/sampler/provider. Detail: architecture.md § `observability/`. -16. **Bearer-token-only CORS posture (security)** — Bearer JWT on every request, no cookies/sessions; `corsMiddleware` deliberately **never** emits `Access-Control-Allow-Credentials` (not needed, and `*` + credentials is a spec violation browsers reject). `cors_allowed_origins` controls who can *read* responses, not cookie scope; CSRF protection is structural. Don't reintroduce cookie auth or `Allow-Credentials` without a design discussion — answers GitHub #29/#30. Code: `internal/api/router.go`. +16. **Bearer-token-only CORS posture (security)** — Bearer JWT on every request, no cookies/sessions; `corsMiddleware` deliberately **never** emits `Access-Control-Allow-Credentials` (not needed, and `*` + credentials is a spec violation browsers reject). `cors.allowed_origins` (settings directory) controls who can *read* responses, not cookie scope; CSRF protection is structural. Don't reintroduce cookie auth or `Allow-Credentials` without a design discussion — answers GitHub #29/#30. Code: `internal/api/router.go`. 17. **Non-fatal boot** — schema-discovery failure on boot is non-fatal: `cmd/wavehouse` records an `api.BootState`, binds `:8080`, serves 503 on `/livez`/`/readyz` with the diagnostic, and retries via `SchemaRegistry.RetryRefresh` (backoff 2s → 60s). Bounds supervisor restart loops. 18. **Health endpoints** — liveness `/livez`, readiness `/readyz` (k8s convention); `/healthz` is a permanent alias of `/livez`; `/health` + `/ready` are deprecated (removal v0.2.0, CHANGELOG #144). `/v1/health` is the SDK's content-free public ping (no ClickHouse check), a `/v1` route so it survives reverse-proxy probe-path filtering. Point k8s at `/livez`/`/readyz`, SDK/online-checks at `/v1/health`, never the deprecated aliases. 19. **Canonical timestamp wire form (fail-open at ingest)** — the HTTP ingest handler rewrites every top-level `DateTime`/`DateTime64` column value it can parse to RFC 3339 UTC (`discovery.CanonicalizeTimestamps`; per-column precision + zone precomputed at schema refresh) after validation + policy checks and **before** the NATS publish, so the one payload every consumer shares — SSE subscribers, the ClickHouse insert, the DLQ — carries the same spelling `/v1/query` renders: live and query reads can't drift on the instant (#372). Zone-less inputs are read in the column's declared zone, else the discovered server default — ClickHouse's own rule, so the spelling changes but never the instant. Deliberately **fail-open**: an unparseable value or unresolvable zone (no tzdata embedded — never a failed refresh, never a silent UTC reinterpretation, which would move instants) publishes verbatim; ingest must not reject a record over its timestamp spelling — fail-closed enforcement belongs to the stream row-filter (#381). Don't re-spell timestamps downstream. Preserve when touching `internal/discovery`, the ingest handler, or the SSE fan-out. Detail: architecture.md § `discovery/` + §Ingest Path; the exact spelling spec (truncation, zero-trimming, `Z`-only) lives in api.md §Timestamp canonicalization — keep it in sync with `canonicalTimestamp`. diff --git a/CHANGELOG.md b/CHANGELOG.md index 950e8781..036a055b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), ### Added +- **Settings-directory hot reload — boot loading, three reload triggers, and the config-key migration** (`internal/settings/` (new: `store.go`, `watch.go`, + tests), `internal/api/settings.go` (new, + tests), `internal/api/{router,ingest,structured_query}.go`, `internal/discovery/discovery.go`, `internal/config/config.go`, `cmd/wavehouse/main.go`, `config.yaml`, `deployments/compose/standalone.yaml`; closes the loop [#500](https://github.com/Wave-RF/WaveHouse/pull/500) opened, tracked by [#48](https://github.com/Wave-RF/WaveHouse/issues/48)): the server now *consumes* the settings directory instead of only validating it. `settings.Store` owns the adopted snapshot: `settings.dir` / `WH_SETTINGS_DIR` is now **required**, boot validates and adopts the directory (missing or invalid refuses to start — the same fail-loud contract as `policy.file_path`); a running instance then re-validates and re-adopts on any of three triggers — a **directory watch** (fsnotify on the directory, not the files, so atomic-writer replaces and Kubernetes ConfigMap symlink swaps aren't lost; bursts debounce into one reload), **`SIGHUP`**, and **`POST /v1/ops/settings/reload`** (admin-gated; returns `{"adopted", "findings"}`, `200` adopted / `422` rejected) — all funneling through one serialized reload path. A reload that fails validation keeps the previous good snapshot (an operator mid-edit degrades to a log line, never a broken server); warnings don't block adoption, matching `wavehouse validate`. The tenant tunables **migrate out of boot config** into the directory's `config.json`: `dedupe.id_field` / `dedupe.require_id` (now with the per-table overrides under `dedupe.tables` that [#222](https://github.com/Wave-RF/WaveHouse/issues/222) asked for, resolved per record through the table → global cascade in one atomic snapshot read, so a reload lands at a record boundary and never mixes documents within one record), `query.default_max_rows` (read per query), `schema.refresh_interval` (re-read after each tick, so a change applies from the next cycle), and the CORS allowlist (`cors.allowed_origins`, resolved per request). The corresponding YAML/env keys are **removed**: `server.cors_allowed_origins`, `query.default_max_rows`, `schema.refresh_interval`, `dedupe.id_field`, `dedupe.require_id` (and `WH_SERVER_CORS_ALLOWED_ORIGINS`, `WH_QUERY_DEFAULT_MAX_ROWS`, `WH_SCHEMA_REFRESH_INTERVAL`, `WH_DEDUPE_ID_FIELD`, `WH_DEDUPE_REQUIRE_ID`); the binary carries **no compiled defaults** — every `config.json` key is required (validation names each missing one), so the adopted snapshot is what the files say, and once adopted it outlives its files (a deleted file or vanished directory is just a rejected reload). Defaults live in one checked-in seed directory (`internal/settings/seed/`, `go:embed`ded) shipped three ways: the new **`wavehouse init-settings `** writes it (refusing a non-empty directory, the `initdb` contract), both container images bake it at `/app/settings` with `WH_SETTINGS_DIR` preset (bare `docker run` boots; bind-mount your own directory over it), and the dev `config.yaml` and e2e fixture point at it. `dedupe.enabled` stays boot config (it owns Pebble's lifecycle — restart-only), as do the platform-infra knobs (SSE keepalives): the split is by owner — the operator's knobs restart, the tenant's reload. Consumers take functions, not values (`IngestHandler.DedupeSettings`, the structured-query handler's `defaultMaxRows func() int`, `corsMiddleware`'s origins getter, `SchemaRegistry.SetIntervalSource`), so `internal/api` stays testable without materializing settings directories. `roles.json`/`policies.json`/`pipes.json` are validated as part of the directory but not yet runtime-adopted — policy and pipes keep their NATS KV stores as the runtime authority; converging those onto the settings directory is the next slice. - **Settings-directory validation — `wavehouse validate [dir]`** (`internal/settings/` (new: `settings.go`, `validate.go`, `decode.go`, `finding.go`, + tests), `cmd/wavehouse/validate.go` (new, + tests), `cmd/wavehouse/main.go`): first piece of the file-based control plane (settings live in a directory of JSON documents — `roles.json`, `policies.json`, `pipes.json`, `config.json` — that a running instance will hot-reload; this change is validation-only — boot loading and reload wiring land separately). `settings.Validate(dir)` is the single gate every consumer of the directory runs: deliberately pure (no network, no ClickHouse — table/column existence stays with schema discovery, per Bring-Your-Own-Schema), and it collects **all** findings in one pass instead of failing on the first. Checks, layered: the directory holds exactly the four files (a missing file is an error — an empty document is `{}`, so absence always means deletion or a wrong path; any unexpected entry — file or directory — is an error so a typoed `polices.json` or a stray backup can't be silently ignored; dot-prefixed entries are the one carve-out, since erroring on vim swap files or the `..data` machinery Kubernetes ConfigMap mounts publish through would break hand editing and the cloud fan-out's mount pattern alike); strict JSON syntax (unknown fields rejected — the JSON form of the retired-config-key trap; empty/truncated files rejected, never read as an empty document; a leading UTF-8 byte order mark named as such instead of surfacing as a cryptic invalid-character error; a directory, unreadable file, or non-regular file (a FIFO would hang the read forever waiting for a writer; a stat gate rejects it — following symlinks, so Kubernetes ConfigMap mounts' symlink layout still passes) squatting on a settings filename named as the one real problem, not double-reported as "missing"; a top-level `null` rejected — the one well-formed document that decodes into a zero value without error, so it would silently read as "no settings"; trailing content rejected; duplicated object keys detected by a token-level pass, since `encoding/json` silently keeps the last copy); per-file shape rules (role names non-empty/unique, pipe names/SQL/param types, `config.json` bounds mirroring boot-config validation — its sections are the *tenant-owned* behavioral tunables (dedupe id_field/require_id plus per-table overrides under `dedupe.tables` — each entry overrides only the fields it names, resolving table → global → compiled default per field, so the effective id_field can never be empty — an explicit empty, whitespace-only, or whitespace-padded id_field is rejected at both levels, since an exact-match JSON key lookup would silently miss every row ([#222](https://github.com/Wave-RF/WaveHouse/issues/222)'s shape, unblocked by the file design since table names are runtime-resolved like policy grants); query default_max_rows, schema refresh_interval, CORS origins); platform-owned knobs like the SSE keepalives deliberately stay boot config); and cross-file referential integrity (every role a policy grant, `default_role`/`admin_role`, or pipe allowlist references must be declared in `roles.json`; an empty role string in a grant or allowlist is named as such — it matches no request and authorizes nobody). Warnings don't invalidate: a grant scoping the admin role (an unconditional bypass — dead config), `default_role` = admin, and a `default` on a required pipe parameter are flagged but legal. An empty `policies.json` means no policy — fail closed, matching deleted-policy semantics — and draws a warning naming the total lockout, so it announces itself at validation time instead of one 403 at a time. The CLI (`cmd/wavehouse/validate.go`, following the `health` subcommand pattern) takes the directory as an argument or from `WH_SETTINGS_DIR`, prints findings, and exits 0/1/2 (valid/invalid/usage) so CI and operators can gate config changes before they reach a running instance. The dispatch in `main.go` also grows `help` and `version` subcommands, and an unknown command is now a usage error instead of silently falling through and starting the server (`wavehouse validat` booting a listener is not a typo anyone wants); each subcommand parses its arguments with a stdlib `flag.FlagSet`, so `wavehouse -h` prints command-specific help and a stray flag or argument is a usage error rather than being silently swallowed. `WH_SETTINGS_DIR` has a single authority: `config.EnvSettingsDir`, with a reflection test pinning the `settings.dir` struct tag to it. The directory's location joins boot config as `settings.dir` (`WH_SETTINGS_DIR`; `internal/config/config.go`, `config.yaml`, `docs/src/content/docs/configuration.mdx`) — boot-tier by necessity, since it's the pointer the reload machinery follows; no default, same silent-misconfiguration reasoning as `policy.file_path`. - **Docs-site analytics for search, code copies, 404s, docs section, and live-demo connectivity** (`docs/src/components/DocsTracking.astro` (new), `docs/src/components/{PostHog,Footer,LiveDemo}.astro`): the site tracked its own CTAs but nothing a reader did on the way to one, so the questions that decide what to write next — what people search for and *don't* find, which snippets get copied, which dead links keep getting followed — had no data behind them. `docs_search` fires a second after the query settles rather than once per keystroke, carrying `query` and `result_count` read off Pagefind's own results message (the rendered list is capped at its page size, so counting the DOM would under-report); `result_count: 0` is the event worth having. `code_copied` (`page`, `language`) watches Expressive Code's copy buttons from the document rather than re-binding every code block on every navigation — the hero's install chip is not an EC block and keeps its own `hero_install_copied`. `docs_404` (`path`, `referrer`) turns broken inbound links into a list instead of a hunch. A `doc_section` property (the first path segment, `home` for `/`) puts every event in a docs area without each tracker carrying its own copy; it's stamped at capture time by a `before_send` hook in `posthog.init()` rather than `register()`, because a queued `register()` replays only after init has already captured the first hard-load `$pageview` — which would then carry the previous visit's persisted value — and `history_change` navigations update the URL before capture fires, so reading `location` in the hook is always current. `live_demo_connected` fires once per mount when the hero's SSE feed comes up rather than on its first row — named for what it measures (the demo backend answered), since a quiet minute on the repo is not a disengaged reader. The three site-wide trackers share one new `DocsTracking.astro` rendered from the footer (like `MermaidZoom` / `ScrollHints`) and delegate from `document`, since Pagefind, Expressive Code, and the 404 route all own their own markup — some of it created after page load. diff --git a/cmd/wavehouse/init_settings.go b/cmd/wavehouse/init_settings.go new file mode 100644 index 00000000..29d4691e --- /dev/null +++ b/cmd/wavehouse/init_settings.go @@ -0,0 +1,51 @@ +package main + +import ( + "errors" + "flag" + "fmt" + "os" + "strings" + + "github.com/Wave-RF/WaveHouse/internal/config" + "github.com/Wave-RF/WaveHouse/internal/settings" +) + +// runInitSettings implements `wavehouse init-settings `: write the +// starter settings directory — all four files, every key at its default — +// so an operator has a complete, valid directory to point WH_SETTINGS_DIR at +// and edit from. The server never does this on its own: a missing directory +// at boot is a refused start, not an invitation to invent one (the +// `initdb` contract). Refuses a non-empty directory. Exit codes: 0 written, +// 1 failed, 2 usage. +func runInitSettings(args []string) int { + fs := flag.NewFlagSet("init-settings", flag.ContinueOnError) + fs.Usage = func() { + _, _ = fmt.Fprintf(fs.Output(), `usage: wavehouse init-settings + +Write a starter settings directory (%s) with every key at its +default into dir, creating it if needed. Refuses a non-empty directory. Point +%s at the result (or pass it to 'wavehouse validate'). + +Exit codes: 0 written, 1 failed, 2 usage. +`, strings.Join(settings.Files(), ", "), config.EnvSettingsDir) + } + if err := fs.Parse(args); err != nil { + if errors.Is(err, flag.ErrHelp) { + return 0 + } + return 2 + } + if fs.NArg() != 1 { + fmt.Fprintf(os.Stderr, "wavehouse init-settings: expected exactly one directory argument, got %d\n", fs.NArg()) + fs.Usage() + return 2 + } + dir := fs.Arg(0) + if err := settings.WriteSeed(dir); err != nil { + fmt.Fprintf(os.Stderr, "wavehouse init-settings: %v\n", err) + return 1 + } + fmt.Printf("wrote %s to %s\n", strings.Join(settings.Files(), ", "), dir) + return 0 +} diff --git a/cmd/wavehouse/init_settings_test.go b/cmd/wavehouse/init_settings_test.go new file mode 100644 index 00000000..d05f4d32 --- /dev/null +++ b/cmd/wavehouse/init_settings_test.go @@ -0,0 +1,42 @@ +package main + +import ( + "os" + "path/filepath" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestRunInitSettings(t *testing.T) { + t.Run("writes a directory validate accepts", func(t *testing.T) { + dir := filepath.Join(t.TempDir(), "settings") + assert.Equal(t, 0, runInitSettings([]string{dir})) + entries, err := os.ReadDir(dir) + require.NoError(t, err) + assert.Len(t, entries, 4) + assert.Equal(t, 0, runValidate([]string{dir}), "the seed must pass its own gate") + }) + + t.Run("refuses a non-empty directory", func(t *testing.T) { + dir := t.TempDir() + require.NoError(t, os.WriteFile(filepath.Join(dir, "keep.txt"), []byte("mine"), 0o600)) + assert.Equal(t, 1, runInitSettings([]string{dir})) + data, err := os.ReadFile(filepath.Join(dir, "keep.txt")) //nolint:gosec // G304: path is rooted in t.TempDir() + require.NoError(t, err) + assert.Equal(t, "mine", string(data)) + }) + + t.Run("missing argument is a usage error", func(t *testing.T) { + assert.Equal(t, 2, runInitSettings(nil)) + }) + + t.Run("too many arguments is a usage error", func(t *testing.T) { + assert.Equal(t, 2, runInitSettings([]string{"a", "b"})) + }) + + t.Run("-h prints help and exits 0", func(t *testing.T) { + assert.Equal(t, 0, runInitSettings([]string{"-h"})) + }) +} diff --git a/cmd/wavehouse/main.go b/cmd/wavehouse/main.go index ae055850..1f2d9614 100644 --- a/cmd/wavehouse/main.go +++ b/cmd/wavehouse/main.go @@ -28,6 +28,7 @@ import ( "github.com/Wave-RF/WaveHouse/internal/observability" "github.com/Wave-RF/WaveHouse/internal/pipes" "github.com/Wave-RF/WaveHouse/internal/policy" + "github.com/Wave-RF/WaveHouse/internal/settings" "github.com/Wave-RF/WaveHouse/internal/stream" ) @@ -98,7 +99,7 @@ func buildInfoFallback() { func main() { // Subcommand dispatch. `health` self-probes /livez for the distroless // Dockerfile HEALTHCHECK; `validate` checks a settings directory without - // starting the server. An unknown command is a usage error — it must + // starting the server; `init-settings` writes a starter one. An unknown command is a usage error — it must // never fall through and silently start the server (`wavehouse validat` // booting a listener is not a typo anyone wants). The switch only routes; // each subcommand owns a stdlib flag.FlagSet, so `wavehouse -h` @@ -111,6 +112,8 @@ func main() { os.Exit(runHealthCheck(os.Args[2:])) case "validate": os.Exit(runValidate(os.Args[2:])) + case "init-settings": + os.Exit(runInitSettings(os.Args[2:])) case "version", "--version", "-v": fmt.Printf("wavehouse %s (commit %s, built %s)\n", Version, GitCommit, BuildTime) os.Exit(0) @@ -130,6 +133,8 @@ func printUsage(w io.Writer) { _, _ = fmt.Fprintf(w, `usage: wavehouse start the server wavehouse validate [dir] validate a settings directory (dir falls back to %s) + wavehouse init-settings + write a starter settings directory (every key at its default) wavehouse health liveness self-probe against the local server (container HEALTHCHECK) wavehouse version print version, commit, and build time wavehouse help show this help @@ -169,6 +174,19 @@ func run() int { logger.Warn("WH_AUTH_JWT_SECRET is using the default insecure value") } + // Settings directory — the hot-reloadable half of configuration (tenant + // tunables: dedupe id_field/require_id, query default_max_rows, schema + // refresh_interval, CORS origins). Required: config.Validate already + // rejected an empty settings.dir, and an invalid directory refuses boot — + // the same fail-loud contract as policy.file_path. The binary carries no + // compiled defaults; `wavehouse init-settings` writes the seed. A *reload* + // of an invalid directory merely keeps the previous snapshot. + settingsStore, _ := settings.Open(cfg.Settings.Dir, logger) + if settingsStore == nil { + logger.Error("settings directory invalid, refusing to start — findings above; `wavehouse validate` reproduces them, `wavehouse init-settings` writes a starter directory", "dir", cfg.Settings.Dir) + return 1 + } + cfg.Auth.OperatorKey = strings.TrimSpace(cfg.Auth.OperatorKey) if cfg.Auth.OperatorKey == "" { logger.Warn("no auth.operator_key set: if you lose the JWT secret, lose control of the JWKS endpoint, or lose your HMAC secret — or the policy is wiped — you will be locked out remotely and will need SSH access to restore the policy file and reboot") @@ -274,6 +292,29 @@ func run() int { ctx, cancel := context.WithCancel(context.Background()) defer cancel() + // Settings reload triggers. All three (SIGHUP here, the directory watcher + // below, POST /v1/ops/settings/reload) funnel into the same serialized + // Store.Reload, and a rejected reload keeps the previous good snapshot. + hup := make(chan os.Signal, 1) + signal.Notify(hup, syscall.SIGHUP) + go func() { + for { + select { + case <-ctx.Done(): + return + case <-hup: + settingsStore.TriggerReload("sighup") + } + } + }() + go func() { + // Watcher setup failure degrades, not fatal: SIGHUP and the ops + // endpoint still reload. + if err := settingsStore.Watch(ctx); err != nil { + logger.Error("settings directory watcher failed; reload via SIGHUP or POST /v1/ops/settings/reload", "error", err) + } + }() + // Schema discovery — non-fatal on boot. If the first Refresh fails // (ClickHouse unreachable, database missing, etc.) we mark the binary // degraded via bootState (which /livez surfaces as 503 + diagnostic) @@ -284,8 +325,10 @@ func run() int { // only after the first successful Refresh (sync or retry) so it never // races RetryRefresh on Refresh calls or on bootState writes. bootState := api.NewBootState(nil) - refreshInterval := time.Duration(cfg.Schema.RefreshInterval) * time.Second - registry := discovery.NewSchemaRegistry(chConn, cfg.ClickHouse.Database, refreshInterval, logger) + // The interval source keeps the cadence live across settings reloads; the + // constructor value is the boot-time resolution of the same setting. + registry := discovery.NewSchemaRegistry(chConn, cfg.ClickHouse.Database, settingsStore.SchemaRefreshInterval(), logger) + registry.SetIntervalSource(settingsStore.SchemaRefreshInterval) if err := registry.Refresh(ctx); err != nil { logger.Warn("schema discovery failed on boot, retrying in background", "error", err) bootState.Set(fmt.Errorf("schema discovery: %w", err)) @@ -427,8 +470,7 @@ func run() int { ingestHandler.PolicyStore = policyStore if dedup != nil { ingestHandler.Dedup = dedup - ingestHandler.IDField = cfg.Dedupe.IDField - ingestHandler.RequireID = cfg.Dedupe.RequireID + ingestHandler.DedupeSettings = settingsStore.DedupeFor } var dlqHandler *api.DLQHandler @@ -486,12 +528,13 @@ func run() int { DLQ: dlqHandler, Policy: api.NewPolicyHandler(policyStore), Pipes: api.NewPipesHandler(pipesStore, policyStore, chConn, cache, cfg.ClickHouse.QueryTimeout, logger), - StructuredQuery: api.NewStructuredQueryHandler(chConn, cache, registry, policyStore, cfg.Cache.TimestampBucketSeconds, cfg.ClickHouse.QueryTimeout, cfg.Query.DefaultMaxRows, logger), + StructuredQuery: api.NewStructuredQueryHandler(chConn, cache, registry, policyStore, cfg.Cache.TimestampBucketSeconds, cfg.ClickHouse.QueryTimeout, settingsStore.DefaultMaxRows, logger), AuthMW: authMW, PolicyStore: policyStore, Logger: logger, JS: js, - CORSOrigins: cfg.Server.CORSAllowedOrigins, + CORSOrigins: settingsStore.CORSOrigins, + Settings: api.NewSettingsHandler(settingsStore, logger), } // Prometheus /metrics routing: same-port → mount on API router, diff --git a/cmd/wavehouse/validate_test.go b/cmd/wavehouse/validate_test.go index 77682062..6b612b58 100644 --- a/cmd/wavehouse/validate_test.go +++ b/cmd/wavehouse/validate_test.go @@ -19,7 +19,7 @@ func writeSettingsDir(t *testing.T, policies string) string { "roles.json": `{"roles": ["public"]}`, "policies.json": policies, "pipes.json": `{}`, - "config.json": `{}`, + "config.json": `{"dedupe": {"id_field": "event_id", "require_id": false}, "query": {"default_max_rows": 10000}, "schema": {"refresh_interval": 60}, "cors": {"allowed_origins": ["*"]}}`, } for name, content := range files { require.NoError(t, os.WriteFile(filepath.Join(dir, name), []byte(content), 0o600)) diff --git a/config.yaml b/config.yaml index cfb3a686..87b7ba75 100644 --- a/config.yaml +++ b/config.yaml @@ -6,16 +6,6 @@ data_dir: ./data server: port: 8080 shutdown_timeout: 10 - # CORS allowlist. "*" allows any browser origin; this is fine in dev because - # WaveHouse is a Bearer-token API (no cookies, no credentials mode — see - # internal/api/router.go corsMiddleware), so cross-origin XHR can't smuggle - # auth. In production with a known frontend domain, replace "*" with the - # explicit list, e.g.: - # cors_allowed_origins: - # - "https://dashboard.wavehouse.example" - # - "http://localhost:3000" # local frontend dev server - cors_allowed_origins: - - "*" otel: enabled: false # master switch — set true to export via OTLP gRPC @@ -47,13 +37,6 @@ clickhouse: password: "" query_timeout: 30s -# Query-shaping defaults. Server-wide RESOURCE limits (memory, rows scanned, -# execution time) live in ClickHouse itself — its settings profiles and quotas -# — so they apply to every query uniformly; see docs/configuration. This block -# holds only the result-LIMIT default. -query: - default_max_rows: 10000 # fallback result LIMIT (<=0 falls back to 10000) - # Server-Sent Events keepalive for GET /v1/stream. A shared ticker pushes a ":" # comment to idle streams so they survive reverse-proxy/tunnel idle timeouts. # keepalive_interval is the effective per-connection period; keep it under your @@ -67,12 +50,11 @@ mq: gap_window_minutes: 15 max_bytes_gb: 50 +# Only the enable switch lives here — it owns Pebble's lifecycle, which only a +# restart can change. id_field / require_id / per-table overrides are tenant +# tunables in the settings directory's config.json. dedupe: enabled: false - id_field: event_id - # Reject rows missing id_field instead of publishing them un-deduped. Off by - # default (they are logged + counted via wavehouse_ingest_dedupe_missing_id_total). - require_id: false cache: l1_max_cost: 67108864 @@ -89,9 +71,6 @@ auth: role_claim: role operator_key: "" # non-JWT full-access operator credential (Authorization: Operator , or X-Operator-Key) for bootstrap/break-glass; empty disables -schema: - refresh_interval: 60 - dlq: enabled: true @@ -103,9 +82,18 @@ dlq: pipes: dir: "" -# Hot-reloadable settings directory: roles.json, policies.json, pipes.json, -# config.json. Check it with `wavehouse validate` (which also honors -# WH_SETTINGS_DIR). Server-side loading/reload is landing incrementally; empty -# leaves it off. +# Settings directory (REQUIRED): roles.json, policies.json, pipes.json, +# config.json. The tenant tunables live in its config.json — dedupe +# id_field/require_id (+ per-table overrides), query.default_max_rows, +# schema.refresh_interval, cors.allowed_origins — and every key is required: +# the binary has no compiled defaults, so what's adopted is exactly what the +# files say. The server validates the directory at boot (invalid or missing +# refuses to start) and reloads it on SIGHUP, on file change, or via +# POST /v1/ops/settings/reload; a reload that fails validation keeps the +# previous settings. `wavehouse init-settings ` writes a starter +# directory with every key at its default; `wavehouse validate [dir]` checks +# one (both honor WH_SETTINGS_DIR). This points at the checked-in seed for +# local dev — to experiment, `wavehouse init-settings ./settings` and point +# here instead. settings: - dir: "" + dir: internal/settings/seed diff --git a/deployments/Dockerfile b/deployments/Dockerfile index 14867d88..261ad3da 100644 --- a/deployments/Dockerfile +++ b/deployments/Dockerfile @@ -30,13 +30,20 @@ RUN --mount=type=cache,target=/go/pkg/mod \ # them here would buy nothing and obscures intent. Named-volume copy-up runs # against `/app/data` regardless; bind mounts mask the image dir entirely # and rely on host-side ownership matching UID 65532. -RUN mkdir -p /app/data /app/pipes && chown -R 65532:65532 /app +# +# /app/settings is the baked seed settings directory (every key at its +# default, the same files `wavehouse init-settings` writes) so a bare +# `docker run` boots; production bind-mounts its own directory over it. +# Deliberately NOT a VOLUME (unlike /app/data): it's config, not state, and an +# anonymous volume would hide a forgotten mount. +RUN mkdir -p /app/data /app/pipes && cp -r /src/internal/settings/seed /app/settings && chown -R 65532:65532 /app # Stage 2: Final minimal image FROM gcr.io/distroless/static-debian12 WORKDIR /app USER nonroot:nonroot +ENV WH_SETTINGS_DIR=/app/settings COPY --from=builder --chown=nonroot:nonroot /app /app COPY --from=builder --chown=nonroot:nonroot /bin/wavehouse /app/wavehouse diff --git a/deployments/Dockerfile.goreleaser b/deployments/Dockerfile.goreleaser index 687ee03e..4c7cfe0a 100644 --- a/deployments/Dockerfile.goreleaser +++ b/deployments/Dockerfile.goreleaser @@ -17,6 +17,10 @@ # once natively and COPY'ing into each target image is correct AND # removes the binfmt/QEMU dependency entirely. FROM --platform=$BUILDPLATFORM alpine:3 AS layout +# The seed settings directory arrives via `extra_files` in .goreleaser.yaml +# (the goreleaser context holds only the dist tree). Baked at /app/settings so +# a bare `docker run` boots; not a VOLUME — it's config, not state. +COPY internal/settings/seed /app/settings RUN mkdir -p /app/data /app/pipes && chown -R 65532:65532 /app FROM gcr.io/distroless/static-debian12 @@ -26,6 +30,7 @@ ARG TARGETPLATFORM WORKDIR /app USER nonroot:nonroot +ENV WH_SETTINGS_DIR=/app/settings # Pre-created state directories owned by the nonroot user (UID 65532) so # the binary can mkdir under /app/data without a volume mount, and so diff --git a/deployments/compose/standalone.yaml b/deployments/compose/standalone.yaml index 246068fd..d2d03cdc 100644 --- a/deployments/compose/standalone.yaml +++ b/deployments/compose/standalone.yaml @@ -39,12 +39,15 @@ services: # quickstart works tokenless. Tune it before production — see that file. WH_POLICY_FILE_PATH: /app/policy.yaml WH_DEDUPE_ENABLED: "false" - # With dedupe on, reject rows missing the id field instead of publishing - # them un-deduped (default false → logged + counted, not rejected). - WH_DEDUPE_REQUIRE_ID: "false" - WH_SCHEMA_REFRESH_INTERVAL: "60" WH_DLQ_ENABLED: "true" - WH_SERVER_CORS_ALLOWED_ORIGINS: "*" + # Settings directory (roles.json, policies.json, pipes.json, config.json) + # — required; the image sets WH_SETTINGS_DIR=/app/settings and bakes the + # seed there (every key at its default: event_id as the dedupe id + # field — applies once WH_DEDUPE_ENABLED is true — 10000-row query + # limit, 60s schema refresh, CORS "*"). Tenant tunables live in its + # config.json and reload live (file watch / SIGHUP / + # POST /v1/ops/settings/reload). To edit them, write your own with + # `wavehouse init-settings ./my-settings` and bind-mount it below. # Optional: bootstrap pipes from .sql files. Mount your pipes dir # read-only and uncomment WH_PIPES_DIR. The directory is a seed for # NATS KV — runtime pipe edits go through the API, not the files. @@ -52,6 +55,7 @@ services: volumes: - wavehouse-data:/app/data - ./dev-policy.yaml:/app/policy.yaml:ro # permissive trial policy (public role, demo tables) — tune for prod + # - ./my-settings:/app/settings # your own settings dir (from `wavehouse init-settings`) # - ./my-pipes:/app/pipes:ro # uncomment alongside WH_PIPES_DIR depends_on: - clickhouse diff --git a/docs/src/content/docs/api.md b/docs/src/content/docs/api.md index c778a82f..c95265a8 100644 --- a/docs/src/content/docs/api.md +++ b/docs/src/content/docs/api.md @@ -594,7 +594,7 @@ Row values of top-level `DateTime`/`DateTime64` columns inside `data` arrive in **Note:** When access control policies are active, streamed events are filtered per the caller's role: tables without `select` permission are skipped, denied columns are removed from each event, and the role's [row-level `filter`](/access-control#row-level-security) is evaluated per subscriber against the caller's JWT claims — supplied by the connection's token (the `Authorization` header, or the `?token=` fallback above), with replayed gap-fill events filtered the same way. For a filter constant the query path's SQL also accepts ([the enforcement caution](/access-control#where-each-rule-is-enforced) gives per-type guidance), a connection is never delivered a row the query path would hide for that role — every comparison the stream can't prove fails closed and withholds the row instead. Numeric comparisons run in the column's storage domain — both operands narrowed the way ClickHouse narrows the stored value and the bound constant — so columns that narrow on insert (`Float32`/`Float64` width, a `Decimal`'s scale) agree with the query path too; the residual payload-vs-stored case is an event whose insert later fails into the DLQ, which the caution documents. The connection's claims are captured once, when the stream is established — a policy change applies from the next live event (an in-flight gap-fill finishes under the policy snapshot taken when the stream opened), but an expired token or changed claims take effect only when the client reconnects. -**CORS:** `/v1/stream` honors the `server.cors_allowed_origins` allowlist like every endpoint. Note that a **header-authenticated stream preflights before it connects** — `Authorization` is not CORS-safelisted — where a bare `EventSource` never preflighted at all: its request is not a `fetch()`, so Fetch's unsafe-request flag is never set and `Last-Event-ID` rides on the plain `GET`. Both headers are allow-listed, so an allowed origin connects *and* resumes cross-origin. +**CORS:** `/v1/stream` honors the `cors.allowed_origins` allowlist (settings directory) like every endpoint. Note that a **header-authenticated stream preflights before it connects** — `Authorization` is not CORS-safelisted — where a bare `EventSource` never preflighted at all: its request is not a `fetch()`, so Fetch's unsafe-request flag is never set and `Last-Event-ID` rides on the plain `GET`. Both headers are allow-listed, so an allowed origin connects *and* resumes cross-origin. :::caution[Behind a proxy: disable response buffering] SSE needs one bit of proxy configuration: disable response buffering, or the proxy holds events until a buffer fills and clients receive nothing in real time. Idle timeouts are handled for you — the `:` keepalive comment above keeps a quiet stream alive under typical proxy/tunnel idle windows ([#226](https://github.com/Wave-RF/WaveHouse/issues/226)), so raising the idle/read timeout is now optional. The TypeScript SDK's stream transport and browser `EventSource` both auto-reconnect (resuming via `Last-Event-ID`) if a connection drops. See [Behind a reverse proxy → Server-Sent Events](/reverse-proxy#server-sent-events-sse) for nginx/Caddy/Cloudflare specifics. @@ -796,6 +796,21 @@ Returns a specific named pipe definition. #### `DELETE /v1/ops/pipes/{name}` — Delete Named Pipe +#### `POST /v1/ops/settings/reload` — Reload Settings Directory + +Re-validates the [settings directory](/configuration#settings-directory) and adopts it when no finding is an error — the same serialized reload path the file watcher and `SIGHUP` use. + +```json +{ + "adopted": true, + "findings": [ + { "severity": "warning", "file": "policies.json", "message": "empty document — no policy; every request will be denied (fail closed)" } + ] +} +``` + +`200` when adopted (warnings allowed); `422` when validation rejected the directory — the previous settings stay in effect, and `findings` says why. + ## Event Message Format ### Internal Wire Format (NATS) diff --git a/docs/src/content/docs/configuration.mdx b/docs/src/content/docs/configuration.mdx index 11646fe3..39698a3c 100644 --- a/docs/src/content/docs/configuration.mdx +++ b/docs/src/content/docs/configuration.mdx @@ -23,6 +23,8 @@ Set `WH_CONFIG` to change the config file path: export WH_CONFIG=/etc/wavehouse/config.yaml ``` +Boot config (YAML + env) holds what the platform operator owns: wiring, lifecycle, secrets. The tenant-owned behavioral tunables — dedupe `id_field`/`require_id`, `query.default_max_rows`, `schema.refresh_interval`, CORS origins — live in the hot-reloadable [settings directory](#settings-directory) instead. + ## Full Reference ### State @@ -37,7 +39,8 @@ export WH_CONFIG=/etc/wavehouse/config.yaml | --- | --- | ------- | ----------- | | `server.port` | `WH_SERVER_PORT` | `8080` | HTTP server listen port. | | `server.shutdown_timeout` | `WH_SERVER_SHUTDOWN_TIMEOUT` | `10` | Graceful shutdown timeout in seconds. | -| `server.cors_allowed_origins` | `WH_SERVER_CORS_ALLOWED_ORIGINS` | `*` | Comma-separated list of allowed CORS origins. `*` allows any browser origin. WaveHouse is a Bearer-token API — `Access-Control-Allow-Credentials` is intentionally never sent, so this allowlist controls *which origins can read responses*, not cookie scope. Tighten to your frontend's exact origin(s) in production (e.g. `https://dashboard.example.com,http://localhost:3000`). | + +The CORS allowlist is a [settings-directory](#settings-directory) tunable (`cors.allowed_origins` in `config.json`), so it reloads without a restart. The server speaks **plain HTTP** — there is no inbound-TLS setting. Terminate TLS at a [reverse proxy](/reverse-proxy#tls) for internet-facing deployments. (The `clickhouse.http_scheme` option below is the *outbound* WaveHouse → ClickHouse hop, unrelated to your clients' TLS.) @@ -64,9 +67,7 @@ The server speaks **plain HTTP** — there is no inbound-TLS setting. Terminate ### Query -| YAML Key | Env Var | Default | Description | -| --- | --- | ------- | ----------- | -| `query.default_max_rows` | `WH_QUERY_DEFAULT_MAX_ROWS` | `10000` | Fallback result `LIMIT` for a structured query when the caller and policy specify none. `0` falls back to the built-in default; a negative value is rejected at startup. | +The fallback result `LIMIT`, `query.default_max_rows`, is a [settings-directory](#settings-directory) tunable in `config.json` (seed default `10000`): the `LIMIT` applied to a structured query when the caller and policy specify none. This is a result-**shaping** default, not a resource limit. Server-wide resource limits (memory, rows scanned, execution time) belong in ClickHouse — see [Server-side resource limits](#server-side-resource-limits) below. @@ -112,9 +113,7 @@ WaveHouse's per-role caps are sent as per-query `SETTINGS` on its connection, so ### Schema Discovery -| YAML Key | Env Var | Default | Description | -| --- | --- | ------- | ----------- | -| `schema.refresh_interval` | `WH_SCHEMA_REFRESH_INTERVAL` | `60` | How often (in seconds) to re-discover ClickHouse table schemas. Also refreshable on-demand via `POST /v1/ops/schema/refresh` (admin-only). | +`schema.refresh_interval` — how often (in seconds) to re-discover ClickHouse table schemas — is a [settings-directory](#settings-directory) tunable in `config.json` (seed default `60`); a reloaded value takes effect from the next refresh cycle. Schemas are also refreshable on-demand via `POST /v1/ops/schema/refresh` (admin-only). ### Message Queue (NATS) @@ -129,9 +128,13 @@ WaveHouse's per-role caps are sent as per-query `SETTINGS` on its connection, so | YAML Key | Env Var | Default | Description | | --- | --- | ------- | ----------- | -| `dedupe.enabled` | `WH_DEDUPE_ENABLED` | `false` | Enable event deduplication. When enabled, the ingest handler checks for duplicates using the configured ID field. | -| `dedupe.id_field` | `WH_DEDUPE_ID_FIELD` | `event_id` | JSON field name in the ingest body used as the dedup key. | -| `dedupe.require_id` | `WH_DEDUPE_REQUIRE_ID` | `false` | With dedupe enabled, controls what happens to a row missing `id_field` (which can't be deduped, so idempotency wouldn't apply to it). Such a row is always logged at `WARN` and counted by `wavehouse_ingest_dedupe_missing_id_total`, in both modes. Default (`false`): it is then published un-deduped. Set `true` to reject it instead (`400` for a single insert; a per-record failure in a batch) — a server-side tripwire for producers that must guarantee the id. | +| `dedupe.enabled` | `WH_DEDUPE_ENABLED` | `false` | Enable event deduplication. Boot config because it owns the embedded Pebble store's lifecycle, which only a restart can change. | + +The behavioral knobs are [settings-directory](#settings-directory) tunables in `config.json`, resolved per record (table override → global value): + +- `dedupe.id_field` (seed default `event_id`) — JSON field name in the ingest body used as the dedup key. +- `dedupe.require_id` (seed default `false`) — controls what happens to a row missing `id_field` (which can't be deduped, so idempotency wouldn't apply to it). Such a row is always logged at `WARN` and counted by `wavehouse_ingest_dedupe_missing_id_total`, in both modes. `false`: it is then published un-deduped. `true` rejects it instead (`400` for a single insert; a per-record failure in a batch) — a server-side tripwire for producers that must guarantee the id. +- `dedupe.tables.
.{id_field, require_id}` — per-table overrides; each entry overrides only the fields it names and inherits the rest. ### Cache @@ -182,11 +185,55 @@ See [API — Authentication](/api#authentication). The settings directory holds WaveHouse's file-based settings as exactly four JSON documents: `roles.json`, `policies.json`, `pipes.json`, and `config.json`. Every file must exist (an empty document is `{}` — a missing file always means deletion or a wrong path, never "defaults"), and any other entry in the directory is an error, so a typoed filename or a stray backup fails loudly instead of being silently ignored. Dot-prefixed entries are the one exception: editor swap files and the `..data` machinery Kubernetes ConfigMap mounts publish through are ignored. -Check a directory with `wavehouse validate [dir]` — the argument falls back to `WH_SETTINGS_DIR`. It validates without starting the server (JSON syntax including unknown fields and duplicate keys, per-file shape rules, and cross-file role references), prints every finding in one pass, and exits `0` for valid (warnings allowed), `1` for invalid, `2` for usage — so operators and CI can gate a settings change before it reaches a running instance. +Create one with `wavehouse init-settings `: it writes all four files with every key at its default and refuses a non-empty directory, so an existing settings directory is never overwritten. The binary carries no compiled defaults — the seed is the one place they live, and what the server adopts is exactly what the files say. The container images bake the same seed at `/app/settings` (and set `WH_SETTINGS_DIR` to it), so a bare `docker run` boots; bind-mount your own directory over that path to edit. + +Check a directory with `wavehouse validate [dir]` — the argument falls back to `WH_SETTINGS_DIR`. It validates without starting the server (JSON syntax including unknown fields and duplicate keys, per-file shape rules including the required keys, and cross-file role references), prints every finding in one pass, and exits `0` for valid (warnings allowed), `1` for invalid, `2` for usage — so operators and CI can gate a settings change before it reaches a running instance. | YAML Key | Env Var | Default | Description | | --- | --- | ------- | ----------- | -| `settings.dir` | `WH_SETTINGS_DIR` | *(empty)* | Path to the settings directory. Its one consumer is the `wavehouse validate` CLI, which — like `wavehouse health` — resolves its input directly from the `[dir]` argument or `WH_SETTINGS_DIR` without loading this file, so a value set only in the YAML key is not picked up. No default, same reasoning as `policy.file_path`: a baked-in path would turn a missing mount into silent misconfiguration instead of an explicit operator choice. | +| `settings.dir` | `WH_SETTINGS_DIR` | *(required)* | Path to the settings directory. The server loads it at boot; the `wavehouse validate` CLI — like `wavehouse health` — resolves its input directly from the `[dir]` argument or `WH_SETTINGS_DIR` without loading this file, so a value set only in the YAML key validates on boot but not via the bare CLI. No default, same reasoning as `policy.file_path`: a baked-in path would turn a missing mount into silent misconfiguration instead of an explicit operator choice. The container images set it to the baked `/app/settings`. | + +#### Loading and hot reload + +The server validates and adopts the directory at boot — a missing or invalid directory refuses to start, the same fail-loud contract as `policy.file_path` (run `wavehouse validate` to reproduce the findings, `wavehouse init-settings` to create a starter directory). + +A running instance re-validates and re-adopts on any of three triggers, all funneling through one serialized reload path: + +- **File change** — the directory is watched; a burst of writes (an editor save, a Kubernetes ConfigMap republish) coalesces into one reload, and a directory that is deleted and recreated (or renamed over) is picked up again. +- **`SIGHUP`** — `kill -HUP `. +- **`POST /v1/ops/settings/reload`** (admin-only) — returns `{"adopted": bool, "findings": [...]}`; `200` when adopted, `422` when rejected. + +A reload that fails validation is logged (and reported by the endpoint) and the previous good settings stay in effect — an operator mid-edit, a deleted file, even a vanished directory degrades to a log line, never to a broken server. Warnings don't block adoption, matching `wavehouse validate`. + +#### `config.json` keys + +The tenant tunables. Every key is required (a missing one is a validation error) except the per-table overrides; the "Seed" column is what `wavehouse init-settings` writes: + +| Key | Seed | Description | +| --- | ---- | ----------- | +| `dedupe.id_field` | `event_id` | Dedup key field — see [Deduplication](#deduplication). | +| `dedupe.require_id` | `false` | Reject rows missing the id field — see [Deduplication](#deduplication). | +| `dedupe.tables.
.{id_field, require_id}` | `{}` | Optional per-table overrides; each entry overrides only the fields it names and inherits the rest. | +| `query.default_max_rows` | `10000` | Fallback result `LIMIT` (`>= 1`) — see [Query](#query). | +| `schema.refresh_interval` | `60` | Seconds between schema re-discoveries (`>= 1`) — see [Schema Discovery](#schema-discovery). | +| `cors.allowed_origins` | `["*"]` | Allowed CORS origins, applied per request. `"*"` allows any browser origin. WaveHouse is a Bearer-token API — `Access-Control-Allow-Credentials` is intentionally never sent, so this allowlist controls *which origins can read responses*, not cookie scope. Tighten to your frontend's exact origin(s) in production (e.g. `["https://dashboard.example.com", "http://localhost:3000"]`). | + +```json +{ + "dedupe": { + "id_field": "event_id", + "require_id": false, + "tables": { + "clicks": { "id_field": "click_id" } + } + }, + "query": { "default_max_rows": 10000 }, + "schema": { "refresh_interval": 60 }, + "cors": { "allowed_origins": ["*"] } +} +``` + +Platform-infra knobs deliberately stay boot config even though they look behavioral — the SSE keepalives exist for the deployment's proxies, `dedupe.enabled` owns a store lifecycle — so the split is by *owner*: the operator's knobs restart, the tenant's reload. ### OTel @@ -234,8 +281,6 @@ data_dir: ./data # nats → ./data/nats, pebble → ./data/pebble server: port: 8080 shutdown_timeout: 10 - cors_allowed_origins: - - "*" stream: keepalive_interval: 30s # max idle time before a keepalive; keep under your proxy's idle window @@ -250,17 +295,13 @@ clickhouse: password: "" query_timeout: 30s -query: - default_max_rows: 10000 - mq: gap_window_minutes: 15 max_bytes_gb: 50 dedupe: - enabled: false - id_field: event_id - require_id: false + enabled: false # id_field / require_id / per-table overrides live in the + # settings directory's config.json cache: l1_max_cost: 67108864 @@ -272,9 +313,6 @@ auth: role_claim: role operator_key: "" # non-JWT full-access operator credential (Authorization: Operator , or X-Operator-Key); empty disables -schema: - refresh_interval: 60 - dlq: enabled: true @@ -287,9 +325,10 @@ pipes: dir: "" # empty = skip bootstrap; set + read-only mount to seed pipes settings: - dir: "" # settings directory (roles/policies/pipes/config .json); - # `wavehouse validate` reads its [dir] argument or - # WH_SETTINGS_DIR (the env var), not this YAML key + dir: "" # REQUIRED: settings directory (roles/policies/pipes/ + # config .json) — validated at boot, reloaded on file + # change, SIGHUP, or POST /v1/ops/settings/reload; + # create one with `wavehouse init-settings ` otel: enabled: false # master switch — set true to export via OTLP gRPC @@ -319,7 +358,6 @@ WH_DATA_DIR=./data WH_SERVER_PORT=8080 WH_SERVER_SHUTDOWN_TIMEOUT=10 -WH_SERVER_CORS_ALLOWED_ORIGINS=* WH_STREAM_KEEPALIVE_INTERVAL=30s WH_STREAM_KEEPALIVE_BUCKETS=3 @@ -332,14 +370,10 @@ WH_CH_USERNAME=default WH_CH_PASSWORD= WH_CH_QUERY_TIMEOUT=30s -WH_QUERY_DEFAULT_MAX_ROWS=10000 - WH_MQ_GAP_WINDOW_MINUTES=15 WH_MQ_MAX_BYTES_GB=50 WH_DEDUPE_ENABLED=false -WH_DEDUPE_ID_FIELD=event_id -WH_DEDUPE_REQUIRE_ID=false WH_CACHE_L1_MAX_COST=67108864 WH_CACHE_TIMESTAMP_BUCKET_SECONDS=60 @@ -349,15 +383,13 @@ WH_AUTH_JWKS_URL= WH_AUTH_ROLE_CLAIM=role WH_AUTH_OPERATOR_KEY= -WH_SCHEMA_REFRESH_INTERVAL=60 - WH_DLQ_ENABLED=true WH_POLICY_FILE_PATH= WH_PIPES_DIR= -WH_SETTINGS_DIR= +WH_SETTINGS_DIR= # required; the container images set /app/settings WH_OTEL_ENABLED=false WH_OTEL_TRACES_ENABLED=true diff --git a/docs/src/content/docs/deployment.md b/docs/src/content/docs/deployment.md index fee146d3..2f8afc73 100644 --- a/docs/src/content/docs/deployment.md +++ b/docs/src/content/docs/deployment.md @@ -59,7 +59,7 @@ Or override any config with environment variables: ```bash WH_CH_ADDR=clickhouse.example.com:9000 \ -WH_SCHEMA_REFRESH_INTERVAL=30 \ +WH_SERVER_PORT=9090 \ ./bin/wavehouse ``` @@ -135,15 +135,6 @@ WH_CH_ADDR=clickhouse:9000 WH_CH_HTTP_PORT=8123 WH_CH_HTTP_SCHEME=http # Scheme for the same (http/https) -# Schema discovery -WH_SCHEMA_REFRESH_INTERVAL=60 # Seconds between schema refreshes - -# CORS — comma-separated allowlist (or "*" for any origin). -# WaveHouse is a Bearer-token API; no cookies are used and the middleware -# deliberately omits Access-Control-Allow-Credentials, so this allowlist only -# controls *which origins can read responses*, not cookie scope. -WH_SERVER_CORS_ALLOWED_ORIGINS=https://app.example.com,https://admin.example.com - # Auth (the JWT middleware always runs — set a secret/JWKS to validate tokens; # without one, every request resolves to the policy default_role) WH_AUTH_JWT_SECRET= @@ -163,15 +154,21 @@ WH_AUTH_OPERATOR_KEY= WH_POLICY_FILE_PATH=/etc/wavehouse/policy.yaml WH_PIPES_DIR=/etc/wavehouse/pipes +# Settings directory (required): roles.json, policies.json, pipes.json, +# config.json. Tenant tunables live in its config.json — dedupe +# id_field/require_id (+ per-table overrides), query default_max_rows, schema +# refresh_interval, CORS allowed_origins — and reload on file change, SIGHUP, +# or POST /v1/ops/settings/reload. Create it with +# `wavehouse init-settings /etc/wavehouse/settings`; the container images +# bake a default at /app/settings. See Configuration — Settings directory. +WH_SETTINGS_DIR=/etc/wavehouse/settings + # Cache tuning WH_CACHE_TIMESTAMP_BUCKET_SECONDS=60 -# Optional dedup +# Optional dedup (the switch is boot config; the id field and strictness are +# settings-directory tunables) WH_DEDUPE_ENABLED=true -WH_DEDUPE_ID_FIELD=event_id -# Reject rows missing the id field instead of publishing them un-deduped -# (default false → such rows are logged + counted, not rejected). -WH_DEDUPE_REQUIRE_ID=false # Standalone tuning WH_MQ_GAP_WINDOW_MINUTES=15 # Minutes of NATS history for SSE gap-fill @@ -369,7 +366,7 @@ CREATE TABLE IF NOT EXISTS clicks ( ORDER BY (page); ``` -WaveHouse discovers this schema on startup and refreshes it every `schema.refresh_interval` seconds (default: 60). You can also trigger an immediate refresh via `POST /v1/ops/schema/refresh` (admin-only). +WaveHouse discovers this schema on startup and refreshes it every `schema.refresh_interval` seconds (settings directory; seed default 60). You can also trigger an immediate refresh via `POST /v1/ops/schema/refresh` (admin-only). ## Dead Letter Queue (DLQ) diff --git a/docs/src/content/docs/development.md b/docs/src/content/docs/development.md index e3f765db..eba84e9f 100644 --- a/docs/src/content/docs/development.md +++ b/docs/src/content/docs/development.md @@ -144,7 +144,7 @@ dev: deps-up $(AIR) **While `make dev` is running you get:** -- WaveHouse on `http://localhost:8080` with `cors_allowed_origins: ["*"]`, so a browser-based app on any localhost port can hit the API directly. +- WaveHouse on `http://localhost:8080` with the default allow-all CORS posture, so a browser-based app on any localhost port can hit the API directly. - A placeholder JWT secret (`change-me-in-production`) ships in `config.yaml`, but **no policy** is seeded — so the stack is fail-closed until you seed one (see [Test the API](#test-the-api)). Override the secret via `WH_AUTH_JWT_SECRET`. - ClickHouse on `http://localhost:8123` (HTTP) and `localhost:9000` (native protocol), Compose project name `wavehouse-dev` so containers/volumes are namespaced. - Hot reload: editing any `.go` file under `cmd/` or `internal/` triggers a debounced rebuild + restart. Config isn't hot-reloaded — `make dev` loads `.config.local.yaml` (a gitignored copy seeded once from `config.yaml`), so edit `.config.local.yaml` and restart to apply config changes. Air's stdout/stderr stream live so you see compile errors and server logs in the same terminal. @@ -228,12 +228,14 @@ curl -s -X POST http://localhost:8080/v1/ops/query \ ### Enable Dedup (Optional) -Set `WH_DEDUPE_ENABLED=true` and `WH_DEDUPE_ID_FIELD=event_id`: +Set `WH_DEDUPE_ENABLED=true`: ```bash -WH_DEDUPE_ENABLED=true WH_DEDUPE_ID_FIELD=event_id make dev +WH_DEDUPE_ENABLED=true make dev ``` +Records dedupe on their `event_id` field by default; the settings directory's `config.json` overrides the field globally or per table (see [Configuration — Deduplication](/configuration#deduplication)). + Then include the dedup field in your ingest body: ```bash diff --git a/docs/src/content/docs/getting-started.md b/docs/src/content/docs/getting-started.md index 8b368d8f..681be98a 100644 --- a/docs/src/content/docs/getting-started.md +++ b/docs/src/content/docs/getting-started.md @@ -95,7 +95,7 @@ curl -N "http://localhost:8080/v1/stream?table=clicks&since=2026-03-24T11:00:00Z The handful of things that most often trip up a first session — each is expected behavior with a quick fix: -- **`404 unknown table: clicks` on the first ingest.** Schema discovery refreshes every 60 seconds (`WH_SCHEMA_REFRESH_INTERVAL`), so a just-created table may not be visible yet. Wait and retry — worst case the next refresh is a full 60 seconds out. (`POST /v1/ops/schema/refresh` forces it, but that endpoint is admin-only — the trial `public` role can't call it.) +- **`404 unknown table: clicks` on the first ingest.** Schema discovery refreshes every 60 seconds (`schema.refresh_interval` in the [settings directory](/configuration#settings-directory)), so a just-created table may not be visible yet. Wait and retry — worst case the next refresh is a full 60 seconds out. (`POST /v1/ops/schema/refresh` forces it, but that endpoint is admin-only — the trial `public` role can't call it.) - **The query returns `[]` right after an ingest succeeded.** Ingest acknowledges as soon as the event is durable in the WAL; the batch worker flushes to ClickHouse every few seconds. If you query within that window the rows simply aren't in ClickHouse yet — re-query after ~5 seconds. (The [SSE stream](#5-subscribe-to-real-time-updates) sees events *immediately* — it's broadcast before the flush.) - **`403` on a table you created yourself.** WaveHouse is fail-closed and the trial policy grants the `public` role access to the *named demo tables only* (`clicks`, `events`). A new table needs a policy entry — see [Access Control](/access-control) for granting roles per table. - **A port is already taken.** The stack binds `8080` (WaveHouse) and `8123`/`9000` (ClickHouse). Stop whatever holds the port or edit the `ports:` mappings in `deployments/compose/standalone.yaml`. @@ -113,4 +113,4 @@ The handful of things that most often trip up a first session — each is expect ## Going further - **Validate JWTs**: set `WH_AUTH_JWT_SECRET=` (the middleware always runs; without a secret every request is the policy `default_role`) and replace the shipped trial policy (`deployments/compose/dev-policy.yaml`) with a least-privilege one — see [API Reference — Authentication](/api#authentication) and [Access Control](/access-control). -- **Enable deduplication**: set `WH_DEDUPE_ENABLED=true` and `WH_DEDUPE_ID_FIELD=event_id` — see [Configuration — Deduplication](/configuration#deduplication). +- **Enable deduplication**: set `WH_DEDUPE_ENABLED=true` — records dedupe on their `event_id` field by default; pick a different field (globally or per table) in the settings directory's `config.json` — see [Configuration — Deduplication](/configuration#deduplication). diff --git a/docs/src/content/docs/reverse-proxy.mdx b/docs/src/content/docs/reverse-proxy.mdx index 63e2ea95..cfc7344e 100644 --- a/docs/src/content/docs/reverse-proxy.mdx +++ b/docs/src/content/docs/reverse-proxy.mdx @@ -22,7 +22,7 @@ WaveHouse and your proxy are layers of one system, not substitutes. The proxy ow | Slow-loris / slow-body mitigation | ✗ | ✓ | | Request-body size limit | Fixed internal backstop (1 MiB parameter/AST bodies, 16 MiB bulk payload bodies) | Tunable outer limit | | Authentication (JWT) | ✓ (validates / resolves role) | Pass through | -| CORS | ✓ (`server.cors_allowed_origins`) | Pass through (don't double it) | +| CORS | ✓ (`cors.allowed_origins`, [settings directory](/configuration#settings-directory)) | Pass through (don't double it) | | Health probes | ✓ (serves `/livez`, `/readyz`, `/v1/health`) | Route / expose appropriately | :::note[Defense in depth, not either/or] @@ -196,7 +196,7 @@ These limit the *whole* request regardless of traffic, so no keepalive extends t WaveHouse does **not** derive a client IP from forwarded headers — it does no per-IP logic (rate limiting and IP allow/deny are the proxy's job) and does not trust `X-Forwarded-For` / `X-Real-IP` / `True-Client-IP` to rewrite the connection's source address. So a forged forwarded header has no effect on WaveHouse, and `r.RemoteAddr` (what OpenTelemetry records as the peer) is the honest immediate peer — your proxy, when one is in front. Still, don't expose `:8080` to untrusted clients: bind WaveHouse to a private interface or firewall the port so the proxy is the only path in. Capturing the real client IP in WaveHouse's own traces and logs — trusted-proxy-aware, so it can't be spoofed — is tracked in [#333](https://github.com/Wave-RF/WaveHouse/issues/333). ::: -- **CORS** — WaveHouse applies its own CORS from `server.cors_allowed_origins`. Let one layer own CORS: either pass it through the proxy untouched (recommended), or strip it from WaveHouse and do it at the proxy — not both, or browsers see duplicate `Access-Control-Allow-Origin` headers and reject the response. +- **CORS** — WaveHouse applies its own CORS from the settings directory's `cors.allowed_origins`. Let one layer own CORS: either pass it through the proxy untouched (recommended), or strip it from WaveHouse and do it at the proxy — not both, or browsers see duplicate `Access-Control-Allow-Origin` headers and reject the response. ## Fencing the admin surface diff --git a/docs/src/content/docs/sdk/index.mdx b/docs/src/content/docs/sdk/index.mdx index f5bc73bc..4ee420bc 100644 --- a/docs/src/content/docs/sdk/index.mdx +++ b/docs/src/content/docs/sdk/index.mdx @@ -401,7 +401,7 @@ Everything else reaches streams normally — including `mode` from the example a ::: :::note[`credentials: 'include'` needs a proxy that owns CORS] -Sending cookies cross-origin is a common reason to reach for this, but it can't work against WaveHouse's own CORS: it deliberately never emits `Access-Control-Allow-Credentials` (it's a Bearer-token API), and the default `cors_allowed_origins: "*"` makes `include` a hard browser failure regardless. It applies where a fronting proxy answers CORS itself — see [Behind a reverse proxy](/reverse-proxy#header-and-auth-forwarding). Same-origin deployments already send cookies without it. +Sending cookies cross-origin is a common reason to reach for this, but it can't work against WaveHouse's own CORS: it deliberately never emits `Access-Control-Allow-Credentials` (it's a Bearer-token API), and the default allow-all CORS allowlist (`cors.allowed_origins`) makes `include` a hard browser failure regardless. It applies where a fronting proxy answers CORS itself — see [Behind a reverse proxy](/reverse-proxy#header-and-auth-forwarding). Same-origin deployments already send cookies without it. ::: The fields the SDK controls — `method`, `headers`, `body`, and `signal` — always win, so this can't corrupt the request itself. In particular `headers` here is ignored; use `options.headers`, which merges properly. diff --git a/go.mod b/go.mod index bada5849..78e24210 100644 --- a/go.mod +++ b/go.mod @@ -20,6 +20,7 @@ require ( github.com/cockroachdb/pebble v1.1.5 github.com/dgraph-io/ristretto/v2 v2.4.2 github.com/dustin/go-humanize v1.0.1 + github.com/fsnotify/fsnotify v1.10.1 github.com/go-chi/chi/v5 v5.3.1 github.com/golang-jwt/jwt/v5 v5.3.1 github.com/google/uuid v1.6.0 @@ -109,7 +110,6 @@ require ( github.com/erikgeiser/coninput v0.0.0-20211004153227-1c3628e74d0f // indirect github.com/fatih/color v1.18.0 // indirect github.com/felixge/httpsnoop v1.1.0 // indirect - github.com/fsnotify/fsnotify v1.9.0 // indirect github.com/getsentry/sentry-go v0.27.0 // indirect github.com/go-delve/delve v1.26.2 // indirect github.com/go-errors/errors v1.5.1 // indirect diff --git a/go.sum b/go.sum index 9c5f51d0..5b51fac2 100644 --- a/go.sum +++ b/go.sum @@ -144,8 +144,8 @@ github.com/fatih/color v1.18.0 h1:S8gINlzdQ840/4pfAwic/ZE0djQEH3wM94VfqLTZcOM= github.com/fatih/color v1.18.0/go.mod h1:4FelSpRwEGDpQ12mAdzqdOukCy4u8WUtOY6lkT/6HfU= github.com/felixge/httpsnoop v1.1.0 h1:3YtUj32ZZkqZtt3sZZsClsymw/QDuVfpNhoA31zeORc= github.com/felixge/httpsnoop v1.1.0/go.mod h1:Zqxgdd+1Rkcz8euOqdr7lqgCRJztwr5hp9vDSi5UZCE= -github.com/fsnotify/fsnotify v1.9.0 h1:2Ml+OJNzbYCTzsxtv8vKSFD9PbJjmhYF14k/jKC7S9k= -github.com/fsnotify/fsnotify v1.9.0/go.mod h1:8jBTzvmWwFyi3Pb8djgCCO5IBqzKJ/Jwo8TRcHyHii0= +github.com/fsnotify/fsnotify v1.10.1 h1:b0/UzAf9yR5rhf3RPm9gf3ehBPpf0oZKIjtpKrx59Ho= +github.com/fsnotify/fsnotify v1.10.1/go.mod h1:TLheqan6HD6GBK6PrDWyDPBaEV8LspOxvPSjC+bVfgo= github.com/getsentry/sentry-go v0.27.0 h1:Pv98CIbtB3LkMWmXi4Joa5OOcwbmnX88sF5qbK3r3Ps= github.com/getsentry/sentry-go v0.27.0/go.mod h1:lc76E2QywIyW8WuBnwl8Lc4bkmQH4+w1gwTf25trprY= github.com/go-chi/chi/v5 v5.3.1 h1:3j4HZLGZQ3JpMCrPJF/Jl3mYJfWLKBfNJ6quurUGCf8= diff --git a/internal/api/ingest.go b/internal/api/ingest.go index 0e90a404..10c0e423 100644 --- a/internal/api/ingest.go +++ b/internal/api/ingest.go @@ -36,13 +36,16 @@ const maxReportedResults = 10000 // IngestHandler handles POST /v1/ingest?table={table} type IngestHandler struct { - Registry *discovery.SchemaRegistry - Dedup dedupe.Deduplicator // nil if dedup disabled - IDField string // dedup key field name (e.g. "event_id") - RequireID bool // reject rows missing IDField instead of publishing un-deduped (dedupe.require_id) - Publisher mq.Publisher - PolicyStore *policy.Store - logger *slog.Logger + Registry *discovery.SchemaRegistry + Dedup dedupe.Deduplicator // nil if dedup disabled + // DedupeSettings resolves the effective dedupe id_field/require_id for a + // table (settings.Store.DedupeFor in production). Called once per record so + // a settings reload lands at a record boundary — one record never mixes two + // documents' values. Dedup is skipped when nil. + DedupeSettings func(table string) (idField string, requireID bool) + Publisher mq.Publisher + PolicyStore *policy.Store + logger *slog.Logger // maxRequestBytes optionally overrides the default inbound request body cap // (maxRequestBodyBytes). When 0, the default applies. Exists so same-package @@ -422,19 +425,23 @@ func (h *IngestHandler) processRecord( // enforces) after the permission checks: check clauses keep pre-#372 semantics. discovery.CanonicalizeTimestamps(schema, data) - // Optional deduplication. - if h.Dedup != nil && h.IDField != "" { - idVal, ok := data[h.IDField] + // Optional deduplication. The id_field/require_id pair resolves per record + // (table override → global; the settings directory always states both, so + // no compiled fallback is needed). A Deduplicator without a settings + // source is a wiring bug, not a mode — main wires both or neither. + if h.Dedup != nil && h.DedupeSettings != nil { + idField, requireID := h.DedupeSettings(table) + idVal, ok := data[idField] if !ok { dedupeMissingIDCounter.Add(ctx, 1, metric.WithAttributes(attribute.String("table", table))) - if h.RequireID { - h.logger.WarnContext(ctx, "dedupe id_field missing; rejecting", "id_field", h.IDField, "table", table) + if requireID { + h.logger.WarnContext(ctx, "dedupe id_field missing; rejecting", "id_field", idField, "table", table) return false, &recordReject{ Status: http.StatusBadRequest, - Message: fmt.Sprintf("missing dedupe id field %q", h.IDField), + Message: fmt.Sprintf("missing dedupe id field %q", idField), }, nil } - h.logger.WarnContext(ctx, "dedupe id_field missing; publishing without idempotency", "id_field", h.IDField, "table", table) + h.logger.WarnContext(ctx, "dedupe id_field missing; publishing without idempotency", "id_field", idField, "table", table) } else { eventID := fmt.Sprint(idVal) dup, err := h.Dedup.CheckAndMark(ctx, eventID) diff --git a/internal/api/ingest_test.go b/internal/api/ingest_test.go index ebe75b39..1adda309 100644 --- a/internal/api/ingest_test.go +++ b/internal/api/ingest_test.go @@ -163,7 +163,7 @@ func TestIngest_Dedup_FirstTime(t *testing.T) { dedup := testutil.NewMockDeduplicator() h := NewIngestHandler(testRegistry(t), pub, testutil.NopLogger()) h.Dedup = dedup - h.IDField = "event_id" + h.DedupeSettings = func(string) (string, bool) { return "event_id", false } req := ingestRequest(t, "clicks", map[string]any{"page": "/home", "event_id": "evt-1"}) w := httptest.NewRecorder() @@ -179,7 +179,7 @@ func TestIngest_Dedup_Duplicate(t *testing.T) { dedup := testutil.NewMockDeduplicator() h := NewIngestHandler(testRegistry(t), pub, testutil.NopLogger()) h.Dedup = dedup - h.IDField = "event_id" + h.DedupeSettings = func(string) (string, bool) { return "event_id", false } // First call. req := ingestRequest(t, "clicks", map[string]any{"page": "/home", "event_id": "dup-1"}) @@ -691,9 +691,9 @@ func TestIngest_Dedup_MissingIDField(t *testing.T) { dedup := testutil.NewMockDeduplicator() h := NewIngestHandler(testRegistry(t), pub, testutil.NopLogger()) h.Dedup = dedup - h.IDField = "event_id" + h.DedupeSettings = func(string) (string, bool) { return "event_id", false } - // Payload omits event_id and require_id is off (the default): the row skips + // Payload omits event_id and require_id is off: the row skips // dedup and is still published — the warn+counter path, not a rejection (#219). req := ingestRequest(t, "clicks", map[string]any{"page": "/home"}) w := httptest.NewRecorder() @@ -710,8 +710,7 @@ func TestIngest_Dedup_RequireID_Rejects(t *testing.T) { pub := &testutil.MockPublisher{} h := NewIngestHandler(testRegistry(t), pub, testutil.NopLogger()) h.Dedup = testutil.NewMockDeduplicator() - h.IDField = "event_id" - h.RequireID = true + h.DedupeSettings = func(string) (string, bool) { return "event_id", true } w := httptest.NewRecorder() h.Handle(w, ingestRequest(t, "clicks", map[string]any{"page": "/home"})) @@ -733,8 +732,7 @@ func TestIngest_NDJSON_RequireID_Rejects(t *testing.T) { pub := &testutil.MockPublisher{} h := NewIngestHandler(testRegistry(t), pub, testutil.NopLogger()) h.Dedup = testutil.NewMockDeduplicator() - h.IDField = "event_id" - h.RequireID = true + h.DedupeSettings = func(string) (string, bool) { return "event_id", true } req := ndjsonRequest(t, "clicks", jsonLine(t, map[string]any{"page": "/a", "event_id": "e1"}), @@ -986,7 +984,7 @@ func TestIngest_NDJSON_Dedup(t *testing.T) { dedup := testutil.NewMockDeduplicator() h := NewIngestHandler(testRegistry(t), pub, testutil.NopLogger()) h.Dedup = dedup - h.IDField = "event_id" + h.DedupeSettings = func(string) (string, bool) { return "event_id", false } req := ndjsonRequest(t, "clicks", jsonLine(t, map[string]any{"page": "/a", "event_id": "e1"}), diff --git a/internal/api/router.go b/internal/api/router.go index 275b2673..73538949 100644 --- a/internal/api/router.go +++ b/internal/api/router.go @@ -28,12 +28,19 @@ type Dependencies struct { Policy *PolicyHandler Pipes *PipesHandler StructuredQuery *StructuredQueryHandler - AuthMW func(http.Handler) http.Handler + // Settings, if non-nil, mounts POST /v1/ops/settings/reload — the API + // trigger for reloading the settings directory. Nil when no settings + // directory is configured (nothing to reload). + Settings *SettingsHandler + AuthMW func(http.Handler) http.Handler // PolicyStore backs the RequireAdmin gate: the admin role (policy.AdminRole) // is read live from the policy, so admin_role changes apply without a restart. PolicyStore *policy.Store JS jetstream.JetStream // for SSE gap-fill - CORSOrigins []string // allowed CORS origins; ["*"] = allow all + // CORSOrigins returns the allowed CORS origins, read per request so a + // settings reload applies immediately (settings.Store.CORSOrigins in + // production). Nil func, an empty list, or ["*"] all mean allow-all. + CORSOrigins func() []string Logger *slog.Logger // MetricsHandler, if non-nil, is mounted at MetricsPath as an unauthenticated // endpoint (Prometheus convention). Wired by main.go from the OTel Prometheus @@ -186,6 +193,9 @@ func NewRouter(deps Dependencies) http.Handler { r.Put("/pipes/{name}", deps.Pipes.Put) r.Delete("/pipes/{name}", deps.Pipes.Delete) } + if deps.Settings != nil { + r.Post("/settings/reload", deps.Settings.Reload) + } }) }) @@ -286,17 +296,7 @@ func RequireAdmin(store *policy.Store, logger *slog.Logger) func(http.Handler) h // // Non-CORS requests (no Origin header) are passed through unchanged — we // don't decorate same-origin responses with CORS noise. -func corsMiddleware(allowedOrigins []string) func(http.Handler) http.Handler { - allowAll := len(allowedOrigins) == 0 - allowedSet := make(map[string]struct{}, len(allowedOrigins)) - for _, o := range allowedOrigins { - if o == "*" { - allowAll = true - continue - } - allowedSet[o] = struct{}{} - } - +func corsMiddleware(origins func() []string) func(http.Handler) http.Handler { return func(next http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { origin := r.Header.Get("Origin") @@ -307,6 +307,24 @@ func corsMiddleware(allowedOrigins []string) func(http.Handler) http.Handler { return } + // Resolved per request (not captured at router build) so a settings + // reload changes the allowlist without a restart. The lists are a + // handful of origins, so a linear scan beats rebuilding a set. + var allowedOrigins []string + if origins != nil { + allowedOrigins = origins() + } + allowAll := len(allowedOrigins) == 0 + originListed := false + for _, o := range allowedOrigins { + if o == "*" { + allowAll = true + } + if o == origin { + originListed = true + } + } + allowed := false switch { case allowAll: @@ -320,7 +338,7 @@ func corsMiddleware(allowedOrigins []string) func(http.Handler) http.Handler { // allowed-origin request, stripping the CORS headers and // breaking the legitimate client. w.Header().Set("Vary", "Origin") - if _, ok := allowedSet[origin]; ok { + if originListed { w.Header().Set("Access-Control-Allow-Origin", origin) allowed = true } diff --git a/internal/api/router_test.go b/internal/api/router_test.go index 4d9462a7..b7dd6b56 100644 --- a/internal/api/router_test.go +++ b/internal/api/router_test.go @@ -5,6 +5,7 @@ import ( "errors" "net/http" "net/http/httptest" + "sync" "testing" "github.com/Wave-RF/WaveHouse/internal/auth" @@ -120,9 +121,43 @@ func TestRequireAdmin_OperatorBypassesNilPolicy(t *testing.T) { assert.Equal(t, http.StatusOK, w.Code, "operator bit admits even a nil-policy request (break-glass)") } +// TestCORSMiddleware_OriginsReloadBetweenRequests pins that the allowlist is +// resolved per request, not captured at router construction: a settings +// reload that changes cors.allowed_origins must apply to the very next +// request without rebuilding the middleware. +func TestCORSMiddleware_OriginsReloadBetweenRequests(t *testing.T) { + t.Parallel() + var mu sync.Mutex + origins := []string{"https://old.example.com"} + handler := corsMiddleware(func() []string { + mu.Lock() + defer mu.Unlock() + return origins + })(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusOK) })) + + get := func(origin string) *httptest.ResponseRecorder { + req := httptest.NewRequestWithContext(context.Background(), http.MethodGet, "/", nil) + req.Header.Set("Origin", origin) + w := httptest.NewRecorder() + handler.ServeHTTP(w, req) + return w + } + + assert.Equal(t, "https://old.example.com", get("https://old.example.com").Header().Get("Access-Control-Allow-Origin")) + assert.Empty(t, get("https://new.example.com").Header().Get("Access-Control-Allow-Origin"), "not yet allowed") + + // "Reload": swap the list the getter returns. + mu.Lock() + origins = []string{"https://new.example.com"} + mu.Unlock() + + assert.Equal(t, "https://new.example.com", get("https://new.example.com").Header().Get("Access-Control-Allow-Origin"), "new allowlist applies on the next request") + assert.Empty(t, get("https://old.example.com").Header().Get("Access-Control-Allow-Origin"), "old origin no longer allowed") +} + func TestCORSMiddleware_Preflight(t *testing.T) { t.Parallel() - handler := corsMiddleware([]string{"*"})(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + handler := corsMiddleware(func() []string { return []string{"*"} })(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { t.Fatal("should not reach handler on OPTIONS") })) @@ -142,7 +177,7 @@ func TestCORSMiddleware_Preflight(t *testing.T) { func TestCORSMiddleware_NormalRequest(t *testing.T) { t.Parallel() var called bool - handler := corsMiddleware([]string{"*"})(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + handler := corsMiddleware(func() []string { return []string{"*"} })(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { called = true w.WriteHeader(http.StatusOK) })) @@ -160,7 +195,7 @@ func TestCORSMiddleware_NormalRequest(t *testing.T) { func TestCORSMiddleware_AllowListedOrigin(t *testing.T) { t.Parallel() var called bool - handler := corsMiddleware([]string{"https://app.example.com"})(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + handler := corsMiddleware(func() []string { return []string{"https://app.example.com"} })(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { called = true w.WriteHeader(http.StatusOK) })) @@ -179,7 +214,7 @@ func TestCORSMiddleware_AllowListedOrigin(t *testing.T) { func TestCORSMiddleware_BlockedOrigin(t *testing.T) { t.Parallel() var called bool - handler := corsMiddleware([]string{"https://allowed.com"})(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + handler := corsMiddleware(func() []string { return []string{"https://allowed.com"} })(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { called = true w.WriteHeader(http.StatusOK) })) @@ -218,7 +253,7 @@ func TestCORSMiddleware_NoCredentialsHeader(t *testing.T) { for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { t.Parallel() - handler := corsMiddleware(tc.allowed)(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + handler := corsMiddleware(func() []string { return tc.allowed })(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusOK) })) req := httptest.NewRequestWithContext(context.Background(), http.MethodGet, "/", nil) @@ -235,7 +270,7 @@ func TestCORSMiddleware_NoCredentialsHeader(t *testing.T) { // callers don't get CORS response headers stamped onto every response. func TestCORSMiddleware_NoOriginIsPassthrough(t *testing.T) { t.Parallel() - handler := corsMiddleware([]string{"*"})(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + handler := corsMiddleware(func() []string { return []string{"*"} })(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusOK) })) req := httptest.NewRequestWithContext(context.Background(), http.MethodGet, "/", nil) @@ -253,7 +288,7 @@ func TestCORSMiddleware_NoOriginIsPassthrough(t *testing.T) { // that as a preflight failure, so the actual request never fires. func TestCORSMiddleware_BlockedOriginPreflight(t *testing.T) { t.Parallel() - handler := corsMiddleware([]string{"https://allowed.com"})(http.HandlerFunc(func(_ http.ResponseWriter, _ *http.Request) { + handler := corsMiddleware(func() []string { return []string{"https://allowed.com"} })(http.HandlerFunc(func(_ http.ResponseWriter, _ *http.Request) { t.Fatal("should not reach handler on OPTIONS") })) @@ -368,7 +403,7 @@ func TestNewRouter_CORSOnStream(t *testing.T) { SSE: NewStreamHandler(hub, nil), Health: &HealthHandler{}, AuthMW: func(next http.Handler) http.Handler { return next }, - CORSOrigins: []string{"https://app.example.com"}, + CORSOrigins: func() []string { return []string{"https://app.example.com"} }, Logger: testutil.NopLogger(), }) diff --git a/internal/api/settings.go b/internal/api/settings.go new file mode 100644 index 00000000..0a22cfa0 --- /dev/null +++ b/internal/api/settings.go @@ -0,0 +1,51 @@ +package api + +import ( + "encoding/json" + "log/slog" + "net/http" + + "github.com/Wave-RF/WaveHouse/internal/settings" +) + +// SettingsHandler serves the ops surface for the hot-reloadable settings +// directory. Constructed only when a settings directory is configured — with +// none there is nothing to reload, so the route is simply absent (the same +// pattern as the DLQ and policy handlers). +type SettingsHandler struct { + Store *settings.Store + logger *slog.Logger +} + +func NewSettingsHandler(store *settings.Store, logger *slog.Logger) *SettingsHandler { + return &SettingsHandler{Store: store, logger: logger} +} + +// reloadResponse is the POST /v1/ops/settings/reload body: whether the +// directory was adopted, and every finding from the validation pass. +// Findings is never null — an empty array keeps clients off the +// nil-vs-empty distinction. +type reloadResponse struct { + Adopted bool `json:"adopted"` + Findings []settings.Finding `json:"findings"` +} + +// Reload handles POST /v1/ops/settings/reload — the API trigger for the same +// serialized reload path SIGHUP and the directory watcher run. 200 when the +// directory was adopted (warnings included in the body), 422 when validation +// rejected it and the previous settings remain in effect. +func (h *SettingsHandler) Reload(w http.ResponseWriter, _ *http.Request) { + findings, adopted := h.Store.TriggerReload("api") + if findings == nil { + findings = []settings.Finding{} + } + status := http.StatusOK + if !adopted { + status = http.StatusUnprocessableEntity + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + if err := json.NewEncoder(w).Encode(reloadResponse{Adopted: adopted, Findings: findings}); err != nil { + h.logger.Error("settings reload response encode", "error", err) + } +} diff --git a/internal/api/settings_test.go b/internal/api/settings_test.go new file mode 100644 index 00000000..3726438e --- /dev/null +++ b/internal/api/settings_test.go @@ -0,0 +1,73 @@ +package api + +import ( + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "testing" + + "github.com/Wave-RF/WaveHouse/internal/settings" + "github.com/Wave-RF/WaveHouse/internal/testutil" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// fullConfig is a complete config.json (every key is required) with the +// given query.default_max_rows. +func fullConfig(maxRows int) string { + return fmt.Sprintf(`{"dedupe": {"id_field": "event_id", "require_id": false}, "query": {"default_max_rows": %d}, "schema": {"refresh_interval": 60}, "cors": {"allowed_origins": ["*"]}}`, maxRows) +} + +// writeSettingsFixture materializes a minimal valid settings directory whose +// config.json content the caller controls. +func writeSettingsFixture(t *testing.T, configJSON string) string { + t.Helper() + dir := t.TempDir() + for name, content := range map[string]string{ + settings.FileRoles: `{"roles": ["public"]}`, + settings.FilePolicies: `{"default_role": "public"}`, + settings.FilePipes: `{}`, + settings.FileConfig: configJSON, + } { + require.NoError(t, os.WriteFile(filepath.Join(dir, name), []byte(content), 0o600)) + } + return dir +} + +// Subtests are ordered on purpose (the 422 case asserts the value the 200 +// case adopted survives), so neither the parent nor the subtests are parallel. +func TestSettingsReload(t *testing.T) { + dir := writeSettingsFixture(t, fullConfig(100)) + store, _ := settings.Open(dir, testutil.NopLogger()) + require.NotNil(t, store) + h := NewSettingsHandler(store, testutil.NopLogger()) + + post := func() (*httptest.ResponseRecorder, reloadResponse) { + rec := httptest.NewRecorder() + h.Reload(rec, httptest.NewRequestWithContext(t.Context(), http.MethodPost, "/v1/ops/settings/reload", nil)) + var body reloadResponse + require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &body)) + return rec, body + } + + t.Run("valid directory adopts with 200", func(t *testing.T) { + require.NoError(t, os.WriteFile(filepath.Join(dir, settings.FileConfig), []byte(fullConfig(200)), 0o600)) + rec, body := post() + assert.Equal(t, http.StatusOK, rec.Code) + assert.True(t, body.Adopted) + assert.NotNil(t, body.Findings, "findings must encode as an array, never null") + assert.Equal(t, 200, store.DefaultMaxRows(), "adopted settings must be live") + }) + + t.Run("invalid directory keeps previous settings with 422", func(t *testing.T) { + require.NoError(t, os.WriteFile(filepath.Join(dir, settings.FileConfig), []byte(`{"unknown_key": true}`), 0o600)) + rec, body := post() + assert.Equal(t, http.StatusUnprocessableEntity, rec.Code) + assert.False(t, body.Adopted) + assert.NotEmpty(t, body.Findings) + assert.Equal(t, 200, store.DefaultMaxRows(), "rejected reload must keep the previous snapshot") + }) +} diff --git a/internal/api/structured_query.go b/internal/api/structured_query.go index 54b4ffa0..3aaba5ec 100644 --- a/internal/api/structured_query.go +++ b/internal/api/structured_query.go @@ -27,8 +27,12 @@ type StructuredQueryHandler struct { BucketSecs int sf singleflight.Group maxQueryTimeout time.Duration - defaultMaxRows int - logger *slog.Logger + // defaultMaxRows returns the current fallback result LIMIT + // (settings.Store.DefaultMaxRows in production) — a func, not an int, so a + // settings reload takes effect on the next query without a restart. Nil or + // a non-positive return means the builder's compiled constant. + defaultMaxRows func() int + logger *slog.Logger // maxRequestBytes optionally overrides the default inbound request body // cap (maxControlBodyBytes). When 0, the default applies. Exists so @@ -45,7 +49,7 @@ func NewStructuredQueryHandler( policyStore *policy.Store, bucketSecs int, queryTimeout time.Duration, - defaultMaxRows int, + defaultMaxRows func() int, logger *slog.Logger, ) *StructuredQueryHandler { return &StructuredQueryHandler{ @@ -115,7 +119,11 @@ func (h *StructuredQueryHandler) Handle(w http.ResponseWriter, r *http.Request) // skip the check. The role's row-filter predicate and max_rows cap are emitted // by Build too, structurally (#322). A policy denial returns a typed error we // map to 403; a malformed query maps to 400. - result, err := query.Build(table, &sq, schema, perms, h.BucketSecs, h.defaultMaxRows) + maxRows := 0 // non-positive → the builder's compiled constant + if h.defaultMaxRows != nil { + maxRows = h.defaultMaxRows() + } + result, err := query.Build(table, &sq, schema, perms, h.BucketSecs, maxRows) if err != nil { // A query that selects nothing — no columns, no aggregations, no // select_all — is a request for no data, not an error: return an empty diff --git a/internal/api/structured_query_test.go b/internal/api/structured_query_test.go index 8ca67e34..cf7fffe4 100644 --- a/internal/api/structured_query_test.go +++ b/internal/api/structured_query_test.go @@ -40,7 +40,7 @@ func newStructuredQueryHandler(t testing.TB) *StructuredQueryHandler { }, }, }) - return NewStructuredQueryHandler(nil, nil, reg, nil, 60, 5*time.Second, 0, testutil.NopLogger()) + return NewStructuredQueryHandler(nil, nil, reg, nil, 60, 5*time.Second, nil, testutil.NopLogger()) } func TestStructuredQuery_MissingTable(t *testing.T) { @@ -295,7 +295,7 @@ func newCapturingHandler(t *testing.T, conn driver.Conn, p *policy.Policy) *Stru }, }, }) - return NewStructuredQueryHandler(conn, nil, reg, policy.NewMemoryStore(p), 60, 5*time.Second, 0, testutil.NopLogger()) + return NewStructuredQueryHandler(conn, nil, reg, policy.NewMemoryStore(p), 60, 5*time.Second, nil, testutil.NopLogger()) } func viewerRequest(t *testing.T, sq query.StructuredQuery) *http.Request { diff --git a/internal/config/config.go b/internal/config/config.go index a04361ec..9548c5d2 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -23,13 +23,11 @@ type Config struct { Dedupe Dedupe `yaml:"dedupe"` Cache Cache `yaml:"cache"` Auth Auth `yaml:"auth"` - Schema Schema `yaml:"schema"` DLQ DLQ `yaml:"dlq"` Policy Policy `yaml:"policy"` Pipes Pipes `yaml:"pipes"` OTel OTel `yaml:"otel"` Prometheus Prometheus `yaml:"prometheus"` - Query Query `yaml:"query"` Stream Stream `yaml:"stream"` Settings Settings `yaml:"settings"` } @@ -44,26 +42,16 @@ const EnvSettingsDir = "WH_SETTINGS_DIR" // Settings locates the hot-reloadable settings directory — the four JSON // documents (roles.json, policies.json, pipes.json, config.json) validated by // internal/settings. Boot-tier by necessity: it's the pointer the reload -// machinery follows, so it can't live behind itself. No default, same -// reasoning as policy.file_path: a baked-in path would turn a missing mount -// into silent misconfiguration instead of an explicit operator choice. +// machinery follows, so it can't live behind itself. REQUIRED, with no +// default, same reasoning as policy.file_path: a baked-in path would turn a +// missing mount into silent misconfiguration instead of an explicit operator +// choice, and the binary has no compiled tunable defaults to fall back on — +// `wavehouse init-settings ` writes the starter directory (the container +// images bake it at /app/settings). type Settings struct { Dir string `yaml:"dir" env:"WH_SETTINGS_DIR"` } -// Query holds query-shaping defaults. Server-wide *resource* limits (memory, -// rows scanned, execution time) deliberately live in ClickHouse itself — its -// settings profiles and quotas, see docs/configuration — so they apply -// uniformly to every query (including raw admin SQL) and compose with the -// per-role caps WaveHouse adds via per-query settings. This block holds only -// the result-shaping default that is genuinely WaveHouse's to own. -type Query struct { - // DefaultMaxRows is the result LIMIT applied to a structured query when the - // caller and policy specify none — the visible, tunable form of what used to - // be the hard-coded query.DefaultMaxRows. 0 falls back to that constant. - DefaultMaxRows int `yaml:"default_max_rows" env:"WH_QUERY_DEFAULT_MAX_ROWS" env-default:"10000"` -} - // OTel configures the OpenTelemetry pipeline. `enabled` is the master switch; // when false, no signals are initialized regardless of the per-signal toggles. // The OTLP destination — endpoint, TLS, custom CA, mutual TLS, and auth headers @@ -117,10 +105,11 @@ type OTelLogs struct { SampleRate float64 `yaml:"sample_rate" env:"WH_OTEL_LOGS_SAMPLE_RATE" env-default:"1.0"` } +// Server holds listener wiring. The CORS allowlist is a tenant tunable and +// lives in the settings directory's config.json (internal/settings). type Server struct { - Port int `yaml:"port" env:"WH_SERVER_PORT" env-default:"8080"` - ShutdownTimeout int `yaml:"shutdown_timeout" env:"WH_SERVER_SHUTDOWN_TIMEOUT" env-default:"10"` - CORSAllowedOrigins []string `yaml:"cors_allowed_origins" env:"WH_SERVER_CORS_ALLOWED_ORIGINS" env-default:"*"` + Port int `yaml:"port" env:"WH_SERVER_PORT" env-default:"8080"` + ShutdownTimeout int `yaml:"shutdown_timeout" env:"WH_SERVER_SHUTDOWN_TIMEOUT" env-default:"10"` } type Stream struct { @@ -150,10 +139,12 @@ type MQ struct { MaxBytesGB int `yaml:"max_bytes_gb" env:"WH_MQ_MAX_BYTES_GB" env-default:"50"` } +// Dedupe holds only the enable switch: it owns Pebble's lifecycle, which only +// a restart can change. The behavioral knobs (id_field, require_id, per-table +// overrides) are tenant tunables and live in the settings directory's +// config.json (internal/settings). type Dedupe struct { - Enabled bool `yaml:"enabled" env:"WH_DEDUPE_ENABLED" env-default:"false"` - IDField string `yaml:"id_field" env:"WH_DEDUPE_ID_FIELD" env-default:"event_id"` - RequireID bool `yaml:"require_id" env:"WH_DEDUPE_REQUIRE_ID" env-default:"false"` + Enabled bool `yaml:"enabled" env:"WH_DEDUPE_ENABLED" env-default:"false"` } type Cache struct { @@ -209,11 +200,6 @@ type Pipes struct { Dir string `yaml:"dir" env:"WH_PIPES_DIR" env-default:""` } -// Schema configures ClickHouse schema discovery. -type Schema struct { - RefreshInterval int `yaml:"refresh_interval" env:"WH_SCHEMA_REFRESH_INTERVAL" env-default:"60"` // seconds -} - // DLQ configures the Dead Letter Queue for failed batch inserts. type DLQ struct { Enabled bool `yaml:"enabled" env:"WH_DLQ_ENABLED" env-default:"true"` @@ -233,6 +219,10 @@ func (c *Config) Validate() error { return fmt.Errorf("server.shutdown_timeout must be non-negative") } + if strings.TrimSpace(c.Settings.Dir) == "" { + return fmt.Errorf("settings.dir (%s) is required: point it at a settings directory, or create one with `wavehouse init-settings `", EnvSettingsDir) + } + if c.Stream.KeepaliveInterval < 0 { return fmt.Errorf("stream.keepalive_interval must be non-negative, got %s", c.Stream.KeepaliveInterval) } @@ -240,22 +230,10 @@ func (c *Config) Validate() error { return fmt.Errorf("stream.keepalive_buckets must be non-negative, got %d", c.Stream.KeepaliveBuckets) } - if c.Schema.RefreshInterval < 1 { - return fmt.Errorf("schema.refresh_interval must be >= 1 second") - } - if c.ClickHouse.QueryTimeout <= time.Duration(0) { return fmt.Errorf("clickhouse.query_timeout must be > 0, got %s", c.ClickHouse.QueryTimeout) } - // query.default_max_rows is the fallback result LIMIT. 0 (or a directly-built - // config that omits it) means "use the built-in query.DefaultMaxRows" — the - // builder substitutes the constant for any non-positive value — so only a - // negative value is an error. - if c.Query.DefaultMaxRows < 0 { - return fmt.Errorf("query.default_max_rows must be non-negative, got %d", c.Query.DefaultMaxRows) - } - if c.MQ.GapWindowMinutes < 0 { return fmt.Errorf("mq.gap_window_minutes must be non-negative") } diff --git a/internal/config/config_test.go b/internal/config/config_test.go index eaf1036f..dbb1be05 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -11,6 +11,28 @@ import ( "github.com/stretchr/testify/require" ) +// TestMain pins WH_SETTINGS_DIR for the whole package: settings.dir is a +// required boot key, and the Load tests run in parallel (so t.Setenv is out). +// Tests that exercise the requirement itself build a Config literal. +func TestMain(m *testing.M) { + if err := os.Setenv("WH_SETTINGS_DIR", "./settings"); err != nil { + panic(err) + } + os.Exit(m.Run()) +} + +func TestValidate_SettingsDirRequired(t *testing.T) { + t.Parallel() + cfg := Config{ + Server: Server{Port: 8080}, + ClickHouse: ClickHouse{HTTPScheme: "http", QueryTimeout: 30 * time.Second}, + } + err := cfg.Validate() + require.Error(t, err) + assert.Contains(t, err.Error(), "settings.dir") + assert.Contains(t, err.Error(), "init-settings") +} + func TestLoad_Defaults(t *testing.T) { t.Parallel() cfg, err := Load("nonexistent.yaml") @@ -28,12 +50,10 @@ func TestLoad_Defaults(t *testing.T) { assert.Equal(t, "role", cfg.Auth.RoleClaim) assert.Empty(t, cfg.Auth.OperatorKey, "operator key is empty by default (feature off)") assert.False(t, cfg.Dedupe.Enabled) - assert.Equal(t, "event_id", cfg.Dedupe.IDField) assert.True(t, cfg.DLQ.Enabled) assert.Empty(t, cfg.Policy.FilePath, "no default bootstrap file — operators opt in explicitly so a missing file never produces a silent fail-closed boot") assert.Equal(t, "", cfg.Pipes.Dir) assert.Equal(t, "./data", cfg.DataDir) - assert.Equal(t, 60, cfg.Schema.RefreshInterval) assert.False(t, cfg.OTel.Enabled) assert.True(t, cfg.OTel.Traces.Enabled) assert.InEpsilon(t, 1.0, cfg.OTel.Traces.SampleRate, 0.0001) @@ -78,50 +98,13 @@ func TestLoad_OperatorKey_FromEnv(t *testing.T) { assert.Equal(t, "env-operator-key", cfg.Auth.OperatorKey) } -func TestLoad_QueryLimits_Defaults(t *testing.T) { - t.Parallel() - cfg, err := Load("nonexistent.yaml") - require.NoError(t, err) - // Only the result-LIMIT default lives in WaveHouse config now; server-wide - // resource limits (memory, rows scanned, time) are ClickHouse's job. - assert.Equal(t, 10000, cfg.Query.DefaultMaxRows) -} - -func TestLoad_QueryLimits_FromYAML(t *testing.T) { - t.Parallel() - dir := t.TempDir() - yamlContent := ` -query: - default_max_rows: 25000 -` - path := filepath.Join(dir, "config.yaml") - require.NoError(t, os.WriteFile(path, []byte(yamlContent), 0o600)) - - cfg, err := Load(path) - require.NoError(t, err) - assert.Equal(t, 25000, cfg.Query.DefaultMaxRows) -} - -func TestValidate_NegativeQueryDefaultMaxRows(t *testing.T) { - t.Parallel() - cfg := Config{ - Server: Server{Port: 8080}, - ClickHouse: ClickHouse{HTTPScheme: "http", QueryTimeout: 30 * time.Second}, - Schema: Schema{RefreshInterval: 60}, - Query: Query{DefaultMaxRows: -1}, - } - err := cfg.Validate() - require.Error(t, err) - assert.Contains(t, err.Error(), "default_max_rows") -} - func TestValidate_KeepaliveValues(t *testing.T) { t.Parallel() base := func() Config { return Config{ Server: Server{Port: 8080}, + Settings: Settings{Dir: "./settings"}, ClickHouse: ClickHouse{HTTPScheme: "http", QueryTimeout: 30 * time.Second}, - Schema: Schema{RefreshInterval: 60}, } } @@ -203,8 +186,8 @@ func TestValidate_PortOutOfRange(t *testing.T) { t.Parallel() cfg := Config{ Server: Server{Port: tt.port}, + Settings: Settings{Dir: "./settings"}, ClickHouse: ClickHouse{HTTPScheme: "http", QueryTimeout: time.Duration(30) * time.Second}, - Schema: Schema{RefreshInterval: 60}, } err := cfg.Validate() require.Error(t, err) @@ -217,32 +200,20 @@ func TestValidate_NegativeShutdownTimeout(t *testing.T) { t.Parallel() cfg := Config{ Server: Server{Port: 8080, ShutdownTimeout: -1}, + Settings: Settings{Dir: "./settings"}, ClickHouse: ClickHouse{HTTPScheme: "http", QueryTimeout: time.Duration(30) * time.Second}, - Schema: Schema{RefreshInterval: 60}, } err := cfg.Validate() require.Error(t, err) assert.Contains(t, err.Error(), "shutdown_timeout") } -func TestValidate_SchemaRefreshIntervalZero(t *testing.T) { - t.Parallel() - cfg := Config{ - Server: Server{Port: 8080}, - ClickHouse: ClickHouse{HTTPScheme: "http", QueryTimeout: time.Duration(30) * time.Second}, - Schema: Schema{RefreshInterval: 0}, - } - err := cfg.Validate() - require.Error(t, err) - assert.Contains(t, err.Error(), "schema.refresh_interval") -} - func TestValidate_NegativeQueryTime(t *testing.T) { t.Parallel() cfg := Config{ Server: Server{Port: 8080}, + Settings: Settings{Dir: "./settings"}, ClickHouse: ClickHouse{HTTPScheme: "http", QueryTimeout: -1}, - Schema: Schema{RefreshInterval: 60}, } err := cfg.Validate() require.Error(t, err) @@ -253,8 +224,8 @@ func TestValidate_ZeroQueryTime(t *testing.T) { t.Parallel() cfg := Config{ Server: Server{Port: 8080}, + Settings: Settings{Dir: "./settings"}, ClickHouse: ClickHouse{HTTPScheme: "http", QueryTimeout: 0}, - Schema: Schema{RefreshInterval: 60}, } err := cfg.Validate() require.Error(t, err) @@ -265,9 +236,9 @@ func TestValidate_NegativeGapWindow(t *testing.T) { t.Parallel() cfg := Config{ Server: Server{Port: 8080}, + Settings: Settings{Dir: "./settings"}, ClickHouse: ClickHouse{HTTPScheme: "http", QueryTimeout: time.Duration(30) * time.Second}, MQ: MQ{GapWindowMinutes: -1}, - Schema: Schema{RefreshInterval: 60}, } err := cfg.Validate() require.Error(t, err) @@ -289,8 +260,8 @@ func TestValidate_TracesSampleRateOutOfRange(t *testing.T) { t.Parallel() cfg := Config{ Server: Server{Port: 8080}, + Settings: Settings{Dir: "./settings"}, ClickHouse: ClickHouse{HTTPScheme: "http", QueryTimeout: time.Duration(30) * time.Second}, - Schema: Schema{RefreshInterval: 60}, OTel: OTel{ Enabled: true, Traces: OTelTraces{Enabled: true, SampleRate: tc.rate}, @@ -308,8 +279,8 @@ func TestValidate_LogsSampleRateOutOfRange(t *testing.T) { t.Parallel() cfg := Config{ Server: Server{Port: 8080}, + Settings: Settings{Dir: "./settings"}, ClickHouse: ClickHouse{HTTPScheme: "http", QueryTimeout: time.Duration(30) * time.Second}, - Schema: Schema{RefreshInterval: 60}, OTel: OTel{ Enabled: true, Traces: OTelTraces{Enabled: true, SampleRate: 0.10}, @@ -328,8 +299,8 @@ func TestValidate_SampleRatesIgnoredWhenObservabilityDisabled(t *testing.T) { // config that they haven't enabled yet. cfg := Config{ Server: Server{Port: 8080}, + Settings: Settings{Dir: "./settings"}, ClickHouse: ClickHouse{HTTPScheme: "http", QueryTimeout: time.Duration(30) * time.Second}, - Schema: Schema{RefreshInterval: 60}, OTel: OTel{ Enabled: false, Traces: OTelTraces{SampleRate: 99}, @@ -345,8 +316,8 @@ func TestValidate_SampleRatesIgnoredWhenSignalDisabled(t *testing.T) { // signal is off, its sample_rate is unused and should not gate startup. cfg := Config{ Server: Server{Port: 8080}, + Settings: Settings{Dir: "./settings"}, ClickHouse: ClickHouse{HTTPScheme: "http", QueryTimeout: time.Duration(30) * time.Second}, - Schema: Schema{RefreshInterval: 60}, OTel: OTel{ Enabled: true, Traces: OTelTraces{Enabled: false, SampleRate: 99}, @@ -370,8 +341,8 @@ func TestValidate_PrometheusPortCollidesWithServerPort(t *testing.T) { t.Parallel() cfg := Config{ Server: Server{Port: 8080}, + Settings: Settings{Dir: "./settings"}, ClickHouse: ClickHouse{HTTPScheme: "http", QueryTimeout: time.Duration(30) * time.Second}, - Schema: Schema{RefreshInterval: 60}, Prometheus: Prometheus{ Enabled: true, Path: "/metrics", @@ -398,8 +369,8 @@ func TestValidate_PrometheusPortOutOfRange(t *testing.T) { t.Parallel() cfg := Config{ Server: Server{Port: 8080}, + Settings: Settings{Dir: "./settings"}, ClickHouse: ClickHouse{HTTPScheme: "http", QueryTimeout: time.Duration(30) * time.Second}, - Schema: Schema{RefreshInterval: 60}, Prometheus: Prometheus{ Enabled: true, Path: "/metrics", @@ -417,8 +388,8 @@ func TestValidate_PrometheusPathMustStartWithSlash(t *testing.T) { t.Parallel() cfg := Config{ Server: Server{Port: 8080}, + Settings: Settings{Dir: "./settings"}, ClickHouse: ClickHouse{HTTPScheme: "http", QueryTimeout: time.Duration(30) * time.Second}, - Schema: Schema{RefreshInterval: 60}, Prometheus: Prometheus{ Enabled: true, Path: "metrics", // missing leading slash @@ -453,8 +424,8 @@ func TestValidate_PrometheusPathReservedConflicts(t *testing.T) { t.Parallel() cfg := Config{ Server: Server{Port: 8080}, + Settings: Settings{Dir: "./settings"}, ClickHouse: ClickHouse{HTTPScheme: "http", QueryTimeout: time.Duration(30) * time.Second}, - Schema: Schema{RefreshInterval: 60}, Prometheus: Prometheus{Enabled: true, Path: tc.path, Port: tc.port}, } err := cfg.Validate() @@ -471,8 +442,8 @@ func TestValidate_PrometheusV1PathAllowedOnSidecarPort(t *testing.T) { // path doesn't collide with the API. Validation should let this through. cfg := Config{ Server: Server{Port: 8080}, + Settings: Settings{Dir: "./settings"}, ClickHouse: ClickHouse{HTTPScheme: "http", QueryTimeout: time.Duration(30) * time.Second}, - Schema: Schema{RefreshInterval: 60}, Prometheus: Prometheus{Enabled: true, Path: "/v1/metrics", Port: 9091}, } assert.NoError(t, cfg.Validate()) @@ -484,8 +455,8 @@ func TestValidate_PrometheusOnly_NoOTel(t *testing.T) { // otel.enabled stays false, prometheus.enabled is true. Must validate. cfg := Config{ Server: Server{Port: 8080}, + Settings: Settings{Dir: "./settings"}, ClickHouse: ClickHouse{HTTPScheme: "http", QueryTimeout: time.Duration(30) * time.Second}, - Schema: Schema{RefreshInterval: 60}, Prometheus: Prometheus{Enabled: true, Path: "/metrics", Port: 0}, } assert.NoError(t, cfg.Validate()) @@ -498,8 +469,8 @@ func TestValidate_PrometheusIgnoredWhenDisabled(t *testing.T) { // get yelled at about unused fields. cfg := Config{ Server: Server{Port: 8080}, + Settings: Settings{Dir: "./settings"}, ClickHouse: ClickHouse{HTTPScheme: "http", QueryTimeout: time.Duration(30) * time.Second}, - Schema: Schema{RefreshInterval: 60}, Prometheus: Prometheus{ Enabled: false, Path: "garbage", @@ -513,8 +484,8 @@ func TestValidate_InvalidHTTPScheme(t *testing.T) { t.Parallel() cfg := Config{ Server: Server{Port: 8080}, + Settings: Settings{Dir: "./settings"}, ClickHouse: ClickHouse{HTTPScheme: "ftp", QueryTimeout: time.Duration(30) * time.Second}, // Intentionally invalid ftp - Schema: Schema{RefreshInterval: 60}, } err := cfg.Validate() diff --git a/internal/discovery/discovery.go b/internal/discovery/discovery.go index 0e45bf0a..5ef77257 100644 --- a/internal/discovery/discovery.go +++ b/internal/discovery/discovery.go @@ -50,9 +50,13 @@ type SchemaRegistry struct { conn driver.Conn database string refreshInterval time.Duration - logger *slog.Logger - mu sync.RWMutex - tables map[string]*TableSchema + // intervalFn, when set via SetIntervalSource, supplies the current + // auto-refresh interval on each tick so a settings reload retunes the + // cadence without restarting the loop. + intervalFn func() time.Duration + logger *slog.Logger + mu sync.RWMutex + tables map[string]*TableSchema } // NewSchemaRegistry creates a registry that discovers schemas from system.columns. @@ -210,10 +214,32 @@ func (sr *SchemaRegistry) RetryRefresh(ctx context.Context, initialBackoff, maxB } } +// SetIntervalSource installs a dynamic source for the auto-refresh interval +// (settings.Store.SchemaRefreshInterval in production), polled after each tick. +// Call before StartAutoRefresh; the constructor's interval remains the +// fallback whenever fn is nil or returns a non-positive duration. +func (sr *SchemaRegistry) SetIntervalSource(fn func() time.Duration) { + sr.intervalFn = fn +} + +// currentInterval resolves the effective auto-refresh interval. +func (sr *SchemaRegistry) currentInterval() time.Duration { + if sr.intervalFn != nil { + if d := sr.intervalFn(); d > 0 { + return d + } + } + return sr.refreshInterval +} + // StartAutoRefresh runs a background goroutine that refreshes schemas -// at the configured interval. Blocks until ctx is cancelled. +// at the configured interval. Blocks until ctx is cancelled. The interval is +// re-read after every tick, so a changed setting applies from the next cycle +// — an in-flight wait finishes at the old cadence rather than resetting, +// which keeps a reload from ever deferring an imminent refresh. func (sr *SchemaRegistry) StartAutoRefresh(ctx context.Context) { - ticker := time.NewTicker(sr.refreshInterval) + interval := sr.currentInterval() + ticker := time.NewTicker(interval) defer ticker.Stop() for { select { @@ -223,6 +249,10 @@ func (sr *SchemaRegistry) StartAutoRefresh(ctx context.Context) { if err := sr.Refresh(ctx); err != nil { sr.logger.Error("schema auto-refresh failed", "error", err) } + if next := sr.currentInterval(); next != interval { + interval = next + ticker.Reset(interval) + } } } } diff --git a/internal/settings/finding.go b/internal/settings/finding.go index 76c7069a..1374bb9e 100644 --- a/internal/settings/finding.go +++ b/internal/settings/finding.go @@ -15,11 +15,13 @@ const ( // Finding is one validation result, located as precisely as the failure // allows: File is empty for directory-level findings, Path is a dotted JSON // path ("tables.clicks.select.analyst") and empty for whole-file findings. +// The JSON shape is part of the ops API: POST /v1/ops/settings/reload returns +// findings verbatim. type Finding struct { - Severity Severity - File string - Path string - Message string + Severity Severity `json:"severity"` + File string `json:"file,omitempty"` + Path string `json:"path,omitempty"` + Message string `json:"message"` } // String renders "error: policies.json: tables.clicks.select.admin: ", diff --git a/internal/settings/seed.go b/internal/settings/seed.go new file mode 100644 index 00000000..1b4d0cd0 --- /dev/null +++ b/internal/settings/seed.go @@ -0,0 +1,61 @@ +package settings + +import ( + "embed" + "fmt" + "io/fs" + "os" + "path/filepath" +) + +// seedFS holds the starter settings directory: every file present, every +// key set to its default. It is the ONE place defaults live — the binary +// has no compiled fallbacks — and it ships three ways: `wavehouse +// init-settings` writes it out, the container images bake it at +// /app/settings, and the compose quickstart serves from that copy. +// +//go:embed seed/*.json +var seedFS embed.FS + +// Seed returns the starter settings directory's files by name. +func Seed() (map[string][]byte, error) { + out := make(map[string][]byte, len(Files())) + for _, name := range Files() { + data, err := fs.ReadFile(seedFS, "seed/"+name) + if err != nil { + return nil, fmt.Errorf("embedded seed %s: %w", name, err) + } + out[name] = data + } + return out, nil +} + +// WriteSeed materializes the starter directory at dir, creating it if +// needed. It refuses a non-empty directory (the `initdb` contract): an +// existing settings directory is someone's config, never something to +// overwrite. +func WriteSeed(dir string) error { + // World-readable on purpose: the operator who runs init-settings is + // routinely not the user the server runs as (root on the host, UID 65532 + // in the container), and the server only needs to read it. + if err := os.MkdirAll(dir, 0o755); err != nil { //nolint:gosec // G301: config directory the server reads as another user + return err + } + entries, err := os.ReadDir(dir) + if err != nil { + return err + } + if len(entries) > 0 { + return fmt.Errorf("%s is not empty — refusing to overwrite an existing settings directory", dir) + } + files, err := Seed() + if err != nil { + return err + } + for _, name := range Files() { + if err := os.WriteFile(filepath.Join(dir, name), files[name], 0o644); err != nil { //nolint:gosec // G306: settings files hold no secrets and are read by the server user + return err + } + } + return nil +} diff --git a/internal/settings/seed/config.json b/internal/settings/seed/config.json new file mode 100644 index 00000000..9528215d --- /dev/null +++ b/internal/settings/seed/config.json @@ -0,0 +1,16 @@ +{ + "dedupe": { + "id_field": "event_id", + "require_id": false, + "tables": {} + }, + "query": { + "default_max_rows": 10000 + }, + "schema": { + "refresh_interval": 60 + }, + "cors": { + "allowed_origins": ["*"] + } +} diff --git a/internal/settings/seed/pipes.json b/internal/settings/seed/pipes.json new file mode 100644 index 00000000..3496521e --- /dev/null +++ b/internal/settings/seed/pipes.json @@ -0,0 +1,3 @@ +{ + "pipes": [] +} diff --git a/internal/settings/seed/policies.json b/internal/settings/seed/policies.json new file mode 100644 index 00000000..0967ef42 --- /dev/null +++ b/internal/settings/seed/policies.json @@ -0,0 +1 @@ +{} diff --git a/internal/settings/seed/roles.json b/internal/settings/seed/roles.json new file mode 100644 index 00000000..a00f09f4 --- /dev/null +++ b/internal/settings/seed/roles.json @@ -0,0 +1,3 @@ +{ + "roles": [] +} diff --git a/internal/settings/settings.go b/internal/settings/settings.go index 8fddd083..f8089817 100644 --- a/internal/settings/settings.go +++ b/internal/settings/settings.go @@ -60,27 +60,28 @@ type PipesFile struct { // that migrate out of boot config (config.yaml/env keeps what the platform // operator owns: wiring, lifecycle, secrets — and platform-infra knobs like // the SSE keepalives, which exist for the deployment's proxies, not the -// tenant). Every field is a pointer or slice — absent means "compiled -// default", so validation only judges values the author actually wrote. +// tenant). Every block and every top-level key inside it is REQUIRED: the +// binary carries no compiled defaults, so the adopted snapshot is exactly +// what the files say. Defaults live in the seed directory (see Seed) that +// `wavehouse init-settings` writes. The fields are pointers only so Validate +// can tell "absent" from the zero value and report it by path. type TenantConfig struct { - Dedupe *DedupeConfig `json:"dedupe,omitempty"` - Query *QueryConfig `json:"query,omitempty"` - Schema *SchemaConfig `json:"schema,omitempty"` - CORS *CORSConfig `json:"cors,omitempty"` + Dedupe *DedupeConfig `json:"dedupe"` + Query *QueryConfig `json:"query"` + Schema *SchemaConfig `json:"schema"` + CORS *CORSConfig `json:"cors"` } // DedupeConfig tunes dedupe behavior. dedupe.enabled stays boot config: it // owns Pebble's lifecycle, which only a restart can change. // -// Each field resolves independently through a three-level cascade: table -// override → the global value here → compiled default. Absent means inherit, -// and an explicit empty, whitespace-only, or whitespace-padded id_field is -// rejected at every level, so the effective id_field can never be empty or -// silently unmatchable — require_id without an id_field simply requires the -// inherited one. +// id_field and require_id are required here and optional per table: a table +// override inherits whichever field it doesn't name. An empty, +// whitespace-only, or whitespace-padded id_field is rejected at every level, +// so the effective id_field can never be empty or silently unmatchable. type DedupeConfig struct { - IDField *string `json:"id_field,omitempty"` - RequireID *bool `json:"require_id,omitempty"` + IDField *string `json:"id_field"` + RequireID *bool `json:"require_id"` // Tables holds per-table overrides keyed by ClickHouse table name (#222). // Names are format-checked only — existence is schema discovery's runtime // concern, same as policies.json table keys. @@ -94,18 +95,25 @@ type TableDedupe struct { RequireID *bool `json:"require_id,omitempty"` } -// QueryConfig mirrors the boot config Query block. +// QueryConfig holds query-shaping defaults. Server-wide *resource* limits +// (memory, rows scanned, execution time) deliberately live in ClickHouse +// itself — its settings profiles and quotas — so they apply uniformly to +// every query; this block holds only the result-LIMIT default. type QueryConfig struct { - DefaultMaxRows *int `json:"default_max_rows,omitempty"` + // DefaultMaxRows is the result LIMIT applied to a structured query when + // the caller and policy specify none. Must be >= 1. + DefaultMaxRows *int `json:"default_max_rows"` } -// SchemaConfig mirrors the boot config Schema block (seconds). +// SchemaConfig tunes ClickHouse schema discovery. type SchemaConfig struct { - RefreshInterval *int `json:"refresh_interval,omitempty"` + // RefreshInterval is the auto-refresh period in seconds. Must be >= 1. + RefreshInterval *int `json:"refresh_interval"` } -// CORSConfig carries the per-request CORS allowlist (boot config -// server.cors_allowed_origins today). +// CORSConfig carries the per-request CORS allowlist. ["*"] allows any +// browser origin; see corsMiddleware in internal/api for why that is safe +// for a Bearer-token API. type CORSConfig struct { - AllowedOrigins []string `json:"allowed_origins,omitempty"` + AllowedOrigins []string `json:"allowed_origins"` } diff --git a/internal/settings/store.go b/internal/settings/store.go new file mode 100644 index 00000000..1d048244 --- /dev/null +++ b/internal/settings/store.go @@ -0,0 +1,131 @@ +package settings + +import ( + "log/slog" + "sync" + "sync/atomic" + "time" +) + +// Store owns the settings snapshot a running instance has adopted. Open +// validates and adopts the directory up front, so a *Store never exists +// without a good document behind it. Reload is the single code path every +// later trigger — SIGHUP, the directory watcher, and POST +// /v1/ops/settings/reload — funnels through: re-validate the directory, and +// swap the snapshot only when no finding is an error, so a bad edit (or a +// deleted file, or a vanished directory) can never evict the last good +// document. Readers go through one lock-free atomic load per lookup; the +// typed accessors below each resolve from a single snapshot load, so a +// reload lands between lookups, never inside one. +// +// There are no compiled defaults here on purpose: every key is required by +// Validate, so the snapshot is exactly what the files said when they were +// adopted. Defaults live in the seed directory (Seed / WriteSeed). +type Store struct { + dir string + logger *slog.Logger + + // mu serializes Reload: concurrent triggers queue rather than racing + // parse-then-swap sequences (a stale parse must not overwrite a newer one). + mu sync.Mutex + snap atomic.Pointer[Document] +} + +// Open validates dir and returns a Store holding its document. A rejected +// directory returns a nil Store with the findings — the caller (boot) +// refuses to start; it must never run without adopted settings. +func Open(dir string, logger *slog.Logger) (*Store, []Finding) { + s := &Store{dir: dir, logger: logger} + findings, adopted := s.TriggerReload("boot") + if !adopted { + return nil, findings + } + return s, findings +} + +// Dir returns the directory this store reads. +func (s *Store) Dir() string { return s.dir } + +// Reload re-validates the directory and adopts the parsed document when no +// finding is an error (warnings don't block adoption, matching `wavehouse +// validate`). On a rejected reload the previous snapshot stays in place. +// The returned bool reports whether the document was adopted. +func (s *Store) Reload() ([]Finding, bool) { + s.mu.Lock() + defer s.mu.Unlock() + doc, findings := parse(s.dir) + if doc == nil { + return findings, false + } + s.snap.Store(doc) + return findings, true +} + +// TriggerReload is Reload plus outcome logging, tagged with the trigger's +// name ("boot", "sighup", "watch", "api") so operators can tell which path +// fired from the log line alone. +func (s *Store) TriggerReload(trigger string) ([]Finding, bool) { + findings, adopted := s.Reload() + if s.logger != nil { + var errs, warns int + for _, f := range findings { + if f.Severity == SeverityError { + errs++ + s.logger.Error("settings finding", "trigger", trigger, "finding", f.String()) + } else { + warns++ + s.logger.Warn("settings finding", "trigger", trigger, "finding", f.String()) + } + } + switch { + case adopted: + s.logger.Info("settings adopted", "trigger", trigger, "dir", s.dir, "warnings", warns) + case s.snap.Load() == nil: + s.logger.Error("settings rejected", "trigger", trigger, "dir", s.dir, "errors", errs, "warnings", warns) + default: + s.logger.Error("settings rejected — keeping previous settings", "trigger", trigger, "dir", s.dir, "errors", errs, "warnings", warns) + } + } + return findings, adopted +} + +// doc returns the current snapshot. Never nil for a Store returned by Open: +// adoption happened before the Store was handed out, and a rejected reload +// leaves the previous document in place. +func (s *Store) doc() *Document { + return s.snap.Load() +} + +// DedupeFor resolves the effective dedupe settings for a table: the table +// override for each field it names, the global value otherwise. Both fields +// resolve from one snapshot load, so a reload can never hand a record the +// id_field of one document and the require_id of another. +func (s *Store) DedupeFor(table string) (idField string, requireID bool) { + d := s.doc().Config.Dedupe + idField, requireID = *d.IDField, *d.RequireID + if td, ok := d.Tables[table]; ok { + if td.IDField != nil { + idField = *td.IDField + } + if td.RequireID != nil { + requireID = *td.RequireID + } + } + return idField, requireID +} + +// DefaultMaxRows returns the fallback result LIMIT for structured queries. +func (s *Store) DefaultMaxRows() int { + return *s.doc().Config.Query.DefaultMaxRows +} + +// SchemaRefreshInterval returns the schema-discovery auto-refresh period. +func (s *Store) SchemaRefreshInterval() time.Duration { + return time.Duration(*s.doc().Config.Schema.RefreshInterval) * time.Second +} + +// CORSOrigins returns the allowed CORS origins; the middleware treats an +// empty list and ["*"] identically (allow-all). +func (s *Store) CORSOrigins() []string { + return s.doc().Config.CORS.AllowedOrigins +} diff --git a/internal/settings/store_test.go b/internal/settings/store_test.go new file mode 100644 index 00000000..73a1b96d --- /dev/null +++ b/internal/settings/store_test.go @@ -0,0 +1,157 @@ +package settings + +import ( + "os" + "path/filepath" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// newLoadedStore materializes a valid directory (with overrides applied) and +// returns a store that has adopted it. +func newLoadedStore(t *testing.T, overrides map[string]string) *Store { + t.Helper() + files := validFiles() + for name, content := range overrides { + files[name] = content + } + s, findings := Open(writeDir(t, files), nil) + require.NotNil(t, s, "findings: %s", findingStrings(findings)) + return s +} + +func TestStore_ReloadAdoptsAndRejects(t *testing.T) { + t.Parallel() + s := newLoadedStore(t, map[string]string{ + FileConfig: configJSON(`{"query": {"default_max_rows": 500}}`), + }) + assert.Equal(t, 500, s.DefaultMaxRows()) + + // Break the directory: the reload must report the error and keep the + // previous snapshot — a bad edit can never evict the last good document. + require.NoError(t, os.WriteFile(filepath.Join(s.Dir(), FileConfig), []byte(configJSON(`{"query": {"default_max_rows": -1}}`)), 0o600)) + findings, adopted := s.Reload() + assert.False(t, adopted) + assert.True(t, HasErrors(findings)) + assert.Equal(t, 500, s.DefaultMaxRows(), "rejected reload must keep the previous snapshot") + + // Fix it: the next reload adopts again. + require.NoError(t, os.WriteFile(filepath.Join(s.Dir(), FileConfig), []byte(configJSON(`{"query": {"default_max_rows": 700}}`)), 0o600)) + findings, adopted = s.Reload() + require.True(t, adopted, "findings: %s", findingStrings(findings)) + assert.Equal(t, 700, s.DefaultMaxRows()) +} + +func TestStore_ReloadWithWarningsAdopts(t *testing.T) { + t.Parallel() + s := newLoadedStore(t, map[string]string{ + FilePolicies: `{}`, // empty policy: legal, warned (total lockout) + FilePipes: `{}`, // drop the pipe so its analyst role reference doesn't dangle + }) + findings, adopted := s.Reload() + assert.True(t, adopted, "warnings alone must not block adoption") + assert.NotEmpty(t, findings) + assert.False(t, HasErrors(findings)) +} + +func TestStore_DedupeFor_Cascade(t *testing.T) { + t.Parallel() + s := newLoadedStore(t, map[string]string{ + FileConfig: configJSON(`{"dedupe": {"require_id": true, "tables": {"clicks": {"id_field": "click_id"}, "views": {"require_id": false}}}}`), + }) + + tests := []struct { + name, table, wantID string + wantRequire bool + }{ + {name: "table overrides id_field, inherits require_id", table: "clicks", wantID: "click_id", wantRequire: true}, + {name: "table overrides require_id, inherits id_field", table: "views", wantID: "event_id", wantRequire: false}, + {name: "unlisted table gets globals", table: "other", wantID: "event_id", wantRequire: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + id, req := s.DedupeFor(tt.table) + assert.Equal(t, tt.wantID, id) + assert.Equal(t, tt.wantRequire, req) + }) + } +} + +// TestStore_OpenRejectsInvalid pins the boot contract: an invalid directory +// yields no Store at all — there is no "store without a document" state and +// no compiled defaults to fall back on. +func TestStore_OpenRejectsInvalid(t *testing.T) { + t.Parallel() + files := validFiles() + files[FileConfig] = `{}` // every key missing + s, findings := Open(writeDir(t, files), nil) + assert.Nil(t, s) + assert.True(t, HasErrors(findings)) + + s, findings = Open(filepath.Join(t.TempDir(), "nope"), nil) + assert.Nil(t, s) + assert.True(t, HasErrors(findings)) +} + +// TestStore_SurvivesVanishedDirectory pins the runtime half of the same +// contract: once adopted, the snapshot outlives its files — deleting the +// directory is just a rejected reload. +func TestStore_SurvivesVanishedDirectory(t *testing.T) { + t.Parallel() + s := newLoadedStore(t, map[string]string{ + FileConfig: configJSON(`{"query": {"default_max_rows": 42}}`), + }) + require.NoError(t, os.RemoveAll(s.Dir())) + findings, adopted := s.Reload() + assert.False(t, adopted) + assert.True(t, HasErrors(findings)) + assert.Equal(t, 42, s.DefaultMaxRows()) + id, req := s.DedupeFor("clicks") + assert.Equal(t, "event_id", id) + assert.False(t, req) +} + +// TestStore_SeedIsValid pins that the shipped starter directory passes its +// own gate: `wavehouse init-settings` must never write something +// `wavehouse validate` rejects, and the defaults are readable back. +func TestStore_SeedIsValid(t *testing.T) { + t.Parallel() + dir := filepath.Join(t.TempDir(), "settings") + require.NoError(t, WriteSeed(dir)) + s, findings := Open(dir, nil) + require.NotNil(t, s, "findings: %s", findingStrings(findings)) + assert.False(t, HasErrors(findings)) + // The one expected finding: an empty policies.json is fail-closed and + // says so. The seed ships no policy on purpose — a policy is a tenant's + // decision (dev-policy.yaml is the opt-in trial one). + assert.Len(t, findings, 1, "findings: %s", findingStrings(findings)) + assert.Contains(t, findingStrings(findings), "no policy") + id, req := s.DedupeFor("anything") + assert.Equal(t, "event_id", id) + assert.False(t, req) + assert.Equal(t, 10000, s.DefaultMaxRows()) + assert.Equal(t, 60*time.Second, s.SchemaRefreshInterval()) + assert.Equal(t, []string{"*"}, s.CORSOrigins()) + + // Non-empty directory: refused, contents untouched. + require.NoError(t, os.WriteFile(filepath.Join(dir, FileConfig), []byte(`{}`), 0o600)) + err := WriteSeed(dir) + require.Error(t, err) + assert.Contains(t, err.Error(), "not empty") + data, _ := os.ReadFile(filepath.Join(dir, FileConfig)) //nolint:gosec // G304: path is rooted in t.TempDir() + assert.Equal(t, `{}`, string(data)) +} + +func TestStore_TypedAccessors(t *testing.T) { + t.Parallel() + s := newLoadedStore(t, map[string]string{ + FileConfig: configJSON(`{"query": {"default_max_rows": 250}, "schema": {"refresh_interval": 5}, "cors": {"allowed_origins": ["https://app.example.com"]}}`), + }) + assert.Equal(t, 250, s.DefaultMaxRows()) + assert.Equal(t, 5*time.Second, s.SchemaRefreshInterval()) + assert.Equal(t, []string{"https://app.example.com"}, s.CORSOrigins()) +} diff --git a/internal/settings/validate.go b/internal/settings/validate.go index 2d18127e..f3b13c1a 100644 --- a/internal/settings/validate.go +++ b/internal/settings/validate.go @@ -283,24 +283,30 @@ func (v *validator) parsePipes(data []byte) []pipes.NamedQuery { return f.Pipes } -// checkIDField rejects an explicit id_field value that could never match a -// JSON key: empty or whitespace-only (the author wrote a value that names -// nothing) and surrounding whitespace (an exact-match lookup would silently -// miss every row). Absent (nil) is always fine — it means inherit, and the -// compiled default floors the cascade. Interior whitespace stays legal: -// "click id" is a valid JSON key. -func (v *validator) checkIDField(path string, val *string, omitHint string) { +// checkIDField rejects an id_field value that could never match a JSON key: +// empty or whitespace-only (the author wrote a value that names nothing) and +// surrounding whitespace (an exact-match lookup would silently miss every +// row). nil is the caller's concern — required at the top level, inherit in +// a table override. Interior whitespace stays legal: "click id" is a valid +// JSON key. +func (v *validator) checkIDField(path string, val *string) { if val == nil { return } switch { case strings.TrimSpace(*val) == "": - v.errorf(FileConfig, path, "must not be empty — %s", omitHint) + v.errorf(FileConfig, path, "must not be empty") case strings.TrimSpace(*val) != *val: v.errorf(FileConfig, path, "id_field %q has surrounding whitespace", *val) } } +// required reports a missing key. Every top-level tunable is required so the +// adopted snapshot never depends on a value the files don't state. +func (v *validator) required(path string) { + v.errorf(FileConfig, path, "required — run `wavehouse init-settings` for a complete starter config.json") +} + func (v *validator) parseConfig(data []byte) TenantConfig { var c TenantConfig if !v.parseFile(FileConfig, data, &c) { @@ -309,8 +315,16 @@ func (v *validator) parseConfig(data []byte) TenantConfig { // files; the document is discarded whenever any error exists. return TenantConfig{} } - if d := c.Dedupe; d != nil { - v.checkIDField("dedupe.id_field", d.IDField, "omit it to use the default") + if d := c.Dedupe; d == nil { + v.required("dedupe") + } else { + if d.IDField == nil { + v.required("dedupe.id_field") + } + if d.RequireID == nil { + v.required("dedupe.require_id") + } + v.checkIDField("dedupe.id_field", d.IDField) // Sorted iteration keeps finding order deterministic across runs. for _, table := range slices.Sorted(maps.Keys(d.Tables)) { td := d.Tables[table] @@ -321,23 +335,31 @@ func (v *validator) parseConfig(data []byte) TenantConfig { case strings.TrimSpace(table) != table: v.errorf(FileConfig, path, "table name %q has surrounding whitespace", table) } - v.checkIDField(path+".id_field", td.IDField, "omit it to inherit the global value") + v.checkIDField(path+".id_field", td.IDField) if td.IDField == nil && td.RequireID == nil { v.warnf(FileConfig, path, "override sets nothing — remove it, or set id_field or require_id") } } } - if q := c.Query; q != nil { - if q.DefaultMaxRows != nil && *q.DefaultMaxRows < 0 { - v.errorf(FileConfig, "query.default_max_rows", "must be non-negative, got %d", *q.DefaultMaxRows) - } + if q := c.Query; q == nil { + v.required("query") + } else if q.DefaultMaxRows == nil { + v.required("query.default_max_rows") + } else if *q.DefaultMaxRows < 1 { + v.errorf(FileConfig, "query.default_max_rows", "must be >= 1, got %d", *q.DefaultMaxRows) } - if s := c.Schema; s != nil { - if s.RefreshInterval != nil && *s.RefreshInterval < 1 { - v.errorf(FileConfig, "schema.refresh_interval", "must be >= 1 second, got %d", *s.RefreshInterval) - } + if s := c.Schema; s == nil { + v.required("schema") + } else if s.RefreshInterval == nil { + v.required("schema.refresh_interval") + } else if *s.RefreshInterval < 1 { + v.errorf(FileConfig, "schema.refresh_interval", "must be >= 1 second, got %d", *s.RefreshInterval) } - if co := c.CORS; co != nil { + if co := c.CORS; co == nil { + v.required("cors") + } else if co.AllowedOrigins == nil { + v.required("cors.allowed_origins") + } else { for i, origin := range co.AllowedOrigins { if strings.TrimSpace(origin) == "" { v.errorf(FileConfig, fmt.Sprintf("cors.allowed_origins[%d]", i), "origin must not be empty") diff --git a/internal/settings/validate_test.go b/internal/settings/validate_test.go index 260bc292..ae775439 100644 --- a/internal/settings/validate_test.go +++ b/internal/settings/validate_test.go @@ -1,6 +1,7 @@ package settings import ( + "encoding/json" "os" "path/filepath" "strings" @@ -26,10 +27,41 @@ func validFiles() map[string]string { FileRoles: `{"roles": ["public", "analyst", "admin"]}`, FilePolicies: `{"default_role": "public", "tables": {"clicks": {"select": {"analyst": {"max_rows": 100}}}}}`, FilePipes: `{"pipes": [{"name": "top_clicks", "sql": "SELECT 1", "allowed_roles": ["analyst"], "parameters": [{"name": "limit", "type": "number"}]}]}`, - FileConfig: `{"dedupe": {"id_field": "event_id", "tables": {"clicks": {"id_field": "click_id"}}}, "schema": {"refresh_interval": 60}}`, + FileConfig: configJSON(`{"dedupe": {"tables": {"clicks": {"id_field": "click_id"}}}}`), } } +// configJSON returns the seed config.json with patch merged over it, one +// level deep (a patched block's keys replace the seed's, the rest of the +// block is kept). Every key is required, so tests that care about one key +// build a complete document from the seed rather than repeating all of them. +func configJSON(patch string) string { + seed, err := Seed() + if err != nil { + panic(err) + } + var base, over map[string]map[string]json.RawMessage + if err := json.Unmarshal(seed[FileConfig], &base); err != nil { + panic(err) + } + if err := json.Unmarshal([]byte(patch), &over); err != nil { + panic(err) + } + for block, keys := range over { + if base[block] == nil { + base[block] = map[string]json.RawMessage{} + } + for k, v := range keys { + base[block][k] = v + } + } + out, err := json.Marshal(base) + if err != nil { + panic(err) + } + return string(out) +} + func findingStrings(findings []Finding) string { parts := make([]string, len(findings)) for i, f := range findings { @@ -61,7 +93,7 @@ func TestValidate_ValidDirectory(t *testing.T) { func TestValidate_EmptyDocuments(t *testing.T) { t.Parallel() doc, findings := parse(writeDir(t, map[string]string{ - FileRoles: `{}`, FilePolicies: `{}`, FilePipes: `{}`, FileConfig: `{}`, + FileRoles: `{}`, FilePolicies: `{}`, FilePipes: `{}`, FileConfig: configJSON(`{}`), })) // Valid — but not silent: the one finding is the no-policy lockout warning, @@ -229,7 +261,14 @@ func TestValidate_ContentRules(t *testing.T) { {"empty override table name", FileConfig, `{"dedupe": {"tables": {"": {"id_field": "x"}}}}`, "table name must not be empty"}, {"override table whitespace", FileConfig, `{"dedupe": {"tables": {" clicks": {"require_id": true}}}}`, "surrounding whitespace"}, {"empty override id_field", FileConfig, `{"dedupe": {"tables": {"clicks": {"id_field": ""}}}}`, "dedupe.tables.clicks.id_field: must not be empty"}, - {"negative max rows", FileConfig, `{"query": {"default_max_rows": -1}}`, "must be non-negative"}, + {"negative max rows", FileConfig, `{"query": {"default_max_rows": -1}}`, "must be >= 1"}, + {"zero max rows", FileConfig, `{"query": {"default_max_rows": 0}}`, "must be >= 1"}, + {"missing dedupe block", FileConfig, configJSON(`{}`)[:0] + `{"query": {"default_max_rows": 1}, "schema": {"refresh_interval": 1}, "cors": {"allowed_origins": []}}`, "dedupe: required"}, + {"missing dedupe.require_id", FileConfig, `{"dedupe": {"id_field": "event_id"}}`, "dedupe.require_id: required"}, + {"missing query.default_max_rows", FileConfig, `{"query": {}}`, "query.default_max_rows: required"}, + {"missing schema.refresh_interval", FileConfig, `{"schema": {}}`, "schema.refresh_interval: required"}, + {"missing cors.allowed_origins", FileConfig, `{"cors": {}}`, "cors.allowed_origins: required"}, + {"empty config document", FileConfig, `{}`, "cors: required"}, {"stream is boot config", FileConfig, `{"stream": {"keepalive_interval": "30s"}}`, "unknown field"}, {"zero refresh interval", FileConfig, `{"schema": {"refresh_interval": 0}}`, "must be >= 1"}, {"empty cors origin", FileConfig, `{"cors": {"allowed_origins": [" "]}}`, "origin must not be empty"}, @@ -293,7 +332,7 @@ func TestValidate_Warnings(t *testing.T) { {"admin grant is dead config", FilePolicies, `{"default_role": "public", "tables": {"clicks": {"select": {"admin": {"max_rows": 5}}}}}`, "unconditional bypass"}, {"default_role equals admin", FilePolicies, `{"default_role": "admin", "tables": {}}`, "every roleless request gets full admin"}, {"admin in pipe allowlist is redundant", FilePipes, `{"pipes": [{"name": "a", "sql": "SELECT 1", "allowed_roles": ["admin"]}]}`, "listing it is redundant"}, - {"empty dedupe override sets nothing", FileConfig, `{"dedupe": {"tables": {"clicks": {}}}}`, "override sets nothing"}, + {"empty dedupe override sets nothing", FileConfig, configJSON(`{"dedupe": {"tables": {"clicks": {}}}}`), "override sets nothing"}, {"default on required parameter", FilePipes, `{"pipes": [{"name": "a", "sql": "SELECT 1", "parameters": [{"name": "x", "required": true, "default": 5}]}]}`, "never used"}, } for _, tt := range tests { @@ -316,17 +355,17 @@ func TestValidate_Warnings(t *testing.T) { func TestValidate_MultipleFaults(t *testing.T) { t.Parallel() doc, findings := parse(writeDir(t, map[string]string{ - FileRoles: `{"roles": ["analyst", "analyst"]}`, // duplicate role - FilePolicies: `{"default_role": "ghost", "tables": {}}`, // undeclared role - FilePipes: `{"pipes": [`, // truncated JSON - FileConfig: `{"query": {"default_max_rows": -1}}`, // bounds violation + FileRoles: `{"roles": ["analyst", "analyst"]}`, // duplicate role + FilePolicies: `{"default_role": "ghost", "tables": {}}`, // undeclared role + FilePipes: `{"pipes": [`, // truncated JSON + FileConfig: configJSON(`{"query": {"default_max_rows": -1}}`), // bounds violation })) assert.Nil(t, doc) out := findingStrings(findings) assert.Contains(t, out, "duplicate role") assert.Contains(t, out, `role "ghost" is not declared`) assert.Contains(t, out, "unexpected EOF") - assert.Contains(t, out, "must be non-negative") + assert.Contains(t, out, "must be >= 1") assert.Len(t, findings, 4, "every fault reported exactly once, no noise:\n%s", out) } @@ -336,7 +375,7 @@ func TestValidate_ErrorAndWarningMix(t *testing.T) { t.Parallel() files := validFiles() files[FilePolicies] = `{"default_role": "public", "tables": {"clicks": {"select": {"admin": {}}}}}` // warning: admin grant is dead config - files[FileConfig] = `{"schema": {"refresh_interval": 0}}` // error: bounds violation + files[FileConfig] = configJSON(`{"schema": {"refresh_interval": 0}}`) // error: bounds violation doc, findings := parse(writeDir(t, files)) assert.Nil(t, doc, "one error rejects the directory even when the rest only warns") require.True(t, HasErrors(findings)) diff --git a/internal/settings/watch.go b/internal/settings/watch.go new file mode 100644 index 00000000..216d4e9d --- /dev/null +++ b/internal/settings/watch.go @@ -0,0 +1,96 @@ +package settings + +import ( + "context" + "fmt" + "path/filepath" + "time" + + "github.com/fsnotify/fsnotify" +) + +// watchDebounce coalesces filesystem event bursts into one reload. Editors +// write-then-rename, Kubernetes ConfigMap updates republish the whole `..data` +// tree, and a human saving four files produces four events — reloading on +// each would validate half-written directories and spam the log. A trailing +// debounce reloads once, after the burst goes quiet. +const watchDebounce = 250 * time.Millisecond + +// Watch blocks until ctx is done, watching the settings directory and running +// the shared reload path when its contents change. A change that fails +// validation is logged and skipped — the previous good snapshot stays — so an +// operator mid-edit degrades to a log line, never to a broken server. +// +// The directory itself is watched (not the four files): atomic-writer flows +// replace files wholesale, and Kubernetes ConfigMap mounts swap a symlink, +// both of which per-file watches lose track of. The parent directory is +// watched too, because fsnotify silently drops a watch whose directory is +// removed or renamed: with only the directory watch, a delete-and-recreate +// (or a rename-over, the atomic way to replace a whole directory) would leave +// every later edit unwatched. Parent events are filtered to the settings +// directory's own name, and the directory watch is re-added on every reload +// so it is restored once the directory exists again. +// +// The setup error is returned (directory missing, fd limits); runtime watcher +// errors are logged and the loop continues — SIGHUP and the ops reload +// endpoint remain as triggers even if the watcher degrades. +func (s *Store) Watch(ctx context.Context) error { + w, err := fsnotify.NewWatcher() + if err != nil { + return fmt.Errorf("settings watcher: %w", err) + } + defer func() { _ = w.Close() }() + dir := filepath.Clean(s.dir) + if err := w.Add(dir); err != nil { + return fmt.Errorf("settings watcher: watch %s: %w", s.dir, err) + } + // Best effort: a parent that can't be watched (e.g. "/" permissions) + // costs only the recreate case, not the watcher. + if parent := filepath.Dir(dir); parent != dir { + if err := w.Add(parent); err != nil && s.logger != nil { + s.logger.Warn("settings watcher: parent directory not watched; a deleted-and-recreated settings directory won't reload until SIGHUP or POST /v1/ops/settings/reload", "parent", parent, "error", err) + } + } + + // The timer starts disarmed; each relevant event re-arms it, so the + // reload fires watchDebounce after the *last* event of a burst. + timer := time.NewTimer(time.Hour) + timer.Stop() + defer timer.Stop() + + for { + select { + case <-ctx.Done(): + return nil + case ev, ok := <-w.Events: + if !ok { + return nil + } + // Chmod-only events can't change content and are noisy on some + // platforms; everything else (create/write/remove/rename) can. + if ev.Op&(fsnotify.Create|fsnotify.Write|fsnotify.Remove|fsnotify.Rename) == 0 { + continue + } + // Parent-directory events matter only when they are about the + // settings directory itself (a sibling changing is noise). + if name := filepath.Clean(ev.Name); filepath.Dir(name) != dir && name != dir { + continue + } + timer.Reset(watchDebounce) + case werr, ok := <-w.Errors: + if !ok { + return nil + } + if s.logger != nil { + s.logger.Error("settings watcher error", "dir", s.dir, "error", werr) + } + case <-timer.C: + // Re-arm the directory watch before reloading: after a remove or + // rename fsnotify has dropped it, and Add is a no-op while it + // still exists. Failure (directory currently absent) is expected + // mid-replace; the next parent event retries. + _ = w.Add(dir) + s.TriggerReload("watch") + } + } +} diff --git a/internal/settings/watch_test.go b/internal/settings/watch_test.go new file mode 100644 index 00000000..a9ffade9 --- /dev/null +++ b/internal/settings/watch_test.go @@ -0,0 +1,103 @@ +package settings + +import ( + "context" + "fmt" + "os" + "path/filepath" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TestStore_Watch_ReloadsOnChange pins the watcher end to end: an edit to a +// settings file lands in the snapshot without any explicit reload call. The +// debounce makes exact timing untestable, so the assertion polls. +// startWatch runs s.Watch in the background and proves the watch is live +// before returning: it writes a valid config.json with maxRows and waits for +// the watcher to adopt it. A fixed sleep could not distinguish "watch +// registered" from "write raced ahead of w.Add", and with the watch proven +// live, the caller's later mutations can't be mistaken for setup races. +func startWatch(ctx context.Context, t *testing.T, s *Store, maxRows int) <-chan error { + t.Helper() + done := make(chan error, 1) + go func() { done <- s.Watch(ctx) }() + // Keep writing until the watcher picks one up: the write and w.Add race, + // and a write that lands before the watch exists emits no event. + // EventuallyWithT rather than Eventually: the condition runs off the test + // goroutine, where a bare require.NoError would FailNow the wrong + // goroutine and stall the poll until timeout instead of failing loudly. + require.EventuallyWithT(t, func(c *assert.CollectT) { + require.NoError(c, os.WriteFile(filepath.Join(s.Dir(), FileConfig), []byte(configJSON(fmt.Sprintf(`{"query": {"default_max_rows": %d}}`, maxRows))), 0o600)) + assert.Equal(c, maxRows, s.DefaultMaxRows()) + }, 5*time.Second, 2*watchDebounce, "watcher should adopt the readiness write") + return done +} + +func TestStore_Watch_ReloadsOnChange(t *testing.T) { + t.Parallel() + s := newLoadedStore(t, map[string]string{ + FileConfig: configJSON(`{"query": {"default_max_rows": 100}}`), + }) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := startWatch(ctx, t, s, 900) + + // An invalid edit is debounced, rejected, and the snapshot survives. + require.NoError(t, os.WriteFile(filepath.Join(s.Dir(), FileConfig), []byte(`not json`), 0o600)) + time.Sleep(4 * watchDebounce) + assert.Equal(t, 900, s.DefaultMaxRows(), "invalid edit must keep the previous snapshot") + + cancel() + assert.NoError(t, <-done) +} + +// TestStore_Watch_MissingDir pins the setup contract: a nonexistent directory +// is a returned error (main.go logs it and degrades to SIGHUP + the ops +// endpoint), not a silent no-op loop. +func TestStore_Watch_MissingDir(t *testing.T) { + t.Parallel() + s := newLoadedStore(t, nil) + require.NoError(t, os.RemoveAll(s.Dir())) + err := s.Watch(context.Background()) + require.Error(t, err) + assert.Contains(t, err.Error(), "settings watcher") +} + +// TestStore_Watch_SurvivesDirectoryRecreate pins the recreate contract: +// fsnotify drops a watch whose directory is removed, so without the parent +// watch + re-add a delete-and-recreate would leave later edits unwatched. +func TestStore_Watch_SurvivesDirectoryRecreate(t *testing.T) { + t.Parallel() + s := newLoadedStore(t, map[string]string{ + FileConfig: configJSON(`{"query": {"default_max_rows": 100}}`), + }) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := startWatch(ctx, t, s, 200) + + // Remove the whole directory: a rejected reload, snapshot survives. + require.NoError(t, os.RemoveAll(s.Dir())) + time.Sleep(4 * watchDebounce) + assert.Equal(t, 200, s.DefaultMaxRows(), "vanished directory must keep the previous snapshot") + + // Recreate it with new content: the watcher must pick it up again. + files := validFiles() + files[FileConfig] = configJSON(`{"query": {"default_max_rows": 300}}`) + require.NoError(t, os.Mkdir(s.Dir(), 0o750)) + for name, content := range files { + require.NoError(t, os.WriteFile(filepath.Join(s.Dir(), name), []byte(content), 0o600)) + } + assert.Eventually(t, func() bool { return s.DefaultMaxRows() == 300 }, + 5*time.Second, 50*time.Millisecond, "recreated directory should be adopted") + + // And edits inside the recreated directory are watched again. + require.NoError(t, os.WriteFile(filepath.Join(s.Dir(), FileConfig), []byte(configJSON(`{"query": {"default_max_rows": 400}}`)), 0o600)) + assert.Eventually(t, func() bool { return s.DefaultMaxRows() == 400 }, + 5*time.Second, 50*time.Millisecond, "edits after recreate should be watched") + + cancel() + assert.NoError(t, <-done) +} diff --git a/tests/e2e/fixtures/config.yaml b/tests/e2e/fixtures/config.yaml index e5a84b3f..a8e51096 100644 --- a/tests/e2e/fixtures/config.yaml +++ b/tests/e2e/fixtures/config.yaml @@ -5,10 +5,6 @@ # the orchestrator via env vars. Everything below is pinned here so the # rig config is visible and editable without recompiling Go. -server: - cors_allowed_origins: - - "*" - mq: # 1 GiB keeps the testcontainer's NATS stream tiny; the suite never # writes anywhere near that. @@ -16,7 +12,6 @@ mq: dedupe: enabled: true - id_field: event_id # JWT validation. The suite signs test tokens with this fixed dev secret; # don't reuse outside the e2e rig. @@ -25,11 +20,6 @@ auth: role_claim: role # operator_key: "" # non-JWT full-access operator credential (Authorization: Operator, or X-Operator-Key); unset in the rig -schema: - # 5s, not the 60s default — tests asserting schema-discovery side - # effects shouldn't wait a minute for the next tick. - refresh_interval: 5 - dlq: enabled: true @@ -38,6 +28,12 @@ policy: # to the repo root. file_path: tests/e2e/fixtures/policy.yaml +# Tenant tunables (schema refresh_interval 5s so schema-discovery tests don't +# wait a minute; dedupe id_field; CORS "*") live in the settings directory — +# see settings/config.json. Same cwd-relative resolution as policy.file_path. +settings: + dir: tests/e2e/fixtures/settings + # OTel on so the export branches get coverage. The endpoint is set by the # orchestrator (scripts/orchestrator) via OTEL_EXPORTER_OTLP_ENDPOINT, default # http://127.0.0.1:4317 — there's no otel.addr key anymore. gRPC exporters init diff --git a/tests/e2e/fixtures/settings/config.json b/tests/e2e/fixtures/settings/config.json new file mode 100644 index 00000000..dcf3d2d8 --- /dev/null +++ b/tests/e2e/fixtures/settings/config.json @@ -0,0 +1,16 @@ +{ + "dedupe": { + "id_field": "event_id", + "require_id": false, + "tables": {} + }, + "query": { + "default_max_rows": 10000 + }, + "schema": { + "refresh_interval": 5 + }, + "cors": { + "allowed_origins": ["*"] + } +} diff --git a/tests/e2e/fixtures/settings/pipes.json b/tests/e2e/fixtures/settings/pipes.json new file mode 100644 index 00000000..0967ef42 --- /dev/null +++ b/tests/e2e/fixtures/settings/pipes.json @@ -0,0 +1 @@ +{} diff --git a/tests/e2e/fixtures/settings/policies.json b/tests/e2e/fixtures/settings/policies.json new file mode 100644 index 00000000..e15bdd5c --- /dev/null +++ b/tests/e2e/fixtures/settings/policies.json @@ -0,0 +1,3 @@ +{ + "default_role": "public" +} diff --git a/tests/e2e/fixtures/settings/roles.json b/tests/e2e/fixtures/settings/roles.json new file mode 100644 index 00000000..754f3e06 --- /dev/null +++ b/tests/e2e/fixtures/settings/roles.json @@ -0,0 +1,3 @@ +{ + "roles": ["public"] +} diff --git a/tests/integration/query_limits_test.go b/tests/integration/query_limits_test.go index b7b58e17..eeb41794 100644 --- a/tests/integration/query_limits_test.go +++ b/tests/integration/query_limits_test.go @@ -99,7 +99,7 @@ func TestStructuredQuery_ResourceCapsEnforcedServerSide(t *testing.T) { }, }) h := api.NewStructuredQueryHandler( - e.chConn, nil, e.registry, store, 60, 30*time.Second, 0, testutil.NopLogger(), + e.chConn, nil, e.registry, store, 60, 30*time.Second, nil, testutil.NopLogger(), ) req := httptest.NewRequest(http.MethodPost,