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

[taitelee]

Summary

This PR makes the NATS JetStream name configurable, enabling multi-environment isolation (e.g., separating Staging and Prod traffic). It replaces the previous hardcoded mq.StreamName() global helper with proper dependency injection to ensure the stream name propagates correctly through the entire system.

• Configurable Namespacing: Added stream_name to MQ configuration; supports YAML and env overrides (WH_MQ_STREAM_NAME).
• Refactored Injection: Updated all handlers (SSE, WS, Sweeper, DLQ) and constructors (NewEmbedded, NewRemote) to accept dynamic stream names.
• Removed Global Dependency: Removed mq.StreamName() global helper to enforce clean dependency injection and prevent hardcoding.
• Documentation & Deployment Sync: Updated docs/configuration.md examples and descriptions to match the dynamic config. Added missing env vars to standalone.yaml and clustered.yaml.
• Testing: Added regression test TestEnsureDLQStream_CustomName to explicitly verify dynamic stream derivation logic.

Related Issues

Closes #36

Configuration Example

Via config.yaml:
mq: stream_name: "WH_STAGING" url: "nats://localhost:4222"

[gemini-code-assist]

Summary of Changes

Hello, I'm Gemini Code Assist¹! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed!

This pull request refactors the NATS JetStream integration to support configurable stream names, enabling better multi-environment isolation. By removing the global mq.StreamName() dependency and implementing proper dependency injection across all message queue handlers and constructors, the system now allows for dynamic stream naming via configuration files or environment variables. This change improves system flexibility and maintainability while ensuring that environment-specific traffic can be effectively separated.

Highlights

• Configurable NATS Stream Name: Replaced the hardcoded mq.StreamName() global helper with dynamic configuration, allowing for environment-specific isolation via stream_name in the MQ configuration.
• Dependency Injection Refactor: Updated constructors for SSE, WS, and Sweeper handlers to accept the stream name as a dependency, ensuring consistent propagation throughout the system.
• Deployment & Documentation: Updated standalone.yaml and clustered.yaml to include the new WH_MQ_STREAM_NAME environment variable and refreshed docs/configuration.md with updated examples.
• Testing: Added a regression test TestEnsureDLQStream_CustomName to verify that the system correctly derives and utilizes custom stream names.

Using Gemini Code Assist

The full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips.

Invoking Gemini

You can request assistance from Gemini at any point by creating a comment using either /gemini <command> or @gemini-code-assist <command>. Below is a summary of the supported commands on the current page.

Feature	Command	Description
Code Review	/gemini review	Performs a code review for the current pull request in its current state.
Pull Request Summary	/gemini summary	Provides a summary of the current pull request in its current state.
Comment	@gemini-code-assist	Responds in comments when explicitly tagged, both in pull request comments and review comments.
Help	/gemini help	Displays a list of available commands.

Customization

To customize the Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a .gemini/ folder in the base of the repository. Detailed instructions can be found here.

Limitations & Feedback

Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counterproductive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for GitHub and other Google products, sign up here.

Footnotes

1. Review the Privacy Notices, Generative AI Prohibited Use Policy, Terms of Service, and learn how to configure Gemini Code Assist in GitHub here. Gemini can make mistakes, so double check it and use code with caution. ↩

[gemini-code-assist]

Code Review

This pull request introduces the ability to configure the NATS JetStream name via the WH_MQ_STREAM_NAME environment variable, replacing the hardcoded "WAVEHOUSE" value. Changes include updates to the configuration struct, constructors for SSE, WebSocket, and Sweeper handlers, and corresponding updates to Docker Compose files and documentation. Feedback focuses on improving encapsulation by unexporting the StreamName field in the SSE and WebSocket handlers to align with the project's dependency injection patterns and avoid accidental mutation. Additionally, it is recommended to avoid hardcoding the stream name in test suites to ensure consistent environment isolation.

I am having trouble creating individual review comments. Click here to see my feedback.

internal/api/stream_sse.go (23)

[SHOULD] The StreamName field should be unexported (streamName) to maintain encapsulation and consistency with other components like the Sweeper. Since it is injected via the constructor, there is no need for it to be public, which prevents accidental modification after initialization and adheres to the project's preference for constructor injection over mutable state.

streamName  string

References

1. Constructor injection is preferred, and structs should avoid global mutable state. Unexporting fields that are injected via the constructor maintains encapsulation. ^(link)

internal/api/stream_sse.go (27)

[SHOULD] Update the constructor to use the unexported field name.

return &SSEHandler{Hub: hub, JS: js, streamName: streamName}

internal/api/stream_sse.go (174)

[SHOULD] Update the usage to the unexported field name.

cons, err := h.JS.CreateOrUpdateConsumer(ctx, h.streamName, jetstream.ConsumerConfig{

internal/api/stream_ws.go (39)

[SHOULD] The StreamName field should be unexported (streamName) for encapsulation and consistency. This aligns with the project's dependency injection patterns and prevents external mutation of the handler's configuration.

streamName     string

References

1. Constructor injection is preferred, and structs should avoid global mutable state. Unexporting fields that are injected via the constructor maintains encapsulation. ^(link)

internal/api/stream_ws.go (43)

[SHOULD] Update the constructor to use the unexported field name.

return &WSHandler{Hub: hub, JS: js, AllowedOrigins: allowedOrigins, streamName: streamName}

internal/api/stream_ws.go (253)

[SHOULD] Update the usage to the unexported field name.

cons, err := h.JS.CreateOrUpdateConsumer(ctx, h.streamName, jetstream.ConsumerConfig{

internal/mq/embedded_test.go (20)

[SHOULD] Avoid hardcoding the stream name 'WAVEHOUSE'. To ensure consistent environment isolation, derive related resource names from a common base configuration value rather than hardcoding them. This prevents drift and facilitates maintenance across test suites.

References

1. To ensure consistent environment isolation, derive related resource names from a common base configuration value rather than hardcoding them.

[claude]

Claude finished @taitelee's task in 3m 1s —— View job

---

PR Review: feat(mq): make NATS stream name configurable

• Read AGENTS.md for project conventions
• Read changed files
• Analyze correctness, security, performance, testing, and doc-sync
• Post final review

---

The core dependency-injection refactor is well-executed — the global mq.StreamName() helper is gone, every constructor is threaded correctly, and the config plumbing (struct tag, env var, default, compose files, config.yaml) is complete. Below are the findings that need attention before this ships.

---

Correctness

[MEDIUM] Subjects are not namespaced — shared-NATS multi-environment isolation is broken

The stated goal is "separating Staging and Prod traffic." The stream name is now configurable, but the stream subjects remain hardcoded:

• internal/mq/embedded.go:97 — Subjects: []string{"ingest.>"}
• internal/mq/remote.go:39 — Subjects: []string{"ingest.>"}
• internal/api/dlq.go:60 — Subjects: []string{"dlq.>"}

NATS JetStream enforces that a given subject belongs to at most one stream. If staging creates stream WH_STAGING claiming ingest.>, prod's attempt to create WH_PROD also claiming ingest.> will fail with a subject-overlap error — the worker process won't start. The feature works fine when each environment has its own NATS instance (embedded standalone, or separate clustered deployments), but it silently fails the primary advertised use-case of sharing a NATS cluster across environments.

If shared-NATS isolation is the goal, subjects need a name-derived prefix too (e.g., strings.ToLower(streamName)+".ingest.>"). If the scope is intentionally limited to separate-NATS deployments, the PR description and docs should say so explicitly to avoid misconfiguration.

[LOW] Redundant empty-string default in config.go:Validate()

// config.go:128-130
if c.MQ.StreamName == "" {
    c.MQ.StreamName = "WAVEHOUSE"
}

cleanenv already applies the env-default:"WAVEHOUSE" tag before Validate() runs, so this branch is unreachable. It looks like defensive code but it's dead code — it survives for the case where someone calls Validate() on a manually-constructed Config{...} with an empty MQ struct, which isn't a supported path. Either document that intent or remove it. Fix this →

[LOW] No NATS stream name validation

NATS rejects stream names that contain spaces, ., *, or >. If someone sets WH_MQ_STREAM_NAME=WAVEHOUSE.STAGING, the process exits at startup with an opaque NATS error. config.go:Validate() is the right place to catch this with a clear message. A simple check would do:

// config.go — inside Validate()
if strings.ContainsAny(c.MQ.StreamName, " \t.*>") {
    return fmt.Errorf("mq.stream_name %q contains invalid characters (spaces, ., *, > are not allowed)", c.MQ.StreamName)
}

---

Encapsulation

[MEDIUM] SSEHandler.StreamName and WSHandler.StreamName should be unexported

stream_sse.go:23 and stream_ws.go:39 expose StreamName string as an exported field. The analogous Sweeper.streamName is correctly unexported. The exported PolicyStore field follows a different pattern — it's intentionally settable post-construction in main.go (it's optional configuration applied after the handler is wired). StreamName is a wiring-time value that must be stable for the handler's lifetime; exporting it allows callers to mutate it between requests.

Fixing this is mechanical (unexport + update replayFromNATS call sites), and Gemini already pointed out the specific lines. Fix this →

---

Testing

[LOW] Sweeper and remote tests still hardcode "WAVEHOUSE"

sweeper_test.go uses NewSweeper(js, "WAVEHOUSE", ...) in every test case. remote_test.go does the same. These are fine for unit tests, but no test exercises the full chain — publish, subscribe, sweep, gap-fill — with a non-default stream name through a real embedded NATS. The new TestEnsureDLQStream_CustomName covers the DLQ path well; a parallel smoke test for the main stream path would close the gap and would catch the subject-namespace issue described above if a shared-NATS test fixture were added.

---

Documentation & Consistency Sync

[LOW] docs/deployment.md is missing WH_MQ_STREAM_NAME

The file already lists the sibling env vars (WH_MQ_GAP_WINDOW_MINUTES at line 195, WH_MQ_MAX_BYTES_GB at line 196, WH_MQ_URL at line 203) but not WH_MQ_STREAM_NAME. Per AGENTS.md's consistency sync table, env var additions must update docs/deployment.md alongside docs/configuration.md (done) and the compose files (done). Fix this →

---

Security

No security findings. The stream name flows only to NATS stream lookup/creation calls (never into SQL, logs, or HTTP responses in a way that would introduce injection or data exposure).

---

Verdict

Iterate — address the subject namespace issue (MEDIUM) or explicitly scope-limit the feature description to separate-NATS deployments, and unexport the StreamName fields (MEDIUM). The LOW items can follow in the same commit or a fast-follow.

[EricAndrechek]

Triage note as part of the pre-alpha priority pass.

#89 deliberately removed the configurable stream name because the previous implementation (#36) shipped a silent drift bug: cfg.MQ.StreamName was threaded into mq.NewEmbedded, api.EnsureDLQStream, ingest.StartIngestWorker, and api.NewDLQHandler, but the Sweeper, SSE handler, and WS handler all called the hardcoded mq.StreamName() function (returning "WAVEHOUSE") instead. Setting WH_MQ_STREAM_NAME=PROD would have created the stream as PROD but left the Sweeper looking up WAVEHOUSE — silent state-sync drift, as documented in the #89 changelog entry.

Since WaveHouse is now fully embedded (each instance has its own NATS process), there is no other tenant to namespace against — the configurable stream name was an artifact of the external-NATS era. Reintroducing it would need either (a) a concrete use case where multiple embedded WaveHouse instances share a NATS process (which contradicts the "embedded NATS per binary" architecture), or (b) a multi-tenant story where one binary serves several logical streams (not currently planned).

Without that, this PR re-introduces a config knob with no use case and a known footgun history.

Recommend: close this PR as superseded by the #89 architectural decision. Reopen with a fresh issue + concrete use case if one emerges. Setting Priority to P3 on the board until then.

@taitelee — you opened this re-iteration; is there a use case I'm missing? If not, I'll close.