docs(oplog-collections-transformer): tighten comments to the 2-line cap

Comment-only cleanup pass: collapse verbose godoc/WHY blocks to ≤2 lines across
the collections-transformer mappers and the migration's added blocks in
inbox-worker / pkg/model. No behavior change.

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/bootstrap.go

@@ -16,12 +16,8 @@ type streamManager interface {
 	CreateOrUpdateStream(ctx context.Context, cfg jetstream.StreamConfig) (oteljetstream.Stream, error)
 }
 
-// bootstrapStreams is a no-op in production (Enabled=false): this service consumes from
-// MIGRATION_OPLOG (owned by the oplog-connector) and publishes to INBOX (owned by inbox-worker),
-// so it owns no streams. When Enabled (local dev / integration), it creates only the
-// MIGRATION_OPLOG_{siteID} schema (Name+Subjects) so the service can stand up against a fresh
-// NATS before the connector runs — createConsumerWithRetry still tolerates the stream's absence.
-// INBOX is never created here; inbox-worker owns it. Federation config stays ops/IaC-owned.
+// bootstrapStreams is a no-op in production (this service owns no streams). When Enabled
+// (dev/integration) it creates only the MIGRATION_OPLOG_{siteID} schema; inbox-worker owns INBOX.
 func bootstrapStreams(ctx context.Context, js streamManager, siteID string, enabled bool) error {
 	if !enabled {
 		return nil

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

@@ -9,25 +9,8 @@ type roomClass struct {
 	Reason   string // exclusion reason (for metrics), set only when Excluded
 }
 
-// classifyRoom maps a source rocketchat_rooms doc to a destination RoomType, or marks it
-// excluded. Inputs are the raw source fields:
-//
-//	t             — source room type: c, p, d, l, v
-//	hasPrid       — whether the room has a parent-room id (prid set) → discussion
-//	hasTeamID     — whether the room has a teamId (team room → plain channel; team-ness dropped)
-//	participantCount — number of members (uids/usernames); for t=d, >2 is a group DM (excluded)
-//	hasBot        — for t=d with 2 participants, whether a participant is a bot (→ botDM)
-//
-// Rules (spec §4.2):
-//
-//	c|p (no prid)        → channel  (no public/private distinction)
-//	p with prid          → discussion
-//	d, 2 participants    → dm, or botDM if hasBot
-//	d, >2 participants   → EXCLUDED "group_dm"
-//	l                    → EXCLUDED "livechat"
-//	v                    → EXCLUDED "voip"
-//	teamId present       → channel (team rooms migrate as plain channel; c/p branch already yields channel)
-//	anything else        → EXCLUDED "unknown_type"
+// classifyRoom maps a source room type t (+ prid/teamId/participant/bot signals) to a destination
+// RoomType or an exclusion (group_dm/livechat/voip/unknown_type). p+prid→discussion, team→channel. §4.2.
 func classifyRoom(t string, hasPrid, hasTeamID bool, hasBot bool, participantCount int) roomClass {
 	// hasTeamID is accepted for caller clarity and future use; c/p branch already returns channel
 	// regardless of teamId, so no separate branch is needed here.

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

@@ -36,13 +36,11 @@ type inboxPublisher interface {
 
 // targetStore is the new-stack per-site Mongo access the transformer needs.
 type targetStore interface {
-	// UpsertUserIfAbsent inserts the user keyed by account only when absent, leaving an
-	// existing doc (owned by the company-wide sync) untouched. inserted reports whether a
-	// new doc was created (false ⇒ already present), so the caller can meter seed outcome.
+	// UpsertUserIfAbsent inserts the user keyed by account only when absent, leaving an existing
+	// doc (owned by the company-wide sync) untouched. inserted is false when already present.
 	UpsertUserIfAbsent(ctx context.Context, u model.User) (inserted bool, err error)
-	// FindThreadRoom looks up the thread room derived from the given parent message id.
-	// Returns roomID, threadRoomID, and siteID (the thread room's home site) when found.
-	// Thread-subs inherit the room's site per spec §6.
+	// FindThreadRoom resolves the thread room for a parent message id, returning roomID,
+	// threadRoomID, and the thread room's home siteID (thread-subs inherit the room's site, §6).
 	FindThreadRoom(ctx context.Context, parentMessageID string) (roomID, threadRoomID, siteID string, found bool, err error)
 	FindUserID(ctx context.Context, account string) (userID string, found bool, err error)
 }
@@ -56,9 +54,8 @@ type handler struct {
 	usersColl      string
 	pub            inboxPublisher
 	target         targetStore
-	// lookups re-read the current source doc on update events, keyed by source collection
-	// name — each watched collection re-reads from ITS OWN collection (the connector forwards
-	// only the delta on update). One SourceLookup per rooms/subs/threadsubs/users collection.
+	// lookups re-read the current source doc on update events (the connector forwards only the
+	// delta), keyed by source collection name — one SourceLookup per watched collection.
 	lookups map[string]migration.SourceLookup
 	metrics *metrics     // nil-safe
 	now     func() int64 // injectable clock, defaults to time.Now().UTC().UnixMilli
@@ -72,9 +69,8 @@ func (h *handler) nowMillis() int64 {
 	return time.Now().UTC().UnixMilli()
 }
 
-// handle dispatches one decoded oplog event by collection. nil = ack+count;
-// migration.ErrSkipped = ack-without-counting (deliberate drop, already metered);
-// migration.ErrPoison => Term; any other error => Nak (transient).
+// handle dispatches one decoded oplog event by collection. nil = ack+count; ErrSkipped =
+// ack-without-count (already metered); ErrPoison => Term; any other error => Nak (transient).
 //
 //nolint:gocritic // ev passed by value: it's the decoded event the consume loop hands off, one per message off the hot path.
 func (h *handler) handle(ctx context.Context, ev oplogEvent) error {
@@ -95,10 +91,8 @@ func (h *handler) handle(ctx context.Context, ev oplogEvent) error {
 	}
 }
 
-// resolveDoc returns the full current source doc bytes for the event, or (nil, true, nil)
-// when the event should be skipped (delete: doc is gone & un-actionable; update lookup miss).
-// insert/replace carry the full doc inline; update re-reads by documentKey._id via the source
-// lookup (the connector forwards only the delta on update). delete is always skip.
+// resolveDoc returns the full current source doc for the event, or (nil, true, nil) to skip.
+// insert/replace carry the doc inline; update re-reads by documentKey._id; delete is always skip.
 //
 //nolint:gocritic // ev passed by value to mirror handle's signature; off the hot path.
 func (h *handler) resolveDoc(ctx context.Context, ev oplogEvent) (doc []byte, skip bool, err error) {
@@ -151,9 +145,8 @@ func documentKeyID(documentKey json.RawMessage) (string, error) {
 	return key.ID, nil
 }
 
-// siteIDFromOrigin returns the record's home siteId. federation.origin absent or the literal
-// "local" → the deployment's siteID; otherwise the first dotted label of the origin domain
-// (e.g. "0030204.tchat-test..." → "0030204").
+// siteIDFromOrigin returns the record's home siteId: the deployment's siteID when origin is
+// absent or "local", else the first dotted label of the origin domain ("0030204.tchat..." → "0030204").
 func siteIDFromOrigin(origin, deploymentSiteID string) string {
 	if origin == "" || origin == "local" {
 		return deploymentSiteID

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

@@ -18,9 +18,8 @@ type jetstreamPublisher struct {
 	publish func(ctx context.Context, msg *nats.Msg, opts ...jetstream.PublishOpt) (*jetstream.PubAck, error)
 }
 
-// Publish emits one InboxEvent onto the INBOX external lane for the event's type,
-// blocking on the pub-ack. The request id flows from ctx into the message headers via
-// natsutil.NewMsg so the transformer→inbox-worker hop shares one correlation id.
+// Publish emits one InboxEvent onto the INBOX external lane, blocking on the pub-ack. The request
+// id flows from ctx into the message headers (natsutil.NewMsg) so transformer→inbox-worker shares it.
 //
 //nolint:gocritic // model.InboxEvent passed by value: one per migrated record, off the hot path.
 func (p *jetstreamPublisher) Publish(ctx context.Context, evt model.InboxEvent) error {

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

@@ -192,9 +192,8 @@ func main() {
 	)
 }
 
-// processOne decodes one event and dispatches it, mapping the outcome to a JetStream disposition:
-// Ack on success, Term on poison (never redelivered), Nak-with-delay on transient up to maxDeliver
-// — then Termed with a distinct metric instead of JetStream's silent drop (see migration.IsFinalDelivery).
+// processOne decodes one event and maps its outcome to a JetStream disposition: Ack on success,
+// Term on poison, Nak-with-delay on transient up to maxDeliver, then Term-with-metric (not silent drop).
 func processOne(ctx context.Context, h *handler, m jetstream.Msg, mtr *metrics, maxDeliver, deleteMaxDeliver int) {
 	// Stamp a correlation id once at entry; it flows via ctx into the inbox publish
 	// (read from ctx through natsutil.NewMsg), so transformer→inbox-worker shares one request_id.

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

@@ -9,9 +9,8 @@ import (
 	"go.opentelemetry.io/otel/metric"
 )
 
-// metrics holds the collections transformer's instruments: processed throughput plus the
-// nak/term/exhausted dispositions that flag a stuck or poison-heavy stream, the user-seed
-// outcome, and the thread-sub FK resolution misses. Nil-safe so unit tests run without a meter.
+// metrics holds the transformer's instruments: processed throughput, nak/term/exhausted
+// dispositions, user-seed outcome, and FK resolution misses. Nil-safe (tests run without a meter).
 type metrics struct {
 	processed   metric.Int64Counter
 	naks        metric.Int64Counter

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

@@ -13,8 +13,7 @@ import (
 	"github.com/hmchangw/chat/pkg/natsutil"
 )
 
-// sourceRoom is the subset of a rocketchat_rooms doc the mapper consumes. Decoded from the
-// connector's relaxed extended JSON via bson.UnmarshalExtJSON, same as the message mapper.
+// sourceRoom is the subset of a rocketchat_rooms doc the mapper decodes (relaxed extended JSON).
 type sourceRoom struct {
 	ID        string    `bson:"_id"`
 	T         string    `bson:"t"`
@@ -33,8 +32,7 @@ type sourceRoom struct {
 	} `bson:"federation"`
 }
 
-// updateDescription is the connector's update delta for a room change event. Only the
-// changed field *keys* matter for event selection, so the values are decoded as opaque any.
+// updateDescription is the connector's update delta; only changed field keys matter, values are opaque.
 type updateDescription struct {
 	UpdatedFields map[string]any `bson:"updatedFields" json:"updatedFields"`
 	RemovedFields []string       `bson:"removedFields" json:"removedFields"`
@@ -56,9 +54,8 @@ func (r *sourceRoom) displayName() string {
 	return r.Name
 }
 
-// handleRoom maps a rocketchat_rooms change event to an inbox InboxEvent (room_sync /
-// room_renamed / room_restricted) per spec §4.2 / §4.0. Returns migration.ErrSkipped for
-// deletes, excluded room types, and update lookup misses (all metered).
+// handleRoom maps a rocketchat_rooms change event to an inbox InboxEvent (§4.2 / §4.0).
+// Returns migration.ErrSkipped for deletes, excluded room types, and update lookup misses.
 //
 //nolint:gocritic // ev passed by value to mirror handle's signature; off the hot path.
 func (h *handler) handleRoom(ctx context.Context, ev oplogEvent) error {
@@ -94,9 +91,8 @@ func (h *handler) handleRoom(ctx context.Context, ev oplogEvent) error {
 		return migration.ErrSkipped
 	}
 
-	// Zero-guard: if the source timestamp is absent (zero), fall back to the handler's now()
-	// so the room doc never carries a year-0001 UpdatedAt, keeping the UpsertRoom high-water-mark
-	// guard functional for first and subsequent syncs.
+	// Zero-guard an absent source timestamp with now() so the room doc never carries a year-0001
+	// UpdatedAt, keeping the UpsertRoom high-water-mark guard functional.
 	nowMillis := h.nowMillis()
 	updatedAt := sr.UpdatedAt.UTC()
 	if updatedAt.IsZero() {
@@ -134,12 +130,8 @@ func (h *handler) handleRoom(ctx context.Context, ev oplogEvent) error {
 	return nil
 }
 
-// roomEvents selects and builds the InboxEvents for the room change.
-//
-//   - insert/replace → [room_sync]
-//   - update, name/fname changed → [room_renamed, room_sync]  (subs + room doc)
-//   - update, ro changed → [room_restricted, room_sync]        (subs + room doc)
-//   - update, other field → [room_sync]
+// 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.
 //
 //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) {
@@ -182,9 +174,8 @@ func (h *handler) roomSyncEvent(room *model.Room) model.InboxEvent {
 }
 
 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.
+	// Use the source _updatedAt millis (zero-guarded in handleRoom) as the nameUpdatedAt high-water
+	// mark so UpdateSubscriptionNamesForRoom matches the companion room_sync guard.
 	return h.inboxEvent(model.InboxRoomRenamed, room.SiteID, mustMarshal(model.RoomRenamedInboxPayload{
 		RoomID:    room.ID,
 		NewName:   room.Name,
@@ -193,9 +184,8 @@ func (h *handler) roomRenamedEvent(room *model.Room) model.InboxEvent {
 }
 
 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.
+	// Use the source _updatedAt millis (zero-guarded in handleRoom) as the visibilityUpdatedAt
+	// high-water mark so ApplySubscriptionVisibility matches the companion room_sync guard.
 	return h.inboxEvent(model.InboxRoomRestricted, room.SiteID, mustMarshal(model.RoomRestrictedInboxPayload{
 		RoomID:         room.ID,
 		Restricted:     room.Restricted,

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

@@ -14,8 +14,7 @@ import (
 	"github.com/hmchangw/chat/pkg/natsutil"
 )
 
-// sourceSubscription is the subset of a rocketchat_subscriptions doc the mapper consumes,
-// decoded from the connector's relaxed extended JSON via bson.UnmarshalExtJSON (handles $date).
+// sourceSubscription is the subset of a rocketchat_subscriptions doc the mapper decodes (handles $date).
 type sourceSubscription struct {
 	ID string `bson:"_id"`
 	U  struct {
@@ -51,8 +50,7 @@ func (s *sourceSubscription) lastSeenMillis() int64 {
 	return ls
 }
 
-// subUpdateDescription is the connector's update delta for a subscription change event. Only the
-// changed field keys matter for event selection, so values are decoded as opaque any.
+// subUpdateDescription is the connector's update delta; only changed field keys matter, values are opaque.
 type subUpdateDescription struct {
 	UpdatedFields map[string]any `bson:"updatedFields" json:"updatedFields"`
 	RemovedFields []string       `bson:"removedFields" json:"removedFields"`
@@ -71,18 +69,14 @@ func subChanged(desc subUpdateDescription, field string) bool {
 	return false
 }
 
-// handleSubscription maps a rocketchat_subscriptions change event to inbox InboxEvents per
-// spec §4.3 / §4.0. insert/replace emit member_added plus the state events that reproduce the
-// source row; update emits the matching event(s) (open toggle drives membership). delete is
-// un-actionable (only the source _id) → skip + metric.
+// handleSubscription maps a rocketchat_subscriptions change event to inbox InboxEvents (§4.3 / §4.0):
+// insert/replace reproduce the source row, update emits the matching event(s), delete maps by _id.
 //
 //nolint:gocritic // ev passed by value to mirror handle's signature; off the hot path.
 func (h *handler) handleSubscription(ctx context.Context, ev oplogEvent) error {
 	if ev.Op == "delete" {
-		// A genuine leave/remove hard-deletes the source row (distinct from the open:false hide
-		// handled as a member_removed update below). The delete event carries only the source _id,
-		// but the destination adopted that _id at member_added time, so emit a subscription_deleted
-		// keyed by it and let inbox-worker delete the row by _id.
+		// A genuine leave hard-deletes the source row (vs the open:false hide → member_removed below).
+		// The destination adopted this _id at member_added time, so delete-by-_id maps back to it.
 		id, idErr := documentKeyID(ev.DocumentKey)
 		if idErr != nil {
 			return idErr
@@ -195,9 +189,8 @@ func (h *handler) publishSubscriptionState(ctx context.Context, ss *sourceSubscr
 	return h.pub.Publish(ctx, h.readEvent(ss, siteID))
 }
 
-// memberAddedEvent builds the member_added InboxEvent. RoomType is classified from the
-// subscription's t alone — a subscription can't distinguish discussion/botDM (no prid/teamId/bot
-// visibility), so those degrade to channel/dm; roles are overridden by the role_updated below.
+// memberAddedEvent builds the member_added InboxEvent. RoomType is classified from t alone
+// (a sub can't see prid/teamId/bot, so discussion/botDM degrade to channel/dm); roles via role_updated.
 func (h *handler) memberAddedEvent(ss *sourceSubscription, siteID string) model.InboxEvent {
 	class := classifyRoom(ss.T, false, false, false, 2)
 	payload := mustMarshal(model.MemberAddEvent{
@@ -215,8 +208,7 @@ func (h *handler) memberAddedEvent(ss *sourceSubscription, siteID string) model.
 }
 
 // subscriptionDeletedEvent builds the subscription_deleted InboxEvent for a source hard-delete,
-// carrying the source subscription _id (adopted as the destination _id at member_added time).
-// Routed to the local inbox-worker like every other migrated row.
+// carrying the source _id (adopted as the destination _id at member_added time).
 func (h *handler) subscriptionDeletedEvent(subID string) model.InboxEvent {
 	payload := mustMarshal(model.SubscriptionDeletedEvent{
 		SubID:     subID,
@@ -302,13 +294,12 @@ func mapSubscriptionRoles(roles []string) []model.Role {
 	return out
 }
 
-// mustMarshal JSON-encodes an inner InboxEvent payload. The argument is always a fixed-shape
-// model struct of plain scalars/slices, so json.Marshal cannot fail; a marshal error would be a
-// programmer error, not a runtime condition, hence the panic.
+// mustMarshal JSON-encodes a fixed-shape model payload; json.Marshal cannot fail on these,
+// so an error is a programmer error and panics.
 func mustMarshal(v any) []byte {
 	b, err := json.Marshal(v)
 	if err != nil {
-		panic(fmt.Sprintf("marshal subscription payload: %v", err))
+		panic(fmt.Sprintf("marshal inbox payload: %v", err))
 	}
 	return b
 }

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

@@ -58,9 +58,8 @@ func (s *mongoTargetStore) UpsertUserIfAbsent(ctx context.Context, u model.User)
 	return res.UpsertedCount > 0, nil
 }
 
-// FindThreadRoom looks up the thread room derived from parentMessageID, returning the
-// owning room id, the thread room id, and the thread room's home site. found=false (no error)
-// when no thread room exists yet for that parent message.
+// FindThreadRoom resolves the thread room for parentMessageID, returning room id, thread room id,
+// and its home site. found=false (no error) 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},