refactor(oplog-collections-transformer): simplify cleanups from /simplify pass

- targetstore: add explicit projections to FindThreadRoom/FindUserID (CLAUDE.md
  §MongoDB "always project precisely"; both run per thread-sub event)
- inboxpublisher: drop the dead `siteID` field (Publish routes on evt.DestSiteID)
- rooms: use mustMarshal in the event builders like the other mappers, dropping
  the (InboxEvent, error) plumbing and collapsing roomEvents' error threading
- users: stamp one nowMillis() per publishUserStatus event instead of two
- threadsubs: fix stale (threadRoomId, userId) comment to userAccount

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

Author: claude

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

@@ -15,7 +15,6 @@ import (
 
 // jetstreamPublisher publishes InboxEvents into the local INBOX stream.
 type jetstreamPublisher struct {
-	siteID  string
 	publish func(ctx context.Context, msg *nats.Msg, opts ...jetstream.PublishOpt) (*jetstream.PubAck, error)
 }
 

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

@@ -21,7 +21,6 @@ func TestJetstreamPublisher_Publish(t *testing.T) {
 	t.Run("publishes event to correct subject with correct data", func(t *testing.T) {
 		var captured *nats.Msg
 		pub := &jetstreamPublisher{
-			siteID: site,
 			publish: func(_ context.Context, msg *nats.Msg, _ ...jetstream.PublishOpt) (*jetstream.PubAck, error) {
 				captured = msg
 				return &jetstream.PubAck{}, nil
@@ -55,7 +54,6 @@ func TestJetstreamPublisher_Publish(t *testing.T) {
 	t.Run("publish error propagates as wrapped error", func(t *testing.T) {
 		publishErr := errors.New("nats down")
 		pub := &jetstreamPublisher{
-			siteID: site,
 			publish: func(_ context.Context, _ *nats.Msg, _ ...jetstream.PublishOpt) (*jetstream.PubAck, error) {
 				return nil, publishErr
 			},

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

@@ -153,7 +153,7 @@ func TestInboxPublisher_RoundTrip(t *testing.T) {
 	})
 	require.NoError(t, err)
 
-	pub := &jetstreamPublisher{siteID: site, publish: js.PublishMsg}
+	pub := &jetstreamPublisher{publish: js.PublishMsg}
 
 	evt := model.InboxEvent{
 		Type:       "room_sync",
@@ -248,7 +248,7 @@ func TestEndToEnd_UserInsert(t *testing.T) {
 		subsColl:       "rocketchat_subscriptions",
 		threadSubsColl: "tsmc_thread_subscriptions",
 		usersColl:      "users",
-		pub:            &jetstreamPublisher{siteID: site, publish: js.PublishMsg},
+		pub:            &jetstreamPublisher{publish: js.PublishMsg},
 		target:         tgtStore,
 		lookups:        map[string]migration.SourceLookup{"users": lookup},
 		now:            func() int64 { return 1700000000000 },
@@ -317,7 +317,7 @@ func TestEndToEnd_RoomInsert_PublishesRoomSync(t *testing.T) {
 		subsColl:       "rocketchat_subscriptions",
 		threadSubsColl: "tsmc_thread_subscriptions",
 		usersColl:      "users",
-		pub:            &jetstreamPublisher{siteID: site, publish: js.PublishMsg},
+		pub:            &jetstreamPublisher{publish: js.PublishMsg},
 		target:         tgtStore,
 		lookups:        map[string]migration.SourceLookup{"rocketchat_rooms": lookup},
 		now:            func() int64 { return 1700000000000 },
@@ -419,7 +419,7 @@ func TestEndToEnd_ThreadSub_NakThenResolve(t *testing.T) {
 		subsColl:       "rocketchat_subscriptions",
 		threadSubsColl: "tsmc_thread_subscriptions",
 		usersColl:      "users",
-		pub:            &jetstreamPublisher{siteID: site, publish: js.PublishMsg},
+		pub:            &jetstreamPublisher{publish: js.PublishMsg},
 		target:         tgtStore,
 		lookups:        map[string]migration.SourceLookup{"tsmc_thread_subscriptions": lookup},
 		now:            func() int64 { return 1700000000000 },

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

@@ -136,7 +136,7 @@ func main() {
 		subsColl:       cfg.SubscriptionsCollection,
 		threadSubsColl: cfg.ThreadSubsCollection,
 		usersColl:      cfg.UsersCollection,
-		pub:            &jetstreamPublisher{siteID: cfg.SiteID, publish: js.PublishMsg},
+		pub:            &jetstreamPublisher{publish: js.PublishMsg},
 		target:         target,
 		lookups:        lookups,
 		metrics:        m,

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

@@ -2,7 +2,6 @@ package main
 
 import (
 	"context"
-	"encoding/json"
 	"fmt"
 	"log/slog"
 	"time"
@@ -145,11 +144,7 @@ func (h *handler) handleRoom(ctx context.Context, ev oplogEvent) error {
 //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) {
 	if ev.Op != "update" {
-		evt, err := h.roomSyncEvent(room)
-		if err != nil {
-			return nil, err
-		}
-		return []model.InboxEvent{evt}, nil
+		return []model.InboxEvent{h.roomSyncEvent(room)}, nil
 	}
 
 	var desc updateDescription
@@ -161,31 +156,11 @@ func (h *handler) roomEvents(ev oplogEvent, room *model.Room) ([]model.InboxEven
 
 	switch {
 	case changed(desc, "name") || changed(desc, "fname"):
-		renamed, err := h.roomRenamedEvent(room)
-		if err != nil {
-			return nil, err
-		}
-		sync, err := h.roomSyncEvent(room)
-		if err != nil {
-			return nil, err
-		}
-		return []model.InboxEvent{renamed, sync}, nil
+		return []model.InboxEvent{h.roomRenamedEvent(room), h.roomSyncEvent(room)}, nil
 	case changed(desc, "ro"):
-		restricted, err := h.roomRestrictedEvent(room)
-		if err != nil {
-			return nil, err
-		}
-		sync, err := h.roomSyncEvent(room)
-		if err != nil {
-			return nil, err
-		}
-		return []model.InboxEvent{restricted, sync}, nil
+		return []model.InboxEvent{h.roomRestrictedEvent(room), h.roomSyncEvent(room)}, nil
 	default:
-		evt, err := h.roomSyncEvent(room)
-		if err != nil {
-			return nil, err
-		}
-		return []model.InboxEvent{evt}, nil
+		return []model.InboxEvent{h.roomSyncEvent(room)}, nil
 	}
 }
 
@@ -202,44 +177,32 @@ func changed(desc updateDescription, field string) bool {
 	return false
 }
 
-func (h *handler) roomSyncEvent(room *model.Room) (model.InboxEvent, error) {
-	payload, err := json.Marshal(room)
-	if err != nil {
-		return model.InboxEvent{}, fmt.Errorf("marshal room_sync payload: %w", err)
-	}
-	return h.inboxEvent(model.InboxEventType("room_sync"), room.SiteID, payload), nil
+func (h *handler) roomSyncEvent(room *model.Room) model.InboxEvent {
+	return h.inboxEvent(model.InboxEventType("room_sync"), room.SiteID, mustMarshal(room))
 }
 
-func (h *handler) roomRenamedEvent(room *model.Room) (model.InboxEvent, error) {
+func (h *handler) roomRenamedEvent(room *model.Room) model.InboxEvent {
 	// Use the source _updatedAt millis (already zero-guarded in handleRoom) as the
 	// nameUpdatedAt high-water mark so UpdateSubscriptionNamesForRoom uses the same guard
 	// timestamp as the companion room_sync.
-	payload, err := json.Marshal(model.RoomRenamedInboxPayload{
+	return h.inboxEvent(model.InboxRoomRenamed, room.SiteID, mustMarshal(model.RoomRenamedInboxPayload{
 		RoomID:    room.ID,
 		NewName:   room.Name,
 		Timestamp: room.UpdatedAt.UnixMilli(),
-	})
-	if err != nil {
-		return model.InboxEvent{}, fmt.Errorf("marshal room_renamed payload: %w", err)
-	}
-	return h.inboxEvent(model.InboxRoomRenamed, room.SiteID, payload), nil
+	}))
 }
 
-func (h *handler) roomRestrictedEvent(room *model.Room) (model.InboxEvent, error) {
+func (h *handler) roomRestrictedEvent(room *model.Room) model.InboxEvent {
 	// Use the source _updatedAt millis (already zero-guarded in handleRoom) as the
 	// visibilityUpdatedAt high-water mark so ApplySubscriptionVisibility uses the same
 	// guard timestamp as the companion room_sync.
-	payload, err := json.Marshal(model.RoomRestrictedInboxPayload{
+	return h.inboxEvent(model.InboxRoomRestricted, room.SiteID, mustMarshal(model.RoomRestrictedInboxPayload{
 		RoomID:         room.ID,
 		Restricted:     room.Restricted,
 		ExternalAccess: room.ExternalAccess,
 		OwnerAccount:   "",
 		Timestamp:      room.UpdatedAt.UnixMilli(),
-	})
-	if err != nil {
-		return model.InboxEvent{}, fmt.Errorf("marshal room_restricted payload: %w", err)
-	}
-	return h.inboxEvent(model.InboxRoomRestricted, room.SiteID, payload), nil
+	}))
 }
 
 // inboxEvent wraps an inner payload in the local-INBOX InboxEvent envelope. SiteID is the

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

@@ -63,7 +63,9 @@ func (s *mongoTargetStore) UpsertUserIfAbsent(ctx context.Context, u model.User)
 // when no thread room exists yet for that parent message.
 func (s *mongoTargetStore) FindThreadRoom(ctx context.Context, parentMessageID string) (string, string, string, bool, error) {
 	var tr model.ThreadRoom
-	err := s.threadRooms.FindOne(ctx, bson.M{"parentMessageId": parentMessageID}).Decode(&tr)
+	err := s.threadRooms.FindOne(ctx, bson.M{"parentMessageId": parentMessageID},
+		options.FindOne().SetProjection(bson.M{"roomId": 1, "siteId": 1}),
+	).Decode(&tr)
 	if errors.Is(err, mongo.ErrNoDocuments) {
 		return "", "", "", false, nil
 	}
@@ -77,7 +79,9 @@ func (s *mongoTargetStore) FindThreadRoom(ctx context.Context, parentMessageID s
 // when the account has not been seeded yet.
 func (s *mongoTargetStore) FindUserID(ctx context.Context, account string) (string, bool, error) {
 	var u model.User
-	err := s.users.FindOne(ctx, bson.M{"account": account}).Decode(&u)
+	err := s.users.FindOne(ctx, bson.M{"account": account},
+		options.FindOne().SetProjection(bson.M{"_id": 1}),
+	).Decode(&u)
 	if errors.Is(err, mongo.ErrNoDocuments) {
 		return "", false, nil
 	}

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

@@ -42,7 +42,7 @@ func (h *handler) handleThreadSub(ctx context.Context, ev oplogEvent) error {
 
 	if ev.Op == "delete" {
 		// Un-actionable: the event carries only the source _id; destination keys by
-		// (threadRoomId, userId) which live in the deleted doc. Also no inbox removal
+		// (threadRoomId, userAccount) which live in the deleted doc. Also no inbox removal
 		// handler for thread-sub unfollows (spec §4.4 / D2).
 		slog.Debug("skip thread_sub delete (un-actionable, no inbox removal handler)",
 			"eventId", ev.EventID, "request_id", reqID)

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

@@ -111,10 +111,11 @@ func (h *handler) handleUser(ctx context.Context, ev oplogEvent) error {
 // not the RocketChat source. A publish failure Naks the whole event; the re-fan is
 // idempotent (status is last-write-wins).
 func (h *handler) publishUserStatus(ctx context.Context, account, statusText string) error {
+	now := h.nowMillis()
 	payload := mustMarshal(model.UserStatusUpdated{
 		Account:    account,
 		StatusText: statusText,
-		Timestamp:  h.nowMillis(),
+		Timestamp:  now,
 	})
 	for _, dest := range h.allSiteIDs {
 		if dest == "" {
@@ -125,7 +126,7 @@ func (h *handler) publishUserStatus(ctx context.Context, account, statusText str
 			SiteID:     h.siteID,
 			DestSiteID: dest,
 			Payload:    payload,
-			Timestamp:  h.nowMillis(),
+			Timestamp:  now,
 		}
 		if err := h.pub.Publish(ctx, evt); err != nil {
 			return fmt.Errorf("publish user_status_updated to %q: %w", dest, err)