diff --git a/internal/rpc/channel_fanout_dispatcher.go b/internal/rpc/channel_fanout_dispatcher.go index 25e8d701..a0a19a34 100644 --- a/internal/rpc/channel_fanout_dispatcher.go +++ b/internal/rpc/channel_fanout_dispatcher.go @@ -955,6 +955,23 @@ func channelMessageFanoutOwnerIDs(res domain.SendChannelMessageResult, extraUser // channelMessagesFanoutOwnerIDs 同上，但取多条结果（批量转发汇成一个 job）的 owner id 并集。 func channelMessagesFanoutOwnerIDs(results []domain.SendChannelMessageResult, extraUserIDs []int64) []int64 { + userIDs, _ := channelMessagesFanoutPeerRefs(results, extraUserIDs) + return peerIDMapKeys(userIDs) +} + +func channelMessagesFanoutUsernamePeers(results []domain.SendChannelMessageResult, extraUserIDs []int64) []domain.Peer { + userIDs, channelIDs := channelMessagesFanoutPeerRefs(results, extraUserIDs) + peers := make([]domain.Peer, 0, len(userIDs)+len(channelIDs)) + for userID := range userIDs { + peers = append(peers, domain.Peer{Type: domain.PeerTypeUser, ID: userID}) + } + for channelID := range channelIDs { + peers = append(peers, domain.Peer{Type: domain.PeerTypeChannel, ID: channelID}) + } + return peers +} + +func channelMessagesFanoutPeerRefs(results []domain.SendChannelMessageResult, extraUserIDs []int64) (map[int64]struct{}, map[int64]struct{}) { userIDs := make(map[int64]struct{}, len(results)+len(extraUserIDs)+4) channelIDs := make(map[int64]struct{}) for _, id := range extraUserIDs { @@ -966,7 +983,7 @@ func channelMessagesFanoutOwnerIDs(results []domain.SendChannelMessageResult, ex collectChannelUpdatePeerRefs(res.Event, res.Channel.ID, userIDs, channelIDs) collectChannelMessagePeerRefs(res.Message, res.Channel.ID, userIDs, channelIDs) } - return peerIDMapKeys(userIDs) + return userIDs, channelIDs } // enqueueChannelMessageFanout 异步 fan-out 单条频道消息并预热跨 viewer 投影（「频道里出现一条新消息」 @@ -976,11 +993,14 @@ func (r *Router) enqueueChannelMessageFanout(ctx context.Context, originUserID i r.enqueueBotAPIChannelMessageUpdate(ctx, originUserID, res) fanoutCache := newViewerPeerCache(r) ownerIDs := channelMessageFanoutOwnerIDs(res, extraUserIDs) + usernamePeers := channelMessagesFanoutUsernamePeers([]domain.SendChannelMessageResult{res}, extraUserIDs) + var usernames map[domain.Peer][]domain.Username skip := skipDeliverySet(res.SkipDeliveryUserIDs) r.enqueueChannelFanoutWithPrefetch(ctx, channelFanoutMessageBox, originUserID, res.Channel.ID, res.Event.Pts, res.Recipients, 0, func(bgCtx context.Context, viewers []int64) { r.prefetchChannelFanoutUsers(bgCtx, fanoutCache, viewers, ownerIDs) + usernames = r.usernameRegistryMap(bgCtx, usernamePeers) }, func(bgCtx context.Context, viewerUserID int64) *tg.Updates { // privacy bot 在 send 时被 SkipDeliveryUserIDs 排除（命令/@/回复以外的消息不可见）。 @@ -991,7 +1011,7 @@ func (r *Router) enqueueChannelMessageFanout(ctx context.Context, originUserID i if _, skipped := skip[viewerUserID]; skipped { return nil } - return r.channelMessageUpdatesWithPeerCache(bgCtx, viewerUserID, res, 0, fanoutCache) + return r.channelMessageUpdatesWithPeerCacheAndUsernames(bgCtx, viewerUserID, res, 0, fanoutCache, usernames) }) } @@ -1030,6 +1050,23 @@ func skipDeliverySet(ids []int64) map[int64]struct{} { // ServiceEvent/ServiceMessage 仅 ServiceEvent.Pts!=0 时收，对应 todo 编辑的服务消息第二容器），使预热 // owner 集与 build 实际下发的 Users 集恰好一致——多收只会无害多预热，但镜像门控让等价测试最紧。 func channelEditMessageFanoutOwnerIDs(res domain.EditChannelMessageResult) []int64 { + userIDs, _ := channelEditMessageFanoutPeerRefs(res) + return peerIDMapKeys(userIDs) +} + +func channelEditMessageFanoutUsernamePeers(res domain.EditChannelMessageResult) []domain.Peer { + userIDs, channelIDs := channelEditMessageFanoutPeerRefs(res) + peers := make([]domain.Peer, 0, len(userIDs)+len(channelIDs)) + for userID := range userIDs { + peers = append(peers, domain.Peer{Type: domain.PeerTypeUser, ID: userID}) + } + for channelID := range channelIDs { + peers = append(peers, domain.Peer{Type: domain.PeerTypeChannel, ID: channelID}) + } + return peers +} + +func channelEditMessageFanoutPeerRefs(res domain.EditChannelMessageResult) (map[int64]struct{}, map[int64]struct{}) { userIDs := make(map[int64]struct{}, 4) channelIDs := make(map[int64]struct{}) if res.Event.Pts != 0 { @@ -1040,7 +1077,7 @@ func channelEditMessageFanoutOwnerIDs(res domain.EditChannelMessageResult) []int collectChannelUpdatePeerRefs(res.ServiceEvent, res.Channel.ID, userIDs, channelIDs) collectChannelMessagePeerRefs(res.ServiceMessage, res.Channel.ID, userIDs, channelIDs) } - return peerIDMapKeys(userIDs) + return userIDs, channelIDs } // enqueueChannelEditMessageFanout 异步 fan-out 一条频道编辑并预热跨 viewer 投影（editMessage/geolive/ @@ -1057,14 +1094,17 @@ func (r *Router) enqueueChannelEditMessageFanout(ctx context.Context, originUser r.enqueueBotAPIChannelEditMessageUpdate(ctx, originUserID, res) fanoutCache := newViewerPeerCache(r) ownerIDs := channelEditMessageFanoutOwnerIDs(res) + usernamePeers := channelEditMessageFanoutUsernamePeers(res) + var usernames map[domain.Peer][]domain.Username nudgePts := max(res.Event.Pts, res.ServiceEvent.Pts) r.enqueueChannelFanoutWithPrefetch(ctx, channelFanoutMessageBox, originUserID, res.Channel.ID, nudgePts, res.Recipients, 0, func(bgCtx context.Context, viewers []int64) { r.prefetchChannelFanoutUsers(bgCtx, fanoutCache, viewers, ownerIDs) + usernames = r.usernameRegistryMap(bgCtx, usernamePeers) }, func(bgCtx context.Context, viewerUserID int64) *tg.Updates { - return r.channelEditMessageUpdatesWithPeerCache(bgCtx, viewerUserID, res, fanoutCache) + return r.channelEditMessageUpdatesWithPeerCacheAndUsernames(bgCtx, viewerUserID, res, fanoutCache, usernames) }) } @@ -1075,13 +1115,16 @@ func (r *Router) enqueueChannelMessagesFanout(ctx context.Context, originUserID, r.enqueueBotAPIChannelMessagesUpdate(ctx, originUserID, results) fanoutCache := newViewerPeerCache(r) ownerIDs := channelMessagesFanoutOwnerIDs(results, extraUserIDs) + usernamePeers := channelMessagesFanoutUsernamePeers(results, extraUserIDs) + var usernames map[domain.Peer][]domain.Username r.enqueueChannelFanoutWithPrefetch(ctx, channelFanoutMessageBox, originUserID, channelID, pts, recipients, int64(len(results))*(64<<10), func(bgCtx context.Context, viewers []int64) { r.prefetchChannelFanoutUsers(bgCtx, fanoutCache, viewers, ownerIDs) + usernames = r.usernameRegistryMap(bgCtx, usernamePeers) }, func(bgCtx context.Context, viewerUserID int64) *tg.Updates { - return r.channelMessagesUpdatesWithPeerCache(bgCtx, viewerUserID, results, nil, false, extraUserIDs, fanoutCache) + return r.channelMessagesUpdatesWithPeerCacheAndUsernames(bgCtx, viewerUserID, results, nil, false, extraUserIDs, fanoutCache, usernames) }) } diff --git a/internal/rpc/channel_fanout_dispatcher_test.go b/internal/rpc/channel_fanout_dispatcher_test.go index 44135338..ebd9e87d 100644 --- a/internal/rpc/channel_fanout_dispatcher_test.go +++ b/internal/rpc/channel_fanout_dispatcher_test.go @@ -346,8 +346,9 @@ func (s *prefetchRecordingUsersService) ByIDsForViewers(_ context.Context, viewe // 回退，prefetch 同步执行）。锁定 edit 路径接入了 O(owner) 预热而非逐 viewer 投影。 func TestChannelEditMessageFanoutInvokesPrefetch(t *testing.T) { users := &prefetchRecordingUsersService{mapUsersService: mapUsersService{users: map[int64]domain.User{}}} + registry := newFakeUsernameRegistry() cs := &captureSessions{} - r := New(Config{}, Deps{Sessions: cs, Users: users}, zaptest.NewLogger(t), clock.System) + r := New(Config{}, Deps{Sessions: cs, Users: users, Usernames: registry}, zaptest.NewLogger(t), clock.System) res := editFanoutTestResult(5, 6) r.enqueueChannelEditMessageFanout(context.Background(), 5, res) @@ -367,6 +368,9 @@ func TestChannelEditMessageFanoutInvokesPrefetch(t *testing.T) { t.Fatalf("prefetch owner ids %v missing %d (must equal channelEditMessageFanoutOwnerIDs)", users.gotOwnerIDs, want) } } + if registry.batchCalls != 1 || registry.peerCalls != 0 { + t.Fatalf("username registry reads = batch %d / peer %d, want one prefetch for all viewers", registry.batchCalls, registry.peerCalls) + } } // nudgeSessions 在 captureSessions 基础上实现 ChannelNudgeProvider 并按 user 记录最近一次推送， diff --git a/internal/rpc/channels_updates.go b/internal/rpc/channels_updates.go index 2c96a831..aeee742c 100644 --- a/internal/rpc/channels_updates.go +++ b/internal/rpc/channels_updates.go @@ -52,7 +52,7 @@ func (r *Router) onUpdatesGetChannelDifference(ctx context.Context, req *tg.Upda r.refreshPublicChannelSubscription(ctx, userID, channelID) } diff = r.enrichChannelDifference(ctx, userID, diff) - out := tgChannelDifference(userID, diff) + out := r.tgChannelDifference(ctx, userID, diff) if linked, ok := r.linkedDiscussionChat(ctx, userID, channelID); ok { switch value := out.(type) { case *tg.UpdatesChannelDifference: @@ -183,12 +183,20 @@ func (r *Router) linkedMonoforumForChannelState(ctx context.Context, userID int6 } func (r *Router) channelMessageUpdatesWithPeerCache(ctx context.Context, viewerUserID int64, res domain.SendChannelMessageResult, randomID int64, cache *viewerPeerCache) *tg.Updates { + updates := r.channelMessageUpdatesWithPeerCacheAndUsernames(ctx, viewerUserID, res, randomID, cache, nil) + if updates != nil { + r.applyUsernamesToPeerObjects(ctx, updates.Users, updates.Chats) + } + return updates +} + +func (r *Router) channelMessageUpdatesWithPeerCacheAndUsernames(ctx context.Context, viewerUserID int64, res domain.SendChannelMessageResult, randomID int64, cache *viewerPeerCache, usernames map[domain.Peer][]domain.Username) *tg.Updates { randomIDs := []int64(nil) includeMessageIDs := randomID != 0 if includeMessageIDs { randomIDs = []int64{randomID} } - return r.channelMessagesUpdatesWithPeerCache(ctx, viewerUserID, []domain.SendChannelMessageResult{res}, randomIDs, includeMessageIDs, nil, cache) + return r.channelMessagesUpdatesWithPeerCacheAndUsernames(ctx, viewerUserID, []domain.SendChannelMessageResult{res}, randomIDs, includeMessageIDs, nil, cache, usernames) } func (r *Router) pushChannelDiscussionUpdate(ctx context.Context, originUserID int64, discussion *domain.SendChannelDiscussionResult) { @@ -209,6 +217,14 @@ func (r *Router) pushChannelDiscussionUpdate(ctx context.Context, originUserID i } func (r *Router) channelMessagesUpdatesWithPeerCache(ctx context.Context, viewerUserID int64, results []domain.SendChannelMessageResult, randomIDs []int64, includeMessageIDs bool, extraUserIDs []int64, cache *viewerPeerCache) *tg.Updates { + updates := r.channelMessagesUpdatesWithPeerCacheAndUsernames(ctx, viewerUserID, results, randomIDs, includeMessageIDs, extraUserIDs, cache, nil) + if updates != nil { + r.applyUsernamesToPeerObjects(ctx, updates.Users, updates.Chats) + } + return updates +} + +func (r *Router) channelMessagesUpdatesWithPeerCacheAndUsernames(ctx context.Context, viewerUserID int64, results []domain.SendChannelMessageResult, randomIDs []int64, includeMessageIDs bool, extraUserIDs []int64, cache *viewerPeerCache, usernames map[domain.Peer][]domain.Username) *tg.Updates { if cache == nil { cache = newViewerPeerCache(r) } @@ -261,13 +277,15 @@ func (r *Router) channelMessagesUpdatesWithPeerCache(ctx context.Context, viewer if date == 0 { date = int(r.clock.Now().Unix()) } - return &tg.Updates{ + out := &tg.Updates{ Updates: updates, Users: tgUsersForViewer(viewerUserID, cache.usersForIDs(ctx, viewerUserID, peerIDMapKeys(userIDs))), Chats: chats, Date: date, Seq: 0, } + applyUsernamesFromRegistry(out.Users, out.Chats, usernames) + return out } func (r *Router) channelEditMessageUpdates(ctx context.Context, viewerUserID int64, res domain.EditChannelMessageResult) *tg.Updates { @@ -275,6 +293,14 @@ func (r *Router) channelEditMessageUpdates(ctx context.Context, viewerUserID int } func (r *Router) channelEditMessageUpdatesWithPeerCache(ctx context.Context, viewerUserID int64, res domain.EditChannelMessageResult, cache *viewerPeerCache) *tg.Updates { + updates := r.channelEditMessageUpdatesWithPeerCacheAndUsernames(ctx, viewerUserID, res, cache, nil) + if updates != nil { + r.applyUsernamesToPeerObjects(ctx, updates.Users, updates.Chats) + } + return updates +} + +func (r *Router) channelEditMessageUpdatesWithPeerCacheAndUsernames(ctx context.Context, viewerUserID int64, res domain.EditChannelMessageResult, cache *viewerPeerCache, usernames map[domain.Peer][]domain.Username) *tg.Updates { if cache == nil { cache = newViewerPeerCache(r) } @@ -297,13 +323,15 @@ func (r *Router) channelEditMessageUpdatesWithPeerCache(ctx context.Context, vie } chats := []tg.ChatClass{tgChannelChatMin(viewerUserID, res.Channel)} chats = append(chats, tgChannels(viewerUserID, cache.channelsForIDs(ctx, viewerUserID, peerIDsExcept(peerIDMapKeys(channelIDs), res.Channel.ID)))...) - return &tg.Updates{ + out := &tg.Updates{ Updates: updates, Users: tgUsersForViewer(viewerUserID, cache.usersForIDs(ctx, viewerUserID, peerIDMapKeys(userIDs))), Chats: chats, Date: int(r.clock.Now().Unix()), Seq: 0, } + applyUsernamesFromRegistry(out.Users, out.Chats, usernames) + return out } func (r *Router) channelDeleteMessagesUpdates(viewerUserID int64, channel domain.Channel, event domain.ChannelUpdateEvent) *tg.Updates { diff --git a/internal/rpc/collectible_usernames_rpc_test.go b/internal/rpc/collectible_usernames_rpc_test.go index 7cc54245..8cac6c4a 100644 --- a/internal/rpc/collectible_usernames_rpc_test.go +++ b/internal/rpc/collectible_usernames_rpc_test.go @@ -3,6 +3,7 @@ package rpc import ( "context" "errors" + "reflect" "strings" "testing" "time" @@ -287,6 +288,140 @@ func TestUsersGetUsersProjectsCollectibleUsernamesInOneBatch(t *testing.T) { } } +func TestMessageEchoProjectsCompleteUsernamesInOneBatch(t *testing.T) { + registry := newFakeUsernameRegistry() + f := newUsernameProjectionFixture(t, registry) + registry.byPeer[domain.Peer{Type: domain.PeerTypeUser, ID: f.owner.ID}] = []domain.Username{ + {Username: "owner_slot", Editable: true, Active: true, SortOrder: 0}, + {Username: "owner_collectible", Active: true, SortOrder: 1, CollectibleID: 21}, + } + registry.byPeer[domain.Peer{Type: domain.PeerTypeUser, ID: f.friend.ID}] = []domain.Username{ + {Username: "friend_slot", Editable: true, Active: true, SortOrder: 0}, + {Username: "friend_collectible", Active: true, SortOrder: 1, CollectibleID: 22}, + } + + users := f.router.usersForMessageUpdate(context.Background(), f.owner.ID, domain.Message{ + OwnerUserID: f.owner.ID, + From: domain.Peer{Type: domain.PeerTypeUser, ID: f.owner.ID}, + Peer: domain.Peer{Type: domain.PeerTypeUser, ID: f.friend.ID}, + }) + if len(users) != 2 { + t.Fatalf("message echo users = %d, want owner and friend", len(users)) + } + want := map[int64][]string{ + f.owner.ID: {"owner_slot", "owner_collectible"}, + f.friend.ID: {"friend_slot", "friend_collectible"}, + } + for _, item := range users { + user, ok := item.(*tg.User) + if !ok { + t.Fatalf("message echo user = %T, want *tg.User", item) + } + vector, set := user.GetUsernames() + if !set || !reflect.DeepEqual(usernameStrings(vector), want[user.ID]) { + t.Fatalf("user %d usernames = %v (set %v), want %v", user.ID, usernameStrings(vector), set, want[user.ID]) + } + } + if registry.batchCalls != 1 || registry.peerCalls != 0 { + t.Fatalf("registry reads = batch %d / peer %d, want one batch read for the response", registry.batchCalls, registry.peerCalls) + } +} + +func TestChannelMessageUpdatesProjectCompleteUsernames(t *testing.T) { + const ( + viewerUserID = int64(1001) + senderUserID = int64(1002) + channelID = int64(2001) + ) + registry := newFakeUsernameRegistry() + registry.byPeer[domain.Peer{Type: domain.PeerTypeUser, ID: senderUserID}] = []domain.Username{ + {Username: "channel_sender", Editable: true, Active: true, SortOrder: 0}, + {Username: "channel_collectible", Active: true, SortOrder: 1, CollectibleID: 31}, + } + router := New(Config{}, Deps{ + Users: mapUsersService{users: map[int64]domain.User{ + senderUserID: {ID: senderUserID, FirstName: "Sender", Username: "channel_sender"}, + }}, + Usernames: registry, + }, zaptest.NewLogger(t), clock.System) + message := domain.ChannelMessage{ + ID: 41, + ChannelID: channelID, + SenderUserID: senderUserID, + From: domain.Peer{Type: domain.PeerTypeUser, ID: senderUserID}, + Date: 1700000500, + Pts: 9, + } + updates := router.channelMessageUpdatesWithPeerCache(context.Background(), viewerUserID, domain.SendChannelMessageResult{ + Channel: domain.Channel{ID: channelID, AccessHash: 22, Title: "group", Megagroup: true, Date: 1700000000}, + Message: message, + Event: domain.ChannelUpdateEvent{ + ChannelID: channelID, + Type: domain.ChannelUpdateNewMessage, + Pts: 9, + PtsCount: 1, + Date: message.Date, + Message: message, + }, + }, 0, newViewerPeerCache(router)) + if updates == nil || len(updates.Users) != 1 { + t.Fatalf("channel updates users = %+v, want one sender", updates) + } + user := updates.Users[0].(*tg.User) + vector, set := user.GetUsernames() + if !set || !reflect.DeepEqual(usernameStrings(vector), []string{"channel_sender", "channel_collectible"}) { + t.Fatalf("channel sender usernames = %v (set %v), want complete vector", usernameStrings(vector), set) + } + if registry.peerCalls != 0 || registry.batchCalls != 1 { + t.Fatalf("registry reads = peer %d / batch %d, want one batched user+channel read", registry.peerCalls, registry.batchCalls) + } +} + +func TestChannelDifferenceProjectsCompleteUsernames(t *testing.T) { + const ( + viewerUserID = int64(1001) + senderUserID = int64(1002) + channelID = int64(2001) + ) + registry := newFakeUsernameRegistry() + registry.byPeer[domain.Peer{Type: domain.PeerTypeUser, ID: senderUserID}] = []domain.Username{ + {Username: "difference_sender", Editable: true, Active: true, SortOrder: 0}, + {Username: "difference_collectible", Active: true, SortOrder: 1, CollectibleID: 41}, + } + router := New(Config{}, Deps{Usernames: registry}, zaptest.NewLogger(t), clock.System) + out := router.tgChannelDifference(context.Background(), viewerUserID, domain.ChannelDifference{ + Final: true, + Pts: 9, + Channel: domain.Channel{ID: channelID, AccessHash: 22, Title: "group", Megagroup: true, Date: 1700000000}, + Self: domain.ChannelMember{ChannelID: channelID, UserID: viewerUserID, Status: domain.ChannelMemberActive}, + Users: []domain.User{{ + ID: senderUserID, + FirstName: "Sender", + Username: "difference_sender", + }}, + NewMessages: []domain.ChannelMessage{{ + ID: 51, + ChannelID: channelID, + SenderUserID: senderUserID, + From: domain.Peer{Type: domain.PeerTypeUser, ID: senderUserID}, + Date: 1700000600, + Pts: 9, + }}, + }) + diff, ok := out.(*tg.UpdatesChannelDifference) + if !ok || len(diff.Users) != 1 { + t.Fatalf("channel difference = %T %+v, want one user", out, out) + } + user := diff.Users[0].(*tg.User) + vector, set := user.GetUsernames() + if !set || !reflect.DeepEqual(usernameStrings(vector), []string{"difference_sender", "difference_collectible"}) { + t.Fatalf("channel difference usernames = %v (set %v), want complete vector", usernameStrings(vector), set) + } + if registry.batchCalls != 1 || registry.peerCalls != 0 { + t.Fatalf("registry reads = batch %d / peer %d, want one batched user+channel read", registry.batchCalls, registry.peerCalls) + } +} + func TestUsersGetUsersDegradesWithoutRegistry(t *testing.T) { f := newUsernameProjectionFixture(t, nil) ctx := WithUserID(context.Background(), f.owner.ID) diff --git a/internal/rpc/fragment.go b/internal/rpc/fragment.go index d0c16966..ca36b661 100644 --- a/internal/rpc/fragment.go +++ b/internal/rpc/fragment.go @@ -255,6 +255,57 @@ func (r *Router) applyUsernamesToPeerObjects(ctx context.Context, users []tg.Use } peers := make([]domain.Peer, 0, len(users)+len(chats)) seen := make(map[domain.Peer]struct{}, len(users)+len(chats)) + peers = appendUsernameProjectionPeers(peers, seen, users, chats) + if len(peers) == 0 { + return + } + byPeer := r.usernameRegistryMap(ctx, peers) + if len(byPeer) == 0 { + return + } + applyUsernamesFromRegistry(users, chats, byPeer) +} + +// applyUsernamesToUpdatesBatch projects one username-registry snapshot over a +// whole outbox claim. A claim may contain repeated peer objects for several +// events and viewers; collecting the peer union first keeps the hot path at one +// registry round trip rather than one read per event or online session. +func (r *Router) applyUsernamesToUpdatesBatch(ctx context.Context, updates []*tg.Updates) { + if r.deps.Usernames == nil || len(updates) == 0 { + return + } + peerCapacity := 0 + for _, update := range updates { + if update != nil { + peerCapacity += len(update.Users) + len(update.Chats) + } + } + if peerCapacity == 0 { + return + } + peers := make([]domain.Peer, 0, peerCapacity) + seen := make(map[domain.Peer]struct{}, peerCapacity) + for _, update := range updates { + if update == nil { + continue + } + peers = appendUsernameProjectionPeers(peers, seen, update.Users, update.Chats) + } + if len(peers) == 0 { + return + } + byPeer := r.usernameRegistryMap(ctx, peers) + if len(byPeer) == 0 { + return + } + for _, update := range updates { + if update != nil { + applyUsernamesFromRegistry(update.Users, update.Chats, byPeer) + } + } +} + +func appendUsernameProjectionPeers(peers []domain.Peer, seen map[domain.Peer]struct{}, users []tg.UserClass, chats []tg.ChatClass) []domain.Peer { addPeer := func(peer domain.Peer) { if peer.ID == 0 { return @@ -275,14 +326,7 @@ func (r *Router) applyUsernamesToPeerObjects(ctx context.Context, users []tg.Use addPeer(domain.Peer{Type: domain.PeerTypeChannel, ID: ch.ID}) } } - if len(peers) == 0 { - return - } - byPeer := r.usernameRegistryMap(ctx, peers) - if len(byPeer) == 0 { - return - } - applyUsernamesFromRegistry(users, chats, byPeer) + return peers } // applyUsernamesFromRegistry applies a previously loaded registry snapshot. diff --git a/internal/rpc/messages_send.go b/internal/rpc/messages_send.go index e253df37..e3beb3d9 100644 --- a/internal/rpc/messages_send.go +++ b/internal/rpc/messages_send.go @@ -596,6 +596,10 @@ func (r *Router) usersForMessageUpdate(ctx context.Context, ownerUserID int64, m if msg.Media != nil && msg.Media.Contact != nil { add(msg.Media.Contact.UserID) } + // A non-min User replaces the cached peer on iOS. Keep the complete + // username vector on synchronous message echoes instead of letting this + // response regress a previously hydrated profile to the legacy scalar. + r.applyUsernamesToPeerObjects(ctx, users, nil) return users } @@ -658,6 +662,7 @@ func (r *Router) usersForMessageUpdates(ctx context.Context, ownerUserID int64, } } } + r.applyUsernamesToPeerObjects(ctx, users, nil) return users } diff --git a/internal/rpc/outbox_dispatcher_test.go b/internal/rpc/outbox_dispatcher_test.go index 32940992..ed831ce4 100644 --- a/internal/rpc/outbox_dispatcher_test.go +++ b/internal/rpc/outbox_dispatcher_test.go @@ -293,6 +293,83 @@ func TestRouterBuildOutboxUpdatesProjectsSenderPerViewerAndCaches(t *testing.T) } } +func TestRouterBuildOutboxUpdatesProjectsUsernamesOncePerClaim(t *testing.T) { + const ( + viewerUserID = int64(1000000003) + senderAUserID = int64(1000000001) + senderBUserID = int64(1000000002) + ) + users := &countingOutboxUsersService{users: map[int64]domain.User{ + senderAUserID: {ID: senderAUserID, FirstName: "Sender A", Username: "sender_a"}, + senderBUserID: {ID: senderBUserID, FirstName: "Sender B", Username: "sender_b"}, + }} + registry := newFakeUsernameRegistry() + registry.byPeer[domain.Peer{Type: domain.PeerTypeUser, ID: senderAUserID}] = []domain.Username{ + {Username: "sender_a", Editable: true, Active: true, SortOrder: 0}, + {Username: "sender_a_collectible", Active: true, SortOrder: 1, CollectibleID: 11}, + } + registry.byPeer[domain.Peer{Type: domain.PeerTypeUser, ID: senderBUserID}] = []domain.Username{ + {Username: "sender_b", Editable: true, Active: true, SortOrder: 0}, + {Username: "sender_b_collectible", Active: true, SortOrder: 1, CollectibleID: 12}, + } + router := New(Config{}, Deps{Users: users, Usernames: registry}, zaptest.NewLogger(t), clock.System) + + senderIDs := []int64{senderAUserID, senderBUserID, senderAUserID} + requests := make([]OutboxUpdateRequest, 0, len(senderIDs)) + for i, senderID := range senderIDs { + msg := domain.Message{ + ID: 20 + i, + OwnerUserID: viewerUserID, + Peer: domain.Peer{Type: domain.PeerTypeUser, ID: senderID}, + From: domain.Peer{Type: domain.PeerTypeUser, ID: senderID}, + Date: 1700000400 + i, + Body: "hello", + Pts: 20 + i, + } + requests = append(requests, OutboxUpdateRequest{ + TargetUserID: viewerUserID, + Event: domain.UpdateEvent{ + UserID: viewerUserID, + Type: domain.UpdateEventNewMessage, + Pts: msg.Pts, + PtsCount: 1, + Date: msg.Date, + Message: msg, + }, + }) + } + + updates := router.BuildOutboxUpdates(context.Background(), requests) + if len(updates) != len(requests) { + t.Fatalf("updates count = %d, want %d", len(updates), len(requests)) + } + for i, update := range updates { + if update == nil || len(update.Users) != 1 { + t.Fatalf("updates[%d].Users = %+v, want one sender", i, update) + } + user, ok := update.Users[0].(*tg.User) + if !ok { + t.Fatalf("updates[%d].Users[0] = %T, want *tg.User", i, update.Users[0]) + } + wantScalar := "sender_a" + wantCollectible := "sender_a_collectible" + if senderIDs[i] == senderBUserID { + wantScalar = "sender_b" + wantCollectible = "sender_b_collectible" + } + if scalar, set := user.GetUsername(); !set || scalar != wantScalar { + t.Fatalf("updates[%d] scalar username = %q (set %v), want %q", i, scalar, set, wantScalar) + } + vector, set := user.GetUsernames() + if !set || !reflect.DeepEqual(usernameStrings(vector), []string{wantScalar, wantCollectible}) { + t.Fatalf("updates[%d] usernames = %v (set %v), want [%s %s]", i, usernameStrings(vector), set, wantScalar, wantCollectible) + } + } + if registry.batchCalls != 1 || registry.peerCalls != 0 { + t.Fatalf("registry reads = batch %d / peer %d, want one batch read for the whole claim", registry.batchCalls, registry.peerCalls) + } +} + func TestRouterBuildOutboxUpdatesSeparatesViewerCache(t *testing.T) { const senderUserID = int64(1000000001) users := &viewerSpecificOutboxUsersService{} diff --git a/internal/rpc/outbox_update_builder.go b/internal/rpc/outbox_update_builder.go index b5b442d1..66fe7a65 100644 --- a/internal/rpc/outbox_update_builder.go +++ b/internal/rpc/outbox_update_builder.go @@ -36,11 +36,15 @@ func (r *Router) BuildOutboxUpdates(ctx context.Context, requests []OutboxUpdate for i, item := range items { update := tgUpdateForOutboxEventForViewer(events[i], viewerUserID) if peers := storyUpdateEventPeers(events[i]); len(peers) > 0 { - update = r.withStoryUpdatePeerObjects(ctx, viewerUserID, update, peers...) + update = r.withStoryUpdatePeerObjectsForOutbox(ctx, viewerUserID, update, peers...) } out[item.index] = update } } + // Username rows are viewer-independent. Project the union once after every + // viewer-specific update has been built so one outbox claim never turns into + // a registry query per event/session. + r.applyUsernamesToUpdatesBatch(ctx, out) return out } diff --git a/internal/rpc/story_peer_projection.go b/internal/rpc/story_peer_projection.go index eb57bc09..a06d7c11 100644 --- a/internal/rpc/story_peer_projection.go +++ b/internal/rpc/story_peer_projection.go @@ -217,6 +217,17 @@ func (r *Router) tgUpdatesDifference(ctx context.Context, viewerUserID int64, di return out } +func (r *Router) tgChannelDifference(ctx context.Context, viewerUserID int64, diff domain.ChannelDifference) tg.UpdatesChannelDifferenceClass { + out := tgChannelDifference(viewerUserID, diff) + switch v := out.(type) { + case *tg.UpdatesChannelDifference: + r.applyPeerReadModels(ctx, viewerUserID, v.Users, v.Chats) + case *tg.UpdatesChannelDifferenceTooLong: + r.applyPeerReadModels(ctx, viewerUserID, v.Users, v.Chats) + } + return out +} + func (r *Router) withStoryUpdatePeerObjects(ctx context.Context, viewerUserID int64, updates *tg.Updates, peers ...domain.Peer) *tg.Updates { if updates == nil { return nil @@ -232,6 +243,28 @@ func (r *Router) withStoryUpdatePeerObjects(ctx context.Context, viewerUserID in return updates } +// withStoryUpdatePeerObjectsForOutbox keeps the viewer-specific story overlay +// local to one update while deferring viewer-independent username projection to +// BuildOutboxUpdates' claim-wide pass. This avoids turning story events into an +// extra username-registry query per event before the final batch projection. +func (r *Router) withStoryUpdatePeerObjectsForOutbox(ctx context.Context, viewerUserID int64, updates *tg.Updates, peers ...domain.Peer) *tg.Updates { + if updates == nil { + return nil + } + users, channels := r.storyPeerObjects(ctx, viewerUserID, peers) + if len(users) > 0 { + projected := tgUsersForViewer(viewerUserID, r.withUsersPresence(users)) + updates.Users = appendUniqueTGUsers(updates.Users, projected...) + } + if len(channels) > 0 { + updates.Chats = appendUniqueTGChats(updates.Chats, tgChannels(viewerUserID, channels)...) + } + r.withBotProfileFlagsForUsers(ctx, updates.Users) + r.applyStoryMaxIDsToPeerObjects(ctx, viewerUserID, updates.Users, updates.Chats) + r.applyBotVerificationIconsToPeerObjects(ctx, updates.Users, updates.Chats) + return updates +} + func (r *Router) withStoryListPeerObjects(ctx context.Context, viewerUserID int64, list domain.StoryList) domain.StoryList { users, channels := r.storyPeerObjects(ctx, viewerUserID, storyListOwnerPeers(list)) if len(users) > 0 { diff --git a/internal/rpc/story_peer_projection_rpc_test.go b/internal/rpc/story_peer_projection_rpc_test.go index 0c050a79..0998a595 100644 --- a/internal/rpc/story_peer_projection_rpc_test.go +++ b/internal/rpc/story_peer_projection_rpc_test.go @@ -2,6 +2,7 @@ package rpc import ( "context" + "reflect" "sync" "testing" "time" @@ -1264,9 +1265,15 @@ func TestBuildOutboxStoryUpdatesHydratesCompanionPeersWithStoriesMaxID(t *testin if _, err := storyStore.UpsertStory(ctx, domain.UpsertStoryRequest{Story: story}); err != nil { t.Fatalf("upsert story: %v", err) } + registry := newFakeUsernameRegistry() + registry.byPeer[ownerPeer] = []domain.Username{ + {Username: "story_owner", Editable: true, Active: true, SortOrder: 0}, + {Username: "story_collectible", Active: true, SortOrder: 1, CollectibleID: 51}, + } r := New(Config{}, Deps{ - Users: appusers.NewService(userStore), - Stories: appstories.NewService(storyStore), + Users: appusers.NewService(userStore), + Stories: appstories.NewService(storyStore), + Usernames: registry, }, zaptest.NewLogger(t), fixedClock{now: time.Unix(1700000300, 0)}) updates := r.BuildOutboxUpdates(ctx, []OutboxUpdateRequest{{ @@ -1290,7 +1297,16 @@ func TestBuildOutboxStoryUpdatesHydratesCompanionPeersWithStoriesMaxID(t *testin if _, ok := updates[0].Updates[0].(*tg.UpdateStory); !ok { t.Fatalf("first update = %T, want updateStory", updates[0].Updates[0]) } - assertUserStoryMaxID(t, findUserClass(updates[0].Users, owner.ID), 13) + ownerUser := findUserClass(updates[0].Users, owner.ID) + assertUserStoryMaxID(t, ownerUser, 13) + projectedOwner := ownerUser.(*tg.User) + vector, set := projectedOwner.GetUsernames() + if !set || !reflect.DeepEqual(usernameStrings(vector), []string{"story_owner", "story_collectible"}) { + t.Fatalf("story owner usernames = %v (set %v), want complete vector", usernameStrings(vector), set) + } + if registry.peerCalls != 1 || registry.batchCalls != 0 { + t.Fatalf("username registry reads = peer %d / batch %d, want one claim-wide read", registry.peerCalls, registry.batchCalls) + } } func TestBuildOutboxNewStoryReactionHydratesReactorUser(t *testing.T) {