refactor(streams): inbox-worker owns INBOX schema; remove search-sync-worker scaffolding

inbox-worker is now the sole owner of INBOX_{siteID} schema bootstrap.
Its bootstrapStreams helper passes Name + Subjects from pkg/stream.Inbox,
matching the canonical schema. Federation (Sources + SubjectTransforms)
remains owned by ops/IaC and never appears in app code.

search-sync-worker stops bootstrapping INBOX. The dead scaffolding
inboxBootstrapStreamConfig (originally added with PR #109 and flagged
in its own doc comment as a temporary stand-in until inbox-worker
migrated) is deleted along with its dedicated unit-test file. The
RemoteSiteIDs config field is removed since its only consumer was the
deleted function. The service's bootstrap loop now skips INBOX while
continuing to create consumers for INBOX-based collections (spotlight,
user-room).

CLAUDE.md gains an explicit ownership rule under "JetStream Streams"
documenting that app code owns Name + Subjects and ops/IaC owns
Sources + SubjectTransforms.

Tests: inbox-worker bootstrap test now asserts Subjects matches
pkg/stream.Inbox(siteID).Subjects and Sources is empty (regression
guard against re-introducing federation in app code).

https://claude.ai/code/session_015Cu3UPeWDU4DaJwP7JZtvc

Author: claude

CLAUDE.md

@@ -216,6 +216,7 @@ All commands are wrapped in the root Makefile. Always use `make` targets — nev
 - `OUTBOX_{siteID}` — Cross-site outbound events
 - `INBOX_{siteID}` — Cross-site inbound events (sourced from remote OUTBOX)
 - **Stream bootstrap is opt-in.** Services that consume from or publish to a stream MUST NOT create it in production — streams are owned by ops/IaC. Each such service's `config` includes `Bootstrap bootstrapConfig` (env prefix `BOOTSTRAP_`) with a single `Enabled` field tagged `env:"STREAMS" envDefault:"false"`. The service's `bootstrap.go` defines a `bootstrapStreams(ctx, js, siteID, enabled) error` helper that no-ops when `Enabled=false`. Local `deploy/docker-compose.yml` sets `BOOTSTRAP_STREAMS=true` so any service can stand up against a fresh NATS in dev. New services that interact with JetStream MUST follow this convention.
+- **Stream bootstrap ownership.** When a service does bootstrap a stream in dev, the helper sets ONLY the stream's schema — `Name + Subjects` from `pkg/stream.<Stream>(siteID)`. Federation config (`Sources` + `SubjectTransforms` for cross-site sourcing) is owned by ops/IaC and MUST NOT appear in any service's `bootstrap.go`. INBOX has a single owning service (`inbox-worker`); other services that consume from INBOX (e.g., `search-sync-worker`) skip it in their bootstrap loop and rely on `inbox-worker` to create the stream.
 
 ### MongoDB
 - Never use ORMs (no GORM, no ent) — use native drivers directly

inbox-worker/bootstrap.go

@@ -30,24 +30,24 @@ type streamCreator interface {
 	CreateOrUpdateStream(ctx context.Context, cfg jetstream.StreamConfig) (oteljetstream.Stream, error)
 }
 
-// bootstrapStreams creates the JetStream streams this service consumes from.
-// No-op when enabled is false (the production path) — streams are owned by
-// ops/IaC. In dev/integration the local docker-compose sets
+// bootstrapStreams creates the JetStream INBOX stream this service consumes
+// from. No-op when enabled is false (the production path) — streams are owned
+// by ops/IaC there. In dev/integration the local docker-compose sets
 // BOOTSTRAP_STREAMS=true so a developer can stand the service up in isolation
 // against a fresh NATS instance.
 //
-// Note: Subjects is intentionally omitted from the StreamConfig. In production
-// the INBOX stream's Sources and SubjectTransforms (cross-site OUTBOX→INBOX
-// sourcing) are configured by ops/IaC. This helper only creates the stream
-// with its Name so the consumer can be attached; the sourcing config is
-// managed externally and must not be overwritten here.
+// Ownership rule: this helper sets only the stream schema (Name + Subjects)
+// from pkg/stream.Inbox. Federation config (Sources + SubjectTransforms for
+// cross-site OUTBOX→INBOX sourcing) belongs to ops/IaC and is layered on in
+// production. App code never sets it.
 func bootstrapStreams(ctx context.Context, js streamCreator, siteID string, enabled bool) error {
 	if !enabled {
 		return nil
 	}
 	inboxCfg := stream.Inbox(siteID)
 	if _, err := js.CreateOrUpdateStream(ctx, jetstream.StreamConfig{
-		Name: inboxCfg.Name,
+		Name:     inboxCfg.Name,
+		Subjects: inboxCfg.Subjects,
 	}); err != nil {
 		return fmt.Errorf("create INBOX stream: %w", err)
 	}

inbox-worker/bootstrap_test.go

@@ -10,6 +10,8 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/Marz32onE/instrumentation-go/otel-nats/oteljetstream"
+
+	"github.com/hmchangw/chat/pkg/stream"
 )
 
 type fakeStreamCreator struct {
@@ -42,7 +44,7 @@ func TestBootstrapStreams(t *testing.T) {
 			wantCreated: nil,
 		},
 		{
-			name:        "enabled - creates INBOX",
+			name:        "enabled - creates INBOX with Name and Subjects",
 			enabled:     true,
 			wantCreated: []string{"INBOX_test"},
 		},
@@ -66,13 +68,16 @@ func TestBootstrapStreams(t *testing.T) {
 			}
 			require.NoError(t, err)
 			require.Len(t, fake.created, len(tc.wantCreated))
+			wantSubjects := stream.Inbox("test").Subjects
 			for i, wantName := range tc.wantCreated {
 				assert.Equal(t, wantName, fake.created[i].Name)
-				// The INBOX stream is created Name-only; ops/IaC owns
-				// cross-site Sources + SubjectTransforms. Lock that
-				// invariant in the test so a future "fix" that adds
-				// Subjects fails loudly here.
-				assert.Empty(t, fake.created[i].Subjects, "INBOX bootstrap must not set Subjects")
+				// App owns the schema (Name + Subjects). Federation
+				// (Sources + SubjectTransforms) belongs to ops/IaC and
+				// must not appear here.
+				assert.Equal(t, wantSubjects, fake.created[i].Subjects,
+					"INBOX bootstrap must set Subjects from pkg/stream.Inbox")
+				assert.Empty(t, fake.created[i].Sources,
+					"federation Sources are owned by ops/IaC and must not be set in app code")
 			}
 		})
 	}

search-sync-worker/inbox_stream.go

@@ -11,47 +11,6 @@ import (
 	"github.com/hmchangw/chat/pkg/subject"
 )
 
-// inboxBootstrapStreamConfig returns the INBOX stream config with cross-site
-// Sources + SubjectTransforms layered on top of the canonical baseline from
-// pkg/stream.Inbox. Only used from the bootstrap path in main.go when
-// BOOTSTRAP_STREAMS=true. In production, inbox-worker owns INBOX creation
-// and search-sync-worker never calls this.
-//
-// Federation mechanics: for each remote site we add a StreamSource pointing
-// at `OUTBOX_{remote}` with a SubjectTransform whose `Source` field acts
-// as both the filter and the rewrite source — `outbox.{remote}.to.{siteID}.>`
-// is matched against the upstream OUTBOX and rewritten to
-// `chat.inbox.{siteID}.aggregate.>` on ingest. (NATS JetStream forbids
-// setting `FilterSubject` and `SubjectTransforms` together on the same
-// source — they are mutually exclusive options; the transform's `Source`
-// covers the filter responsibility.) This lets consumers tell local events
-// apart from federated ones by the presence of the `aggregate` segment.
-//
-// This helper stays local to search-sync-worker because it's bootstrap-only
-// — inbox-worker will own an equivalent construction (as a proper feature,
-// not a test toggle) when it migrates in its own PR.
-//
-// Requires NATS Server 2.10+ for SubjectTransforms support.
-func inboxBootstrapStreamConfig(siteID string, remoteSiteIDs []string) jetstream.StreamConfig {
-	baseline := stream.Inbox(siteID)
-	destPattern := fmt.Sprintf("chat.inbox.%s.aggregate.>", siteID)
-	sources := make([]*jetstream.StreamSource, 0, len(remoteSiteIDs))
-	for _, remote := range remoteSiteIDs {
-		sourcePattern := fmt.Sprintf("outbox.%s.to.%s.>", remote, siteID)
-		sources = append(sources, &jetstream.StreamSource{
-			Name: fmt.Sprintf("OUTBOX_%s", remote),
-			SubjectTransforms: []jetstream.SubjectTransformConfig{
-				{Source: sourcePattern, Destination: destPattern},
-			},
-		})
-	}
-	return jetstream.StreamConfig{
-		Name:     baseline.Name,
-		Subjects: baseline.Subjects,
-		Sources:  sources,
-	}
-}
-
 // inboxMemberCollection is the shared base for collections that index
 // subscription lifecycle events (member_added, member_removed) off the
 // INBOX stream. It centralizes stream config and subject filters so
@@ -60,9 +19,9 @@ func inboxBootstrapStreamConfig(siteID string, remoteSiteIDs []string) jetstream
 //
 // The stream name + local subject pattern come straight from pkg/stream.Inbox
 // so there's one canonical definition for every consumer of INBOX.
-// Cross-site federation (Sources + SubjectTransforms) is a deployment
-// concern owned by whichever service creates the INBOX stream and is
-// layered on separately — see inboxBootstrapStreamConfig.
+// inbox-worker owns INBOX schema bootstrap; cross-site federation (Sources
+// + SubjectTransforms) is owned by ops/IaC. search-sync-worker is a pure
+// consumer of INBOX.
 type inboxMemberCollection struct{}
 
 func (b *inboxMemberCollection) StreamConfig(siteID string) jetstream.StreamConfig {

search-sync-worker/inbox_stream_test.go

@@ -21,24 +21,22 @@ import (
 
 // bootstrapConfig groups every field that is ONLY meaningful when the worker
 // is being stood up in dev or integration tests without its normal upstream
-// services. In production none of these fields should be set — streams are
-// owned by their publisher services (message-gatekeeper for
-// MESSAGES_CANONICAL, inbox-worker for INBOX) and search-sync-worker only
-// manages its own durable consumers.
+// services. In production Enabled must remain false — streams are owned by
+// their publisher services (message-gatekeeper for MESSAGES_CANONICAL,
+// inbox-worker for INBOX) and search-sync-worker only manages its own
+// durable consumers.
+//
+// search-sync-worker NEVER bootstraps INBOX, even when Enabled=true; that
+// stream's schema is owned by inbox-worker and its federation by ops/IaC.
 //
 // Env vars in this group are all prefixed `BOOTSTRAP_` so they're easy to
 // spot in deployment manifests and obvious to grep.
 type bootstrapConfig struct {
 	// Enabled (BOOTSTRAP_STREAMS) toggles whether the worker calls
 	// CreateOrUpdateStream at startup for each collection's stream. Leave
-	// false in production.
+	// false in production. INBOX is intentionally excluded from this loop
+	// — inbox-worker owns INBOX schema bootstrap.
 	Enabled bool `env:"STREAMS" envDefault:"false"`
-	// RemoteSiteIDs (BOOTSTRAP_REMOTE_SITE_IDS) lists the other sites whose
-	// OUTBOX streams should be sourced into this site's INBOX when the
-	// worker is creating it itself. Used to build the cross-site Sources +
-	// SubjectTransforms config during bootstrap. Only consulted when
-	// Enabled is true; unused in production.
-	RemoteSiteIDs []string `env:"REMOTE_SITE_IDS" envSeparator:","`
 }
 
 type config struct {
@@ -165,29 +163,22 @@ func main() {
 	// we don't redundantly call CreateOrUpdateStream per collection.
 	createdStreams := make(map[string]struct{}, len(collections))
 
-	// Canonical INBOX stream name, used below to decide when to layer on
-	// cross-site Sources + SubjectTransforms during bootstrap.
+	// INBOX is owned by inbox-worker — see the skip in the loop below.
 	inboxName := stream.Inbox(cfg.SiteID).Name
 
 	for _, coll := range collections {
 		streamCfg := coll.StreamConfig(cfg.SiteID)
-		if cfg.Bootstrap.Enabled {
-			bootstrapCfg := streamCfg
-			// The INBOX stream is the only one that needs cross-site Sources
-			// + SubjectTransforms. Collections return a minimal baseline
-			// (name + local subjects from pkg/stream.Inbox) and the
-			// bootstrap path layers on the federation config here, keeping
-			// the cross-site topology out of the Collection type entirely.
-			if streamCfg.Name == inboxName {
-				bootstrapCfg = inboxBootstrapStreamConfig(cfg.SiteID, cfg.Bootstrap.RemoteSiteIDs)
-			}
-			if _, alreadyCreated := createdStreams[bootstrapCfg.Name]; !alreadyCreated {
-				if _, err := js.CreateOrUpdateStream(ctx, bootstrapCfg); err != nil {
-					slog.Error("create stream failed", "stream", bootstrapCfg.Name, "error", err)
+		// Skip INBOX bootstrap — inbox-worker owns its schema, ops/IaC
+		// owns its federation. Consumer creation still runs for
+		// INBOX-based collections (spotlight, user-room).
+		if cfg.Bootstrap.Enabled && streamCfg.Name != inboxName {
+			if _, alreadyCreated := createdStreams[streamCfg.Name]; !alreadyCreated {
+				if _, err := js.CreateOrUpdateStream(ctx, streamCfg); err != nil {
+					slog.Error("create stream failed", "stream", streamCfg.Name, "error", err)
 					os.Exit(1)
 				}
-				createdStreams[bootstrapCfg.Name] = struct{}{}
-				slog.Info("stream bootstrapped", "stream", bootstrapCfg.Name)
+				createdStreams[streamCfg.Name] = struct{}{}
+				slog.Info("stream bootstrapped", "stream", streamCfg.Name)
 			}
 		}