diff --git a/internal/mtprotoedge/session_manager.go b/internal/mtprotoedge/session_manager.go index cd7691f1..07dbec49 100644 --- a/internal/mtprotoedge/session_manager.go +++ b/internal/mtprotoedge/session_manager.go @@ -1307,24 +1307,6 @@ func (m *SessionManager) ReceivesUpdatesForAuthKey(authKeyID [8]byte, sessionID return hasProfile && c.receivesUpdates.Load() && c.membershipsSynced.Load() } -// UpdatesActivationStartedForAuthKey reports whether the session has completed -// membership synchronization and either became ready or is draining its older -// pending FIFO. Unlike ReceivesUpdatesForAuthKey, flushing is a positive result: -// another delivery callback must not rebuild memberships or enqueue a duplicate -// cache-convergence update behind the same pending batch. -func (m *SessionManager) UpdatesActivationStartedForAuthKey(authKeyID [8]byte, sessionID int64) bool { - m.mu.RLock() - key := sessionKey{authKeyID: authKeyID, sessionID: sessionID} - c, ok := m.bySession[key] - flushing := m.flushing[key] - m.mu.RUnlock() - if !ok { - return false - } - _, hasProfile := c.LayerProfile() - return hasProfile && c.membershipsSynced.Load() && (c.receivesUpdates.Load() || flushing) -} - // SetReceivesUpdatesForAuthKey 标记指定 raw auth_key_id + session_id 是否接收主动 updates。 func (m *SessionManager) SetReceivesUpdatesForAuthKey(authKeyID [8]byte, sessionID int64, receives bool) { m.mu.Lock() diff --git a/internal/mtprotoedge/session_ready_test.go b/internal/mtprotoedge/session_ready_test.go index 38c9f7f4..dc0a2725 100644 --- a/internal/mtprotoedge/session_ready_test.go +++ b/internal/mtprotoedge/session_ready_test.go @@ -26,36 +26,11 @@ func TestReceivesUpdatesForAuthKeyRequiresMembershipSync(t *testing.T) { if sm.ReceivesUpdatesForAuthKey(raw, 42) { t.Fatal("ready before membership sync — a failed sync would never be retried") } - if sm.UpdatesActivationStartedForAuthKey(raw, 42) { - t.Fatal("activation started before membership sync") - } sm.SetSessionChannelMemberships(raw, 42, 100, []int64{7}, sm.ChannelMembershipGeneration(raw, 42)) if !sm.ReceivesUpdatesForAuthKey(raw, 42) { t.Fatal("not ready after successful membership sync") } - if !sm.UpdatesActivationStartedForAuthKey(raw, 42) { - t.Fatal("activation not started after successful membership sync") - } - - // Pending FIFO drain counts as an activation already in progress even though - // the session is not fully ready yet. A stale RPC callback must not enqueue a - // duplicate cache-convergence update behind that same FIFO. - key := sessionKey{authKeyID: raw, sessionID: 42} - sm.mu.Lock() - c.receivesUpdates.Store(false) - sm.flushing[key] = true - sm.mu.Unlock() - if sm.ReceivesUpdatesForAuthKey(raw, 42) { - t.Fatal("flushing session reported fully ready") - } - if !sm.UpdatesActivationStartedForAuthKey(raw, 42) { - t.Fatal("flushing session did not report activation started") - } - sm.mu.Lock() - delete(sm.flushing, key) - c.receivesUpdates.Store(true) - sm.mu.Unlock() // 没有任何频道的账号：空列表的成功同步同样算就绪。 sm.SetSessionChannelMemberships(raw, 42, 100, nil, sm.ChannelMembershipGeneration(raw, 42)) diff --git a/internal/rpc/account.go b/internal/rpc/account.go index d17ad220..06c33fb3 100644 --- a/internal/rpc/account.go +++ b/internal/rpc/account.go @@ -1100,13 +1100,15 @@ func (r *Router) tgAccountPrivacyRules(ctx context.Context, viewerUserID int64, return nil, internalErr() } } - return &tg.AccountPrivacyRules{ + out := &tg.AccountPrivacyRules{ Rules: tgPrivacyRules(rules.Rules), // viewer 可能把自己（inputUserSelf）写进隐私名单，须带 self 标志，否则下发的 // self=false user 会被 DrKLO putUsers 覆盖账号缓存。 Users: tgUsersForViewer(viewerUserID, users), Chats: []tg.ChatClass{}, - }, nil + } + r.applyPeerReadModels(ctx, viewerUserID, out.Users, out.Chats) + return out, nil } func (r *Router) domainPrivacyRulesFromInput(ctx context.Context, userID int64, in []tg.InputPrivacyRuleClass) ([]domain.PrivacyRule, error) { @@ -1534,8 +1536,7 @@ func (r *Router) onAccountUpdateProfile(ctx context.Context, req *tg.AccountUpda return nil, profileErr(err) } r.invalidateRPCProjectionForUser(u.ID) - r.pushUsernameUpdate(ctx, u) - return r.tgSelfUser(u), nil + return r.pushUsernameUpdate(ctx, u), nil } func (r *Router) onAccountCheckUsername(ctx context.Context, username string) (bool, error) { @@ -1568,8 +1569,7 @@ func (r *Router) onAccountUpdateUsername(ctx context.Context, username string) ( return nil, usernameErr(err) } r.invalidateRPCProjectionForUser(u.ID) - r.pushUsernameUpdate(ctx, u) - return r.tgSelfUser(u), nil + return r.pushUsernameUpdate(ctx, u), nil } // onAccountReorderUsernames rewrites the caller's active username order @@ -1730,12 +1730,13 @@ func (r *Router) onAccountUpdateEmojiStatus(ctx context.Context, status tg.Emoji } r.invalidateRPCProjectionForUser(u.ID) update := &tg.UpdateUserEmojiStatus{UserID: u.ID, EmojiStatus: tgUserEmojiStatusValue(value)} + self := r.tgSelfUserWithUsernames(ctx, u) if durableWrite { if sessionID != 0 { r.bookkeepAuxPtsForCurrentSession(ctx, event) } r.pushUserUpdatesIfNoReliableDispatch(ctx, u.ID, &tg.Updates{ - Updates: []tg.UpdateClass{update}, Users: []tg.UserClass{r.tgSelfUser(u)}, Date: event.Date, + Updates: []tg.UpdateClass{update}, Users: []tg.UserClass{self}, Date: event.Date, }) } else if updates, ok := r.deps.Updates.(UserEmojiStatusUpdatesService); ok { event, _, recordErr := updates.RecordUserEmojiStatus(ctx, authKeyID, userID, value, rawAuthKeyIDForOrigin(ctx), sessionID) @@ -1746,13 +1747,13 @@ func (r *Router) onAccountUpdateEmojiStatus(ctx context.Context, status tg.Emoji r.bookkeepAuxPtsForCurrentSession(ctx, event) } r.pushUserUpdatesIfNoReliableDispatch(ctx, u.ID, &tg.Updates{ - Updates: []tg.UpdateClass{update}, Users: []tg.UserClass{r.tgSelfUser(u)}, Date: event.Date, + Updates: []tg.UpdateClass{update}, Users: []tg.UserClass{self}, Date: event.Date, }) } else { // Lightweight test deployments without the durable extension retain the // previous online-only behavior; production wiring implements it. r.pushUserUpdates(ctx, u.ID, &tg.Updates{ - Updates: []tg.UpdateClass{update}, Users: []tg.UserClass{r.tgSelfUser(u)}, Date: int(r.clock.Now().Unix()), + Updates: []tg.UpdateClass{update}, Users: []tg.UserClass{self}, Date: int(r.clock.Now().Unix()), }) } return true, nil @@ -1826,7 +1827,7 @@ func (r *Router) onAccountUpdateColor(ctx context.Context, req *tg.AccountUpdate r.invalidateRPCProjectionForUser(u.ID) r.pushUserUpdates(ctx, u.ID, &tg.Updates{ Updates: []tg.UpdateClass{&tg.UpdateUser{UserID: u.ID}}, - Users: []tg.UserClass{r.tgSelfUser(u)}, + Users: []tg.UserClass{r.tgSelfUserWithUsernames(ctx, u)}, Date: int(r.clock.Now().Unix()), }) return true, nil @@ -1930,18 +1931,22 @@ func (r *Router) onAccountGetCollectibleEmojiStatuses(ctx context.Context, hash return &tg.AccountEmojiStatuses{Hash: catalogHash, Statuses: statuses}, nil } -func (r *Router) pushUsernameUpdate(ctx context.Context, u domain.User) { - if u.ID == 0 { - return - } +func (r *Router) pushUsernameUpdate(ctx context.Context, u domain.User) *tg.User { // updateUserName carries the vector clients persist, so it has to be the full // registry list when one exists. One peer, one registry read; the overlay // degrades to tgUsernames(u.Username) whenever the registry is unavailable. + // Keep the RPC result and pushed update as distinct tg.User objects: TL Encode + // recomputes flags, so sharing one pointer across response and push delivery + // would make otherwise independent encoders mutate the same object. self := r.tgSelfUser(u) - users := []tg.UserClass{self} + pushedSelf := r.tgSelfUser(u) + users := []tg.UserClass{self, pushedSelf} r.applyUsernamesToPeerObjects(ctx, users, nil) + if u.ID == 0 { + return self + } usernames := tgUsernames(u.Username) - if vector, ok := self.GetUsernames(); ok && len(vector) > 0 { + if vector, ok := pushedSelf.GetUsernames(); ok && len(vector) > 0 { usernames = vector } r.pushUserUpdates(ctx, u.ID, &tg.Updates{ @@ -1951,9 +1956,10 @@ func (r *Router) pushUsernameUpdate(ctx context.Context, u domain.User) { LastName: u.LastName, Usernames: usernames, }}, - Users: users, + Users: []tg.UserClass{pushedSelf}, Date: int(r.clock.Now().Unix()), }) + return self } func (r *Router) pushSelfUserChangedUpdate(ctx context.Context, u domain.User) { diff --git a/internal/rpc/account_business.go b/internal/rpc/account_business.go index 629fd3f8..594ec716 100644 --- a/internal/rpc/account_business.go +++ b/internal/rpc/account_business.go @@ -239,6 +239,7 @@ func (r *Router) onAccountResolveBusinessChatLink(ctx context.Context, slug stri } } } + r.applyPeerReadModels(ctx, viewerID, users, nil) return &tg.AccountResolvedBusinessChatLinks{ Peer: &tg.PeerUser{UserID: link.OwnerUserID}, Message: link.Message, @@ -271,10 +272,12 @@ func (r *Router) onAccountGetConnectedBots(ctx context.Context) (*tg.AccountConn if !found { return &tg.AccountConnectedBots{ConnectedBots: []tg.ConnectedBot{}, Users: []tg.UserClass{}}, nil } - return &tg.AccountConnectedBots{ + out := &tg.AccountConnectedBots{ ConnectedBots: []tg.ConnectedBot{tgConnectedBot(bot)}, Users: []tg.UserClass{r.tgUser(botUser)}, - }, nil + } + r.applyPeerReadModels(ctx, userID, out.Users, nil) + return out, nil } func (r *Router) onAccountUpdateConnectedBot(ctx context.Context, req *tg.AccountUpdateConnectedBotRequest) (tg.UpdatesClass, error) { @@ -301,7 +304,7 @@ func (r *Router) onAccountUpdateConnectedBot(ctx context.Context, req *tg.Accoun return nil, businessAutomationErr(err) } r.invalidateRPCProjectionForViewer(userID) - return r.connectedBusinessBotEmptyUpdates(botUser), nil + return r.connectedBusinessBotEmptyUpdates(ctx, userID, botUser), nil } recipients, err := r.domainBusinessBotRecipients(ctx, userID, req.Recipients) if err != nil { @@ -316,7 +319,7 @@ func (r *Router) onAccountUpdateConnectedBot(ctx context.Context, req *tg.Accoun return nil, businessAutomationErr(err) } r.invalidateRPCProjectionForViewer(userID) - return r.connectedBusinessBotEmptyUpdates(botUser, tgConnectedBot(saved)), nil + return r.connectedBusinessBotEmptyUpdates(ctx, userID, botUser, tgConnectedBot(saved)), nil } func (r *Router) onAccountToggleConnectedBotPaused(ctx context.Context, req *tg.AccountToggleConnectedBotPausedRequest) (bool, error) { @@ -410,17 +413,19 @@ func connectedBusinessBotUsable(u domain.User) bool { return u.Bot && u.ID != 0 && u.ID != domain.BotFatherUserID } -func (r *Router) connectedBusinessBotEmptyUpdates(botUser domain.User, bots ...tg.ConnectedBot) *tg.Updates { +func (r *Router) connectedBusinessBotEmptyUpdates(ctx context.Context, viewerUserID int64, botUser domain.User, bots ...tg.ConnectedBot) *tg.Updates { users := []tg.UserClass{} if botUser.ID != 0 { users = append(users, r.tgUser(botUser)) } - return &tg.Updates{ + out := &tg.Updates{ Updates: []tg.UpdateClass{}, Users: users, Chats: []tg.ChatClass{}, Date: int(r.clock.Now().Unix()), } + r.applyPeerReadModels(ctx, viewerUserID, out.Users, out.Chats) + return out } func (r *Router) connectedBusinessBotPeerSettings(ctx context.Context, ownerUserID int64, peer domain.Peer, settings domain.PeerSettings) (domain.PeerSettings, error) { diff --git a/internal/rpc/account_notify.go b/internal/rpc/account_notify.go index 8003c5e8..b6c7abda 100644 --- a/internal/rpc/account_notify.go +++ b/internal/rpc/account_notify.go @@ -183,12 +183,14 @@ func (r *Router) onAccountGetNotifyExceptions(ctx context.Context, req *tg.Accou chats = appendUniqueTGChats(chats, tgCommunityChats(views)...) } } - return &tg.Updates{ + out := &tg.Updates{ Updates: updates, Users: r.tgUsersForIDs(ctx, userID, userIDs), Chats: chats, Date: int(r.clock.Now().Unix()), - }, nil + } + r.applyPeerReadModels(ctx, userID, out.Users, out.Chats) + return out, nil } // notifyExceptionQualifies 判定一条异常是否纳入 getNotifyExceptions 结果。 diff --git a/internal/rpc/account_phone.go b/internal/rpc/account_phone.go index b96e5c01..794c9e4f 100644 --- a/internal/rpc/account_phone.go +++ b/internal/rpc/account_phone.go @@ -87,5 +87,5 @@ func (r *Router) onAccountChangePhone(ctx context.Context, req *tg.AccountChange } r.bookkeepAuxPtsForCurrentSession(ctx, result.Event) } - return r.tgSelfUser(result.User), nil + return r.tgSelfUserWithUsernames(ctx, result.User), nil } diff --git a/internal/rpc/account_phone_rpc_test.go b/internal/rpc/account_phone_rpc_test.go index 6a8c5450..3fb76dd5 100644 --- a/internal/rpc/account_phone_rpc_test.go +++ b/internal/rpc/account_phone_rpc_test.go @@ -21,7 +21,7 @@ func TestAccountChangePhoneRPCReturnsSelfPushesOthersAndReplaysDifference(t *tes auths := memory.NewAuthorizationStore() codes := memory.NewCodeStore() events := memory.NewUpdateEventStore() - user, err := users.Create(ctx, domain.User{AccessHash: 401, Phone: "15550013001", FirstName: "Alice"}) + user, err := users.Create(ctx, domain.User{AccessHash: 401, Phone: "15550013001", FirstName: "Alice", Username: "Alice"}) if err != nil { t.Fatalf("create user: %v", err) } @@ -34,8 +34,13 @@ func TestAccountChangePhoneRPCReturnsSelfPushesOthersAndReplaysDifference(t *tes appaccount.WithUsers(users), appaccount.WithPhoneChange(memory.NewPhoneChangeStore(users, events), auths, codes, nil, "12345", time.Minute, 5), ) + registry := newFakeUsernameRegistry() + registry.byPeer[domain.Peer{Type: domain.PeerTypeUser, ID: user.ID}] = []domain.Username{ + {Username: "Alice", Editable: true, Active: true, SortOrder: 0}, + {Username: "aliceCollect0728b", Active: true, SortOrder: 1, CollectibleID: 2}, + } sessions := &captureSessions{} - r := New(Config{}, Deps{Account: accountSvc, Sessions: sessions}, zaptest.NewLogger(t), clock.System) + r := New(Config{}, Deps{Account: accountSvc, Sessions: sessions, Usernames: registry}, zaptest.NewLogger(t), clock.System) reqCtx := WithSessionID(WithAuthKeyID(WithUserID(ctx, user.ID), authKeyID), 77) sentClass, err := r.onAccountSendChangePhoneCode(reqCtx, &tg.AccountSendChangePhoneCodeRequest{PhoneNumber: "+1 555 001 3002"}) @@ -62,6 +67,7 @@ func TestAccountChangePhoneRPCReturnsSelfPushesOthersAndReplaysDifference(t *tes if !ok || self.ID != user.ID || self.Phone != "15550013002" { t.Fatalf("returned self = %T %+v", userClass, userClass) } + assertVectorOnlyUsernames(t, "account.changePhone", self, []string{"Alice", "aliceCollect0728b"}) otherPush, ok := sessions.lastUserPush().(*tg.Updates) if !ok || len(otherPush.Updates) != 2 { diff --git a/internal/rpc/bots_inline.go b/internal/rpc/bots_inline.go index 30c4520d..75765b42 100644 --- a/internal/rpc/bots_inline.go +++ b/internal/rpc/bots_inline.go @@ -1036,6 +1036,7 @@ func (r *Router) tgBotInlineResults(ctx context.Context, viewerUserID int64, in out.Users = append(out.Users, r.tgUser(u)) } } + r.applyPeerReadModels(ctx, viewerUserID, out.Users, nil) return out } diff --git a/internal/rpc/bots_longtail.go b/internal/rpc/bots_longtail.go index e24419cf..ae18d8ba 100644 --- a/internal/rpc/bots_longtail.go +++ b/internal/rpc/bots_longtail.go @@ -464,7 +464,7 @@ func (r *Router) onBotsUpdateUserEmojiStatus(ctx context.Context, req *tg.BotsUp UserID: u.ID, EmojiStatus: tgUserEmojiStatus(u, r.clock.Now().Unix()), }}, - Users: []tg.UserClass{r.tgUser(u)}, + Users: []tg.UserClass{r.tgSelfUserWithUsernames(ctx, u)}, Date: int(r.clock.Now().Unix()), }) return true, nil diff --git a/internal/rpc/channels_core.go b/internal/rpc/channels_core.go index 30505006..6afff68d 100644 --- a/internal/rpc/channels_core.go +++ b/internal/rpc/channels_core.go @@ -144,11 +144,13 @@ func (r *Router) onChannelsGetFullChannel(ctx context.Context, input tg.InputCha view.State.NotifySettings = © } } - return &tg.MessagesChatFull{ + out := &tg.MessagesChatFull{ FullChat: tgCommunityFull(view), Chats: tgCommunityHydratedChats(userID, view), Users: tgUsers(view.Users), - }, nil + } + r.applyPeerReadModels(ctx, userID, out.Users, out.Chats) + return out, nil } if errors.Is(communityErrValue, domain.ErrCommunityPrivate) { return nil, communityErr(communityErrValue) @@ -177,11 +179,12 @@ func (r *Router) onChannelsGetFullChannel(ctx context.Context, input tg.InputCha chats := append([]tg.ChatClass(nil), cached.chats...) chats = r.appendLinkedDiscussionChat(ctx, userID, ref.ID, chats) r.trackChannelInterest(ctx, userID, ref.ID) - r.applyPeerReadModels(ctx, userID, nil, chats) + users := r.tgUsersForIDs(ctx, userID, cached.userIDs) + r.applyPeerReadModels(ctx, userID, users, chats) return &tg.MessagesChatFull{ FullChat: &full, Chats: chats, - Users: r.tgUsersForIDs(ctx, userID, cached.userIDs), + Users: users, }, nil } view, err := r.channelFullReadView(ctx, userID, input) @@ -225,11 +228,12 @@ func (r *Router) onChannelsGetFullChannel(ctx context.Context, input tg.InputCha // projection cache so a revoked badge cannot outlive the change by a cache TTL. r.applyBotVerificationToChannelFull(ctx, view.Channel.ID, full) r.applyAndroidChannelReactionEditorCompat(ctx, full, canChangeInfo) - r.applyPeerReadModels(ctx, userID, nil, chats) + users := r.tgUsersForIDs(ctx, userID, userIDs) + r.applyPeerReadModels(ctx, userID, users, chats) return &tg.MessagesChatFull{ FullChat: full, Chats: chats, - Users: r.tgUsersForIDs(ctx, userID, userIDs), + Users: users, }, nil } @@ -321,11 +325,13 @@ func (r *Router) onChannelsGetSendAs(ctx context.Context, req *tg.ChannelsGetSen chats = append(chats, tgChannels(userID, extras)...) } } - return &tg.ChannelsSendAsPeers{ + out := &tg.ChannelsSendAsPeers{ Peers: peers, Chats: chats, Users: r.tgUsersForIDs(ctx, userID, []int64{userID}), - }, nil + } + r.applyPeerReadModels(ctx, userID, out.Users, out.Chats) + return out, nil } func (r *Router) applyPendingJoinRequestsToFullChannel(ctx context.Context, full *tg.ChannelFull, channelID int64, userIDs []int64) []int64 { diff --git a/internal/rpc/channels_dialogs.go b/internal/rpc/channels_dialogs.go index 62db939a..3c72b0f7 100644 --- a/internal/rpc/channels_dialogs.go +++ b/internal/rpc/channels_dialogs.go @@ -25,6 +25,7 @@ func (r *Router) onChannelsGetLeftChannels(ctx context.Context, offset int) (tg. for _, item := range list.Channels { chats = append(chats, tgChannelChat(userID, item.Channel, &item.Self)) } + r.applyUsernamesToPeerObjects(ctx, nil, chats) if len(chats) == 0 && list.Count > 0 { return &tg.MessagesChatsSlice{Count: list.Count, Chats: chats}, nil } @@ -56,6 +57,7 @@ func (r *Router) onChannelsGetInactiveChannels(ctx context.Context) (*tg.Message dates = append(dates, date) chats = append(chats, tgChannelChatMin(userID, channel)) } + r.applyUsernamesToPeerObjects(ctx, nil, chats) return &tg.MessagesInactiveChats{Dates: dates, Chats: chats, Users: []tg.UserClass{}}, nil } @@ -71,5 +73,7 @@ func (r *Router) onChannelsGetGroupsForDiscussion(ctx context.Context) (tg.Messa if err != nil { return nil, channelInvalidErr(err) } - return &tg.MessagesChats{Chats: tgChannels(userID, channels)}, nil + chats := tgChannels(userID, channels) + r.applyUsernamesToPeerObjects(ctx, nil, chats) + return &tg.MessagesChats{Chats: chats}, nil } diff --git a/internal/rpc/channels_invites.go b/internal/rpc/channels_invites.go index 26d49f11..2f77b58e 100644 --- a/internal/rpc/channels_invites.go +++ b/internal/rpc/channels_invites.go @@ -59,9 +59,12 @@ func (r *Router) onMessagesCheckChatInvite(ctx context.Context, hash string) (tg return nil, channelInviteErr(err) } if res.Already { - // chatInviteAlready#5a686d7c wraps a full Chat, so the badge flags already - // travel through tgChannelChat; nothing to re-apply here. - return &tg.ChatInviteAlready{Chat: tgChannelChat(userID, res.Channel, &res.Self)}, nil + // chatInviteAlready#5a686d7c wraps a full Chat. Run the shared peer + // read-model pass before nesting it so collectible usernames and badge + // facts cannot be lost behind a scalar-only complete Channel. + chat := tgChannelChat(userID, res.Channel, &res.Self) + r.applyPeerReadModels(ctx, userID, nil, []tg.ChatClass{chat}) + return &tg.ChatInviteAlready{Chat: chat}, nil } invite := &tg.ChatInvite{ Channel: true, diff --git a/internal/rpc/channels_legacy_chat.go b/internal/rpc/channels_legacy_chat.go index 1ce197d6..b3d653ff 100644 --- a/internal/rpc/channels_legacy_chat.go +++ b/internal/rpc/channels_legacy_chat.go @@ -149,6 +149,7 @@ func (r *Router) onMessagesGetChats(ctx context.Context, ids []int64) (tg.Messag } } } + r.applyUsernamesToPeerObjects(ctx, nil, chats) return &tg.MessagesChats{Chats: chats}, nil } diff --git a/internal/rpc/channels_members.go b/internal/rpc/channels_members.go index a44a9217..0ad8b911 100644 --- a/internal/rpc/channels_members.go +++ b/internal/rpc/channels_members.go @@ -28,13 +28,17 @@ func (r *Router) onChannelsGetAdminedPublicChannels(ctx context.Context, req *tg if err != nil { return nil, internalErr() } - return &tg.MessagesChats{Chats: tgChannels(userID, channels)}, nil + chats := tgChannels(userID, channels) + r.applyUsernamesToPeerObjects(ctx, nil, chats) + return &tg.MessagesChats{Chats: chats}, nil } channels, err := r.deps.Channels.ListAdminedPublicChannels(ctx, userID) if err != nil { return nil, internalErr() } - return &tg.MessagesChats{Chats: tgChannels(userID, channels)}, nil + chats := tgChannels(userID, channels) + r.applyUsernamesToPeerObjects(ctx, nil, chats) + return &tg.MessagesChats{Chats: chats}, nil } func (r *Router) onChannelsDeleteParticipantHistory(ctx context.Context, req *tg.ChannelsDeleteParticipantHistoryRequest) (*tg.MessagesAffectedHistory, error) { @@ -110,6 +114,7 @@ func (r *Router) onChannelsGetMessageAuthor(ctx context.Context, req *tg.Channel if len(users) == 0 { return nil, peerIDInvalidErr() } + r.applyPeerReadModels(ctx, userID, users, nil) return users[0], nil } diff --git a/internal/rpc/channels_settings.go b/internal/rpc/channels_settings.go index 3157d526..4a1aba97 100644 --- a/internal/rpc/channels_settings.go +++ b/internal/rpc/channels_settings.go @@ -289,6 +289,7 @@ func (r *Router) onChannelsGetChannelRecommendations(ctx context.Context, req *t return nil, channelInvalidErr(err) } chats := tgChannels(userID, res.Channels) + r.applyUsernamesToPeerObjects(ctx, nil, chats) if res.Count > len(chats) { return &tg.MessagesChatsSlice{Count: res.Count, Chats: chats}, nil } diff --git a/internal/rpc/chat_automation_rpc_test.go b/internal/rpc/chat_automation_rpc_test.go index ba5ca95d..67a0ce5d 100644 --- a/internal/rpc/chat_automation_rpc_test.go +++ b/internal/rpc/chat_automation_rpc_test.go @@ -129,6 +129,15 @@ func TestBusinessChatLinkRPCs(t *testing.T) { const userID int64 = 1000000002 ctx := WithUserID(context.Background(), userID) r, _ := newChatAutomationTestRouter(t) + r.deps.Users = mapUsersService{users: map[int64]domain.User{ + userID: {ID: userID, AccessHash: 2002, FirstName: "Business", Username: "business_slot"}, + }} + registry := newFakeUsernameRegistry() + registry.byPeer[domain.Peer{Type: domain.PeerTypeUser, ID: userID}] = []domain.Username{ + {Username: "business_slot", Editable: true, Active: true, SortOrder: 0}, + {Username: "business_collectible", Active: true, SortOrder: 1, CollectibleID: 22}, + } + r.deps.Usernames = registry created, err := r.onAccountCreateBusinessChatLink(ctx, tg.InputBusinessChatLink{ Message: "Prefilled message", @@ -156,6 +165,13 @@ func TestBusinessChatLinkRPCs(t *testing.T) { if !ok || peer.UserID != userID || resolved.Message != "Prefilled message" { t.Fatalf("resolved = %+v", resolved) } + if len(resolved.Users) != 1 { + t.Fatalf("resolved users = %+v, want one", resolved.Users) + } + assertVectorOnlyUsernames(t, "resolved business chat owner", resolved.Users[0].(*tg.User), []string{"business_slot", "business_collectible"}) + if registry.peerCalls != 1 || registry.batchCalls != 0 { + t.Fatalf("resolved business username reads = peer:%d batch:%d, want 1/0", registry.peerCalls, registry.batchCalls) + } list, err = r.onAccountGetBusinessChatLinks(ctx) if err != nil || len(list.Links) != 1 || list.Links[0].Views != 1 { t.Fatalf("post-resolve links = %+v err=%v", list, err) diff --git a/internal/rpc/collectible_usernames_rpc_test.go b/internal/rpc/collectible_usernames_rpc_test.go index 52f96088..cd0bfd5e 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" + "fmt" "reflect" "strings" "testing" @@ -232,6 +233,101 @@ func usernameStrings(list []tg.Username) []string { return out } +type tgUsernamePeer interface { + SetFlags() + GetUsername() (string, bool) + GetUsernames() ([]tg.Username, bool) +} + +func assertVectorOnlyUsernames(t *testing.T, stage string, peer tgUsernamePeer, want []string) []tg.Username { + t.Helper() + peer.SetFlags() + if scalar, set := peer.GetUsername(); set || scalar != "" { + t.Fatalf("%s scalar username = %q (set %v), want absent when usernames vector is present", stage, scalar, set) + } + vector, set := peer.GetUsernames() + if !set || !reflect.DeepEqual(usernameStrings(vector), want) { + t.Fatalf("%s usernames = %v (set %v), want %v", stage, usernameStrings(vector), set, want) + } + return vector +} + +func assertScalarOnlyUsername(t *testing.T, stage string, peer tgUsernamePeer, want string) { + t.Helper() + peer.SetFlags() + if vector, set := peer.GetUsernames(); set || len(vector) != 0 { + t.Fatalf("%s usernames = %+v (set %v), want vector absent", stage, vector, set) + } + if scalar, set := peer.GetUsername(); !set || scalar != want { + t.Fatalf("%s scalar username = %q (set %v), want %q", stage, scalar, set, want) + } +} + +func assertUsernameUpdateEnvelope(t *testing.T, stage string, updates *tg.Updates, want []string) { + t.Helper() + assertValue := func(valueStage string, value *tg.Updates) { + t.Helper() + if value == nil || len(value.Updates) != 1 || len(value.Users) != 1 { + t.Fatalf("%s envelope = %+v, want one update and one user", valueStage, value) + } + update, ok := value.Updates[0].(*tg.UpdateUserName) + if !ok { + t.Fatalf("%s update = %T, want *tg.UpdateUserName", valueStage, value.Updates[0]) + } + if got := usernameStrings(update.Usernames); !reflect.DeepEqual(got, want) { + t.Fatalf("%s update usernames = %v, want %v", valueStage, got, want) + } + user, ok := value.Users[0].(*tg.User) + if !ok { + t.Fatalf("%s companion user = %T, want *tg.User", valueStage, value.Users[0]) + } + assertVectorOnlyUsernames(t, valueStage+" companion", user, want) + } + assertValue(stage, updates) + for _, profile := range []tlprofile.Profile{ + tlprofile.Profile225, + tlprofile.Profile226, + tlprofile.Profile227, + tlprofile.Profile228, + } { + profileStage := fmt.Sprintf("%s Layer %d", stage, profile) + var wire bin.Buffer + if err := tlprofile.EncodeObject(profile, updates, &wire); err != nil { + t.Fatalf("encode %s: %v", profileStage, err) + } + decoded, err := tlprofile.DecodeObject(profile, &bin.Buffer{Buf: wire.Copy()}, tlprofile.Limits{}) + if err != nil { + t.Fatalf("decode %s: %v", profileStage, err) + } + decodedUpdates, ok := decoded.(*tg.Updates) + if !ok { + t.Fatalf("decoded %s = %T, want *tg.Updates", profileStage, decoded) + } + assertValue(profileStage, decodedUpdates) + } +} + +func TestApplyUsernamesFromRegistryKeepsEditableOnlyPeersScalarOnly(t *testing.T) { + user := &tg.User{ID: 11} + user.SetUsername("editable_user") + channel := &tg.Channel{ID: 22} + channel.SetUsername("editable_channel") + applyUsernamesFromRegistry( + []tg.UserClass{user}, + []tg.ChatClass{channel}, + map[domain.Peer][]domain.Username{ + {Type: domain.PeerTypeUser, ID: user.ID}: { + {Username: "editable_user", Editable: true, Active: true}, + }, + {Type: domain.PeerTypeChannel, ID: channel.ID}: { + {Username: "editable_channel", Editable: true, Active: true}, + }, + }, + ) + assertScalarOnlyUsername(t, "editable-only user", user, "editable_user") + assertScalarOnlyUsername(t, "editable-only channel", channel, "editable_channel") +} + func TestUsersGetUsersProjectsCollectibleUsernamesInOneBatch(t *testing.T) { registry := newFakeUsernameRegistry() f := newUsernameProjectionFixture(t, registry) @@ -256,16 +352,7 @@ func TestUsersGetUsersProjectsCollectibleUsernamesInOneBatch(t *testing.T) { t.Fatalf("users = %d, want 2", len(out)) } self := out[0].(*tg.User) - if scalar, ok := self.GetUsername(); !ok || scalar != "nft4" { - t.Fatalf("self scalar username = %q (set %v), want primary collectible nft4", scalar, ok) - } - vector, ok := self.GetUsernames() - if !ok { - t.Fatalf("self usernames unset, want registry vector") - } - if got := usernameStrings(vector); len(got) != 2 || got[0] != "nft4" || got[1] != "owner_slot" { - t.Fatalf("self usernames = %v, want [nft4 owner_slot]", got) - } + vector := assertVectorOnlyUsernames(t, "self", self, []string{"nft4", "owner_slot"}) if vector[0].Editable || !vector[0].Active { t.Fatalf("primary collectible flags = %+v, want non-editable+active", vector[0]) } @@ -273,13 +360,7 @@ func TestUsersGetUsersProjectsCollectibleUsernamesInOneBatch(t *testing.T) { t.Fatalf("editable slot flags = %+v, want editable+active", vector[1]) } friend := out[1].(*tg.User) - if scalar, ok := friend.GetUsername(); !ok || scalar != "friend_slot" { - t.Fatalf("friend scalar username = %q (set %v), want friend_slot", scalar, ok) - } - friendVector, _ := friend.GetUsernames() - if got := usernameStrings(friendVector); len(got) != 2 || got[1] != "gem4" { - t.Fatalf("friend usernames = %v, want [friend_slot gem4]", got) - } + friendVector := assertVectorOnlyUsernames(t, "friend", friend, []string{"friend_slot", "gem4"}) if friendVector[1].Active { t.Fatalf("inactive collectible projected active: %+v", friendVector[1]) } @@ -289,7 +370,7 @@ func TestUsersGetUsersProjectsCollectibleUsernamesInOneBatch(t *testing.T) { } } -func TestResolveUsernamePreservesCompleteUsernamesThroughLayer228(t *testing.T) { +func TestResolveUsernamePreservesCompleteUsernamesThroughLayers225To228(t *testing.T) { registry := newFakeUsernameRegistry() f := newUsernameProjectionFixture(t, registry) registry.byPeer[domain.Peer{Type: domain.PeerTypeUser, ID: f.owner.ID}] = []domain.Username{ @@ -314,70 +395,62 @@ func TestResolveUsernamePreservesCompleteUsernamesThroughLayer228(t *testing.T) if !ok { t.Fatalf("%s resolved user = %T, want *tg.User", stage, value.Users[0]) } - if scalar, set := user.GetUsername(); !set || scalar != "owner_slot" { - t.Fatalf("%s scalar username = %q (set %v), want owner_slot", stage, scalar, set) - } - vector, set := user.GetUsernames() want := []string{"owner_slot", "owner_collectible_b", "owner_collectible_a"} - if !set || !reflect.DeepEqual(usernameStrings(vector), want) { - t.Fatalf("%s usernames = %v (set %v), want %v", stage, usernameStrings(vector), set, want) - } + assertVectorOnlyUsernames(t, stage, user, want) } assertResolvedUsernames("canonical", resolved) - var wire bin.Buffer - if err := tlprofile.EncodeObject(tlprofile.Profile228, resolved, &wire); err != nil { - t.Fatalf("encode Layer 228 resolved peer: %v", err) - } - decoded, err := tlprofile.DecodeObject( + for _, profile := range []tlprofile.Profile{ + tlprofile.Profile225, + tlprofile.Profile226, + tlprofile.Profile227, tlprofile.Profile228, - &bin.Buffer{Buf: wire.Copy()}, - tlprofile.Limits{}, - ) - if err != nil { - t.Fatalf("decode Layer 228 resolved peer: %v", err) - } - decodedResolved, ok := decoded.(*tg.ContactsResolvedPeer) - if !ok { - t.Fatalf("decoded Layer 228 object = %T, want *tg.ContactsResolvedPeer", decoded) - } - assertResolvedUsernames("Layer 228", decodedResolved) + } { + stage := fmt.Sprintf("Layer %d", profile) + var wire bin.Buffer + if err := tlprofile.EncodeObject(profile, resolved, &wire); err != nil { + t.Fatalf("encode %s resolved peer: %v", stage, err) + } + decoded, err := tlprofile.DecodeObject(profile, &bin.Buffer{Buf: wire.Copy()}, tlprofile.Limits{}) + if err != nil { + t.Fatalf("decode %s resolved peer: %v", stage, err) + } + decodedResolved, ok := decoded.(*tg.ContactsResolvedPeer) + if !ok { + t.Fatalf("decoded %s object = %T, want *tg.ContactsResolvedPeer", stage, decoded) + } + assertResolvedUsernames(stage, decodedResolved) - requestBody := encodeExactLayerRPC(t, tlprofile.Profile228, &tg.ContactsResolveUsernameRequest{ - Username: "@owner_slot", - }) - admitted, err := f.router.AdmitLayer(tlprofile.Profile228, &requestBody, tlprofile.Limits{}) - if err != nil { - t.Fatalf("admit Layer 228 contacts.resolveUsername: %v", err) + requestBody := encodeExactLayerRPC(t, profile, &tg.ContactsResolveUsernameRequest{Username: "@owner_slot"}) + admitted, err := f.router.AdmitLayer(profile, &requestBody, tlprofile.Limits{}) + if err != nil { + t.Fatalf("admit %s contacts.resolveUsername: %v", stage, err) + } + result, method, err := f.router.DispatchAdmitted( + WithUserID(context.Background(), f.friend.ID), + [8]byte{1}, + 1, + 1, + 1, + admitted, + ) + if err != nil || method != "contacts.resolveUsername" { + t.Fatalf("dispatch %s method=%q err=%v", stage, method, err) + } + var resultWire bin.Buffer + if err := result.Encode(&resultWire); err != nil { + t.Fatalf("encode %s method result: %v", stage, err) + } + decodedResult, err := tlprofile.DecodeObject(profile, &bin.Buffer{Buf: resultWire.Copy()}, tlprofile.Limits{}) + if err != nil { + t.Fatalf("decode %s method result: %v", stage, err) + } + methodResolved, ok := decodedResult.(*tg.ContactsResolvedPeer) + if !ok { + t.Fatalf("decoded %s method result = %T, want *tg.ContactsResolvedPeer", stage, decodedResult) + } + assertResolvedUsernames(stage+" method result", methodResolved) } - result, method, err := f.router.DispatchAdmitted( - WithUserID(context.Background(), f.friend.ID), - [8]byte{1}, - 1, - 1, - 1, - admitted, - ) - if err != nil || method != "contacts.resolveUsername" { - t.Fatalf("dispatch Layer 228 method=%q err=%v", method, err) - } - var resultWire bin.Buffer - if err := result.Encode(&resultWire); err != nil { - t.Fatalf("encode Layer 228 method result: %v", err) - } - decodedResult, err := tlprofile.DecodeObject( - tlprofile.Profile228, - &bin.Buffer{Buf: resultWire.Copy()}, - tlprofile.Limits{}, - ) - if err != nil { - t.Fatalf("decode Layer 228 method result: %v", err) - } - methodResolved, ok := decodedResult.(*tg.ContactsResolvedPeer) - if !ok { - t.Fatalf("decoded Layer 228 method result = %T, want *tg.ContactsResolvedPeer", decodedResult) - } - assertResolvedUsernames("Layer 228 method result", methodResolved) } func TestAuthLoginTokenSuccessProjectsCompleteSelfUsernames(t *testing.T) { @@ -406,29 +479,30 @@ func TestAuthLoginTokenSuccessProjectsCompleteSelfUsernames(t *testing.T) { t.Fatalf("authorization user = %T, want *tg.User", authorization.User) } want := []string{"owner_slot", "owner_collectible_b", "owner_collectible_a"} - vector, set := self.GetUsernames() - if !set || !reflect.DeepEqual(usernameStrings(vector), want) { - t.Fatalf("authorization usernames = %v (set %v), want %v", usernameStrings(vector), set, want) - } + assertVectorOnlyUsernames(t, "authorization", self, want) - var wire bin.Buffer - if err := tlprofile.EncodeObject(tlprofile.Profile228, success, &wire); err != nil { - t.Fatalf("encode Layer 228 login token success: %v", err) - } - decoded, err := tlprofile.DecodeObject(tlprofile.Profile228, &bin.Buffer{Buf: wire.Copy()}, tlprofile.Limits{}) - if err != nil { - t.Fatalf("decode Layer 228 login token success: %v", err) - } - decodedSuccess, ok := decoded.(*tg.AuthLoginTokenSuccess) - if !ok { - t.Fatalf("decoded login token success = %T", decoded) - } - decodedAuthorization := decodedSuccess.Authorization.(*tg.AuthAuthorization) - decodedSelf := decodedAuthorization.User.(*tg.User) - decodedVector, decodedSet := decodedSelf.GetUsernames() - if !decodedSet || !reflect.DeepEqual(usernameStrings(decodedVector), want) { - t.Fatalf("decoded authorization usernames = %v (set %v), want %v", - usernameStrings(decodedVector), decodedSet, want) + for _, profile := range []tlprofile.Profile{ + tlprofile.Profile225, + tlprofile.Profile226, + tlprofile.Profile227, + tlprofile.Profile228, + } { + stage := fmt.Sprintf("Layer %d authorization", profile) + var wire bin.Buffer + if err := tlprofile.EncodeObject(profile, success, &wire); err != nil { + t.Fatalf("encode %s: %v", stage, err) + } + decoded, err := tlprofile.DecodeObject(profile, &bin.Buffer{Buf: wire.Copy()}, tlprofile.Limits{}) + if err != nil { + t.Fatalf("decode %s: %v", stage, err) + } + decodedSuccess, ok := decoded.(*tg.AuthLoginTokenSuccess) + if !ok { + t.Fatalf("decoded %s = %T", stage, decoded) + } + decodedAuthorization := decodedSuccess.Authorization.(*tg.AuthAuthorization) + decodedSelf := decodedAuthorization.User.(*tg.User) + assertVectorOnlyUsernames(t, stage, decodedSelf, want) } if registry.peerCalls != 1 || registry.batchCalls != 0 { t.Fatalf("authorization username reads = peer:%d batch:%d, want 1/0", @@ -436,6 +510,122 @@ func TestAuthLoginTokenSuccessProjectsCompleteSelfUsernames(t *testing.T) { } } +func TestAccountProfileMutationResponsesKeepVectorOnlyUsernames(t *testing.T) { + registry := newFakeUsernameRegistry() + f := newUsernameProjectionFixture(t, registry) + sessions := &captureSessions{} + f.router.deps.Sessions = sessions + peer := domain.Peer{Type: domain.PeerTypeUser, ID: f.owner.ID} + registry.byPeer[peer] = []domain.Username{ + {Username: "owner_slot", Editable: true, Active: true, SortOrder: 0}, + {Username: "owner_collectible", Active: true, SortOrder: 1, CollectibleID: 21}, + } + ctx := WithUserID(context.Background(), f.owner.ID) + profile := &tg.AccountUpdateProfileRequest{} + profile.SetFirstName("Updated") + + updated, err := f.router.onAccountUpdateProfile(ctx, profile) + if err != nil { + t.Fatalf("account.updateProfile: %v", err) + } + self, ok := updated.(*tg.User) + if !ok { + t.Fatalf("account.updateProfile user = %T, want *tg.User", updated) + } + assertVectorOnlyUsernames(t, "account.updateProfile", self, []string{"owner_slot", "owner_collectible"}) + profileEnvelope := sessions.lastUserPush().(*tg.Updates) + pushed := profileEnvelope.Users[0].(*tg.User) + if pushed == self { + t.Fatal("account.updateProfile reused one mutable tg.User for RPC result and push") + } + assertUsernameUpdateEnvelope(t, "account.updateProfile push", profileEnvelope, []string{"owner_slot", "owner_collectible"}) + if registry.peerCalls != 1 || registry.batchCalls != 0 { + t.Fatalf("account.updateProfile username reads = peer:%d batch:%d, want 1/0", registry.peerCalls, registry.batchCalls) + } + + registry.byPeer[peer] = []domain.Username{ + {Username: "renamed_slot", Editable: true, Active: true, SortOrder: 0}, + {Username: "owner_collectible", Active: true, SortOrder: 1, CollectibleID: 21}, + } + updated, err = f.router.onAccountUpdateUsername(ctx, "renamed_slot") + if err != nil { + t.Fatalf("account.updateUsername: %v", err) + } + self, ok = updated.(*tg.User) + if !ok { + t.Fatalf("account.updateUsername user = %T, want *tg.User", updated) + } + assertVectorOnlyUsernames(t, "account.updateUsername", self, []string{"renamed_slot", "owner_collectible"}) + usernameEnvelope := sessions.lastUserPush().(*tg.Updates) + pushed = usernameEnvelope.Users[0].(*tg.User) + if pushed == self { + t.Fatal("account.updateUsername reused one mutable tg.User for RPC result and push") + } + assertUsernameUpdateEnvelope(t, "account.updateUsername push", usernameEnvelope, []string{"renamed_slot", "owner_collectible"}) + if registry.peerCalls != 2 || registry.batchCalls != 0 { + t.Fatalf("profile mutation username reads = peer:%d batch:%d, want 2/0", registry.peerCalls, registry.batchCalls) + } +} + +func TestPremiumStatusProjectionSkipsOfflineUsernameRead(t *testing.T) { + const userID = int64(1000000888) + registry := newFakeUsernameRegistry() + registry.byPeer[domain.Peer{Type: domain.PeerTypeUser, ID: userID}] = []domain.Username{ + {Username: "premium_slot", Editable: true, Active: true, SortOrder: 0}, + {Username: "premium_collectible", Active: true, SortOrder: 1, CollectibleID: 88}, + } + sessions := &captureSessions{} + r := New(Config{}, Deps{Sessions: sessions, Usernames: registry}, zaptest.NewLogger(t), clock.System) + u := domain.User{ID: userID, FirstName: "Premium", Username: "premium_slot"} + + r.pushPremiumStatusUpdate(context.Background(), u) + if registry.peerCalls != 0 || sessions.lastUserPush() != nil { + t.Fatalf("offline premium push = registry reads:%d message:%T, want no work", registry.peerCalls, sessions.lastUserPush()) + } + + sessions.onlineUserIDs = []int64{userID} + r.pushPremiumStatusUpdate(context.Background(), u) + if registry.peerCalls != 1 { + t.Fatalf("online premium username reads = %d, want 1", registry.peerCalls) + } + updates, ok := sessions.lastUserPush().(*tg.Updates) + if !ok || len(updates.Users) != 1 { + t.Fatalf("online premium push = %T %+v", sessions.lastUserPush(), sessions.lastUserPush()) + } + assertVectorOnlyUsernames(t, "premium status update", updates.Users[0].(*tg.User), []string{"premium_slot", "premium_collectible"}) +} + +func TestPremiumStatusProjectionBatchesOnlineUsers(t *testing.T) { + registry := newFakeUsernameRegistry() + users := []domain.User{ + {ID: 101, FirstName: "One", Username: "one_slot"}, + {ID: 102, FirstName: "Two", Username: "two_slot"}, + {ID: 103, FirstName: "Offline", Username: "offline_slot"}, + } + for _, u := range users { + registry.byPeer[domain.Peer{Type: domain.PeerTypeUser, ID: u.ID}] = []domain.Username{ + {Username: u.Username, Editable: true, Active: true, SortOrder: 0}, + {Username: u.Username + "_collectible", Active: true, SortOrder: 1, CollectibleID: u.ID}, + } + } + sessions := &captureSessions{onlineUserIDs: []int64{101, 102}} + r := New(Config{}, Deps{Sessions: sessions, Usernames: registry}, zaptest.NewLogger(t), clock.System) + + r.pushPremiumStatusUpdates(context.Background(), []domain.User{users[0], users[1], users[0], users[2]}) + + if registry.batchCalls != 1 || registry.peerCalls != 0 { + t.Fatalf("premium batch username reads = batch:%d peer:%d, want 1/0", registry.batchCalls, registry.peerCalls) + } + if got, want := sessions.pushedUserIDs(), []int64{101, 102}; !reflect.DeepEqual(got, want) { + t.Fatalf("premium batch pushed users = %v, want %v", got, want) + } + updates, ok := sessions.lastUserPush().(*tg.Updates) + if !ok || len(updates.Users) != 1 { + t.Fatalf("premium batch last push = %T %+v", sessions.lastUserPush(), sessions.lastUserPush()) + } + assertVectorOnlyUsernames(t, "premium batch last user", updates.Users[0].(*tg.User), []string{"two_slot", "two_slot_collectible"}) +} + func TestMessageEchoProjectsCompleteUsernamesInOneBatch(t *testing.T) { registry := newFakeUsernameRegistry() f := newUsernameProjectionFixture(t, registry) @@ -465,10 +655,7 @@ func TestMessageEchoProjectsCompleteUsernamesInOneBatch(t *testing.T) { 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]) - } + assertVectorOnlyUsernames(t, fmt.Sprintf("message echo user %d", user.ID), user, 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) @@ -516,10 +703,7 @@ func TestChannelMessageUpdatesProjectCompleteUsernames(t *testing.T) { 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) - } + assertVectorOnlyUsernames(t, "channel sender", user, []string{"channel_sender", "channel_collectible"}) 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) } @@ -561,10 +745,7 @@ func TestChannelDifferenceProjectsCompleteUsernames(t *testing.T) { 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) - } + assertVectorOnlyUsernames(t, "channel difference sender", user, []string{"difference_sender", "difference_collectible"}) 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) } @@ -580,14 +761,7 @@ func TestUsersGetUsersDegradesWithoutRegistry(t *testing.T) { } self := out[0].(*tg.User) // Without a collectible registry the official legacy shape is scalar-only. - self.SetFlags() - vector, ok := self.GetUsernames() - if ok || len(vector) != 0 { - t.Fatalf("usernames = %+v (set %v), want vector absent", vector, ok) - } - if scalar, ok := self.GetUsername(); !ok || scalar != "owner_slot" { - t.Fatalf("scalar username = %q (set %v), want owner_slot", scalar, ok) - } + assertScalarOnlyUsername(t, "self without registry", self, "owner_slot") } func TestUsersGetUsersDegradesWhenRegistryFails(t *testing.T) { @@ -608,14 +782,7 @@ func TestUsersGetUsersDegradesWhenRegistryFails(t *testing.T) { } for i, want := range []string{"owner_slot", "friend_slot"} { user := out[i].(*tg.User) - user.SetFlags() - vector, ok := user.GetUsernames() - if ok || len(vector) != 0 { - t.Fatalf("users[%d] usernames = %+v, want vector absent", i, vector) - } - if scalar, ok := user.GetUsername(); !ok || scalar != want { - t.Fatalf("users[%d] scalar username = %q (set %v), want %q", i, scalar, ok, want) - } + assertScalarOnlyUsername(t, fmt.Sprintf("users[%d] with failed registry", i), user, want) } } @@ -963,15 +1130,27 @@ func TestChannelsGetChannelsProjectsCollectibleUsernames(t *testing.T) { if len(out) != 1 { t.Fatalf("chats = %d, want 1", len(out)) } - vector, ok := out[0].(*tg.Channel).GetUsernames() - if !ok { - t.Fatalf("channel usernames unset, want registry vector") - } - if got := usernameStrings(vector); len(got) != 2 || got[0] != "chan_nft" || got[1] != "chan_slot" { - t.Fatalf("channel usernames = %v, want [chan_nft chan_slot]", got) - } - if scalar, ok := out[0].(*tg.Channel).GetUsername(); !ok || scalar != "chan_nft" { - t.Fatalf("scalar channel username = %q (set %v), want primary collectible chan_nft", scalar, ok) + assertVectorOnlyUsernames(t, "channel", out[0].(*tg.Channel), []string{"chan_nft", "chan_slot"}) + for _, profile := range []tlprofile.Profile{ + tlprofile.Profile225, + tlprofile.Profile226, + tlprofile.Profile227, + tlprofile.Profile228, + } { + stage := fmt.Sprintf("Layer %d channel", profile) + var wire bin.Buffer + if err := tlprofile.EncodeObject(profile, chats, &wire); err != nil { + t.Fatalf("encode %s: %v", stage, err) + } + decoded, err := tlprofile.DecodeObject(profile, &bin.Buffer{Buf: wire.Copy()}, tlprofile.Limits{}) + if err != nil { + t.Fatalf("decode %s: %v", stage, err) + } + decodedChats, ok := decoded.(*tg.MessagesChats) + if !ok || len(decodedChats.Chats) != 1 { + t.Fatalf("decoded %s = %T %+v", stage, decoded, decoded) + } + assertVectorOnlyUsernames(t, stage, decodedChats.Chats[0].(*tg.Channel), []string{"chan_nft", "chan_slot"}) } } @@ -985,11 +1164,5 @@ func TestChannelsGetChannelsDegradesWithoutRegistry(t *testing.T) { if err != nil { t.Fatalf("get channels: %v", err) } - vector, ok := chats.(*tg.MessagesChats).Chats[0].(*tg.Channel).GetUsernames() - if ok || len(vector) != 0 { - t.Fatalf("legacy channel usernames = %+v (set %v), want vector absent", vector, ok) - } - if scalar, ok := chats.(*tg.MessagesChats).Chats[0].(*tg.Channel).GetUsername(); !ok || scalar != "chan_slot" { - t.Fatalf("legacy scalar username = %q (set %v), want chan_slot", scalar, ok) - } + assertScalarOnlyUsername(t, "channel without registry", chats.(*tg.MessagesChats).Chats[0].(*tg.Channel), "chan_slot") } diff --git a/internal/rpc/contacts.go b/internal/rpc/contacts.go index 630bf20a..6967c513 100644 --- a/internal/rpc/contacts.go +++ b/internal/rpc/contacts.go @@ -478,6 +478,7 @@ func (r *Router) onContactsGetBlocked(ctx context.Context, req *tg.ContactsGetBl }) users = append(users, r.tgUser(item.User)) } + r.applyUsernamesToPeerObjects(ctx, users, nil) if list.Count > len(blocked)+req.Offset { return &tg.ContactsBlockedSlice{Count: list.Count, Blocked: blocked, Chats: []tg.ChatClass{}, Users: users}, nil } @@ -835,6 +836,7 @@ func (r *Router) onContactsDeleteContacts(ctx context.Context, ids []tg.InputUse } } r.invalidateRPCProjectionForViewer(userID) + r.applyUsernamesToPeerObjects(ctx, users, nil) out := &tg.Updates{Updates: updates, Users: users, Date: int(r.clock.Now().Unix())} r.pushUserUpdatesIfNoReliableDispatch(ctx, userID, out) return out, nil @@ -874,7 +876,7 @@ func (r *Router) onContactsUpdateContactNote(ctx context.Context, req *tg.Contac // private note into the shared update log. r.pushContactNoteRefreshIfReliableDispatch(ctx, userID, peerUser) } else { - r.pushUserUpdates(ctx, userID, r.contactNoteRefreshUpdates(peerUser, int(r.clock.Now().Unix()), true)) + r.pushUserUpdates(ctx, userID, r.contactNoteRefreshUpdates(ctx, userID, peerUser, int(r.clock.Now().Unix()), true)) } return true, nil } @@ -1062,17 +1064,19 @@ func contactUserForUpdates(contact domain.Contact) domain.User { return peerUser } -func (r *Router) contactNoteRefreshUpdates(peerUser domain.User, date int, includeContactsReset bool) *tg.Updates { +func (r *Router) contactNoteRefreshUpdates(ctx context.Context, viewerUserID int64, peerUser domain.User, date int, includeContactsReset bool) *tg.Updates { updates := make([]tg.UpdateClass, 0, 2) if includeContactsReset { updates = append(updates, &tg.UpdateContactsReset{}) } updates = append(updates, &tg.UpdateUser{UserID: peerUser.ID}) - return &tg.Updates{ + out := &tg.Updates{ Updates: updates, Users: []tg.UserClass{r.tgUser(peerUser)}, Date: date, } + r.applyUsernamesToPeerObjects(ctx, out.Users, nil) + return out } // pushContactNoteRefreshIfReliableDispatch complements the durable @@ -1087,7 +1091,7 @@ func (r *Router) pushContactNoteRefreshIfReliableDispatch(ctx context.Context, u ctx, userID, "push contact note full-user refresh", - r.contactNoteRefreshUpdates(peerUser, int(r.clock.Now().Unix()), false), + r.contactNoteRefreshUpdates(ctx, userID, peerUser, int(r.clock.Now().Unix()), false), ) } diff --git a/internal/rpc/contacts_users_rpc_test.go b/internal/rpc/contacts_users_rpc_test.go index 0c198c7e..af179b2f 100644 --- a/internal/rpc/contacts_users_rpc_test.go +++ b/internal/rpc/contacts_users_rpc_test.go @@ -76,12 +76,9 @@ func TestContactsSearchFindsUsers(t *testing.T) { t.Fatalf("peer = %T %+v, want friend", box.Results[0], box.Results[0]) } user := box.Users[0].(*tg.User) - if scalar, ok := user.GetUsername(); !ok || scalar != "search_friend" { - t.Fatalf("search result scalar username = %q (set %v), want search_friend", scalar, ok) - } - vector, ok := user.GetUsernames() - if !ok || len(vector) != 2 || vector[1].Username != "nft4" || !vector[1].Active { - t.Fatalf("search result username vector = %+v (set %v), want active nft4 alias", vector, ok) + vector := assertVectorOnlyUsernames(t, "contacts.search result", user, []string{"search_friend", "nft4"}) + if !vector[1].Active { + t.Fatalf("search result username vector = %+v, want active nft4 alias", vector) } } diff --git a/internal/rpc/deps.go b/internal/rpc/deps.go index 8d827ef9..6cda9794 100644 --- a/internal/rpc/deps.go +++ b/internal/rpc/deps.go @@ -93,12 +93,10 @@ type ImmediateSessionPusher interface { PushToSessionForAuthKeyImmediate(ctx context.Context, rawAuthKeyID [8]byte, sessionID int64, t proto.MessageType, msg tg.UpdatesClass) error } -// SessionUpdatesStateProvider 暴露连接当前的 updates 激活状态（可选能力）。 -// started 同时覆盖 ready 与 pending FIFO 正在 flushing；两者都证明当前物理 session -// 已完成 membership 建立并开始激活，后续陈旧 post-response callback 不得重复查库或推送。 +// SessionUpdatesStateProvider 暴露连接当前的 updates 接收状态（可选能力）。 +// 用于按 RPC 置位 receivesUpdates 时的幂等短路；不实现时每次都走完整置位（幂等，仅多余开销）。 type SessionUpdatesStateProvider interface { ReceivesUpdatesForAuthKey(rawAuthKeyID [8]byte, sessionID int64) bool - UpdatesActivationStartedForAuthKey(rawAuthKeyID [8]byte, sessionID int64) bool } // ClientLayerBinder 把协商 TL layer 即时下推到连接（可选能力）。 diff --git a/internal/rpc/fragment.go b/internal/rpc/fragment.go index ca36b661..4352ee7a 100644 --- a/internal/rpc/fragment.go +++ b/internal/rpc/fragment.go @@ -346,16 +346,12 @@ func applyUsernamesFromRegistry(users []tg.UserClass, chats []tg.ChatClass, byPe continue } if vector := tgUsernamesFromRegistry(list, u.Username); len(vector) > 0 { - // Layer 228 defines username as the main active username, not as a - // legacy alternative to usernames. TDesktop seeds its local search - // index from this scalar before consuming the complete vector, so - // both fields must be projected together. - if primary := domain.ActiveUsername(list); primary != "" { - u.SetUsername(primary) - } else { - u.Flags.Unset(3) - u.Username = "" - } + // Official clients treat the legacy scalar and the complete vector as + // alternative representations. TDLib rejects a User carrying both and + // discards the complete username set, while TDesktop and DrKLO derive + // the primary username from the first active vector entry. + u.Flags.Unset(3) + u.Username = "" u.SetUsernames(vector) } } @@ -372,12 +368,8 @@ func applyUsernamesFromRegistry(users []tg.UserClass, chats []tg.ChatClass, byPe // when unset, which is exactly the fallback tgUsernamesFromRegistry wants. scalar, _ := ch.GetUsername() if vector := tgUsernamesFromRegistry(list, scalar); len(vector) > 0 { - if primary := domain.ActiveUsername(list); primary != "" { - ch.SetUsername(primary) - } else { - ch.Flags.Unset(6) - ch.Username = "" - } + ch.Flags.Unset(6) + ch.Username = "" ch.SetUsernames(vector) } } diff --git a/internal/rpc/messages_attach_menu.go b/internal/rpc/messages_attach_menu.go index 7423e84e..bb23b34a 100644 --- a/internal/rpc/messages_attach_menu.go +++ b/internal/rpc/messages_attach_menu.go @@ -228,6 +228,7 @@ func (r *Router) attachMenuUsers(ctx context.Context, viewerID int64, ids []int6 } out = append(out, tgUser) } + r.applyPeerReadModels(ctx, viewerID, out, nil) return out } diff --git a/internal/rpc/messages_bot_longtail.go b/internal/rpc/messages_bot_longtail.go index 513122a8..1bf63426 100644 --- a/internal/rpc/messages_bot_longtail.go +++ b/internal/rpc/messages_bot_longtail.go @@ -455,6 +455,7 @@ func (r *Router) onMessagesGetPreparedInlineMessage(ctx context.Context, req *tg out.Users = append(out.Users, r.tgUser(u)) } } + r.applyPeerReadModels(ctx, userID, out.Users, nil) return out, nil } diff --git a/internal/rpc/messages_compat.go b/internal/rpc/messages_compat.go index 03993183..a9444f76 100644 --- a/internal/rpc/messages_compat.go +++ b/internal/rpc/messages_compat.go @@ -197,6 +197,7 @@ func (r *Router) onMessagesGetCommonChats(ctx context.Context, req *tg.MessagesG for _, ch := range common.Channels { chats = append(chats, tgChannelChatMin(userID, ch)) } + r.applyUsernamesToPeerObjects(ctx, nil, chats) return &tg.MessagesChats{Chats: chats}, nil } diff --git a/internal/rpc/messages_dialogs.go b/internal/rpc/messages_dialogs.go index a4e3359e..1c05a9bc 100644 --- a/internal/rpc/messages_dialogs.go +++ b/internal/rpc/messages_dialogs.go @@ -50,10 +50,11 @@ func (r *Router) onMessagesSaveDraft(ctx context.Context, req *tg.MessagesSaveDr if draft.TopMessageID > 0 { update.SetTopMsgID(draft.TopMessageID) } + users, chats := r.peerObjectsForDraftUpdate(ctx, userID, peer) updates := &tg.Updates{ Updates: appendAuxPtsBookkeeping([]tg.UpdateClass{update}, recorded), - Users: r.usersForDraftUpdate(ctx, userID, peer), - Chats: r.chatsForDraftUpdate(ctx, userID, peer), + Users: users, + Chats: chats, Date: date, Seq: 0, } @@ -95,8 +96,7 @@ func (r *Router) onMessagesGetAllDrafts(ctx context.Context) (tg.UpdatesClass, e return nil, dialogDraftErr(err) } updates := make([]tg.UpdateClass, 0, len(drafts)) - users := r.usersForDrafts(ctx, userID, drafts) - chats := r.chatsForDrafts(ctx, userID, drafts) + users, chats := r.peerObjectsForDrafts(ctx, userID, drafts) for _, draft := range drafts { peer := tgPeer(draft.Peer) if peer == nil { @@ -141,10 +141,11 @@ func (r *Router) onMessagesClearAllDrafts(ctx context.Context) (bool, error) { } } r.bookkeepAuxPtsForCurrentSession(ctx, events...) + users, chats := r.peerObjectsForDrafts(ctx, userID, drafts) r.pushUserUpdatesIfNoReliableDispatch(ctx, userID, &tg.Updates{ Updates: updates, - Users: r.usersForDrafts(ctx, userID, drafts), - Chats: r.chatsForDrafts(ctx, userID, drafts), + Users: users, + Chats: chats, Date: date, Seq: 0, }) @@ -346,6 +347,20 @@ func (r *Router) chatsForDrafts(ctx context.Context, userID int64, drafts []doma return chats } +func (r *Router) peerObjectsForDraftUpdate(ctx context.Context, userID int64, peer domain.Peer) ([]tg.UserClass, []tg.ChatClass) { + users := r.usersForDraftUpdate(ctx, userID, peer) + chats := r.chatsForDraftUpdate(ctx, userID, peer) + r.applyUsernamesToPeerObjects(ctx, users, chats) + return users, chats +} + +func (r *Router) peerObjectsForDrafts(ctx context.Context, userID int64, drafts []domain.DialogDraft) ([]tg.UserClass, []tg.ChatClass) { + users := r.usersForDrafts(ctx, userID, drafts) + chats := r.chatsForDrafts(ctx, userID, drafts) + r.applyUsernamesToPeerObjects(ctx, users, chats) + return users, chats +} + func dialogDraftErr(err error) error { switch { case err == nil: @@ -382,10 +397,11 @@ func (r *Router) clearDraftAfterSend(ctx context.Context, userID int64, peer dom } recorded := r.recordDraftMessageEvent(ctx, userID, peer, topMessageID, &date) r.bookkeepAuxPtsForCurrentSession(ctx, recorded) + users, chats := r.peerObjectsForDraftUpdate(ctx, userID, peer) r.pushUserUpdatesIfNoReliableDispatch(ctx, userID, &tg.Updates{ Updates: appendAuxPtsBookkeeping([]tg.UpdateClass{update}, recorded), - Users: r.usersForDraftUpdate(ctx, userID, peer), - Chats: r.chatsForDraftUpdate(ctx, userID, peer), + Users: users, + Chats: chats, Date: date, Seq: 0, }) diff --git a/internal/rpc/messages_quick_replies.go b/internal/rpc/messages_quick_replies.go index 94e11a21..c837c1b7 100644 --- a/internal/rpc/messages_quick_replies.go +++ b/internal/rpc/messages_quick_replies.go @@ -314,7 +314,7 @@ func (r *Router) quickReplyUsers(ctx context.Context, userID int64) []tg.UserCla if err != nil || self.ID == 0 { return []tg.UserClass{} } - return []tg.UserClass{r.tgSelfUser(self)} + return []tg.UserClass{r.tgSelfUserWithUsernames(ctx, self)} } func (r *Router) quickReplyMutationUpdates(ctx context.Context, userID int64, mutation domain.QuickReplyMutation, prefix []tg.UpdateClass) (*tg.Updates, error) { diff --git a/internal/rpc/messages_saved_dialogs.go b/internal/rpc/messages_saved_dialogs.go index fb61bcf5..e405e001 100644 --- a/internal/rpc/messages_saved_dialogs.go +++ b/internal/rpc/messages_saved_dialogs.go @@ -333,7 +333,9 @@ func (r *Router) savedDialogsProjection(ctx context.Context, userID int64, list } if len(userIDs) > 0 { if found, err := r.deps.Users.ByIDs(ctx, userID, userIDs); err == nil { - users = append(users, r.tgUsers(found)...) + projected := tgUsersForViewer(userID, r.withUsersPresence(found)) + r.withBotProfileFlagsForUsers(ctx, projected) + users = append(users, projected...) } } } @@ -351,6 +353,8 @@ func (r *Router) savedDialogsProjection(ctx context.Context, userID int64, list } } } + r.applyUsernamesToPeerObjects(ctx, users, chats) + r.applyBotVerificationIconsToPeerObjects(ctx, users, chats) return users, chats } diff --git a/internal/rpc/outbox_dispatcher_test.go b/internal/rpc/outbox_dispatcher_test.go index ed831ce4..b6598af5 100644 --- a/internal/rpc/outbox_dispatcher_test.go +++ b/internal/rpc/outbox_dispatcher_test.go @@ -3,6 +3,7 @@ package rpc import ( "context" "errors" + "fmt" "reflect" "sync" "testing" @@ -357,13 +358,7 @@ func TestRouterBuildOutboxUpdatesProjectsUsernamesOncePerClaim(t *testing.T) { 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) - } + assertVectorOnlyUsernames(t, fmt.Sprintf("updates[%d]", i), user, []string{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) diff --git a/internal/rpc/phone_group_call.go b/internal/rpc/phone_group_call.go index 4edf74dd..7f99f612 100644 --- a/internal/rpc/phone_group_call.go +++ b/internal/rpc/phone_group_call.go @@ -147,20 +147,24 @@ func (r *Router) onPhoneGetGroupCallJoinAs(ctx context.Context, peer tg.InputPee Chats: []tg.ChatClass{}, Users: r.tgUsersForIDs(ctx, userID, []int64{userID}), } + finish := func() *tg.PhoneJoinAsPeers { + r.applyPeerReadModels(ctx, userID, out.Users, out.Chats) + return out + } if r.deps.Channels == nil { - return out, nil + return finish(), nil } dp, err := r.checkedDomainPeerFromInputPeer(ctx, userID, peer) if err != nil || dp.Type != domain.PeerTypeChannel || dp.ID == 0 { - return out, nil + return finish(), nil } view, err := r.deps.Channels.GetChannel(ctx, userID, dp.ID) if err != nil || view.Self.Status != domain.ChannelMemberActive || !channelMemberIsAdmin(view.Self) { - return out, nil + return finish(), nil } out.Peers = append(out.Peers, &tg.PeerChannel{ChannelID: view.Channel.ID}) out.Chats = append(out.Chats, tgChannel(userID, view.Channel, &view.Self)) - return out, nil + return finish(), nil } func (r *Router) conferenceCallCanAccess(ctx context.Context, callID, userID int64) (bool, error) { diff --git a/internal/rpc/photos.go b/internal/rpc/photos.go index 25f733c0..a12c3f77 100644 --- a/internal/rpc/photos.go +++ b/internal/rpc/photos.go @@ -486,8 +486,11 @@ func (r *Router) photosPhotoForSelf(ctx context.Context, userID int64, photo dom if kind == domain.ProfilePhotoKindProfile { applyProfilePhotoToUser(&self, photo) } - out.Users = append(out.Users, r.tgSelfUser(self)) - r.pushSelfPhotoUpdate(ctx, self) + projected := r.tgSelfUser(self) + pushed := r.tgSelfUser(self) + r.applyUsernamesToPeerObjects(ctx, []tg.UserClass{projected, pushed}, nil) + out.Users = append(out.Users, projected) + r.pushSelfPhotoUpdateWithUser(ctx, self, pushed) return out } @@ -501,6 +504,7 @@ func (r *Router) photosPhotoForUser(ctx context.Context, viewerUserID, targetUse return out } out.Users = append(out.Users, r.tgUser(user)) + r.applyUsernamesToPeerObjects(ctx, out.Users, nil) return out } @@ -560,7 +564,14 @@ func (r *Router) pushSelfPhotoUpdate(ctx context.Context, self domain.User) { if self.ID == 0 { return } - updates := selfPhotoUpdates(self, int(r.clock.Now().Unix()), r.tgSelfUser(self)) + r.pushSelfPhotoUpdateWithUser(ctx, self, r.tgSelfUserWithUsernames(ctx, self)) +} + +func (r *Router) pushSelfPhotoUpdateWithUser(ctx context.Context, self domain.User, projected *tg.User) { + if self.ID == 0 || projected == nil { + return + } + updates := selfPhotoUpdates(self, int(r.clock.Now().Unix()), projected) r.pushUserUpdates(ctx, self.ID, updates) r.pushSelfPhotoUpdateToCurrentSession(ctx, updates) } diff --git a/internal/rpc/premium_sweeper.go b/internal/rpc/premium_sweeper.go index 29d89f36..7f59a56d 100644 --- a/internal/rpc/premium_sweeper.go +++ b/internal/rpc/premium_sweeper.go @@ -49,9 +49,7 @@ func (r *Router) sweepExpiredPremium(ctx context.Context, batch int) { r.log.Warn("premium sweep failed", zap.Error(err)) return } - for _, u := range users { - r.pushPremiumStatusUpdate(ctx, u) - } + r.pushPremiumStatusUpdates(ctx, users) // 不满一批说明已扫完当前积压；满批则继续，避免长停机后积压跨多个周期。 if len(users) < batch { return @@ -153,14 +151,50 @@ func (r *Router) NotifyUserModerationFlagsChanged(ctx context.Context, u domain. // 授予、到期与 admin 认证变更共用：updateUser 触发客户端用随附的 self user // 对象刷新 premium/verified 等基础 flag（TDesktop processUser 按 flag 翻转）。 func (r *Router) pushPremiumStatusUpdate(ctx context.Context, u domain.User) { - if u.ID == 0 { + r.pushPremiumStatusUpdates(ctx, []domain.User{u}) +} + +// pushPremiumStatusUpdates projects one username-registry snapshot over the +// whole online subset. Premium expiry is swept in batches of up to 500 users; +// reading each peer separately here would turn one maintenance batch into a +// serial registry N+1 even though offline users need no immediate push. +func (r *Router) pushPremiumStatusUpdates(ctx context.Context, candidates []domain.User) { + if len(candidates) == 0 || r.deps.Sessions == nil { + return + } + online, hasOnlineIndex := r.deps.Sessions.(OnlineUserProvider) + users := make([]domain.User, 0, len(candidates)) + peers := make([]domain.Peer, 0, len(candidates)) + seen := make(map[int64]struct{}, len(candidates)) + for _, u := range candidates { + if u.ID == 0 { + continue + } + if _, ok := seen[u.ID]; ok { + continue + } + seen[u.ID] = struct{}{} + if hasOnlineIndex && !online.IsUserOnline(u.ID) { + continue + } + users = append(users, u) + peers = append(peers, domain.Peer{Type: domain.PeerTypeUser, ID: u.ID}) + } + if len(users) == 0 { return } pushCtx, cancel := context.WithTimeout(ctx, 10*time.Second) defer cancel() - r.pushUserUpdates(pushCtx, u.ID, &tg.Updates{ - Updates: []tg.UpdateClass{&tg.UpdateUser{UserID: u.ID}}, - Users: []tg.UserClass{r.tgSelfUserWithUsernames(pushCtx, u)}, - Date: int(r.clock.Now().Unix()), - }) + usernames := r.usernameRegistryMap(pushCtx, peers) + date := int(r.clock.Now().Unix()) + for _, u := range users { + projected := r.tgSelfUser(u) + projectedUsers := []tg.UserClass{projected} + applyUsernamesFromRegistry(projectedUsers, nil, usernames) + r.pushUserUpdates(pushCtx, u.ID, &tg.Updates{ + Updates: []tg.UpdateClass{&tg.UpdateUser{UserID: u.ID}}, + Users: projectedUsers, + Date: date, + }) + } } diff --git a/internal/rpc/receives_updates_mark_test.go b/internal/rpc/receives_updates_mark_test.go index e8785d7d..6d1bb739 100644 --- a/internal/rpc/receives_updates_mark_test.go +++ b/internal/rpc/receives_updates_mark_test.go @@ -5,14 +5,11 @@ import ( "errors" "reflect" "sync" - "sync/atomic" "testing" - "time" "github.com/iamxvbaba/td/bin" "github.com/iamxvbaba/td/clock" "github.com/iamxvbaba/td/tg" - "github.com/iamxvbaba/td/tlprofile" "go.uber.org/zap/zaptest" appchannels "telesrv/internal/app/channels" @@ -65,262 +62,8 @@ func TestDispatchMarksSessionReceivesUpdates(t *testing.T) { if sessions.sessionID != 42 { t.Fatalf("marked session_id = %d, want 42", sessions.sessionID) } -} - -type updatesStateCaptureSessions struct { - *captureSessions -} - -type selfCountingUsersService struct { - staticUsersService - selfCalls atomic.Int32 -} - -func (s *selfCountingUsersService) Self(ctx context.Context, userID int64) (domain.User, error) { - s.selfCalls.Add(1) - return s.staticUsersService.Self(ctx, userID) -} - -func (s *updatesStateCaptureSessions) ReceivesUpdatesForAuthKey([8]byte, int64) bool { - return s.snapshot().receives -} - -func (s *updatesStateCaptureSessions) UpdatesActivationStartedForAuthKey([8]byte, int64) bool { - return s.snapshot().receives -} - -func TestDispatchPushesCompleteSelfProfileOnceWhenSessionBecomesReady(t *testing.T) { - const ( - userID = int64(1000000311) - sessionID = int64(311) - ) - rawAuthKeyID := [8]byte{31} - self := domain.User{ - ID: userID, - AccessHash: 3111, - FirstName: "Alice", - Username: "Alice", - } - peer := domain.Peer{Type: domain.PeerTypeUser, ID: userID} - registry := newFakeUsernameRegistry() - registry.byPeer[peer] = []domain.Username{ - {Username: "Alice", Active: true, Editable: true, SortOrder: 0}, - {Username: "aliceCollect0728b", Active: true, SortOrder: 1, CollectibleID: 2}, - {Username: "aliceCollect0728a", Active: true, SortOrder: 2, CollectibleID: 1}, - } - sessions := &updatesStateCaptureSessions{captureSessions: &captureSessions{}} - users := &selfCountingUsersService{staticUsersService: staticUsersService{user: self}} - r := New(Config{DC: 2, IP: "127.0.0.1", Port: 2398}, Deps{ - Sessions: sessions, - Users: users, - Usernames: registry, - }, zaptest.NewLogger(t), clock.System) - - dispatch := func() context.Context { - t.Helper() - var in bin.Buffer - if err := (&tg.HelpGetConfigRequest{}).Encode(&in); err != nil { - t.Fatalf("encode help.getConfig: %v", err) - } - ctx := postresponse.WithCallbacks(WithUserID(context.Background(), userID)) - if _, err := r.Dispatch(ctx, rawAuthKeyID, sessionID, &in); err != nil { - t.Fatalf("dispatch help.getConfig: %v", err) - } - return ctx - } - - ctx := dispatch() - staleCtx := dispatch() // staged before the first delivery makes the session ready - if got := sessions.snapshot(); got.receives || got.sessionPushCalls != 0 { - t.Fatalf("pre-delivery readiness = receives:%v pushes:%d, want false/0", got.receives, got.sessionPushCalls) - } - postresponse.Run(ctx) - postresponse.Run(staleCtx) - - got := sessions.snapshot() - if !got.receives || got.receivesCalls != 1 || got.sessionPushCalls != 1 { - t.Fatalf("post-delivery readiness = receives:%v ready_calls:%d pushes:%d, want true/1/1", - got.receives, got.receivesCalls, got.sessionPushCalls) - } - updates, ok := got.message.(*tg.Updates) - if !ok || len(updates.Updates) != 1 || len(updates.Users) != 1 { - t.Fatalf("self refresh = %T %+v, want one updateUserName and one user", got.message, got.message) - } - nameRefresh, ok := updates.Updates[0].(*tg.UpdateUserName) - if !ok || nameRefresh.UserID != userID || nameRefresh.FirstName != "Alice" || nameRefresh.LastName != "" { - t.Fatalf("self name refresh = %T %+v, want updateUserName(%d, Alice)", updates.Updates[0], updates.Updates[0], userID) - } - wantUsernames := []string{"Alice", "aliceCollect0728b", "aliceCollect0728a"} - if !reflect.DeepEqual(usernameStrings(nameRefresh.Usernames), wantUsernames) { - t.Fatalf("self updateUserName usernames = %v, want %v", usernameStrings(nameRefresh.Usernames), wantUsernames) - } - projected, ok := updates.Users[0].(*tg.User) - if !ok { - t.Fatalf("self refresh user = %T, want *tg.User", updates.Users[0]) - } - vector, set := projected.GetUsernames() - if !set || !reflect.DeepEqual(usernameStrings(vector), wantUsernames) { - t.Fatalf("self refresh usernames = %v (set %v), want %v", usernameStrings(vector), set, wantUsernames) - } - if scalar, set := projected.GetUsername(); !set || scalar != "Alice" { - t.Fatalf("self refresh scalar username = %q (set %v), want Alice", scalar, set) - } - if updates.Seq != 0 { - t.Fatalf("self refresh seq = %d, want 0", updates.Seq) - } - - for _, profile := range []tlprofile.Profile{ - tlprofile.Profile225, - tlprofile.Profile226, - tlprofile.Profile227, - tlprofile.Profile228, - } { - var wire bin.Buffer - if err := tlprofile.EncodeObject(profile, updates, &wire); err != nil { - t.Fatalf("encode Layer %d self refresh: %v", profile, err) - } - decoded, err := tlprofile.DecodeObject(profile, &bin.Buffer{Buf: wire.Copy()}, tlprofile.Limits{}) - if err != nil { - t.Fatalf("decode Layer %d self refresh: %v", profile, err) - } - decodedUpdates, ok := decoded.(*tg.Updates) - if !ok || len(decodedUpdates.Updates) != 1 || len(decodedUpdates.Users) != 1 { - t.Fatalf("decoded Layer %d self refresh = %T %+v", profile, decoded, decoded) - } - decodedNameRefresh, ok := decodedUpdates.Updates[0].(*tg.UpdateUserName) - if !ok || !reflect.DeepEqual(usernameStrings(decodedNameRefresh.Usernames), wantUsernames) { - t.Fatalf("decoded Layer %d updateUserName = %T usernames=%v, want %v", - profile, decodedUpdates.Updates[0], usernameStrings(decodedNameRefresh.Usernames), wantUsernames) - } - decodedUser := decodedUpdates.Users[0].(*tg.User) - decodedVector, decodedSet := decodedUser.GetUsernames() - if !decodedSet || !reflect.DeepEqual(usernameStrings(decodedVector), wantUsernames) { - t.Fatalf("decoded Layer %d usernames = %v (set %v), want %v", - profile, usernameStrings(decodedVector), decodedSet, wantUsernames) - } - } - - // A session that is already fully ready must not receive the bootstrap again - // on every ordinary RPC. - postresponse.Run(dispatch()) - got = sessions.snapshot() - if got.receivesCalls != 1 || got.sessionPushCalls != 1 || users.selfCalls.Load() != 1 || registry.peerCalls != 1 { - t.Fatalf("repeat dispatch effects = ready_calls:%d pushes:%d self_reads:%d registry_reads:%d, want 1/1/1/1", - got.receivesCalls, got.sessionPushCalls, users.selfCalls.Load(), registry.peerCalls) - } -} - -type blockingActivationSessions struct { - *updatesStateCaptureSessions - setEntered chan struct{} - releaseSet chan struct{} - blockOnce sync.Once -} - -func (s *blockingActivationSessions) SetReceivesUpdatesForAuthKey(rawAuthKeyID [8]byte, sessionID int64, receives bool) { - s.blockOnce.Do(func() { - close(s.setEntered) - <-s.releaseSet - }) - s.captureSessions.SetReceivesUpdatesForAuthKey(rawAuthKeyID, sessionID, receives) -} - -func TestConcurrentDeliveredRPCsClaimSessionActivationOnce(t *testing.T) { - const ( - userID = int64(1000000313) - sessionID = int64(313) - ) - rawAuthKeyID := [8]byte{33} - peer := domain.Peer{Type: domain.PeerTypeUser, ID: userID} - registry := newFakeUsernameRegistry() - registry.byPeer[peer] = []domain.Username{ - {Username: "Alice", Active: true, Editable: true, SortOrder: 0}, - {Username: "aliceCollect0728b", Active: true, SortOrder: 1, CollectibleID: 2}, - } - sessions := &blockingActivationSessions{ - updatesStateCaptureSessions: &updatesStateCaptureSessions{captureSessions: &captureSessions{}}, - setEntered: make(chan struct{}), - releaseSet: make(chan struct{}), - } - users := &selfCountingUsersService{staticUsersService: staticUsersService{user: domain.User{ - ID: userID, FirstName: "Alice", Username: "Alice", - }}} - r := New(Config{DC: 2, IP: "127.0.0.1", Port: 2398}, Deps{ - Sessions: sessions, - Users: users, - Usernames: registry, - }, zaptest.NewLogger(t), clock.System) - - dispatch := func() context.Context { - t.Helper() - var in bin.Buffer - if err := (&tg.HelpGetConfigRequest{}).Encode(&in); err != nil { - t.Fatalf("encode help.getConfig: %v", err) - } - ctx := postresponse.WithCallbacks(WithUserID(context.Background(), userID)) - if _, err := r.Dispatch(ctx, rawAuthKeyID, sessionID, &in); err != nil { - t.Fatalf("dispatch help.getConfig: %v", err) - } - return ctx - } - first, second := dispatch(), dispatch() - firstDone := make(chan struct{}) - go func() { - postresponse.Run(first) - close(firstDone) - }() - <-sessions.setEntered - - secondDone := make(chan struct{}) - go func() { - postresponse.Run(second) - close(secondDone) - }() - select { - case <-secondDone: - close(sessions.releaseSet) - <-firstDone - t.Fatal("second activation callback overtook the in-flight activation") - case <-time.After(50 * time.Millisecond): - // The waiter shares the owner's activation result, preserving the later - // bootstrap phase ordering without repeating membership work. - } - close(sessions.releaseSet) - <-firstDone - <-secondDone - - got := sessions.snapshot() - if !got.receives || got.receivesCalls != 1 || got.sessionPushCalls != 1 || users.selfCalls.Load() != 1 || registry.peerCalls != 1 { - t.Fatalf("concurrent activation effects = receives:%v ready_calls:%d pushes:%d self_reads:%d registry_reads:%d, want true/1/1/1/1", - got.receives, got.receivesCalls, got.sessionPushCalls, users.selfCalls.Load(), registry.peerCalls) - } -} - -func TestDispatchSuppressesSelfProfileWhenUsernameRegistryFails(t *testing.T) { - const userID = int64(1000000312) - registry := newFakeUsernameRegistry() - registry.err = errors.New("registry unavailable") - sessions := &updatesStateCaptureSessions{captureSessions: &captureSessions{}} - r := New(Config{DC: 2, IP: "127.0.0.1", Port: 2398}, Deps{ - Sessions: sessions, - Users: staticUsersService{user: domain.User{ - ID: userID, FirstName: "Alice", Username: "Alice", - }}, - Usernames: registry, - }, zaptest.NewLogger(t), clock.System) - - var in bin.Buffer - if err := (&tg.HelpGetConfigRequest{}).Encode(&in); err != nil { - t.Fatalf("encode help.getConfig: %v", err) - } - ctx := postresponse.WithCallbacks(WithUserID(context.Background(), userID)) - if _, err := r.Dispatch(ctx, [8]byte{32}, 312, &in); err != nil { - t.Fatalf("dispatch help.getConfig: %v", err) - } - postresponse.Run(ctx) - got := sessions.snapshot() - if !got.receives || got.sessionPushCalls != 0 { - t.Fatalf("registry failure effects = receives:%v pushes:%d, want true/0", got.receives, got.sessionPushCalls) + if sessions.message != nil { + t.Fatalf("session readiness emitted unsolicited update %T; readiness must only open delivery", sessions.message) } } diff --git a/internal/rpc/router.go b/internal/rpc/router.go index 5ff3f843..badcc97c 100644 --- a/internal/rpc/router.go +++ b/internal/rpc/router.go @@ -149,13 +149,6 @@ type Router struct { // online 过滤前），按 userID 短 TTL；零值 sync.Map 即可用，无需构造器初始化。候选集变动 // 很慢（加好友/开新私聊），短 TTL 内复用避免 updateStatus 每次重跑 ~25-30 条 hydration 查询。 presenceCandidateCache sync.Map // userID(int64) -> *presenceCandidateEntry - // updatesReadySF serializes the delivery-gated activation of one physical - // raw auth-key + session. Cold-start RPCs are concurrent and can all be staged - // before the first rpc_result is delivered; only the claim owner may rebuild - // memberships and emit the one-shot self cache-convergence update. Waiters - // share completion so their later bootstrap phase cannot overtake activation; - // SessionUpdatesStateProvider retains ready/flushing for stale callbacks. - updatesReadySF singleflight.Group // botStatus 永久缓存 userID->是否 bot。bot 标志按账号不可变（BotFather 注册即定，普通用户永不变 bot）， // 故可无 TTL 缓存。userIsBot 在 PFS 连接上被 announceSessionOnline 每 RPC 调用，不缓存则每次一发 diff --git a/internal/rpc/rpc_testkit_sessions_test.go b/internal/rpc/rpc_testkit_sessions_test.go index 450fe7fb..457e8ad7 100644 --- a/internal/rpc/rpc_testkit_sessions_test.go +++ b/internal/rpc/rpc_testkit_sessions_test.go @@ -20,7 +20,6 @@ type captureSessions struct { authKeyResolved bool receives bool receivesCalls int - sessionPushCalls int messageType proto.MessageType message bin.Encoder userMessage bin.Encoder // 最近一次 PushToUser* 的消息（与 message 区分：message 也被 PushToSession 覆盖） @@ -34,32 +33,30 @@ type captureSessions struct { } type captureSessionsSnapshot struct { - sessionID int64 - userID int64 - userResolved bool - authKeyID [8]byte - authKeyResolved bool - receives bool - receivesCalls int - sessionPushCalls int - messageType proto.MessageType - message bin.Encoder + sessionID int64 + userID int64 + userResolved bool + authKeyID [8]byte + authKeyResolved bool + receives bool + receivesCalls int + messageType proto.MessageType + message bin.Encoder } func (s *captureSessions) snapshot() captureSessionsSnapshot { s.mu.Lock() defer s.mu.Unlock() return captureSessionsSnapshot{ - sessionID: s.sessionID, - userID: s.userID, - userResolved: s.userResolved, - authKeyID: s.authKeyID, - authKeyResolved: s.authKeyResolved, - receives: s.receives, - receivesCalls: s.receivesCalls, - sessionPushCalls: s.sessionPushCalls, - messageType: s.messageType, - message: s.message, + sessionID: s.sessionID, + userID: s.userID, + userResolved: s.userResolved, + authKeyID: s.authKeyID, + authKeyResolved: s.authKeyResolved, + receives: s.receives, + receivesCalls: s.receivesCalls, + messageType: s.messageType, + message: s.message, } } @@ -159,7 +156,6 @@ func (s *captureSessions) PushToSessionForAuthKey(_ context.Context, rawAuthKeyI s.sessionID = sessionID s.messageType = t s.message = msg - s.sessionPushCalls++ return nil } diff --git a/internal/rpc/self_profile_refresh.go b/internal/rpc/self_profile_refresh.go deleted file mode 100644 index d1ca31c3..00000000 --- a/internal/rpc/self_profile_refresh.go +++ /dev/null @@ -1,89 +0,0 @@ -package rpc - -import ( - "context" - "fmt" - - "github.com/iamxvbaba/td/tg" - "go.uber.org/zap" - - "telesrv/internal/domain" -) - -// tgSelfUserWithUsernames is the narrow single-object projection used by -// authorization results and self-profile updates. Those constructors already -// carry a User, so it must include the complete username registry vector and -// never downgrade a client's collectible usernames to the legacy scalar. Story -// and bot-verification read models are unrelated and deliberately not queried. -func (r *Router) tgSelfUserWithUsernames(ctx context.Context, u domain.User) *tg.User { - self := r.tgSelfUser(u) - users := []tg.UserClass{self} - r.applyUsernamesToPeerObjects(ctx, users, nil) - return self -} - -// pushUpdatesReadySelfProfile repairs the current session's cached self user -// when it first becomes eligible for proactive updates. updateUserName is the -// authoritative basic-name/username cache write in TDLib and DrKLO. A companion -// username-complete User remains in updates.users so DrKLO's normal peer merge -// persists the vector, but no updateUser is sent because it only invalidates -// userFull (and makes DrKLO clear photo caches). -// -// This is an ephemeral cache-convergence update: it allocates no PTS, writes no -// durable update event and targets only the physical session that just became -// ready. A username-registry read failure suppresses the refresh instead of -// replacing a possibly richer client cache with the legacy scalar-only shape. -func (r *Router) pushUpdatesReadySelfProfile(ctx context.Context, userID int64) { - updates, err := r.updatesReadySelfProfile(ctx, userID) - if err != nil { - r.log.Warn("build updates-ready self profile", - zap.Int64("user_id", userID), - zap.Error(err)) - return - } - if updates == nil { - return - } - r.pushCurrentSessionMessage(ctx, "push updates-ready self profile", updates) -} - -func (r *Router) updatesReadySelfProfile(ctx context.Context, userID int64) (*tg.Updates, error) { - if userID == 0 || r.deps.Users == nil { - return nil, nil - } - u, err := r.deps.Users.Self(ctx, userID) - if err != nil { - return nil, fmt.Errorf("load self user: %w", err) - } - if u.ID != userID || u.Deleted { - return nil, fmt.Errorf("invalid self user: requested %d, got %d deleted=%v", userID, u.ID, u.Deleted) - } - - self := r.tgSelfUser(u) - users := []tg.UserClass{self} - if r.deps.Usernames != nil { - peer := domain.Peer{Type: domain.PeerTypeUser, ID: userID} - list, err := r.deps.Usernames.PeerUsernames(ctx, peer) - if err != nil { - return nil, fmt.Errorf("load self usernames: %w", err) - } - if len(list) != 0 { - applyUsernamesFromRegistry(users, nil, map[domain.Peer][]domain.Username{peer: list}) - } - } - usernames := tgUsernames(u.Username) - if vector, ok := self.GetUsernames(); ok && len(vector) != 0 { - usernames = vector - } - return &tg.Updates{ - Updates: []tg.UpdateClass{&tg.UpdateUserName{ - UserID: userID, - FirstName: u.FirstName, - LastName: u.LastName, - Usernames: usernames, - }}, - Users: users, - Date: int(r.clock.Now().Unix()), - Seq: 0, - }, nil -} diff --git a/internal/rpc/self_user_projection.go b/internal/rpc/self_user_projection.go new file mode 100644 index 00000000..fea9329c --- /dev/null +++ b/internal/rpc/self_user_projection.go @@ -0,0 +1,25 @@ +package rpc + +import ( + "context" + + "github.com/iamxvbaba/td/tg" + + "telesrv/internal/domain" +) + +// tgSelfUserWithUsernames is the narrow single-object projection used by +// authorization results and self-profile updates. Those constructors already +// carry a complete User, so it must include the username registry without +// querying unrelated story or bot-verification read models. +// +// When a collectible registry vector exists, applyUsernamesToPeerObjects clears +// the legacy scalar username. Official clients treat the scalar and vector as +// alternative representations; emitting both makes TDLib reject the username +// set as malformed. +func (r *Router) tgSelfUserWithUsernames(ctx context.Context, u domain.User) *tg.User { + self := r.tgSelfUser(u) + users := []tg.UserClass{self} + r.applyUsernamesToPeerObjects(ctx, users, nil) + return self +} diff --git a/internal/rpc/story_peer_projection.go b/internal/rpc/story_peer_projection.go index a06d7c11..701cb4ea 100644 --- a/internal/rpc/story_peer_projection.go +++ b/internal/rpc/story_peer_projection.go @@ -539,8 +539,10 @@ func appendUniqueTGChats(base []tg.ChatClass, extra ...tg.ChatClass) []tg.ChatCl // whole user/chat set, so a handler pays one call per read model per response // rather than one per peer. // -// It is the single hook every handler uses; adding a read model here reaches all -// of them at once. +// It is the shared hook for handlers that return complete peer envelopes. New +// builders still have to call it explicitly at their response boundary; nested +// single-peer fields and fan-out builders use the corresponding narrow/preload +// helpers instead. func (r *Router) applyPeerReadModels(ctx context.Context, viewerUserID int64, users []tg.UserClass, chats []tg.ChatClass) { r.applyStoryMaxIDsToPeerObjects(ctx, viewerUserID, users, chats) r.applyUsernamesToPeerObjects(ctx, users, chats) diff --git a/internal/rpc/story_peer_projection_rpc_test.go b/internal/rpc/story_peer_projection_rpc_test.go index 0998a595..b97f5565 100644 --- a/internal/rpc/story_peer_projection_rpc_test.go +++ b/internal/rpc/story_peer_projection_rpc_test.go @@ -2,7 +2,6 @@ package rpc import ( "context" - "reflect" "sync" "testing" "time" @@ -1300,10 +1299,7 @@ func TestBuildOutboxStoryUpdatesHydratesCompanionPeersWithStoriesMaxID(t *testin 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) - } + assertVectorOnlyUsernames(t, "story owner", projectedOwner, []string{"story_owner", "story_collectible"}) 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) } diff --git a/internal/rpc/telegram_login.go b/internal/rpc/telegram_login.go index 352a5b9b..7a87fe86 100644 --- a/internal/rpc/telegram_login.go +++ b/internal/rpc/telegram_login.go @@ -74,6 +74,7 @@ func (r *Router) telegramLoginRequestResult(ctx context.Context, viewerUserID in return nil, telegramLoginOAuthInvalidErr() } botTL := r.withBotProfileFlags(ctx, r.tgUser(bot)) + r.applyPeerReadModels(ctx, viewerUserID, []tg.UserClass{botTL}, nil) out := &tg.URLAuthResultRequest{ RequestWriteAccess: request.Requests(domain.TelegramLoginScopeBotAccess), RequestPhoneNumber: request.Requests(domain.TelegramLoginScopePhone), @@ -379,6 +380,7 @@ func (r *Router) onAccountGetWebAuthorizations(ctx context.Context) (*tg.Account } } } + r.applyPeerReadModels(ctx, userID, result.Users, nil) return result, nil } diff --git a/internal/rpc/updates.go b/internal/rpc/updates.go index b292ed50..eab8fd6a 100644 --- a/internal/rpc/updates.go +++ b/internal/rpc/updates.go @@ -210,39 +210,26 @@ func (r *Router) maybeMarkSessionReceivesUpdates(ctx context.Context) { if !ok { return } - if r.sessionUpdatesActivationStarted(ctx) { - return + if provider, ok := r.deps.Sessions.(SessionUpdatesStateProvider); ok { + rawAuthKeyID, okRaw := RawAuthKeyIDFrom(ctx) + sessionID, okSess := SessionIDFrom(ctx) + if okRaw && okSess && provider.ReceivesUpdatesForAuthKey(rawAuthKeyID, sessionID) { + return + } } r.stageSessionUpdatesReadyAfterDelivery(ctx, userID) } -func (r *Router) sessionUpdatesActivationStarted(ctx context.Context) bool { - provider, ok := r.deps.Sessions.(SessionUpdatesStateProvider) - if !ok { - return false - } - rawAuthKeyID, okRaw := RawAuthKeyIDFrom(ctx) - sessionID, okSession := SessionIDFrom(ctx) - return okRaw && okSession && provider.UpdatesActivationStartedForAuthKey(rawAuthKeyID, sessionID) -} - -func (r *Router) markSessionReceivesUpdatesNow(ctx context.Context, userID int64) bool { +func (r *Router) markSessionReceivesUpdatesNow(ctx context.Context, userID int64) { if r.deps.Sessions == nil { - return false + return } r.syncSessionChannelMemberships(ctx, userID) sessionID, ok := SessionIDFrom(ctx) if !ok { - return false + return } r.deps.Sessions.SetReceivesUpdatesForAuthKey(rawAuthKeyIDForOrigin(ctx), sessionID, true) - if _, ok := r.deps.Sessions.(SessionUpdatesStateProvider); !ok { - // Alternate lightweight binders cannot expose flushing/membership state. - // Preserve their historical behavior; the Router claim still collapses - // callbacks which overlap this activation attempt. - return true - } - return r.sessionUpdatesActivationStarted(ctx) } func ptr[T any](v T) *T { return &v } diff --git a/internal/rpc/updates_delivery.go b/internal/rpc/updates_delivery.go index d4932303..3458213a 100644 --- a/internal/rpc/updates_delivery.go +++ b/internal/rpc/updates_delivery.go @@ -2,7 +2,6 @@ package rpc import ( "context" - "strconv" "time" "go.uber.org/zap" @@ -203,8 +202,7 @@ func (r *Router) registerUpdatesDeliveryPlan(ctx context.Context, plan *updatesD // 1. commit the exact account cursor carried by the delivered result; // 2. the just-delivered difference may retire its projected secret-chat events; // 3. membership routing is rebuilt before SetReceivesUpdates starts FIFO flush; -// 4. one claim owner activates the session and emits its username convergence update; -// 5. bootstrap jobs are published last, so they queue behind older pending updates. +// 4. bootstrap jobs are published last, so they queue behind older pending updates. // // Each phase gets an independent timeout so one failed side effect cannot starve // the remaining delivery-safe transitions. @@ -237,53 +235,11 @@ func (r *Router) runUpdatesDeliveryPlan(plan updatesDeliveryPlan) { } } if plan.markSessionReady { - r.activateSessionUpdates(baseCtx, plan.readyUserID) + ctx, cancel := context.WithTimeout(baseCtx, updatesDeliveryPhaseTimeout) + r.markSessionReceivesUpdatesNow(ctx, plan.readyUserID) + cancel() } if plan.publishBootstrap { r.publishBootstrapAfterBaseline(baseCtx, plan.bootstrapUserID) } } - -func (r *Router) activateSessionUpdates(baseCtx context.Context, userID int64) { - rawAuthKeyID, okRaw := RawAuthKeyIDFrom(baseCtx) - sessionID, okSession := SessionIDFrom(baseCtx) - if userID == 0 { - return - } - if !okRaw || !okSession { - // Direct handler tests and lightweight embeddings may not carry the full - // physical identity. They cannot benefit from a cross-request claim, but - // must retain the historical delivery-gated activation behavior. - ctx, cancel := context.WithTimeout(baseCtx, updatesDeliveryPhaseTimeout) - started := r.markSessionReceivesUpdatesNow(ctx, userID) - cancel() - if started { - ctx, cancel = context.WithTimeout(baseCtx, updatesDeliveryPhaseTimeout) - r.pushUpdatesReadySelfProfile(ctx, userID) - cancel() - } - return - } - if r.sessionUpdatesActivationStarted(baseCtx) { - return - } - key := string(rawAuthKeyID[:]) + strconv.FormatInt(sessionID, 10) - _, _, _ = r.updatesReadySF.Do(key, func() (any, error) { - // A callback may have been staged before another callback completed the - // same activation. Re-check under singleflight before touching storage. - if r.sessionUpdatesActivationStarted(baseCtx) { - return nil, nil - } - ctx, cancel := context.WithTimeout(baseCtx, updatesDeliveryPhaseTimeout) - started := r.markSessionReceivesUpdatesNow(ctx, userID) - cancel() - if !started { - return nil, nil - } - - ctx, cancel = context.WithTimeout(baseCtx, updatesDeliveryPhaseTimeout) - r.pushUpdatesReadySelfProfile(ctx, userID) - cancel() - return nil, nil - }) -} diff --git a/internal/rpc/username_notify_test.go b/internal/rpc/username_notify_test.go index 0fdb5919..239fd8c6 100644 --- a/internal/rpc/username_notify_test.go +++ b/internal/rpc/username_notify_test.go @@ -42,13 +42,7 @@ func TestNotifyPeerUsernamesChangedUserPushesPreloadedVector(t *testing.T) { } updates := sessions.lastUserPush().(*tg.Updates) user := updates.Users[0].(*tg.User) - if scalar, ok := user.GetUsername(); !ok || scalar != "nft4" { - t.Fatalf("pushed scalar username = %q (set %v), want primary collectible nft4", scalar, ok) - } - vector, ok := user.GetUsernames() - if !ok || len(vector) != 2 || vector[0].Username != "nft4" { - t.Fatalf("pushed username vector = %+v (set %v)", vector, ok) - } + assertVectorOnlyUsernames(t, "pushed user", user, []string{"nft4", "owner_slot"}) } func TestNotifyPeerUsernamesChangedChannelPushesPreloadedVector(t *testing.T) { @@ -84,11 +78,5 @@ func TestNotifyPeerUsernamesChangedChannelPushesPreloadedVector(t *testing.T) { } updates := sessions.lastUserPush().(*tg.Updates) channel := updates.Chats[0].(*tg.Channel) - if scalar, ok := channel.GetUsername(); !ok || scalar != "collectible" { - t.Fatalf("pushed scalar username = %q (set %v), want primary collectible", scalar, ok) - } - vector, ok := channel.GetUsernames() - if !ok || len(vector) != 2 || vector[0].Username != "collectible" { - t.Fatalf("pushed username vector = %+v (set %v)", vector, ok) - } + assertVectorOnlyUsernames(t, "pushed channel", channel, []string{"collectible", "channel_slot"}) }