fix(stream): apply policy row-filter per subscriber on SSE

CHANGELOG.md

@@ -31,6 +31,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Security
 
+- **Live SSE streams now apply a role's row-`filter` per subscriber, closing a query/stream row-level-security drift** (`internal/stream/hub.go`, `internal/stream/subscriber.go`, `internal/policy/policy.go`, `internal/policy/rowfilter.go` (new), `internal/discovery/validation.go`, `internal/api/stream.go`, `cmd/wavehouse/main.go`, `docs/src/content/docs/access-control.mdx`, `docs/src/content/docs/architecture.md`, plus tests in `internal/policy/rowfilter_test.go` (new), `internal/discovery/validation_test.go`, `internal/stream/hub_test.go`): closes #319. The SSE delivery path stripped denied columns but never applied a role's row-level `filter` predicate, so a subscriber received rows the structured-query path would have filtered out for that same role — a data-exposure on the streaming surface for any table that combines a row-policy with a shared or role-scoped stream (harmless on the public Stats table today, which carries no restrictive row-policy, but real for any private/PII table fronted by a stream). The row-filter is now resolved once into predicates that feed **both** read surfaces — the query path renders them to SQL, the stream evaluates them in memory (`ResolvedPermissions.RowVisible`, evaluated per subscriber against that subscriber's claims via the same `Evaluate` call the query path uses) — so the two can't drift (the row-level analogue of the shared `IsColumnAllowed` decision from #223). Because a row-filter resolves against each subscriber's JWT claims, the #294/#353 once-per-role projection is now claims-aware: a role **without** a filter keeps the pure once-per-role fast path unchanged (zero regression on the public stream), while a role **with** a filter keeps the shared column projection but delivers it only to the subscribers whose claims admit each row (evaluated against the full event, so a filter may key on a column the role can't select). Equality and set predicates (`_eq`/`_neq`/`_in` — the tenant/user scoping row-security actually uses) are enforced exactly — numeric ties beyond `float64` precision resolve at arbitrary precision (`math/big`), so the string-encoded 64-bit IDs ingest accepts to survive JS precision loss never falsely collide, and a `NaN` operand fails closed rather than comparing equal to everything — with one representation caveat: a `Bool`/`DateTime` filter value compares against the event's canonical text form, without ClickHouse's cross-representation coercion (`true`, not `1`); ordering predicates (`_gt`/`_lt`) are schema-informed — numeric columns compare numerically (matching ClickHouse), other types lexicographically — which is best-effort for exotic types stored as numbers (a documented boundary in access-control.mdx), with every ambiguous or uncomparable value failing closed (the row is withheld). Replay (gap-fill) applies the same per-connection row check. The resource limits (`max_rows`, `max_execution_time`, …) remain a query-path property and are still **not** applied to the stream (a separate, documented boundary). Supersedes the "stream path applies no row-level filter" invariant noted in the #294/#353 Changed entry below.
 - **Policy `_in` is now enforced on both the row-`filter` and insert-`check` paths, closing a fail-open row-security gap** (`internal/policy/policy.go`, `internal/api/ingest.go`, `docs/src/content/docs/access-control.mdx`, plus tests in `internal/policy/policy_test.go`, `internal/api/ingest_test.go`): closes #224. The `Filter` schema accepted `_in` but the engine never read it: on the row-`filter`/SELECT path `resolveFilters` had no `_in` branch, so a row-security filter like `tenant_id: { _in: … }` produced **no `WHERE` predicate** and the role saw every row instead of its tenant subset (a fail-open, same family as #223); on the `check`/INSERT path only `_eq` was honored, silently dropping any other operator. `_in` now takes a single claim that resolves to a JSON **array** (the multi-tenant case — a token's `tenant_ids` list) and emits `col IN (?, …)` with one bound param per element; a scalar claim is a one-element set, and an empty/absent claim matches **no rows** (fail-closed) rather than widening to all of them. On the insert path an `_in` check requires the column be present and one of the set — there is no single value to auto-inject as `_eq` does, so an omitted column is rejected (`403 check failed`). The comparison operators are enforced on `filter` (`_eq`/`_neq`/`_gt`/`_lt`/`_in` all produce predicates now, so nothing is rejected there) and, on `check`, `_neq`/`_gt`/`_lt` become a loud config-load rejection (no insert-time semantics; `check` honors `_eq` + `_in`). The `_in` value stays a single templated string in the wire schema (Go `Filter.In`, SDK `PolicyFilter._in`), matching the established "set = array" shape of the caller-query `in` operator.
 - **`denied_aggregations` is now enforced case-insensitively against the caller-supplied function name, closing a policy bypass** (`internal/policy/policy.go`, plus tests in `internal/policy/policy_test.go`): closes #318. `IsAggregationAllowed` lower-cased the aggregation name only *after* the deny-list loop and the empty-allow-list early return, so the deny check compared a lower-cased deny entry against the raw caller input — a denied aggregation slipped past simply by changing case (`SUM` bypassed a `sum` deny entry, and with an empty allow list the call was then permitted). `isValidAggFn` already accepts any casing and the SQL builder emits the function name verbatim, so the denied aggregation actually executed. The case fold now happens once, before the deny check, so deny wins regardless of caller casing — matching the case-insensitive contract the access-control docs already specified.
 - **A role's structured-query resource caps are now enforced server-side by ClickHouse, so a public read can't outrun its budget during a server-side scan/merge/aggregation phase** (`internal/api/ch_settings.go` (new), `internal/api/structured_query.go`, `internal/policy/policy.go`, `internal/policy/scalars.go` (new), `internal/config/config.go`, `internal/query/builder.go`, `cmd/wavehouse/main.go`, `config.yaml`, `clients/ts/src/types.ts`, `docs/src/content/docs/{access-control.mdx,configuration.mdx}`, plus tests in `internal/api/ch_settings_test.go`, `internal/policy/{policy,scalars}_test.go`, `internal/config/config_test.go`, `internal/query/builder_test.go`, `tests/integration/query_limits_test.go` (all new/expanded), `tests/e2e/sdk/query.test.ts`): closes #316. The native ClickHouse connection passed **no per-query `Settings`**, so a read's policy caps bound it only client-side — a Go `context` deadline (which cancels only while the client is reading result blocks) plus a SQL `LIMIT`. Memory and rows scanned were **never bounded server-side**, so a heavy aggregation could allocate gigabytes of state or scan an entire table well within the time budget; and because `clickhouse-go` derives a `max_execution_time` setting from the context deadline only for deadlines `> 1s`, a sub-second time cap reached ClickHouse with no server-side time bound at all. The structured-query path now attaches per-query `Settings` derived from the role's resolved permissions: `max_execution_time` (fractional seconds, emitted explicitly so the sub-second case is enforced), `max_result_rows` + `result_overflow_mode=throw` (defense-in-depth behind the SQL `LIMIT`), `max_rows_to_read` + `read_overflow_mode=throw`, and `max_memory_usage` — so a query that exceeds its budget is rejected by the server (ClickHouse codes 158 / 241) rather than running to completion. **Boundary:** WaveHouse owns the *dynamic, per-role* caps (sent as per-query settings); the *global, static* backstop — which applies to every query including named pipes and raw admin SQL — is configured in **ClickHouse's own settings profiles and quotas** (documented in `configuration.mdx`), composes with the per-role caps, and holds even against a WaveHouse bug. **Schema (pre-launch):** the per-role policy fields are human-readable in / numeric out — `max_execution_time_ms` (int) → **`max_execution_time`** (set as a duration string `"5s"` or a bare ms number; read back as ms), the new **`max_memory_usage`** (set as a size string `"4GiB"` — IEC/SI respected via `dustin/go-humanize`, so `4GB` ≠ `4GiB` — or a bare byte number; read back as bytes), and the new **`max_rows_to_read`** (int), backed by two small `Millis`/`ByteSize` types in the `policy` package. The formerly hard-coded `query.DefaultMaxRows = 10000` result-LIMIT becomes the documented, tunable `query.default_max_rows` config knob (`Build` takes it as a parameter). Raw admin SQL remains unbounded by WaveHouse (governed by ClickHouse). Verified RED before the fix: the handler-level integration test confirmed a capped read returned the full result set when the settings weren't sent.

cmd/wavehouse/main.go

@@ -292,7 +292,7 @@ func run() int {
 	// handler (write counts), and the Hub that projects/serializes each event once
 	// per (topic, role) and pushes it to that role's subscribers.
 	sseMetrics := stream.NewMetrics()
-	streamHub := stream.NewHub(policyStore, sseMetrics)
+	streamHub := stream.NewHub(policyStore, registry, sseMetrics)
 
 	// Start policy watch for cluster-wide updates.
 	go policyStore.Watch(ctx)

docs/src/content/docs/access-control.mdx

@@ -370,12 +370,12 @@ The same policy drives every data path, but not every field is meaningful on eve
 | ------- | -------- | -------- |
 | Structured read | `POST /v1/query?table={table}` | table+role `select` required, then `allow`/`deny_columns`, row `filter`, aggregation rules, and the per-role `max_rows` / `max_execution_time` / `max_rows_to_read` / `max_memory_usage` caps (over the [ClickHouse server-wide limits](/configuration#server-side-resource-limits)) |
 | Ingest (write) | `POST /v1/ingest?table={table}` | table+role `insert` required, then `allow`/`deny_columns` and `check` (enforced and auto-injected) |
-| Live stream | `GET /v1/stream` | table+role `select` required (a table the role can't read is skipped), then denied columns are masked from each event |
+| Live stream | `GET /v1/stream` | table+role `select` required (a table the role can't read is skipped), then denied columns are masked from each event and the role's row `filter` is applied per subscriber against their JWT claims (see caution below) |
 | Raw SQL | `POST /v1/admin/query` | `admin_role` only — no per-statement policy; the role gate is the entire authorization story |
 | Named pipe | `GET/POST /v1/pipes/{name}` | per-pipe `allowed_roles` (not the policy engine; see [Named Pipes](/pipes)). Resource limits come from ClickHouse's [server-wide settings](/configuration#server-side-resource-limits), not per-role policy caps |
 
-:::caution[Live streams enforce access, not row filters]
-SSE subscribers are checked for table-level `select` permission and have denied columns stripped from each event, but the row-level `filter` predicates and the resource limits (`max_rows`, `max_execution_time`, `max_rows_to_read`, `max_memory_usage`) are a property of the SQL query path and are **not** applied to the live event stream. If a role must never observe another tenant's rows in real time, don't grant it stream access to a shared table — scope the data at the table level.
+:::caution[Live streams enforce column and row policy, but not resource limits]
+SSE subscribers are checked for table-level `select` permission, have denied columns stripped from each event, and — like the query path — receive only the rows their role's row-level `filter` predicates admit, evaluated per subscriber against their JWT claims. Equality and set predicates (`_eq`, `_neq`, `_in`) — the usual tenant/user scoping — are enforced exactly: numeric ties resolve at full precision (an ID beyond `float64`'s 2^53 never falsely matches a neighbor) and an unparseable or `NaN` operand withholds the row. One representation caveat: a `Bool` (or `DateTime`) value compares by its canonical text form, without ClickHouse's cross-representation coercion — write the filter value the way your events carry it (`true`, not `1`). Ordering predicates (`_gt`, `_lt`) are best-effort: the stream compares numeric columns numerically (matching ClickHouse) but lacks ClickHouse's full per-type coercion, so an ordering filter on an exotic type (a `Decimal`/`Int128` beyond `float64` precision, or a `DateTime` ingested as a Unix number) can admit or withhold a row the query path wouldn't. The resource limits (`max_rows`, `max_execution_time`, `max_rows_to_read`, `max_memory_usage`) remain a property of the SQL query path and are **not** applied to the live event stream; if those caps are part of a role's isolation story, don't rely on them over the stream.
 :::
 
 ## Managing the policy

docs/src/content/docs/architecture.md

@@ -85,7 +85,7 @@ The API layer uses [Chi](https://github.com/go-chi/chi) for routing with Request
 
 The SSE fan-out, factored out of `api/` so the delivery hot path ([#294](https://github.com/Wave-RF/WaveHouse/issues/294)) lives next to the keepalive primitives it shares. One abstraction per file.
 
-- **hub.go** — `Hub`, the event fan-out. Subscribers register under `(topic, role)`; `Broadcast` decodes each event once, applies each subscribed role's column policy once, builds one SSE frame per role, and fans it to every member of that role's `Bucket` — collapsing the prior per-subscriber `unmarshal → evaluate → filter → marshal` into one pass per distinct `(role, table)` output shape (the [#294](https://github.com/Wave-RF/WaveHouse/issues/294) lever; the measured ceiling was ~2 270 deliveries/s from re-projecting per subscriber). The `(topic, role)` key is sufficient because column visibility derives only from the role+table policy entry, never from JWT claims (claims feed only the row-level `WHERE`/`CHECK`, which the stream path does not apply). `ReplayFrame` shares the same projection for the handler's per-connection gap-fill.
+- **hub.go** — `Hub`, the event fan-out. Subscribers register under `(topic, role)`; `Broadcast` decodes each event once, applies each subscribed role's column policy once, builds one SSE frame per role, and fans it to every member of that role's `Bucket` — collapsing the prior per-subscriber `unmarshal → evaluate → filter → marshal` into one pass per distinct `(role, table)` output shape (the [#294](https://github.com/Wave-RF/WaveHouse/issues/294) lever; the measured ceiling was ~2 270 deliveries/s from re-projecting per subscriber). The column projection is claims-independent, so it is shared across a role's whole bucket; the role's row-level `filter` predicate is not — it is resolved against each subscriber's JWT claims, so for a role that carries a filter `Broadcast` keeps the shared column projection but delivers it only to the subscribers whose claims admit each row (`ResolvedPermissions.RowVisible`, evaluated against the full event via the numeric-aware comparison seeded from the schema registry). This is the [#319](https://github.com/Wave-RF/WaveHouse/issues/319) fix that closes the query/stream row-level-security drift; roles without a filter keep the pure once-per-role fast path. `ReplayFrame` shares the same projection and per-connection row check for the handler's gap-fill.
 - **subscriber.go** — `Subscriber`, the per-connection handle. It owns a single ready-to-write outbound queue of `Frame`s (each tagged with its `kind`, so the handler labels the write where it happens): producers — the keepalive wheel and the event `Hub` — fan frames in with `Send` (non-blocking; a full queue drops and the `Hub` counts it), and the handler drains `Frames()` to the client verbatim. The queue is sized for buffering live events (cap 64, up from the keepalive-only cap 1; #152 will make it a knob), and an `Evicted()` channel is the seam the slow-consumer follow-up closes to disconnect a wedged consumer.
 - **bucket.go** — `Bucket`, the reusable fan-out primitive: a concurrency-safe set of subscribers. `Push` delivers a shared `Frame` to each fire-and-forget (the keepalive wheel's ring); `Snapshot` exposes the members so the event `Hub` can fan out while inspecting each `Send` result (to count drops). The `Hub` holds one `Bucket` per `(topic, role)` so a projected frame is built once and sent to every member instead of re-projected per subscriber.
 - **heartbeat.go** — The keepalive wheel (`Heartbeater`). A single process-wide ticker fans a minimal `:` comment across the ring of `Bucket`s, waking ~1/N of live streams per tick so the writes don't synchronize. The effective per-connection keepalive period is `stream.keepalive_interval` (the wheel ticks every `keepalive_interval ÷ keepalive_buckets`, so one rotation spans the interval); the owning handler goroutine does the actual write, so the shared ticker never touches a `ResponseWriter` directly.

go.mod

@@ -1,6 +1,6 @@
 module github.com/Wave-RF/WaveHouse
 
-go 1.26.4
+go 1.26.5
 
 tool (
 	github.com/Zxilly/go-size-analyzer/cmd/gsa

internal/api/errors_test.go

@@ -194,7 +194,7 @@ func TestAuthzDenied_LogsChiRoutePattern(t *testing.T) {
 	router := NewRouter(Dependencies{
 		Ingest:      NewIngestHandler(reg, &testutil.MockPublisher{}, logger),
 		Query:       &QueryHandler{},
-		SSE:         NewStreamHandler(stream.NewHub(nil, nil), nil),
+		SSE:         NewStreamHandler(stream.NewHub(nil, nil, nil), nil),
 		Health:      &HealthHandler{},
 		Schema:      NewSchemaHandler(reg),
 		AuthMW:      func(next http.Handler) http.Handler { return next },

internal/api/router_test.go

@@ -283,7 +283,7 @@ func TestNewRouter_RoutesRegistered(t *testing.T) {
 		{Name: "events", Columns: []discovery.Column{{Name: "id", Type: "String"}}},
 	})
 	pub := &testutil.MockPublisher{}
-	hub := stream.NewHub(nil, nil)
+	hub := stream.NewHub(nil, nil, nil)
 
 	deps := Dependencies{
 		Ingest:  NewIngestHandler(reg, pub, testutil.NopLogger()),
@@ -339,7 +339,7 @@ func TestNewRouter_RoutesRegistered(t *testing.T) {
 func TestNewRouter_CORSOnStream(t *testing.T) {
 	t.Parallel()
 
-	hub := stream.NewHub(nil, nil)
+	hub := stream.NewHub(nil, nil, nil)
 	router := NewRouter(Dependencies{
 		SSE:         NewStreamHandler(hub, nil),
 		Health:      &HealthHandler{},
@@ -411,7 +411,7 @@ func TestNewRouter_RawSQLAdminGate(t *testing.T) {
 
 	reg := discovery.NewSchemaRegistryFromMap(nil)
 	pub := &testutil.MockPublisher{}
-	hub := stream.NewHub(nil, nil)
+	hub := stream.NewHub(nil, nil, nil)
 
 	router := NewRouter(Dependencies{
 		Ingest:      NewIngestHandler(reg, pub, testutil.NopLogger()),
@@ -472,7 +472,7 @@ func TestNewRouter_OptionalDepsNil(t *testing.T) {
 
 	reg := discovery.NewSchemaRegistryFromMap(nil)
 	pub := &testutil.MockPublisher{}
-	hub := stream.NewHub(nil, nil)
+	hub := stream.NewHub(nil, nil, nil)
 
 	deps := Dependencies{
 		Ingest:      NewIngestHandler(reg, pub, testutil.NopLogger()),
@@ -523,7 +523,7 @@ func TestNewRouter_NotFoundEmitsJSON(t *testing.T) {
 
 	reg := discovery.NewSchemaRegistryFromMap(nil)
 	pub := &testutil.MockPublisher{}
-	hub := stream.NewHub(nil, nil)
+	hub := stream.NewHub(nil, nil, nil)
 	deps := Dependencies{
 		Ingest: NewIngestHandler(reg, pub, testutil.NopLogger()),
 		Query:  &QueryHandler{},
@@ -548,7 +548,7 @@ func TestNewRouter_MethodNotAllowedEmitsJSON(t *testing.T) {
 
 	reg := discovery.NewSchemaRegistryFromMap(nil)
 	pub := &testutil.MockPublisher{}
-	hub := stream.NewHub(nil, nil)
+	hub := stream.NewHub(nil, nil, nil)
 	deps := Dependencies{
 		Ingest: NewIngestHandler(reg, pub, testutil.NopLogger()),
 		Query:  &QueryHandler{},
@@ -645,7 +645,7 @@ func TestNewRouter_SchemaAdminOnly(t *testing.T) {
 	t.Parallel()
 	reg := discovery.NewSchemaRegistryFromMap(nil)
 	pub := &testutil.MockPublisher{}
-	hub := stream.NewHub(nil, nil)
+	hub := stream.NewHub(nil, nil, nil)
 
 	router := NewRouter(Dependencies{
 		Ingest:      NewIngestHandler(reg, pub, testutil.NopLogger()),