diff --git a/cmd/telesrv/main.go b/cmd/telesrv/main.go index c518471c..9fb09526 100644 --- a/cmd/telesrv/main.go +++ b/cmd/telesrv/main.go @@ -942,21 +942,22 @@ func run(logger *zap.Logger) error { go readModelListener.Run(ctx) activeSessions.SetLifecycleObserver(router) adminService.Configure(adminapp.Dependencies{ - Auth: authService, - Revoker: router, - Users: usersService, - Stars: starsService, - StarsNotifier: router, - UserNotifier: router, - FreezeNotifier: router, - Channels: channelsService, - ChannelNotifier: router, - Messages: messagesService, - Gifts: giftsService, - GiftGranter: router, - Bots: botsService, - Emoji: filesService, - Moderation: moderationService, + Auth: authService, + Revoker: router, + Users: usersService, + Stars: starsService, + StarsNotifier: router, + UserNotifier: router, + UserModerationNotifier: router, + FreezeNotifier: router, + Channels: channelsService, + ChannelNotifier: router, + Messages: messagesService, + Gifts: giftsService, + GiftGranter: router, + Bots: botsService, + Emoji: filesService, + Moderation: moderationService, }) moderationActionOptions := []moderationapp.ActionExecutorOption{} if cfg.PublicLinkWebAddr != "" { diff --git a/internal/admin/service.go b/internal/admin/service.go index cb66aa95..0c51cb38 100644 --- a/internal/admin/service.go +++ b/internal/admin/service.go @@ -107,6 +107,10 @@ type UserNotifier interface { NotifyUserChanged(ctx context.Context, u domain.User) error } +type UserModerationNotifier interface { + NotifyUserModerationFlagsChanged(ctx context.Context, u domain.User) error +} + type AccountFreezeNotifier interface { NotifyAccountFreezeChanged(ctx context.Context, freeze domain.AccountFreeze) error } @@ -184,47 +188,49 @@ type GiftGranter interface { } type Dependencies struct { - Commands CommandRepository - Restrictions RestrictionStore - Auth AuthService - Revoker AuthKeyRevoker - Users UsersService - Stars StarsService - StarsNotifier StarsNotifier - UserNotifier UserNotifier - FreezeNotifier AccountFreezeNotifier - Channels ChannelsService - ChannelNotifier ChannelNotifier - Messages MessagesService - Gifts GiftsService - GiftGranter GiftGranter - OfficialGifts OfficialGiftsSource - Bots BotService - Emoji EmojiService - Moderation ModerationService - Now func() time.Time + Commands CommandRepository + Restrictions RestrictionStore + Auth AuthService + Revoker AuthKeyRevoker + Users UsersService + Stars StarsService + StarsNotifier StarsNotifier + UserNotifier UserNotifier + UserModerationNotifier UserModerationNotifier + FreezeNotifier AccountFreezeNotifier + Channels ChannelsService + ChannelNotifier ChannelNotifier + Messages MessagesService + Gifts GiftsService + GiftGranter GiftGranter + OfficialGifts OfficialGiftsSource + Bots BotService + Emoji EmojiService + Moderation ModerationService + Now func() time.Time } type Service struct { - commands CommandRepository - restrictions RestrictionStore - auth AuthService - revoker AuthKeyRevoker - users UsersService - stars StarsService - starsNotifier StarsNotifier - userNotifier UserNotifier - freezeNotifier AccountFreezeNotifier - channels ChannelsService - channelNotifier ChannelNotifier - messages MessagesService - gifts GiftsService - giftGranter GiftGranter - officialGifts OfficialGiftsSource - bots BotService - emoji EmojiService - moderation ModerationService - now func() time.Time + commands CommandRepository + restrictions RestrictionStore + auth AuthService + revoker AuthKeyRevoker + users UsersService + stars StarsService + starsNotifier StarsNotifier + userNotifier UserNotifier + userModerationNotifier UserModerationNotifier + freezeNotifier AccountFreezeNotifier + channels ChannelsService + channelNotifier ChannelNotifier + messages MessagesService + gifts GiftsService + giftGranter GiftGranter + officialGifts OfficialGiftsSource + bots BotService + emoji EmojiService + moderation ModerationService + now func() time.Time } func NewService(deps Dependencies) *Service { @@ -257,6 +263,9 @@ func (s *Service) Configure(deps Dependencies) *Service { if deps.UserNotifier != nil { s.userNotifier = deps.UserNotifier } + if deps.UserModerationNotifier != nil { + s.userModerationNotifier = deps.UserModerationNotifier + } if deps.FreezeNotifier != nil { s.freezeNotifier = deps.FreezeNotifier } @@ -969,7 +978,7 @@ func (s *Service) SetUserFlags(ctx context.Context, req SetUserFlagsRequest) (Co } details["updated_scam"] = updated.Scam details["updated_fake"] = updated.Fake - if err := s.notifyUserChanged(ctx, updated); err != nil { + if err := s.notifyUserModerationFlagsChanged(ctx, updated); err != nil { details["notify_error"] = err.Error() } return CommandResult{Message: "user flags updated", Details: details}, nil @@ -2232,6 +2241,13 @@ func (s *Service) notifyUserChanged(ctx context.Context, u domain.User) error { return s.userNotifier.NotifyUserChanged(ctx, u) } +func (s *Service) notifyUserModerationFlagsChanged(ctx context.Context, u domain.User) error { + if s == nil || s.userModerationNotifier == nil { + return s.notifyUserChanged(ctx, u) + } + return s.userModerationNotifier.NotifyUserModerationFlagsChanged(ctx, u) +} + func (s *Service) notifyAccountFreezeChanged(ctx context.Context, freeze domain.AccountFreeze) error { if s == nil || s.freezeNotifier == nil { return nil diff --git a/internal/admin/service_test.go b/internal/admin/service_test.go index 4eb952c4..ee653cf1 100644 --- a/internal/admin/service_test.go +++ b/internal/admin/service_test.go @@ -143,6 +143,39 @@ func TestModerationFlagsRejectImpossibleScamFakeState(t *testing.T) { } } +func TestSetUserFlagsUsesNonPTSModerationNotifier(t *testing.T) { + ctx := context.Background() + users := &fakeUsersService{users: map[int64]domain.User{ + 1001: {ID: 1001, FirstName: "Alice"}, + }} + ordinaryNotifier := &fakeUserNotifier{} + moderationNotifier := &fakeUserModerationNotifier{} + svc := NewService(Dependencies{ + Commands: newMemoryCommandRepo(), + Users: users, + UserNotifier: ordinaryNotifier, + UserModerationNotifier: moderationNotifier, + Now: fixedNow, + }) + + if _, err := svc.SetUserFlags(ctx, SetUserFlagsRequest{ + CommandMeta: CommandMeta{CommandID: "set-user-scam", Actor: "ops", Reason: "confirmed report"}, + UserID: 1001, + Scam: true, + }); err != nil { + t.Fatalf("SetUserFlags: %v", err) + } + if got := users.users[1001]; !got.Scam || got.Fake { + t.Fatalf("updated user = %+v", got) + } + if len(moderationNotifier.users) != 1 || moderationNotifier.users[0] != 1001 { + t.Fatalf("moderation notifications = %v", moderationNotifier.users) + } + if len(ordinaryNotifier.users) != 0 { + t.Fatalf("ordinary notifications = %v, want dedicated non-PTS path", ordinaryNotifier.users) + } +} + func TestAccountFreezesBatchesAndReturnsOnlyActiveFacts(t *testing.T) { now := fixedNow() store := &fakeBatchRestrictionStore{fakeRestrictionStore: fakeRestrictionStore{items: map[int64]domain.AccountFreeze{ @@ -863,6 +896,15 @@ func (f *fakeUserNotifier) NotifyUserChanged(_ context.Context, u domain.User) e return nil } +type fakeUserModerationNotifier struct { + users []int64 +} + +func (f *fakeUserModerationNotifier) NotifyUserModerationFlagsChanged(_ context.Context, u domain.User) error { + f.users = append(f.users, u.ID) + return nil +} + type fakeChannelsService struct { channels map[int64]domain.Channel verifiedCalls int diff --git a/internal/app/users/service.go b/internal/app/users/service.go index 2b050fa1..97703e97 100644 --- a/internal/app/users/service.go +++ b/internal/app/users/service.go @@ -33,6 +33,10 @@ type usernameAvailabilityStore interface { CheckUsername(ctx context.Context, userID int64, username string) (bool, error) } +type moderationFlagAudienceStore interface { + ModerationFlagAudience(ctx context.Context, userID int64, limit int) ([]int64, error) +} + // Option 调整用户服务可选依赖。 type Option func(*Service) @@ -176,27 +180,6 @@ 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 @@ -421,6 +404,23 @@ func (s *Service) SetScamFake(ctx context.Context, userID int64, scam, fake bool return updated, nil } +// ModerationFlagAudience returns the bounded set of existing viewers that may +// need an immediate updateUser after SCAM/FAKE changes. This is an online +// accelerator only: it does not allocate PTS or create durable update events. +func (s *Service) ModerationFlagAudience(ctx context.Context, userID int64, limit int) ([]int64, error) { + if userID == 0 { + return nil, ErrNotAuthorized + } + if limit <= 0 || limit > 4096 { + limit = 4096 + } + audience, ok := s.users.(moderationFlagAudienceStore) + if !ok { + return []int64{userID}, nil + } + return audience.ModerationFlagAudience(ctx, userID, limit) +} + // SetSupport 设置/取消用户的 support 标记（官方客服账号）。写后刷新基础缓存。 func (s *Service) SetSupport(ctx context.Context, userID int64, support bool) (domain.User, error) { if userID == 0 { diff --git a/internal/app/users/service_test.go b/internal/app/users/service_test.go index d6a39878..f515a195 100644 --- a/internal/app/users/service_test.go +++ b/internal/app/users/service_test.go @@ -259,49 +259,6 @@ 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/domain/update_event.go b/internal/domain/update_event.go index 07ac5099..0c1aa7b8 100644 --- a/internal/domain/update_event.go +++ b/internal/domain/update_event.go @@ -36,12 +36,7 @@ const ( // UpdateEventUserEmojiStatus carries the exact immutable status snapshot. // It consumes account pts even though updateUserEmojiStatus has no pts. UpdateEventUserEmojiStatus UpdateEventType = "user_emoji_status" - // UpdateEventUserProfile is a durable viewer-scoped refresh for another - // user's absolute profile flags (currently moderation SCAM/FAKE state). - // The changed user is carried in Peer; hydration attaches the authoritative - // viewer projection to Users before online dispatch or getDifference. - UpdateEventUserProfile UpdateEventType = "user_profile" - UpdateEventDeleteMessages UpdateEventType = "delete_messages" + UpdateEventDeleteMessages UpdateEventType = "delete_messages" // UpdateEventPinnedMessages 映射 updatePinnedMessages（私聊置顶/取消 // 置顶；MessageIDs 是该 owner 自己视角的 box id，Bool 为 pinned）。 // TL 构造器自带账号 pts/pts_count，不属于 LacksWirePts。 @@ -145,7 +140,6 @@ func (e UpdateEvent) LacksWirePts() bool { UpdateEventPeerStoryBlocked, UpdateEventUserPhone, UpdateEventUserEmojiStatus, - UpdateEventUserProfile, UpdateEventDialogFilter, UpdateEventDialogFilterOrder, UpdateEventDialogFilters, diff --git a/internal/rpc/convert_updates.go b/internal/rpc/convert_updates.go index 6fef3f78..efdb6bea 100644 --- a/internal/rpc/convert_updates.go +++ b/internal/rpc/convert_updates.go @@ -242,11 +242,6 @@ func tgOtherUpdateFromEvent(event domain.UpdateEvent) tg.UpdateClass { return nil } return &tg.UpdateUserEmojiStatus{UserID: event.UserID, EmojiStatus: tgUserEmojiStatusValue(event.EmojiStatus)} - case domain.UpdateEventUserProfile: - if event.Peer.Type != domain.PeerTypeUser || event.Peer.ID == 0 { - return nil - } - return &tg.UpdateUser{UserID: event.Peer.ID} case domain.UpdateEventChannelState: if event.Peer.Type != domain.PeerTypeChannel || event.Peer.ID == 0 { return nil diff --git a/internal/rpc/deps.go b/internal/rpc/deps.go index 888717a7..7375f64c 100644 --- a/internal/rpc/deps.go +++ b/internal/rpc/deps.go @@ -249,15 +249,6 @@ 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. diff --git a/internal/rpc/moderation_profile_update_test.go b/internal/rpc/moderation_profile_update_test.go index 5c5bd3da..50e02fe9 100644 --- a/internal/rpc/moderation_profile_update_test.go +++ b/internal/rpc/moderation_profile_update_test.go @@ -13,16 +13,17 @@ import ( type moderationProjectionUsers struct { UsersService - freshCalls int + audience []int64 + viewers []int64 } -func (s *moderationProjectionUsers) ByIDs(_ context.Context, _ int64, ids []int64) ([]domain.User, error) { - return []domain.User{{ID: ids[0], FirstName: "stale"}}, nil +func (s *moderationProjectionUsers) ByIDs(_ context.Context, viewerUserID int64, ids []int64) ([]domain.User, error) { + s.viewers = append(s.viewers, viewerUserID) + return []domain.User{{ID: ids[0], FirstName: "Flagged", Scam: true}}, 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 +func (s *moderationProjectionUsers) ModerationFlagAudience(_ context.Context, _ int64, _ int) ([]int64, error) { + return append([]int64(nil), s.audience...), nil } type moderationProjectionChannels struct { @@ -42,84 +43,90 @@ func (s *moderationProjectionChannels) GetChannelsAuthoritative(_ context.Contex }}, nil } -func TestModerationProfileUpdateCarriesStandardFlagsAndPts(t *testing.T) { - const ( - viewerID = int64(1001) - targetID = int64(2002) - ) - event := domain.UpdateEvent{ - UserID: viewerID, - Type: domain.UpdateEventUserProfile, - Pts: 7, PtsCount: 1, Date: 1700000000, - Peer: domain.Peer{Type: domain.PeerTypeUser, ID: targetID}, - Users: []domain.User{{ - ID: targetID, AccessHash: 22, FirstName: "Flagged", Scam: true, - }}, - } +func (s *moderationProjectionChannels) FilterActiveMemberIDs(_ context.Context, _ int64, userIDs []int64) ([]int64, error) { + return append([]int64(nil), userIDs...), nil +} - updates := tgUpdateForOutboxEventForViewer(event, viewerID) - if updates == nil || len(updates.Updates) != 2 { - t.Fatalf("updates = %+v", updates) +func TestUserModerationFlagsPushStandardNonPTSUpdate(t *testing.T) { + const ( + targetID = int64(2002) + onlineViewerID = int64(1001) + offlineViewerID = int64(3003) + ) + users := &moderationProjectionUsers{ + audience: []int64{targetID, onlineViewerID, offlineViewerID}, + } + sessions := &captureSessions{onlineUserIDs: []int64{targetID, onlineViewerID}} + r := New(Config{}, Deps{Users: users, Sessions: sessions}, zap.NewNop(), clock.System) + + if err := r.NotifyUserModerationFlagsChanged(context.Background(), domain.User{ + ID: targetID, FirstName: "Flagged", Scam: true, + }); err != nil { + t.Fatalf("notify moderation flags: %v", err) + } + pushed := sessions.pushedUserIDs() + if len(pushed) != 2 || pushed[0] != targetID || pushed[1] != onlineViewerID { + t.Fatalf("pushed user ids = %v", pushed) + } + if len(users.viewers) != 2 || users.viewers[0] != targetID || users.viewers[1] != onlineViewerID { + t.Fatalf("projected viewers = %v", users.viewers) + } + updates, ok := sessions.lastUserPush().(*tg.Updates) + if !ok || len(updates.Updates) != 1 { + t.Fatalf("updates = %T %+v", sessions.lastUserPush(), sessions.lastUserPush()) } refresh, ok := updates.Updates[0].(*tg.UpdateUser) if !ok || refresh.UserID != targetID { t.Fatalf("refresh = %T %+v", updates.Updates[0], updates.Updates[0]) } - bookkeeping, ok := updates.Updates[1].(*tg.UpdateDeleteMessages) - if !ok || bookkeeping.Pts != 7 || bookkeeping.PtsCount != 1 || len(bookkeeping.Messages) != 0 { - t.Fatalf("bookkeeping = %T %+v", updates.Updates[1], updates.Updates[1]) + for _, update := range updates.Updates { + if _, syntheticDelete := update.(*tg.UpdateDeleteMessages); syntheticDelete { + t.Fatalf("synthetic delete bookkeeping leaked into moderation update: %+v", update) + } } if len(updates.Users) != 1 { t.Fatalf("users = %+v", updates.Users) } - user, ok := updates.Users[0].(*tg.User) - if !ok || user.ID != targetID || !user.Scam || user.Fake { + if user, ok := updates.Users[0].(*tg.User); !ok || user.ID != targetID || !user.Scam || user.Fake { t.Fatalf("user = %T %+v", updates.Users[0], updates.Users[0]) } - - difference := tgUpdatesDifference(viewerID, domain.UpdateDifference{ - State: domain.UpdateState{Pts: 7, Date: event.Date}, - Events: []domain.UpdateEvent{event}, - }) - full, ok := difference.(*tg.UpdatesDifference) - if !ok || len(full.OtherUpdates) != 1 || len(full.Users) != 1 { - t.Fatalf("difference = %T %+v", difference, difference) - } - if refresh, ok := full.OtherUpdates[0].(*tg.UpdateUser); !ok || refresh.UserID != targetID { - t.Fatalf("difference refresh = %T %+v", full.OtherUpdates[0], full.OtherUpdates[0]) - } - if user, ok := full.Users[0].(*tg.User); !ok || !user.Scam || user.Fake { - t.Fatalf("difference user = %T %+v", full.Users[0], full.Users[0]) - } } -func TestChannelModerationUpdateCarriesStandardFlagsAndPts(t *testing.T) { +func TestChannelModerationFlagsPushStandardNonPTSUpdate(t *testing.T) { const ( - viewerID = int64(3003) + ownerID = int64(3003) + memberID = int64(3004) channelID = int64(4004) ) - event := domain.UpdateEvent{ - UserID: viewerID, - Type: domain.UpdateEventChannelState, - Pts: 9, PtsCount: 1, Date: 1700000001, - Peer: domain.Peer{Type: domain.PeerTypeChannel, ID: channelID}, - Channels: []domain.Channel{{ - ID: channelID, AccessHash: 44, CreatorUserID: viewerID, - Title: "Flagged channel", Megagroup: true, Scam: true, - }}, + channels := &moderationProjectionChannels{} + sessions := &captureSessions{ + onlineUserIDs: []int64{ownerID, memberID}, + channelMembers: map[int64][]int64{channelID: {ownerID, memberID}}, + } + r := New(Config{}, Deps{Channels: channels, Sessions: sessions}, zap.NewNop(), clock.System) + if err := r.NotifyChannelChanged(context.Background(), domain.Channel{ + ID: channelID, AccessHash: 44, CreatorUserID: ownerID, + Title: "Flagged channel", Megagroup: true, Scam: true, + }); err != nil { + t.Fatalf("notify channel flags: %v", err) } - updates := tgUpdateForOutboxEventForViewer(event, viewerID) - if updates == nil || len(updates.Updates) != 2 { - t.Fatalf("updates = %+v", updates) + pushed := sessions.pushedUserIDs() + if len(pushed) != 2 { + t.Fatalf("pushed user ids = %v", pushed) + } + updates, ok := sessions.lastUserPush().(*tg.Updates) + if !ok || len(updates.Updates) != 1 { + t.Fatalf("updates = %T %+v", sessions.lastUserPush(), sessions.lastUserPush()) } refresh, ok := updates.Updates[0].(*tg.UpdateChannel) if !ok || refresh.ChannelID != channelID { t.Fatalf("refresh = %T %+v", updates.Updates[0], updates.Updates[0]) } - bookkeeping, ok := updates.Updates[1].(*tg.UpdateDeleteMessages) - if !ok || bookkeeping.Pts != 9 || bookkeeping.PtsCount != 1 || len(bookkeeping.Messages) != 0 { - t.Fatalf("bookkeeping = %T %+v", updates.Updates[1], updates.Updates[1]) + for _, update := range updates.Updates { + if _, syntheticDelete := update.(*tg.UpdateDeleteMessages); syntheticDelete { + t.Fatalf("synthetic delete bookkeeping leaked into moderation update: %+v", update) + } } if len(updates.Chats) != 1 { t.Fatalf("chats = %+v", updates.Chats) @@ -130,27 +137,18 @@ func TestChannelModerationUpdateCarriesStandardFlagsAndPts(t *testing.T) { } } -func TestModerationRefreshEventsBypassServerProjectionCaches(t *testing.T) { - users := &moderationProjectionUsers{} +func TestChannelStateRefreshEventBypassesServerProjectionCache(t *testing.T) { channels := &moderationProjectionChannels{} - r := New(Config{}, Deps{Users: users, Channels: channels}, zap.NewNop(), clock.System) + r := New(Config{}, Deps{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) + if channels.freshCalls != 1 || len(events[0].Channels) != 1 || + events[0].Channels[0].Title != "fresh" || !events[0].Channels[0].Scam { + t.Fatalf("authoritative channel refresh = calls:%d channels:%+v", channels.freshCalls, events[0].Channels) } } diff --git a/internal/rpc/outbox_dispatcher.go b/internal/rpc/outbox_dispatcher.go index fa9dab6a..f4850deb 100644 --- a/internal/rpc/outbox_dispatcher.go +++ b/internal/rpc/outbox_dispatcher.go @@ -543,17 +543,6 @@ func tgUpdateForOutboxEventForViewer(event domain.UpdateEvent, viewerUserID int6 Date: event.Date, Seq: 0, // 私聊不维护账号级 seq，恒 0 } - case domain.UpdateEventUserProfile: - update := tgOtherUpdateFromEvent(event) - if update == nil { - return nil - } - return &tg.Updates{ - Updates: appendAuxPtsBookkeeping([]tg.UpdateClass{update}, event), - Users: tgUsersForViewer(viewerUserID, event.Users), - Date: event.Date, - Seq: 0, - } case domain.UpdateEventChannelState: update := tgOtherUpdateFromEvent(event) if update == nil { diff --git a/internal/rpc/premium_sweeper.go b/internal/rpc/premium_sweeper.go index 2e0377a7..c06cf667 100644 --- a/internal/rpc/premium_sweeper.go +++ b/internal/rpc/premium_sweeper.go @@ -78,6 +78,66 @@ func (r *Router) NotifyUserChanged(ctx context.Context, u domain.User) error { return nil } +type moderationUserAudienceService interface { + ModerationFlagAudience(ctx context.Context, userID int64, limit int) ([]int64, error) +} + +// NotifyUserModerationFlagsChanged sends the standard, non-PTS updateUser +// shape to online accounts that already know the peer. Offline accounts +// converge when their next authoritative peer/dialog read carries the updated +// User flags; no synthetic message-box event is created. +func (r *Router) NotifyUserModerationFlagsChanged(ctx context.Context, u domain.User) error { + if r == nil || u.ID == 0 { + return nil + } + r.invalidateRPCProjectionForUser(u.ID) + if r.deps.Users == nil { + return nil + } + audience := []int64{u.ID} + if service, ok := r.deps.Users.(moderationUserAudienceService); ok { + viewers, err := service.ModerationFlagAudience(ctx, u.ID, 4096) + if err != nil { + r.log.Warn("list moderation user update audience", + zap.Int64("target_user_id", u.ID), + zap.Error(err)) + } else if len(viewers) != 0 { + audience = viewers + } + } + + pushCtx, cancel := context.WithTimeout(ctx, 10*time.Second) + defer cancel() + seen := make(map[int64]struct{}, len(audience)) + for _, viewerUserID := range audience { + if viewerUserID == 0 { + continue + } + if _, ok := seen[viewerUserID]; ok { + continue + } + seen[viewerUserID] = struct{}{} + if online, ok := r.deps.Sessions.(OnlineUserProvider); ok && !online.IsUserOnline(viewerUserID) { + continue + } + users, err := r.deps.Users.ByIDs(pushCtx, viewerUserID, []int64{u.ID}) + if err != nil || len(users) == 0 { + r.log.Warn("project moderation user update", + zap.Int64("viewer_user_id", viewerUserID), + zap.Int64("target_user_id", u.ID), + zap.Error(err)) + continue + } + r.pushUserUpdates(pushCtx, viewerUserID, &tg.Updates{ + Updates: []tg.UpdateClass{&tg.UpdateUser{UserID: u.ID}}, + Users: tgUsersForViewer(viewerUserID, users), + Date: int(r.clock.Now().Unix()), + Seq: 0, + }) + } + return nil +} + // pushPremiumStatusUpdate 向用户本人的全部在线 session 推送会员状态变化。 // 授予、到期与 admin 认证变更共用：updateUser 触发客户端用随附的 self user // 对象刷新 premium/verified 等基础 flag（TDesktop processUser 按 flag 翻转）。 diff --git a/internal/rpc/update_peer_refs.go b/internal/rpc/update_peer_refs.go index fd548a11..98916767 100644 --- a/internal/rpc/update_peer_refs.go +++ b/internal/rpc/update_peer_refs.go @@ -27,20 +27,6 @@ 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}) diff --git a/internal/store/postgres/channel_settings.go b/internal/store/postgres/channel_settings.go index b9ecd267..c3f07aed 100644 --- a/internal/store/postgres/channel_settings.go +++ b/internal/store/postgres/channel_settings.go @@ -10,7 +10,6 @@ import ( "github.com/jackc/pgx/v5" "telesrv/internal/domain" - "telesrv/internal/store/postgres/sqlcgen" ) func (s *ChannelStore) EditChannelTitle(ctx context.Context, req domain.EditChannelTitleRequest) (domain.EditChannelTitleResult, error) { @@ -338,34 +337,6 @@ FOR UPDATE`, channelID).Scan(¤tScam, ¤tFake); err != nil { if _, err := tx.Exec(ctx, `UPDATE channels SET scam = $2, fake = $3, updated_at = now() WHERE id = $1 AND NOT deleted`, channelID, scam, fake); err != nil { return domain.Channel{}, fmt.Errorf("set channel scam/fake: %w", err) } - audience, err := channelModerationFlagAudience(ctx, tx, channelID, maxModerationFlagAudience) - if err != nil { - return domain.Channel{}, err - } - qtx := sqlcgen.New(tx) - date := nowUnix() - for _, viewerUserID := range audience { - pts, err := reserveUserPts(ctx, tx, viewerUserID, 1) - if err != nil { - return domain.Channel{}, fmt.Errorf("reserve channel moderation flag pts for viewer %d: %w", viewerUserID, err) - } - event := domain.UpdateEvent{ - UserID: viewerUserID, - Type: domain.UpdateEventChannelState, - Pts: pts, PtsCount: 1, Date: date, - Peer: domain.Peer{Type: domain.PeerTypeChannel, ID: channelID}, - } - if err := appendUserUpdateEvent(ctx, tx, qtx, viewerUserID, event); err != nil { - return domain.Channel{}, fmt.Errorf("append channel moderation flag event for viewer %d: %w", viewerUserID, err) - } - if err := enqueueDispatch(ctx, qtx, sqlcgen.EnqueueDispatchParams{ - TargetUserID: viewerUserID, - Pts: int32(pts), - EventType: string(event.Type), - }); err != nil { - return domain.Channel{}, fmt.Errorf("enqueue channel moderation flag dispatch for viewer %d: %w", viewerUserID, err) - } - } if err := tx.Commit(ctx); err != nil { return domain.Channel{}, fmt.Errorf("commit channel scam/fake: %w", err) } @@ -378,37 +349,6 @@ FOR UPDATE`, channelID).Scan(¤tScam, ¤tFake); err != nil { return channel, nil } -func channelModerationFlagAudience(ctx context.Context, db sqlcgen.DBTX, channelID int64, limit int) ([]int64, error) { - if channelID <= 0 || limit <= 0 { - return nil, nil - } - rows, err := db.Query(ctx, ` -SELECT cm.user_id -FROM channel_members cm -JOIN users u ON u.id = cm.user_id AND u.deleted_at IS NULL -WHERE cm.channel_id = $1 AND cm.status = 'active' -ORDER BY cm.user_id -LIMIT $2`, channelID, limit) - if err != nil { - return nil, fmt.Errorf("list channel moderation flag audience: %w", err) - } - defer rows.Close() - out := make([]int64, 0) - for rows.Next() { - var userID int64 - if err := rows.Scan(&userID); err != nil { - return nil, fmt.Errorf("scan channel moderation flag audience: %w", err) - } - if userID != 0 { - out = append(out, userID) - } - } - if err := rows.Err(); err != nil { - return nil, fmt.Errorf("iterate channel moderation flag audience: %w", err) - } - return out, nil -} - // SetChannelAdminSettings applies an admin-direct moderation-settings patch // (no membership/permission checks). nil fields are left unchanged. func (s *ChannelStore) SetChannelAdminSettings(ctx context.Context, channelID int64, patch domain.ChannelAdminSettings) (domain.Channel, error) { diff --git a/internal/store/postgres/moderation_flags_integration_test.go b/internal/store/postgres/moderation_flags_integration_test.go index 516a6238..92f089e6 100644 --- a/internal/store/postgres/moderation_flags_integration_test.go +++ b/internal/store/postgres/moderation_flags_integration_test.go @@ -49,7 +49,7 @@ func TestModerationFlagsRejectImpossibleStateAtPostgresBoundary(t *testing.T) { } } -func TestUserModerationFlagsCreateDurableViewerProfileEvents(t *testing.T) { +func TestUserModerationFlagsDoNotAdvanceAccountPts(t *testing.T) { pool := testPool(t) ctx := context.Background() suffix := randomSuffix(t) @@ -72,6 +72,15 @@ func TestUserModerationFlagsCreateDurableViewerProfileEvents(t *testing.T) { }); err != nil { t.Fatalf("save reverse contact: %v", err) } + viewers := []domain.User{target, savedTargetViewer, savedByTargetViewer, unrelated} + baseline := make(map[int64]int, len(viewers)) + for _, viewer := range viewers { + pts, err := events.MaxContiguousPts(ctx, viewer.ID) + if err != nil { + t.Fatalf("viewer %d baseline pts: %v", viewer.ID, err) + } + baseline[viewer.ID] = pts + } updated, err := users.SetScamFake(ctx, target.ID, true, false) if err != nil { @@ -81,20 +90,16 @@ func TestUserModerationFlagsCreateDurableViewerProfileEvents(t *testing.T) { t.Fatalf("updated flags = scam:%v fake:%v", updated.Scam, updated.Fake) } - for _, viewer := range []domain.User{target, savedTargetViewer, savedByTargetViewer} { - got, err := events.ListAfter(ctx, viewer.ID, 0, 10) - if err != nil { - t.Fatalf("list viewer %d events: %v", viewer.ID, err) + for _, viewer := range viewers { + pts, err := events.MaxContiguousPts(ctx, viewer.ID) + if err != nil || pts != baseline[viewer.ID] { + t.Fatalf("viewer %d pts=%d want=%d err=%v", viewer.ID, pts, baseline[viewer.ID], err) } - if len(got) != 1 || got[0].Type != domain.UpdateEventUserProfile || - got[0].Peer != (domain.Peer{Type: domain.PeerTypeUser, ID: target.ID}) || - got[0].Pts != 1 || got[0].PtsCount != 1 { - t.Fatalf("viewer %d events = %+v", viewer.ID, got) + got, err := events.ListAfter(ctx, viewer.ID, baseline[viewer.ID], 10) + if err != nil || len(got) != 0 { + t.Fatalf("viewer %d moderation events=%+v err=%v", viewer.ID, got, err) } } - if got, err := events.ListAfter(ctx, unrelated.ID, 0, 10); err != nil || len(got) != 0 { - t.Fatalf("unrelated events = %+v err=%v", got, err) - } var outboxCount int if err := pool.QueryRow(ctx, ` SELECT count(*) @@ -102,33 +107,42 @@ FROM dispatch_outbox WHERE target_user_id = ANY($1::bigint[]) AND event_type = 'user_profile'`, []int64{target.ID, savedTargetViewer.ID, savedByTargetViewer.ID, unrelated.ID}, - ).Scan(&outboxCount); err != nil || outboxCount != 3 { + ).Scan(&outboxCount); err != nil || outboxCount != 0 { t.Fatalf("profile outbox count=%d err=%v", outboxCount, err) } + audience, err := users.ModerationFlagAudience(ctx, target.ID, 4096) + if err != nil { + t.Fatalf("moderation audience: %v", err) + } + audienceSet := make(map[int64]struct{}, len(audience)) + for _, userID := range audience { + audienceSet[userID] = struct{}{} + } + for _, viewer := range []domain.User{target, savedTargetViewer, savedByTargetViewer} { + if _, ok := audienceSet[viewer.ID]; !ok { + t.Fatalf("viewer %d missing from audience %v", viewer.ID, audience) + } + } + if _, ok := audienceSet[unrelated.ID]; ok { + t.Fatalf("unrelated viewer included in audience %v", audience) + } + if _, err := users.SetScamFake(ctx, target.ID, true, false); err != nil { t.Fatalf("repeat same flags: %v", err) } - for _, viewer := range []domain.User{target, savedTargetViewer, savedByTargetViewer} { - got, err := events.ListAfter(ctx, viewer.ID, 0, 10) - if err != nil || len(got) != 1 { - t.Fatalf("same-state viewer %d events = %+v err=%v", viewer.ID, got, err) - } - } - if _, err := users.SetScamFake(ctx, target.ID, false, true); err != nil { t.Fatalf("switch to fake: %v", err) } - for _, viewer := range []domain.User{target, savedTargetViewer, savedByTargetViewer} { - got, err := events.ListAfter(ctx, viewer.ID, 1, 10) - if err != nil || len(got) != 1 || got[0].Pts != 2 || - got[0].Type != domain.UpdateEventUserProfile { - t.Fatalf("second viewer %d events = %+v err=%v", viewer.ID, got, err) + for _, viewer := range viewers { + pts, err := events.MaxContiguousPts(ctx, viewer.ID) + if err != nil || pts != baseline[viewer.ID] { + t.Fatalf("viewer %d final pts=%d want=%d err=%v", viewer.ID, pts, baseline[viewer.ID], err) } } } -func TestChannelModerationFlagsCreateDurableMemberStateEvents(t *testing.T) { +func TestChannelModerationFlagsDoNotAdvanceMemberAccountPts(t *testing.T) { pool := testPool(t) ctx := context.Background() suffix := randomSuffix(t) @@ -166,17 +180,15 @@ func TestChannelModerationFlagsCreateDurableMemberStateEvents(t *testing.T) { if !updated.Scam || updated.Fake { t.Fatalf("updated flags = scam:%v fake:%v", updated.Scam, updated.Fake) } - for _, viewer := range []domain.User{owner, member} { - got, err := events.ListAfter(ctx, viewer.ID, baseline[viewer.ID], 10) - if err != nil || len(got) != 1 || - got[0].Type != domain.UpdateEventChannelState || - got[0].Peer != (domain.Peer{Type: domain.PeerTypeChannel, ID: created.Channel.ID}) || - got[0].Pts != baseline[viewer.ID]+1 || got[0].PtsCount != 1 { - t.Fatalf("viewer %d events = %+v err=%v", viewer.ID, got, err) + for _, viewer := range []domain.User{owner, member, unrelated} { + pts, err := events.MaxContiguousPts(ctx, viewer.ID) + if err != nil || pts != baseline[viewer.ID] { + t.Fatalf("viewer %d pts=%d want=%d err=%v", viewer.ID, pts, baseline[viewer.ID], err) + } + got, err := events.ListAfter(ctx, viewer.ID, baseline[viewer.ID], 10) + if err != nil || len(got) != 0 { + t.Fatalf("viewer %d moderation events=%+v err=%v", viewer.ID, got, err) } - } - if got, err := events.ListAfter(ctx, unrelated.ID, baseline[unrelated.ID], 10); err != nil || len(got) != 0 { - t.Fatalf("unrelated events = %+v err=%v", got, err) } var outboxCount int @@ -187,28 +199,20 @@ WHERE target_user_id = ANY($1::bigint[]) AND event_type = 'channel_state' AND pts > 0`, []int64{owner.ID, member.ID, unrelated.ID}, - ).Scan(&outboxCount); err != nil || outboxCount != 2 { + ).Scan(&outboxCount); err != nil || outboxCount != 0 { t.Fatalf("channel state outbox count=%d err=%v", outboxCount, err) } if _, err := channels.SetChannelScamFake(ctx, created.Channel.ID, true, false); err != nil { t.Fatalf("repeat same channel flags: %v", err) } - for _, viewer := range []domain.User{owner, member} { - got, err := events.ListAfter(ctx, viewer.ID, baseline[viewer.ID], 10) - if err != nil || len(got) != 1 { - t.Fatalf("same-state viewer %d events = %+v err=%v", viewer.ID, got, err) - } - } - if _, err := channels.SetChannelScamFake(ctx, created.Channel.ID, false, true); err != nil { t.Fatalf("switch channel to fake: %v", err) } - for _, viewer := range []domain.User{owner, member} { - got, err := events.ListAfter(ctx, viewer.ID, baseline[viewer.ID]+1, 10) - if err != nil || len(got) != 1 || got[0].Pts != baseline[viewer.ID]+2 || - got[0].Type != domain.UpdateEventChannelState { - t.Fatalf("second viewer %d events = %+v err=%v", viewer.ID, got, err) + for _, viewer := range []domain.User{owner, member, unrelated} { + pts, err := events.MaxContiguousPts(ctx, viewer.ID) + if err != nil || pts != baseline[viewer.ID] { + t.Fatalf("viewer %d final pts=%d want=%d err=%v", viewer.ID, pts, baseline[viewer.ID], err) } } } diff --git a/internal/store/postgres/user.go b/internal/store/postgres/user.go index c3fa169f..4846f014 100644 --- a/internal/store/postgres/user.go +++ b/internal/store/postgres/user.go @@ -391,33 +391,6 @@ FOR UPDATE`, userID).Scan(¤tScam, ¤tFake); err != nil { } return domain.User{}, fmt.Errorf("set user scam/fake: %w", err) } - audience, err := moderationFlagAudience(ctx, tx, userID, maxModerationFlagAudience) - if err != nil { - return domain.User{}, err - } - date := int(time.Now().Unix()) - for _, viewerUserID := range audience { - pts, err := reserveUserPts(ctx, tx, viewerUserID, 1) - if err != nil { - return domain.User{}, fmt.Errorf("reserve moderation flag pts for viewer %d: %w", viewerUserID, err) - } - event := domain.UpdateEvent{ - UserID: viewerUserID, - Type: domain.UpdateEventUserProfile, - Pts: pts, PtsCount: 1, Date: date, - Peer: domain.Peer{Type: domain.PeerTypeUser, ID: userID}, - } - if err := appendUserUpdateEvent(ctx, tx, qtx, viewerUserID, event); err != nil { - return domain.User{}, fmt.Errorf("append moderation flag event for viewer %d: %w", viewerUserID, err) - } - if err := enqueueDispatch(ctx, qtx, sqlcgen.EnqueueDispatchParams{ - TargetUserID: viewerUserID, - Pts: int32(pts), - EventType: string(event.Type), - }); err != nil { - return domain.User{}, fmt.Errorf("enqueue moderation flag dispatch for viewer %d: %w", viewerUserID, err) - } - } if err := tx.Commit(ctx); err != nil { return domain.User{}, fmt.Errorf("commit user scam/fake: %w", err) } @@ -427,10 +400,17 @@ FOR UPDATE`, userID).Scan(¤tScam, ¤tFake); err != nil { const maxModerationFlagAudience = 4096 -// moderationFlagAudience returns the bounded set of accounts that can already -// observe the target through a direct contact or private dialog. The final -// user_id ordering is deliberate: concurrent moderation changes acquire user -// pts watermarks in the same global order and cannot deadlock by target order. +// ModerationFlagAudience returns the bounded set of accounts that can already +// observe the target through a direct contact or private dialog. It is used +// only for best-effort, non-PTS updateUser fanout after the authoritative flag +// mutation commits. +func (s *UserStore) ModerationFlagAudience(ctx context.Context, userID int64, limit int) ([]int64, error) { + if limit > maxModerationFlagAudience { + limit = maxModerationFlagAudience + } + return moderationFlagAudience(ctx, s.db, userID, limit) +} + func moderationFlagAudience(ctx context.Context, db sqlcgen.DBTX, userID int64, limit int) ([]int64, error) { if userID <= 0 || limit <= 0 { return nil, nil