feat: caching for local table/scope invalidation - #177
Conversation
|
Caution Review failedFailed to post review comments 📝 WalkthroughSummary by CodeRabbit
WalkthroughRedesigns cache to be namespace+scope-aware with VersionManager and LocalCache; propagates scope through ingest and streaming subjects; invalidates cache after ClickHouse inserts; removes TieredCache; adds per-handler ClickHouse query timeouts; removes client-side cache_ttl; updates docs and tests. ChangesCache architecture and ingest integration
Sequence Diagram (cache invalidation flow): sequenceDiagram
participant Bento as clickhouseOutput
participant CH as ClickHouse
participant Cache as LocalCache
Bento->>CH: INSERT batch (table, rows)
CH-->>Bento: 200 OK
Bento->>Cache: InvalidateCache(table, scopes)
Cache->>Cache: IncrementVersion(table[.scope]) for each scope
Estimated code review effort 🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related issues
Possibly related PRs
Suggested reviewers
🚥 Pre-merge checks | ✅ 2 | ❌ 3❌ Failed checks (3 warnings)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
✨ Simplify code
Comment |
github-actions
Bot
added
documentation
Summary of ChangesHello, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed! This pull request introduces a robust caching invalidation mechanism designed to improve data consistency within the system. By refactoring the cache layer to support table and scope-based invalidation and wiring this into the Bento ingest pipeline, the system now ensures that cached query results are invalidated immediately upon successful data ingestion. These changes also lay the groundwork for future multi-tenancy support through the introduction of scope-based metadata. Highlights
Using Gemini Code AssistThe full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips. Invoking Gemini You can request assistance from Gemini at any point by creating a comment using either
Customization To customize the Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a Limitations & Feedback Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counterproductive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for GitHub and other Google products, sign up here. Footnotes
|
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request refactors the caching architecture by replacing TieredCache with a generic Cache interface and introducing scoped invalidation. The ingest pipeline now supports a scope field, and cache invalidation is triggered after ClickHouse inserts. However, several critical issues were identified: using an evicting Ristretto cache for the versionTable risks serving stale data, and generateInvalidationKeys fails to invalidate global table entries during scoped inserts. Furthermore, the removal of received_timestamp injection may impact analytics, and the scope identifier lacks regex validation. Performance improvements are also suggested for Ristretto cost configurations and memory allocation across sub-caches. Iterate.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
github-project-automation
Bot
moved this from Backlog
to In progress
in WaveHouse Task Board
|
/review |
There was a problem hiding this comment.
Code Review
This pull request introduces a significant overhaul of the caching system, moving from a tiered cache to a version-based invalidation strategy using a VersionManager. It also adds support for query timeouts and scopes (tenancy) across the ingest and query paths. Key changes include the removal of the TieredCache, updates to the Cache interface to support namespaces and scopes, and the introduction of a QueryTimeToTTL function for dynamic cache expiration. However, several critical issues were identified: a regression where received_timestamp is no longer injected into ClickHouse inserts, a bug in the LocalCache.Get implementation that causes misses for keys with zero TTL, and a potential memory leak in the VersionManager due to unbounded map growth. Additionally, redundant version increments were found in the invalidation logic. Iterate.
![claude[bot]](https://avatars.githubusercontent.com/in/1236702?s=60&v=4){kind=small}
Review summary — 4 [MUST], 1 [SHOULD], 0 [MAY]See inline threads for detail. Highlights:
Iterate — address the |
There was a problem hiding this comment.
Actionable comments posted: 11
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (7)
docs/src/content/docs/api.md (1)
568-606:⚠️ Potential issue | 🟠 Major | ⚡ Quick win
EventMessagedocs are missing thescopefield and now violate the wire-format contract.The Internal Wire Format and Client-Facing SSE/WS examples still omit
scope, which makes this page inconsistent with the current event payload shape.Proposed doc fix
{ "table_name": "clicks", + "scope": "", "received_timestamp": "2026-03-24T12:00:00.123456789Z", "data": { "page": "/home", "button": "signup", "score": 42.5 } } | Field | Type | Description | | ----- | ---- | ----------- | | `table_name` | string | Target ClickHouse table (from URL). | +| `scope` | string | Optional scope namespace used for subject routing/cache invalidation context. | | `received_timestamp` | string | RFC 3339 nano timestamp when WaveHouse received the event. | | `data` | object | The original flat JSON body. |As per coding guidelines, “
internal/ingest/types.go: EventMessage JSON tags must agree with docs/src/content/docs/api.md event format, SSE/WS examples, and ClickHouse INSERT columns”.docs/src/content/docs/architecture.md (1)
194-195:⚠️ Potential issue | 🟡 Minor | ⚡ Quick winQuery timeout is documented as fixed
30s, but it is now configurable.This section should reference
clickhouse.query_timeoutinstead of a hardcoded 30s value to match current behavior and the configuration reference.docs/src/content/docs/configuration.md (1)
168-172:⚠️ Potential issue | 🟡 Minor | ⚡ Quick winExample config still includes removed
cache.default_ttl.The cache section in the example YAML is stale and contradicts the updated reference table. Please remove
default_ttlfrom the example block.Proposed doc fix
cache: l1_max_cost: 67108864 - default_ttl: 300 timestamp_bucket_seconds: 60As per coding guidelines, “
internal/config/**/*.go: Config struct tags in internal/config/config.go must agree with docs/src/content/docs/configuration.md, config.yaml, and compose env blocks”.internal/ingest/bento.go (1)
390-399:⚠️ Potential issue | 🟠 Major | 🏗️ Heavy lift
registerOncemakes the injected cache effectively process-global.Line 398 closes over
cacheinside a factory that is registered only once. After the firstStartIngestWorkercall, later workers will keep using the first call's cache/JetStream wiring, so restarts and tests become order-dependent and can invalidate the wrong cache instance.As per coding guidelines "Do not use global state; pass dependencies explicitly through constructor injection".
internal/api/pipes.go (1)
88-103:⚠️ Potential issue | 🔴 Critical | ⚡ Quick winFix empty-role authorization fail-open in allowlist checks.
Line 90 currently skips allowlist enforcement when
role == "", which can permit unintended access. Also guard against empty allowlist entries matching empty roles.🔒 Proposed fix
// Check role permissions. if len(q.AllowedRoles) > 0 { role := RoleFromContext(r.Context()) - if role != "" { - allowed := false - for _, ar := range q.AllowedRoles { - if ar == role || ar == "*" { - allowed = true - break - } - } - if !allowed { - writeJSONError(w, http.StatusForbidden, "forbidden") - return - } - } + allowed := false + for _, ar := range q.AllowedRoles { + if ar == "*" || (ar != "" && ar == role) { + allowed = true + break + } + } + if !allowed { + writeJSONError(w, http.StatusForbidden, "forbidden") + return + } }Based on learnings: remove any outer guard that skips allowlist evaluation for
role == "", and require only non-empty allowlist entries to match (ar != "" && ar == role).internal/api/query.go (1)
92-96:⚠️ Potential issue | 🟡 Minor | ⚡ Quick winUpdate constructor docs to match configurable timeout behavior.
Line 92 still documents a fixed 30s request deadline, but
NewQueryHandlernow acceptsqueryTimeout(Line 97), so this comment is outdated.internal/api/query_test.go (1)
511-512: 🛠️ Refactor suggestion | 🟠 Major | ⚡ Quick winReplace this TODO with an actual short-timeout expiry test.
Please add a deterministic test that sets a very small
queryTimeout, blocks the fake ClickHouse handler, and asserts timeout-driven failure status/body.As per coding guidelines: "Every new function should have corresponding test cases."
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: 8bcb45f4-437d-45e2-af7e-8e0c2cdc0cb1
📒 Files selected for processing (38)
clients/ts/src/query-builder.test.tsclients/ts/src/query-builder.tsclients/ts/src/types.tscmd/wavehouse/main.goconfig.yamldocs/src/content/docs/api.mddocs/src/content/docs/architecture.mddocs/src/content/docs/configuration.mddocs/src/content/docs/sdk.mdinternal/api/ingest.gointernal/api/pipes.gointernal/api/query.gointernal/api/query_test.gointernal/api/stream_sse.gointernal/api/stream_ws.gointernal/api/structured_query.gointernal/api/structured_query_test.gointernal/cache/cache.gointernal/cache/cache_test.gointernal/cache/local.gointernal/cache/local_test.gointernal/cache/tiered.gointernal/cache/tiered_test.gointernal/cache/version_manager.gointernal/cache/version_manager_test.gointernal/config/config.gointernal/config/config_test.gointernal/ingest/bento.gointernal/ingest/bento_test.gointernal/ingest/types.gointernal/query/ast.gointernal/query/ident.gointernal/query/ident_test.gointernal/testutil/mocks.gotests/e2e/sdk/cache.test.tstests/e2e/sdk/query.test.tstests/integration/dlq_test.gotests/integration/setup_test.go
💤 Files with no reviewable changes (9)
- clients/ts/src/types.ts
- internal/ingest/bento_test.go
- internal/cache/tiered_test.go
- internal/cache/tiered.go
- internal/query/ast.go
- internal/testutil/mocks.go
- clients/ts/src/query-builder.ts
- docs/src/content/docs/sdk.md
- clients/ts/src/query-builder.test.ts
📜 Review details
🧰 Additional context used
📓 Path-based instructions (14)
**/*.go
📄 CodeRabbit inference engine (AGENTS.md)
**/*.go: Use Go interfaces for core behaviors (Cache, Deduplicator, Publisher, Subscriber) to enable interface-first design with multiple implementations for standalone and clustered modes
Apply gofumpt strict formatting enforced by CI
Use structured logging with log/slog (JSON handler) instead of other logging approaches
Return errors and wrap with fmt.Errorf("context: %w", err) instead of panicking
Do not use global state; pass dependencies explicitly through constructor injection
Use lowercase, single-word (or abbreviated) package names in Go
Every new function should have corresponding test cases
Aim for 80%+ coverage on new code; project-wide CI enforces 80% minimum via merged unit + integration + e2e profiles
Go version 1.26 with strict gofumpt formatting enforced by CI; use golangci-lint v2.11.4 pinned in Makefile; most dev tools pinned in go.mod via tool directives
Files:
internal/api/stream_ws.gointernal/query/ident_test.gointernal/ingest/types.gointernal/api/stream_sse.gointernal/api/ingest.gointernal/cache/cache_test.gointernal/api/query.gotests/integration/setup_test.gointernal/api/query_test.gointernal/api/structured_query_test.gointernal/query/ident.gointernal/config/config.gointernal/cache/version_manager_test.gointernal/cache/cache.gointernal/api/structured_query.gotests/integration/dlq_test.gocmd/wavehouse/main.gointernal/cache/version_manager.gointernal/config/config_test.gointernal/api/pipes.gointernal/ingest/bento.gointernal/cache/local.gointernal/cache/local_test.go
internal/api/**/*.go
📄 CodeRabbit inference engine (AGENTS.md)
internal/api/**/*.go: Use Chi v5 for HTTP routing in WaveHouse
Use Bearer-token-only CORS posture: never emit Access-Control-Allow-Credentials, rely on Authorization: Bearer headers, not cookies
Schema-driven ingest: POST /v1/ingest/{table} accepts flat JSON body, validates against discovered schema (unknown fields rejected, types checked, nullable constraints enforced), returns 200 immediately after optional dedup + MQ publish
/health returns 200 after first successful schema Refresh; /ready returns 503 with latest diagnostic until Refresh succeeds
Files:
internal/api/stream_ws.gointernal/api/stream_sse.gointernal/api/ingest.gointernal/api/query.gointernal/api/query_test.gointernal/api/structured_query_test.gointernal/api/structured_query.gointernal/api/pipes.go
internal/api/**/*stream*.go
📄 CodeRabbit inference engine (AGENTS.md)
Use NATS DeliverByStartTime for SSE/WS gap-fill; no in-process ring buffer
Files:
internal/api/stream_ws.gointernal/api/stream_sse.go
**/*_test.go
📄 CodeRabbit inference engine (AGENTS.md)
**/*_test.go: Use table-driven tests with t.Run(tt.name, ...) for test cases in Go
Use shared mocks from internal/testutil/ (MockPublisher, MockCache, MockDeduplicator, MockSubscriber) instead of creating ad-hoc mocks
Use testutil.NewTestSchemaRegistry(tables) or discovery.NewSchemaRegistryFromMap(tables) for schema-aware tests
Files:
internal/query/ident_test.gointernal/cache/cache_test.gotests/integration/setup_test.gointernal/api/query_test.gointernal/api/structured_query_test.gointernal/cache/version_manager_test.gotests/integration/dlq_test.gointernal/config/config_test.gointernal/cache/local_test.go
internal/query/**/*.go
📄 CodeRabbit inference engine (AGENTS.md)
Structured queries (POST /v1/tables/{table}/query) are type-safe query AST validated against schema, with permission enforcement, timestamp bucketing for cache optimization, and 10,000 row DefaultMaxRows limit cap
Files:
internal/query/ident_test.gointernal/query/ident.go
internal/ingest/**/*.go
📄 CodeRabbit inference engine (AGENTS.md)
to prevent silent data loss; DLQ is controlled by dlq.enabled config
internal/ingest/**/*.go: Ingest is insert-only: the EventMessage wire format carries {table_name, received_timestamp, data} and nothing else; non-insert mutations (DELETE/UPDATE/TRUNCATE) go through POST /v1/admin/query under admin/service role
Per-table batching: the Bento ingest pipeline groups events by table name and performs dynamic INSERTs using schema column order; each table batch is independent
Failed batch inserts are published to a separate NATS stream (WAVEHOUSE_DLQ) with subjects dlq.
Active Sweeper purges NATS messages that are both ACKed (written to ClickHouse) and older than the gap windowFiles:
internal/ingest/types.gointernal/ingest/bento.gointernal/ingest/types.go
📄 CodeRabbit inference engine (AGENTS.md)
EventMessage JSON tags must agree with docs/src/content/docs/api.md event format, SSE/WS examples, and ClickHouse INSERT columns
Files:
internal/ingest/types.gotests/e2e/sdk/**/*.test.ts
📄 CodeRabbit inference engine (AGENTS.md)
E2E tests in tests/e2e/sdk/*.test.ts exercise the full pipeline and use helpers from tests/e2e/sdk/helpers.ts; run with make test-e2e
Files:
tests/e2e/sdk/query.test.tstests/e2e/sdk/cache.test.tsinternal/api/**/*ingest*.go
📄 CodeRabbit inference engine (AGENTS.md)
If NATS stream is full on ingest, return 503 + Retry-After instead of blocking
Files:
internal/api/ingest.gointernal/cache/**/*.go
📄 CodeRabbit inference engine (AGENTS.md)
TieredCache uses golang.org/x/sync/singleflight to prevent cache stampede
Files:
internal/cache/cache_test.gointernal/cache/version_manager_test.gointernal/cache/cache.gointernal/cache/version_manager.gointernal/cache/local.gointernal/cache/local_test.gotests/integration/**/*_test.go
📄 CodeRabbit inference engine (AGENTS.md)
Integration tests must be marked with //go:build integration and use Docker testcontainers for ClickHouse; E2E tests use the SDK as the primary test harness in tests/e2e/sdk/
Files:
tests/integration/setup_test.gotests/integration/dlq_test.go**/api/**/*_test.go
📄 CodeRabbit inference engine (AGENTS.md)
**/api/**/*_test.go: Use testutil.MakeJWT(t, claims) and testutil.MakeExpiredJWT(t, claims) for JWT authentication tests
Use testutil.AssertJSONResponse(t, rec, status, expected) and testutil.AssertJSONContains(t, rec, status, substring) for HTTP handler assertionsFiles:
internal/api/query_test.gointernal/api/structured_query_test.gointernal/config/**/*.go
📄 CodeRabbit inference engine (AGENTS.md)
Config struct tags in internal/config/config.go must agree with docs/src/content/docs/configuration.md, config.yaml, and compose env blocks
Files:
internal/config/config.gointernal/config/config_test.gocmd/wavehouse/**/*.go
📄 CodeRabbit inference engine (AGENTS.md)
Schema-discovery failure on boot is non-fatal: record api.BootState diagnostic, bind :8080, and retry via SchemaRegistry.RetryRefresh (exp backoff 2s → 60s) in background
Files:
cmd/wavehouse/main.go🧠 Learnings (3)
📚 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/query_test.gointernal/api/structured_query_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/query_test.gointernal/api/structured_query_test.go📚 Learning: 2026-05-20T20:30:15.808Z
Learnt from: taitelee Repo: Wave-RF/WaveHouse PR: 172 File: internal/api/pipes_test.go:106-118 Timestamp: 2026-05-20T20:30:15.808Z Learning: For WaveHouse pipes authorization allowlist checks, fix the empty-role fail-open behavior by (1) removing any outer guard that prevents allowlist evaluation when the incoming `role` is `""` (e.g., don’t short-circuit with `if role != "" { ... }`), and (2) during allowlist scanning, ensure only non-empty allowlist entries can match—e.g., require `ar != "" && ar == role` (so a malformed allowlist like `["" ]` cannot grant access to an empty incoming role via `"" == ""`).Applied to files:
internal/api/pipes.go🪛 LanguageTool
docs/src/content/docs/architecture.md
[style] ~108-~108: Consider using the typographical ellipsis character here instead.
Context: ...which publishes the inner data payload ({"id":"abc","field":...}) todlq.{table}NATS subjects when ...(ELLIPSIS)
🔇 Additional comments (7)
internal/cache/cache.go (1)
10-34: LGTM!Also applies to: 37-80
internal/cache/version_manager.go (1)
13-49: LGTM!internal/cache/local_test.go (1)
18-22: LGTM!Also applies to: 31-41, 51-61, 70-78, 87-98, 100-131
tests/integration/setup_test.go (1)
319-319: LGTM!tests/integration/dlq_test.go (1)
62-62: LGTM!Also applies to: 106-106, 131-131
tests/e2e/sdk/query.test.ts (1)
190-225: LGTM!config.yaml (1)
46-46: LGTM!
github-project-automation
Bot
moved this from In progress
to In review
in WaveHouse Task Board
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 (5)
docs/src/content/docs/api.md (1)
117-122:⚠️ Potential issue | 🟠 Major | ⚡ Quick winEndpoint shape drift: document ingest as path param, not query param.
Line 117 (and the example at Line 626) documents
POST /v1/ingest?table={table}, which conflicts with the API contract usingPOST /v1/ingest/{table}. This will send users to the wrong URL shape.Suggested doc fix
-### `POST /v1/ingest?table={table}` — Ingest Data +### `POST /v1/ingest/{table}` — Ingest Data @@ -curl -X POST http://localhost:8080/v1/ingest?table=clicks \ +curl -X POST http://localhost:8080/v1/ingest/clicks \ @@ -curl -X POST http://localhost:8080/v1/ingest?table=clicks \ +curl -X POST http://localhost:8080/v1/ingest/clicks \As per coding guidelines: “Schema-driven ingest:
POST /v1/ingest/{table}accepts flat JSON body…”.Also applies to: 626-626
internal/cache/local.go (2)
19-23:⚠️ Potential issue | 🟡 Minor | ⚡ Quick win
NumCountersbecomes zero for smallmaxCostvalues due to integer division.The expression
maxCost / 100 * 10uses integer division. IfmaxCost < 100, this evaluates to0, which may cause Ristretto initialization to fail or behave unexpectedly.Proposed fix
func NewLocal(maxCost int64) (*LocalCache, error) { cache, err := ristretto.NewCache(&ristretto.Config[string, []byte]{ - NumCounters: maxCost / 100 * 10, + NumCounters: maxCost / 10, // Simplified; consider max(maxCost/10, 1000) for small caches MaxCost: maxCost, BufferItems: 64, })
53-60: 🧹 Nitpick | 🔵 Trivial | 💤 Low valueUnused
ctxparameter inInvalidateCache.The
ctxparameter is accepted but never used. While this matches the interface signature (needed for future Redis implementations), consider adding a brief comment or using_ context.Contextto signal intent.Suggested clarification
-func (l *LocalCache) InvalidateCache(ctx context.Context, table string, scopes map[string]struct{}) (uint64, error) { +func (l *LocalCache) InvalidateCache(_ context.Context, table string, scopes map[string]struct{}) (uint64, error) { + // ctx unused for local cache; required by interface for distributed implementations keys := generateInvalidationKeys(table, scopes)internal/ingest/bento.go (1)
391-410: 🧹 Nitpick | 🔵 Trivial | ⚖️ Poor tradeoff
registerOnce.Doclosure capturescachefrom first invocation only.The
registerOnce.Doblock capturescache,host,chHTTPPort, and other parameters from the first call toStartIngestWorker. Subsequent calls with different values will silently use the original configuration. While this is likely intentional (single worker per process), consider documenting this constraint or returning an error on subsequent calls with different parameters.internal/api/pipes.go (1)
88-103:⚠️ Potential issue | 🔴 Critical | ⚡ Quick winEnforce allowlist checks even when the incoming role is empty.
Line 90skips authorization whenrole == "", which creates a fail-open path for pipes withAllowedRoles. Evaluate the allowlist unconditionally and only allow non-empty exact matches (plus*).Based on learnings: remove outer guards that skip allowlist checks for empty roles, and require non-empty allowlist entries for exact role matches.🔒 Proposed fix
if len(q.AllowedRoles) > 0 { role := RoleFromContext(r.Context()) - if role != "" { - allowed := false - for _, ar := range q.AllowedRoles { - if ar == role || ar == "*" { - allowed = true - break - } - } - if !allowed { - writeJSONError(w, http.StatusForbidden, "forbidden") - return - } - } + allowed := false + for _, ar := range q.AllowedRoles { + if ar == "*" || (ar != "" && ar == role) { + allowed = true + break + } + } + if !allowed { + writeJSONError(w, http.StatusForbidden, "forbidden") + return + } }
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: 61ddf52e-b286-45cf-b053-79c96cc196c6
📒 Files selected for processing (12)
docs/src/content/docs/api.mddocs/src/content/docs/architecture.mddocs/src/content/docs/configuration.mdinternal/api/pipes.gointernal/api/query.gointernal/api/stream_sse.gointernal/cache/local.gointernal/config/config.gointernal/config/config_test.gointernal/ingest/bento.gotests/e2e/sdk/cache.test.tstests/integration/setup_test.go
💤 Files with no reviewable changes (1)
- docs/src/content/docs/configuration.md
📜 Review details
🧰 Additional context used
📓 Path-based instructions (9)
**/*.go
📄 CodeRabbit inference engine (AGENTS.md)
**/*.go: Use Go interfaces for core behaviors (Cache, Deduplicator, Publisher, Subscriber) to enable interface-first design with multiple implementations for standalone and clustered modes
Apply gofumpt strict formatting enforced by CI
Use structured logging with log/slog (JSON handler) instead of other logging approaches
Return errors and wrap with fmt.Errorf("context: %w", err) instead of panicking
Do not use global state; pass dependencies explicitly through constructor injection
Use lowercase, single-word (or abbreviated) package names in Go
Every new function should have corresponding test cases
Aim for 80%+ coverage on new code; project-wide CI enforces 80% minimum via merged unit + integration + e2e profiles
Go version 1.26 with strict gofumpt formatting enforced by CI; use golangci-lint v2.11.4 pinned in Makefile; most dev tools pinned in go.mod via tool directives
Files:
internal/api/query.gointernal/api/pipes.gointernal/api/stream_sse.gointernal/config/config.gointernal/cache/local.gointernal/ingest/bento.gointernal/config/config_test.gotests/integration/setup_test.go
internal/api/**/*.go
📄 CodeRabbit inference engine (AGENTS.md)
internal/api/**/*.go: Use Chi v5 for HTTP routing in WaveHouse
Use Bearer-token-only CORS posture: never emit Access-Control-Allow-Credentials, rely on Authorization: Bearer headers, not cookies
Schema-driven ingest: POST /v1/ingest/{table} accepts flat JSON body, validates against discovered schema (unknown fields rejected, types checked, nullable constraints enforced), returns 200 immediately after optional dedup + MQ publish
/health returns 200 after first successful schema Refresh; /ready returns 503 with latest diagnostic until Refresh succeeds
Files:
internal/api/query.gointernal/api/pipes.gointernal/api/stream_sse.go
tests/e2e/sdk/**/*.test.ts
📄 CodeRabbit inference engine (AGENTS.md)
E2E tests in tests/e2e/sdk/*.test.ts exercise the full pipeline and use helpers from tests/e2e/sdk/helpers.ts; run with make test-e2e
Files:
tests/e2e/sdk/cache.test.ts
internal/api/**/*stream*.go
📄 CodeRabbit inference engine (AGENTS.md)
Use NATS DeliverByStartTime for SSE/WS gap-fill; no in-process ring buffer
Files:
internal/api/stream_sse.go
internal/config/**/*.go
📄 CodeRabbit inference engine (AGENTS.md)
Config struct tags in internal/config/config.go must agree with docs/src/content/docs/configuration.md, config.yaml, and compose env blocks
Files:
internal/config/config.gointernal/config/config_test.go
internal/cache/**/*.go
📄 CodeRabbit inference engine (AGENTS.md)
TieredCache uses golang.org/x/sync/singleflight to prevent cache stampede
Files:
internal/cache/local.go
internal/ingest/**/*.go
📄 CodeRabbit inference engine (AGENTS.md)
to prevent silent data loss; DLQ is controlled by dlq.enabled config
internal/ingest/**/*.go: Ingest is insert-only: the EventMessage wire format carries {table_name, received_timestamp, data} and nothing else; non-insert mutations (DELETE/UPDATE/TRUNCATE) go through POST /v1/admin/query under admin/service role
Per-table batching: the Bento ingest pipeline groups events by table name and performs dynamic INSERTs using schema column order; each table batch is independent
Failed batch inserts are published to a separate NATS stream (WAVEHOUSE_DLQ) with subjects dlq.
Active Sweeper purges NATS messages that are both ACKed (written to ClickHouse) and older than the gap windowFiles:
internal/ingest/bento.go**/*_test.go
📄 CodeRabbit inference engine (AGENTS.md)
**/*_test.go: Use table-driven tests with t.Run(tt.name, ...) for test cases in Go
Use shared mocks from internal/testutil/ (MockPublisher, MockCache, MockDeduplicator, MockSubscriber) instead of creating ad-hoc mocks
Use testutil.NewTestSchemaRegistry(tables) or discovery.NewSchemaRegistryFromMap(tables) for schema-aware testsFiles:
internal/config/config_test.gotests/integration/setup_test.gotests/integration/**/*_test.go
📄 CodeRabbit inference engine (AGENTS.md)
Integration tests must be marked with //go:build integration and use Docker testcontainers for ClickHouse; E2E tests use the SDK as the primary test harness in tests/e2e/sdk/
Files:
tests/integration/setup_test.go🧠 Learnings (1)
📚 Learning: 2026-05-20T20:30:15.808Z
Learnt from: taitelee Repo: Wave-RF/WaveHouse PR: 172 File: internal/api/pipes_test.go:106-118 Timestamp: 2026-05-20T20:30:15.808Z Learning: For WaveHouse pipes authorization allowlist checks, fix the empty-role fail-open behavior by (1) removing any outer guard that prevents allowlist evaluation when the incoming `role` is `""` (e.g., don’t short-circuit with `if role != "" { ... }`), and (2) during allowlist scanning, ensure only non-empty allowlist entries can match—e.g., require `ar != "" && ar == role` (so a malformed allowlist like `["" ]` cannot grant access to an empty incoming role via `"" == ""`).Applied to files:
internal/api/pipes.go🔇 Additional comments (13)
tests/integration/setup_test.go (1)
165-170: LGTM!Also applies to: 175-175
tests/e2e/sdk/cache.test.ts (1)
76-76: LGTM!docs/src/content/docs/architecture.md (1)
194-195: LGTM!docs/src/content/docs/api.md (1)
577-577: LGTM!Also applies to: 590-590
internal/ingest/bento.go (3)
331-332: The past review concern about returning success when invalidation fails is acknowledged in the TODO comment. For LocalCache,InvalidateCachealways returnsnil, so this error path is currently unreachable. The TODO correctly identifies that this must be addressed before adding distributed cache implementations like Redis.
159-183: LGTM!
254-279: LGTM!internal/cache/local.go (1)
47-50: LGTM!internal/api/pipes.go (1)
131-132: Known invalidation gap is already tracked.internal/api/query.go (1)
92-97: LGTM!Also applies to: 196-197
internal/config/config.go (1)
172-173: LGTM!internal/config/config_test.go (1)
186-208: ⚡ Quick winMerge query-timeout boundary checks into a table-driven
t.Runtest.
internal/config/config_test.go:186-208has two standaloneTestValidate_*cases covering the sameclickhouse.query_timeoutvalidation (negative and zero). Combine them into one table-driven test usingt.Run(tt.name, ...)per the**/*_test.gotesting guideline.internal/api/stream_sse.go (1)
48-53: LGTM!
7cfe381 to
e6c166f
Compare
github-actions
Bot
removed
the
dependencies
github-project-automation
Bot
moved this from In review
to In progress
in WaveHouse Task Board
github-project-automation
Bot
moved this from In progress
to Done
in WaveHouse Task Board
2 participants
Summary
Refactored caching package to handle table (and eventually table + scope) based cache invalidation, wiring it up to Bento ingest worker to invalidate the relevant caches when new data is pushed to clickhouse successfully.
Related Issues
Closes #85, #73