fix(oplog-collections-transformer): address CodeRabbit findings on PR #363

- handler: recover degraded insert/replace (empty fullDocument + Degraded) via source
  lookup instead of poisoning — mirrors oplog-transformer, stops dropping documents.
- rooms: a single update changing both name/fname AND ro now emits room_renamed AND
  room_restricted (plus room_sync), not just the first; wrap bare returns with context.
- subscriptions: zero-guard ls/lr (→0) and ts (→now) so absent source times don't leak
  negative year-0001 timestamps; emit role_updated even when roles are cleared on an
  update; wrap resolver/documentKey returns with context.
- config: trim + non-empty-validate SITE_ID/NATS_URL/SOURCE_MONGO_URI/TARGET_MONGO_URI so
  a whitespace-only required var fails at startup, not later.
- main: warn once at startup when ALL_SITE_IDS is empty (status fan-out disabled) — the
  config half of the ALL_SITE_IDS handling.
- tests: combined name+ro and degraded-recovery room tests; zero-timestamp and role-clear
  subscription tests; whitespace-config tests; tighten dispatch/error/fixture assertions
  and restore the global meter provider in metrics_test.
- docs(SOURCE_DATA): read timestamp is max(ls,lr) (resolved), not an open question.

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

@@ -60,11 +60,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

@@ -99,31 +99,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 +122,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

@@ -119,11 +119,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 +159,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 +168,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

@@ -37,6 +37,13 @@ func main() {
 	}
 	slog.SetDefault(slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{Level: parseLevel(cfg.LogLevel)})))
 
+	// Surface an empty ALL_SITE_IDS once at startup: user statusText changes won't propagate
+	// (publishUserStatus skips with a per-event metric). Legitimate for a rooms/subs-only partial
+	// deployment; a misconfig otherwise. (Future: promote to a hard-fail once the modes are known.)
+	if !hasDestinationSite(cfg.AllSiteIDs) {
+		slog.Warn("ALL_SITE_IDS is empty — user status fan-out is disabled (intentional only for a partial deployment)")
+	}
+
 	ctx := context.Background()
 
 	tracerShutdown, err := otelutil.InitTracer(ctx, "oplog-collections-transformer")
@@ -252,6 +259,16 @@ func processOne(ctx context.Context, h *handler, m jetstream.Msg, mtr *metrics,
 
 func nowMs() int64 { return time.Now().UTC().UnixMilli() }
 
+// hasDestinationSite reports whether sites has at least one non-empty entry (a real fan-out target).
+func hasDestinationSite(sites []string) bool {
+	for _, s := range sites {
+		if s != "" {
+			return true
+		}
+	}
+	return false
+}
+
 // streamWaitTimeout bounds how long startup waits for the connector to bootstrap MIGRATION_OPLOG.
 const streamWaitTimeout = 60 * time.Second
 

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