From 3dd9c345d716e61b4723a23c79c27e207fabe9cd Mon Sep 17 00:00:00 2001 From: iamxvbaba <28732408+iamxvbaba@users.noreply.github.com> Date: Fri, 24 Jul 2026 11:57:00 +0800 Subject: [PATCH] fix: sync moderation flag convergence --- ...47_user_moderation_profile_events.down.sql | 26 +++ ...0147_user_moderation_profile_events.up.sql | 19 ++ internal/domain/update_event.go | 6 + internal/rpc/convert_updates.go | 5 + .../rpc/moderation_profile_update_test.go | 97 +++++++++++ internal/rpc/outbox_dispatcher.go | 22 +++ internal/store/postgres/channel_settings.go | 103 ++++++++++- .../moderation_flags_integration_test.go | 164 ++++++++++++++++++ ...ft_lifecycle_migration_integration_test.go | 4 +- internal/store/postgres/user.go | 123 ++++++++++++- 10 files changed, 564 insertions(+), 5 deletions(-) create mode 100644 deploy/migrations/0147_user_moderation_profile_events.down.sql create mode 100644 deploy/migrations/0147_user_moderation_profile_events.up.sql create mode 100644 internal/rpc/moderation_profile_update_test.go diff --git a/deploy/migrations/0147_user_moderation_profile_events.down.sql b/deploy/migrations/0147_user_moderation_profile_events.down.sql new file mode 100644 index 00000000..53f0a3dd --- /dev/null +++ b/deploy/migrations/0147_user_moderation_profile_events.down.sql @@ -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[]) +); diff --git a/deploy/migrations/0147_user_moderation_profile_events.up.sql b/deploy/migrations/0147_user_moderation_profile_events.up.sql new file mode 100644 index 00000000..b8527dd2 --- /dev/null +++ b/deploy/migrations/0147_user_moderation_profile_events.up.sql @@ -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[]) +); diff --git a/internal/domain/update_event.go b/internal/domain/update_event.go index bf72007a..65361dec 100644 --- a/internal/domain/update_event.go +++ b/internal/domain/update_event.go @@ -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, diff --git a/internal/rpc/convert_updates.go b/internal/rpc/convert_updates.go index d187cfc3..6b36eac1 100644 --- a/internal/rpc/convert_updates.go +++ b/internal/rpc/convert_updates.go @@ -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 diff --git a/internal/rpc/moderation_profile_update_test.go b/internal/rpc/moderation_profile_update_test.go new file mode 100644 index 00000000..aec28b7b --- /dev/null +++ b/internal/rpc/moderation_profile_update_test.go @@ -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]) + } +} diff --git a/internal/rpc/outbox_dispatcher.go b/internal/rpc/outbox_dispatcher.go index e0e12af9..fa9dab6a 100644 --- a/internal/rpc/outbox_dispatcher.go +++ b/internal/rpc/outbox_dispatcher.go @@ -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: diff --git a/internal/store/postgres/channel_settings.go b/internal/store/postgres/channel_settings.go index 733152e8..b9ecd267 100644 --- a/internal/store/postgres/channel_settings.go +++ b/internal/store/postgres/channel_settings.go @@ -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(¤tScam, ¤tFake); 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) { diff --git a/internal/store/postgres/moderation_flags_integration_test.go b/internal/store/postgres/moderation_flags_integration_test.go index 5c007c75..516a6238 100644 --- a/internal/store/postgres/moderation_flags_integration_test.go +++ b/internal/store/postgres/moderation_flags_integration_test.go @@ -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) + } + } +} diff --git a/internal/store/postgres/star_gift_lifecycle_migration_integration_test.go b/internal/store/postgres/star_gift_lifecycle_migration_integration_test.go index 4a72ba9e..05576a5d 100644 --- a/internal/store/postgres/star_gift_lifecycle_migration_integration_test.go +++ b/internal/store/postgres/star_gift_lifecycle_migration_integration_test.go @@ -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) } } diff --git a/internal/store/postgres/user.go b/internal/store/postgres/user.go index fbea492f..c3fa169f 100644 --- a/internal/store/postgres/user.go +++ b/internal/store/postgres/user.go @@ -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(¤tScam, ¤tFake); 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 {