feat(mq): make NATS stream name configurable (Re-Iteration)

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{

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")

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{

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"

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"

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

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)
+}

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 },

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,

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,

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)