From 412ff6b99436a2b6f1fda6065f4341b8071a8e76 Mon Sep 17 00:00:00 2001 From: taitelee Date: Tue, 21 Apr 2026 15:50:21 -0400 Subject: [PATCH 1/5] Configurable stream names --- cmd/wavehouse-api/main.go | 2 +- cmd/wavehouse-worker/main.go | 3 ++- cmd/wavehouse/main.go | 3 ++- config.yaml | 1 + internal/api/dlq_test.go | 8 ++++---- internal/api/router.go | 1 + internal/config/config.go | 1 + internal/ingest/bento.go | 4 ++-- internal/mq/embedded.go | 7 ++++--- internal/mq/remote.go | 7 ++++--- tests/integration_test.go | 5 ++++- 11 files changed, 26 insertions(+), 16 deletions(-) diff --git a/cmd/wavehouse-api/main.go b/cmd/wavehouse-api/main.go index ded0bac9..7db56f53 100644 --- a/cmd/wavehouse-api/main.go +++ b/cmd/wavehouse-api/main.go @@ -126,7 +126,7 @@ func main() { // Remote MQ (NATS). maxBytes := int64(cfg.MQ.MaxBytesGB) * 1024 * 1024 * 1024 - remoteMQ, err := mq.NewRemote(cfg.MQ.URL, maxBytes) + remoteMQ, err := mq.NewRemote(cfg.MQ.URL, cfg.MQ.StreamName, maxBytes) if err != nil { logger.Error("mq init", "error", err) os.Exit(1) diff --git a/cmd/wavehouse-worker/main.go b/cmd/wavehouse-worker/main.go index 692e4c31..5c7e516b 100644 --- a/cmd/wavehouse-worker/main.go +++ b/cmd/wavehouse-worker/main.go @@ -96,7 +96,7 @@ func main() { // Remote MQ (NATS). maxBytes := int64(cfg.MQ.MaxBytesGB) * 1024 * 1024 * 1024 - remoteMQ, err := mq.NewRemote(cfg.MQ.URL, maxBytes) + remoteMQ, err := mq.NewRemote(cfg.MQ.URL, cfg.MQ.StreamName, maxBytes) if err != nil { logger.Error("mq init", "error", err) os.Exit(1) @@ -121,6 +121,7 @@ func main() { ingestStream, err := ingest.StartIngestWorker( ctx, remoteMQ.NatsConn(), + cfg.MQ.StreamName, chConn, cfg.ClickHouse.Addr, cfg.ClickHouse.HTTPPort, // Uses 8123 by default diff --git a/cmd/wavehouse/main.go b/cmd/wavehouse/main.go index 6a1b1a7a..4c0ce66d 100644 --- a/cmd/wavehouse/main.go +++ b/cmd/wavehouse/main.go @@ -120,7 +120,7 @@ func main() { // Embedded MQ (NATS). maxBytes := int64(cfg.MQ.MaxBytesGB) * 1024 * 1024 * 1024 - embeddedMQ, err := mq.NewEmbedded(cfg.MQ.EmbeddedDir, maxBytes) + embeddedMQ, err := mq.NewEmbedded(cfg.MQ.EmbeddedDir, cfg.MQ.StreamName, maxBytes) if err != nil { logger.Error("mq init", "error", err) os.Exit(1) @@ -183,6 +183,7 @@ func main() { ingestStream, err := ingest.StartIngestWorker( ctx, embeddedMQ.NatsConn(), + cfg.MQ.StreamName, chConn, cfg.ClickHouse.Addr, cfg.ClickHouse.HTTPPort, // Uses 8123 by default diff --git a/config.yaml b/config.yaml index 11ce9665..e09195f3 100644 --- a/config.yaml +++ b/config.yaml @@ -14,6 +14,7 @@ clickhouse: password: "" mq: + stream_name: WAVEHOUSE embedded_dir: ./data/nats url: nats://localhost:4222 gap_window_minutes: 15 diff --git a/internal/api/dlq_test.go b/internal/api/dlq_test.go index ff56cb47..aec65824 100644 --- a/internal/api/dlq_test.go +++ b/internal/api/dlq_test.go @@ -16,7 +16,7 @@ import ( func TestDLQStats_EmptyWhenNoStream(t *testing.T) { dir := t.TempDir() - emb, err := mq.NewEmbedded(dir, 1024*1024, testutil.NopLogger()) + emb, err := mq.NewEmbedded(dir, "WAVEHOUSE", 1024*1024, testutil.NopLogger()) require.NoError(t, err) defer emb.Close() @@ -40,7 +40,7 @@ func TestDLQStats_EmptyWhenNoStream(t *testing.T) { func TestDLQStats_ReturnsCorrectCounts(t *testing.T) { dir := t.TempDir() - emb, err := mq.NewEmbedded(dir, 1024*1024, testutil.NopLogger()) + emb, err := mq.NewEmbedded(dir, "WAVEHOUSE", 1024*1024, testutil.NopLogger()) require.NoError(t, err) defer emb.Close() @@ -80,7 +80,7 @@ func TestDLQStats_ReturnsCorrectCounts(t *testing.T) { func TestDLQStats_SingleTable(t *testing.T) { dir := t.TempDir() - emb, err := mq.NewEmbedded(dir, 1024*1024, testutil.NopLogger()) + emb, err := mq.NewEmbedded(dir, "WAVEHOUSE", 1024*1024, testutil.NopLogger()) require.NoError(t, err) defer emb.Close() @@ -110,7 +110,7 @@ func TestDLQStats_SingleTable(t *testing.T) { func TestEnsureDLQStream_Idempotent(t *testing.T) { dir := t.TempDir() - emb, err := mq.NewEmbedded(dir, 1024*1024, testutil.NopLogger()) + emb, err := mq.NewEmbedded(dir, "WAVEHOUSE", 1024*1024, testutil.NopLogger()) require.NoError(t, err) defer emb.Close() diff --git a/internal/api/router.go b/internal/api/router.go index 26436873..df366b4d 100644 --- a/internal/api/router.go +++ b/internal/api/router.go @@ -25,6 +25,7 @@ type Dependencies struct { Pipes *PipesHandler StructuredQuery *StructuredQueryHandler AuthMW func(http.Handler) http.Handler + AuthEnabled bool JS jetstream.JetStream // for SSE/WS gap-fill CORSOrigins []string // allowed CORS origins; ["*"] = allow all LogLevel *slog.LevelVar diff --git a/internal/config/config.go b/internal/config/config.go index 27f0dda4..2d52b48d 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -45,6 +45,7 @@ type ClickHouse struct { } type MQ struct { + StreamName string `yaml:"stream_name" env:"WH_MQ_STREAM_NAME" env-default:"WAVEHOUSE"` EmbeddedDir string `yaml:"embedded_dir" env:"WH_MQ_EMBEDDED_DIR" env-default:"./data/nats"` URL string `yaml:"url" env:"WH_MQ_URL" env-default:"nats://localhost:4222"` GapWindowMinutes int `yaml:"gap_window_minutes" env:"WH_MQ_GAP_WINDOW_MINUTES" env-default:"15"` diff --git a/internal/ingest/bento.go b/internal/ingest/bento.go index 22bd9941..431e038e 100644 --- a/internal/ingest/bento.go +++ b/internal/ingest/bento.go @@ -213,7 +213,7 @@ func (d *dlqOutput) WriteBatch(ctx context.Context, batch service.MessageBatch) // running stream for lifecycle management. Callers should call stream.Stop(ctx) // during graceful shutdown to drain in-flight batches. The provided ctx controls // the stream's lifetime — cancelling it initiates shutdown of the Bento pipeline. -func StartIngestWorker(ctx context.Context, nc *nats.Conn, chConn driver.Conn, chHost, chHTTPPort, chUser, chPassword, chDB string) (*service.Stream, error) { +func StartIngestWorker(ctx context.Context, nc *nats.Conn, streamName string, chConn driver.Conn, chHost, chHTTPPort, chUser, chPassword, chDB string) (*service.Stream, error) { host, _, err := net.SplitHostPort(chHost) if err != nil { host = chHost @@ -229,7 +229,7 @@ func StartIngestWorker(ctx context.Context, nc *nats.Conn, chConn driver.Conn, c setupCtx, setupCancel := context.WithTimeout(ctx, 10*time.Second) defer setupCancel() - cons, err := js.CreateOrUpdateConsumer(setupCtx, "WAVEHOUSE", jetstream.ConsumerConfig{ + cons, err := js.CreateOrUpdateConsumer(setupCtx, streamName, jetstream.ConsumerConfig{ Durable: "buffer-consumer", FilterSubject: "ingest.>", AckPolicy: jetstream.AckExplicitPolicy, diff --git a/internal/mq/embedded.go b/internal/mq/embedded.go index 67e09b15..0db8182e 100644 --- a/internal/mq/embedded.go +++ b/internal/mq/embedded.go @@ -46,12 +46,13 @@ type EmbeddedNATS struct { server *natsserver.Server conn *nats.Conn js jetstream.JetStream + streamName string } // NewEmbedded starts an embedded NATS server with JetStream enabled. // An optional *slog.Logger can be passed to control server log output; // if omitted, slog.Default() is used. -func NewEmbedded(storeDir string, maxBytes int64, logger ...*slog.Logger) (*EmbeddedNATS, error) { +func NewEmbedded(storeDir, streamName string, maxBytes int64, logger ...*slog.Logger) (*EmbeddedNATS, error) { l := slog.Default() if len(logger) > 0 && logger[0] != nil { l = logger[0] @@ -105,7 +106,7 @@ func NewEmbedded(storeDir string, maxBytes int64, logger ...*slog.Logger) (*Embe return nil, fmt.Errorf("create stream: %w", err) } - return &EmbeddedNATS{server: ns, conn: nc, js: js}, nil + return &EmbeddedNATS{server: ns, conn: nc, js: js, streamName: streamName}, nil } func (e *EmbeddedNATS) Publish(ctx context.Context, subject string, data []byte) error { @@ -118,7 +119,7 @@ func (e *EmbeddedNATS) Publish(ctx context.Context, subject string, data []byte) } func (e *EmbeddedNATS) Subscribe(ctx context.Context, subject, consumerName string, handler func(msg *Message) error) error { - cons, err := e.js.CreateOrUpdateConsumer(ctx, streamName, jetstream.ConsumerConfig{ + cons, err := e.js.CreateOrUpdateConsumer(ctx, e.streamName, jetstream.ConsumerConfig{ Durable: consumerName, FilterSubject: subject, AckPolicy: jetstream.AckExplicitPolicy, diff --git a/internal/mq/remote.go b/internal/mq/remote.go index 22f317b1..b56ead31 100644 --- a/internal/mq/remote.go +++ b/internal/mq/remote.go @@ -14,10 +14,11 @@ import ( type RemoteNATS struct { conn *nats.Conn js jetstream.JetStream + streamName string } // NewRemote connects to a NATS cluster at the given URL. -func NewRemote(url string, maxBytes int64) (*RemoteNATS, error) { +func NewRemote(url, streamName string, maxBytes int64) (*RemoteNATS, error) { nc, err := nats.Connect(url) if err != nil { return nil, fmt.Errorf("connect to nats: %w", err) @@ -45,7 +46,7 @@ func NewRemote(url string, maxBytes int64) (*RemoteNATS, error) { return nil, fmt.Errorf("create stream: %w", err) } - return &RemoteNATS{conn: nc, js: js}, nil + return &RemoteNATS{conn: nc, js: js, streamName: streamName}, nil } func (r *RemoteNATS) Publish(ctx context.Context, subject string, data []byte) error { @@ -58,7 +59,7 @@ func (r *RemoteNATS) Publish(ctx context.Context, subject string, data []byte) e } func (r *RemoteNATS) Subscribe(ctx context.Context, subject, consumerName string, handler func(msg *Message) error) error { - cons, err := r.js.CreateOrUpdateConsumer(ctx, streamName, jetstream.ConsumerConfig{ + cons, err := r.js.CreateOrUpdateConsumer(ctx, r.streamName, jetstream.ConsumerConfig{ Durable: consumerName, FilterSubject: subject, AckPolicy: jetstream.AckExplicitPolicy, diff --git a/tests/integration_test.go b/tests/integration_test.go index 03610884..d8bbc4be 100644 --- a/tests/integration_test.go +++ b/tests/integration_test.go @@ -92,6 +92,7 @@ func setupTestEnv(t *testing.T) *testEnv { t.Helper() ctx := context.Background() logger := slog.Default() + testStream := "WAVEHOUSE" // Start ClickHouse container. chReq := testcontainers.ContainerRequest{ @@ -132,7 +133,7 @@ func setupTestEnv(t *testing.T) *testEnv { require.NoError(t, err) // Start embedded NATS. - embeddedMQ, err := mq.NewEmbedded(t.TempDir(), 10*1024*1024, testutil.NopLogger()) // 10 MB + embeddedMQ, err := mq.NewEmbedded(t.TempDir(), testStream, 10*1024*1024, testutil.NopLogger()) // 10 MB require.NoError(t, err) js := embeddedMQ.JetStream() @@ -148,6 +149,7 @@ func setupTestEnv(t *testing.T) *testEnv { _, err = ingest.StartIngestWorker( ctx, embeddedMQ.NatsConn(), + testStream, chConn, chAddr, httpPort.Port(), @@ -181,6 +183,7 @@ func setupTestEnv(t *testing.T) *testEnv { Schema: api.NewSchemaHandler(registry), DLQ: dlqHandler, AuthMW: api.JWTAuthMiddleware(api.AuthConfig{Enabled: false}), + AuthEnabled: false, JS: js, } From bc78f22fc25c26cd4907f133b8ca65a4cc97cd90 Mon Sep 17 00:00:00 2001 From: taitelee Date: Wed, 22 Apr 2026 12:28:06 -0400 Subject: [PATCH 2/5] Fixed integration test and dlq bugs and removed scope creep --- CHANGELOG.md | 2 ++ cmd/wavehouse-api/main.go | 4 ++-- cmd/wavehouse-worker/main.go | 2 +- cmd/wavehouse/main.go | 4 ++-- docs/configuration.md | 1 + internal/api/dlq.go | 13 ++++++------- internal/api/dlq_test.go | 24 +++++++++++++++--------- internal/api/router.go | 1 - internal/config/config.go | 4 ++++ internal/mq/embedded.go | 1 - tests/integration_test.go | 5 ++--- 11 files changed, 35 insertions(+), 26 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 18ff484c..6918eb95 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,8 @@ 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 +### Added +- Configurable NATS JetStream stream names via `WH_MQ_STREAM_NAME` to support multi-tenant deployments. ### Fixed diff --git a/cmd/wavehouse-api/main.go b/cmd/wavehouse-api/main.go index 7db56f53..982065a5 100644 --- a/cmd/wavehouse-api/main.go +++ b/cmd/wavehouse-api/main.go @@ -135,7 +135,7 @@ func main() { // DLQ stream. if cfg.DLQ.Enabled { - if err := api.EnsureDLQStream(context.Background(), remoteMQ.JetStream(), maxBytes/10); err != nil { + if err := api.EnsureDLQStream(context.Background(), remoteMQ.JetStream(), cfg.MQ.StreamName, maxBytes/10); err != nil { logger.Error("dlq stream init", "error", err) os.Exit(1) } @@ -207,7 +207,7 @@ func main() { var dlqHandler *api.DLQHandler if cfg.DLQ.Enabled { - dlqHandler = api.NewDLQHandler(js, logger) + dlqHandler = api.NewDLQHandler(js, cfg.MQ.StreamName, logger) } queryHandler := api.NewQueryHandler(chConn, tiered, time.Duration(cfg.Cache.DefaultTTL)*time.Second) diff --git a/cmd/wavehouse-worker/main.go b/cmd/wavehouse-worker/main.go index 5c7e516b..6c655136 100644 --- a/cmd/wavehouse-worker/main.go +++ b/cmd/wavehouse-worker/main.go @@ -105,7 +105,7 @@ func main() { // DLQ stream. if cfg.DLQ.Enabled { - if err := api.EnsureDLQStream(context.Background(), remoteMQ.JetStream(), maxBytes/10); err != nil { + if err := api.EnsureDLQStream(context.Background(), remoteMQ.JetStream(), cfg.MQ.StreamName, maxBytes/10); err != nil { logger.Error("dlq stream init", "error", err) os.Exit(1) } diff --git a/cmd/wavehouse/main.go b/cmd/wavehouse/main.go index 4c0ce66d..c9d3d71a 100644 --- a/cmd/wavehouse/main.go +++ b/cmd/wavehouse/main.go @@ -133,7 +133,7 @@ func main() { // DLQ stream. if cfg.DLQ.Enabled { - if err := api.EnsureDLQStream(context.Background(), embeddedMQ.JetStream(), maxBytes/10); err != nil { + if err := api.EnsureDLQStream(context.Background(), embeddedMQ.JetStream(), cfg.MQ.StreamName, maxBytes/10); err != nil { logger.Error("dlq stream init", "error", err) os.Exit(1) } @@ -224,7 +224,7 @@ func main() { var dlqHandler *api.DLQHandler if cfg.DLQ.Enabled { - dlqHandler = api.NewDLQHandler(js, logger) + dlqHandler = api.NewDLQHandler(js, cfg.MQ.StreamName, logger) } queryHandler := api.NewQueryHandler(chConn, tiered, time.Duration(cfg.Cache.DefaultTTL)*time.Second) diff --git a/docs/configuration.md b/docs/configuration.md index 4cecdd17..6d1a87f3 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -50,6 +50,7 @@ export WH_CONFIG=/etc/wavehouse/config.yaml | YAML Key | Env Var | Default | Description | | -------- | ------- | ------- | ----------- | +| `mq.stream_name` | `WH_MQ_STREAM_NAME` | `WAVEHOUSE` | The JetStream namespace to isolate environments (e.g., Staging vs Prod). | | `mq.embedded_dir` | `WH_MQ_EMBEDDED_DIR` | `./data/nats` | Data directory for the embedded NATS server (standalone mode). | | `mq.url` | `WH_MQ_URL` | `nats://localhost:4222` | NATS server URL (clustered mode). | | `mq.gap_window_minutes` | `WH_MQ_GAP_WINDOW_MINUTES` | `15` | How many minutes of messages to retain in NATS for SSE/WS gap-fill. The Active Sweeper will not purge messages newer than this window. | diff --git a/internal/api/dlq.go b/internal/api/dlq.go index fb49fb68..300f5f0d 100644 --- a/internal/api/dlq.go +++ b/internal/api/dlq.go @@ -12,20 +12,19 @@ import ( // DLQHandler exposes Dead Letter Queue statistics. type DLQHandler struct { JS jetstream.JetStream + DLQStreamName string Logger *slog.Logger } -func NewDLQHandler(js jetstream.JetStream, logger *slog.Logger) *DLQHandler { - return &DLQHandler{JS: js, Logger: logger} +func NewDLQHandler(js jetstream.JetStream, baseStreamName string, logger *slog.Logger) *DLQHandler { + return &DLQHandler{JS: js, DLQStreamName: baseStreamName + "_DLQ", Logger: logger} } -// DLQStreamName is the NATS JetStream stream used for dead-lettered events. -const DLQStreamName = "WAVEHOUSE_DLQ" // Stats returns per-table message counts in the DLQ stream. // Supports optional ?table= query parameter to filter by table name. func (h *DLQHandler) Stats(w http.ResponseWriter, r *http.Request) { - stream, err := h.JS.Stream(r.Context(), DLQStreamName) + stream, err := h.JS.Stream(r.Context(), h.DLQStreamName) if err != nil { // Stream may not exist yet if no failures have occurred. w.Header().Set("Content-Type", "application/json") @@ -56,9 +55,9 @@ func (h *DLQHandler) Stats(w http.ResponseWriter, r *http.Request) { } // EnsureDLQStream creates the DLQ JetStream stream if it doesn't exist. -func EnsureDLQStream(ctx context.Context, js jetstream.JetStream, maxBytes int64) error { +func EnsureDLQStream(ctx context.Context, js jetstream.JetStream, baseStreamName string, maxBytes int64) error { _, err := js.CreateOrUpdateStream(ctx, jetstream.StreamConfig{ - Name: DLQStreamName, + Name: baseStreamName + "_DLQ", Subjects: []string{"dlq.>"}, Retention: jetstream.LimitsPolicy, MaxBytes: maxBytes, diff --git a/internal/api/dlq_test.go b/internal/api/dlq_test.go index aec65824..8fb8a736 100644 --- a/internal/api/dlq_test.go +++ b/internal/api/dlq_test.go @@ -16,11 +16,12 @@ import ( func TestDLQStats_EmptyWhenNoStream(t *testing.T) { dir := t.TempDir() + streamName := "WAVEHOUSE" emb, err := mq.NewEmbedded(dir, "WAVEHOUSE", 1024*1024, testutil.NopLogger()) require.NoError(t, err) defer emb.Close() - handler := NewDLQHandler(emb.JetStream(), slog.Default()) + handler := NewDLQHandler(emb.JetStream(), streamName, slog.Default()) req := httptest.NewRequest(http.MethodGet, "/v1/dlq/stats", nil) rec := httptest.NewRecorder() @@ -40,6 +41,7 @@ func TestDLQStats_EmptyWhenNoStream(t *testing.T) { func TestDLQStats_ReturnsCorrectCounts(t *testing.T) { dir := t.TempDir() + streamName := "WAVEHOUSE" emb, err := mq.NewEmbedded(dir, "WAVEHOUSE", 1024*1024, testutil.NopLogger()) require.NoError(t, err) defer emb.Close() @@ -48,7 +50,7 @@ func TestDLQStats_ReturnsCorrectCounts(t *testing.T) { ctx := context.Background() // Create the DLQ stream. - require.NoError(t, EnsureDLQStream(ctx, js, 1024*1024)) + require.NoError(t, EnsureDLQStream(ctx, js, streamName, 1024*1024)) // Publish messages to DLQ subjects. for i := 0; i < 3; i++ { @@ -60,7 +62,7 @@ func TestDLQStats_ReturnsCorrectCounts(t *testing.T) { require.NoError(t, err) } - handler := NewDLQHandler(js, slog.Default()) + handler := NewDLQHandler(js, streamName, slog.Default()) req := httptest.NewRequest(http.MethodGet, "/v1/dlq/stats", nil) rec := httptest.NewRecorder() @@ -80,6 +82,7 @@ func TestDLQStats_ReturnsCorrectCounts(t *testing.T) { func TestDLQStats_SingleTable(t *testing.T) { dir := t.TempDir() + streamName := "WAVEHOUSE" emb, err := mq.NewEmbedded(dir, "WAVEHOUSE", 1024*1024, testutil.NopLogger()) require.NoError(t, err) defer emb.Close() @@ -87,12 +90,12 @@ func TestDLQStats_SingleTable(t *testing.T) { js := emb.JetStream() ctx := context.Background() - require.NoError(t, EnsureDLQStream(ctx, js, 1024*1024)) + require.NoError(t, EnsureDLQStream(ctx, js, streamName, 1024*1024)) _, err = js.Publish(ctx, "dlq.orders", []byte(`{"table_name":"orders"}`)) require.NoError(t, err) - handler := NewDLQHandler(js, slog.Default()) + handler := NewDLQHandler(js, streamName, slog.Default()) req := httptest.NewRequest(http.MethodGet, "/v1/dlq/stats", nil) rec := httptest.NewRecorder() @@ -110,6 +113,7 @@ func TestDLQStats_SingleTable(t *testing.T) { func TestEnsureDLQStream_Idempotent(t *testing.T) { dir := t.TempDir() + streamName := "WAVEHOUSE" emb, err := mq.NewEmbedded(dir, "WAVEHOUSE", 1024*1024, testutil.NopLogger()) require.NoError(t, err) defer emb.Close() @@ -118,15 +122,17 @@ func TestEnsureDLQStream_Idempotent(t *testing.T) { ctx := context.Background() // Calling twice should not error. - require.NoError(t, EnsureDLQStream(ctx, js, 1024*1024)) - require.NoError(t, EnsureDLQStream(ctx, js, 1024*1024)) + require.NoError(t, EnsureDLQStream(ctx, js, streamName, 1024*1024)) + require.NoError(t, EnsureDLQStream(ctx, js, streamName, 1024*1024)) + handler := NewDLQHandler(js, streamName, slog.Default()) + // Stream should be accessible. - stream, err := js.Stream(ctx, DLQStreamName) + stream, err := js.Stream(ctx, handler.DLQStreamName) require.NoError(t, err) info, err := stream.Info(ctx) require.NoError(t, err) - assert.Equal(t, DLQStreamName, info.Config.Name) + assert.Equal(t, handler.DLQStreamName, info.Config.Name) assert.Equal(t, []string{"dlq.>"}, info.Config.Subjects) } diff --git a/internal/api/router.go b/internal/api/router.go index df366b4d..26436873 100644 --- a/internal/api/router.go +++ b/internal/api/router.go @@ -25,7 +25,6 @@ type Dependencies struct { Pipes *PipesHandler StructuredQuery *StructuredQueryHandler AuthMW func(http.Handler) http.Handler - AuthEnabled bool JS jetstream.JetStream // for SSE/WS gap-fill CORSOrigins []string // allowed CORS origins; ["*"] = allow all LogLevel *slog.LevelVar diff --git a/internal/config/config.go b/internal/config/config.go index 2d52b48d..c31d46a1 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -125,6 +125,10 @@ func (c *Config) Validate() error { return fmt.Errorf("mq.gap_window_minutes must be non-negative") } + if c.MQ.StreamName == "" { + return fmt.Errorf("mq.stream_name cannot be empty") + } + return nil } diff --git a/internal/mq/embedded.go b/internal/mq/embedded.go index 0db8182e..24bc5ff2 100644 --- a/internal/mq/embedded.go +++ b/internal/mq/embedded.go @@ -39,7 +39,6 @@ func (s *slogNATSLogger) Tracef(format string, v ...any) { s.l.Debug(fmt.Sprintf(format, v...), "component", "nats") } -const streamName = "WAVEHOUSE" // EmbeddedNATS runs an in-process NATS server with JetStream. type EmbeddedNATS struct { diff --git a/tests/integration_test.go b/tests/integration_test.go index d8bbc4be..1ee3eac5 100644 --- a/tests/integration_test.go +++ b/tests/integration_test.go @@ -139,7 +139,7 @@ func setupTestEnv(t *testing.T) *testEnv { js := embeddedMQ.JetStream() // Create DLQ stream. - require.NoError(t, api.EnsureDLQStream(ctx, js, 1024*1024)) + require.NoError(t, api.EnsureDLQStream(ctx, js, testStream, 1024*1024)) // Schema registry. registry := discovery.NewSchemaRegistry(chConn, "default", time.Minute, logger) @@ -172,7 +172,7 @@ func setupTestEnv(t *testing.T) *testEnv { queryHandler := api.NewQueryHandler(chConn, tiered, 5*time.Second) sseHandler := api.NewSSEHandler(hub, js) wsHandler := api.NewWSHandler(hub, js, nil) - dlqHandler := api.NewDLQHandler(js, logger) + dlqHandler := api.NewDLQHandler(js, testStream, logger) deps := api.Dependencies{ Ingest: ingestHandler, @@ -183,7 +183,6 @@ func setupTestEnv(t *testing.T) *testEnv { Schema: api.NewSchemaHandler(registry), DLQ: dlqHandler, AuthMW: api.JWTAuthMiddleware(api.AuthConfig{Enabled: false}), - AuthEnabled: false, JS: js, } From b94d1148388080393c4956975864c066eea89d6f Mon Sep 17 00:00:00 2001 From: taitelee Date: Wed, 22 Apr 2026 13:28:18 -0400 Subject: [PATCH 3/5] Ran make fmt --- internal/api/dlq.go | 5 ++--- internal/api/dlq_test.go | 4 ++-- internal/mq/embedded.go | 7 +++---- internal/mq/remote.go | 4 ++-- 4 files changed, 9 insertions(+), 11 deletions(-) diff --git a/internal/api/dlq.go b/internal/api/dlq.go index 300f5f0d..a5335d7b 100644 --- a/internal/api/dlq.go +++ b/internal/api/dlq.go @@ -11,16 +11,15 @@ import ( // DLQHandler exposes Dead Letter Queue statistics. type DLQHandler struct { - JS jetstream.JetStream + JS jetstream.JetStream DLQStreamName string - Logger *slog.Logger + Logger *slog.Logger } func NewDLQHandler(js jetstream.JetStream, baseStreamName string, logger *slog.Logger) *DLQHandler { return &DLQHandler{JS: js, DLQStreamName: baseStreamName + "_DLQ", Logger: logger} } - // Stats returns per-table message counts in the DLQ stream. // Supports optional ?table= query parameter to filter by table name. func (h *DLQHandler) Stats(w http.ResponseWriter, r *http.Request) { diff --git a/internal/api/dlq_test.go b/internal/api/dlq_test.go index 8fb8a736..3f5725d8 100644 --- a/internal/api/dlq_test.go +++ b/internal/api/dlq_test.go @@ -123,10 +123,10 @@ func TestEnsureDLQStream_Idempotent(t *testing.T) { // Calling twice should not error. require.NoError(t, EnsureDLQStream(ctx, js, streamName, 1024*1024)) - require.NoError(t, EnsureDLQStream(ctx, js, streamName, 1024*1024)) + require.NoError(t, EnsureDLQStream(ctx, js, streamName, 1024*1024)) handler := NewDLQHandler(js, streamName, slog.Default()) - + // Stream should be accessible. stream, err := js.Stream(ctx, handler.DLQStreamName) require.NoError(t, err) diff --git a/internal/mq/embedded.go b/internal/mq/embedded.go index 24bc5ff2..88375072 100644 --- a/internal/mq/embedded.go +++ b/internal/mq/embedded.go @@ -39,12 +39,11 @@ func (s *slogNATSLogger) Tracef(format string, v ...any) { s.l.Debug(fmt.Sprintf(format, v...), "component", "nats") } - // EmbeddedNATS runs an in-process NATS server with JetStream. type EmbeddedNATS struct { - server *natsserver.Server - conn *nats.Conn - js jetstream.JetStream + server *natsserver.Server + conn *nats.Conn + js jetstream.JetStream streamName string } diff --git a/internal/mq/remote.go b/internal/mq/remote.go index b56ead31..5fe6bfee 100644 --- a/internal/mq/remote.go +++ b/internal/mq/remote.go @@ -12,8 +12,8 @@ import ( // RemoteNATS connects to an external NATS cluster. type RemoteNATS struct { - conn *nats.Conn - js jetstream.JetStream + conn *nats.Conn + js jetstream.JetStream streamName string } From 0c24024a8619f4bda638c2f80672910a7bcd25c2 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Thu, 23 Apr 2026 14:55:16 +0000 Subject: [PATCH 4/5] fix(config,tests): default empty mq stream name and use request contexts in dlq tests Agent-Logs-Url: https://github.com/Wave-RF/WaveHouse/sessions/7b39cf64-429d-44ba-8ce7-57ef24b415be Co-authored-by: EricAndrechek <35144594+EricAndrechek@users.noreply.github.com> --- internal/api/dlq_test.go | 4 ++-- internal/config/config.go | 2 +- internal/config/config_test.go | 13 +++++++++++++ 3 files changed, 16 insertions(+), 3 deletions(-) diff --git a/internal/api/dlq_test.go b/internal/api/dlq_test.go index 87ae807b..76c5092a 100644 --- a/internal/api/dlq_test.go +++ b/internal/api/dlq_test.go @@ -63,7 +63,7 @@ func TestDLQStats_ReturnsCorrectCounts(t *testing.T) { } handler := NewDLQHandler(js, streamName, slog.Default()) - req := httptest.NewRequest(http.MethodGet, "/v1/dlq/stats", nil) + req := httptest.NewRequestWithContext(context.Background(), http.MethodGet, "/v1/dlq/stats", nil) rec := httptest.NewRecorder() handler.Stats(rec, req) @@ -96,7 +96,7 @@ func TestDLQStats_SingleTable(t *testing.T) { require.NoError(t, err) handler := NewDLQHandler(js, streamName, slog.Default()) - req := httptest.NewRequest(http.MethodGet, "/v1/dlq/stats", nil) + req := httptest.NewRequestWithContext(context.Background(), http.MethodGet, "/v1/dlq/stats", nil) rec := httptest.NewRecorder() handler.Stats(rec, req) diff --git a/internal/config/config.go b/internal/config/config.go index c31d46a1..f0232b63 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -126,7 +126,7 @@ func (c *Config) Validate() error { } if c.MQ.StreamName == "" { - return fmt.Errorf("mq.stream_name cannot be empty") + c.MQ.StreamName = "WAVEHOUSE" } return nil diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 2c3f26e4..01b57918 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -211,6 +211,19 @@ func TestValidate_NegativeGapWindow(t *testing.T) { assert.Contains(t, err.Error(), "gap_window_minutes") } +func TestValidate_DefaultsStreamNameWhenEmpty(t *testing.T) { + t.Parallel() + cfg := Config{ + Mode: ModeStandalone, + Server: Server{Port: 8080}, + Schema: Schema{RefreshInterval: 60}, + } + + err := cfg.Validate() + require.NoError(t, err) + assert.Equal(t, "WAVEHOUSE", cfg.MQ.StreamName) +} + func TestLoad_AuthEnabledNoSecret_FailsValidation(t *testing.T) { t.Parallel() dir := t.TempDir() From 3b2caa607038d7eb636e67ff5dce09309a258772 Mon Sep 17 00:00:00 2001 From: taitelee Date: Wed, 6 May 2026 14:07:47 -0400 Subject: [PATCH 5/5] fix(mq): fully wire dynamic stream name through handlers, add custom DLQ tests, and sync config docs --- cmd/wavehouse-api/main.go | 4 ++-- cmd/wavehouse-worker/main.go | 2 +- cmd/wavehouse/main.go | 6 ++--- deployments/compose/clustered.yaml | 4 ++++ deployments/compose/standalone.yaml | 1 + docs/configuration.md | 1 + internal/api/dlq_test.go | 34 +++++++++++++++++++++++++++++ internal/api/router_test.go | 16 +++++++------- internal/api/stream_sse.go | 8 +++---- internal/api/stream_ws.go | 8 +++---- internal/ingest/sweeper.go | 21 +++++++++--------- internal/ingest/sweeper_test.go | 32 +++++++++++++-------------- internal/mq/embedded_test.go | 4 ++-- internal/mq/mq.go | 5 ----- internal/mq/mq_test.go | 5 ----- internal/mq/remote_test.go | 6 ++--- tests/integration_test.go | 4 ++-- 17 files changed, 96 insertions(+), 65 deletions(-) diff --git a/cmd/wavehouse-api/main.go b/cmd/wavehouse-api/main.go index 3d8a8e71..35f33429 100644 --- a/cmd/wavehouse-api/main.go +++ b/cmd/wavehouse-api/main.go @@ -221,10 +221,10 @@ func run() int { queryHandler := api.NewQueryHandler(chConn, tiered, time.Duration(cfg.Cache.DefaultTTL)*time.Second) queryHandler.PolicyStore = policyStore - sseHandler := api.NewSSEHandler(hub, remoteMQ.JetStream()) + sseHandler := api.NewSSEHandler(hub, remoteMQ.JetStream(), cfg.MQ.StreamName) sseHandler.PolicyStore = policyStore - wsHandler := api.NewWSHandler(hub, remoteMQ.JetStream(), cfg.Server.CORSAllowedOrigins) + wsHandler := api.NewWSHandler(hub, remoteMQ.JetStream(), cfg.Server.CORSAllowedOrigins, cfg.MQ.StreamName) wsHandler.PolicyStore = policyStore deps := api.Dependencies{ diff --git a/cmd/wavehouse-worker/main.go b/cmd/wavehouse-worker/main.go index c6cb06a0..fdaa3786 100644 --- a/cmd/wavehouse-worker/main.go +++ b/cmd/wavehouse-worker/main.go @@ -143,7 +143,7 @@ func run() int { // Active sweeper — purges processed + expired messages every minute. gapWindow := time.Duration(cfg.MQ.GapWindowMinutes) * time.Minute - sweeper := ingest.NewSweeper(remoteMQ.JetStream(), gapWindow, logger) + sweeper := ingest.NewSweeper(remoteMQ.JetStream(), cfg.MQ.StreamName, gapWindow, logger) go sweeper.Start(ctx) logger.Info("worker started", "mode", "clustered") diff --git a/cmd/wavehouse/main.go b/cmd/wavehouse/main.go index b81c5812..00d70d64 100644 --- a/cmd/wavehouse/main.go +++ b/cmd/wavehouse/main.go @@ -173,7 +173,7 @@ func run() int { // Active sweeper — purges messages that are both written to CH and // older than the SSE gap window. Runs every minute. gapWindow := time.Duration(cfg.MQ.GapWindowMinutes) * time.Minute - sweeper := ingest.NewSweeper(embeddedMQ.JetStream(), gapWindow, logger) + sweeper := ingest.NewSweeper(embeddedMQ.JetStream(), cfg.MQ.StreamName, gapWindow, logger) // Hub for streaming fan-out. hub := api.NewHub() @@ -238,10 +238,10 @@ func run() int { queryHandler := api.NewQueryHandler(chConn, tiered, time.Duration(cfg.Cache.DefaultTTL)*time.Second) queryHandler.PolicyStore = policyStore - sseHandler := api.NewSSEHandler(hub, js) + sseHandler := api.NewSSEHandler(hub, js, cfg.MQ.StreamName) sseHandler.PolicyStore = policyStore - wsHandler := api.NewWSHandler(hub, js, cfg.Server.CORSAllowedOrigins) + wsHandler := api.NewWSHandler(hub, js, cfg.Server.CORSAllowedOrigins, cfg.MQ.StreamName) wsHandler.PolicyStore = policyStore deps := api.Dependencies{ diff --git a/deployments/compose/clustered.yaml b/deployments/compose/clustered.yaml index a69f5f80..96065a4d 100644 --- a/deployments/compose/clustered.yaml +++ b/deployments/compose/clustered.yaml @@ -48,6 +48,7 @@ services: environment: WH_MODE: clustered WH_CH_ADDR: clickhouse:9000 + WH_MQ_STREAM_NAME: WAVEHOUSE WH_MQ_URL: nats://nats:4222 WH_CACHE_REDIS_URL: redis://redis:6379 WH_AUTH_ENABLED: "false" @@ -71,6 +72,7 @@ services: environment: WH_MODE: clustered WH_CH_ADDR: clickhouse:9000 + WH_MQ_STREAM_NAME: WAVEHOUSE WH_MQ_URL: nats://nats:4222 WH_CACHE_REDIS_URL: redis://redis:6379 WH_AUTH_ENABLED: "false" @@ -94,6 +96,7 @@ services: environment: WH_MODE: clustered WH_CH_ADDR: clickhouse:9000 + WH_MQ_STREAM_NAME: WAVEHOUSE WH_MQ_URL: nats://nats:4222 WH_SCHEMA_REFRESH_INTERVAL: "60" WH_DLQ_ENABLED: "true" @@ -109,6 +112,7 @@ services: environment: WH_MODE: clustered WH_CH_ADDR: clickhouse:9000 + WH_MQ_STREAM_NAME: WAVEHOUSE WH_MQ_URL: nats://nats:4222 WH_SCHEMA_REFRESH_INTERVAL: "60" WH_DLQ_ENABLED: "true" diff --git a/deployments/compose/standalone.yaml b/deployments/compose/standalone.yaml index 74869c26..a68c1f55 100644 --- a/deployments/compose/standalone.yaml +++ b/deployments/compose/standalone.yaml @@ -19,6 +19,7 @@ services: environment: WH_MODE: standalone WH_CH_ADDR: clickhouse:9000 + WH_MQ_STREAM_NAME: WAVEHOUSE WH_MQ_EMBEDDED_DIR: /data/nats WH_AUTH_ENABLED: "false" WH_DEDUPE_ENABLED: "false" diff --git a/docs/configuration.md b/docs/configuration.md index 6d1a87f3..06a59157 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -120,6 +120,7 @@ clickhouse: password: "" mq: + stream_name: WAVEHOUSE embedded_dir: ./data/nats url: nats://localhost:4222 gap_window_minutes: 15 diff --git a/internal/api/dlq_test.go b/internal/api/dlq_test.go index 76c5092a..9c35ae4a 100644 --- a/internal/api/dlq_test.go +++ b/internal/api/dlq_test.go @@ -136,3 +136,37 @@ func TestEnsureDLQStream_Idempotent(t *testing.T) { assert.Equal(t, handler.DLQStreamName, info.Config.Name) assert.Equal(t, []string{"dlq.>"}, info.Config.Subjects) } + +func TestEnsureDLQStream_CustomName(t *testing.T) { + t.Parallel() + dir := t.TempDir() + + // Use a non-default stream name + customStreamName := "MY_CUSTOM_TENANT" + expectedDLQName := customStreamName + "_DLQ" + + emb, err := mq.NewEmbedded(dir, customStreamName, 1024*1024, testutil.NopLogger()) + require.NoError(t, err) + defer func() { _ = emb.Close() }() + + js := emb.JetStream() + ctx := context.Background() + + // 1. Ensure the DLQ stream is created + require.NoError(t, EnsureDLQStream(ctx, js, customStreamName, 1024*1024)) + + // 2. Instantiate the handler + handler := NewDLQHandler(js, customStreamName, slog.Default()) + + // 3. Assert the handler stored the correct derived name + assert.Equal(t, expectedDLQName, handler.DLQStreamName) + + // 4. Assert NATS actually created a stream with that exact derived name + stream, err := js.Stream(ctx, expectedDLQName) + require.NoError(t, err) + + info, err := stream.Info(ctx) + require.NoError(t, err) + assert.Equal(t, expectedDLQName, info.Config.Name) + assert.Equal(t, []string{"dlq.>"}, info.Config.Subjects) +} diff --git a/internal/api/router_test.go b/internal/api/router_test.go index 18afee6d..bfac710f 100644 --- a/internal/api/router_test.go +++ b/internal/api/router_test.go @@ -138,8 +138,8 @@ func TestNewRouter_RoutesRegistered(t *testing.T) { deps := Dependencies{ Ingest: NewIngestHandler(reg, pub), Query: &QueryHandler{}, - SSE: NewSSEHandler(hub, nil), - WS: NewWSHandler(hub, nil, nil), + SSE: NewSSEHandler(hub, nil, "WAVEHOUSE"), + WS: NewWSHandler(hub, nil, nil, "WAVEHOUSE"), Health: &HealthHandler{}, Schema: NewSchemaHandler(reg), AuthMW: func(next http.Handler) http.Handler { return next }, @@ -180,8 +180,8 @@ func TestNewRouter_OptionalDepsNil(t *testing.T) { deps := Dependencies{ Ingest: NewIngestHandler(reg, pub), Query: &QueryHandler{}, - SSE: NewSSEHandler(hub, nil), - WS: NewWSHandler(hub, nil, nil), + SSE: NewSSEHandler(hub, nil, "WAVEHOUSE"), + WS: NewWSHandler(hub, nil, nil, "WAVEHOUSE"), Health: &HealthHandler{}, Schema: NewSchemaHandler(reg), AuthMW: func(next http.Handler) http.Handler { return next }, @@ -247,8 +247,8 @@ func TestNewRouter_NotFoundEmitsJSON(t *testing.T) { deps := Dependencies{ Ingest: NewIngestHandler(reg, pub), Query: &QueryHandler{}, - SSE: NewSSEHandler(hub, nil), - WS: NewWSHandler(hub, nil, nil), + SSE: NewSSEHandler(hub, nil, "WAVEHOUSE"), + WS: NewWSHandler(hub, nil, nil, "WAVEHOUSE"), Health: &HealthHandler{}, Schema: NewSchemaHandler(reg), AuthMW: func(next http.Handler) http.Handler { return next }, @@ -272,8 +272,8 @@ func TestNewRouter_MethodNotAllowedEmitsJSON(t *testing.T) { deps := Dependencies{ Ingest: NewIngestHandler(reg, pub), Query: &QueryHandler{}, - SSE: NewSSEHandler(hub, nil), - WS: NewWSHandler(hub, nil, nil), + SSE: NewSSEHandler(hub, nil, "WAVEHOUSE"), + WS: NewWSHandler(hub, nil, nil, "WAVEHOUSE"), Health: &HealthHandler{}, Schema: NewSchemaHandler(reg), AuthMW: func(next http.Handler) http.Handler { return next }, diff --git a/internal/api/stream_sse.go b/internal/api/stream_sse.go index 60f2d9b3..35eabe5d 100644 --- a/internal/api/stream_sse.go +++ b/internal/api/stream_sse.go @@ -8,7 +8,6 @@ import ( "time" "github.com/Wave-RF/WaveHouse/internal/ingest" - "github.com/Wave-RF/WaveHouse/internal/mq" "github.com/Wave-RF/WaveHouse/internal/policy" "github.com/nats-io/nats.go/jetstream" @@ -21,10 +20,11 @@ type SSEHandler struct { Hub *Hub JS jetstream.JetStream PolicyStore *policy.Store + StreamName string } -func NewSSEHandler(hub *Hub, js jetstream.JetStream) *SSEHandler { - return &SSEHandler{Hub: hub, JS: js} +func NewSSEHandler(hub *Hub, js jetstream.JetStream, streamName string) *SSEHandler { + return &SSEHandler{Hub: hub, JS: js, StreamName: streamName} } func (h *SSEHandler) Handle(w http.ResponseWriter, r *http.Request) { @@ -171,7 +171,7 @@ func (h *SSEHandler) applyStreamPolicy(raw []byte, role string, claims map[strin // replayFromNATS creates an ephemeral NATS consumer starting at the given time // and sends all available messages to the callback until caught up. func (h *SSEHandler) replayFromNATS(ctx context.Context, since time.Time, subject string, send func([]byte) bool) { - cons, err := h.JS.CreateOrUpdateConsumer(ctx, mq.StreamName(), jetstream.ConsumerConfig{ + cons, err := h.JS.CreateOrUpdateConsumer(ctx, h.StreamName, jetstream.ConsumerConfig{ FilterSubject: subject, DeliverPolicy: jetstream.DeliverByStartTimePolicy, OptStartTime: &since, diff --git a/internal/api/stream_ws.go b/internal/api/stream_ws.go index 2dcd0767..c64bcd8e 100644 --- a/internal/api/stream_ws.go +++ b/internal/api/stream_ws.go @@ -8,7 +8,6 @@ import ( "time" "github.com/Wave-RF/WaveHouse/internal/ingest" - "github.com/Wave-RF/WaveHouse/internal/mq" "github.com/Wave-RF/WaveHouse/internal/observability" "github.com/Wave-RF/WaveHouse/internal/policy" @@ -37,10 +36,11 @@ type WSHandler struct { JS jetstream.JetStream PolicyStore *policy.Store AllowedOrigins []string + StreamName string } -func NewWSHandler(hub *Hub, js jetstream.JetStream, allowedOrigins []string) *WSHandler { - return &WSHandler{Hub: hub, JS: js, AllowedOrigins: allowedOrigins} +func NewWSHandler(hub *Hub, js jetstream.JetStream, allowedOrigins []string, streamName string) *WSHandler { + return &WSHandler{Hub: hub, JS: js, AllowedOrigins: allowedOrigins, StreamName: streamName} } // wsCommand represents an in-band subscribe/unsubscribe command. @@ -250,7 +250,7 @@ func (h *WSHandler) applyStreamPolicy(raw []byte, role string, claims map[string // replayFromNATS creates an ephemeral NATS consumer starting at the given time. func (h *WSHandler) replayFromNATS(ctx context.Context, since time.Time, subject string, send func([]byte) bool) { - cons, err := h.JS.CreateOrUpdateConsumer(ctx, mq.StreamName(), jetstream.ConsumerConfig{ + cons, err := h.JS.CreateOrUpdateConsumer(ctx, h.StreamName, jetstream.ConsumerConfig{ FilterSubject: subject, DeliverPolicy: jetstream.DeliverByStartTimePolicy, OptStartTime: &since, diff --git a/internal/ingest/sweeper.go b/internal/ingest/sweeper.go index f2e5e3d8..f6e03901 100644 --- a/internal/ingest/sweeper.go +++ b/internal/ingest/sweeper.go @@ -6,7 +6,6 @@ import ( "log/slog" "time" - "github.com/Wave-RF/WaveHouse/internal/mq" "github.com/nats-io/nats.go/jetstream" ) @@ -20,18 +19,20 @@ import ( // ClickHouse down freezes purging; catastrophic outage fills to MaxBytes // and triggers backpressure via DiscardNew. type Sweeper struct { - js jetstream.JetStream - gapWindow time.Duration - logger *slog.Logger + js jetstream.JetStream + streamName string + gapWindow time.Duration + logger *slog.Logger } // NewSweeper creates an Active Sweeper. // TODO: need leader election or shared lock to only run one instance of the sweeper in clustered mode -func NewSweeper(js jetstream.JetStream, gapWindow time.Duration, logger *slog.Logger) *Sweeper { +func NewSweeper(js jetstream.JetStream, streamName string, gapWindow time.Duration, logger *slog.Logger) *Sweeper { return &Sweeper{ - js: js, - gapWindow: gapWindow, - logger: logger, + js: js, + streamName: streamName, + gapWindow: gapWindow, + logger: logger, } } @@ -50,14 +51,14 @@ func (s *Sweeper) Start(ctx context.Context) { } func (s *Sweeper) sweep(ctx context.Context) { - stream, err := s.js.Stream(ctx, mq.StreamName()) + stream, err := s.js.Stream(ctx, s.streamName) if err != nil { s.logger.Error("sweeper: get stream", "error", err) return } // 1. Get buffer consumer's AckFloor — highest contiguous ACKed sequence. - cons, err := s.js.Consumer(ctx, mq.StreamName(), BufferConsumerName) + cons, err := s.js.Consumer(ctx, s.streamName, BufferConsumerName) if err != nil { // Consumer may not exist yet if no messages have been ingested. s.logger.Warn("sweeper: buffer consumer not found (may not exist yet)", "error", err) diff --git a/internal/ingest/sweeper_test.go b/internal/ingest/sweeper_test.go index 22b3ebf2..b4344d7b 100644 --- a/internal/ingest/sweeper_test.go +++ b/internal/ingest/sweeper_test.go @@ -116,7 +116,7 @@ func TestSweep_GapSeqIsBottleneck(t *testing.T) { consumerFn: func(context.Context, string, string) (jetstream.Consumer, error) { return mc, nil }, } - s := NewSweeper(js, gapWindow, nopLogger()) + s := NewSweeper(js, "WAVEHOUSE", gapWindow, nopLogger()) s.sweep(context.Background()) // gapSeq ~101, ackFloor+1 = 151 → target = min(151, 101) = 101 @@ -152,7 +152,7 @@ func TestSweep_AckFloorIsBottleneck(t *testing.T) { consumerFn: func(context.Context, string, string) (jetstream.Consumer, error) { return mc, nil }, } - s := NewSweeper(js, gapWindow, nopLogger()) + s := NewSweeper(js, "WAVEHOUSE", gapWindow, nopLogger()) s.sweep(context.Background()) // gapSeq ~51, ackFloor+1 = 31 → target = min(31, 51) = 31 @@ -184,7 +184,7 @@ func TestSweep_AllWithinWindow_NoPurge(t *testing.T) { consumerFn: func(context.Context, string, string) (jetstream.Consumer, error) { return mc, nil }, } - s := NewSweeper(js, gapWindow, nopLogger()) + s := NewSweeper(js, "WAVEHOUSE", gapWindow, nopLogger()) s.sweep(context.Background()) assert.Empty(t, ms.purged, "no purge when all messages within gap window") @@ -205,7 +205,7 @@ func TestSweep_ConsumerNotFound_NoPurge(t *testing.T) { }, } - s := NewSweeper(js, 5*time.Minute, nopLogger()) + s := NewSweeper(js, "WAVEHOUSE", 5*time.Minute, nopLogger()) s.sweep(context.Background()) assert.Empty(t, ms.purged, "no purge when consumer not found") @@ -219,7 +219,7 @@ func TestSweep_StreamError_NoPurge(t *testing.T) { }, } - s := NewSweeper(js, 5*time.Minute, nopLogger()) + s := NewSweeper(js, "WAVEHOUSE", 5*time.Minute, nopLogger()) // Should not panic. s.sweep(context.Background()) } @@ -249,7 +249,7 @@ func TestSweep_TargetLessOrEqualOne_NoPurge(t *testing.T) { consumerFn: func(context.Context, string, string) (jetstream.Consumer, error) { return mc, nil }, } - s := NewSweeper(js, gapWindow, nopLogger()) + s := NewSweeper(js, "WAVEHOUSE", gapWindow, nopLogger()) s.sweep(context.Background()) // ackFloor+1 = 1, gapSeq = some large number → target = min(1, X) = 1 → skip @@ -271,7 +271,7 @@ func TestSweep_ConsumerInfoError_NoPurge(t *testing.T) { consumerFn: func(context.Context, string, string) (jetstream.Consumer, error) { return mc, nil }, } - s := NewSweeper(js, 5*time.Minute, nopLogger()) + s := NewSweeper(js, "WAVEHOUSE", 5*time.Minute, nopLogger()) s.sweep(context.Background()) assert.Empty(t, ms.purged, "no purge when consumer info fails") @@ -307,7 +307,7 @@ func TestSweep_PurgeError_Logged(t *testing.T) { consumerFn: func(context.Context, string, string) (jetstream.Consumer, error) { return mc, nil }, } - s := NewSweeper(js, gapWindow, nopLogger()) + s := NewSweeper(js, "WAVEHOUSE", gapWindow, nopLogger()) // Should not panic even when purge fails. s.sweep(context.Background()) } @@ -324,7 +324,7 @@ func TestFindGapSequence_EmptyStream(t *testing.T) { }, } - s := NewSweeper(nil, 5*time.Minute, nopLogger()) + s := NewSweeper(nil, "WAVEHOUSE", 5*time.Minute, nopLogger()) seq, err := s.findGapSequence(context.Background(), ms) assert.NoError(t, err) assert.Equal(t, uint64(0), seq) @@ -338,7 +338,7 @@ func TestFindGapSequence_FirstSeqGTLastSeq(t *testing.T) { }, } - s := NewSweeper(nil, 5*time.Minute, nopLogger()) + s := NewSweeper(nil, "WAVEHOUSE", 5*time.Minute, nopLogger()) seq, err := s.findGapSequence(context.Background(), ms) assert.NoError(t, err) assert.Equal(t, uint64(0), seq) @@ -356,7 +356,7 @@ func TestFindGapSequence_AllWithinWindow(t *testing.T) { }, } - s := NewSweeper(nil, 10*time.Minute, nopLogger()) + s := NewSweeper(nil, "WAVEHOUSE", 10*time.Minute, nopLogger()) seq, err := s.findGapSequence(context.Background(), ms) assert.NoError(t, err) assert.Equal(t, uint64(0), seq, "all messages within window → 0") @@ -374,7 +374,7 @@ func TestFindGapSequence_AllExpired(t *testing.T) { }, } - s := NewSweeper(nil, 5*time.Minute, nopLogger()) + s := NewSweeper(nil, "WAVEHOUSE", 5*time.Minute, nopLogger()) seq, err := s.findGapSequence(context.Background(), ms) assert.NoError(t, err) // All expired → result should be last+1 = 101 (binary search default) @@ -400,7 +400,7 @@ func TestFindGapSequence_BoundaryDetection(t *testing.T) { }, } - s := NewSweeper(nil, gapWindow, nopLogger()) + s := NewSweeper(nil, "WAVEHOUSE", gapWindow, nopLogger()) seq, err := s.findGapSequence(context.Background(), ms) assert.NoError(t, err) assert.Equal(t, uint64(61), seq, "first seq within gap window should be 61") @@ -428,7 +428,7 @@ func TestFindGapSequence_SparseSequences(t *testing.T) { }, } - s := NewSweeper(nil, gapWindow, nopLogger()) + s := NewSweeper(nil, "WAVEHOUSE", gapWindow, nopLogger()) seq, err := s.findGapSequence(context.Background(), ms) assert.NoError(t, err) // With sparse seqs, binary search skips missing seqs. @@ -440,7 +440,7 @@ func TestFindGapSequence_StreamInfoError(t *testing.T) { t.Parallel() ms := &mockStream{infoErr: errors.New("stream info unavailable")} - s := NewSweeper(nil, 5*time.Minute, nopLogger()) + s := NewSweeper(nil, "WAVEHOUSE", 5*time.Minute, nopLogger()) _, err := s.findGapSequence(context.Background(), ms) assert.Error(t, err) assert.Contains(t, err.Error(), "stream info") @@ -458,7 +458,7 @@ func TestStart_ContextCancellation(t *testing.T) { }, } - s := NewSweeper(js, 5*time.Minute, nopLogger()) + s := NewSweeper(js, "WAVEHOUSE", 5*time.Minute, nopLogger()) ctx, cancel := context.WithCancel(context.Background()) cancel() // Cancel immediately. diff --git a/internal/mq/embedded_test.go b/internal/mq/embedded_test.go index 2cf31b34..06aff8a5 100644 --- a/internal/mq/embedded_test.go +++ b/internal/mq/embedded_test.go @@ -17,7 +17,7 @@ import ( func newTestEmbedded(t *testing.T) *EmbeddedNATS { t.Helper() logger := slog.New(slog.NewTextHandler(io.Discard, nil)) - e, err := NewEmbedded(t.TempDir(), StreamName(), 64<<20, logger) + e, err := NewEmbedded(t.TempDir(), "WAVEHOUSE", 64<<20, logger) require.NoError(t, err) t.Cleanup(func() { _ = e.Close() }) return e @@ -73,7 +73,7 @@ func TestEmbeddedNATS_Accessors(t *testing.T) { func TestEmbeddedNATS_DefaultLogger(t *testing.T) { // NewEmbedded without a logger should not panic — it falls back to the // default slog logger. - e, err := NewEmbedded(t.TempDir(), StreamName(), 64<<20) + e, err := NewEmbedded(t.TempDir(), "WAVEHOUSE", 64<<20) require.NoError(t, err) t.Cleanup(func() { _ = e.Close() }) } diff --git a/internal/mq/mq.go b/internal/mq/mq.go index 93a6eff2..aaf0e1b7 100644 --- a/internal/mq/mq.go +++ b/internal/mq/mq.go @@ -45,8 +45,3 @@ type Subscriber interface { Subscribe(ctx context.Context, subject, consumerName string, handler func(msg *Message) error) error Close() error } - -// StreamName returns the JetStream stream name used by all MQ implementations. -func StreamName() string { - return "WAVEHOUSE" -} diff --git a/internal/mq/mq_test.go b/internal/mq/mq_test.go index 46d9f81a..9588d506 100644 --- a/internal/mq/mq_test.go +++ b/internal/mq/mq_test.go @@ -7,11 +7,6 @@ import ( "github.com/stretchr/testify/assert" ) -func TestStreamName(t *testing.T) { - t.Parallel() - assert.Equal(t, "WAVEHOUSE", StreamName()) -} - func TestMessage_AckNak(t *testing.T) { t.Parallel() diff --git a/internal/mq/remote_test.go b/internal/mq/remote_test.go index 6f14c866..3f4ba50b 100644 --- a/internal/mq/remote_test.go +++ b/internal/mq/remote_test.go @@ -34,7 +34,7 @@ func startListeningNATS(t *testing.T) string { func TestRemoteNATS_PublishSubscribe(t *testing.T) { url := startListeningNATS(t) - r, err := NewRemote(url, StreamName(), 64<<20) + r, err := NewRemote(url, "WAVEHOUSE", 64<<20) require.NoError(t, err) t.Cleanup(func() { _ = r.Close() }) @@ -73,7 +73,7 @@ func TestRemoteNATS_PublishSubscribe(t *testing.T) { func TestRemoteNATS_Accessors(t *testing.T) { url := startListeningNATS(t) - r, err := NewRemote(url, StreamName(), 64<<20) + r, err := NewRemote(url, "WAVEHOUSE", 64<<20) require.NoError(t, err) t.Cleanup(func() { _ = r.Close() }) @@ -85,6 +85,6 @@ func TestRemoteNATS_Accessors(t *testing.T) { func TestRemoteNATS_BadURL(t *testing.T) { // Connecting to an obviously wrong URL should return an error rather than // hanging or panicking. - _, err := NewRemote("nats://127.0.0.1:1", StreamName(), 1<<20) + _, err := NewRemote("nats://127.0.0.1:1", "WAVEHOUSE", 1<<20) require.Error(t, err) } diff --git a/tests/integration_test.go b/tests/integration_test.go index 969538a7..64f5c077 100644 --- a/tests/integration_test.go +++ b/tests/integration_test.go @@ -214,8 +214,8 @@ func setupTestEnv(t *testing.T) *testEnv { // Wire handlers. ingestHandler := api.NewIngestHandler(registry, embeddedMQ) queryHandler := api.NewQueryHandler(chConn, tiered, 5*time.Second) - sseHandler := api.NewSSEHandler(hub, js) - wsHandler := api.NewWSHandler(hub, js, nil) + sseHandler := api.NewSSEHandler(hub, js, testStream) + wsHandler := api.NewWSHandler(hub, js, nil, testStream) dlqHandler := api.NewDLQHandler(js, testStream, logger) deps := api.Dependencies{