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

fix: sync non-PTS moderation updates

Browse source
This commit is contained in:
iamxvbaba 2026-07-24 11:57:00 +08:00
parent cc76cd3679
commit 70e57b4d07
15 changed files with 331 additions and 378 deletions
Download patch file Download diff file Expand all files Collapse all files

94
internal/admin/service.go
View file

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

42
internal/admin/service_test.go
View file

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

42
internal/app/users/service.go
View file

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

43
internal/app/users/service_test.go
View file

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

8
internal/domain/update_event.go
View file

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

5
internal/rpc/convert_updates.go
View file

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

9
internal/rpc/deps.go
View file

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

146
internal/rpc/moderation_profile_update_test.go
View file

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

11
internal/rpc/outbox_dispatcher.go
View file

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

60
internal/rpc/premium_sweeper.go
View file

@ -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 翻转）。

14
internal/rpc/update_peer_refs.go
View file

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

60
internal/store/postgres/channel_settings.go
View file

@ -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(&currentScam, &currentFake); 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(&currentScam, &currentFake); 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) {

102
internal/store/postgres/moderation_flags_integration_test.go
View file

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

42
internal/store/postgres/user.go
View file

@ -391,33 +391,6 @@ FOR UPDATE`, userID).Scan(&currentScam, &currentFake); 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(&currentScam, &currentFake); 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