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

fix: sync moderation flag convergence

Browse source
This commit is contained in:
iamxvbaba 2026-07-24 11:57:00 +08:00
parent 5d5883a3d1
commit 3dd9c345d7
10 changed files with 564 additions and 5 deletions
Download patch file Download diff file Expand all files Collapse all files

26
deploy/migrations/0147_user_moderation_profile_events.down.sql Normal file
View file

@ -0,0 +1,26 @@
-- Preserve account pts continuity if this development migration is reversed:
-- a profile refresh is absolute state, so replacing it with noop is safer than
-- deleting business history and leaving a permanent difference gap.
UPDATE public.user_update_events
SET event_type = 'noop', peer_type = NULL, peer_id = NULL
WHERE event_type = 'user_profile';
UPDATE public.dispatch_outbox
SET event_type = 'noop'
WHERE event_type = 'user_profile';
ALTER TABLE public.user_update_events DROP CONSTRAINT IF EXISTS user_update_events_type_check;
ALTER TABLE public.user_update_events ADD CONSTRAINT user_update_events_type_check CHECK (
(event_type)::text = ANY (ARRAY[
'new_message', 'read_history_inbox', 'read_history_outbox', 'read_message_contents',
'edit_message', 'web_page', 'message_reactions', 'message_poll', 'draft_message', 'quick_replies',
'new_quick_reply', 'delete_quick_reply', 'quick_reply_message', 'delete_quick_reply_messages',
'contacts_reset', 'dialog_pinned', 'pinned_dialogs', 'pinned_messages', 'dialog_unread_mark',
'peer_settings', 'peer_story_blocked', 'user_phone', 'user_emoji_status', 'privacy', 'delete_messages',
'dialog_filter', 'dialog_filter_order', 'dialog_filters', 'folder_peers',
'channel_available_messages', 'channel_view_forum_as_messages', 'channel_state',
'saved_dialog_pinned', 'pinned_saved_dialogs', 'story', 'read_stories',
'sent_story_reaction', 'new_story_reaction', 'noop',
'read_channel_discussion_inbox', 'read_channel_discussion_outbox'
]::text[])
);

19
deploy/migrations/0147_user_moderation_profile_events.up.sql Normal file
View file

@ -0,0 +1,19 @@
-- SCAM/FAKE is viewer-visible absolute profile state. Each affected direct
-- contact/dialog viewer receives a durable account-pts refresh so online
-- outbox delivery and offline getDifference converge even when TDesktop keeps
-- its local contact/dialog cache across restarts.
ALTER TABLE public.user_update_events DROP CONSTRAINT IF EXISTS user_update_events_type_check;
ALTER TABLE public.user_update_events ADD CONSTRAINT user_update_events_type_check CHECK (
(event_type)::text = ANY (ARRAY[
'new_message', 'read_history_inbox', 'read_history_outbox', 'read_message_contents',
'edit_message', 'web_page', 'message_reactions', 'message_poll', 'draft_message', 'quick_replies',
'new_quick_reply', 'delete_quick_reply', 'quick_reply_message', 'delete_quick_reply_messages',
'contacts_reset', 'dialog_pinned', 'pinned_dialogs', 'pinned_messages', 'dialog_unread_mark',
'peer_settings', 'peer_story_blocked', 'user_phone', 'user_emoji_status', 'user_profile', 'privacy',
'delete_messages', 'dialog_filter', 'dialog_filter_order', 'dialog_filters', 'folder_peers',
'channel_available_messages', 'channel_view_forum_as_messages', 'channel_state',
'saved_dialog_pinned', 'pinned_saved_dialogs', 'story', 'read_stories',
'sent_story_reaction', 'new_story_reaction', 'noop',
'read_channel_discussion_inbox', 'read_channel_discussion_outbox'
]::text[])
);

6
internal/domain/update_event.go
View file

@ -36,6 +36,11 @@ 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"
// UpdateEventPrivacy carries the immutable account privacy key/rule
// snapshot committed at this pts. updatePrivacy has no wire pts.
UpdateEventPrivacy UpdateEventType = "privacy"
@ -144,6 +149,7 @@ func (e UpdateEvent) LacksWirePts() bool {
UpdateEventPeerStoryBlocked,
UpdateEventUserPhone,
UpdateEventUserEmojiStatus,
UpdateEventUserProfile,
UpdateEventPrivacy,
UpdateEventDialogFilter,
UpdateEventDialogFilterOrder,

5
internal/rpc/convert_updates.go
View file

@ -242,6 +242,11 @@ 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.UpdateEventPrivacy:
if event.Privacy.OwnerUserID == 0 || event.Privacy.Key == "" || len(event.Privacy.Rules) == 0 {
return nil

97
internal/rpc/moderation_profile_update_test.go Normal file
View file

@ -0,0 +1,97 @@
package rpc
import (
"testing"
"github.com/iamxvbaba/td/tg"
"telesrv/internal/domain"
)
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,
}},
}
updates := tgUpdateForOutboxEventForViewer(event, viewerID)
if updates == nil || len(updates.Updates) != 2 {
t.Fatalf("updates = %+v", updates)
}
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])
}
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 {
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) {
const (
viewerID = int64(3003)
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,
}},
}
updates := tgUpdateForOutboxEventForViewer(event, viewerID)
if updates == nil || len(updates.Updates) != 2 {
t.Fatalf("updates = %+v", updates)
}
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])
}
if len(updates.Chats) != 1 {
t.Fatalf("chats = %+v", updates.Chats)
}
channel, ok := updates.Chats[0].(*tg.Channel)
if !ok || channel.ID != channelID || !channel.Scam || channel.Fake {
t.Fatalf("channel = %T %+v", updates.Chats[0], updates.Chats[0])
}
}

22
internal/rpc/outbox_dispatcher.go
View file

@ -543,6 +543,28 @@ 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 {
return nil
}
return &tg.Updates{
Updates: appendAuxPtsBookkeeping([]tg.UpdateClass{update}, event),
Chats: tgChannels(viewerUserID, event.Channels),
Date: event.Date,
Seq: 0,
}
case domain.UpdateEventNoop:
return nil
default:

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

@ -6,7 +6,11 @@ import (
"errors"
"fmt"
"strings"
"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) {
@ -292,16 +296,80 @@ func (s *ChannelStore) SetChannelScamFake(ctx context.Context, channelID int64,
if scam && fake {
return domain.Channel{}, domain.ErrPeerModerationFlagsInvalid
}
channel, err := s.channelByID(ctx, s.db, channelID)
beginner, ok := s.db.(txBeginner)
if !ok {
return domain.Channel{}, fmt.Errorf("set channel scam/fake: db does not support transactions")
}
tx, err := beginner.Begin(ctx)
if err != nil {
return domain.Channel{}, fmt.Errorf("begin set channel scam/fake: %w", err)
}
committed := false
defer func() {
if !committed {
_ = tx.Rollback(ctx)
}
}()
var currentScam, currentFake bool
if err := tx.QueryRow(ctx, `
SELECT scam, fake
FROM channels
WHERE id = $1 AND NOT deleted
FOR UPDATE`, channelID).Scan(&currentScam, &currentFake); err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return domain.Channel{}, domain.ErrChannelInvalid
}
return domain.Channel{}, fmt.Errorf("lock channel scam/fake: %w", err)
}
channel, err := s.channelByID(ctx, tx, channelID)
if err != nil {
return domain.Channel{}, err
}
if channel.Scam == scam && channel.Fake == fake {
if err := tx.Commit(ctx); err != nil {
return domain.Channel{}, fmt.Errorf("commit unchanged channel scam/fake: %w", err)
}
committed = true
return channel, nil
}
if _, err := s.db.Exec(ctx, `UPDATE channels SET scam = $2, fake = $3, updated_at = now() WHERE id = $1 AND NOT deleted`, channelID, scam, fake); err != nil {
if currentScam != channel.Scam || currentFake != channel.Fake {
return domain.Channel{}, fmt.Errorf("channel scam/fake snapshot changed while locked")
}
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)
}
committed = true
if s.rowCache != nil {
s.rowCache.delete(channelID)
}
@ -310,6 +378,37 @@ func (s *ChannelStore) SetChannelScamFake(ctx context.Context, channelID int64,
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) {

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

@ -48,3 +48,167 @@ func TestModerationFlagsRejectImpossibleStateAtPostgresBoundary(t *testing.T) {
t.Fatalf("channel after rejected writes=%+v err=%v", gotChannel, err)
}
}
func TestUserModerationFlagsCreateDurableViewerProfileEvents(t *testing.T) {
pool := testPool(t)
ctx := context.Background()
suffix := randomSuffix(t)
users := NewUserStore(pool)
contacts := NewContactStore(pool)
events := NewUpdateEventStore(pool)
target := createTestUser(t, ctx, users, "+1782"+suffix+"71", "FlagTarget", "")
savedTargetViewer := createTestUser(t, ctx, users, "+1782"+suffix+"72", "SavedTarget", "")
savedByTargetViewer := createTestUser(t, ctx, users, "+1782"+suffix+"73", "SavedByTarget", "")
unrelated := createTestUser(t, ctx, users, "+1782"+suffix+"74", "Unrelated", "")
if _, err := contacts.Upsert(ctx, savedTargetViewer.ID, domain.ContactInput{
ContactUserID: target.ID, FirstName: target.FirstName,
}); err != nil {
t.Fatalf("save target contact: %v", err)
}
if _, err := contacts.Upsert(ctx, target.ID, domain.ContactInput{
ContactUserID: savedByTargetViewer.ID, FirstName: savedByTargetViewer.FirstName,
}); err != nil {
t.Fatalf("save reverse contact: %v", err)
}
updated, err := users.SetScamFake(ctx, target.ID, true, false)
if err != nil {
t.Fatalf("set scam: %v", err)
}
if !updated.Scam || updated.Fake {
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)
}
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)
}
}
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(*)
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 {
t.Fatalf("profile outbox count=%d err=%v", outboxCount, err)
}
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)
}
}
}
func TestChannelModerationFlagsCreateDurableMemberStateEvents(t *testing.T) {
pool := testPool(t)
ctx := context.Background()
suffix := randomSuffix(t)
users := NewUserStore(pool)
channels := NewChannelStore(pool)
events := NewUpdateEventStore(pool)
owner := createTestUser(t, ctx, users, "+1783"+suffix+"71", "FlagOwner", "")
member := createTestUser(t, ctx, users, "+1783"+suffix+"72", "FlagMember", "")
unrelated := createTestUser(t, ctx, users, "+1783"+suffix+"73", "FlagUnrelated", "")
created, err := channels.CreateChannel(ctx, domain.CreateChannelRequest{
CreatorUserID: owner.ID,
MemberUserIDs: []int64{member.ID},
Title: "Flagged Channel " + suffix,
Megagroup: true,
Date: 1700003000,
})
if err != nil {
t.Fatalf("create channel: %v", err)
}
baseline := make(map[int64]int)
for _, viewer := range []domain.User{owner, member, unrelated} {
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 := channels.SetChannelScamFake(ctx, created.Channel.ID, true, false)
if err != nil {
t.Fatalf("set channel scam: %v", err)
}
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)
}
}
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
if err := pool.QueryRow(ctx, `
SELECT count(*)
FROM dispatch_outbox
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 {
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)
}
}
}

4
internal/store/postgres/star_gift_lifecycle_migration_integration_test.go
View file

@ -14,7 +14,7 @@ func TestStarGiftLifecycleMigrationsApply(t *testing.T) {
if err != nil {
t.Fatalf("migrate star gift lifecycle schema: %v", err)
}
if status.Dirty || status.Empty || status.Version != 145 {
t.Fatalf("migration status = %+v, want clean version 145", status)
if status.Dirty || status.Empty || status.Version != 147 {
t.Fatalf("migration status = %+v, want clean version 147", status)
}
}

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

@ -341,7 +341,46 @@ func (s *UserStore) SetScamFake(ctx context.Context, userID int64, scam, fake bo
if scam && fake {
return domain.User{}, domain.ErrPeerModerationFlagsInvalid
}
row, err := s.q.SetUserScamFake(ctx, sqlcgen.SetUserScamFakeParams{
beginner, ok := s.db.(txBeginner)
if !ok {
return domain.User{}, fmt.Errorf("set user scam/fake: db does not support transactions")
}
tx, err := beginner.Begin(ctx)
if err != nil {
return domain.User{}, fmt.Errorf("begin set user scam/fake: %w", err)
}
committed := false
defer func() {
if !committed {
_ = tx.Rollback(ctx)
}
}()
qtx := s.q.WithTx(tx)
var currentScam, currentFake bool
if err := tx.QueryRow(ctx, `
SELECT scam, fake
FROM users
WHERE id = $1
FOR UPDATE`, userID).Scan(&currentScam, &currentFake); err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return domain.User{}, domain.ErrUserNotFound
}
return domain.User{}, fmt.Errorf("lock user scam/fake: %w", err)
}
if currentScam == scam && currentFake == fake {
row, err := qtx.GetUserByID(ctx, userID)
if err != nil {
return domain.User{}, fmt.Errorf("reload unchanged user scam/fake: %w", err)
}
if err := tx.Commit(ctx); err != nil {
return domain.User{}, fmt.Errorf("commit unchanged user scam/fake: %w", err)
}
committed = true
return userFromModel(row), nil
}
row, err := qtx.SetUserScamFake(ctx, sqlcgen.SetUserScamFakeParams{
ID: userID,
Scam: scam,
Fake: fake,
@ -352,9 +391,91 @@ func (s *UserStore) SetScamFake(ctx context.Context, userID int64, scam, fake bo
}
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)
}
committed = true
return userFromModel(row), 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.
func moderationFlagAudience(ctx context.Context, db sqlcgen.DBTX, userID int64, limit int) ([]int64, error) {
if userID <= 0 || limit <= 0 {
return nil, nil
}
rows, err := db.Query(ctx, `
SELECT picked.user_id
FROM (
SELECT candidates.user_id
FROM (
SELECT $1::bigint AS user_id, 0 AS priority, 2147483647::bigint AS activity
UNION ALL
SELECT contact_user_id, 1, 0 FROM contacts WHERE user_id = $1
UNION ALL
SELECT user_id, 1, 0 FROM contacts WHERE contact_user_id = $1
UNION ALL
SELECT peer_id, 2, top_message_date FROM dialogs WHERE user_id = $1 AND peer_type = 'user'
UNION ALL
SELECT user_id, 2, top_message_date FROM dialogs WHERE peer_type = 'user' AND peer_id = $1
) candidates
JOIN users u ON u.id = candidates.user_id AND u.deleted_at IS NULL
GROUP BY candidates.user_id
ORDER BY min(candidates.priority), max(candidates.activity) DESC, candidates.user_id
LIMIT $2
) picked
ORDER BY picked.user_id`, userID, limit)
if err != nil {
return nil, fmt.Errorf("list moderation flag audience: %w", err)
}
defer rows.Close()
out := make([]int64, 0)
for rows.Next() {
var id int64
if err := rows.Scan(&id); err != nil {
return nil, fmt.Errorf("scan moderation flag audience: %w", err)
}
if id != 0 {
out = append(out, id)
}
}
if err := rows.Err(); err != nil {
return nil, fmt.Errorf("iterate moderation flag audience: %w", err)
}
return out, nil
}
// SweepExpiredPremium 清空到期会员行并返回清理后的用户。
func (s *UserStore) SweepExpiredPremium(ctx context.Context, now int64, limit int) ([]domain.User, error) {
if limit <= 0 {