From 6cafa40c7be5f35c19caca2f2d63cce2c5d2dab0 Mon Sep 17 00:00:00 2001 From: iamxvbaba <28732408+iamxvbaba@users.noreply.github.com> Date: Fri, 24 Jul 2026 11:57:00 +0800 Subject: [PATCH] fix: sync moderation refresh cache bypass --- internal/app/channels/service.go | 15 +++++ internal/app/users/service.go | 21 +++++++ internal/app/users/service_test.go | 43 ++++++++++++++ internal/rpc/deps.go | 15 +++++ .../rpc/moderation_profile_update_test.go | 59 +++++++++++++++++++ internal/rpc/update_peer_refs.go | 34 +++++++++++ 6 files changed, 187 insertions(+) diff --git a/internal/app/channels/service.go b/internal/app/channels/service.go index 0a7c4bc6..fc42eb56 100644 --- a/internal/app/channels/service.go +++ b/internal/app/channels/service.go @@ -216,6 +216,21 @@ func (s *Service) GetChannels(ctx context.Context, userID int64, channelIDs []in return s.channels.GetChannels(ctx, userID, ids) } +// GetChannelsAuthoritative bypasses the app-level versioned read model for a +// durable channel_state refresh. PostgreSQL GetChannels is a bounded direct +// projection query, so the returned flag snapshot cannot be the pre-commit +// value that the event is intended to invalidate. +func (s *Service) GetChannelsAuthoritative(ctx context.Context, userID int64, channelIDs []int64) ([]domain.ChannelView, error) { + if s == nil || s.channels == nil || userID == 0 { + return nil, domain.ErrChannelInvalid + } + ids := uniqueNonZero(channelIDs) + if len(ids) == 0 { + return nil, nil + } + return s.channels.GetChannels(ctx, userID, ids) +} + // GetJoinableChannel returns a channel shell so RPC can verify access hash before join. func (s *Service) GetJoinableChannel(ctx context.Context, userID, channelID int64) (domain.Channel, error) { if s == nil || s.channels == nil || userID == 0 || channelID == 0 { diff --git a/internal/app/users/service.go b/internal/app/users/service.go index 02322cdf..2b050fa1 100644 --- a/internal/app/users/service.go +++ b/internal/app/users/service.go @@ -176,6 +176,27 @@ func (s *Service) ByIDs(ctx context.Context, currentUserID int64, userIDs []int6 return s.projectUsers(ctx, currentUserID, users) } +// ByIDsAuthoritative reloads an explicit profile-refresh target from the +// durable store, then replaces the shared base cache before applying the +// viewer projection. It is intentionally reserved for durable update +// delivery; ordinary reads continue to use ByIDs. +func (s *Service) ByIDsAuthoritative(ctx context.Context, currentUserID int64, userIDs []int64) ([]domain.User, error) { + if currentUserID == 0 { + return nil, ErrNotAuthorized + } + ids := uniqueUserIDs(userIDs, maxBatchUsers) + if len(ids) == 0 { + return nil, nil + } + s.dropCachedUsers(ctx, ids...) + users, err := s.users.ByIDs(ctx, ids) + if err != nil { + return nil, err + } + s.putCachedUsers(ctx, users...) + return s.projectUsers(ctx, currentUserID, users) +} + // ByIDsForViewers 跨多个 viewer 批量投影同一组 user（fan-out 模板化）：base user 只加载一次， // 隐私/改名/头像投影经 userprojection.ForViewers 压成 O(owner) 查询。返回 map[viewerID][]User， // 每个切片与 ByIDs(viewer, ids) 字节等价——**唯一例外是 personal photo overlay**（ForViewers v1 diff --git a/internal/app/users/service_test.go b/internal/app/users/service_test.go index f515a195..d6a39878 100644 --- a/internal/app/users/service_test.go +++ b/internal/app/users/service_test.go @@ -259,6 +259,49 @@ func TestServiceUsesBaseCacheWithoutCachingViewerOverlay(t *testing.T) { } } +func TestServiceAuthoritativeUserReloadReplacesStaleBaseCache(t *testing.T) { + ctx := context.Background() + base := memory.NewUserStore() + viewer, err := base.Create(ctx, domain.User{AccessHash: 1, Phone: "15550000011", FirstName: "Viewer"}) + if err != nil { + t.Fatalf("create viewer: %v", err) + } + target, err := base.Create(ctx, domain.User{AccessHash: 2, Phone: "15550000012", FirstName: "Target"}) + if err != nil { + t.Fatalf("create target: %v", err) + } + store := &countingUserStore{UserStore: base} + cache := newMemoryBaseUserCache() + svc := NewService(store, WithBaseUserCache(cache)) + + primed, err := svc.ByIDs(ctx, viewer.ID, []int64{target.ID}) + if err != nil || len(primed) != 1 || primed[0].Scam { + t.Fatalf("prime ByIDs users=%+v err=%v", primed, err) + } + if _, err := base.SetScamFake(ctx, target.ID, true, false); err != nil { + t.Fatalf("commit moderation flags behind cache: %v", err) + } + stale, err := svc.ByIDs(ctx, viewer.ID, []int64{target.ID}) + if err != nil || len(stale) != 1 || stale[0].Scam { + t.Fatalf("ordinary cached ByIDs users=%+v err=%v, want stale scam=false", stale, err) + } + + fresh, err := svc.ByIDsAuthoritative(ctx, viewer.ID, []int64{target.ID}) + if err != nil || len(fresh) != 1 || !fresh[0].Scam || fresh[0].Fake { + t.Fatalf("authoritative ByIDs users=%+v err=%v, want scam=true fake=false", fresh, err) + } + if store.byIDsCalls != 2 { + t.Fatalf("store ByIDs calls=%d, want prime + authoritative reload", store.byIDsCalls) + } + cached, err := svc.ByIDs(ctx, viewer.ID, []int64{target.ID}) + if err != nil || len(cached) != 1 || !cached[0].Scam || cached[0].Fake { + t.Fatalf("replaced cache ByIDs users=%+v err=%v, want scam=true fake=false", cached, err) + } + if store.byIDsCalls != 2 { + t.Fatalf("store ByIDs calls after cached read=%d, want unchanged", store.byIDsCalls) + } +} + func TestServiceRefreshesBaseCacheAfterProfileUpdate(t *testing.T) { ctx := context.Background() base := memory.NewUserStore() diff --git a/internal/rpc/deps.go b/internal/rpc/deps.go index f8c301bd..73f754bc 100644 --- a/internal/rpc/deps.go +++ b/internal/rpc/deps.go @@ -249,6 +249,15 @@ type UsersService interface { ByIDs(ctx context.Context, currentUserID int64, userIDs []int64) ([]domain.User, error) } +// UserAuthoritativeProjectionService bypasses viewer-independent base caches +// for an explicit durable profile-refresh event. The event exists precisely +// because a just-committed absolute user fact must replace client and server +// caches; re-reading a stale Redis value would acknowledge the outbox row +// without ever exposing the committed state. +type UserAuthoritativeProjectionService interface { + ByIDsAuthoritative(ctx context.Context, currentUserID int64, userIDs []int64) ([]domain.User, error) +} + // TelegramLoginService is the domain-only boundary shared by the MTProto RPC // edge and the public OIDC provider. PostgreSQL remains authoritative for all // consent transitions; the RPC layer only projects domain state to TL. @@ -781,6 +790,12 @@ type ChannelsService interface { FilterActiveMemberIDs(ctx context.Context, channelID int64, userIDs []int64) ([]int64, error) } +// ChannelAuthoritativeProjectionService bypasses long-lived channel read +// models for a durable channel_state refresh emitted by an admin mutation. +type ChannelAuthoritativeProjectionService interface { + GetChannelsAuthoritative(ctx context.Context, userID int64, channelIDs []int64) ([]domain.ChannelView, error) +} + // CommunitiesService abstracts the Layer 228 Community aggregation domain. // Community containers never expose tg types and never own message/read/pts state. type CommunitiesService interface { diff --git a/internal/rpc/moderation_profile_update_test.go b/internal/rpc/moderation_profile_update_test.go index aec28b7b..5c5bd3da 100644 --- a/internal/rpc/moderation_profile_update_test.go +++ b/internal/rpc/moderation_profile_update_test.go @@ -1,13 +1,47 @@ package rpc import ( + "context" "testing" + "github.com/iamxvbaba/td/clock" "github.com/iamxvbaba/td/tg" + "go.uber.org/zap" "telesrv/internal/domain" ) +type moderationProjectionUsers struct { + UsersService + freshCalls int +} + +func (s *moderationProjectionUsers) ByIDs(_ context.Context, _ int64, ids []int64) ([]domain.User, error) { + return []domain.User{{ID: ids[0], FirstName: "stale"}}, nil +} + +func (s *moderationProjectionUsers) ByIDsAuthoritative(_ context.Context, _ int64, ids []int64) ([]domain.User, error) { + s.freshCalls++ + return []domain.User{{ID: ids[0], FirstName: "fresh", Scam: true}}, nil +} + +type moderationProjectionChannels struct { + ChannelsService + freshCalls int +} + +func (s *moderationProjectionChannels) GetChannels(_ context.Context, _ int64, ids []int64) ([]domain.ChannelView, error) { + return []domain.ChannelView{{Channel: domain.Channel{ID: ids[0], Title: "stale", Megagroup: true}}}, nil +} + +func (s *moderationProjectionChannels) GetChannelsAuthoritative(_ context.Context, viewerUserID int64, ids []int64) ([]domain.ChannelView, error) { + s.freshCalls++ + return []domain.ChannelView{{ + Channel: domain.Channel{ID: ids[0], Title: "fresh", Megagroup: true, Scam: true}, + Self: domain.ChannelMember{ChannelID: ids[0], UserID: viewerUserID, Status: domain.ChannelMemberActive}, + }}, nil +} + func TestModerationProfileUpdateCarriesStandardFlagsAndPts(t *testing.T) { const ( viewerID = int64(1001) @@ -95,3 +129,28 @@ func TestChannelModerationUpdateCarriesStandardFlagsAndPts(t *testing.T) { t.Fatalf("channel = %T %+v", updates.Chats[0], updates.Chats[0]) } } + +func TestModerationRefreshEventsBypassServerProjectionCaches(t *testing.T) { + users := &moderationProjectionUsers{} + channels := &moderationProjectionChannels{} + r := New(Config{}, Deps{Users: users, Channels: channels}, zap.NewNop(), clock.System) + const viewerID = int64(5005) + events := r.enrichUpdateEvents(context.Background(), viewerID, []domain.UpdateEvent{ + { + UserID: viewerID, Type: domain.UpdateEventUserProfile, + Peer: domain.Peer{Type: domain.PeerTypeUser, ID: 6006}, + }, + { + UserID: viewerID, Type: domain.UpdateEventChannelState, + Peer: domain.Peer{Type: domain.PeerTypeChannel, ID: 7007}, + }, + }) + if users.freshCalls != 1 || len(events[0].Users) != 1 || + events[0].Users[0].FirstName != "fresh" || !events[0].Users[0].Scam { + t.Fatalf("authoritative user refresh = calls:%d users:%+v", users.freshCalls, events[0].Users) + } + if channels.freshCalls != 1 || len(events[1].Channels) != 1 || + events[1].Channels[0].Title != "fresh" || !events[1].Channels[0].Scam { + t.Fatalf("authoritative channel refresh = calls:%d channels:%+v", channels.freshCalls, events[1].Channels) + } +} diff --git a/internal/rpc/update_peer_refs.go b/internal/rpc/update_peer_refs.go index c98c8578..fd548a11 100644 --- a/internal/rpc/update_peer_refs.go +++ b/internal/rpc/update_peer_refs.go @@ -3,6 +3,8 @@ package rpc import ( "context" + "go.uber.org/zap" + "telesrv/internal/domain" ) @@ -25,6 +27,38 @@ func (r *Router) enrichUpdateEventsWithPeerCache(ctx context.Context, viewerUser allUserIDs := make(map[int64]struct{}) allChannelIDs := make(map[int64]struct{}) for i := range out { + if out[i].Type == domain.UpdateEventUserProfile { + if service, ok := r.deps.Users.(UserAuthoritativeProjectionService); ok { + users, err := service.ByIDsAuthoritative(ctx, viewerUserID, []int64{out[i].Peer.ID}) + if err != nil { + r.log.Warn("reload authoritative user profile event", + zap.Int64("viewer_user_id", viewerUserID), + zap.Int64("target_user_id", out[i].Peer.ID), + zap.Error(err)) + } else { + out[i].Users = users + cache.primeUsers(viewerUserID, users) + } + } + } + if out[i].Type == domain.UpdateEventChannelState { + if service, ok := r.deps.Channels.(ChannelAuthoritativeProjectionService); ok { + views, err := service.GetChannelsAuthoritative(ctx, viewerUserID, []int64{out[i].Peer.ID}) + if err != nil { + r.log.Warn("reload authoritative channel state event", + zap.Int64("viewer_user_id", viewerUserID), + zap.Int64("channel_id", out[i].Peer.ID), + zap.Error(err)) + } else { + out[i].Channels = out[i].Channels[:0] + for _, view := range views { + if view.Channel.ID != 0 { + out[i].Channels = append(out[i].Channels, view.Channel) + } + } + } + } + } if out[i].Type == domain.UpdateEventMessageReactions { out[i] = r.enrichMessageReactionEvent(ctx, viewerUserID, out[i]) }