astra/owpengram-server
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

fix(rpc): sync preserve complete username projections

Browse source
This commit is contained in:
iamxvbaba 2026-07-31 20:42:25 +08:00
parent 464d4edb4a
commit ee74d941bb
10 changed files with 411 additions and 22 deletions
Download patch file Download diff file Expand all files Collapse all files

53
internal/rpc/channel_fanout_dispatcher.go
View file

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

6
internal/rpc/channel_fanout_dispatcher_test.go
View file

@ -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 记录最近一次推送，

36
internal/rpc/channels_updates.go
View file

@ -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 {

135
internal/rpc/collectible_usernames_rpc_test.go
View file

@ -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)

60
internal/rpc/fragment.go
View file

@ -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.

5
internal/rpc/messages_send.go
View file

@ -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
}

77
internal/rpc/outbox_dispatcher_test.go
View file

@ -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{}

6
internal/rpc/outbox_update_builder.go
View file

@ -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
}

33
internal/rpc/story_peer_projection.go
View file

@ -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 {

22
internal/rpc/story_peer_projection_rpc_test.go
View file

@ -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) {