fix(data-migration): post-review changes for the collections CDC migration

Collapses all changes made in response to review (Nitish, CodeRabbit, and a
deep self-review) into one commit on top of the feature commit, so the
review-response delta reads as a single diff.

Subscriptions / inbox-worker (review fixes, kept):
- Missing-subscription mute/favorite/read now Nak (like roles) instead of a
  silent no-op, so a field event that races ahead of member_added redelivers
  until the sub lands (extends Nitish's roles observation to the siblings).
- Mute/favorite/role events stamp the source _updatedAt as their high-water
  mark, so a redelivered insert snapshot can't out-rank a newer update; a
  cleared roles set now propagates.

Transformer correctness (CodeRabbit):
- Degraded insert/replace recovers the doc via a source lookup instead of
  poisoning; a combined name+ro update emits both room_renamed and
  room_restricted; zero-time guards keep absent source timestamps from
  becoming year-0001 high-water values; required config scalars are trimmed
  and validated non-empty; SOURCE_DATA read timestamp = max(ls, lr); bare
  returns wrapped with context.

Spec-conformance revert (the design spec + CDC_COVERAGE are decision-complete):
- Subscription true-delete is un-actionable -> skip + metric (spec 4.0/4.3,
  coverage row 8). Removed the out-of-spec subscription_deleted machinery:
  InboxSubscriptionDeleted, SubscriptionDeletedEvent, MemberAddEvent.SubID,
  handleSubscriptionDeleted, DeleteSubscriptionByID, and the source-_id
  adoption in handleMemberAdded (back to generated UUIDv7).
- Post-seed user updates are NOT propagated (spec 4.1/9, coverage row 15).
  Removed the user_status_updated fan-out: publishUserStatus, the statusText
  update path, ALL_SITE_IDS config + wiring, and the empty-sites startup
  warning. inbox-worker UpdateUserStatus reverted to its base last-write-wins
  form (drops the statusUpdatedAt guard, which only supported the fan-out).

This keeps inbox-worker at the single spec-sanctioned change (handleMemberAdded
skip->error) and removes the subscription_deleted membership-lane ordering
hazard at its root. Build, full unit suite, lint, integration-compile, and
gosec all green.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012X9qhQT4NwmCHjdndwNtFD

Author: claude

data-migration/SOURCE_DATA.md

@@ -131,7 +131,7 @@ One row per (user, room). ✅ Unique index `{ rid:1, 'u._id':1 }`.
 | `ts` | date | Join time (set once, stable across re-joins) | ✅ |
 | `roles[]` | string[] | `owner`/`moderator`/`leader`/`user` (role-based ownership) | ✅ |
 | `ls` | date | Last **seen** (scrolled cursor) | ✅ |
-| `lr` | date | Last **read** (explicit mark) — *we plan to use this* | ✅ |
+| `lr` | date | Last **read** (explicit mark) | ✅ |
 | `alert` | bool | True if **any** unread content (not just mentions) | ✅ |
 | `userMentions` / `groupMentions` | int | Unread `@user` / `@all`,`@here` counts | ✅ |
 | `tunread[]` | string[] | Parent-message ids (`tmid`) of threads with any unread | ✅ |
@@ -142,7 +142,7 @@ One row per (user, room). ✅ Unique index `{ rid:1, 'u._id':1 }`.
 | `name` / `fname` | string | Machine name / friendly display name | ✅ |
 | `federation.origin` | string (opt) | Origin site (assumed consistent with room) | ✅ ❓ |
 
-Derived: "has mention" = `userMentions>0 || groupMentions>0`; "muted" = `disableNotifications`. **Open:** read timestamp = `lr` or `ls`? (we lean `lr`).
+Derived: "has mention" = `userMentions>0 || groupMentions>0`; "muted" = `disableNotifications`; **read timestamp (`lastSeenAt`) = `max(ls, lr)`** (resolved per design D1 — the furthest point consumed by either the scrolled cursor or the explicit mark-read).
 
 ## 5. `tsmc_thread_subscriptions`
 

data-migration/oplog-collections-transformer/config.go

@@ -12,10 +12,6 @@ import (
 type config struct {
 	SiteID string `env:"SITE_ID,required"`
 
-	// AllSiteIDs is every federated site. statusText updates fan to all of them (incl.
-	// ours) — status is global-visibility, not home-routed like rooms/subs.
-	AllSiteIDs []string `env:"ALL_SITE_IDS" envDefault:"" envSeparator:","`
-
 	NatsURL       string `env:"NATS_URL,required"`
 	NatsCredsFile string `env:"NATS_CREDS_FILE" envDefault:""`
 
@@ -60,11 +56,22 @@ func parseConfig() (config, error) {
 	if err != nil {
 		return config{}, fmt.Errorf("parse config: %w", err)
 	}
+	// caarlos0/env `required` only rejects an unset var, not a whitespace-only one. Trim and
+	// re-validate the required scalars too, so a value like "  " fails here rather than breaking
+	// subject building / connections later at runtime.
+	cfg.SiteID = strings.TrimSpace(cfg.SiteID)
+	cfg.NatsURL = strings.TrimSpace(cfg.NatsURL)
+	cfg.SourceMongoURI = strings.TrimSpace(cfg.SourceMongoURI)
+	cfg.TargetMongoURI = strings.TrimSpace(cfg.TargetMongoURI)
 	cfg.RoomsCollection = strings.TrimSpace(cfg.RoomsCollection)
 	cfg.SubscriptionsCollection = strings.TrimSpace(cfg.SubscriptionsCollection)
 	cfg.ThreadSubsCollection = strings.TrimSpace(cfg.ThreadSubsCollection)
 	cfg.UsersCollection = strings.TrimSpace(cfg.UsersCollection)
 	for name, v := range map[string]string{
+		"SITE_ID":                  cfg.SiteID,
+		"NATS_URL":                 cfg.NatsURL,
+		"SOURCE_MONGO_URI":         cfg.SourceMongoURI,
+		"TARGET_MONGO_URI":         cfg.TargetMongoURI,
 		"ROOMS_COLLECTION":         cfg.RoomsCollection,
 		"SUBSCRIPTIONS_COLLECTION": cfg.SubscriptionsCollection,
 		"THREAD_SUBS_COLLECTION":   cfg.ThreadSubsCollection,

data-migration/oplog-collections-transformer/config_test.go

@@ -137,3 +137,19 @@ func TestParseConfig_CustomCollectionNames(t *testing.T) {
 	assert.Equal(t, "my_thread_subs", cfg.ThreadSubsCollection)
 	assert.Equal(t, "my_users", cfg.UsersCollection)
 }
+
+func TestParseConfig_WhitespaceSiteID_Errors(t *testing.T) {
+	setRequiredEnv(t)
+	t.Setenv("SITE_ID", "   ")
+	_, err := parseConfig()
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "SITE_ID must be non-empty")
+}
+
+func TestParseConfig_WhitespaceSourceMongoURI_Errors(t *testing.T) {
+	setRequiredEnv(t)
+	t.Setenv("SOURCE_MONGO_URI", "  ")
+	_, err := parseConfig()
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "SOURCE_MONGO_URI must be non-empty")
+}

data-migration/oplog-collections-transformer/handler.go

@@ -47,7 +47,6 @@ type targetStore interface {
 
 type handler struct {
 	siteID         string
-	allSiteIDs     []string
 	roomsColl      string
 	subsColl       string
 	threadSubsColl string
@@ -99,31 +98,20 @@ func (h *handler) resolveDoc(ctx context.Context, ev oplogEvent) (doc []byte, sk
 	switch ev.Op {
 	case "insert", "replace":
 		if len(ev.FullDocument) == 0 {
-			// The connector always carries the doc for insert/replace — a missing one is a
-			// contract violation that can never succeed on redelivery. Poison.
-			return nil, false, fmt.Errorf("%w: %s without fullDocument", migration.ErrPoison, ev.Op)
+			if !ev.Degraded {
+				// The connector always carries the doc for insert/replace — a non-degraded missing
+				// one is a contract violation that can never succeed on redelivery. Poison.
+				return nil, false, fmt.Errorf("%w: %s without fullDocument", migration.ErrPoison, ev.Op)
+			}
+			// Degraded: the connector couldn't encode fullDocument (left nil) but still published.
+			// Recover the live source doc by _id rather than drop it — mirrors oplog-transformer.
+			slog.Warn("recovering degraded insert/replace via source lookup",
+				"eventId", ev.EventID, "reason", ev.DegradedReason, "request_id", natsutil.RequestIDFromContext(ctx))
+			return h.resolveBySourceLookup(ctx, ev)
 		}
 		return ev.FullDocument, false, nil
 	case "update":
-		id, idErr := documentKeyID(ev.DocumentKey)
-		if idErr != nil {
-			return nil, false, idErr
-		}
-		lk := h.lookups[ev.Collection]
-		if lk == nil {
-			// An update for a collection we don't have a source lookup for is a misconfiguration
-			// (filter subjects and the lookups map disagree) — it can never succeed. Poison.
-			return nil, false, fmt.Errorf("%w: no source lookup for collection %q", migration.ErrPoison, ev.Collection)
-		}
-		got, lookupErr := lk.FindByID(ctx, id)
-		if lookupErr != nil {
-			return nil, false, fmt.Errorf("lookup %q: %w", id, lookupErr)
-		}
-		if got == nil {
-			// Doc vanished from source between the change event and our re-read — nothing to apply.
-			return nil, true, nil
-		}
-		return got, false, nil
+		return h.resolveBySourceLookup(ctx, ev)
 	case "delete":
 		// Un-actionable: only documentKey._id, and the destination doesn't key by source _id.
 		return nil, true, nil
@@ -133,6 +121,33 @@ func (h *handler) resolveDoc(ctx context.Context, ev oplogEvent) (doc []byte, sk
 	}
 }
 
+// resolveBySourceLookup re-reads the full current source doc by documentKey._id — used for updates
+// (the connector forwards only the delta) and degraded insert/replace (fullDocument couldn't encode).
+// skip=true when the doc vanished from source between the event and our re-read.
+//
+//nolint:gocritic // ev passed by value to mirror resolveDoc's signature; off the hot path.
+func (h *handler) resolveBySourceLookup(ctx context.Context, ev oplogEvent) (doc []byte, skip bool, err error) {
+	id, idErr := documentKeyID(ev.DocumentKey)
+	if idErr != nil {
+		return nil, false, idErr
+	}
+	lk := h.lookups[ev.Collection]
+	if lk == nil {
+		// No source lookup for this collection is a misconfiguration (filter subjects and the
+		// lookups map disagree) — it can never succeed. Poison.
+		return nil, false, fmt.Errorf("%w: no source lookup for collection %q", migration.ErrPoison, ev.Collection)
+	}
+	got, lookupErr := lk.FindByID(ctx, id)
+	if lookupErr != nil {
+		return nil, false, fmt.Errorf("lookup %q: %w", id, lookupErr)
+	}
+	if got == nil {
+		// Doc vanished from source between the change event and our re-read — nothing to apply.
+		return nil, true, nil
+	}
+	return got, false, nil
+}
+
 // documentKeyID decodes documentKey → _id (the common string case). Returns migration.ErrPoison
 // when missing/malformed — mirrors the message transformer's documentKeyID.
 func documentKeyID(documentKey json.RawMessage) (string, error) {

data-migration/oplog-collections-transformer/handler_test.go

@@ -73,7 +73,6 @@ const (
 func newTestHandler(pub inboxPublisher, target targetStore, lookup migration.SourceLookup) *handler {
 	return &handler{
 		siteID:         testSiteID,
-		allSiteIDs:     []string{testSiteID, "s2"},
 		roomsColl:      roomsColl,
 		subsColl:       subsColl,
 		threadSubsColl: threadSubColl,
@@ -119,11 +118,13 @@ func TestHandle_Dispatch(t *testing.T) {
 		assert.NotEmpty(t, pub.events)
 	})
 
-	t.Run("thread-subs delete routes to handleThreadSub and skips", func(t *testing.T) {
+	t.Run("thread-subs routes to handleThreadSub (poison is branch-specific)", func(t *testing.T) {
 		h := newTestHandler(&fakePublisher{}, &fakeTarget{}, &fakeLookup{})
-		err := h.handle(context.Background(), oplogEvent{Op: "delete", Collection: threadSubColl,
-			DocumentKey: json.RawMessage(`{"_id":"ts1"}`)})
-		assert.ErrorIs(t, err, migration.ErrSkipped)
+		// Empty parentMessage._id poisons ONLY on the thread-sub branch; the default-skip would
+		// return ErrSkipped, so ErrPoison proves the event actually reached handleThreadSub.
+		err := h.handle(context.Background(), oplogEvent{Op: "insert", Collection: threadSubColl,
+			FullDocument: json.RawMessage(`{"_id":"ts1","u":{"username":"alice"},"parentMessage":{"_id":""}}`)})
+		assert.ErrorIs(t, err, migration.ErrPoison)
 	})
 
 	t.Run("unknown collection skipped", func(t *testing.T) {
@@ -157,6 +158,7 @@ func TestHandleRoom_PublishError(t *testing.T) {
 	err := h.handleRoom(context.Background(), roomEv("insert", doc, ""))
 	require.Error(t, err)
 	assert.NotErrorIs(t, err, migration.ErrSkipped)
+	assert.NotErrorIs(t, err, migration.ErrPoison, "a transient publish failure must Nak, not poison")
 }
 
 func TestHandleUser_UpsertError(t *testing.T) {
@@ -165,6 +167,7 @@ func TestHandleUser_UpsertError(t *testing.T) {
 	err := h.handleUser(context.Background(), userEv("insert", `{"_id":"u1","username":"alice"}`))
 	require.Error(t, err)
 	assert.NotErrorIs(t, err, migration.ErrSkipped)
+	assert.NotErrorIs(t, err, migration.ErrPoison, "a transient upsert failure must Nak, not poison")
 }
 
 func TestResolveDoc(t *testing.T) {

data-migration/oplog-collections-transformer/inboxpublisher_test.go

@@ -27,9 +27,11 @@ func TestJetstreamPublisher_Publish(t *testing.T) {
 			},
 		}
 
+		// Distinct SiteID (home) vs DestSiteID (routing) so a source/dest mix-up in subject
+		// routing would be caught — the subject must use DestSiteID.
 		evt := model.InboxEvent{
 			Type:       "room_sync",
-			SiteID:     site,
+			SiteID:     "home-site",
 			DestSiteID: site,
 			Payload:    []byte(`{"id":"r1"}`),
 			Timestamp:  1700000000000,
@@ -39,8 +41,8 @@ func TestJetstreamPublisher_Publish(t *testing.T) {
 		require.NoError(t, err)
 		require.NotNil(t, captured)
 
-		wantSubject := subject.InboxExternal(site, "room_sync")
-		assert.Equal(t, wantSubject, captured.Subject, "subject must be chat.inbox.s1.aggregate.room_sync")
+		wantSubject := subject.InboxExternal(evt.DestSiteID, "room_sync")
+		assert.Equal(t, wantSubject, captured.Subject, "subject must route on DestSiteID, not SiteID")
 
 		var got model.InboxEvent
 		require.NoError(t, json.Unmarshal(captured.Data, &got))

data-migration/oplog-collections-transformer/main.go

@@ -131,7 +131,6 @@ func main() {
 
 	h := &handler{
 		siteID:         cfg.SiteID,
-		allSiteIDs:     cfg.AllSiteIDs,
 		roomsColl:      cfg.RoomsCollection,
 		subsColl:       cfg.SubscriptionsCollection,
 		threadSubsColl: cfg.ThreadSubsCollection,

data-migration/oplog-collections-transformer/metrics_test.go

@@ -43,6 +43,9 @@ func TestMetrics_NilSafe(t *testing.T) {
 // TestMetrics_DispositionCountersCarryCollection verifies the disposition counters are labelled
 // by both op and collection, so ops can see which source collection is stuck/poisoning.
 func TestMetrics_DispositionCountersCarryCollection(t *testing.T) {
+	// Restore the global meter provider so this test doesn't leak its manual reader into siblings.
+	prev := otel.GetMeterProvider()
+	t.Cleanup(func() { otel.SetMeterProvider(prev) })
 	reader := sdkmetric.NewManualReader()
 	otel.SetMeterProvider(sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader)))
 

data-migration/oplog-collections-transformer/rooms.go

@@ -69,7 +69,7 @@ func (h *handler) handleRoom(ctx context.Context, ev oplogEvent) error {
 
 	doc, skip, err := h.resolveDoc(ctx, ev)
 	if err != nil {
-		return err
+		return fmt.Errorf("resolve room doc: %w", err)
 	}
 	if skip {
 		h.metrics.onSkipped(ctx, ev.Op+"_skip")
@@ -120,18 +120,18 @@ func (h *handler) handleRoom(ctx context.Context, ev oplogEvent) error {
 
 	evts, err := h.roomEvents(ev, &room)
 	if err != nil {
-		return err
+		return fmt.Errorf("build room events: %w", err)
 	}
 	for _, evt := range evts {
 		if err := h.pub.Publish(ctx, evt); err != nil {
-			return err
+			return fmt.Errorf("publish room event %q: %w", evt.Type, err)
 		}
 	}
 	return nil
 }
 
-// roomEvents builds the InboxEvents for a room change: always room_sync, preceded by
-// room_renamed (name/fname changed) or room_restricted (ro changed) on the matching update.
+// roomEvents builds the InboxEvents for a room change: always room_sync, preceded by room_renamed
+// (name/fname changed) and/or room_restricted (ro changed) — both when a single update changes both.
 //
 //nolint:gocritic // ev passed by value to mirror handle's signature; off the hot path.
 func (h *handler) roomEvents(ev oplogEvent, room *model.Room) ([]model.InboxEvent, error) {
@@ -146,14 +146,18 @@ func (h *handler) roomEvents(ev oplogEvent, room *model.Room) ([]model.InboxEven
 		}
 	}
 
-	switch {
-	case changed(desc, "name") || changed(desc, "fname"):
-		return []model.InboxEvent{h.roomRenamedEvent(room), h.roomSyncEvent(room)}, nil
-	case changed(desc, "ro"):
-		return []model.InboxEvent{h.roomRestrictedEvent(room), h.roomSyncEvent(room)}, nil
-	default:
-		return []model.InboxEvent{h.roomSyncEvent(room)}, nil
+	// A single update delta can change name/fname AND ro together — emit every matching event,
+	// not just the first, so a combined rename+restrict doesn't drop the visibility change.
+	var evts []model.InboxEvent
+	if changed(desc, "name") || changed(desc, "fname") {
+		evts = append(evts, h.roomRenamedEvent(room))
+	}
+	if changed(desc, "ro") {
+		evts = append(evts, h.roomRestrictedEvent(room))
 	}
+	// room_sync always trails so the room doc itself converges alongside the subscription-side events.
+	evts = append(evts, h.roomSyncEvent(room))
+	return evts, nil
 }
 
 // changed reports whether the named field appears in the update delta (set or removed).

data-migration/oplog-collections-transformer/rooms_test.go

@@ -291,3 +291,40 @@ func TestHandleRoom_UpdateOtherFieldEmitsOnlyRoomSync(t *testing.T) {
 	require.Len(t, pub.events, 1, "unrelated field update must emit exactly one room_sync")
 	assert.Equal(t, model.InboxEventType("room_sync"), pub.events[0].Type)
 }
+
+func TestHandleRoom_UpdateNameAndRo_EmitsRenamedRestrictedAndSync(t *testing.T) {
+	pub := &fakePublisher{}
+	full := `{"_id":"r1","t":"c","fname":"New","ro":true,"uids":["u1"],"_updatedAt":{"$date":"2024-05-01T12:00:00.000Z"}}`
+	h := newTestHandler(pub, &fakeTarget{}, &fakeLookup{doc: json.RawMessage(full)})
+
+	err := h.handleRoom(context.Background(), roomEv("update", "", `{"updatedFields":{"fname":"New","ro":true}}`))
+	require.NoError(t, err)
+
+	require.Len(t, pub.events, 3, "a combined name+ro update must emit room_renamed, room_restricted, and room_sync")
+	types := []model.InboxEventType{pub.events[0].Type, pub.events[1].Type, pub.events[2].Type}
+	assert.Contains(t, types, model.InboxRoomRenamed)
+	assert.Contains(t, types, model.InboxRoomRestricted)
+	assert.Contains(t, types, model.InboxEventType("room_sync"))
+}
+
+func TestHandleRoom_DegradedInsertRecoversViaSourceLookup(t *testing.T) {
+	pub := &fakePublisher{}
+	full := `{"_id":"r1","t":"c","fname":"Recovered","uids":["u1"]}`
+	h := newTestHandler(pub, &fakeTarget{}, &fakeLookup{doc: json.RawMessage(full)})
+
+	// Degraded insert: the connector couldn't encode fullDocument (empty) but flagged Degraded.
+	ev := oplogEvent{Op: "insert", Collection: roomsColl, EventID: "e1",
+		DocumentKey: json.RawMessage(`{"_id":"r1"}`), Degraded: true, DegradedReason: "fullDocument encode failed"}
+	require.NoError(t, h.handleRoom(context.Background(), ev))
+
+	require.Len(t, pub.events, 1)
+	var room model.Room
+	require.NoError(t, json.Unmarshal(pub.events[0].Payload, &room))
+	assert.Equal(t, "Recovered", room.Name)
+}
+
+func TestHandleRoom_NonDegradedInsertWithoutFullDocument_Poisons(t *testing.T) {
+	h := newTestHandler(&fakePublisher{}, &fakeTarget{}, &fakeLookup{})
+	ev := oplogEvent{Op: "insert", Collection: roomsColl, DocumentKey: json.RawMessage(`{"_id":"r1"}`)}
+	assert.ErrorIs(t, h.handleRoom(context.Background(), ev), migration.ErrPoison)
+}