thread-inbox: subscription-sourced room fields, app-unsubscribe filter, DM HR info

docs/client-api.md

@@ -4378,7 +4378,7 @@ Returns the user's thread subscriptions across **all sites** as one globally-ord
 |---|---|---|
 | `siteId` | string | The thread's owning site. |
 | `roomId` | string | The room the thread belongs to. |
-| `roomName` | string | The owning room's name (empty if the room doc is unavailable). |
+| `roomName` | string | Per-subscriber display label, sourced from the user's subscription: `channel` → room name; `dm` → counterpart account; `botDM` → app name. |
 | `roomType` | string | The owning room's type (`channel`, `dm`, `botDM`, `discussion`). |
 | `threadRoomId` | string | The thread room ID. |
 | `parentMessageId` | string | The thread's parent (top-level) message ID. |
@@ -4388,6 +4388,7 @@ Returns the user's thread subscriptions across **all sites** as one globally-ord
 | `lastMsgAt` | number | UTC ms of the thread's last activity — the global sort key. |
 | `parentMessage` | [Message](#message-schema) | Optional. The hydrated parent message; reply count rides on its `tcount`. |
 | `lastMessage` | [Message](#message-schema) | Optional. The hydrated last reply. |
+| `hrInfo` | [SubscriptionHRInfo](#subscriptionhrinfo) | Optional. Present **only on `dm` rows** — the counterpart's HR record, resolved from `roomName`. Omitted when the directory lookup degrades. |
 
 ```json
 {
@@ -4415,6 +4416,18 @@ Returns the user's thread subscriptions across **all sites** as one globally-ord
         "sender": { "id": "01970a4f8c2d7c9a01970a4f8c2d7c9b", "account": "bob" },
         "msg": "shipping it"
       }
+    },
+    {
+      "siteId": "site-a",
+      "roomId": "01970a4f8c2d7c9aDM",
+      "roomName": "bob",
+      "roomType": "dm",
+      "threadRoomId": "01970a4f8c2d7c9aTHR2",
+      "parentMessageId": "01970a4f8c2d7c9aPQRS",
+      "hasMention": false,
+      "unread": false,
+      "lastMsgAt": 1746518100000,
+      "hrInfo": { "account": "bob", "name": "鮑伯", "engName": "Bob" }
     }
   ],
   "nextCursor": "eyJsYXN0TXNnQXQiOjE3NDY1MTg0MDAwMDAsInRocmVhZFJvb21JZCI6IjAxOTcwYTRmOGMyZDdjOWFUSFJEIn0=",

history-service/internal/mongorepo/pipelines.go

@@ -36,21 +36,23 @@ func followingThreadsPipeline(roomID, account string, accessSince *time.Time) bs
 // from thread_subscriptions (the per-user filter) and joins thread_rooms for the
 // activity/parent fields.
 //
-// $lookup justification: three joins, none avoidable.
+// $lookup justification: two joins, none avoidable.
 //  1. subscriptions (membership) runs FIRST, on the thread_subscription's own
 //     roomId — the room subscription, not the thread subscription, is the source
 //     of truth for whether the user still belongs to the room (purged on leave;
-//     thread_subscriptions rows are not). Filtering here, before $limit, keeps the
-//     page exact; doing it before the thread_rooms join means that join runs only
-//     for accessible threads. Indexed point read on (u.account, roomId).
+//     thread_subscriptions rows are not). For botDM rooms, where unsubscribe is a
+//     soft toggle that retains the row, the same join also gates on isSubscribed so
+//     an unsubscribed app's threads drop out. Filtering here, before $limit, keeps
+//     the page exact; doing it before the thread_rooms join means that join runs
+//     only for accessible threads. Indexed point read on (u.account, roomId). This
+//     join also carries roomName and roomType: the subscription holds the
+//     per-subscriber display name (dm/botDM room docs have none) and the roomType,
+//     so the page needs no rooms lookup.
 //  2. thread_rooms supplies the inbox sort key (lastMsgAt) and parent/activity
 //     fields, which live there rather than on thread_subscriptions, so we must
 //     sort and paginate on the looked-up field. Denormalizing lastMsgAt onto every
 //     subscription was rejected because it would write-amplify across all
 //     subscribers on every reply (see docs/design/user-thread-list.md §5).
-//  3. rooms (name/type) runs AFTER $limit, so it enriches only the ≤limit+1 page
-//     rows with indexed _id point reads — cheaper than a separate round trip, and
-//     it avoids bolting a non-cacheable GetRoomsMeta onto the cached RoomRepository.
 func userThreadSubscriptionsPipeline(account string, cursorLastMsgAt *time.Time, cursorThreadRoomID string, limit int) bson.A {
 	pipeline := bson.A{
 		bson.D{{Key: "$match", Value: bson.M{"userAccount": account}}},
@@ -63,13 +65,29 @@ func userThreadSubscriptionsPipeline(account string, cursorLastMsgAt *time.Time,
 				bson.D{{Key: "$match", Value: bson.M{
 					"u.account": account,
 					"$expr":     bson.M{"$eq": bson.A{"$roomId", "$$rid"}},
+					// botDM unsubscribe is a soft toggle (isSubscribed=false, row
+					// retained), unlike a room leave that purges the row. Gate botDM
+					// rooms on isSubscribed so an unsubscribed app's threads drop out
+					// of the inbox; channel/dm/discussion rooms pass through untouched.
+					"$or": bson.A{
+						bson.M{"roomType": bson.M{"$ne": "botDM"}},
+						bson.M{"isSubscribed": true},
+					},
 				}}},
-				bson.D{{Key: "$project", Value: bson.M{"_id": 1}}},
+				bson.D{{Key: "$project", Value: bson.M{"_id": 1, "name": 1, "roomType": 1}}},
 			},
 			"as": "sub",
 		}}},
 		// {$ne: []} — $lookup sets "sub" to [] when no subscription matched; non-empty means subscribed.
 		bson.D{{Key: "$match", Value: bson.M{"sub": bson.M{"$ne": bson.A{}}}}},
+		// roomName and roomType come from the user's own subscription, not a room doc:
+		// dm/botDM rooms store an empty room name (the display name is per-subscriber),
+		// and the subscription already carries roomType — so no rooms lookup is needed.
+		// Lifted to scalars here so the sub array can be dropped before $sort.
+		bson.D{{Key: "$set", Value: bson.M{
+			"roomName": bson.M{"$arrayElemAt": bson.A{"$sub.name", 0}},
+			"roomType": bson.M{"$arrayElemAt": bson.A{"$sub.roomType", 0}},
+		}}},
 		bson.D{{Key: "$project", Value: bson.M{"sub": 0}}}, // drop before $sort — mirrors unreadThreadsPipeline
 		bson.D{{Key: "$lookup", Value: bson.M{
 			"from":         threadRoomsCollection,
@@ -96,24 +114,14 @@ func userThreadSubscriptionsPipeline(account string, cursorLastMsgAt *time.Time,
 	pipeline = append(pipeline,
 		bson.D{{Key: "$sort", Value: bson.D{{Key: "tr.lastMsgAt", Value: -1}, {Key: "threadRoomId", Value: -1}}}},
 		bson.D{{Key: "$limit", Value: int64(limit + 1)}},
-		// Page-scoped (post-$limit) room name/type enrichment; preserve rows whose
-		// room doc is missing (degrade to empty name/type).
-		bson.D{{Key: "$lookup", Value: bson.M{
-			"from":         roomsCollection,
-			"localField":   "tr.roomId",
-			"foreignField": "_id",
-			"as":           "room",
-			"pipeline": bson.A{
-				bson.D{{Key: "$project", Value: bson.M{"name": 1, "type": 1}}},
-			},
-		}}},
-		bson.D{{Key: "$unwind", Value: bson.M{"path": "$room", "preserveNullAndEmptyArrays": true}}},
+		// roomName/roomType both ride in on the membership subscription (join 1), so
+		// the page needs no rooms lookup.
 		bson.D{{Key: "$project", Value: bson.M{
 			"_id":             "$threadRoomId",
 			"roomId":          "$tr.roomId",
 			"siteId":          "$tr.siteId",
-			"roomName":        "$room.name",
-			"roomType":        "$room.type",
+			"roomName":        1, // sourced from the subscription above, not a room doc
+			"roomType":        1, // sourced from the subscription above, not a room doc
 			"parentMessageId": "$tr.parentMessageId",
 			"lastMsgId":       "$tr.lastMsgId",
 			"lastMsgAt":       "$tr.lastMsgAt",

history-service/internal/mongorepo/threadsubscription_test.go

@@ -9,10 +9,99 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.mongodb.org/mongo-driver/v2/bson"
+	"go.mongodb.org/mongo-driver/v2/mongo"
 
 	"github.com/hmchangw/chat/pkg/model"
 )
 
+// insertBotDMSubscription seeds a botDM room subscription with an explicit
+// isSubscribed flag, mirroring how user-service's SetAppSubscribed persists the
+// soft-unsubscribe toggle ($set isSubscribed:false, row retained). Raw bson, not
+// model.Subscription, because the struct's `omitempty` would drop a false
+// isSubscribed — and explicit false is exactly the production state under test.
+func insertBotDMSubscription(t *testing.T, db *mongo.Database, account, roomID string, subscribed bool) {
+	t.Helper()
+	_, err := db.Collection("subscriptions").InsertOne(context.Background(), bson.M{
+		"_id":          account + ":" + roomID,
+		"u":            bson.M{"account": account},
+		"roomId":       roomID,
+		"siteId":       "site-a",
+		"roomType":     string(model.RoomTypeBotDM),
+		"isSubscribed": subscribed,
+	})
+	require.NoError(t, err)
+}
+
+// insertNamedSubscription seeds a room subscription carrying the per-subscriber
+// display Name and roomType, mirroring how room-worker's newSub persists them
+// (channel: room name; dm/botDM: counterpart account / app name).
+func insertNamedSubscription(t *testing.T, db *mongo.Database, account, roomID, name string, roomType model.RoomType) {
+	t.Helper()
+	_, err := db.Collection("subscriptions").InsertOne(context.Background(), model.Subscription{
+		ID:       account + ":" + roomID,
+		User:     model.SubscriptionUser{Account: account},
+		RoomID:   roomID,
+		SiteID:   "site-a",
+		Name:     name,
+		RoomType: roomType,
+	})
+	require.NoError(t, err)
+}
+
+// roomName and roomType are filled from the user's own subscription, not a room
+// document: dm/botDM rooms carry an empty room name (the display name is
+// per-subscriber). No rooms collection is seeded — the pipeline must resolve both
+// fields without it.
+func TestThreadSubscriptionRepo_ListUserThreadSubscriptions_RoomNameFromSubscription(t *testing.T) {
+	db := setupMongo(t)
+	repo := NewThreadSubscriptionRepo(db)
+	ctx := context.Background()
+	base := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)
+
+	// A DM thread with no rooms document at all; alice's subscription names the
+	// counterpart (bob) and carries the room type.
+	insertThreadRoom(t, db, model.ThreadRoom{ID: "tr-dm", RoomID: "dm1", ParentMessageID: "p-dm", LastMsgID: "m-dm", SiteID: "site-a", LastMsgAt: base.Add(time.Hour), CreatedAt: base, UpdatedAt: base})
+	insertThreadSubscription(t, db, model.ThreadSubscription{ID: "ts-dm", ThreadRoomID: "tr-dm", RoomID: "dm1", ParentMessageID: "p-dm", UserAccount: "alice", SiteID: "site-a", CreatedAt: base, UpdatedAt: base})
+	insertNamedSubscription(t, db, "alice", "dm1", "bob", model.RoomTypeDM)
+
+	rows, _, err := repo.ListUserThreadSubscriptions(ctx, "alice", nil, "", 10)
+	require.NoError(t, err)
+	require.Len(t, rows, 1)
+	assert.Equal(t, "bob", rows[0].RoomName, "DM roomName must come from the subscription Name, not the empty room name")
+	assert.Equal(t, model.RoomTypeDM, rows[0].RoomType, "DM roomType must come from the subscription, with no rooms lookup")
+}
+
+// An unsubscribed app's botDM thread must not appear in the inbox, while a
+// still-subscribed app's botDM thread does. botDM unsubscribe is a soft toggle
+// (isSubscribed=false) that leaves the subscription row in place — unlike a room
+// leave, which purges it — so the membership join must gate botDM rows on
+// isSubscribed.
+func TestThreadSubscriptionRepo_ListUserThreadSubscriptions_UnsubscribedApp(t *testing.T) {
+	db := setupMongo(t)
+	repo := NewThreadSubscriptionRepo(db)
+	ctx := context.Background()
+	base := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)
+
+	// Two botDM rooms, each hosting one thread alice follows. bot-off has the newer
+	// activity, so a missing filter would surface it first.
+	insertThreadRoom(t, db, model.ThreadRoom{ID: "tr-on", RoomID: "bot-on", ParentMessageID: "p-on", LastMsgID: "m-on", SiteID: "site-a", LastMsgAt: base.Add(3 * time.Hour), CreatedAt: base, UpdatedAt: base})
+	insertThreadRoom(t, db, model.ThreadRoom{ID: "tr-off", RoomID: "bot-off", ParentMessageID: "p-off", LastMsgID: "m-off", SiteID: "site-a", LastMsgAt: base.Add(5 * time.Hour), CreatedAt: base, UpdatedAt: base})
+
+	insertThreadSubscription(t, db, model.ThreadSubscription{ID: "ts-on", ThreadRoomID: "tr-on", RoomID: "bot-on", ParentMessageID: "p-on", UserAccount: "alice", SiteID: "site-a", CreatedAt: base, UpdatedAt: base})
+	insertThreadSubscription(t, db, model.ThreadSubscription{ID: "ts-off", ThreadRoomID: "tr-off", RoomID: "bot-off", ParentMessageID: "p-off", UserAccount: "alice", SiteID: "site-a", CreatedAt: base, UpdatedAt: base})
+
+	// alice still subscribes to bot-on; she has unsubscribed from bot-off.
+	insertBotDMSubscription(t, db, "alice", "bot-on", true)
+	insertBotDMSubscription(t, db, "alice", "bot-off", false)
+
+	rows, hasMore, err := repo.ListUserThreadSubscriptions(ctx, "alice", nil, "", 10)
+	require.NoError(t, err)
+	assert.False(t, hasMore)
+	require.Len(t, rows, 1)
+	assert.Equal(t, "tr-on", rows[0].ThreadRoomID, "unsubscribed-app thread must be filtered out")
+}
+
 func TestThreadSubscriptionRepo_ListUserThreadSubscriptions(t *testing.T) {
 	db := setupMongo(t)
 	repo := NewThreadSubscriptionRepo(db)
@@ -37,15 +126,12 @@ func TestThreadSubscriptionRepo_ListUserThreadSubscriptions(t *testing.T) {
 	insertThreadSubscription(t, db, model.ThreadSubscription{ID: "ts-left", ThreadRoomID: "tr-left", RoomID: "r-left", ParentMessageID: "p-left", UserAccount: "alice", SiteID: "site-a", CreatedAt: base, UpdatedAt: base})
 
 	// alice's room subscriptions — the membership $lookup keeps only threads whose
-	// room she is still subscribed to. She is in r1 and r2, but not r-left.
-	insertSubscription(t, db, "alice", "r1")
+	// room she is still subscribed to. She is in r1 and r2, but not r-left. The
+	// subscription is also where roomName/roomType come from: r1 carries its channel
+	// name and type; r2 is left nameless/typeless to exercise the empty pass-through.
+	insertNamedSubscription(t, db, "alice", "r1", "general", model.RoomTypeChannel)
 	insertSubscription(t, db, "alice", "r2")
 
-	// rooms feed the name/type $lookup; r2 is intentionally unseeded to exercise
-	// the missing-room degrade (empty name/type, row still returned).
-	_, err := db.Collection("rooms").InsertOne(ctx, model.Room{ID: "r1", Name: "general", Type: model.RoomTypeChannel, SiteID: "site-a"})
-	require.NoError(t, err)
-
 	// Page 1, limit 2: newest two (tr-1 5h, tr-2 3h), hasMore true.
 	rows, hasMore, err := repo.ListUserThreadSubscriptions(ctx, "alice", nil, "", 2)
 	require.NoError(t, err)
@@ -64,7 +150,7 @@ func TestThreadSubscriptionRepo_ListUserThreadSubscriptions(t *testing.T) {
 	assert.WithinDuration(t, base.Add(2*time.Hour), *rows[0].LastSeenAt, time.Second)
 	assert.Nil(t, rows[1].LastSeenAt) // never-seen thread
 
-	// Room name/type ride in via the rooms $lookup; r2 unseeded ⇒ degrade to empty.
+	// Room name/type ride in on the membership subscription; r2's sub is nameless ⇒ empty.
 	assert.Equal(t, "general", rows[0].RoomName)
 	assert.Equal(t, model.RoomTypeChannel, rows[0].RoomType)
 	assert.Empty(t, rows[1].RoomName)

history-service/internal/mongorepo/threadsubscription_unit_test.go

@@ -30,10 +30,11 @@ func TestUserThreadSubscriptionsPipeline_FirstPageHasNoCursorMatch(t *testing.T)
 	p := userThreadSubscriptionsPipeline("alice", nil, "", 20)
 	// First page: userAccount $match + membership $match (sub != []), no value-cursor $match.
 	assert.Equal(t, 2, countStages(t, p, "$match"))
-	// Three joins: thread_rooms (sort/cursor), subscriptions (membership), rooms (name/type).
-	assert.Equal(t, 3, countStages(t, p, "$lookup"))
-	// thread_rooms and rooms are unwound; the membership join uses {$ne: []}, not $unwind.
-	assert.Equal(t, 2, countStages(t, p, "$unwind"))
+	// Two joins: thread_rooms (sort/cursor) and subscriptions (membership). roomName
+	// and roomType both ride in on the membership join, so there is no rooms lookup.
+	assert.Equal(t, 2, countStages(t, p, "$lookup"))
+	// Only thread_rooms is unwound; the membership join uses {$ne: []}, not $unwind.
+	assert.Equal(t, 1, countStages(t, p, "$unwind"))
 	assert.Equal(t, 1, countStages(t, p, "$sort"))
 	// Only the outer page $limit is top-level; the membership join's inner $limit:1
 	// is nested inside the $lookup pipeline and not counted here.

pkg/model/threadlist.go

@@ -25,6 +25,11 @@ type ThreadListItem struct {
 	// Hydrated message bodies, subject to the thread access window.
 	ParentMessage *cassandra.Message `json:"parentMessage,omitempty" bson:"parentMessage,omitempty"`
 	LastMessage   *cassandra.Message `json:"lastMessage,omitempty"   bson:"lastMessage,omitempty"`
+
+	// HRInfo carries the DM counterpart's HR-directory record (native + English
+	// name). The user-service aggregator resolves it from RoomName (which holds the
+	// counterpart account for DM rooms); present on DM rows only.
+	HRInfo *SubscriptionHRInfo `json:"hrInfo,omitempty" bson:"hrInfo,omitempty"`
 }
 
 // ThreadSubscriptionListRequest is the server-to-server leaf request the

pkg/model/threadlist_test.go

@@ -29,6 +29,27 @@ func TestThreadListItemJSON(t *testing.T) {
 	roundTrip(t, &src, &model.ThreadListItem{})
 }
 
+// A DM thread row carries the counterpart's HR record, which survives a round trip.
+func TestThreadListItemJSON_WithHRInfo(t *testing.T) {
+	src := model.ThreadListItem{
+		SiteID: "site-a", RoomID: "dm1", RoomName: "bob", RoomType: model.RoomTypeDM,
+		ThreadRoomID: "thr-1", LastMsgAt: 1746518400000,
+		HRInfo: &model.SubscriptionHRInfo{Account: "bob", Name: "鮑勃", EngName: "Bob Chen"},
+	}
+	roundTrip(t, &src, &model.ThreadListItem{})
+}
+
+// hrInfo is omitted for non-DM rows (no counterpart record).
+func TestThreadListItemJSON_OmitsNilHRInfo(t *testing.T) {
+	src := model.ThreadListItem{SiteID: "site-a", RoomID: "room-1", RoomType: model.RoomTypeChannel, ThreadRoomID: "thr-1"}
+	data, err := json.Marshal(&src)
+	require.NoError(t, err)
+	var raw map[string]any
+	require.NoError(t, json.Unmarshal(data, &raw))
+	_, hasHR := raw["hrInfo"]
+	assert.False(t, hasHR, "nil hrInfo must be omitted")
+}
+
 // LastSeenAt is omitted when nil (never-seen thread).
 func TestThreadListItemJSON_OmitsNilLastSeenAt(t *testing.T) {
 	src := model.ThreadListItem{SiteID: "site-a", RoomID: "room-1", ThreadRoomID: "thr-1"}