refactor: full api --> ingest --> clickhouse --> dlq refactor - #182
Conversation
github-actions
Bot
added
dependencies
|
Caution Review failedFailed to post review comments 📝 WalkthroughSummary by CodeRabbit
WalkthroughReplaces Bento with a native IngestWorker (NATS→batched ClickHouse inserts, per-row fallback, DLQ routing), refactors cache invalidation to accept version-key lists, adds NATS PublishOpt helpers and test mocks, broadens discovery validation, makes TS batch inserts concurrent, and updates tests/docs/build config accordingly. ChangesIngest Pipeline Modernization
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 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 |
coderabbitai
Bot
added
documentation
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 8
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
internal/api/ingest.go (2)
61-83:⚠️ Potential issue | 🟠 Major | ⚡ Quick winPrevent table-enumeration via auth-before-existence checks.
Line 61-66 runs schema existence checks before policy evaluation, which lets unauthorized callers distinguish unknown tables (
404) from known-but-forbidden (403). Evaluate authorization first (using the provided table name), then only reveal existence to authorized callers.Suggested fix
- schema := h.Registry.Get(table) - if schema == nil { - slog.WarnContext(ctx, "unknown table requested", "table", table) - writeJSONError(w, http.StatusNotFound, fmt.Sprintf("unknown table: %s", table)) - return - } - // FAST AUTH: Table-level policy check (Before spending CPU parsing JSON) var perms *policy.ResolvedPermissions var role string @@ if !perms.Allowed { slog.WarnContext(ctx, "policy enforcement rejected request", "role", role, "table", table) writeJSONError(w, http.StatusForbidden, "forbidden") return } } + +schema := h.Registry.Get(table) +if schema == nil { + slog.WarnContext(ctx, "unknown table requested", "table", table) + writeJSONError(w, http.StatusNotFound, fmt.Sprintf("unknown table: %s", table)) + return +}
86-92:⚠️ Potential issue | 🟡 Minor | ⚡ Quick winReject trailing tokens after JSON decode in
internal/api/ingest.go.
json.Decoder.Decode(&data)will stop after the first JSON value; without an EOF/trailing-token check, malformed multi-value bodies can be accepted and then pass schema validation. Add a secondDecodeexpectingio.EOF(and importio) so trailing garbage returns400, matching the existing pattern ininternal/api/query.go.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: 389411c2-cd6e-4f22-bc73-a9581d789577
⛔ Files ignored due to path filters (1)
go.sumis excluded by!**/*.sum
📒 Files selected for processing (25)
clients/ts/src/table.tscmd/wavehouse/main.gogo.modinternal/api/ingest.gointernal/cache/cache.gointernal/cache/cache_test.gointernal/cache/local.gointernal/cache/local_test.gointernal/cache/version_manager.gointernal/discovery/discovery_test.gointernal/discovery/validation.gointernal/discovery/validation_test.gointernal/ingest/bento.gointernal/ingest/bento_test.gointernal/ingest/worker.gointernal/mq/embedded.gointernal/mq/mq.gointernal/testutil/mocks.gotests/e2e/sdk/batching.test.tstests/e2e/sdk/cache.test.tstests/e2e/sdk/dlq.test.tstests/e2e/sdk/ingest.test.tstests/e2e/sdk/streaming.test.tstests/e2e/sdk/stress.test.tstests/integration/setup_test.go
💤 Files with no reviewable changes (5)
- internal/ingest/bento.go
- internal/ingest/bento_test.go
- tests/integration/setup_test.go
- internal/discovery/discovery_test.go
- internal/cache/cache_test.go
📜 Review details
⏰ Context from checks skipped due to timeout of 300000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (1)
- GitHub Check: CI
🧰 Additional context used
📓 Path-based instructions (9)
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/batching.test.tstests/e2e/sdk/stress.test.tstests/e2e/sdk/streaming.test.tstests/e2e/sdk/cache.test.tstests/e2e/sdk/dlq.test.tstests/e2e/sdk/ingest.test.ts
**/*.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/mq/embedded.gointernal/cache/local.gointernal/mq/mq.gointernal/cache/version_manager.gointernal/discovery/validation_test.gointernal/discovery/validation.gointernal/testutil/mocks.gointernal/cache/local_test.gointernal/api/ingest.gointernal/ingest/worker.gocmd/wavehouse/main.gointernal/cache/cache.go
internal/cache/**/*.go
📄 CodeRabbit inference engine (AGENTS.md)
TieredCache uses golang.org/x/sync/singleflight to prevent cache stampede
Files:
internal/cache/local.gointernal/cache/version_manager.gointernal/cache/local_test.gointernal/cache/cache.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/discovery/validation_test.gointernal/cache/local_test.go
clients/ts/**/*.{ts,tsx,json}
📄 CodeRabbit inference engine (AGENTS.md)
Use pnpm >= 11.1 and Node.js 22 LTS (pinned via .nvmrc) for SDK and E2E test harnesses; SDK and tests shell out to pnpm
Files:
clients/ts/src/table.ts
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/ingest.go
internal/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.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/worker.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🔇 Additional comments (17)
internal/mq/mq.go (1)
54-69: LGTM!internal/mq/embedded.go (1)
111-123: LGTM!internal/testutil/mocks.go (1)
25-33: LGTM!internal/cache/cache.go (1)
24-28: LGTM!Also applies to: 57-62
internal/cache/local.go (1)
53-58: LGTM!internal/cache/local_test.go (1)
120-125: LGTM!internal/cache/version_manager.go (1)
28-32: LGTM!internal/ingest/worker.go (6)
44-61: LGTM!
63-110: LGTM!
112-161: LGTM!
163-228: LGTM!
230-266: LGTM!
268-311: LGTM!cmd/wavehouse/main.go (1)
282-296: LGTM!tests/e2e/sdk/ingest.test.ts (1)
36-36: LGTM!Also applies to: 64-64, 114-114, 151-151, 203-203, 258-258, 341-341
tests/e2e/sdk/stress.test.ts (1)
1-60: LGTM!go.mod (1)
114-114: LGTM!
github-project-automation
Bot
moved this from Backlog
to In review
in WaveHouse Task Board
|
Warning Gemini encountered an error creating the summary. You can try again by commenting |
github-actions
Bot
added
area/pipes
There was a problem hiding this comment.
Actionable comments posted: 5
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
internal/ingest/worker.go (1)
325-331:⚠️ Potential issue | 🔴 Critical | ⚡ Quick winDo not ack the source message when DLQ publish fails.
If
PublishMsgfails at Line 325, Line 331 stillDoubleAcks the original message, which can permanently lose that event.Proposed fix
_, pubErr := w.js.PublishMsg(ctx, msg) if pubErr != nil { w.logger.ErrorContext(ctx, "NATS DLQ publish failed", "subject", subject, "error", pubErr) + return } // DoubleAck original message so NATS doesn't redeliver the corrupt data _ = pm.natsMsg.DoubleAck(ctx)
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: b2d542d6-ea19-4ca8-8eb6-b674d95de47e
📒 Files selected for processing (19)
.golangci.ymlclients/ts/src/stream/controller.tsclients/ts/src/table.tscmd/wavehouse/main.gointernal/api/dlq.gointernal/api/dlq_test.gointernal/api/ingest.gointernal/api/pipes.gointernal/api/pipes_test.gointernal/api/stream_sse.gointernal/api/structured_query.gointernal/ingest/worker.gointernal/pipes/pipes.gotests/e2e/sdk/batching.test.tstests/e2e/sdk/cache.test.tstests/e2e/sdk/dlq.test.tstests/e2e/sdk/streaming.test.tstests/e2e/sdk/stress.test.tstests/integration/dlq_test.go
💤 Files with no reviewable changes (1)
- tests/e2e/sdk/stress.test.ts
📜 Review details
⏰ Context from checks skipped due to timeout of 300000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (1)
- GitHub Check: CI
🧰 Additional context used
📓 Path-based instructions (12)
**/*.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_sse.gointernal/api/structured_query.gointernal/pipes/pipes.gointernal/api/dlq.gointernal/api/dlq_test.gointernal/api/pipes.gointernal/api/pipes_test.gointernal/api/ingest.gointernal/ingest/worker.gotests/integration/dlq_test.gocmd/wavehouse/main.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_sse.gointernal/api/structured_query.gointernal/api/dlq.gointernal/api/dlq_test.gointernal/api/pipes.gointernal/api/pipes_test.gointernal/api/ingest.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_sse.go
internal/pipes/**/*.go
📄 CodeRabbit inference engine (AGENTS.md)
Named query pipes: pre-defined SQL templates with parameter binding, role restrictions, and caching; stored in NATS KV with .sql file directory bootstrap
Files:
internal/pipes/pipes.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/dlq.test.tstests/e2e/sdk/cache.test.tstests/e2e/sdk/batching.test.tstests/e2e/sdk/streaming.test.ts
**/*_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/api/dlq_test.gointernal/api/pipes_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 assertions
Files:
internal/api/dlq_test.gointernal/api/pipes_test.go
clients/ts/**/*.{ts,tsx,json}
📄 CodeRabbit inference engine (AGENTS.md)
Use pnpm >= 11.1 and Node.js 22 LTS (pinned via .nvmrc) for SDK and E2E test harnesses; SDK and tests shell out to pnpm
Files:
clients/ts/src/table.tsclients/ts/src/stream/controller.ts
internal/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.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/worker.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/dlq_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/dlq_test.gointernal/api/pipes_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/dlq_test.gointernal/api/pipes_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.gointernal/api/pipes_test.go🧬 Code graph analysis (6)
tests/e2e/sdk/dlq.test.ts (1)
tests/e2e/sdk/helpers.ts (3)
waitForCondition(88-99)chQuery(109-125)testId(103-105)internal/api/dlq.go (1)
internal/query/ident.go (2)
SafeEncodeNATS(11-24)SafeDecodeNATS(27-31)clients/ts/src/stream/controller.ts (1)
clients/ts/src/types.ts (1)
StreamSubscriber(41-50)internal/api/pipes_test.go (2)
internal/pipes/pipes.go (1)
NewMemoryStore(294-303)internal/api/pipes.go (1)
NewPipesHandler(26-28)tests/integration/dlq_test.go (1)
internal/query/ident.go (1)
SafeEncodeNATS(11-24)tests/e2e/sdk/streaming.test.ts (1)
clients/ts/src/stream/controller.ts (1)
status(67-69)🔇 Additional comments (9)
clients/ts/src/table.ts (1)
51-63: Unbounded concurrent insert fan-out remains.This concern was already raised earlier for this segment (add concurrency limiting/chunking).
internal/api/structured_query.go (1)
56-56: LGTM!internal/pipes/pipes.go (1)
211-212: LGTM!tests/e2e/sdk/dlq.test.ts (1)
14-15: LGTM!Also applies to: 20-20, 44-47, 51-52, 58-60, 63-66
internal/api/dlq.go (1)
8-8: LGTM!Also applies to: 11-11, 40-40, 51-55
internal/api/dlq_test.go (1)
76-77: LGTM!Also applies to: 107-107
tests/e2e/sdk/cache.test.ts (1)
89-90: TTL units are still inflated by* 1000and can exceed test timing bounds.This appears to be the same unresolved issue already raised in prior review feedback.
Also applies to: 110-110
tests/e2e/sdk/streaming.test.ts (1)
53-53: LGTM!Also applies to: 85-87, 91-91, 132-132, 167-167
cmd/wavehouse/main.go (1)
432-434: LGTM!
coderabbitai
Bot
added
the
documentation
github-project-automation
Bot
moved this from In review
to Done
in WaveHouse Task Board
2 participants
Summary
Fully refactored the API --> nats --> ingest --> clickhouse --> DLQ flow to drop bento for better control + speed, dropped false negative api parsing, retrying clickhouse fails (not entire batch on DLQ).
Test plan
Related Issues
Closes #91, closes #34, starts building what's needed for #50, starts on phase 1 of #175