Gate JetStream stream creation behind BOOTSTRAP_STREAMS across services

CLAUDE.md

@@ -194,7 +194,7 @@ All commands are wrapped in the root Makefile. Always use `make` targets — nev
 - Use `natsutil.ReplyJSON` for success responses, `natsutil.ReplyError` for errors
 - Define all stream configs in `pkg/stream/stream.go` with name pattern `<STREAM>_<siteID>`
 - Use durable consumers named after the service
-- Use `js.CreateOrUpdateStream` at startup — it's idempotent
+- Stream creation is gated by `BOOTSTRAP_STREAMS` (see below); when enabled, use `js.CreateOrUpdateStream` (it's idempotent) via the service's `bootstrapStreams` helper, never inline
 
 ### Event Timestamps
 - Every NATS event struct in `pkg/model` must include a `Timestamp int64 \`json:"timestamp" bson:"timestamp"\`` field
@@ -215,6 +215,8 @@ All commands are wrapped in the root Makefile. Always use `make` targets — nev
 - `ROOMS_{siteID}` — Member invite requests
 - `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

broadcast-worker/bootstrap.go

@@ -0,0 +1,50 @@
+package main
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/nats-io/nats.go/jetstream"
+
+	"github.com/Marz32onE/instrumentation-go/otel-nats/oteljetstream"
+
+	"github.com/hmchangw/chat/pkg/stream"
+)
+
+// bootstrapConfig groups every field that is ONLY meaningful when the
+// service is being stood up in dev or integration tests against a NATS
+// instance where the streams it consumes do not yet exist. In production
+// streams are pre-provisioned by ops/IaC and Bootstrap.Enabled must remain
+// false; the service only creates its own durable consumer.
+type bootstrapConfig struct {
+	// Enabled (BOOTSTRAP_STREAMS) toggles whether the service calls
+	// CreateOrUpdateStream at startup for the streams it consumes.
+	// Leave false in production.
+	Enabled bool `env:"STREAMS" envDefault:"false"`
+}
+
+// streamCreator is the minimal JetStream surface bootstrapStreams depends on.
+// Kept service-local so we don't pollute pkg/ with a one-method type and so
+// tests can inject a fake without mockgen.
+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
+// BOOTSTRAP_STREAMS=true so a developer can stand the service up in isolation
+// against a fresh NATS instance.
+func bootstrapStreams(ctx context.Context, js streamCreator, siteID string, enabled bool) error {
+	if !enabled {
+		return nil
+	}
+	canonicalCfg := stream.MessagesCanonical(siteID)
+	if _, err := js.CreateOrUpdateStream(ctx, jetstream.StreamConfig{
+		Name:     canonicalCfg.Name,
+		Subjects: canonicalCfg.Subjects,
+	}); err != nil {
+		return fmt.Errorf("create MESSAGES_CANONICAL stream: %w", err)
+	}
+	return nil
+}

broadcast-worker/bootstrap_test.go

@@ -0,0 +1,71 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/nats-io/nats.go/jetstream"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/Marz32onE/instrumentation-go/otel-nats/oteljetstream"
+)
+
+type fakeStreamCreator struct {
+	created []string
+	failOn  string // stream name to fail on; empty = never fail
+	failErr error  // error to return when failing
+}
+
+// Returns nil for the Stream value because bootstrapStreams discards it.
+func (f *fakeStreamCreator) CreateOrUpdateStream(_ context.Context, cfg jetstream.StreamConfig) (oteljetstream.Stream, error) { //nolint:gocritic // hugeParam: cfg is passed by value to satisfy the streamCreator interface
+	if f.failOn != "" && cfg.Name == f.failOn {
+		return nil, f.failErr
+	}
+	f.created = append(f.created, cfg.Name)
+	return nil, nil
+}
+
+func TestBootstrapStreams(t *testing.T) {
+	tests := []struct {
+		name        string
+		enabled     bool
+		failOn      string
+		failErr     error
+		wantCreated []string
+		wantErrSub  string
+	}{
+		{
+			name:        "disabled - skips creation",
+			enabled:     false,
+			wantCreated: nil,
+		},
+		{
+			name:        "enabled - creates MESSAGES_CANONICAL",
+			enabled:     true,
+			wantCreated: []string{"MESSAGES_CANONICAL_test"},
+		},
+		{
+			name:       "enabled - wraps MESSAGES_CANONICAL creator error",
+			enabled:    true,
+			failOn:     "MESSAGES_CANONICAL_test",
+			failErr:    errors.New("nats down"),
+			wantErrSub: "create MESSAGES_CANONICAL stream",
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			fake := &fakeStreamCreator{failOn: tc.failOn, failErr: tc.failErr}
+			err := bootstrapStreams(context.Background(), fake, "test", tc.enabled)
+			if tc.wantErrSub != "" {
+				require.Error(t, err)
+				assert.Contains(t, err.Error(), tc.wantErrSub)
+				assert.ErrorIs(t, err, tc.failErr)
+				return
+			}
+			require.NoError(t, err)
+			assert.Equal(t, tc.wantCreated, fake.created)
+		})
+	}
+}

broadcast-worker/deploy/docker-compose.yml

@@ -24,6 +24,7 @@ services:
       - USER_CACHE_TTL=5m
       - VALKEY_ADDR=valkey:6379
       - VALKEY_KEY_GRACE_PERIOD=24h
+      - BOOTSTRAP_STREAMS=true
     volumes:
       - ../../docker-local/backend.creds:/etc/nats/backend.creds:ro
     networks:

broadcast-worker/main.go

@@ -24,19 +24,20 @@ import (
 )
 
 type config struct {
-	NatsURL              string        `env:"NATS_URL"                  envDefault:"nats://localhost:4222"`
-	NatsCredsFile        string        `env:"NATS_CREDS_FILE"           envDefault:""`
-	SiteID               string        `env:"SITE_ID"                   envDefault:"default"`
-	MongoURI             string        `env:"MONGO_URI"                 envDefault:"mongodb://localhost:27017"`
-	MongoDB              string        `env:"MONGO_DB"                  envDefault:"chat"`
-	MongoUsername        string        `env:"MONGO_USERNAME"            envDefault:""`
-	MongoPassword        string        `env:"MONGO_PASSWORD"            envDefault:""`
-	MaxWorkers           int           `env:"MAX_WORKERS"               envDefault:"100"`
-	UserCacheSize        int           `env:"USER_CACHE_SIZE"           envDefault:"10000"`
-	UserCacheTTL         time.Duration `env:"USER_CACHE_TTL"            envDefault:"5m"`
-	ValkeyAddr           string        `env:"VALKEY_ADDR,required"`
-	ValkeyPassword       string        `env:"VALKEY_PASSWORD"           envDefault:""`
-	ValkeyKeyGracePeriod time.Duration `env:"VALKEY_KEY_GRACE_PERIOD,required"`
+	NatsURL              string          `env:"NATS_URL"                  envDefault:"nats://localhost:4222"`
+	NatsCredsFile        string          `env:"NATS_CREDS_FILE"           envDefault:""`
+	SiteID               string          `env:"SITE_ID"                   envDefault:"default"`
+	MongoURI             string          `env:"MONGO_URI"                 envDefault:"mongodb://localhost:27017"`
+	MongoDB              string          `env:"MONGO_DB"                  envDefault:"chat"`
+	MongoUsername        string          `env:"MONGO_USERNAME"            envDefault:""`
+	MongoPassword        string          `env:"MONGO_PASSWORD"            envDefault:""`
+	MaxWorkers           int             `env:"MAX_WORKERS"               envDefault:"100"`
+	UserCacheSize        int             `env:"USER_CACHE_SIZE"           envDefault:"10000"`
+	UserCacheTTL         time.Duration   `env:"USER_CACHE_TTL"            envDefault:"5m"`
+	ValkeyAddr           string          `env:"VALKEY_ADDR,required"`
+	ValkeyPassword       string          `env:"VALKEY_PASSWORD"           envDefault:""`
+	ValkeyKeyGracePeriod time.Duration   `env:"VALKEY_KEY_GRACE_PERIOD,required"`
+	Bootstrap            bootstrapConfig `envPrefix:"BOOTSTRAP_"`
 }
 
 func main() {
@@ -93,15 +94,13 @@ func main() {
 		os.Exit(1)
 	}
 
-	canonicalCfg := stream.MessagesCanonical(cfg.SiteID)
-	if _, err = js.CreateOrUpdateStream(ctx, jetstream.StreamConfig{
-		Name:     canonicalCfg.Name,
-		Subjects: canonicalCfg.Subjects,
-	}); err != nil {
-		slog.Error("create MESSAGES_CANONICAL stream failed", "error", err)
+	if err := bootstrapStreams(ctx, js, cfg.SiteID, cfg.Bootstrap.Enabled); err != nil {
+		slog.Error("bootstrap streams failed", "error", err)
 		os.Exit(1)
 	}
 
+	canonicalCfg := stream.MessagesCanonical(cfg.SiteID)
+
 	cons, err := js.CreateOrUpdateConsumer(ctx, canonicalCfg.Name, jetstream.ConsumerConfig{
 		Durable:   "broadcast-worker",
 		AckPolicy: jetstream.AckExplicitPolicy,