fix(stream): apply policy row-filter per subscriber on SSE - #381
Conversation
github-actions
Bot
added
documentation
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughSummary by CodeRabbit
WalkthroughThe change adds shared, schema-aware row-filter enforcement to live and replayed SSE delivery. Subscriber claims are captured immutably. Numeric and timestamp comparisons fail closed when values are invalid or unsupported. Tests cover isolation, replay, metrics, concurrency, and ClickHouse parity. ChangesStreaming row-level security
Estimated code review effort: 5 (Critical) | ~120 minutes Merge Risk: 🔵 Low · up to The SSE change applies row-level filtering per subscriber, but a concurrency test may finish without actually exercising that filtered path, while a changelog cleanup and duplicated numeric mapping still need owner follow-up. The PR is mergeable with explicit awareness of these bounded test and release-maintenance risks. Sequence Diagram(s)sequenceDiagram
participant Client
participant StreamHandler
participant Subscriber
participant Hub
participant ResolvedPermissions
participant SchemaRegistry
Client->>StreamHandler: connect with JWT
StreamHandler->>Subscriber: snapshot claims
StreamHandler->>Hub: register subscriber
Hub->>ResolvedPermissions: resolve row predicates
Hub->>SchemaRegistry: resolve column types
Hub->>Subscriber: deliver visible projected event
🚥 Pre-merge checks | ✅ 3 | ❌ 2❌ Failed checks (2 warnings)
✅ Passed checks (3 passed)
✨ Finishing Touches 💡 1📝 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://7ddba31b-wavehouse-docs.wave-rf.workers.dev |
Code Coverage OverviewLanguages: Go GoThe overall coverage in commit a84fd3d 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!
There was a problem hiding this comment.
Pull request overview
Applies row-level policies to SSE live and replay delivery, addressing #319.
Changes:
- Shares resolved predicates between SQL queries and in-memory stream filtering.
- Adds claims-aware subscriber filtering, schema-informed comparisons, and withheld-row metrics.
- Expands unit, integration, E2E, security, and user documentation.
Reviewed changes
Copilot reviewed 30 out of 30 changed files in this pull request and generated 1 comment.
Show a summary per file
| File | Description |
|---|---|
AGENTS.md |
Records the new streaming security invariant. |
CHANGELOG.md |
Documents the SSE row-filter security fix. |
SECURITY.md |
Updates the access-control posture. |
cmd/wavehouse/main.go |
Supplies schemas to the stream hub. |
docs/src/content/docs/access-control.mdx |
Documents stream enforcement and comparison boundaries. |
docs/src/content/docs/api.md |
Updates SSE and timestamp behavior. |
docs/src/content/docs/architecture.md |
Describes claims-aware fan-out architecture. |
docs/src/content/docs/sdk/streaming.md |
Documents server-side stream filtering. |
internal/api/errors_test.go |
Updates hub construction. |
internal/api/router_test.go |
Updates router test wiring. |
internal/api/stream.go |
Passes claims into live and replay filtering. |
internal/api/stream_test.go |
Updates stream-handler setup. |
internal/discovery/timestamp.go |
Exposes timestamp parsing for policy comparison. |
internal/discovery/timestamp_test.go |
Tests timestamp comparison parsing. |
internal/discovery/validation.go |
Exposes numeric and string type classifiers. |
internal/discovery/validation_test.go |
Tests exported type classification. |
internal/policy/policy.go |
Creates shared resolved row predicates. |
internal/policy/policy_test.go |
Tests numeric claim canonicalization. |
internal/policy/rowfilter.go |
Implements in-memory row visibility evaluation. |
internal/policy/rowfilter_test.go |
Tests row-filter operators and edge cases. |
internal/stream/bucket_test.go |
Updates subscriber construction. |
internal/stream/doc.go |
Documents per-subscriber row filtering. |
internal/stream/heartbeat_test.go |
Updates subscriber construction. |
internal/stream/hub.go |
Applies row filters during live delivery and replay. |
internal/stream/hub_test.go |
Tests isolation, replay, metrics, types, and concurrency. |
internal/stream/metrics.go |
Adds the withheld-row counter. |
internal/stream/subscriber.go |
Stores immutable connection claims. |
internal/stream/subscriber_test.go |
Updates subscriber setup. |
tests/e2e/sdk/streaming.test.ts |
Verifies per-claim SSE isolation end to end. |
tests/integration/setup_test.go |
Wires schemas into the integration hub. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
EricAndrechek
left a comment
EricAndrechek
left a comment
There was a problem hiding this comment.
So I am going to send this back for changes before finishing reviewing it all so that the larger requirements can start while I do more reading, but I think some changes larger than you may have been anticipating are going to be required here, meaning I will have to pretty heavily re-review after them anyway.
Basically, a LOT of the code I am reading and reviewing now is giving me flashbacks to the PR we JUST finished on (#457) and cases we already spent many iterations going back and forth on there, etc finding so so many niche edge cases and closing them, etc. It feels like we now have a lot of duplicate code or code attempting to do basically the exact same things in two different places – half of it pre-existing or already a WIP in this PR prior to #457, and the other half a more robust version iterated on and approved in #457.
For instance, in #457 we spent a lot of time working on coercing and comparing types and canonicalizing them safely, like in internal/policy/policy.go with the CanonicalScalar function that had that digit bound check for loop and max digits const and everything – yet (as far as I can tell) we are doing something very very similar in the new internal/policy/rowfilter.go file and its scalarString function. Obviously there are some differences across the code, but you can immediately see how much less robust the code is in rowfilter.go compared to the new code from #457, and how similar many of the functions are in what they are trying to do.
I think that the merge in from #457 was more complicated than we'd think it was going to be, and that we need to do a significant amount of refactoring what was already in flight in this PR to adopt/extend the work done in #457 and build off of it and its more rigorously tested type coercions and comparison logic.
github-actions
Bot
added
the
dependencies
|
@coderabbitai review |
✅ Action performedReview finished.
|
There was a problem hiding this comment.
Actionable comments posted: 2
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: cf14c09a-26c5-4317-b695-5ff0f3f35fbe
📒 Files selected for processing (19)
AGENTS.mdCHANGELOG.mddocs/src/content/docs/access-control.mdxdocs/src/content/docs/api.mddocs/src/content/docs/architecture.mddocs/src/content/docs/sdk/streaming.mdgo.modinternal/discovery/timestamp.gointernal/discovery/timestamp_test.gointernal/discovery/validation.gointernal/discovery/validation_test.gointernal/policy/policy.gointernal/policy/policy_test.gointernal/policy/rowfilter.gointernal/policy/rowfilter_test.gointernal/stream/hub.gointernal/stream/hub_test.gotests/e2e/sdk/streaming.test.tstests/integration/rowfilter_narrowing_test.go
📜 Review details
🧰 Additional context used
📓 Path-based instructions (5)
**/*.go
📄 CodeRabbit inference engine (AGENTS.md)
**/*.go: - Go 1.26, strict formatting (gofumpt, enforced by CI)
- Table-driven tests: Use
tests := []struct{ name string; ... }witht.Run(tt.name, ...)for test cases.
Files:
internal/discovery/timestamp.gointernal/discovery/timestamp_test.gointernal/discovery/validation_test.gointernal/policy/policy_test.gotests/integration/rowfilter_narrowing_test.gointernal/policy/policy.gointernal/discovery/validation.gointernal/stream/hub.gointernal/policy/rowfilter.gointernal/policy/rowfilter_test.gointernal/stream/hub_test.go
internal/**/*.go
📄 CodeRabbit inference engine (AGENTS.md)
internal/**/*.go: - No global state: Dependencies are passed explicitly (constructor injection).
- Comment the why, not the what. Add a comment only when the reason isn't obvious from the code; a line that matches the surrounding pattern needs none.
- DRY — one source of truth. Before adding logic, look for an existing helper, type, or constant to reuse; before duplicating a rule, factor it into one place every caller reads.
Files:
internal/discovery/timestamp.gointernal/discovery/timestamp_test.gointernal/discovery/validation_test.gointernal/policy/policy_test.gointernal/policy/policy.gointernal/discovery/validation.gointernal/stream/hub.gointernal/policy/rowfilter.gointernal/policy/rowfilter_test.gointernal/stream/hub_test.go
docs/src/content/docs/**/*.mdx
📄 CodeRabbit inference engine (AGENTS.md)
Every code change should update the corresponding docs in the same PR. A code change without its doc update is incomplete.
Files:
docs/src/content/docs/access-control.mdx
**/*_test.go
📄 CodeRabbit inference engine (AGENTS.md)
- Every new function should have corresponding test cases. Run
make lintandmake testbefore considering work complete.
Files:
internal/discovery/timestamp_test.gointernal/discovery/validation_test.gointernal/policy/policy_test.gotests/integration/rowfilter_narrowing_test.gointernal/policy/rowfilter_test.gointernal/stream/hub_test.go
tests/e2e/sdk/*.test.ts
📄 CodeRabbit inference engine (AGENTS.md)
tests/e2e/sdk/*.test.ts: Add new E2E scenarios astests/e2e/sdk/*.test.tsfiles using helpers fromtests/e2e/sdk/helpers.ts.
A new test file must (1) add its suite name toSUITESintables.tsand (2) get its names viaconst T = suiteTables("<suite>"), then referenceT.clicksetc. — never a bareclicks.
Files:
tests/e2e/sdk/streaming.test.ts
🧠 Learnings (8)
📚 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:
AGENTS.mddocs/src/content/docs/api.mddocs/src/content/docs/sdk/streaming.mddocs/src/content/docs/architecture.mdCHANGELOG.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/discovery/timestamp.gointernal/discovery/timestamp_test.gointernal/discovery/validation_test.gointernal/policy/policy_test.gotests/integration/rowfilter_narrowing_test.gointernal/policy/policy.gointernal/discovery/validation.gointernal/stream/hub.gointernal/policy/rowfilter.gointernal/policy/rowfilter_test.gointernal/stream/hub_test.go
📚 Learning: 2026-08-11T21:55:53.726Z
Learnt from: jfwoods
Repo: Wave-RF/WaveHouse PR: 434
File: clients/go/stream_test.go:159-176
Timestamp: 2026-08-11T21:55:53.726Z
Learning: For Go files in this repository, do not report direct type assertions solely because the `forcetypeassert` rule is commented out in `.golangci.yml`. Only flag a type assertion when there is an independent correctness, safety, or maintainability issue.
Applied to files:
internal/discovery/timestamp.gointernal/discovery/timestamp_test.gointernal/discovery/validation_test.gointernal/policy/policy_test.gotests/integration/rowfilter_narrowing_test.gointernal/policy/policy.gointernal/discovery/validation.gointernal/stream/hub.gointernal/policy/rowfilter.gointernal/policy/rowfilter_test.gointernal/stream/hub_test.go
📚 Learning: 2026-08-13T12:17:52.620Z
Learnt from: EricAndrechek
Repo: Wave-RF/WaveHouse PR: 470
File: docs/src/content/docs/reverse-proxy.mdx:137-144
Timestamp: 2026-08-13T12:17:52.620Z
Learning: For Wave-RF/WaveHouse documentation, verify claims about implementation control flow against the authoritative implementation source (for example, internal/auth/auth.go) rather than relying solely on docs/** content. Documentation may lag behind or paraphrase behavior, so control-flow claims should be confirmed in source code.
Applied to files:
docs/src/content/docs/access-control.mdx
📚 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/discovery/timestamp_test.gointernal/discovery/validation_test.gointernal/policy/policy_test.gotests/integration/rowfilter_narrowing_test.gointernal/policy/rowfilter_test.gointernal/stream/hub_test.go
📚 Learning: 2026-08-12T21:55:01.712Z
Learnt from: CR
Repo: Wave-RF/WaveHouse PR: 0
File: AGENTS.md:0-0
Timestamp: 2026-08-12T21:55:01.712Z
Learning: Applies to internal/**/*.go : - **Table-driven tests**: Use `tests := []struct{ name string; ... }` with `t.Run(tt.name, ...)` for test cases.
Applied to files:
internal/discovery/validation_test.gointernal/policy/rowfilter_test.go
📚 Learning: 2026-07-08T12:46:29.364Z
Learnt from: taitelee
Repo: Wave-RF/WaveHouse PR: 381
File: internal/stream/hub.go:142-178
Timestamp: 2026-07-08T12:46:29.364Z
Learning: In `internal/stream/hub.go`, the per-subscriber `policy.Evaluate(...)` call in `Broadcast` (on the row-filtered path, when `perms.HasRowFilter()` is true) is intentionally not memoized per distinct claim set. Rationale from maintainer taitelee: the claims-independent fast path (`!HasRowFilter()`) already ensures high-fanout public streams without a row-filter pay no per-subscriber cost; for topics that do carry a row-filter, visibility is inherently per-connection (different JWT claims → different rows) so the per-subscriber evaluation can't be hoisted without losing correctness, and memoization by claim set would rarely hit since subscribers in a row-filtered bucket typically have distinct tenant claims (plus `map[string]any` claims aren't cheaply hashable). This tradeoff is intentional; don't flag it as a perf issue unless profiling on a real filtered-high-fanout topic shows it matters.
Applied to files:
internal/stream/hub.go
📚 Learning: 2026-06-10T19:54:03.032Z
Learnt from: EricAndrechek
Repo: Wave-RF/WaveHouse PR: 312
File: CHANGELOG.md:0-0
Timestamp: 2026-06-10T19:54:03.032Z
Learning: In the Wave-RF/WaveHouse repository, CHANGELOG.md entries under `[Unreleased]` use descriptive Keep-a-Changelog leads (e.g. "The structured-query column allowlist is now a hard cap…"), NOT the Conventional Commit PR title verbatim. Do not flag CHANGELOG entry leads for not matching the PR title — that is not a rule in this repo. There is no `.coderabbit.yaml`, and neither `AGENTS.md` nor `CONTRIBUTING.md` requires CHANGELOG leads to match PR titles.
Applied to files:
CHANGELOG.md
🪛 ast-grep (0.45.1)
tests/integration/rowfilter_narrowing_test.go
[warning] 135-135: Narrowing a non-constant integer to a smaller fixed-width type (int8/int16/int32, uint8/uint16/uint32) can silently overflow or wrap, yielding negative or truncated values that are dangerous in size, length, or index logic. Validate the source value is within the target type's range before converting (e.g. bounds-check, or use a checked helper), and avoid narrowing untrusted or len()/parsed values.
Context: uint32(i)
Note: [CWE-190] Integer Overflow or Wraparound.
(integer-overflow-narrowing-conversion-go)
[warning] 149-149: Narrowing a non-constant integer to a smaller fixed-width type (int8/int16/int32, uint8/uint16/uint32) can silently overflow or wrap, yielding negative or truncated values that are dangerous in size, length, or index logic. Validate the source value is within the target type's range before converting (e.g. bounds-check, or use a checked helper), and avoid narrowing untrusted or len()/parsed values.
Context: uint32(id)
Note: [CWE-190] Integer Overflow or Wraparound.
(integer-overflow-narrowing-conversion-go)
[warning] 159-159: Narrowing a non-constant integer to a smaller fixed-width type (int8/int16/int32, uint8/uint16/uint32) can silently overflow or wrap, yielding negative or truncated values that are dangerous in size, length, or index logic. Validate the source value is within the target type's range before converting (e.g. bounds-check, or use a checked helper), and avoid narrowing untrusted or len()/parsed values.
Context: uint32(id)
Note: [CWE-190] Integer Overflow or Wraparound.
(integer-overflow-narrowing-conversion-go)
🪛 LanguageTool
docs/src/content/docs/access-control.mdx
[style] ~284-~284: Consider using “who” when you are referring to a person instead of an object.
Context: ...e rows are also invisible to the writer that produced them. (A Float column may in...
(THAT_WHO)
[typographical] ~435-~435: The word ‘When’ starts a question. Add a question mark (“?”) at the end of the sentence.
Context: ...lidation**, or WaveHouse refuses to boot. That turns a typo, a missing mount, or ...
(WRB_QUESTION_MARK)
[typographical] ~436-~436: The word ‘When’ starts a question. Add a question mark (“?”) at the end of the sentence.
Context: ...s denied (logged loudly, admin included). Seed one via PUT /v1/admin/policy usi...
(WRB_QUESTION_MARK)
🔇 Additional comments (35)
tests/e2e/sdk/streaming.test.ts (1)
40-46: LGTM!Also applies to: 259-292
tests/integration/rowfilter_narrowing_test.go (2)
30-230: LGTM!
30-230: 📐 Maintainability & Code QualityVerify the required test commands.
Provide CI or local evidence that
make lintandmake testpass before merge.As per coding guidelines: “Run
make lintandmake testbefore considering work complete.”Source: Coding guidelines
AGENTS.md (1)
44-44: LGTM!Also applies to: 61-61
CHANGELOG.md (1)
34-34: LGTM!Also applies to: 48-49
docs/src/content/docs/access-control.mdx (1)
181-181: LGTM!Also applies to: 222-239, 281-284, 381-404, 417-417, 435-439, 552-553
docs/src/content/docs/api.md (1)
217-217: LGTM!Also applies to: 242-243, 284-284, 293-294, 584-584
docs/src/content/docs/architecture.md (1)
79-79: LGTM!Also applies to: 88-92, 117-118, 143-144, 154-154
docs/src/content/docs/sdk/streaming.md (1)
113-113: LGTM!go.mod (1)
3-3: LGTM!internal/discovery/validation.go (1)
171-217: LGTM!Also applies to: 219-240, 242-281
internal/discovery/timestamp.go (1)
54-81: LGTM!internal/discovery/timestamp_test.go (1)
105-144: LGTM!Also applies to: 146-171
internal/policy/policy.go (3)
195-212: LGTM!Also applies to: 214-237, 259-289, 292-329
452-492: LGTM!Also applies to: 511-516, 526-590, 600-630
806-846: LGTM!Also applies to: 848-905
internal/policy/rowfilter.go (4)
20-35: LGTM!Also applies to: 37-63, 65-99, 101-122
124-188: LGTM!Also applies to: 190-209, 211-234
246-293: LGTM!Also applies to: 295-338, 342-376
385-489: LGTM!Also applies to: 491-524, 526-535
internal/discovery/validation_test.go (1)
288-327: LGTM!internal/policy/policy_test.go (3)
164-188: LGTM!Also applies to: 421-482, 484-497
499-574: LGTM!Also applies to: 576-609
824-987: LGTM!Also applies to: 1005-1064, 1094-1168, 1170-1198
internal/policy/rowfilter_test.go (4)
12-17: LGTM!Also applies to: 62-80, 90-112
117-123: LGTM!Also applies to: 129-136, 143-160, 162-183, 185-201, 203-226
228-288: LGTM!Also applies to: 314-318, 320-356
358-398: LGTM!Also applies to: 400-422, 424-444, 446-471, 473-525, 527-543
internal/stream/hub.go (4)
26-32: LGTM!Also applies to: 41-46
138-142: LGTM!Also applies to: 144-181
183-213: LGTM!Also applies to: 225-231
279-305: LGTM!internal/stream/hub_test.go (3)
26-47: LGTM!Also applies to: 153-179
242-274: LGTM!Also applies to: 391-419
545-586: LGTM!Also applies to: 708-713, 726-764
EricAndrechek
left a comment
EricAndrechek
left a comment
There was a problem hiding this comment.
Much better use of what we merged in from the prior PR, but policy/policy.go and policy/rowfilter.go are quickly become massive files that are very difficult to read. I think the only BLOCKING changes to this PR I have are on these files.
|
@coderabbitai review |
✅ Action performedReview finished.
|
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
internal/stream/hub_test.go (1)
686-714: 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick winGuarantee row-filter execution in this race test.
The hub starts with no subscribers. A broadcaster can complete all calls before a churner calls
Add.Broadcastthen returns with no role buckets and never readssub.claims.Add one persistent claims-bearing subscriber before starting the goroutines. Remove it before the final
hub.Lenassertion.Proposed test adjustment
hub := NewHub(policy.NewMemoryStore(rowFilterPolicy()), nil, nil) const topic = "ingest.clicks" raw := rawEvent(t, "clicks", "t", map[string]any{"tenant_id": "acme", "page": "/a"}) +stable := NewSubscriber(map[string]any{"tenant": "acme"}, nil) +hub.Add(topic, "viewer", stable) // ... start broadcasters and churners ... wg.Wait() +hub.Remove(topic, "viewer", stable) assert.Equal(t, 0, hub.Len(topic), "every subscriber is removed")
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: 71719911-166e-46a6-b14c-a1cca6e1150c
📒 Files selected for processing (18)
CHANGELOG.mddocs/src/content/docs/access-control.mdxdocs/src/content/docs/api.mddocs/src/content/docs/architecture.mddocs/src/content/docs/sdk/streaming.mdinternal/api/stream.gointernal/policy/canonical.gointernal/policy/numeric.gointernal/policy/policy.gointernal/policy/rowfilter.gointernal/stream/bucket.gointernal/stream/bucket_test.gointernal/stream/heartbeat_test.gointernal/stream/hub.gointernal/stream/hub_test.gointernal/stream/subscriber.gointernal/stream/subscriber_test.gotests/integration/rowfilter_narrowing_test.go
💤 Files with no reviewable changes (2)
- internal/policy/rowfilter.go
- internal/policy/policy.go
📜 Review details
🧰 Additional context used
🧠 Learnings (6)
📚 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/bucket.gointernal/stream/subscriber_test.gointernal/stream/bucket_test.gointernal/stream/heartbeat_test.gointernal/policy/numeric.gointernal/api/stream.gotests/integration/rowfilter_narrowing_test.gointernal/stream/subscriber.gointernal/policy/canonical.gointernal/stream/hub.gointernal/stream/hub_test.go
📚 Learning: 2026-08-11T21:55:53.726Z
Learnt from: jfwoods
Repo: Wave-RF/WaveHouse PR: 434
File: clients/go/stream_test.go:159-176
Timestamp: 2026-08-11T21:55:53.726Z
Learning: For Go files in this repository, do not report direct type assertions solely because the `forcetypeassert` rule is commented out in `.golangci.yml`. Only flag a type assertion when there is an independent correctness, safety, or maintainability issue.
Applied to files:
internal/stream/bucket.gointernal/stream/subscriber_test.gointernal/stream/bucket_test.gointernal/stream/heartbeat_test.gointernal/policy/numeric.gointernal/api/stream.gotests/integration/rowfilter_narrowing_test.gointernal/stream/subscriber.gointernal/policy/canonical.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/subscriber_test.gointernal/stream/bucket_test.gointernal/stream/heartbeat_test.gotests/integration/rowfilter_narrowing_test.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/api.mdCHANGELOG.mddocs/src/content/docs/sdk/streaming.mddocs/src/content/docs/architecture.md
📚 Learning: 2026-07-08T12:46:29.364Z
Learnt from: taitelee
Repo: Wave-RF/WaveHouse PR: 381
File: internal/stream/hub.go:142-178
Timestamp: 2026-07-08T12:46:29.364Z
Learning: In `internal/stream/hub.go`, the per-subscriber `policy.Evaluate(...)` call in `Broadcast` (on the row-filtered path, when `perms.HasRowFilter()` is true) is intentionally not memoized per distinct claim set. Rationale from maintainer taitelee: the claims-independent fast path (`!HasRowFilter()`) already ensures high-fanout public streams without a row-filter pay no per-subscriber cost; for topics that do carry a row-filter, visibility is inherently per-connection (different JWT claims → different rows) so the per-subscriber evaluation can't be hoisted without losing correctness, and memoization by claim set would rarely hit since subscribers in a row-filtered bucket typically have distinct tenant claims (plus `map[string]any` claims aren't cheaply hashable). This tradeoff is intentional; don't flag it as a perf issue unless profiling on a real filtered-high-fanout topic shows it matters.
Applied to files:
internal/stream/hub.go
📚 Learning: 2026-08-13T12:17:52.620Z
Learnt from: EricAndrechek
Repo: Wave-RF/WaveHouse PR: 470
File: docs/src/content/docs/reverse-proxy.mdx:137-144
Timestamp: 2026-08-13T12:17:52.620Z
Learning: For Wave-RF/WaveHouse documentation, verify claims about implementation control flow against the authoritative implementation source (for example, internal/auth/auth.go) rather than relying solely on docs/** content. Documentation may lag behind or paraphrase behavior, so control-flow claims should be confirmed in source code.
Applied to files:
docs/src/content/docs/access-control.mdx
🔇 Additional comments (17)
internal/policy/numeric.go (1)
54-84: LGTM!Also applies to: 94-107, 117-173, 186-195, 202-220, 230-260
internal/policy/canonical.go (1)
41-51: LGTM!Also applies to: 73-113, 132-137, 147-211, 222-246, 260-273
internal/stream/hub.go (1)
154-155: LGTM!Also applies to: 168-187, 221-221, 231-245, 287-295, 312-312
internal/stream/subscriber.go (1)
45-59: LGTM!Also applies to: 92-96, 106-117
internal/stream/bucket.go (1)
6-9: LGTM!internal/stream/hub_test.go (1)
22-22: LGTM!Also applies to: 101-101, 133-134, 197-197, 251-253, 280-282, 297-297, 308-331, 341-341, 358-359, 397-397, 410-410, 437-437, 450-450, 500-500, 521-521, 557-576, 667-667, 746-747, 786-786, 818-819, 857-857, 929-952
tests/integration/rowfilter_narrowing_test.go (1)
16-16: LGTM!Also applies to: 174-181
internal/api/stream.go (1)
45-51: LGTM!Also applies to: 84-84, 101-103
internal/stream/bucket_test.go (1)
15-16: LGTM!Also applies to: 34-34, 53-53, 83-84
internal/stream/heartbeat_test.go (1)
45-45: LGTM!Also applies to: 69-69, 104-104, 143-149, 176-176
internal/stream/subscriber_test.go (1)
11-11: LGTM!Also applies to: 32-32
CHANGELOG.md (1)
34-34: LGTM!Also applies to: 42-44, 48-49
docs/src/content/docs/access-control.mdx (2)
381-404: LGTM!Also applies to: 552-552
181-181: 📐 Maintainability & Code QualityNo issue found in the policy-contract documentation.
The documented predicate resolution, fail-closed claims, canonical numeric bounds, and per-subscriber stream enforcement match the implementation.
docs/src/content/docs/api.md (1)
284-284: LGTM!Also applies to: 293-294, 584-584
docs/src/content/docs/architecture.md (1)
79-79: LGTM!Also applies to: 88-92, 96-96, 117-118, 143-146, 268-277
docs/src/content/docs/sdk/streaming.md (1)
111-117: LGTM!
EricAndrechek
left a comment
EricAndrechek
left a comment
There was a problem hiding this comment.
This is MUCH better, much cleaner with the canonical.go and numeric.go broken out. Nothing significant enough to block merge of this, lgtm – but I'm triggering a final copilot run – make sure it doesn't flag anything and that all the comments are indeed resolved first before hitting merge. Thanks!
github-project-automation
Bot
moved this from In review
to Done
in WaveHouse Task Board
## Summary Moves `.stream()` / `.liveQuery()` off `EventSource` and onto `fetch`, so the JWT travels as `Authorization: Bearer` instead of `?token=` in the request URI — where every proxy, CDN, and load balancer in front of WaveHouse is free to log it. **Server untouched.** `bearerToken()` has always preferred the header, and the CORS preflight has allow-listed `Authorization` + `Last-Event-ID` since #215 — `internal/api/router_test.go:352` is an existing test naming #203 as its reason. `?token=` stays accepted for clients that genuinely can't set headers. **Advances #203; does not close it.** Tasks 1 and 2 of 3 are done (header auth, cURL flow). The third — retiring `?token=` server-side — is #468, and closing that closes this. **This does not make streams authenticated, and (regarding #203) does not close it.** `/v1/stream` stays ungated: an expired or missing token still resolves to `default_role` and gets a filtered `200`, never a `401`. What changes is that the JWT stops appearing in request URIs, and therefore in proxy, CDN, and load-balancer logs. Enforcing expiry on a live stream is #239 and is deliberately not delivered here. Part of #194. ## Decisions worth your eyes **1. Redirects are refused only when the request carries a credential.** Platforms strip `Authorization` on a cross-origin redirect ([whatwg/fetch#1544](whatwg/fetch#1544) — the mitigation for the class behind [CVE-2022-1650](GHSA-6h5x-7c5m-7cr7), which was this exact bug in the `eventsource` package) while forwarding other headers intact. A credentialed hop therefore either silently downgrades the stream to `default_role` — this endpoint answers an unauthenticated caller rather than rejecting them — or hands a configured proxy secret to whatever the redirect names. Refusing costs nothing, because following would never have produced an *authenticated* stream anyway. With no credential there's nothing to protect, so redirects are followed and CDN canonicalization, geo/LB indirection, and http→https upgrades all work. `options.fetch` overrides `redirect` if you need the credentialed case followed regardless. `manual` rather than `error`: `error` rejects with a bare `TypeError` indistinguishable from a connection failure, which the reconnect loop would retry forever against a redirect that will never stop happening. **2. `FetchLike`'s URL parameter narrows to `string`.** The design note on #269 said the type was narrow so hand-written `(url: string, init?) => Promise<Response>` middleware would assign. What shipped was `string | URL | Request`, which — parameters being contravariant — rejects exactly that. Purely additive for implementers; only code that *imported* `FetchLike` and called through it with a `URL` breaks. **3. The SDK gains its first runtime dependency: `eventsource-parser@^3.1.0`.** MIT, **zero transitive deps**, npm provenance-attested (SLSA v1), no install scripts, 61.7M downloads/week, same maintainer as the canonical `eventsource`. Dual CJS/ESM, so it composes with our `dist/index.cjs`. Measured cost in the CDN IIFE bundle: 3402 B minified, 1430 B gzipped. A caret rather than an exact pin because pinning in a *published library* duplicates the package in any consumer tree already resolving a 3.x and freezes them out of patch/security releases until we cut one — our lockfile still governs CI. The range stops below the ESM-only 4.0.0. Why rent rather than hand-roll: the correct parser is ~440 lines, and the parts that matter — reassembling a frame split across chunk boundaries without quadratic recopying, disambiguating a trailing `\r` at a chunk boundary between a bare-CR terminator and half a split `\r\n`, and capping buffered input against a hostile stream — are what a naive version gets wrong, and are unavoidable even though we control the server, since HTTP/2 and intermediaries re-chunk freely. The buffer cap is set explicitly to 16 MiB; the parser defaults to unbounded. **This retires the "zero-dependency" claim** everywhere it appeared — both READMEs, AGENTS.md invariant 14, four site pages, and an older `Unreleased` CHANGELOG entry that would otherwise have shipped in the same release notes as the entry introducing the dependency. The only surviving mentions are in released CHANGELOG history, which is a record rather than a claim. ## What this fixes beyond the headline - **Expired-token silent downgrade.** `auth()` was called once and baked into the URL; `EventSource` then reconnected forever with that token. Once expired the stream stayed open serving a reduced view. This is why #203 is a **prerequisite for #239**. - **Blank `id:` clearing resumption.** The hub emits `id: ` for passthrough payloads; per spec an empty `id` *clears* the last-event-id. The transport retains the last **non-empty** id. - **Real errors.** A rejection carries its actual status and message instead of `EventSource`'s status-free `onerror` — which, in the old transport, meant a gateway `401` surfaced as a silent `closed` with no `error` callback at all. Note the limit: a browser stream going cross-origin only sees the status if the rejection passes CORS and the gateway answered the `Authorization` preflight; otherwise it degrades to a retryable network error. - **Non-SSE `200`s refused** (`SSE_BAD_CONTENT_TYPE`). An auth gateway's login page would otherwise feed HTML to the parser, which per the SSE grammar parses to *nothing* — leaving the stream live and permanently silent. - **No Node polyfill.** `polyfills.ts` and the `eventsource` devDependency are deleted. - **`options.fetch` / `headers` / `fetchOptions` reach streams**, closing the carve-out documented in #456. ## Behavior changes to be aware of - **A credentialed cross-origin browser stream now preflights on the initial connect.** `EventSource` never preflighted at all (its request isn't a `fetch()`, so Fetch's unsafe-request flag is never set). A proxy that answers CORS itself must allow `Authorization` on `OPTIONS /v1/stream` or the stream never opens. Documented in `reverse-proxy.mdx`. - **A proxy that strips `Authorization` on `/v1/stream`, or redirects it while credentialed, now breaks.** Both previously "worked" by accident. - **Resumption is at-least-once and time-bounded** — this was always true; the docs now say so. The last event you saw is *certainly* redelivered (the id is a `received_timestamp` and replay is inclusive), the SDK does not dedupe live frames, and replay is capped by `mq.gap_window_minutes` (15 default). ## Testing `sse.test.ts` is rewritten against an injected fetch returning a scripted `ReadableStream`. The old harness stubbed a global `FakeEventSource` and could only assert on URL strings — framing, reconnect, and resumption had **no coverage at all**. 204 tests now, with 14 behavioral fixes mutation-verified: reverting each one makes a test fail, and — after review caught a case where it didn't — fail on the assertion that names it. The e2e auth test was rewritten to be discriminating: `anon` is denied `payload` on the events table, so dropping the `Authorization` header flips the assertion. The previous version could not fail. ## Beyond the nominal scope Two REST-path fixes ride along, both surfaced by review of the streaming work and both in `clients/ts/src/http.ts`. Flagging them because an SSE PR is not where you would look for them, and either can be split out on request. - **A cancelled request could throw instead of returning `ABORTED`.** The network-error backoff is the one `sleep` inside `request()`'s catch, so its rejection had no handler and escaped as a raw `DOMException`. Nothing wraps `request()`, so it reached callers as an unhandled rejection — and the `AbortController` example in our own reference demonstrated a branch that could not be taken against an unreachable server. - **Abort is now classified from the signal, not the rejection's type.** Keying off the error made the outcome depend on `maxRetries`: `AbortSignal.timeout()` raises a `TimeoutError`, so it reported `NETWORK_ERROR` at `maxRetries: 0` and `ABORTED` at `2`. It also mis-handled middleware — an `options.fetch` enforcing its own per-attempt deadline aborts an internal controller while the caller never cancelled, which is transient and should be retried, not reported as a terminal `ABORTED`. The same rule then had to be applied to the stream transport, where the old error-type check was worse: an `AbortError` from `auth()` or a custom `fetch` ended the stream terminally and emitted **nothing**. ## Review Fifteen pre-push rounds against both gating reviewers, who verify by executing the code rather than reading it. Worth knowing what they caught, since none of it was reachable by CI: | | | |---|---| | Behavior bugs | `SSE_CONNECT_ERROR` never reaching a subscriber; a closed stream stuck reporting `live`; a consumed-body guard that didn't guard; an unhandled rejection that killed the host process; a stranded reconnect timer | | Regression vs `EventSource` | `close()` from inside a handler no longer stopped delivery | | Coverage holes | deleting the bearer half of the credentialed-redirect rule left the suite green | | False claims in docs | `EventSource` "preflighted on reconnect" (it never preflights); "WaveHouse does not reject a stream" (it 400s on a missing table); "the SDK isolates a throwing handler" (true only inside the transport — several paths outside it are not, now enumerated in the SDK reference and filed as #473) | Two recurring shapes, both worth knowing before you read the diff. **In the code: a guard or cleanup applied to N−1 of N call sites.** Six defects shared it. `if (this._closed) return` now appears eight times in `sse.ts`, several added a round apart. Assume any new early-exit path in this transport is the one that got missed. **In the prose: a true mechanism attached to a wider case set than it holds for.** This accounts for essentially every documentation defect found here, and it recurred for eight consecutive rounds — three times *inside the sentence written to fix the previous instance*. The reviewers' diagnosis is the useful part: none of these were factual errors about the system, they were missing quantifiers. Nearly every claim in this area is a function of a variable the docs cannot name — the reader's token-provider latency, which origin, which credentials mode — so the domain lives only in the author's head at the moment of writing, and the next revision reaches for the deepest true mechanism and silently re-attaches it to the whole case set. The empirical tell was sharp: the rule-shaped sentences never needed correcting; the value-shaped ones were corrected every round. The Live Queries failure section is written to that conclusion — it states rules and gives the reader a test, rather than reporting which outcome is typical. Two amplifiers were also removed: sentences that counted table rows (a ninth row would have silently falsified five of them) and facts restated independently in four or five files. **If you are reviewing prose here, the question that finds bugs is "for which cases is this true?", not "is this true?"** Four axes account for essentially every defect found on this branch, and a claim that is silent about which side it means is the shape to distrust: 1. **Who rejected it** — WaveHouse (a `400` on the stream route; a `404`/`405` off it) versus something in front. Never "the server said 401": `/v1/stream` is ungated. 2. **Where the caller runs** — server-side or same-origin (statuses visible) versus browser cross-origin, where CORS can make any rejection, including a rejected preflight, indistinguishable from a network drop. 3. **Whether the request carries a credential** — decides `redirect: "manual"` versus `"follow"`, and the test is `Authorization`-or-`headers`, so cookies are *not* credentials by it (#478). 4. **Whether the failure is semantically transient** — the 4xx-is-terminal rule is a transport mechanism, not a claim about the world (#469). One structural note so it isn't rediscovered: `CHANGELOG.md` is denylisted from the docs-prose gate (`scripts/docs-prose.sh:38`), so that entry has never been read by the automated docs reviewer. It is worth reading at docs scrutiny rather than skimming as boilerplate — a false claim survived three rounds there for exactly that reason. **The Go diff is one comment, so this looks deployment-free. It isn't.** A credentialed cross-origin browser stream now preflights where `EventSource` never did, so a proxy that answers CORS itself must allow `Authorization` on `OPTIONS /v1/stream` or streams stop opening — silently, in a retry loop, not with a visible error. That break is documented in `reverse-proxy.mdx`; a reviewer reading only `internal/` will conclude nothing operational changed. ## Follow-ups filed Design work deferred out of this PR: - **#465** — gzip on `/v1/stream` (per-frame flush; measure before adopting) - **#466** — normalize a schemeless `baseURL`, with a loopback exception - **#467** — binary framing negotiated via `Accept`, sequenced behind #465 - **#468** — retire `?token=` server-side; closes #203 - **#204** — commented with the POST-body analysis that unblocks multiplexing Defects found by review of this branch and left unfixed here, each because the fix lands outside the transport or carries a design question I didn't want to answer unilaterally in a PR about auth: - **#469** — a stream `429` is terminal; should honor `Retry-After` - **#471** — a rejected resumption preflight leaves a stream re-dialing forever - **#473** — a throwing subscriber silently stops delivery to the others. Widened during review to cover every path outside the transport's guard — `.subscribe()`'s initial unguarded `status` call (worse via `liveQuery()`, which returns no handle at all), the fan-out dropping the event for a concurrent `for await` and leaving it un-terminated on a terminal close, a throwing `status` handler making `.connected()` time out against a live stream, and `liveQuery()`'s backfill flush discarding its buffer. The docs and CHANGELOG now describe the real contract rather than the one I first wrote. - **#476** — `http.ts`'s `sleep()` leaks an abort listener when the timer wins - **#484** — transport hardening against a non-conforming peer, raised as "what I'd watch" in the final review: a parser-buffer overflow re-dials at a flat rate forever because the backoff reset counts the overflowing connection as healthy, and `FetchLike`'s contract never states that `init.signal` must be honored. Neither is reachable with a conforming peer; both are cheap now. - **#477** — `StreamController` retains every event when nothing iterates. The buffer's only drain is the async iterator's `next()`, so a `.subscribe()`-only consumer — the pattern the docs lead with — holds every event it has ever received, unbounded. Doubled on a filtered or live stream, since both controller layers buffer. - **#478** — a cookie-authenticated stream bypasses the redirect guard. `credentialed` tests for a bearer token or configured `headers`; cookies are neither, so the request follows a redirect and can arrive unauthenticated. - **#449** — pre-existing liveQuery dedup boundary, re-confirmed by review - **#445** — not touched here, but surfaced again while reviewing the streaming docs and worth a person's eye: `wh.pipe(name).stream()` is documented as working in three places while `pipes.ts` streams `?table=<pipeName>`, so it subscribes to a topic nothing publishes and silently yields nothing. The DLQ variant carries an inline caveat for the same class of gap; the pipe one doesn't. ## Reviewer notes The interesting file is `clients/ts/src/stream/sse.ts`. `controller.ts` is untouched — the transport sits behind the same `StreamTransport` interface — so the diff is scoped to the transport, its tests, and the docs the change invalidated. `clients/ts/src/stream/live-query.test.ts` is new and is the first test coverage `LiveQuery` has had. It pins one thing worth knowing about: the backfill, not the stream, spends the first `auth()` call, and that ordering is emergent from four independent details rather than declared anywhere. One added `await` on the REST path silently swaps the two failure modes the docs describe, so the test exists to make that a red build rather than a documentation drift. Merged with `main` at `1064a4fe`, which brought #381 (per-subscriber SSE row filtering) and #457. Both conflicted textually with this branch and both were resolved keeping each side; #381 adds no new status code, so this PR's claim that `/v1/stream` raises exactly one 4xx itself still holds. --------- Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
Three catch-up changes, all client-side. The server is untouched. Routes: main merged every admin-gated endpoint under /v1/ops (#479) with no aliases, so thirteen call sites were 404ing against a current server — schema list/refresh, DLQ stats, raw SQL, policy get/put/validate, and pipes CRUD, plus the codegen CLI's schema fetch. Rewrote them along with the tests, the shared wire_cases.json fixture, and the Go SDK docs. The fixture is replayed by both conformance runners, so the stale paths broke the TypeScript half too; `make test-conformance-ts` is back to 45/45. ClientOptions.Headers: the TypeScript SDK gained options.headers in #456 and Go had no equivalent. Headers now apply to every request the client makes, REST and SSE alike — which is also how an operator sends the server's non-JWT X-Operator-Key. The SDK's own headers are set afterwards and win a collision; net/http canonicalizes names, so matching is case-insensitive; the map is copied at construction so later mutation can't reach into requests. SSE robustness, mirroring main's fetch-based rewrite (#470). The Go SDK already authenticated by header, so that part was never stale, but three gaps were: - A credentialed stream followed redirects. net/http drops Authorization on a cross-host hop while forwarding custom headers verbatim, so a redirect either downgraded the stream to default_role in silence or handed configured secrets to wherever it pointed. Now refused with a terminal SSE_REDIRECT. Uncredentialed streams still follow. - A 200 with any content type was treated as an event stream, so an auth gateway's login page left the stream sitting in StatusLive delivering nothing. Now a terminal SSE_BAD_CONTENT_TYPE. - Every failure collapsed into one retryable SSE_ERROR, and malformed frames came back as a bare fmt.Errorf, so errors.As and IsRetryable didn't work on them. Replaced with the taxonomy the TypeScript SDK uses — SSE_AUTH_ERROR, SSE_NETWORK_ERROR, SSE_CONNECT_ERROR, SSE_REDIRECT, SSE_BAD_CONTENT_TYPE, SSE_PARSE_ERROR, SSE_READ_ERROR — each with its own retryable flag, all delivered as *Error. Also documents what main changed underneath the Go SDK without changing its code: DateTime values arrive canonicalized to RFC 3339 UTC (#402), SSE applies policy row-filters per subscriber and fails closed (#381, #457), /v1/stream is ungated so WaveHouse never 401s a stream, and /v1/ops/dlq/stats is absent (404) when the DLQ is disabled rather than returning empty stats. Tests: terminal-failure table (bad content type, missing content type, credentialed redirect, non-HTTP scheme), redirect-followed-when- uncredentialed, typed retryable parse errors, and header precedence and copying on both transports.
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