diff --git a/internal/rpc/account.go b/internal/rpc/account.go index 837cb481..49a9407f 100644 --- a/internal/rpc/account.go +++ b/internal/rpc/account.go @@ -1962,7 +1962,7 @@ func (r *Router) pushSelfUserChangedUpdate(ctx context.Context, u domain.User) { } 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.tgSelfUserWithReadModels(ctx, u)}, Date: int(r.clock.Now().Unix()), }) } diff --git a/internal/rpc/auth.go b/internal/rpc/auth.go index 31f827ce..5fa52603 100644 --- a/internal/rpc/auth.go +++ b/internal/rpc/auth.go @@ -319,7 +319,7 @@ func (r *Router) authLoginTokenSuccess(ctx context.Context, a domain.Authorizati return nil, internalErr() } return &tg.AuthLoginTokenSuccess{ - Authorization: &tg.AuthAuthorization{User: r.tgSelfUser(u)}, + Authorization: &tg.AuthAuthorization{User: r.tgSelfUserWithReadModels(ctx, u)}, }, nil } @@ -528,7 +528,7 @@ func (r *Router) finishAuthSignIn(ctx context.Context, u domain.User, needSignUp } r.bindSessionUser(ctx, u.ID) r.pushSignInServiceNotificationToOthers(ctx, u) - return &tg.AuthAuthorization{User: r.tgSelfUser(u)}, nil + return &tg.AuthAuthorization{User: r.tgSelfUserWithReadModels(ctx, u)}, nil } func (r *Router) onAuthResendCode(ctx context.Context, req *tg.AuthResendCodeRequest) (tg.AuthSentCodeClass, error) { @@ -637,7 +637,7 @@ func (r *Router) onAuthCheckPassword(ctx context.Context, password tg.InputCheck if err != nil { return nil, internalErr() } - return &tg.AuthAuthorization{User: r.tgSelfUser(u)}, nil + return &tg.AuthAuthorization{User: r.tgSelfUserWithReadModels(ctx, u)}, nil } func (r *Router) onAuthRequestPasswordRecovery(ctx context.Context) (*tg.AuthPasswordRecovery, error) { @@ -678,7 +678,7 @@ func (r *Router) onAuthRecoverPassword(ctx context.Context, req *tg.AuthRecoverP if err != nil { return nil, internalErr() } - return &tg.AuthAuthorization{User: r.tgSelfUser(u)}, nil + return &tg.AuthAuthorization{User: r.tgSelfUserWithReadModels(ctx, u)}, nil } func (r *Router) onAuthCheckRecoveryPassword(ctx context.Context, code string) (bool, error) { @@ -809,7 +809,7 @@ func (r *Router) onAuthFinishPasskeyLogin(ctx context.Context, req *tg.AuthFinis r.setAuthUserCache(id, u.ID, true) } r.bindSessionUser(ctx, u.ID) - return &tg.AuthAuthorization{User: r.tgSelfUser(u)}, nil + return &tg.AuthAuthorization{User: r.tgSelfUserWithReadModels(ctx, u)}, nil } func emailVerificationCode(v tg.EmailVerificationClass) string { @@ -840,7 +840,7 @@ func (r *Router) onAuthImportBotAuthorization(ctx context.Context, req *tg.AuthI r.setAuthUserCache(id, u.ID, true) } r.bindSessionUser(ctx, u.ID) - return &tg.AuthAuthorization{User: r.tgSelfUser(u)}, nil + return &tg.AuthAuthorization{User: r.tgSelfUserWithReadModels(ctx, u)}, nil } // onAuthSignUp 处理 auth.signUp：创建用户并绑定授权。 @@ -854,7 +854,7 @@ func (r *Router) onAuthSignUp(ctx context.Context, req *tg.AuthSignUpRequest) (t } r.bindSessionUser(ctx, u.ID) r.enqueueLoginMessageBootstrap(ctx, loginMessage) - return &tg.AuthAuthorization{User: r.tgSelfUser(u)}, nil + return &tg.AuthAuthorization{User: r.tgSelfUserWithReadModels(ctx, u)}, nil } // onAuthLogOut 处理 auth.logOut：解绑当前 auth_key 的授权。 diff --git a/internal/rpc/collectible_usernames_rpc_test.go b/internal/rpc/collectible_usernames_rpc_test.go index 8cac6c4a..cfa4b004 100644 --- a/internal/rpc/collectible_usernames_rpc_test.go +++ b/internal/rpc/collectible_usernames_rpc_test.go @@ -12,6 +12,7 @@ import ( "github.com/iamxvbaba/td/clock" "github.com/iamxvbaba/td/tg" "github.com/iamxvbaba/td/tgerr" + "github.com/iamxvbaba/td/tlprofile" "go.uber.org/zap/zaptest" appchannels "telesrv/internal/app/channels" @@ -288,6 +289,149 @@ func TestUsersGetUsersProjectsCollectibleUsernamesInOneBatch(t *testing.T) { } } +func TestResolveUsernamePreservesCompleteUsernamesThroughLayer228(t *testing.T) { + registry := newFakeUsernameRegistry() + f := newUsernameProjectionFixture(t, registry) + registry.byPeer[domain.Peer{Type: domain.PeerTypeUser, ID: f.owner.ID}] = []domain.Username{ + {Username: "owner_slot", Editable: true, Active: true, SortOrder: 0}, + {Username: "owner_collectible_b", Active: true, SortOrder: 1, CollectibleID: 22}, + {Username: "owner_collectible_a", Active: true, SortOrder: 2, CollectibleID: 21}, + } + + resolved, err := f.router.onContactsResolveUsername( + WithUserID(context.Background(), f.friend.ID), + &tg.ContactsResolveUsernameRequest{Username: "@owner_slot"}, + ) + if err != nil { + t.Fatalf("resolve username: %v", err) + } + assertResolvedUsernames := func(stage string, value *tg.ContactsResolvedPeer) { + t.Helper() + if value == nil || len(value.Users) != 1 { + t.Fatalf("%s resolved users = %+v, want one user", stage, value) + } + user, ok := value.Users[0].(*tg.User) + 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) + } + } + 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( + 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) + + 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) + } + 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) { + registry := newFakeUsernameRegistry() + f := newUsernameProjectionFixture(t, registry) + registry.byPeer[domain.Peer{Type: domain.PeerTypeUser, ID: f.owner.ID}] = []domain.Username{ + {Username: "owner_slot", Editable: true, Active: true, SortOrder: 0}, + {Username: "owner_collectible_b", Active: true, SortOrder: 1, CollectibleID: 22}, + {Username: "owner_collectible_a", Active: true, SortOrder: 2, CollectibleID: 21}, + } + + result, err := f.router.authLoginTokenSuccess(context.Background(), domain.Authorization{UserID: f.owner.ID}) + if err != nil { + t.Fatalf("auth login token success: %v", err) + } + success, ok := result.(*tg.AuthLoginTokenSuccess) + if !ok { + t.Fatalf("auth login token result = %T, want *tg.AuthLoginTokenSuccess", result) + } + authorization, ok := success.Authorization.(*tg.AuthAuthorization) + if !ok { + t.Fatalf("authorization = %T, want *tg.AuthAuthorization", success.Authorization) + } + self, ok := authorization.User.(*tg.User) + if !ok { + 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) + } + + 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) + } +} + func TestMessageEchoProjectsCompleteUsernamesInOneBatch(t *testing.T) { registry := newFakeUsernameRegistry() f := newUsernameProjectionFixture(t, registry) diff --git a/internal/rpc/premium_sweeper.go b/internal/rpc/premium_sweeper.go index 69a92592..7c3ee3b8 100644 --- a/internal/rpc/premium_sweeper.go +++ b/internal/rpc/premium_sweeper.go @@ -160,7 +160,7 @@ func (r *Router) pushPremiumStatusUpdate(ctx context.Context, u domain.User) { defer cancel() r.pushUserUpdates(pushCtx, u.ID, &tg.Updates{ Updates: []tg.UpdateClass{&tg.UpdateUser{UserID: u.ID}}, - Users: []tg.UserClass{r.tgSelfUser(u)}, + Users: []tg.UserClass{r.tgSelfUserWithReadModels(pushCtx, u)}, Date: int(r.clock.Now().Unix()), }) } diff --git a/internal/rpc/receives_updates_mark_test.go b/internal/rpc/receives_updates_mark_test.go index a1232416..4e3cb550 100644 --- a/internal/rpc/receives_updates_mark_test.go +++ b/internal/rpc/receives_updates_mark_test.go @@ -10,6 +10,7 @@ import ( "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" @@ -64,6 +65,145 @@ func TestDispatchMarksSessionReceivesUpdates(t *testing.T) { } } +type updatesStateCaptureSessions struct { + *captureSessions +} + +func (s *updatesStateCaptureSessions) ReceivesUpdatesForAuthKey([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{}} + r := New(Config{DC: 2, IP: "127.0.0.1", Port: 2398}, Deps{ + Sessions: sessions, + Users: staticUsersService{user: self}, + 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() + 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) + + 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 update and one user", got.message, got.message) + } + refresh, ok := updates.Updates[0].(*tg.UpdateUser) + if !ok || refresh.UserID != userID { + t.Fatalf("self refresh update = %T %+v, want updateUser(%d)", updates.Updates[0], updates.Updates[0], userID) + } + 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() + wantUsernames := []string{"Alice", "aliceCollect0728b", "aliceCollect0728a"} + 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) + } + + var wire bin.Buffer + if err := tlprofile.EncodeObject(tlprofile.Profile228, updates, &wire); err != nil { + t.Fatalf("encode Layer 228 self refresh: %v", err) + } + decoded, err := tlprofile.DecodeObject(tlprofile.Profile228, &bin.Buffer{Buf: wire.Copy()}, tlprofile.Limits{}) + if err != nil { + t.Fatalf("decode Layer 228 self refresh: %v", err) + } + decodedUpdates, ok := decoded.(*tg.Updates) + if !ok || len(decodedUpdates.Users) != 1 { + t.Fatalf("decoded Layer 228 self refresh = %T %+v", decoded, decoded) + } + decodedUser := decodedUpdates.Users[0].(*tg.User) + decodedVector, decodedSet := decodedUser.GetUsernames() + if !decodedSet || !reflect.DeepEqual(usernameStrings(decodedVector), wantUsernames) { + t.Fatalf("decoded Layer 228 usernames = %v (set %v), want %v", + 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 || registry.peerCalls != 1 { + t.Fatalf("repeat dispatch effects = ready_calls:%d pushes:%d registry_reads:%d, want 1/1/1", + got.receivesCalls, got.sessionPushCalls, 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) + } +} + // TestDispatchSkipsReceivesUpdatesForInvokeWithoutUpdates 验证 invokeWithoutUpdates // 包装的请求（media/temp 连接）不会把该 session 标记为 updates 接收者。 func TestDispatchSkipsReceivesUpdatesForInvokeWithoutUpdates(t *testing.T) { diff --git a/internal/rpc/rpc_testkit_sessions_test.go b/internal/rpc/rpc_testkit_sessions_test.go index 457e8ad7..450fe7fb 100644 --- a/internal/rpc/rpc_testkit_sessions_test.go +++ b/internal/rpc/rpc_testkit_sessions_test.go @@ -20,6 +20,7 @@ type captureSessions struct { authKeyResolved bool receives bool receivesCalls int + sessionPushCalls int messageType proto.MessageType message bin.Encoder userMessage bin.Encoder // 最近一次 PushToUser* 的消息（与 message 区分：message 也被 PushToSession 覆盖） @@ -33,30 +34,32 @@ type captureSessions struct { } type captureSessionsSnapshot struct { - sessionID int64 - userID int64 - userResolved bool - authKeyID [8]byte - authKeyResolved bool - receives bool - receivesCalls int - messageType proto.MessageType - message bin.Encoder + sessionID int64 + userID int64 + userResolved bool + authKeyID [8]byte + authKeyResolved bool + receives bool + receivesCalls int + sessionPushCalls 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, - 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, + sessionPushCalls: s.sessionPushCalls, + messageType: s.messageType, + message: s.message, } } @@ -156,6 +159,7 @@ 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 new file mode 100644 index 00000000..2e811f08 --- /dev/null +++ b/internal/rpc/self_profile_refresh.go @@ -0,0 +1,84 @@ +package rpc + +import ( + "context" + "fmt" + + "github.com/iamxvbaba/td/tg" + "go.uber.org/zap" + + "telesrv/internal/domain" +) + +// tgSelfUserWithReadModels is the single-object response-boundary projection +// used by authorization results and self-profile updates. tgSelfUser itself is +// intentionally a pure domain -> TL conversion because many list paths call it +// in loops; the read-model pass belongs here, where it stays one query per +// response. +func (r *Router) tgSelfUserWithReadModels(ctx context.Context, u domain.User) *tg.User { + self := r.tgSelfUser(u) + users := []tg.UserClass{self} + r.applyPeerReadModels(ctx, u.ID, users, nil) + return self +} + +// pushUpdatesReadySelfProfile repairs the current session's cached self user +// when it first becomes eligible for proactive updates. Telegram's updateUser +// contract requires the complete User to be carried by the outer updates.users +// vector; TDLib applies that vector before invalidating userFull for updateUser. +// +// 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}) + } + } + // These read models are optional projections. Their existing response-boundary + // contract degrades independently; the username registry above is handled + // strictly because losing that vector is the cache corruption fixed here. + r.applyStoryMaxIDsToPeerObjects(ctx, userID, users, nil) + r.applyBotVerificationIconsToPeerObjects(ctx, users, nil) + + return &tg.Updates{ + Updates: []tg.UpdateClass{&tg.UpdateUser{UserID: userID}}, + Users: users, + Date: int(r.clock.Now().Unix()), + Seq: 0, + }, nil +} diff --git a/internal/rpc/updates_delivery.go b/internal/rpc/updates_delivery.go index 3458213a..c723327b 100644 --- a/internal/rpc/updates_delivery.go +++ b/internal/rpc/updates_delivery.go @@ -202,7 +202,8 @@ 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. bootstrap jobs are published last, so they queue behind older pending updates. +// 4. the current session receives one complete self profile after becoming ready; +// 5. 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. @@ -238,6 +239,10 @@ func (r *Router) runUpdatesDeliveryPlan(plan updatesDeliveryPlan) { ctx, cancel := context.WithTimeout(baseCtx, updatesDeliveryPhaseTimeout) r.markSessionReceivesUpdatesNow(ctx, plan.readyUserID) cancel() + + ctx, cancel = context.WithTimeout(baseCtx, updatesDeliveryPhaseTimeout) + r.pushUpdatesReadySelfProfile(ctx, plan.readyUserID) + cancel() } if plan.publishBootstrap { r.publishBootstrapAfterBaseline(baseCtx, plan.bootstrapUserID)