fix(stream): apply policy row-filter per subscriber on SSE - #381
fix(stream): apply policy row-filter per subscriber on SSE#381taitelee wants to merge 9 commits into
Conversation
github-actions
Bot
added
documentation
📝 WalkthroughSummary by CodeRabbit
WalkthroughThis PR makes SSE live-stream delivery apply row-level policy filters per subscriber, using claims-aware policy resolution, schema-aware numeric comparisons, and replay-path filtering. It also updates wiring, tests, and documentation to match the new streaming behavior. ChangesStreaming row-level security enforcement
Estimated code review effort: 4 (Complex) | ~60 minutes Sequence Diagram(s)sequenceDiagram
participant StreamHandler
participant Subscriber
participant Hub
participant Policy
StreamHandler->>Subscriber: SetClaims(claims)
StreamHandler->>Hub: register subscriber
Hub->>Policy: Evaluate(role, claims)
Policy-->>Hub: resolved rowFilter
Hub->>Hub: projectColumns(role, event)
Hub->>Hub: RowVisible(row, numericCols)
Hub-->>Subscriber: deliver frame if visible
Possibly related issues
Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 3 | ❌ 2❌ Failed checks (2 warnings)
✅ Passed checks (3 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
✨ Simplify code
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
|
📚 Docs preview is live → https://c89f5022-wavehouse-docs.wave-rf.workers.dev |
Code Coverage OverviewLanguages: Go GoThe overall coverage in the Show a code coverage summary of the most impacted files.
Updated |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
docs/src/content/docs/access-control.mdx (1)
373-373: 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick winUpdate the Live stream row in the enforcement table to include row-level filtering.
The "Where each rule is enforced" table at line 373 still reads "denied columns are masked from each event" with no mention of row-level
filter, but this PR adds exactly that enforcement. The Structured read row (line 371) explicitly lists "row filter", so the table is now inconsistent with both the caution block below (lines 377–378) and the actual code (hub.goBroadcastcallsRowVisibleper subscriber). Per the docs code↔docs sync guideline, the table should reflect the changed behavior.📝 Proposed fix to update the table row
-| 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 row-level `filter` predicates are evaluated per subscriber against their JWT claims (see caution below) |Source: Coding guidelines
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: 2efc09f7-0227-4f93-844b-1b18b486200f
📒 Files selected for processing (17)
CHANGELOG.mdcmd/wavehouse/main.godocs/src/content/docs/access-control.mdxdocs/src/content/docs/architecture.mdinternal/api/errors_test.gointernal/api/router_test.gointernal/api/stream.gointernal/api/stream_test.gointernal/discovery/validation.gointernal/discovery/validation_test.gointernal/policy/policy.gointernal/policy/rowfilter.gointernal/policy/rowfilter_test.gointernal/stream/hub.gointernal/stream/hub_test.gointernal/stream/subscriber.gotests/integration/setup_test.go
📜 Review details
⏰ Context from checks skipped due to timeout. (5)
- GitHub Check: E2E tests
- GitHub Check: Docs build
- GitHub Check: Coverage
- GitHub Check: Lint
- GitHub Check: Analyze (go)
⚠️ CI failures not shown inline (2)
GitHub Actions: PR housekeeping / 0_PR housekeeping.txt: fix(stream): apply policy row-filter per subscriber on SSE
Conclusion: failure
##[group]Run # Single source of truth for the rule: scripts/lint-pr-title.sh — the
�[36;1m# Single source of truth for the rule: scripts/lint-pr-title.sh — the�[0m
�[36;1m# SAME validator the local agent gate runs (.claude/hooks/agent-bash-gate.sh),�[0m
�[36;1m# so CI and local can't drift. The checkout above is ref: main, so this is�[0m
�[36;1m# always the default-branch script. Dependabot's grouped-update titles�[0m
�[36;1m# routinely exceed the 72-char subject cap and the format isn't�[0m
�[36;1m# configurable, so Dependabot PRs are exempt from the length check�[0m
�[36;1m# (the format check still applies).�[0m
�[36;1mif [[ "$PR_AUTHOR" == "dependabot[bot]" || "$PR_AUTHOR" == "app/dependabot" ]]; then�[0m
�[36;1m export PR_TITLE_SKIP_LENGTH=1�[0m
�[36;1mfi�[0m
�[36;1m�[0m
�[36;1mif reason=$(bash scripts/lint-pr-title.sh "$PR_TITLE" 2>&1); then�[0m
�[36;1m echo "passed=true" >> "$GITHUB_OUTPUT"�[0m
�[36;1m echo "PR title OK: $PR_TITLE"�[0m
�[36;1melse�[0m
�[36;1m echo "passed=false" >> "$GITHUB_OUTPUT"�[0m
�[36;1m printf '%s\n' "$reason"�[0m
�[36;1m echo "::error::$(printf '%s' "$reason" | head -1)"�[0m
GitHub Actions: PR housekeeping / PR housekeeping: fix(stream): apply policy row-filter per subscriber on SSE
Conclusion: failure
##[group]Run # Single source of truth for the rule: scripts/lint-pr-title.sh — the
�[36;1m# Single source of truth for the rule: scripts/lint-pr-title.sh — the�[0m
�[36;1m# SAME validator the local agent gate runs (.claude/hooks/agent-bash-gate.sh),�[0m
�[36;1m# so CI and local can't drift. The checkout above is ref: main, so this is�[0m
�[36;1m# always the default-branch script. Dependabot's grouped-update titles�[0m
�[36;1m# routinely exceed the 72-char subject cap and the format isn't�[0m
�[36;1m# configurable, so Dependabot PRs are exempt from the length check�[0m
�[36;1m# (the format check still applies).�[0m
�[36;1mif [[ "$PR_AUTHOR" == "dependabot[bot]" || "$PR_AUTHOR" == "app/dependabot" ]]; then�[0m
�[36;1m export PR_TITLE_SKIP_LENGTH=1�[0m
�[36;1mfi�[0m
�[36;1m�[0m
�[36;1mif reason=$(bash scripts/lint-pr-title.sh "$PR_TITLE" 2>&1); then�[0m
�[36;1m echo "passed=true" >> "$GITHUB_OUTPUT"�[0m
�[36;1m echo "PR title OK: $PR_TITLE"�[0m
�[36;1melse�[0m
�[36;1m echo "passed=false" >> "$GITHUB_OUTPUT"�[0m
�[36;1m printf '%s\n' "$reason"�[0m
�[36;1m echo "::error::$(printf '%s' "$reason" | head -1)"�[0m
🧰 Additional context used
📓 Path-based instructions (2)
**/*_test.go
📄 CodeRabbit inference engine (AGENTS.md)
**/*_test.go: Write tests in table-driven form witht.Run(tt.name, ...)for multiple cases.
Use shared mocks frominternal/testutil/instead of ad-hoc mocks in tests.
Use the repo’s JWT, schema, policy, pipes, and JSON response test helpers (testutil.MakeJWT,testutil.MakeExpiredJWT,NewTestSchemaRegistry,policy.NewMemoryStore,pipes.NewMemoryStore,AssertJSONResponse,AssertJSONContains) where applicable.
Every new function should have corresponding test cases, and new code should aim for 80%+ coverage.
Files:
internal/api/errors_test.gotests/integration/setup_test.gointernal/api/stream_test.gointernal/api/router_test.gointernal/discovery/validation_test.gointernal/policy/rowfilter_test.gointernal/stream/hub_test.go
docs/src/content/docs/**/*.{md,mdx}
📄 CodeRabbit inference engine (AGENTS.md)
Documentation prose under the Starlight docs site must stay accurate against code, include runnable examples where relevant, and reflect code↔docs sync for changed behavior.
Files:
docs/src/content/docs/access-control.mdxdocs/src/content/docs/architecture.md
🧠 Learnings (5)
📚 Learning: 2026-05-20T01:02:00.784Z
Learnt from: EricAndrechek
Repo: Wave-RF/WaveHouse PR: 164
File: internal/api/router_test.go:289-350
Timestamp: 2026-05-20T01:02:00.784Z
Learning: In WaveHouse’s internal API tests (files matching internal/api/**/*_test.go), follow the existing separation-of-concerns convention for testing the RequireRole middleware: inject `ContextKeyRole` directly into the request `context.Context` instead of using `testutil.MakeJWT`/JWT-driven flows. Do not refactor role-gate tests to use JWT tokens—JWT parsing and token handling are covered separately in `middleware_test.go` (the dedicated JWT parsing tests), and mixing those concerns would expand the failure surface and reduce isolation.
Applied to files:
internal/api/errors_test.gointernal/api/stream_test.gointernal/api/router_test.go
📚 Learning: 2026-05-23T01:23:59.268Z
Learnt from: EricAndrechek
Repo: Wave-RF/WaveHouse PR: 174
File: internal/api/ingest_test.go:111-111
Timestamp: 2026-05-23T01:23:59.268Z
Learning: In WaveHouse Go tests in internal/api/**/*_test.go, use internal/testutil.AssertJSONErrorResponse(t, w) for HTTP error-path JSON assertions. Do not use (or reintroduce) package-local assertJSONErrorResponse helpers. AssertJSONErrorResponse verifies the response Content-Type is application/json, includes the X-Content-Type-Options: nosniff header, and that the JSON body contains an "error" field.
Applied to files:
internal/api/errors_test.gointernal/api/stream_test.gointernal/api/router_test.go
📚 Learning: 2026-06-26T12:23:22.696Z
Learnt from: EricAndrechek
Repo: Wave-RF/WaveHouse PR: 346
File: internal/stream/subscriber_test.go:9-28
Timestamp: 2026-06-26T12:23:22.696Z
Learning: In this Go repository, prefer table-driven tests (e.g., `[]struct{...}` with `t.Run(...)`) only for tests that cover multiple scenarios/inputs and can be cleanly enumerated. Do not artificially rewrite a clear single-scenario sequential behavioral-flow test into a table-driven form just to fit the pattern; if there’s only one meaningful scenario, keep the test as a straightforward linear flow (as in `TestSubscriber_SendDeliversThenDropsWhenFull`).
Applied to files:
internal/api/errors_test.gotests/integration/setup_test.gointernal/api/stream_test.gointernal/api/router_test.gointernal/discovery/validation_test.gointernal/policy/rowfilter_test.gointernal/stream/hub_test.go
📚 Learning: 2026-07-07T12:38:12.052Z
Learnt from: EricAndrechek
Repo: Wave-RF/WaveHouse PR: 378
File: internal/auth/auth.go:119-132
Timestamp: 2026-07-07T12:38:12.052Z
Learning: In this repo, do not add or recommend logging/tracing client IP addresses using naive or untrusted sources (e.g., `r.RemoteAddr` or directly trusting/deriving `X-Forwarded-For`) anywhere in the Go codebase. `middleware.RealIP` was removed due to IP-spoofing risks, and proper trusted-proxy-aware client-IP handling is intentionally deferred to issue `#333`. During code review, if proposed changes would record client IPs (including in audit paths such as `internal/auth/auth.go`), reject/redirect until `#333` lands with correct trusted-proxy configuration and safeguards.
Applied to files:
internal/api/errors_test.gocmd/wavehouse/main.gointernal/discovery/validation.gotests/integration/setup_test.gointernal/api/stream.gointernal/api/stream_test.gointernal/api/router_test.gointernal/discovery/validation_test.gointernal/policy/policy.gointernal/policy/rowfilter.gointernal/stream/subscriber.gointernal/policy/rowfilter_test.gointernal/stream/hub.gointernal/stream/hub_test.go
📚 Learning: 2026-06-10T15:01:09.027Z
Learnt from: EricAndrechek
Repo: Wave-RF/WaveHouse PR: 312
File: docs/src/content/docs/development.md:0-0
Timestamp: 2026-06-10T15:01:09.027Z
Learning: In this repo’s Markdown review (all .md files), do not flag capitalization/style issues for literal paths starting with ".github/" (or any substring that is a path beginning with ".github/"). Treat ".github" as the correct lowercase dotfile directory name, even when it appears inside prose or code spans; automated checks such as LanguageTool’s "(GITHUB)" rule commonly produce false positives for this literal filesystem path.
Applied to files:
docs/src/content/docs/architecture.mdCHANGELOG.md
🔇 Additional comments (20)
CHANGELOG.md (1)
34-34: LGTM!docs/src/content/docs/access-control.mdx (1)
377-378: LGTM!docs/src/content/docs/architecture.md (1)
88-88: LGTM!cmd/wavehouse/main.go (1)
295-295: LGTM!tests/integration/setup_test.go (1)
315-315: LGTM!internal/api/errors_test.go (1)
197-197: LGTM!internal/api/router_test.go (1)
286-286: LGTM!Also applies to: 342-342, 414-414, 475-475, 526-526, 551-551, 648-648
internal/api/stream_test.go (1)
25-25: LGTM!Also applies to: 53-53, 78-78, 117-117
internal/stream/hub_test.go (2)
41-51: LGTM!
68-68: LGTM!Also applies to: 100-100, 136-136, 288-288, 331-331, 352-352, 373-373, 390-390, 411-411, 440-440
internal/policy/policy.go (1)
64-82: LGTM!Also applies to: 197-208, 228-313
internal/policy/rowfilter.go (2)
13-15: LGTM!
17-103: 🔒 Security & PrivacyLexicographic fallback is intentional
NewHubis already wired with a schema registry incmd/wavehouse/main.go, andnumericColsdeliberately falls back to string comparison when schema information is unavailable. That makes>/<best-effort by design, so this isn’t a security bug; the doc comment is the part that should be softened.> Likely an incorrect or invalid review comment.internal/policy/rowfilter_test.go (1)
1-108: LGTM!internal/discovery/validation.go (1)
51-69: LGTM!Also applies to: 152-158
internal/discovery/validation_test.go (1)
227-255: LGTM!internal/api/stream.go (1)
45-51: LGTM!Also applies to: 85-85, 103-103
internal/stream/hub.go (2)
25-31: LGTM!Also applies to: 43-45
185-198: LGTM!Also applies to: 218-233, 245-277
internal/stream/subscriber.go (1)
26-31: LGTM!Also applies to: 51-55
github-project-automation
Bot
moved this from In progress
to In review
in WaveHouse Task Board
github-actions
Bot
added
the
dependencies
There was a problem hiding this comment.
Actionable comments posted: 1
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: c2faf18f-1e6f-44b3-92f4-aa3152c98c0c
📒 Files selected for processing (6)
go.modinternal/policy/policy.gointernal/policy/rowfilter.gointernal/policy/rowfilter_test.gointernal/stream/hub.gointernal/stream/hub_test.go
📜 Review details
⏰ Context from checks skipped due to timeout. (7)
- GitHub Check: Integration tests
- GitHub Check: E2E tests
- GitHub Check: Docs build
- GitHub Check: Unit tests
- GitHub Check: Coverage
- GitHub Check: Lint
- GitHub Check: Analyze (go)
⚠️ CI failures not shown inline (2)
GitHub Actions: PR housekeeping / PR housekeeping: fix(stream): apply policy row-filter per subscriber on SSE
Conclusion: failure
##[group]Run # Single source of truth for the rule: scripts/lint-pr-title.sh — the
�[36;1m# Single source of truth for the rule: scripts/lint-pr-title.sh — the�[0m
�[36;1m# SAME validator the local agent gate runs (.claude/hooks/agent-bash-gate.sh),�[0m
�[36;1m# so CI and local can't drift. The checkout above is ref: main, so this is�[0m
�[36;1m# always the default-branch script. Dependabot's grouped-update titles�[0m
�[36;1m# routinely exceed the 72-char subject cap and the format isn't�[0m
�[36;1m# configurable, so Dependabot PRs are exempt from the length check�[0m
�[36;1m# (the format check still applies).�[0m
�[36;1mif [[ "$PR_AUTHOR" == "dependabot[bot]" || "$PR_AUTHOR" == "app/dependabot" ]]; then�[0m
�[36;1m export PR_TITLE_SKIP_LENGTH=1�[0m
�[36;1mfi�[0m
�[36;1m�[0m
�[36;1mif reason=$(bash scripts/lint-pr-title.sh "$PR_TITLE" 2>&1); then�[0m
�[36;1m echo "passed=true" >> "$GITHUB_OUTPUT"�[0m
�[36;1m echo "PR title OK: $PR_TITLE"�[0m
�[36;1melse�[0m
�[36;1m echo "passed=false" >> "$GITHUB_OUTPUT"�[0m
�[36;1m printf '%s\n' "$reason"�[0m
�[36;1m echo "::error::$(printf '%s' "$reason" | head -1)"�[0m
GitHub Actions: PR housekeeping / 0_PR housekeeping.txt: fix(stream): apply policy row-filter per subscriber on SSE
Conclusion: failure
##[group]Run # Single source of truth for the rule: scripts/lint-pr-title.sh — the
�[36;1m# Single source of truth for the rule: scripts/lint-pr-title.sh — the�[0m
�[36;1m# SAME validator the local agent gate runs (.claude/hooks/agent-bash-gate.sh),�[0m
�[36;1m# so CI and local can't drift. The checkout above is ref: main, so this is�[0m
�[36;1m# always the default-branch script. Dependabot's grouped-update titles�[0m
�[36;1m# routinely exceed the 72-char subject cap and the format isn't�[0m
�[36;1m# configurable, so Dependabot PRs are exempt from the length check�[0m
�[36;1m# (the format check still applies).�[0m
�[36;1mif [[ "$PR_AUTHOR" == "dependabot[bot]" || "$PR_AUTHOR" == "app/dependabot" ]]; then�[0m
�[36;1m export PR_TITLE_SKIP_LENGTH=1�[0m
�[36;1mfi�[0m
�[36;1m�[0m
�[36;1mif reason=$(bash scripts/lint-pr-title.sh "$PR_TITLE" 2>&1); then�[0m
�[36;1m echo "passed=true" >> "$GITHUB_OUTPUT"�[0m
�[36;1m echo "PR title OK: $PR_TITLE"�[0m
�[36;1melse�[0m
�[36;1m echo "passed=false" >> "$GITHUB_OUTPUT"�[0m
�[36;1m printf '%s\n' "$reason"�[0m
�[36;1m echo "::error::$(printf '%s' "$reason" | head -1)"�[0m
🧰 Additional context used
📓 Path-based instructions (1)
**/*_test.go
📄 CodeRabbit inference engine (AGENTS.md)
**/*_test.go: Write tests in table-driven form witht.Run(tt.name, ...)for multiple cases.
Use shared mocks frominternal/testutil/instead of ad-hoc mocks in tests.
Use the repo’s JWT, schema, policy, pipes, and JSON response test helpers (testutil.MakeJWT,testutil.MakeExpiredJWT,NewTestSchemaRegistry,policy.NewMemoryStore,pipes.NewMemoryStore,AssertJSONResponse,AssertJSONContains) where applicable.
Every new function should have corresponding test cases, and new code should aim for 80%+ coverage.
Files:
internal/policy/rowfilter_test.gointernal/stream/hub_test.go
🧠 Learnings (2)
📚 Learning: 2026-07-07T12:38:12.052Z
Learnt from: EricAndrechek
Repo: Wave-RF/WaveHouse PR: 378
File: internal/auth/auth.go:119-132
Timestamp: 2026-07-07T12:38:12.052Z
Learning: In this repo, do not add or recommend logging/tracing client IP addresses using naive or untrusted sources (e.g., `r.RemoteAddr` or directly trusting/deriving `X-Forwarded-For`) anywhere in the Go codebase. `middleware.RealIP` was removed due to IP-spoofing risks, and proper trusted-proxy-aware client-IP handling is intentionally deferred to issue `#333`. During code review, if proposed changes would record client IPs (including in audit paths such as `internal/auth/auth.go`), reject/redirect until `#333` lands with correct trusted-proxy configuration and safeguards.
Applied to files:
internal/stream/hub.gointernal/policy/rowfilter.gointernal/policy/policy.gointernal/policy/rowfilter_test.gointernal/stream/hub_test.go
📚 Learning: 2026-06-26T12:23:22.696Z
Learnt from: EricAndrechek
Repo: Wave-RF/WaveHouse PR: 346
File: internal/stream/subscriber_test.go:9-28
Timestamp: 2026-06-26T12:23:22.696Z
Learning: In this Go repository, prefer table-driven tests (e.g., `[]struct{...}` with `t.Run(...)`) only for tests that cover multiple scenarios/inputs and can be cleanly enumerated. Do not artificially rewrite a clear single-scenario sequential behavioral-flow test into a table-driven form just to fit the pattern; if there’s only one meaningful scenario, keep the test as a straightforward linear flow (as in `TestSubscriber_SendDeliversThenDropsWhenFull`).
Applied to files:
internal/policy/rowfilter_test.gointernal/stream/hub_test.go
🔇 Additional comments (8)
internal/policy/policy.go (2)
111-175: LGTM!Also applies to: 177-233
235-260: LGTM!Also applies to: 262-461
internal/policy/rowfilter.go (1)
78-139: LGTM!internal/policy/rowfilter_test.go (1)
111-167: LGTM!Also applies to: 169-232
internal/stream/hub.go (2)
142-171: LGTM!Also applies to: 221-236
172-173: 🩺 Stability & AvailabilityNo race here —
Subscriber.claimsis set once viaSetClaimsbeforeHub.Addand documented as read-only for the subscriber’s lifetime, soBroadcast’s read does not have a concurrent writer.> Likely an incorrect or invalid review comment.internal/stream/hub_test.go (1)
22-29: LGTM!Also applies to: 188-214, 440-464, 519-526, 530-545
go.mod (1)
3-3: LGTM!
…ties at full precision
EricAndrechek
added
the
area/streaming
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
internal/stream/hub_test.go (1)
11-11: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winUse the shared schema test helper.
This test constructs the registry with
discovery.NewSchemaRegistryFromMapdirectly. Replace it withinternal/testutil.NewTestSchemaRegistryto keep schema setup consistent with the repository’s test infrastructure. (github.com)As per coding guidelines, use the repo’s shared schema test helper (
NewTestSchemaRegistry) where applicable.Also applies to: 254-284
Source: Coding guidelines
tests/e2e/sdk/streaming.test.ts (1)
133-182: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winMake the cross-country absence checks deterministic.
The test waits for each matching event, then immediately checks the other subscriber’s array. A leaked event can still be in flight when
some(...)runs, so the test can pass before observing the leak. Insert a unique per-country barrier event after the two target events, wait for both subscribers to receive their barriers, then assert that each stream lacks the other country’s target ID. (github.com)
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: 88138154-07d6-437e-8dfc-6c2b5b2e0acc
📒 Files selected for processing (8)
CHANGELOG.mddocs/src/content/docs/architecture.mdinternal/policy/policy.gointernal/policy/rowfilter.gointernal/policy/rowfilter_test.gointernal/stream/hub.gointernal/stream/hub_test.gotests/e2e/sdk/streaming.test.ts
💤 Files with no reviewable changes (1)
- internal/policy/rowfilter_test.go
📜 Review details
🧰 Additional context used
📓 Path-based instructions (2)
docs/src/content/docs/**/*.{md,mdx}
📄 CodeRabbit inference engine (AGENTS.md)
Documentation prose under the Starlight docs site must stay accurate against code, include runnable examples where relevant, and reflect code↔docs sync for changed behavior.
Files:
docs/src/content/docs/architecture.md
**/*_test.go
📄 CodeRabbit inference engine (AGENTS.md)
**/*_test.go: Write tests in table-driven form witht.Run(tt.name, ...)for multiple cases.
Use shared mocks frominternal/testutil/instead of ad-hoc mocks in tests.
Use the repo’s JWT, schema, policy, pipes, and JSON response test helpers (testutil.MakeJWT,testutil.MakeExpiredJWT,NewTestSchemaRegistry,policy.NewMemoryStore,pipes.NewMemoryStore,AssertJSONResponse,AssertJSONContains) where applicable.
Every new function should have corresponding test cases, and new code should aim for 80%+ coverage.
Files:
internal/stream/hub_test.go
🧠 Learnings (3)
📚 Learning: 2026-06-10T15:01:09.027Z
Learnt from: EricAndrechek
Repo: Wave-RF/WaveHouse PR: 312
File: docs/src/content/docs/development.md:0-0
Timestamp: 2026-06-10T15:01:09.027Z
Learning: In this repo’s Markdown review (all .md files), do not flag capitalization/style issues for literal paths starting with ".github/" (or any substring that is a path beginning with ".github/"). Treat ".github" as the correct lowercase dotfile directory name, even when it appears inside prose or code spans; automated checks such as LanguageTool’s "(GITHUB)" rule commonly produce false positives for this literal filesystem path.
Applied to files:
CHANGELOG.mddocs/src/content/docs/architecture.md
📚 Learning: 2026-07-07T12:38:12.052Z
Learnt from: EricAndrechek
Repo: Wave-RF/WaveHouse PR: 378
File: internal/auth/auth.go:119-132
Timestamp: 2026-07-07T12:38:12.052Z
Learning: In this repo, do not add or recommend logging/tracing client IP addresses using naive or untrusted sources (e.g., `r.RemoteAddr` or directly trusting/deriving `X-Forwarded-For`) anywhere in the Go codebase. `middleware.RealIP` was removed due to IP-spoofing risks, and proper trusted-proxy-aware client-IP handling is intentionally deferred to issue `#333`. During code review, if proposed changes would record client IPs (including in audit paths such as `internal/auth/auth.go`), reject/redirect until `#333` lands with correct trusted-proxy configuration and safeguards.
Applied to files:
internal/policy/policy.gointernal/policy/rowfilter.gointernal/stream/hub.gointernal/stream/hub_test.go
📚 Learning: 2026-06-26T12:23:22.696Z
Learnt from: EricAndrechek
Repo: Wave-RF/WaveHouse PR: 346
File: internal/stream/subscriber_test.go:9-28
Timestamp: 2026-06-26T12:23:22.696Z
Learning: In this Go repository, prefer table-driven tests (e.g., `[]struct{...}` with `t.Run(...)`) only for tests that cover multiple scenarios/inputs and can be cleanly enumerated. Do not artificially rewrite a clear single-scenario sequential behavioral-flow test into a table-driven form just to fit the pattern; if there’s only one meaningful scenario, keep the test as a straightforward linear flow (as in `TestSubscriber_SendDeliversThenDropsWhenFull`).
Applied to files:
internal/stream/hub_test.go
🔇 Additional comments (8)
internal/policy/policy.go (2)
128-206: LGTM!
239-261: 📐 Maintainability & Code QualityNo change needed. Predicate resolution and rendering already have end-to-end
Evaluatecoverage for resolved claim values, empty_invalues, combined predicates, and matchingWhereClause/WhereParams.internal/policy/rowfilter.go (1)
40-44: LGTM!internal/stream/hub.go (1)
22-22: LGTM!Also applies to: 40-44, 109-114, 115-199, 203-232, 274-276, 282-293
internal/stream/hub_test.go (1)
21-26: LGTM!Also applies to: 41-51, 68-68, 100-100, 136-136, 163-253, 286-288, 331-331, 352-352, 373-373, 390-390, 411-411, 425-425, 438-462, 464-466
CHANGELOG.md (1)
34-34: LGTM!docs/src/content/docs/architecture.md (1)
88-88: LGTM!tests/e2e/sdk/streaming.test.ts (1)
3-10: LGTM!Also applies to: 28-39
| // NOT applied here; the caller checks perms.RowVisible per subscriber (with that | ||
| // subscriber's claims) when perms.HasRowFilter(). ok=false means skip: the role can't |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Name the claims-bound permissions in this comment.
projectColumns returns the claims-independent perms. The callers resolve subPerms with each subscriber’s claims before calling RowVisible. Change perms.RowVisible to subPerms.RowVisible, or use “the per-subscriber resolved permissions,” to prevent future code from reusing the nil-claims permissions for row visibility. (github.com)
|
Two doc-sync findings that land on lines this PR doesn't touch, so they can't be line-anchored as review comments.
That enumeration is now incomplete in a security-relevant way. Worth noting here too that the stream now depends on claims, so a connection authenticated via
The column section immediately above it (line 177) already names both read surfaces — "the structured-query and live-stream paths defer to the same per-column decision" — and after this PR the row filter has the identical property: one |
EricAndrechek
left a comment
EricAndrechek
left a comment
There was a problem hiding this comment.
Nice work. Reviewing this PR took a lot longer than I expected/wanted it to – it really made me realize how complex our policy engine and stuff is starting to become... I originally had a BUNCH of things I was adding to this PR review about policy stuff I didn't like, but I think it's more important to get this security fixing PR merged and then to refactor and make policy more maintainable separately. Also some efficiency/perf type things, especially in the claim evaluation predicate stuff, SQL building, etc – like I noticed in this review just how much of that is wasteful recomputation and could be memoized or smth, and how much happens on the hot path, etc., but same idea: let's tackle that later in another PR so this one can just get shipped. I also forgot about our discussion about the row policy logic and how it being done correctly would technically require a CH SQL engine executing, which is what the query path does, but iirc what we decided on and you built was to just do a best effort of it now that fails closed until we can figure that out so we at least don't have a gaping security hole, as we'd rather that failed closed behavior occasionally accidentally blocking data from being streamed that should have been allowed as opposed to the inverse.
My notes on this PR include:
- a case in
policy/rowfilter.gowhere I think we actually fail open? Would love a test case to confirm that first, and then if my theory is correct, for it be closed and test case to enforce it so it can't regress. So like"9" > "100"being a test that should NOT pass I think, right? Also need to make sure_neqproperly works and fails closed too for instance, where like UUIDs that have different caps applied but that CH might see as the same etc need to be equal but the current code would pass them through to trivial string comparisons I think. - a reference to some of the perf / memoization stuff in the code to a new issue I opened (#435) for it just to document
- a small req for us to add in some o11y for when rls filters/withholds data and some other minor o11y fixes
- test case expansion, specifically for the hub stuff: I don't think that the dual goroutine interaction was properly being tested with race coverage for sub claims?
- a minor immutability req on
stream/subscriber.go's claims since those (I don't think need to be mutable? Feel free to pushback if I missed something though...) shouldn't change/hot-reload or at least have no mechanism to mid-stream right now (which kind of leads into the broader #239 issue...) - a bunch of smaller stale comment or docs type things – not sure if you have been or not, but HIGHLY recommend/suggest you get in the practice of running
/prepushor/docs-reviewfor just docs (or/pr-review-locallyfor other PR reviews) inclaudeon your local worktree/branch before requesting human review etc as it does a good job at catching low hanging fruit, especially these sorts of doc inconsistencies for you
| return false | ||
| } | ||
| for _, pred := range p.rowFilter { | ||
| if !pred.matches(row, numericCols[pred.Column]) { |
There was a problem hiding this comment.
numericCols[pred.Column] collapses three distinct states into false: a known text column, a column absent from the schema, and no schema at all — Hub.numericCols returns nil when the registry is nil or when registry.Get(table) misses. The first is correct. The other two silently downgrade _gt/_lt to lexicographic comparison, which admits rows ClickHouse would exclude: "9" > "100" is true as text, false as numbers.
This contradicts the contract stated at lines 31–33 of this file — "Every ambiguous or uncomparable case fails closed — the row is hidden, never leaked" — and rowfilter_test.go:64 currently pins the leak as expected behavior.
The no-schema state is reachable in production, not just in the nil-registry test hub: cmd/wavehouse/main.go:198 treats a failed boot-time registry.Refresh as non-fatal and serves while retrying in the background, so every table reads as unknown for that window. A table dropped or renamed since the replayed events were published lands in the same place.
| func TestHub_ConcurrentAddRemoveBroadcast_Race(t *testing.T) { | ||
| t.Parallel() | ||
| hub := NewHub(nil, nil) | ||
| hub := NewHub(nil, nil, nil) |
There was a problem hiding this comment.
This is the only test that drives the hub from multiple goroutines, and it builds a passthrough hub. With a nil policy store projectColumns returns a nil perms, so HasRowFilter() is false and the new per-subscriber sub.claims read at hub.go:169 never executes here.
The tests that do reach that line (TestHub_RowFilter_*) call SetClaims, Add and Broadcast from a single goroutine, so the cross-goroutine read this PR introduces is never observed by the race detector anywhere in the suite. -race covers unit and integration only (Makefile:666, :680), and the e2e binary is built -cover -covermode=atomic under CGO_ENABLED=0 (scripts/build.sh), so the new e2e row-filter coverage doesn't close the gap either.
| :::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. |
There was a problem hiding this comment.
Two problems with the representation caveat.
Scope. Numeric comparison is selected only when discovery.IsNumericType is true, and that covers Int*/UInt*/Float*/Decimal* only (internal/discovery/validation.go:161). So every other type compares as text — Enum, UUID, IPv4/IPv6, Date, String — not just Bool and DateTime as written here.
Direction. The failure isn't uniformly fail-closed. A representation mismatch withholds the row under _eq/_in, but under _neq the text compare returns "not equal" and admits a row the query path would have excluded — an uppercase UUID carried in a claim, or 1 for a Bool.
| 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. |
There was a problem hiding this comment.
This bullet is updated, but the Streaming Path flow block at lines 261–263 still reads "Per-role policy filtering (historical + live): denied tables skipped, denied columns stripped. Live projection runs once per role (Hub.Broadcast)". It doesn't mention row filtering, and "once per role" now holds only for roles without a filter. The two descriptions in this file disagree.
| 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. | ||
| ::: |
There was a problem hiding this comment.
The caution doesn't cover the fail-closed case a policy author is most likely to hit. rowfilter.go:56-58 withholds the row whenever the filter's column is absent from the event payload, and the stream evaluates against the ingested JSON rather than the stored row. So a filter keyed on a column clients don't send — or on a ClickHouse DEFAULT/MATERIALIZED/ALIAS column, which never appears in an SSE payload at all — silently withholds every event for that subscriber while the query path returns those rows normally.
| numericResolved = true | ||
| } | ||
| for _, sub := range rb.bucket.Snapshot() { | ||
| subPerms := policy.Evaluate(p, rb.role, evt.TableName, "select", sub.claims) |
There was a problem hiding this comment.
Filed as #435 — not asking for it in this PR, but flagging it here so the cost is recorded rather than silently absorbed.
Nothing in this Evaluate call depends on the event. role and table are fixed for the connection (one subscriber = one topic = one table), claims are fixed at connect, and p changes only on policy reload — yet it runs once per event per subscriber. A subscriber connected across 10k events recomputes an identical ResolvedPermissions 10k times, each one paying a resolveTemplate regexp pass per filter value plus a predicatesToSQL render whose WhereClause/WhereParams the stream never reads.
Worth being explicit in the PR description or CHANGELOG that this trades some of the #294/#353 per-role fan-out gain for the RLS fix — right now the tradeoff isn't recorded anywhere, and af7afb9/8987480 (which had solved it) were reverted in 2379bc4 with an empty commit body.
| for _, sub := range rb.bucket.Snapshot() { | ||
| subPerms := policy.Evaluate(p, rb.role, evt.TableName, "select", sub.claims) | ||
| if !subPerms.RowVisible(evt.Data, numericCols) { | ||
| continue // this row is filtered out for this subscriber |
There was a problem hiding this comment.
A row withheld here is invisible to operators. Metrics exposes only ConnOpened / ConnClosed / FrameSent / FrameDropped (internal/stream/metrics.go), and FrameDropped means a full queue — an RLS withhold isn't counted anywhere.
That matters more than it would for an ordinary filter, because this path fails closed silently in several distinct ways: a filter column absent from the event payload (rowfilter.go:56), a non-scalar value, an unparseable or NaN operand, and — today — any table the registry has no schema for. All of them produce the same outcome: the subscriber quietly receives nothing.
An operator debugging "this subscriber isn't getting events" has no way to separate no matching rows from every row withheld by a misconfigured filter, and no signal at all when a policy edit starts withholding everything for a role. A counter here (ideally labelled by table/role) would make a fail-closed security feature observable.
| } | ||
| if perms.HasRowFilter() { | ||
| subPerms := policy.Evaluate(p, role, evt.TableName, "select", claims) | ||
| if !subPerms.RowVisible(evt.Data, h.numericCols(evt.TableName)) { |
There was a problem hiding this comment.
h.numericCols(evt.TableName) runs once per replayed event. replayFromNATS loops (internal/api/stream.go:168), so a gap-fill of N events does N registry lookups plus N make(map[string]bool, len(schema.Columns)) allocations, each walking the full column list.
Broadcast hoists exactly this behind numericResolved (lines 164-167) and reuses it across every role and subscriber for the event. Replay doesn't, even though a Last-Event-ID reconnect after an outage is precisely the case where N gets large. Low stakes — replay is one-time and bounded — but the pattern is already sitting next door in the same file.
| // therefore keeps the shared column projection but evaluates row visibility PER | ||
| // subscriber (ResolvedPermissions.RowVisible) before delivering — closing the | ||
| // query/stream RLS drift in #319. Roles without a row-filter keep the pure | ||
| // once-per-role fast path unchanged. See projectColumns. |
There was a problem hiding this comment.
This doc comment is updated, but the AGENTS.md invariant index isn't — and per AGENTS.md §Documentation Sync an architecture change maps to both architecture.md and AGENTS.md. (Commenting here because AGENTS.md isn't in this diff, so it can't be line-anchored.)
AGENTS.md:44 still describes the package as "Broadcast projects + serializes each event once per role, the #294 delivery hot path" — now true only for roles without a filter.
AGENTS.md:61 (Key Design Decision #12) closes with "Structured and live-stream (stream.filterColumns) reads share the one per-column decision policy.IsColumnAllowed, so column visibility can't drift." This PR builds the row-level twin of exactly that — one resolvePredicates call feeding both predicatesToSQL and RowVisible — and it isn't recorded anywhere in the index. That index is what the next agent touching internal/stream or internal/policy reads to learn what has to stay true, and "row visibility can't drift either" is precisely the kind of invariant it exists to hold.
There was a problem hiding this comment.
BenchmarkBroadcast_RowFilteredFanout existed at 8987480 and was removed by 2379bc4 along with the optimization it measured. HEAD has no benchmarks in this package at all.
This connects to how the performance thread on hub.go:178 was closed. The reply there was: "we'd rather add a benchmark and optimize if profiling on a real filtered-high-fanout topic shows it matters, than add the complexity speculatively" — and CodeRabbit recorded a persistent learning from that exchange, so the flag is unlikely to resurface on its own in future reviews.
The branch's final state has neither the optimization nor the benchmark, so the condition that thread was resolved on can't actually be evaluated by anyone. Restoring just the benchmark — without the compiled-filter machinery that was reverted — would settle the question on data and guard the path against future regressions. See #435.
2 participants
Summary
The Server-Sent Events stream applied a role's column allowlist but never its row-level
filter, so a subscriber could receive rows the structured-query path would have hidden for that same role — a row-level-security bypass on the streaming surface. This applies the same row-filter on the stream:ResolvedPermissions.RowVisible) on the stream — so the two can't drift._gt/_lt) are schema-informed — numeric columns compare numerically via the schema registry now wired into the hub — while equality/set (_eq/_neq/_in) stay exact. Ambiguous values fail closed; the best-effort ordering boundary is documented in access-control docs.Related Issues
Closes #319