From f64447bbdd6db8bc5f10c0dafff90fa349c8c612 Mon Sep 17 00:00:00 2001 From: Eric Andrechek Date: Tue, 12 May 2026 18:34:10 -0400 Subject: [PATCH 1/5] chore(api)!: drop hub wildcard fan-out; SSE/WS use ?table=, not ?topic= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes #100 (part of #87). After the #89 MVP cuts every producer publishes a concrete ingest. subject and the SDK only ever subscribes to one concrete subject, so the NATS-style `*` / `>` wildcard fan-out in `Hub.Broadcast` was unused machinery — kept it and we paid for it in code, tests, and docs that implied the API was more general than it actually is. Server (BREAKING — public HTTP API): - `internal/api/hub.go`: deleted `matchTopic`, the wildcard pattern loop, the `sent` dedup map, and the `strings` import. `Broadcast` now does a single exact-topic delivery loop. - `internal/api/hub_test.go`: removed the 8 wildcard tests plus `TestMatchTopic` (~200 LOC of tests, none of which exercise a code path the SDK can reach today). - `internal/api/stream_sse.go`: replaced `?topic=` (which defaulted to `ingest.>`) with `?table=`. Returns 400 when missing. Subject is built server-side (`ingest.
`) — no more NATS subject convention leaking into the public HTTP API. - `internal/api/stream_ws.go`: same query-param swap. In-band command field renamed `"topic"` → `"table"` (raw name, no `ingest.` prefix), and the outbound envelope is now `{"table":"...","data":{...}}` instead of `{"topic":"ingest.
","data":{...}}`. SDK (TypeScript, BREAKING for direct transport users — high-level `wh.from('clicks').stream()` API is unchanged): - `ws-manager.ts`: dropped the now-unreachable wildcard dispatch loop + the `matchTopicPattern` helper. `SharedWSManager` is keyed by table name; subscribe/unsubscribe commands and the inbound envelope use the new `table` field. - `client.ts`: stopped pre-building `ingest.${table}` — passes the raw table name through to transports. - `sse.ts` / `ws.ts`: `SSEOptions`/`WSOptions` carry `table` instead of `topic`; URLs use `?table=`. Docs / changelog: - `docs/api.md`: stream endpoint tables updated (`topic` row replaced by required `table`, wildcard sentence removed); curl examples and the in-band JS example rewritten. - `docs/architecture.md`: `stream_sse.go`/`stream_ws.go` blurb describes the explicit-table contract. - `docs/development.md`: dev-loop curl examples switched to `?table=`. - `CHANGELOG.md` `[Unreleased]`: Removed (hub wildcards) + Changed (SSE/WS query param, WS in-band field, WS envelope) entries. Local: `make verify` clean (tidy + fmt + vulncheck + lint), `make test-unit` 424 pass, `make test-sdk` 120 pass, `make test-integration` 3 pass, `make test-e2e` 30 pass (incl. WS streaming), merged coverage gate 80.8% (threshold 80%). Out of scope per #100: raw SQL (`/v1/query`), pipes (`/v1/pipes/*`, `/v1/admin/pipes/*`), structured query (`/v1/tables/{table}/query`). Bonus cleanup (`clients/ts/src/pipes.ts:40-42` dead `PipeRef.stream()`) deferred to keep diff focused. Co-Authored-By: Claude Opus 4.7 (1M context) --- CHANGELOG.md | 13 ++ clients/ts/src/client.ts | 7 +- clients/ts/src/stream/sse.ts | 4 +- clients/ts/src/stream/ws-manager.ts | 86 ++++-------- clients/ts/src/stream/ws.ts | 4 +- docs/api.md | 35 +++-- docs/architecture.md | 2 +- docs/development.md | 9 +- internal/api/hub.go | 50 ------- internal/api/hub_test.go | 201 ---------------------------- internal/api/stream_sse.go | 8 +- internal/api/stream_test.go | 12 +- internal/api/stream_ws.go | 80 +++++------ 13 files changed, 119 insertions(+), 392 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e434e347..75b4be5c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,19 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ## Unreleased +### Removed +- **Hub wildcard subscriptions** (`internal/api/hub.go`, `internal/api/hub_test.go`): dropped the NATS-style `*` / `>` pattern matching from `Hub.Broadcast`, the wildcard pattern loop, the `sent` dedup map, the `matchTopic` helper, and the eight wildcard tests (plus `TestMatchTopic`). After the #89 MVP cuts every producer publishes a concrete `ingest.
` subject and the SDK only ever subscribes to one concrete subject, so the wildcard fan-out was unused machinery. Closes #100 (part of #87). Net −210 lines (mostly tests). + +### Changed +- **BREAKING:** SSE/WS streaming endpoints replaced `?topic=` with `?table=` (issue #100, option B). The NATS subject convention no longer leaks into the public HTTP API; the server builds `ingest.
` internally. SSE returns `400 Bad Request` when `?table=` is missing (was: defaulted to `ingest.>`). WS `?table=` is optional — clients can still defer to in-band subscribe commands. +- **BREAKING:** WebSocket in-band subscribe/unsubscribe commands use `"table"` (raw name) instead of `"topic"` (NATS subject): + - Before: `{"action":"subscribe","topic":"ingest.clicks"}` + - After: `{"action":"subscribe","table":"clicks"}` +- **BREAKING:** WebSocket outbound envelope is now labelled `"table"` (raw name) instead of `"topic"` (NATS subject): + - Before: `{"topic":"ingest.clicks","data":{"table_name":"clicks",...}}` + - After: `{"table":"clicks","data":{"table_name":"clicks",...}}` +- TypeScript SDK (`@wavehouse/sdk`) updated to the new contract: `SharedWSManager` is now keyed by table name (one less string concat per dispatch), `SSEOptions` / `WSOptions` carry `table` instead of `topic`, wildcard dispatch loop + `matchTopicPattern` helper removed from `ws-manager.ts`. The high-level SDK API (`wh.from('clicks').stream()`) is unchanged. + ### Removed - **`project-orchestrator.yml` workflow + its three composite-action artifacts** (`.github/workflows/project-orchestrator.yml`, `.github/actions/board-upsert-status/`, `.github/actions/set-linked-issues-status/`, `.github/scripts/board-fetch-item.sh`, `AGENTS.md`, `CHANGELOG.md`): −887 lines net. The orchestrator was the largest single source of cross-trigger complexity on this repo (3-4 workflow_run-chained runs per PR push, `statusCheckRollup` GraphQL perms quirks, integration-token `NONE` for private-org members) for behaviour that is mostly either provided natively by GitHub or a one-click manual operation on a 4-person team. Replaced by: reviewer-assign step in `housekeeping.yml` that fires once on `pull_request_target: opened` / `ready_for_review` (not per-synchronize, so it doesn't re-spam after `dismiss_stale_reviews_on_push`), plus GitHub's native Projects v2 workflows (`Auto-add to project`, `Item added`, `Pull request merged`) configured in the project UI. Trade-offs explicit in the PR body: drafts no longer auto-flip on bot-clean, `CHANGES_REQUESTED` doesn't auto-move the board card, linked-issue card mirroring is dropped. AGENTS.md §"Governance Files" + §"Task Board state machine" + §"Review tooling reference" all rewritten to match. `dependabot-automerge.yml` trimmed in parallel: no more board-upsert step (native handles placement), `PROJECT_BOARD_TOKEN` guard removed (no longer used in this workflow), reviewer list sourced from `board-config.env`'s `ADMINS` via `replace()`, major-bump comment uses the marker-comment upsert pattern from `housekeeping.yml`. - **`STATUS_*` and old `ADMINS` consumers in `board-config.env`** — STATUS option IDs had only orchestrator-side consumers and are now unreferenced. `ADMINS` was restored to `board-config.env` after the initial orchestrator-removal commit dropped it (Gemini and Claude both flagged the resulting drift across three inlined copies); both `housekeeping.yml` and `dependabot-automerge.yml` now load `ADMINS` from `board-config.env`. `admin-approval.yml` keeps its own inline copy with the documented latency-avoidance reasoning. diff --git a/clients/ts/src/client.ts b/clients/ts/src/client.ts index 4cd144ea..774f40a9 100644 --- a/clients/ts/src/client.ts +++ b/clients/ts/src/client.ts @@ -81,13 +81,12 @@ export class WaveHouseClient { return sql(this._ctx, query, params, opts); } - /** @internal Create a stream for the given table/topic. */ + /** @internal Create a stream for the given table. */ private _createStream>( table: string, opts?: StreamOptions, ): StreamController { const transportType = opts?.transport ?? this._config.transport ?? 'auto'; - const topic = `ingest.${table}`; // The Smart 'auto' Logic let useWS = transportType === 'ws'; @@ -126,7 +125,7 @@ export class WaveHouseClient { connect() { // Subscribe to the manager; forward events to the transport callbacks. const unsub = mgr.subscribe( - topic, + table, (event) => this.onEvent?.(event), (status) => this.onStatus?.(status), (error) => this.onError?.(error), @@ -154,7 +153,7 @@ export class WaveHouseClient { const transport = new SSETransport({ baseURL: this._ctx.baseURL, - topic, + table, since: opts?.since, }); const controller = new StreamController(transport); diff --git a/clients/ts/src/stream/sse.ts b/clients/ts/src/stream/sse.ts index 02bbe000..8f22bd19 100644 --- a/clients/ts/src/stream/sse.ts +++ b/clients/ts/src/stream/sse.ts @@ -3,7 +3,7 @@ import type { StreamTransport } from './controller.js'; export interface SSEOptions { baseURL: string; - topic: string; + table: string; since?: string; } @@ -33,7 +33,7 @@ export class SSETransport> implements StreamTranspor } const url = new URL('/v1/stream/sse', this._opts.baseURL); - url.searchParams.set('topic', this._opts.topic); + url.searchParams.set('table', this._opts.table); if (this._opts.since) { url.searchParams.set('since', this._opts.since); } diff --git a/clients/ts/src/stream/ws-manager.ts b/clients/ts/src/stream/ws-manager.ts index ad89d2d2..9780a4a9 100644 --- a/clients/ts/src/stream/ws-manager.ts +++ b/clients/ts/src/stream/ws-manager.ts @@ -14,14 +14,14 @@ interface Subscription { * Manages a single multiplexed WebSocket connection. * * Instead of one WebSocket per stream, all subscriptions share a single - * connection. Topics are subscribed/unsubscribed via in-band JSON commands: + * connection. Tables are subscribed/unsubscribed via in-band JSON commands: * - * {"action":"subscribe","topic":"ingest.clicks"} - * {"action":"unsubscribe","topic":"ingest.clicks"} + * {"action":"subscribe","table":"clicks"} + * {"action":"unsubscribe","table":"clicks"} * - * Incoming messages have a topic envelope: + * Incoming messages carry a table envelope: * - * {"topic":"ingest.clicks","data":{"table_name":"clicks",...}} + * {"table":"clicks","data":{"table_name":"clicks",...}} */ export class SharedWSManager { private _baseURL: string; @@ -40,25 +40,25 @@ export class SharedWSManager { } /** - * Subscribe to a topic. Opens the WebSocket if not already connected. + * Subscribe to a table. Opens the WebSocket if not already connected. * Returns an unsubscribe function. */ subscribe>( - topic: string, + table: string, callback: WSEventCallback, onStatus?: WSStatusCallback, onError?: WSErrorCallback, ): () => void { const sub: Subscription = { callback, onStatus, onError }; - let topicSubs = this._subs.get(topic); - const isNewTopic = !topicSubs || topicSubs.size === 0; + let tableSubs = this._subs.get(table); + const isNewTable = !tableSubs || tableSubs.size === 0; - if (!topicSubs) { - topicSubs = new Set(); - this._subs.set(topic, topicSubs); + if (!tableSubs) { + tableSubs = new Set(); + this._subs.set(table, tableSubs); } - topicSubs.add(sub); + tableSubs.add(sub); // Ensure connection is open. if (!this._ws && !this._closed) { @@ -69,16 +69,16 @@ export class SharedWSManager { onStatus?.(this._connected ? 'live' : 'connecting'); } - // Send subscribe command for new topics. - if (isNewTopic) { - this._send(JSON.stringify({ action: 'subscribe', topic })); + // Send subscribe command for new tables. + if (isNewTable) { + this._send(JSON.stringify({ action: 'subscribe', table })); } return () => { - topicSubs!.delete(sub); - if (topicSubs!.size === 0) { - this._subs.delete(topic); - this._send(JSON.stringify({ action: 'unsubscribe', topic })); + tableSubs!.delete(sub); + if (tableSubs!.size === 0) { + this._subs.delete(table); + this._send(JSON.stringify({ action: 'unsubscribe', table })); } // Close connection if no subscriptions remain. if (this._subs.size === 0) { @@ -137,23 +137,23 @@ export class SharedWSManager { } this._pendingCommands = []; - // Re-subscribe all active topics. - for (const topic of this._subs.keys()) { - this._ws?.send(JSON.stringify({ action: 'subscribe', topic })); + // Re-subscribe all active tables. + for (const table of this._subs.keys()) { + this._ws?.send(JSON.stringify({ action: 'subscribe', table })); } }; this._ws.onmessage = (e) => { try { const envelope = JSON.parse(e.data as string) as { - topic: string; + table: string; data: { table_name: string; received_timestamp: string; data: unknown; }; }; - if (!envelope.topic || !envelope.data) return; + if (!envelope.table || !envelope.data) return; const event: StreamEvent = { table: envelope.data.table_name, @@ -161,23 +161,12 @@ export class SharedWSManager { data: envelope.data.data as Record, }; - // Dispatch to exact topic subscribers. - const exact = this._subs.get(envelope.topic); - if (exact) { - for (const sub of exact) { + const subs = this._subs.get(envelope.table); + if (subs) { + for (const sub of subs) { sub.callback(event); } } - - // Dispatch to wildcard subscribers (e.g. "ingest.>"). - for (const [pattern, subs] of this._subs) { - if (pattern === envelope.topic) continue; // already handled - if (matchTopicPattern(pattern, envelope.topic)) { - for (const sub of subs) { - sub.callback(event); - } - } - } } catch { // ignore malformed messages } @@ -234,22 +223,3 @@ export class SharedWSManager { } } } - -/** - * Client-side NATS-style topic matching for dispatching messages. - * - `*` matches exactly one token - * - `>` as the last token matches one or more tokens - */ -function matchTopicPattern(pattern: string, subject: string): boolean { - const pTokens = pattern.split('.'); - const sTokens = subject.split('.'); - - for (let i = 0; i < pTokens.length; i++) { - if (pTokens[i] === '>') { - return i < sTokens.length; - } - if (i >= sTokens.length) return false; - if (pTokens[i] !== '*' && pTokens[i] !== sTokens[i]) return false; - } - return pTokens.length === sTokens.length; -} diff --git a/clients/ts/src/stream/ws.ts b/clients/ts/src/stream/ws.ts index f5c3cc0e..f9d63000 100644 --- a/clients/ts/src/stream/ws.ts +++ b/clients/ts/src/stream/ws.ts @@ -3,7 +3,7 @@ import type { StreamTransport } from './controller.js'; export interface WSOptions { baseURL: string; - topic: string; + table: string; since?: string; auth?: () => Promise | string; } @@ -48,7 +48,7 @@ export class WSTransport> implements StreamTransport private async _doConnect(): Promise { const wsBase = this._opts.baseURL.replace(/^http/, 'ws'); const url = new URL('/v1/stream/ws', wsBase); - url.searchParams.set('topic', this._opts.topic); + url.searchParams.set('table', this._opts.table); if (this._opts.since) { url.searchParams.set('since', this._opts.since); } diff --git a/docs/api.md b/docs/api.md index 6bec3636..2abbb855 100644 --- a/docs/api.md +++ b/docs/api.md @@ -304,7 +304,7 @@ Opens a persistent SSE connection for real-time event streaming. Supports histor | Param | Type | Default | Description | | ----- | ---- | ------- | ----------- | -| `topic` | string | `ingest.>` | NATS subject to subscribe to. Supports NATS wildcards: `*` matches one token, `>` matches one or more remaining tokens. | +| `table` | string | (required) | Table name to subscribe to. Returns 400 if missing. | | `since` | string | — | RFC 3339 timestamp. If provided, replays historical events from NATS before switching to live streaming. | | `token` | string | — | JWT token (alternative to `Authorization` header, useful for `EventSource`). Stripped from URL after extraction. | @@ -329,28 +329,25 @@ data: {"table_name":"page_views","received_timestamp":"2026-03-24T12:00:01.456Z" **curl example:** ```bash -# All tables -curl -N http://localhost:8080/v1/stream/sse - -# Specific table -curl -N "http://localhost:8080/v1/stream/sse?topic=ingest.clicks" +# Subscribe to a specific table +curl -N "http://localhost:8080/v1/stream/sse?table=clicks" # With gap-fill -curl -N "http://localhost:8080/v1/stream/sse?since=2026-03-24T11:00:00Z" +curl -N "http://localhost:8080/v1/stream/sse?table=clicks&since=2026-03-24T11:00:00Z" ``` --- ### `GET /v1/stream/ws` — WebSocket Stream -Opens a WebSocket connection for real-time event streaming. Supports in-band multiplexing — a single WebSocket can subscribe to multiple topics dynamically. +Opens a WebSocket connection for real-time event streaming. Supports in-band multiplexing — a single WebSocket can subscribe to multiple tables dynamically. **Query Parameters:** | Param | Type | Default | Description | | ----- | ---- | ------- | ----------- | -| `topic` | string | — | Optional initial topic to subscribe to (backward compatible). If omitted, the client must send subscribe commands. | -| `since` | string | — | RFC 3339 timestamp for gap-fill on the initial `?topic=` subscription. | +| `table` | string | — | Optional initial table to subscribe to. If omitted, the client must send subscribe commands. | +| `since` | string | — | RFC 3339 timestamp for gap-fill on the initial `?table=` subscription. | | `token` | string | — | JWT token (alternative to `Authorization` header). Stripped from URL after extraction. | **In-band commands (client → server):** @@ -358,17 +355,17 @@ Opens a WebSocket connection for real-time event streaming. Supports in-band mul After connecting, send JSON commands to manage subscriptions: ```json -{"action": "subscribe", "topic": "ingest.clicks"} -{"action": "subscribe", "topic": "ingest.page_views"} -{"action": "unsubscribe", "topic": "ingest.clicks"} +{"action": "subscribe", "table": "clicks"} +{"action": "subscribe", "table": "page_views"} +{"action": "unsubscribe", "table": "clicks"} ``` **Outbound message format (server → client):** -Each message is wrapped in an envelope with the topic: +Each message is wrapped in an envelope labelled with the table name: ```json -{"topic": "ingest.clicks", "data": {"table_name": "clicks", "received_timestamp": "...", "data": {...}}} +{"table": "clicks", "data": {"table_name": "clicks", "received_timestamp": "...", "data": {...}}} ``` **JavaScript example:** @@ -376,12 +373,12 @@ Each message is wrapped in an envelope with the topic: ```javascript const ws = new WebSocket("ws://localhost:8080/v1/stream/ws?token="); ws.onopen = () => { - ws.send(JSON.stringify({ action: "subscribe", topic: "ingest.clicks" })); - ws.send(JSON.stringify({ action: "subscribe", topic: "ingest.page_views" })); + ws.send(JSON.stringify({ action: "subscribe", table: "clicks" })); + ws.send(JSON.stringify({ action: "subscribe", table: "page_views" })); }; ws.onmessage = (event) => { - const { topic, data } = JSON.parse(event.data); - console.log(`[${topic}]`, data.table_name, data.data); + const { table, data } = JSON.parse(event.data); + console.log(`[${table}]`, data.table_name, data.data); }; ``` diff --git a/docs/architecture.md b/docs/architecture.md index 9d62a1c1..79479f95 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -90,7 +90,7 @@ The API layer uses [Chi](https://github.com/go-chi/chi) for routing with standar - **structured_query.go** — Handler for `POST /v1/tables/{table}/query`: validates query AST, enforces permissions, builds and executes SQL. - **ingest.go** — Accepts flat JSON body for `POST /v1/ingest/{table}`, validates against discovered schema, optional dedup, publishes to NATS subject `ingest.{table}`. - **query.go** — Executes SQL queries directly against ClickHouse. Results are cached. UUID/DateTime columns are converted to strings. -- **stream_sse.go** / **stream_ws.go** — Real-time streaming via SSE and WebSocket. Default topic is `ingest.>` (all tables). Supports gap-fill from NATS JetStream using `DeliverByStartTime`. +- **stream_sse.go** / **stream_ws.go** — Real-time streaming via SSE and WebSocket. Callers select a table with the `?table=` query parameter (required for SSE); WS additionally accepts in-band `{"action":"subscribe","table":"..."}` commands. Supports gap-fill from NATS JetStream using `DeliverByStartTime`. - **transform.go** — Shared `transformForClient` function: passes through `table_name`, `received_timestamp`, and `data` from the wire format. - **schema.go** — Schema discovery API: list all schemas, get one table, trigger refresh. - **dlq.go** — DLQ stats endpoint and `EnsureDLQStream` helper for creating the `WAVEHOUSE_DLQ` NATS stream. diff --git a/docs/development.md b/docs/development.md index bee31019..32004278 100644 --- a/docs/development.md +++ b/docs/development.md @@ -89,14 +89,11 @@ curl -s -X POST http://localhost:8080/v1/query \ -H "Content-Type: application/json" \ -d '{"sql": "SELECT * FROM clicks LIMIT 10"}' -# Open an SSE stream for all tables (Ctrl+C to stop) -curl -N http://localhost:8080/v1/stream/sse - -# Open an SSE stream for a specific table -curl -N "http://localhost:8080/v1/stream/sse?topic=ingest.clicks" +# Open an SSE stream for a specific table (Ctrl+C to stop) +curl -N "http://localhost:8080/v1/stream/sse?table=clicks" # With gap-fill (replays events since the given timestamp, then switches to live) -curl -N "http://localhost:8080/v1/stream/sse?since=2026-03-24T11:00:00Z" +curl -N "http://localhost:8080/v1/stream/sse?table=clicks&since=2026-03-24T11:00:00Z" # Health check (no auth required) curl http://localhost:8080/health diff --git a/internal/api/hub.go b/internal/api/hub.go index 26e94155..0e6ce960 100644 --- a/internal/api/hub.go +++ b/internal/api/hub.go @@ -3,7 +3,6 @@ package api import ( "context" "encoding/json" - "strings" "sync" "github.com/Wave-RF/WaveHouse/internal/mq" @@ -84,59 +83,10 @@ func (h *Hub) Broadcast(topic string, msg *mq.Message) { h.mu.RLock() defer h.mu.RUnlock() - sent := make(map[chan []byte]struct{}) - - // Exact match. Only mark a channel as "sent" after the send actually - // succeeds — if the channel is full and we hit the default case, the - // wildcard loop below gets a second chance. for ch := range h.subscribers[topic] { select { case ch <- data: - sent[ch] = struct{}{} default: } } - - // Wildcard match: iterate all subscriber patterns. - for pattern, chs := range h.subscribers { - if pattern == topic { - continue // already handled above - } - if !matchTopic(pattern, topic) { - continue - } - for ch := range chs { - if _, dup := sent[ch]; dup { - continue - } - select { - case ch <- data: - sent[ch] = struct{}{} - default: - } - } - } -} - -// matchTopic checks whether a NATS-style pattern matches a subject. -// Tokens are separated by ".". -// - "*" matches exactly one token -// - ">" as the last pattern token matches one or more remaining tokens -func matchTopic(pattern, subject string) bool { - pTokens := strings.Split(pattern, ".") - sTokens := strings.Split(subject, ".") - - for i, pt := range pTokens { - if pt == ">" { - // ">" must be the last token and matches 1+ remaining subject tokens. - return i < len(sTokens) - } - if i >= len(sTokens) { - return false - } - if pt != "*" && pt != sTokens[i] { - return false - } - } - return len(pTokens) == len(sTokens) } diff --git a/internal/api/hub_test.go b/internal/api/hub_test.go index 5519276c..79babd30 100644 --- a/internal/api/hub_test.go +++ b/internal/api/hub_test.go @@ -208,204 +208,3 @@ func TestHub_UnsubscribeCleansEmptyTopic(t *testing.T) { hub.mu.RUnlock() assert.False(t, exists, "topic should be removed when last subscriber leaves") } - -func TestHub_WildcardGreaterThan(t *testing.T) { - t.Parallel() - hub := NewHub() - - // Subscribe with "ingest.>" wildcard. - ch := make(chan []byte, 10) - hub.Subscribe("ingest.>", ch) - defer hub.Unsubscribe("ingest.>", ch) - - hub.Broadcast("ingest.clicks", &mq.Message{ - Ctx: context.Background(), - Subject: "ingest.clicks", - Data: []byte(`{"table_name":"clicks"}`), - }) - - select { - case msg := <-ch: - payload := unwrapTestMessage(t, msg) - assert.Contains(t, string(payload), "clicks") - case <-time.After(time.Second): - t.Fatal("wildcard subscriber should have received ingest.clicks") - } -} - -func TestHub_WildcardStar(t *testing.T) { - t.Parallel() - hub := NewHub() - - ch := make(chan []byte, 10) - hub.Subscribe("ingest.*", ch) - defer hub.Unsubscribe("ingest.*", ch) - - hub.Broadcast("ingest.clicks", &mq.Message{ - Ctx: context.Background(), - Subject: "ingest.clicks", - Data: []byte(`{"table_name":"clicks"}`), - }) - - select { - case msg := <-ch: - payload := unwrapTestMessage(t, msg) - assert.Contains(t, string(payload), "clicks") - case <-time.After(time.Second): - t.Fatal("star wildcard subscriber should have received") - } -} - -func TestHub_WildcardStarNoMultiToken(t *testing.T) { - t.Parallel() - hub := NewHub() - - ch := make(chan []byte, 10) - hub.Subscribe("ingest.*", ch) - defer hub.Unsubscribe("ingest.*", ch) - - // "ingest.*" should NOT match "ingest.clicks.subpath" (star = one token). - hub.Broadcast("ingest.clicks.subpath", &mq.Message{ - Ctx: context.Background(), - Subject: "ingest.clicks.subpath", - Data: []byte(`{"table_name":"clicks"}`), - }) - - select { - case <-ch: - t.Fatal("star wildcard should NOT match multi-token subject") - case <-time.After(50 * time.Millisecond): - // expected — no message - } -} - -func TestHub_WildcardGreaterThanMultiToken(t *testing.T) { - t.Parallel() - hub := NewHub() - - ch := make(chan []byte, 10) - hub.Subscribe("ingest.>", ch) - defer hub.Unsubscribe("ingest.>", ch) - - // "ingest.>" should match multi-token subjects. - hub.Broadcast("ingest.clicks.subpath", &mq.Message{ - Ctx: context.Background(), - Subject: "ingest.clicks.subpath", - Data: []byte(`{"table_name":"clicks"}`), - }) - - select { - case msg := <-ch: - payload := unwrapTestMessage(t, msg) - assert.Contains(t, string(payload), "clicks") - case <-time.After(time.Second): - t.Fatal("> wildcard should match multi-token subjects") - } -} - -func TestHub_WildcardDoesNotMatchExact(t *testing.T) { - t.Parallel() - hub := NewHub() - - ch := make(chan []byte, 10) - hub.Subscribe("ingest.>", ch) - defer hub.Unsubscribe("ingest.>", ch) - - // "ingest.>" should NOT match "ingest" alone (> requires 1+ tokens after). - hub.Broadcast("ingest", &mq.Message{ - Ctx: context.Background(), - Subject: "ingest", - Data: []byte(`{"table_name":"ingest"}`), - }) - - select { - case <-ch: - t.Fatal("> should not match bare prefix") - case <-time.After(50 * time.Millisecond): - // good - } -} - -func TestHub_BareGreaterThanMatchesAll(t *testing.T) { - t.Parallel() - hub := NewHub() - - ch := make(chan []byte, 10) - hub.Subscribe(">", ch) - defer hub.Unsubscribe(">", ch) - - hub.Broadcast("ingest.clicks", &mq.Message{ - Ctx: context.Background(), - Subject: "ingest.clicks", - Data: []byte(`{"table_name":"clicks"}`), - }) - - select { - case <-ch: - // expected - case <-time.After(time.Second): - t.Fatal("bare > should match everything") - } -} - -func TestHub_WildcardNoDuplicateDelivery(t *testing.T) { - t.Parallel() - hub := NewHub() - - ch := make(chan []byte, 10) - // Subscribe with both exact and wildcard that would match. - hub.Subscribe("ingest.clicks", ch) - hub.Subscribe("ingest.>", ch) - defer hub.Unsubscribe("ingest.clicks", ch) - defer hub.Unsubscribe("ingest.>", ch) - - hub.Broadcast("ingest.clicks", &mq.Message{ - Ctx: context.Background(), - Subject: "ingest.clicks", - Data: []byte(`{"table_name":"clicks"}`), - }) - - // Should receive exactly one message, not two. - select { - case <-ch: - // got first - case <-time.After(time.Second): - t.Fatal("should have received at least one message") - } - - select { - case <-ch: - t.Fatal("should NOT receive a duplicate") - case <-time.After(50 * time.Millisecond): - // good - } -} - -func TestMatchTopic(t *testing.T) { - t.Parallel() - tests := []struct { - pattern string - subject string - want bool - }{ - {"ingest.clicks", "ingest.clicks", true}, - {"ingest.clicks", "ingest.users", false}, - {"ingest.*", "ingest.clicks", true}, - {"ingest.*", "ingest.clicks.sub", false}, - {"ingest.>", "ingest.clicks", true}, - {"ingest.>", "ingest.clicks.sub", true}, - {"ingest.>", "ingest", false}, - {">", "anything", true}, - {">", "a.b.c", true}, - {"*.*", "ingest.clicks", true}, - {"*.*", "ingest", false}, - {"a.*.c", "a.b.c", true}, - {"a.*.c", "a.b.d", false}, - } - for _, tt := range tests { - t.Run(tt.pattern+"_vs_"+tt.subject, func(t *testing.T) { - t.Parallel() - assert.Equal(t, tt.want, matchTopic(tt.pattern, tt.subject)) - }) - } -} diff --git a/internal/api/stream_sse.go b/internal/api/stream_sse.go index 60f2d9b3..581a82a9 100644 --- a/internal/api/stream_sse.go +++ b/internal/api/stream_sse.go @@ -34,10 +34,12 @@ func (h *SSEHandler) Handle(w http.ResponseWriter, r *http.Request) { return } - topic := r.URL.Query().Get("topic") - if topic == "" { - topic = "ingest.>" + table := r.URL.Query().Get("table") + if table == "" { + writeJSONError(w, http.StatusBadRequest, "missing required query parameter: table") + return } + topic := "ingest." + table // Resolve stream permissions for this request. role := RoleFromContext(r.Context()) diff --git a/internal/api/stream_test.go b/internal/api/stream_test.go index 141c8e5f..8bad3a8c 100644 --- a/internal/api/stream_test.go +++ b/internal/api/stream_test.go @@ -138,13 +138,13 @@ func TestWS_ApplyStreamPolicy_FiltersColumns(t *testing.T) { } raw, _ := json.Marshal(evt) - out := h.applyStreamPolicy(raw, "user", nil, "ingest.events") + out := h.applyStreamPolicy(raw, "user", nil, "events") require.NotNil(t, out) var got map[string]any require.NoError(t, json.Unmarshal(out, &got)) - // WS wraps in topic envelope. - assert.Equal(t, "ingest.events", got["topic"]) + // WS wraps in table envelope. + assert.Equal(t, "events", got["table"]) inner := got["data"].(map[string]any) data := inner["data"].(map[string]any) assert.Equal(t, "click", data["name"]) @@ -162,13 +162,13 @@ func TestWS_ApplyStreamPolicy_NoPolicy(t *testing.T) { } raw, _ := json.Marshal(evt) - out := h.applyStreamPolicy(raw, "", nil, "ingest.clicks") + out := h.applyStreamPolicy(raw, "", nil, "clicks") require.NotNil(t, out) var got map[string]any require.NoError(t, json.Unmarshal(out, &got)) - // WS wraps in topic envelope. - assert.Equal(t, "ingest.clicks", got["topic"]) + // WS wraps in table envelope. + assert.Equal(t, "clicks", got["table"]) inner := got["data"].(map[string]any) assert.Equal(t, "clicks", inner["table_name"]) } diff --git a/internal/api/stream_ws.go b/internal/api/stream_ws.go index 2dcd0767..6c2d9e7d 100644 --- a/internal/api/stream_ws.go +++ b/internal/api/stream_ws.go @@ -22,16 +22,16 @@ import ( // WSHandler handles GET /v1/stream/ws. // Supports multiplexed subscriptions via in-band JSON commands: // -// {"action":"subscribe","topic":"ingest.clicks"} -// {"action":"unsubscribe","topic":"ingest.clicks"} +// {"action":"subscribe","table":"clicks"} +// {"action":"unsubscribe","table":"clicks"} // -// Outbound messages are wrapped in a topic envelope: +// Outbound messages are wrapped in a table envelope: // -// {"topic":"ingest.clicks","data":{...event...}} +// {"table":"clicks","data":{...event...}} // -// For backward compatibility, the ?topic= query parameter auto-subscribes -// on connect. If no ?topic= is set, the connection starts with no -// subscriptions and waits for in-band subscribe commands. +// The optional ?table= query parameter auto-subscribes on connect. If no +// ?table= is set, the connection starts with no subscriptions and waits +// for in-band subscribe commands. type WSHandler struct { Hub *Hub JS jetstream.JetStream @@ -46,7 +46,7 @@ func NewWSHandler(hub *Hub, js jetstream.JetStream, allowedOrigins []string) *WS // wsCommand represents an in-band subscribe/unsubscribe command. type wsCommand struct { Action string `json:"action"` - Topic string `json:"topic"` + Table string `json:"table"` } func (h *WSHandler) Handle(w http.ResponseWriter, r *http.Request) { @@ -71,21 +71,21 @@ func (h *WSHandler) Handle(w http.ResponseWriter, r *http.Request) { // Merged channel receives messages from all subscribed topics. merged := make(chan []byte, 64) - // Track active topic subscriptions and their per-topic channels. + // Track active subscriptions by table name, with their per-table channels. var mu sync.Mutex - subs := make(map[string]chan []byte) // topic → per-topic channel + subs := make(map[string]chan []byte) // table → per-table channel - subscribeTopic := func(topic string) { + subscribeTable := func(table string) { mu.Lock() defer mu.Unlock() - if _, exists := subs[topic]; exists { + if _, exists := subs[table]; exists { return } ch := make(chan []byte, 64) - subs[topic] = ch - h.Hub.Subscribe(topic, ch) + subs[table] = ch + h.Hub.Subscribe("ingest."+table, ch) - // Pump per-topic channel into merged channel + // Pump per-table channel into merged channel go func() { for msg := range ch { select { @@ -96,40 +96,40 @@ func (h *WSHandler) Handle(w http.ResponseWriter, r *http.Request) { }() } - unsubscribeTopic := func(topic string) { + unsubscribeTable := func(table string) { mu.Lock() - ch, exists := subs[topic] + ch, exists := subs[table] if !exists { mu.Unlock() return } - delete(subs, topic) + delete(subs, table) mu.Unlock() - h.Hub.Unsubscribe(topic, ch) // closes ch, which stops the pump goroutine + h.Hub.Unsubscribe("ingest."+table, ch) // closes ch, which stops the pump goroutine } unsubscribeAll := func() { mu.Lock() - topics := make([]string, 0, len(subs)) + tables := make([]string, 0, len(subs)) for t := range subs { - topics = append(topics, t) + tables = append(tables, t) } mu.Unlock() - for _, t := range topics { - unsubscribeTopic(t) + for _, t := range tables { + unsubscribeTable(t) } } defer unsubscribeAll() - // Backward compat: auto-subscribe if ?topic= is set. - if topic := r.URL.Query().Get("topic"); topic != "" { - subscribeTopic(topic) + // Auto-subscribe if ?table= is set. + if table := r.URL.Query().Get("table"); table != "" { + subscribeTable(table) // Gap fill from NATS. if since := r.URL.Query().Get("since"); since != "" { if ts, parseErr := time.Parse(time.RFC3339, since); parseErr == nil && h.JS != nil { - h.replayFromNATS(ctx, ts, topic, func(data []byte) bool { - out := h.applyStreamPolicy(data, role, map[string]any(claims), topic) + h.replayFromNATS(ctx, ts, "ingest."+table, func(data []byte) bool { + out := h.applyStreamPolicy(data, role, map[string]any(claims), table) if out == nil { return true } @@ -147,14 +147,14 @@ func (h *WSHandler) Handle(w http.ResponseWriter, r *http.Request) { return } var cmd wsCommand - if json.Unmarshal(data, &cmd) != nil || cmd.Topic == "" { + if json.Unmarshal(data, &cmd) != nil || cmd.Table == "" { continue } switch cmd.Action { case "subscribe": - subscribeTopic(cmd.Topic) + subscribeTable(cmd.Table) case "unsubscribe": - unsubscribeTopic(cmd.Topic) + unsubscribeTable(cmd.Table) } } }() @@ -166,7 +166,7 @@ func (h *WSHandler) Handle(w http.ResponseWriter, r *http.Request) { select { case <-ctx.Done(): return - // Determine the event's table_name to use as the topic in the envelope. + // Determine the event's table_name to label the envelope. case data := <-merged: var envelope struct { TraceHeaders map[string]string `json:"trace_headers"` @@ -179,9 +179,9 @@ func (h *WSHandler) Handle(w http.ResponseWriter, r *http.Request) { var rawEvt struct { TableName string `json:"table_name"` } - evtTopic := "" - if json.Unmarshal(envelope.Payload, &rawEvt) == nil && rawEvt.TableName != "" { - evtTopic = "ingest." + rawEvt.TableName + evtTable := "" + if json.Unmarshal(envelope.Payload, &rawEvt) == nil { + evtTable = rawEvt.TableName } parentCtx := otel.GetTextMapPropagator().Extract( @@ -191,7 +191,7 @@ func (h *WSHandler) Handle(w http.ResponseWriter, r *http.Request) { _, pushSpan := tracer.Start(parentCtx, "WS.PushEvent") - out := h.applyStreamPolicy(envelope.Payload, role, map[string]any(claims), evtTopic) + out := h.applyStreamPolicy(envelope.Payload, role, map[string]any(claims), evtTable) if out == nil { pushSpan.End() continue @@ -208,14 +208,14 @@ func (h *WSHandler) Handle(w http.ResponseWriter, r *http.Request) { // applyStreamPolicy transforms raw event data for the client, filtering columns // based on the caller's policy permissions. Returns nil if the event should be skipped. -// The result is wrapped in a topic envelope: {"topic":"...","data":{...}}. -func (h *WSHandler) applyStreamPolicy(raw []byte, role string, claims map[string]any, topic string) []byte { +// The result is wrapped in a table envelope: {"table":"...","data":{...}}. +func (h *WSHandler) applyStreamPolicy(raw []byte, role string, claims map[string]any, table string) []byte { var evt ingest.EventMessage if err := json.Unmarshal(raw, &evt); err != nil || evt.TableName == "" { if !json.Valid(raw) { return nil } - envelope := map[string]any{"topic": topic, "data": json.RawMessage(raw)} + envelope := map[string]any{"table": table, "data": json.RawMessage(raw)} data, err := json.Marshal(envelope) if err != nil { return nil @@ -238,7 +238,7 @@ func (h *WSHandler) applyStreamPolicy(raw []byte, role string, claims map[string "data": evt.Data, } envelope := map[string]any{ - "topic": "ingest." + evt.TableName, + "table": evt.TableName, "data": inner, } data, err := json.Marshal(envelope) From c9a8da6546e1293e183e2c2c0089a97d3dd09883 Mon Sep 17 00:00:00 2001 From: Eric Andrechek Date: Tue, 12 May 2026 18:47:28 -0400 Subject: [PATCH 2/5] fix(api): validate ?table= and cmd.table to block NATS wildcard injection MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses two MUST findings from Claude review + Gemini review on PR #124. Root cause: the `?table=` value lands in NATS `FilterSubject` via `replayFromNATS` on the SSE/WS gap-fill path. NATS subjects honour `*` and `>` wildcards, so `?table=>&since=2020-01-01T00:00:00Z` builds `FilterSubject: "ingest.>"` and replays *every* `ingest.*` message — defeating the whole point of removing wildcards from the Hub. Fix: - New package-level `validTableNameRe` matching `^[a-zA-Z_][a-zA-Z0-9_]*$`, the same safe-identifier shape that `internal/ingest.safeIdentifierRe` and `internal/query.validIdentifierRe` already enforce for SQL identifiers. The dot/`*`/`>` exclusion is the load-bearing part for this security fix. - SSE: 400 when `?table=` is missing or fails the regex. - WS: 400 (pre-`websocket.Accept`) when `?table=` is set but fails the regex. Empty `?table=` is still allowed (in-band subscribe path). - WS in-band `cmd.Table`: silently drop wildcard / malformed table names. Hub uses exact-match so the impact today is nil — defensive consistency only, per Gemini's flag. Other fixes from Claude review: - `README.md` quick-start `curl -N .../stream/sse` updated to `?table=clicks` (the bare URL now returns 400). - `docs/api.md` SSE response example previously showed events from both `clicks` and `page_views` over the same connection — that's no longer possible since SSE is one-table-per-connection. Replaced the second event with another `clicks` row and added a one-liner pointing multi-table consumers at the WebSocket endpoint. Tests: - `TestSSE_RejectsMissingOrInvalidTable` covers missing, `>`, `*`, `ingest.>`, `clicks.subpath`, leading-digit, space-bearing inputs. - `TestSSE_AcceptsSafeTableName` confirms the validation gate doesn't reject the canonical case (cancelled-context trick exits the live loop after validation). - `TestWS_RejectsInvalidTableOnQuery` covers the WS `?table=` path. - `TestValidTableNameRe` is a focused regex truth table. Verification: `make verify` clean; `make test-unit` 454 pass (was 424), unit coverage 74.1% (was 72.9%); `make test-integration` 3 pass; `make test-e2e` 30 pass; merged coverage gate 81.9% (was 80.8%). Co-Authored-By: Claude Opus 4.7 (1M context) --- CHANGELOG.md | 2 +- README.md | 4 +- docs/api.md | 4 +- internal/api/stream_sse.go | 12 ++++ internal/api/stream_test.go | 106 ++++++++++++++++++++++++++++++++++++ internal/api/stream_ws.go | 16 ++++++ 6 files changed, 140 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 75b4be5c..c0ba03cd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,7 +10,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - **Hub wildcard subscriptions** (`internal/api/hub.go`, `internal/api/hub_test.go`): dropped the NATS-style `*` / `>` pattern matching from `Hub.Broadcast`, the wildcard pattern loop, the `sent` dedup map, the `matchTopic` helper, and the eight wildcard tests (plus `TestMatchTopic`). After the #89 MVP cuts every producer publishes a concrete `ingest.
` subject and the SDK only ever subscribes to one concrete subject, so the wildcard fan-out was unused machinery. Closes #100 (part of #87). Net −210 lines (mostly tests). ### Changed -- **BREAKING:** SSE/WS streaming endpoints replaced `?topic=` with `?table=` (issue #100, option B). The NATS subject convention no longer leaks into the public HTTP API; the server builds `ingest.
` internally. SSE returns `400 Bad Request` when `?table=` is missing (was: defaulted to `ingest.>`). WS `?table=` is optional — clients can still defer to in-band subscribe commands. +- **BREAKING:** SSE/WS streaming endpoints replaced `?topic=` with `?table=` (issue #100, option B). The NATS subject convention no longer leaks into the public HTTP API; the server builds `ingest.
` internally. SSE returns `400 Bad Request` when `?table=` is missing or fails the `^[a-zA-Z_][a-zA-Z0-9_]*$` safe-identifier check (the same regex `internal/ingest` and `internal/query` use for SQL identifiers — crucially this rejects NATS wildcard characters `*` and `>` before they reach the gap-fill `FilterSubject`). WS `?table=` is optional — clients can still defer to in-band subscribe commands — but is validated by the same regex when present. - **BREAKING:** WebSocket in-band subscribe/unsubscribe commands use `"table"` (raw name) instead of `"topic"` (NATS subject): - Before: `{"action":"subscribe","topic":"ingest.clicks"}` - After: `{"action":"subscribe","table":"clicks"}` diff --git a/README.md b/README.md index 6d6429b6..f1213917 100644 --- a/README.md +++ b/README.md @@ -68,8 +68,8 @@ curl -s -X POST http://localhost:8080/v1/query \ -H "Content-Type: application/json" \ -d '{"sql": "SELECT * FROM clicks LIMIT 10"}' -# Open a real-time SSE stream (Ctrl+C to stop) -curl -N http://localhost:8080/v1/stream/sse +# Open a real-time SSE stream for a specific table (Ctrl+C to stop) +curl -N "http://localhost:8080/v1/stream/sse?table=clicks" ``` WaveHouse is now accepting API requests on `http://localhost:8080`. diff --git a/docs/api.md b/docs/api.md index 2abbb855..30f07377 100644 --- a/docs/api.md +++ b/docs/api.md @@ -321,9 +321,11 @@ id: 2026-03-24T12:00:00.123Z data: {"table_name":"clicks","received_timestamp":"2026-03-24T12:00:00.123Z","data":{"page":"/home","button":"signup"}} id: 2026-03-24T12:00:01.456Z -data: {"table_name":"page_views","received_timestamp":"2026-03-24T12:00:01.456Z","data":{"url":"/dashboard"}} +data: {"table_name":"clicks","received_timestamp":"2026-03-24T12:00:01.456Z","data":{"page":"/pricing"}} ``` +Each SSE connection is bound to a single `?table=`; to consume multiple tables, open one connection per table or use the WebSocket endpoint with in-band multiplexing. + **Note:** When access control policies are active, streamed events are filtered per the caller's role — denied columns are removed and tables without select permission are skipped. **curl example:** diff --git a/internal/api/stream_sse.go b/internal/api/stream_sse.go index 581a82a9..ed993009 100644 --- a/internal/api/stream_sse.go +++ b/internal/api/stream_sse.go @@ -5,6 +5,7 @@ import ( "encoding/json" "fmt" "net/http" + "regexp" "time" "github.com/Wave-RF/WaveHouse/internal/ingest" @@ -16,6 +17,13 @@ import ( "go.opentelemetry.io/otel/propagation" ) +// validTableNameRe matches safe table identifiers and — critically — rejects +// the NATS subject wildcards `*` and `>`. The `?table=` value is concatenated +// into a NATS FilterSubject in the gap-fill path; without this guard, +// `?table=>` would build `ingest.>` and replay every ingest subject. +// Matches the same shape as ingest.safeIdentifierRe / query.validIdentifierRe. +var validTableNameRe = regexp.MustCompile(`^[a-zA-Z_][a-zA-Z0-9_]*$`) + // SSEHandler handles GET /v1/stream/sse. type SSEHandler struct { Hub *Hub @@ -39,6 +47,10 @@ func (h *SSEHandler) Handle(w http.ResponseWriter, r *http.Request) { writeJSONError(w, http.StatusBadRequest, "missing required query parameter: table") return } + if !validTableNameRe.MatchString(table) { + writeJSONError(w, http.StatusBadRequest, "invalid table name") + return + } topic := "ingest." + table // Resolve stream permissions for this request. diff --git a/internal/api/stream_test.go b/internal/api/stream_test.go index 8bad3a8c..57e1e08b 100644 --- a/internal/api/stream_test.go +++ b/internal/api/stream_test.go @@ -1,7 +1,11 @@ package api import ( + "context" "encoding/json" + "net/http" + "net/http/httptest" + "net/url" "testing" "time" @@ -173,6 +177,108 @@ func TestWS_ApplyStreamPolicy_NoPolicy(t *testing.T) { assert.Equal(t, "clicks", inner["table_name"]) } +// TestSSE_RejectsMissingOrInvalidTable verifies that the SSE handler returns +// 400 when the ?table= parameter is missing or contains characters that could +// be interpreted as NATS wildcards (regression guard for the fix that landed +// alongside the wildcard fan-out removal in #100). +func TestSSE_RejectsMissingOrInvalidTable(t *testing.T) { + t.Parallel() + h := &SSEHandler{Hub: NewHub()} + + cases := []struct { + name string + table string + }{ + {"missing", ""}, + {"nats greater wildcard", ">"}, + {"nats star wildcard", "*"}, + {"dot separator", "ingest.clicks"}, + {"nested wildcard", "ingest.>"}, + {"trailing wildcard", "clicks.>"}, + {"space", "click s"}, + {"leading digit", "1clicks"}, + {"empty after url decode", ""}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + target := "/v1/stream/sse" + if tc.table != "" { + target += "?table=" + url.QueryEscape(tc.table) + } + req := httptest.NewRequestWithContext(context.Background(), http.MethodGet, target, nil) + w := httptest.NewRecorder() + h.Handle(w, req) + assert.Equal(t, http.StatusBadRequest, w.Code) + assert.Equal(t, "application/json", w.Header().Get("Content-Type")) + }) + } +} + +func TestSSE_AcceptsSafeTableName(t *testing.T) { + t.Parallel() + h := &SSEHandler{Hub: NewHub()} + + // Use a request context that's already cancelled so the handler exits + // the live-stream select loop immediately instead of blocking the test. + ctx, cancel := context.WithCancel(context.Background()) + cancel() + req := httptest.NewRequestWithContext(ctx, http.MethodGet, "/v1/stream/sse?table=clicks", nil) + w := httptest.NewRecorder() + h.Handle(w, req) + // Past the validation gate — header set to text/event-stream, not the + // 400-path application/json. + assert.Equal(t, http.StatusOK, w.Code) + assert.Equal(t, "text/event-stream", w.Header().Get("Content-Type")) +} + +func TestWS_RejectsInvalidTableOnQuery(t *testing.T) { + t.Parallel() + h := &WSHandler{Hub: NewHub()} + + cases := []string{">", "*", "ingest.>", "clicks ", "clicks.subpath"} + for _, tbl := range cases { + t.Run(tbl, func(t *testing.T) { + t.Parallel() + target := "/v1/stream/ws?table=" + url.QueryEscape(tbl) + req := httptest.NewRequestWithContext(context.Background(), http.MethodGet, target, nil) + w := httptest.NewRecorder() + h.Handle(w, req) + // Validation runs before websocket.Accept, so we get a plain 400 + // rather than an upgrade-attempt error. + assert.Equal(t, http.StatusBadRequest, w.Code) + assert.Equal(t, "application/json", w.Header().Get("Content-Type")) + }) + } +} + +func TestValidTableNameRe(t *testing.T) { + t.Parallel() + cases := []struct { + in string + want bool + }{ + {"clicks", true}, + {"page_views", true}, + {"_internal", true}, + {"events_v2", true}, + {">", false}, + {"*", false}, + {"ingest.>", false}, + {"clicks.subpath", false}, + {"clicks ", false}, + {"clicks*", false}, + {"1clicks", false}, + {"", false}, + } + for _, tc := range cases { + t.Run(tc.in, func(t *testing.T) { + t.Parallel() + assert.Equal(t, tc.want, validTableNameRe.MatchString(tc.in)) + }) + } +} + func TestExtractEventTimestamp(t *testing.T) { t.Parallel() tests := []struct { diff --git a/internal/api/stream_ws.go b/internal/api/stream_ws.go index 6c2d9e7d..f19849ff 100644 --- a/internal/api/stream_ws.go +++ b/internal/api/stream_ws.go @@ -50,6 +50,16 @@ type wsCommand struct { } func (h *WSHandler) Handle(w http.ResponseWriter, r *http.Request) { + // Validate ?table= before upgrading. Empty is OK — the client can + // still subscribe via in-band commands after connect. Reject NATS + // wildcards / unsafe chars here so the gap-fill path that builds + // FilterSubject: "ingest."+table can't be tricked into a wildcard + // consumer. + if t := r.URL.Query().Get("table"); t != "" && !validTableNameRe.MatchString(t) { + writeJSONError(w, http.StatusBadRequest, "invalid table name") + return + } + origins := h.AllowedOrigins if len(origins) == 0 { origins = []string{"*"} @@ -150,6 +160,12 @@ func (h *WSHandler) Handle(w http.ResponseWriter, r *http.Request) { if json.Unmarshal(data, &cmd) != nil || cmd.Table == "" { continue } + // Hub lookups are exact-match so wildcards here are inert, but + // reject them anyway for consistency with the ?table= path and + // to keep the contract crisp. + if !validTableNameRe.MatchString(cmd.Table) { + continue + } switch cmd.Action { case "subscribe": subscribeTable(cmd.Table) From 1aa7ea7fe1e4ff6e64b39bf61aa89f7e4ef7533c Mon Sep 17 00:00:00 2001 From: Eric Andrechek Date: Wed, 13 May 2026 16:30:50 -0400 Subject: [PATCH 3/5] chore(api): WS RFC3339Nano + table-tagged dispatch; doc-sync from #7 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses two [SHOULD]s from Gemini's re-review of #124 plus the doc-sync gaps that surfaced after merging origin/main (#7 docs site). * `internal/api/stream_ws.go`: - `?since=` gap-fill now parses with `time.RFC3339Nano` instead of plain `time.RFC3339`. Go's `RFC3339Nano` layout treats the fractional seconds as optional, so this is a strict superset (plain `RFC3339` still parses). Matches the SSE handler's existing acceptance of high-precision client timestamps. - The merged channel now carries `{table, data}` (new `wsOutbound` type) instead of raw `[]byte`. The per-table pump goroutine tags each message with the table it was subscribed under, so the write loop no longer needs to re-unmarshal the payload just to extract `table_name` for the envelope. Also fixes the edge case Gemini flagged: when the payload isn't a standard `ingest.EventMessage` (raw-JSON pass-through), the outbound envelope previously fell back to `"table": ""`. It now keeps the subscription's table name so clients always know which subscription a message belongs to. * Doc-sync from the origin/main merge: - `docs/src/content/docs/getting-started.md` quick-start: the `?topic=ingest.clicks` curl example would 400 against the new contract. Replaced with `?table=clicks` (and the previously-shown "All tables" form, which had no `?table=`, is now gone since that's a 400 too). - `docs/src/content/docs/sdk.md` `SharedWSManager` section: said "topic subscriptions" + "NATS-style wildcard matching … on the client side." Updated to per-table subscriptions and a note that the legacy `*` / `>` wildcards are no longer accepted server-side. CHANGELOG `[Unreleased]` updated to call out both server-side changes plus the doc-sync. Local gates: - `make verify` clean (tidy + fmt + vulncheck + lint) - `make test-unit` 500 pass, 74.6% - `make test-sdk` 54.6% - `make test-integration` 20.1% - `make test-e2e` 30 pass + 1 skip, 51.4% - `make cov` merged 81.5% Co-Authored-By: Claude Opus 4.7 (1M context) --- CHANGELOG.md | 3 ++ docs/src/content/docs/getting-started.md | 11 +++---- docs/src/content/docs/sdk.md | 8 ++--- internal/api/stream_ws.go | 41 +++++++++++++----------- 4 files changed, 35 insertions(+), 28 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 376ad4cc..9600cd9d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -18,6 +18,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Before: `{"topic":"ingest.clicks","data":{"table_name":"clicks",...}}` - After: `{"table":"clicks","data":{"table_name":"clicks",...}}` - TypeScript SDK (`@wavehouse/sdk`) updated to the new contract: `SharedWSManager` is now keyed by table name (one less string concat per dispatch), `SSEOptions` / `WSOptions` carry `table` instead of `topic`, wildcard dispatch loop + `matchTopicPattern` helper removed from `ws-manager.ts`. The high-level SDK API (`wh.from('clicks').stream()`) is unchanged. +- WebSocket `?since=` gap-fill now accepts `RFC3339Nano` (with fractional seconds) in addition to `RFC3339`, matching the SSE handler's existing acceptance of high-precision client timestamps. +- WebSocket internal dispatch carries the subscribing table name through the merged channel instead of re-unmarshalling each payload to extract `table_name`. Drops the per-message redundant unmarshal and — for non-`EventMessage` payloads (raw-JSON pass-through) — keeps the outbound envelope's `"table"` field labelled with the actual subscription instead of falling back to `""`. No behaviour change for the standard `EventMessage` path. +- Doc-sync for the `?topic=` → `?table=` rename: `docs/src/content/docs/getting-started.md` quick-start now shows `?table=clicks` (the previous `?topic=ingest.clicks` example would 400 against the new contract), and `docs/src/content/docs/sdk.md` `SharedWSManager` section now describes per-table (not per-topic) subscriptions and notes that the legacy NATS-style wildcards are no longer accepted server-side. ### Removed - **`project-orchestrator.yml` workflow + its three composite-action artifacts** (`.github/workflows/project-orchestrator.yml`, `.github/actions/board-upsert-status/`, `.github/actions/set-linked-issues-status/`, `.github/scripts/board-fetch-item.sh`, `AGENTS.md`, `CHANGELOG.md`): −887 lines net. The orchestrator was the largest single source of cross-trigger complexity on this repo (3-4 workflow_run-chained runs per PR push, `statusCheckRollup` GraphQL perms quirks, integration-token `NONE` for private-org members) for behaviour that is mostly either provided natively by GitHub or a one-click manual operation on a 4-person team. Replaced by: reviewer-assign step in `housekeeping.yml` that fires once on `pull_request_target: opened` / `ready_for_review` (not per-synchronize, so it doesn't re-spam after `dismiss_stale_reviews_on_push`), plus GitHub's native Projects v2 workflows (`Auto-add to project`, `Item added`, `Pull request merged`) configured in the project UI. Trade-offs explicit in the PR body: drafts no longer auto-flip on bot-clean, `CHANGES_REQUESTED` doesn't auto-move the board card, linked-issue card mirroring is dropped. AGENTS.md §"Governance Files" + §"Task Board state machine" + §"Review tooling reference" all rewritten to match. `dependabot-automerge.yml` trimmed in parallel: no more board-upsert step (native handles placement), `PROJECT_BOARD_TOKEN` guard removed (no longer used in this workflow), reviewer list sourced from `board-config.env`'s `ADMINS` via `replace()`, major-bump comment uses the marker-comment upsert pattern from `housekeeping.yml`. diff --git a/docs/src/content/docs/getting-started.md b/docs/src/content/docs/getting-started.md index 17b2fca9..2a2b6267 100644 --- a/docs/src/content/docs/getting-started.md +++ b/docs/src/content/docs/getting-started.md @@ -78,16 +78,15 @@ Prefer a type-safe query builder over raw SQL? See the [structured query endpoin Every ingested event is broadcast to SSE and WebSocket subscribers **before** it's flushed to ClickHouse, so dashboards see new data with zero perceived lag. ```bash -# All tables -curl -N http://localhost:8080/v1/stream/sse - -# Specific table -curl -N "http://localhost:8080/v1/stream/sse?topic=ingest.clicks" +# Specific table (?table= is required) +curl -N "http://localhost:8080/v1/stream/sse?table=clicks" # With historical replay (RFC 3339 timestamp) -curl -N "http://localhost:8080/v1/stream/sse?since=2026-03-24T11:00:00Z" +curl -N "http://localhost:8080/v1/stream/sse?table=clicks&since=2026-03-24T11:00:00Z" ``` +To consume multiple tables on a single connection, use the WebSocket endpoint (`/v1/stream/ws`) with in-band `subscribe` commands — see the [API reference](api.md) for the envelope format. + ## Next steps - **[Architecture](architecture.md)** — how ingest, query, cache, and streaming fit together. diff --git a/docs/src/content/docs/sdk.md b/docs/src/content/docs/sdk.md index 54d6d704..49183568 100644 --- a/docs/src/content/docs/sdk.md +++ b/docs/src/content/docs/sdk.md @@ -535,12 +535,12 @@ interface StreamEvent { ### `SharedWSManager` -When using WebSocket transport, the SDK automatically multiplexes all topic subscriptions over a single WebSocket connection per client via `SharedWSManager`. This is transparent — `.stream()` calls route through it automatically. +When using WebSocket transport, the SDK automatically multiplexes all per-table subscriptions over a single WebSocket connection per client via `SharedWSManager`. This is transparent — `.stream()` calls route through it automatically. Key behaviors: -- Ref-counted subscriptions: unsubscribing removes only when the last subscriber for a topic disconnects. -- Auto-reconnect with exponential backoff; all active topics resubscribed on reconnect. -- NATS-style wildcard matching (`*` = one token, `>` = one-or-more) on the client side. +- Ref-counted subscriptions: unsubscribing removes only when the last subscriber for a table disconnects. +- Auto-reconnect with exponential backoff; all active table subscriptions resubscribed on reconnect. +- Exact-match dispatch by table name. The legacy NATS-style wildcards (`*`, `>`) are no longer accepted server-side; subscribe to one concrete table per call. ### Client-Side Stream Filtering diff --git a/internal/api/stream_ws.go b/internal/api/stream_ws.go index f19849ff..4cd07037 100644 --- a/internal/api/stream_ws.go +++ b/internal/api/stream_ws.go @@ -78,8 +78,12 @@ func (h *WSHandler) Handle(w http.ResponseWriter, r *http.Request) { ctx := r.Context() - // Merged channel receives messages from all subscribed topics. - merged := make(chan []byte, 64) + // Merged channel receives messages from all subscribed tables. Each message + // carries the subscribing table name so the dispatcher doesn't need to + // re-unmarshal the payload to label the outbound envelope — and so the + // envelope's "table" field is correct even when the payload isn't an + // ingest.EventMessage (e.g. raw JSON pass-through). + merged := make(chan wsOutbound, 64) // Track active subscriptions by table name, with their per-table channels. var mu sync.Mutex @@ -95,11 +99,12 @@ func (h *WSHandler) Handle(w http.ResponseWriter, r *http.Request) { subs[table] = ch h.Hub.Subscribe("ingest."+table, ch) - // Pump per-table channel into merged channel + // Pump per-table channel into merged channel, tagging each message with + // the subscribing table so the writer keeps the subscription context. go func() { for msg := range ch { select { - case merged <- msg: + case merged <- wsOutbound{table: table, data: msg}: default: } } @@ -135,9 +140,11 @@ func (h *WSHandler) Handle(w http.ResponseWriter, r *http.Request) { if table := r.URL.Query().Get("table"); table != "" { subscribeTable(table) - // Gap fill from NATS. + // Gap fill from NATS. RFC3339Nano subsumes RFC3339 (the fractional + // component is optional), matching the SSE handler's acceptance of + // high-precision client timestamps. if since := r.URL.Query().Get("since"); since != "" { - if ts, parseErr := time.Parse(time.RFC3339, since); parseErr == nil && h.JS != nil { + if ts, parseErr := time.Parse(time.RFC3339Nano, since); parseErr == nil && h.JS != nil { h.replayFromNATS(ctx, ts, "ingest."+table, func(data []byte) bool { out := h.applyStreamPolicy(data, role, map[string]any(claims), table) if out == nil { @@ -182,24 +189,15 @@ func (h *WSHandler) Handle(w http.ResponseWriter, r *http.Request) { select { case <-ctx.Done(): return - // Determine the event's table_name to label the envelope. - case data := <-merged: + case m := <-merged: var envelope struct { TraceHeaders map[string]string `json:"trace_headers"` Payload []byte `json:"payload"` } - if err := json.Unmarshal(data, &envelope); err != nil { + if err := json.Unmarshal(m.data, &envelope); err != nil { continue } - var rawEvt struct { - TableName string `json:"table_name"` - } - evtTable := "" - if json.Unmarshal(envelope.Payload, &rawEvt) == nil { - evtTable = rawEvt.TableName - } - parentCtx := otel.GetTextMapPropagator().Extract( context.Background(), propagation.MapCarrier(envelope.TraceHeaders), @@ -207,7 +205,7 @@ func (h *WSHandler) Handle(w http.ResponseWriter, r *http.Request) { _, pushSpan := tracer.Start(parentCtx, "WS.PushEvent") - out := h.applyStreamPolicy(envelope.Payload, role, map[string]any(claims), evtTable) + out := h.applyStreamPolicy(envelope.Payload, role, map[string]any(claims), m.table) if out == nil { pushSpan.End() continue @@ -222,6 +220,13 @@ func (h *WSHandler) Handle(w http.ResponseWriter, r *http.Request) { } } +// wsOutbound pairs a message with the subscribing table name so the WS write +// loop can label the outbound envelope without re-unmarshalling the payload. +type wsOutbound struct { + table string + data []byte +} + // applyStreamPolicy transforms raw event data for the client, filtering columns // based on the caller's policy permissions. Returns nil if the event should be skipped. // The result is wrapped in a table envelope: {"table":"...","data":{...}}. From 35592f3b39e3184de45f71f5c085efaa9b018b10 Mon Sep 17 00:00:00 2001 From: Eric Andrechek Date: Wed, 13 May 2026 16:40:48 -0400 Subject: [PATCH 4/5] fix(sdk): WS envelope parsing; doc + test cleanup from CodeRabbit MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes the CodeRabbit MAJOR + minor items raised on 1aa7ea7, plus Claude's [SHOULD] on the api.md `since` row from the same round. ### CodeRabbit MAJOR (real bug): `clients/ts/src/stream/ws.ts` `WSTransport.onmessage` still parsed the pre-#100 envelope shape — top-level `{ table_name, received_timestamp, data }`. The server now emits the table-wrapped envelope `{ table, data: { table_name, received_timestamp, data } }`, so every event coming through the direct WS transport would have surfaced as `{ table: undefined, timestamp: undefined, data: }` to the caller. The high-level `wh.from('clicks').stream()` path was not affected because it goes through `SharedWSManager` (which was already updated in this PR), but anything reaching into `WSTransport` directly was broken. Both `ws.ts` and `ws-manager.ts` now use the same envelope shape: ```ts const envelope = JSON.parse(e.data) as { table: string; data: { table_name?: string; received_timestamp?: string; data: ...; }; }; if (!envelope.table || !envelope.data) return; const event = { table: envelope.data.table_name ?? envelope.table, timestamp: envelope.data.received_timestamp ?? '', data: envelope.data.data, }; ``` The `?? envelope.table` fallback on the inner `table_name` lines up with the server-side change in the previous commit: for non- `EventMessage` payloads (raw-JSON pass-through), the inner object may not carry `table_name`, but the envelope's `table` field always holds the subscribing table. ### CodeRabbit minor + Claude [SHOULD] on docs - `docs/src/content/docs/api.md` SSE `?table=` row now spells out the validation regex and that 400 covers both missing AND invalid (was only "Returns 400 if missing"). WS `?table=` row gets the same regex note. Both `?since=` rows (SSE + WS) now say "RFC 3339 or RFC 3339 Nano" instead of just "RFC 3339" — sub-second precision is accepted, which Claude flagged as a doc gap on the WS row but applies equally to SSE. - `CHANGELOG.md` Unreleased block: added blank lines around `### Removed` / `### Changed` headings to satisfy markdownlint MD022 (heading-spacing rule). ### CodeRabbit refactor: `internal/api/stream_test.go` `TestSSE_RejectsMissingOrInvalidTable` and `TestWS_RejectsInvalidTableOnQuery` previously asserted the `http.StatusBadRequest` status + `application/json` Content-Type header. Switched to `testutil.AssertJSONContains(t, w, http.StatusBadRequest, map[string]any{"error": expectedMsg})` — the JSON unmarshal proves the body is JSON without checking the header directly, AND we now pin the actual error wording. The 9 SSE cases split into "missing required query parameter: table" (empty / unset table param) and "invalid table name" (everything else); the 5 WS cases all hit "invalid table name". ### Local gates - `make verify` clean - `make test-unit` 500 pass, 74.6% (`internal/api` 63.3%) - `make test-sdk` 120 pass, 54.6% - `make test-e2e` 30 pass + 1 skip, 51.3% - `make cov` merged 81.4% Co-Authored-By: Claude Opus 4.7 (1M context) --- CHANGELOG.md | 3 +++ clients/ts/src/stream/ws-manager.ts | 13 +++++++++---- clients/ts/src/stream/ws.ts | 22 ++++++++++++++------- docs/src/content/docs/api.md | 8 ++++---- internal/api/stream_test.go | 30 ++++++++++++++--------------- 5 files changed, 46 insertions(+), 30 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9600cd9d..1a494c98 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,10 +6,13 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ## Unreleased + ### Removed + - **Hub wildcard subscriptions** (`internal/api/hub.go`, `internal/api/hub_test.go`): dropped the NATS-style `*` / `>` pattern matching from `Hub.Broadcast`, the wildcard pattern loop, the `sent` dedup map, the `matchTopic` helper, and the eight wildcard tests (plus `TestMatchTopic`). After the #89 MVP cuts every producer publishes a concrete `ingest.
` subject and the SDK only ever subscribes to one concrete subject, so the wildcard fan-out was unused machinery. Closes #100 (part of #87). Net −210 lines (mostly tests). ### Changed + - **BREAKING:** SSE/WS streaming endpoints replaced `?topic=` with `?table=` (issue #100, option B). The NATS subject convention no longer leaks into the public HTTP API; the server builds `ingest.
` internally. SSE returns `400 Bad Request` when `?table=` is missing or fails the `^[a-zA-Z_][a-zA-Z0-9_]*$` safe-identifier check (the same regex `internal/ingest` and `internal/query` use for SQL identifiers — crucially this rejects NATS wildcard characters `*` and `>` before they reach the gap-fill `FilterSubject`). WS `?table=` is optional — clients can still defer to in-band subscribe commands — but is validated by the same regex when present. - **BREAKING:** WebSocket in-band subscribe/unsubscribe commands use `"table"` (raw name) instead of `"topic"` (NATS subject): - Before: `{"action":"subscribe","topic":"ingest.clicks"}` diff --git a/clients/ts/src/stream/ws-manager.ts b/clients/ts/src/stream/ws-manager.ts index 9780a4a9..73c9bfd9 100644 --- a/clients/ts/src/stream/ws-manager.ts +++ b/clients/ts/src/stream/ws-manager.ts @@ -145,19 +145,24 @@ export class SharedWSManager { this._ws.onmessage = (e) => { try { + // Server envelope: {table, data: {table_name, received_timestamp, data}}. + // For non-EventMessage (raw-JSON pass-through) payloads, the inner + // table_name may be absent — fall back to the envelope's table so the + // StreamEvent always carries the subscribing table. envelope.table is + // also what we key dispatch by, so the two are guaranteed to agree. const envelope = JSON.parse(e.data as string) as { table: string; data: { - table_name: string; - received_timestamp: string; + table_name?: string; + received_timestamp?: string; data: unknown; }; }; if (!envelope.table || !envelope.data) return; const event: StreamEvent = { - table: envelope.data.table_name, - timestamp: envelope.data.received_timestamp, + table: envelope.data.table_name ?? envelope.table, + timestamp: envelope.data.received_timestamp ?? '', data: envelope.data.data as Record, }; diff --git a/clients/ts/src/stream/ws.ts b/clients/ts/src/stream/ws.ts index f9d63000..66f936d5 100644 --- a/clients/ts/src/stream/ws.ts +++ b/clients/ts/src/stream/ws.ts @@ -70,15 +70,23 @@ export class WSTransport> implements StreamTransport this._ws.onmessage = (e) => { try { - const msg = JSON.parse(e.data as string) as { - table_name: string; - received_timestamp: string; - data: T; + // Server WS envelope: {table, data: {table_name, received_timestamp, data}}. + // For non-EventMessage (raw-JSON pass-through) payloads, table_name on + // the inner object may be absent — fall back to the envelope's table so + // the StreamEvent always carries the subscribing table name. + const envelope = JSON.parse(e.data as string) as { + table: string; + data: { + table_name?: string; + received_timestamp?: string; + data: T; + }; }; + if (!envelope.table || !envelope.data) return; const event: StreamEvent = { - table: msg.table_name, - timestamp: msg.received_timestamp, - data: msg.data, + table: envelope.data.table_name ?? envelope.table, + timestamp: envelope.data.received_timestamp ?? '', + data: envelope.data.data, }; this.onEvent?.(event); } catch { diff --git a/docs/src/content/docs/api.md b/docs/src/content/docs/api.md index 1f6f10de..502f780e 100644 --- a/docs/src/content/docs/api.md +++ b/docs/src/content/docs/api.md @@ -311,8 +311,8 @@ Opens a persistent SSE connection for real-time event streaming. Supports histor | Param | Type | Default | Description | | ----- | ---- | ------- | ----------- | -| `table` | string | (required) | Table name to subscribe to. Returns 400 if missing. | -| `since` | string | — | RFC 3339 timestamp. If provided, replays historical events from NATS before switching to live streaming. | +| `table` | string | (required) | Table name to subscribe to. Must match `^[a-zA-Z_][a-zA-Z0-9_]*$` (rejects NATS wildcards `*` / `>`). Returns 400 if missing or invalid. | +| `since` | string | — | RFC 3339 or RFC 3339 Nano timestamp. If provided, replays historical events from NATS before switching to live streaming. | | `token` | string | — | JWT token (alternative to `Authorization` header, useful for `EventSource`). Stripped from URL after extraction. | **Headers:** @@ -355,8 +355,8 @@ Opens a WebSocket connection for real-time event streaming. Supports in-band mul | Param | Type | Default | Description | | ----- | ---- | ------- | ----------- | -| `table` | string | — | Optional initial table to subscribe to. If omitted, the client must send subscribe commands. | -| `since` | string | — | RFC 3339 timestamp for gap-fill on the initial `?table=` subscription. | +| `table` | string | — | Optional initial table to subscribe to. If omitted, the client must send subscribe commands. When present, must match `^[a-zA-Z_][a-zA-Z0-9_]*$` — invalid values return 400 before the WebSocket upgrade. | +| `since` | string | — | RFC 3339 or RFC 3339 Nano timestamp for gap-fill on the initial `?table=` subscription. | | `token` | string | — | JWT token (alternative to `Authorization` header). Stripped from URL after extraction. | **In-band commands (client → server):** diff --git a/internal/api/stream_test.go b/internal/api/stream_test.go index 57e1e08b..5710f16a 100644 --- a/internal/api/stream_test.go +++ b/internal/api/stream_test.go @@ -11,6 +11,7 @@ import ( "github.com/Wave-RF/WaveHouse/internal/ingest" "github.com/Wave-RF/WaveHouse/internal/policy" + "github.com/Wave-RF/WaveHouse/internal/testutil" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -186,18 +187,19 @@ func TestSSE_RejectsMissingOrInvalidTable(t *testing.T) { h := &SSEHandler{Hub: NewHub()} cases := []struct { - name string - table string + name string + table string + errBody string }{ - {"missing", ""}, - {"nats greater wildcard", ">"}, - {"nats star wildcard", "*"}, - {"dot separator", "ingest.clicks"}, - {"nested wildcard", "ingest.>"}, - {"trailing wildcard", "clicks.>"}, - {"space", "click s"}, - {"leading digit", "1clicks"}, - {"empty after url decode", ""}, + {"missing", "", "missing required query parameter: table"}, + {"nats greater wildcard", ">", "invalid table name"}, + {"nats star wildcard", "*", "invalid table name"}, + {"dot separator", "ingest.clicks", "invalid table name"}, + {"nested wildcard", "ingest.>", "invalid table name"}, + {"trailing wildcard", "clicks.>", "invalid table name"}, + {"space", "click s", "invalid table name"}, + {"leading digit", "1clicks", "invalid table name"}, + {"empty after url decode", "", "missing required query parameter: table"}, } for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { @@ -209,8 +211,7 @@ func TestSSE_RejectsMissingOrInvalidTable(t *testing.T) { req := httptest.NewRequestWithContext(context.Background(), http.MethodGet, target, nil) w := httptest.NewRecorder() h.Handle(w, req) - assert.Equal(t, http.StatusBadRequest, w.Code) - assert.Equal(t, "application/json", w.Header().Get("Content-Type")) + testutil.AssertJSONContains(t, w, http.StatusBadRequest, map[string]any{"error": tc.errBody}) }) } } @@ -246,8 +247,7 @@ func TestWS_RejectsInvalidTableOnQuery(t *testing.T) { h.Handle(w, req) // Validation runs before websocket.Accept, so we get a plain 400 // rather than an upgrade-attempt error. - assert.Equal(t, http.StatusBadRequest, w.Code) - assert.Equal(t, "application/json", w.Header().Get("Content-Type")) + testutil.AssertJSONContains(t, w, http.StatusBadRequest, map[string]any{"error": "invalid table name"}) }) } } From 84a5448a7f3dbff604a3930b4267173f0852e65c Mon Sep 17 00:00:00 2001 From: Eric Andrechek Date: Wed, 13 May 2026 16:47:55 -0400 Subject: [PATCH 5/5] =?UTF-8?q?fix(sdk):=20SharedWSManager=20=E2=80=94=20d?= =?UTF-8?q?rop=20dup-subscribe=20+=20idempotent=20unsubscribe?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two latent bugs in `clients/ts/src/stream/ws-manager.ts` that CodeRabbit caught on the round-2 re-review of #124. ### 1. Duplicate subscribe frames on first connect The previous flow on first subscribe-before-connection-opens was: 1. `subscribe()` records the table in `_subs` and calls `_send({"action":"subscribe","table":"clicks"})`. 2. `_send` sees `this._ws` is null / not OPEN and queues the frame in `_pendingCommands`. 3. `_doConnect()` resolves auth, opens the socket. `onopen` fires. 4. `onopen` flushes `_pendingCommands` — sends `subscribe(clicks)` (1). 5. `onopen` then iterates `_subs.keys()` and sends `subscribe(clicks)` (2). Result: server sees the same table subscribed twice on a fresh connection. Server is idempotent (Hub dedups by exact subject), so the symptom isn't visibly broken, but it's wasted bandwidth and noise in any traffic capture. Fix: drop the `_pendingCommands` queue entirely and only send subscribe frames from `subscribe()` when the socket is already open. When the socket isn't open yet, `onopen`'s `_subs.keys()` reconciliation is the single source of truth — it sends one subscribe per active table. No queuing means no duplication. The same applies to `unsubscribe`: if we tear a sub down before the socket opens, the table is already removed from `_subs`, so `onopen` will simply not include it in its re-subscribe loop. Sending an explicit `unsubscribe` frame to a server that never saw the `subscribe` is a no-op, so we skip that frame when the socket isn't open. ### 2. Stale `tableSubs` reference in unsubscribe closure The unsubscribe closure captured the `tableSubs` Set at subscribe-time: ```ts let tableSubs = this._subs.get(table); if (!tableSubs) { tableSubs = new Set(); this._subs.set(table, tableSubs); } tableSubs.add(sub); return () => { tableSubs!.delete(sub); if (tableSubs!.size === 0) { this._subs.delete(table); ... } }; ``` If the closure ran twice (idempotent unsubscribe), or if a new subscriber arrived after the first sub torn down and got a *fresh* Set placed at the same `table` key, the closure's captured `tableSubs` would still point at the now-detached old Set. Calling `delete(sub)` on the old Set would be a no-op, but `_subs.delete(table)` on the second call would yank the NEW subscriber's Set out from under it. Fix: look up `this._subs.get(table)` inside the closure each call, and short-circuit if the sub isn't a member of the current Set (already unsubscribed, or our Set was replaced). That makes the closure idempotent and safe against subscribe/unsubscribe re-entrancy. ### Local gates - `make test-sdk` 120 pass, 54.3% - `make test-e2e` 30 pass + 1 skip, 51.3% Co-Authored-By: Claude Opus 4.7 (1M context) --- clients/ts/src/stream/ws-manager.ts | 51 ++++++++++++++++------------- 1 file changed, 29 insertions(+), 22 deletions(-) diff --git a/clients/ts/src/stream/ws-manager.ts b/clients/ts/src/stream/ws-manager.ts index 73c9bfd9..3bc66370 100644 --- a/clients/ts/src/stream/ws-manager.ts +++ b/clients/ts/src/stream/ws-manager.ts @@ -32,7 +32,6 @@ export class SharedWSManager { private _reconnectTimer: ReturnType | null = null; private _closed = false; private _connected = false; - private _pendingCommands: string[] = []; constructor(baseURL: string, auth?: () => Promise | string) { this._baseURL = baseURL; @@ -69,16 +68,31 @@ export class SharedWSManager { onStatus?.(this._connected ? 'live' : 'connecting'); } - // Send subscribe command for new tables. - if (isNewTable) { - this._send(JSON.stringify({ action: 'subscribe', table })); + // Send subscribe frame only if the socket is open. If we're still + // connecting (or reconnecting), onopen will reconcile by re-subscribing + // every table in this._subs — queuing the frame here would just produce + // a duplicate subscribe once the socket opens. + if (isNewTable && this._wsOpen()) { + this._ws!.send(JSON.stringify({ action: 'subscribe', table })); } return () => { - tableSubs!.delete(sub); - if (tableSubs!.size === 0) { + // Idempotent — look up the *current* set each call rather than the one + // captured at subscribe-time. If this closure runs twice (or after a + // subsequent subscribe replaced this table's set), the captured + // reference would otherwise let the second call delete the wrong entry + // from this._subs. + const current = this._subs.get(table); + if (!current || !current.has(sub)) return; + current.delete(sub); + if (current.size === 0) { this._subs.delete(table); - this._send(JSON.stringify({ action: 'unsubscribe', table })); + if (this._wsOpen()) { + this._ws!.send(JSON.stringify({ action: 'unsubscribe', table })); + } + // If not open, onopen only re-subscribes what's still in this._subs — + // the just-deleted table is excluded, so no explicit unsubscribe + // needs to reach the server. } // Close connection if no subscriptions remain. if (this._subs.size === 0) { @@ -87,6 +101,10 @@ export class SharedWSManager { }; } + private _wsOpen(): boolean { + return this._ws !== null && this._ws.readyState === WebSocket.OPEN; + } + /** Close the WebSocket and release all resources. */ close(): void { this._closed = true; @@ -131,13 +149,10 @@ export class SharedWSManager { this._connected = true; this._notifyAllStatus('live'); - // Flush pending commands. - for (const cmd of this._pendingCommands) { - this._ws?.send(cmd); - } - this._pendingCommands = []; - - // Re-subscribe all active tables. + // Reconcile server state from this._subs — sends one subscribe per + // currently-active table. Sub/unsub frames while the socket was opening + // (or reconnecting) intentionally aren't queued; this loop is the + // single source of truth for what the server should be streaming. for (const table of this._subs.keys()) { this._ws?.send(JSON.stringify({ action: 'subscribe', table })); } @@ -195,14 +210,6 @@ export class SharedWSManager { }; } - private _send(data: string): void { - if (this._ws && this._ws.readyState === WebSocket.OPEN) { - this._ws.send(data); - } else { - this._pendingCommands.push(data); - } - } - private _scheduleReconnect(): void { this._notifyAllStatus('reconnecting'); const delay = Math.min(1000 * 2 ** this._reconnectAttempt, 30_000);