From 73cf6a81847b172e5deba1e8f79594567b0f67ee Mon Sep 17 00:00:00 2001 From: iamxvbaba <28732408+iamxvbaba@users.noreply.github.com> Date: Wed, 29 Jul 2026 16:19:04 +0800 Subject: [PATCH] fix(channels): sync preserve locally cleared dialogs --- deploy/migrations/0001_init.up.sql | 2 +- ...er_update_events_discussion_types.down.sql | 2 +- ...user_update_events_discussion_types.up.sql | 2 +- .../0015_user_update_events_web_page.down.sql | 2 +- .../0015_user_update_events_web_page.up.sql | 2 +- .../0061_user_phone_change_update.down.sql | 2 +- .../0061_user_phone_change_update.up.sql | 2 +- .../0114_collectible_emoji_status.down.sql | 2 +- .../0114_collectible_emoji_status.up.sql | 2 +- ...47_user_moderation_profile_events.down.sql | 2 +- ...0147_user_moderation_profile_events.up.sql | 2 +- ...0161_channel_history_clear_anchor.down.sql | 4 + .../0161_channel_history_clear_anchor.up.sql | 14 ++ ...62_channel_history_clear_recovery.down.sql | 7 + ...0162_channel_history_clear_recovery.up.sql | 22 ++ internal/app/channels/service.go | 11 +- internal/app/dialogs/service.go | 34 +-- internal/app/updates/service.go | 10 - internal/domain/channel.go | 148 +++++++---- internal/domain/dialog.go | 41 +-- internal/domain/update_event.go | 11 +- internal/rpc/channels_messages.go | 12 +- internal/rpc/channels_messages_rpc_test.go | 149 +++++++++-- internal/rpc/channels_stubs.go | 21 -- internal/rpc/convert_channels_core.go | 2 + internal/rpc/convert_updates.go | 36 ++- internal/rpc/deps.go | 1 - internal/rpc/messages_delete.go | 13 +- internal/rpc/messages_delete_rpc_test.go | 63 +++++ internal/rpc/messages_register.go | 17 +- internal/rpc/rpc_testkit_updates_test.go | 9 - internal/rpc/updates.go | 26 +- internal/store/memory/channel_dialogs.go | 63 +++-- internal/store/memory/channel_helpers.go | 24 +- internal/store/memory/channel_members.go | 2 + .../store/memory/channel_message_delete.go | 28 ++- .../store/memory/channel_message_history.go | 113 ++++++++- internal/store/memory/channel_store.go | 22 +- internal/store/memory/channel_test.go | 178 ++++++++++++- internal/store/postgres/channel_core.go | 4 +- internal/store/postgres/channel_dialogs.go | 238 +++++++++++++----- internal/store/postgres/channel_helpers.go | 53 +++- .../store/postgres/channel_member_helpers.go | 20 +- .../store/postgres/channel_member_join.go | 1 + .../store/postgres/channel_member_list.go | 7 +- .../store/postgres/channel_message_delete.go | 64 ++++- .../store/postgres/channel_message_history.go | 172 ++++++++++++- .../channel_message_integration_test.go | 180 ++++++++++++- .../postgres/channel_read_integration_test.go | 6 +- ...ft_lifecycle_migration_integration_test.go | 4 +- 50 files changed, 1525 insertions(+), 327 deletions(-) create mode 100644 deploy/migrations/0161_channel_history_clear_anchor.down.sql create mode 100644 deploy/migrations/0161_channel_history_clear_anchor.up.sql create mode 100644 deploy/migrations/0162_channel_history_clear_recovery.down.sql create mode 100644 deploy/migrations/0162_channel_history_clear_recovery.up.sql diff --git a/deploy/migrations/0001_init.up.sql b/deploy/migrations/0001_init.up.sql index bb0d8541..2ef5e6b0 100644 --- a/deploy/migrations/0001_init.up.sql +++ b/deploy/migrations/0001_init.up.sql @@ -2948,7 +2948,7 @@ CREATE TABLE public.user_update_events ( story_payload jsonb DEFAULT '{}'::jsonb NOT NULL, reaction_payload jsonb DEFAULT '{}'::jsonb NOT NULL, CONSTRAINT user_update_events_peer_type_check CHECK (((peer_type IS NULL) OR ((peer_type)::text = ANY (ARRAY[('user'::character varying)::text, ('channel'::character varying)::text])))), - CONSTRAINT user_update_events_type_check CHECK (((event_type)::text = ANY (ARRAY[('new_message'::character varying)::text, ('read_history_inbox'::character varying)::text, ('read_history_outbox'::character varying)::text, ('read_message_contents'::character varying)::text, ('edit_message'::character varying)::text, ('message_poll'::character varying)::text, ('draft_message'::character varying)::text, ('quick_replies'::character varying)::text, ('new_quick_reply'::character varying)::text, ('delete_quick_reply'::character varying)::text, ('quick_reply_message'::character varying)::text, ('delete_quick_reply_messages'::character varying)::text, ('contacts_reset'::character varying)::text, ('dialog_pinned'::character varying)::text, ('pinned_dialogs'::character varying)::text, ('pinned_messages'::character varying)::text, ('dialog_unread_mark'::character varying)::text, ('peer_settings'::character varying)::text, ('peer_story_blocked'::character varying)::text, ('delete_messages'::character varying)::text, ('dialog_filter'::character varying)::text, ('dialog_filter_order'::character varying)::text, ('dialog_filters'::character varying)::text, ('folder_peers'::character varying)::text, ('channel_available_messages'::character varying)::text, ('channel_view_forum_as_messages'::character varying)::text, ('channel_state'::character varying)::text, ('saved_dialog_pinned'::character varying)::text, ('pinned_saved_dialogs'::character varying)::text, ('story'::character varying)::text, ('read_stories'::character varying)::text, ('sent_story_reaction'::character varying)::text, ('new_story_reaction'::character varying)::text, ('noop'::character varying)::text]))) + CONSTRAINT user_update_events_type_check CHECK (((event_type)::text = ANY (ARRAY[('new_message'::character varying)::text, ('read_history_inbox'::character varying)::text, ('read_history_outbox'::character varying)::text, ('read_message_contents'::character varying)::text, ('edit_message'::character varying)::text, ('message_poll'::character varying)::text, ('draft_message'::character varying)::text, ('quick_replies'::character varying)::text, ('new_quick_reply'::character varying)::text, ('delete_quick_reply'::character varying)::text, ('quick_reply_message'::character varying)::text, ('delete_quick_reply_messages'::character varying)::text, ('contacts_reset'::character varying)::text, ('dialog_pinned'::character varying)::text, ('pinned_dialogs'::character varying)::text, ('pinned_messages'::character varying)::text, ('dialog_unread_mark'::character varying)::text, ('peer_settings'::character varying)::text, ('peer_story_blocked'::character varying)::text, ('delete_messages'::character varying)::text, ('dialog_filter'::character varying)::text, ('dialog_filter_order'::character varying)::text, ('dialog_filters'::character varying)::text, ('folder_peers'::character varying)::text, ('channel_view_forum_as_messages'::character varying)::text, ('channel_state'::character varying)::text, ('saved_dialog_pinned'::character varying)::text, ('pinned_saved_dialogs'::character varying)::text, ('story'::character varying)::text, ('read_stories'::character varying)::text, ('sent_story_reaction'::character varying)::text, ('new_story_reaction'::character varying)::text, ('noop'::character varying)::text]))) ); diff --git a/deploy/migrations/0003_user_update_events_discussion_types.down.sql b/deploy/migrations/0003_user_update_events_discussion_types.down.sql index 1f4d3113..acf8fbb0 100644 --- a/deploy/migrations/0003_user_update_events_discussion_types.down.sql +++ b/deploy/migrations/0003_user_update_events_discussion_types.down.sql @@ -7,7 +7,7 @@ ALTER TABLE public.user_update_events ADD CONSTRAINT user_update_events_type_che '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', 'delete_messages', 'dialog_filter', 'dialog_filter_order', - 'dialog_filters', 'folder_peers', 'channel_available_messages', 'channel_view_forum_as_messages', + 'dialog_filters', 'folder_peers', 'channel_view_forum_as_messages', 'channel_state', 'saved_dialog_pinned', 'pinned_saved_dialogs', 'story', 'read_stories', 'sent_story_reaction', 'new_story_reaction', 'noop' ]::text[]) diff --git a/deploy/migrations/0003_user_update_events_discussion_types.up.sql b/deploy/migrations/0003_user_update_events_discussion_types.up.sql index 9e8d7d6e..1edd0a35 100644 --- a/deploy/migrations/0003_user_update_events_discussion_types.up.sql +++ b/deploy/migrations/0003_user_update_events_discussion_types.up.sql @@ -12,7 +12,7 @@ ALTER TABLE public.user_update_events ADD CONSTRAINT user_update_events_type_che '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', 'delete_messages', 'dialog_filter', 'dialog_filter_order', - 'dialog_filters', 'folder_peers', 'channel_available_messages', 'channel_view_forum_as_messages', + 'dialog_filters', 'folder_peers', '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' diff --git a/deploy/migrations/0015_user_update_events_web_page.down.sql b/deploy/migrations/0015_user_update_events_web_page.down.sql index ee8842ec..6a721f53 100644 --- a/deploy/migrations/0015_user_update_events_web_page.down.sql +++ b/deploy/migrations/0015_user_update_events_web_page.down.sql @@ -7,7 +7,7 @@ ALTER TABLE public.user_update_events ADD CONSTRAINT user_update_events_type_che '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', 'delete_messages', 'dialog_filter', 'dialog_filter_order', - 'dialog_filters', 'folder_peers', 'channel_available_messages', 'channel_view_forum_as_messages', + 'dialog_filters', 'folder_peers', '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' diff --git a/deploy/migrations/0015_user_update_events_web_page.up.sql b/deploy/migrations/0015_user_update_events_web_page.up.sql index 0187d6f7..bea20414 100644 --- a/deploy/migrations/0015_user_update_events_web_page.up.sql +++ b/deploy/migrations/0015_user_update_events_web_page.up.sql @@ -10,7 +10,7 @@ ALTER TABLE public.user_update_events ADD CONSTRAINT user_update_events_type_che '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', 'delete_messages', 'dialog_filter', 'dialog_filter_order', - 'dialog_filters', 'folder_peers', 'channel_available_messages', 'channel_view_forum_as_messages', + 'dialog_filters', 'folder_peers', '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' diff --git a/deploy/migrations/0061_user_phone_change_update.down.sql b/deploy/migrations/0061_user_phone_change_update.down.sql index cca70746..eb3c24df 100644 --- a/deploy/migrations/0061_user_phone_change_update.down.sql +++ b/deploy/migrations/0061_user_phone_change_update.down.sql @@ -6,7 +6,7 @@ ALTER TABLE public.user_update_events ADD CONSTRAINT user_update_events_type_che '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', 'delete_messages', 'dialog_filter', 'dialog_filter_order', - 'dialog_filters', 'folder_peers', 'channel_available_messages', 'channel_view_forum_as_messages', + 'dialog_filters', 'folder_peers', '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' diff --git a/deploy/migrations/0061_user_phone_change_update.up.sql b/deploy/migrations/0061_user_phone_change_update.up.sql index 130a719c..83841569 100644 --- a/deploy/migrations/0061_user_phone_change_update.up.sql +++ b/deploy/migrations/0061_user_phone_change_update.up.sql @@ -9,7 +9,7 @@ ALTER TABLE public.user_update_events ADD CONSTRAINT user_update_events_type_che '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', 'delete_messages', 'dialog_filter', - 'dialog_filter_order', 'dialog_filters', 'folder_peers', 'channel_available_messages', + 'dialog_filter_order', 'dialog_filters', 'folder_peers', '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', diff --git a/deploy/migrations/0114_collectible_emoji_status.down.sql b/deploy/migrations/0114_collectible_emoji_status.down.sql index ea3bb7ec..f900feb6 100644 --- a/deploy/migrations/0114_collectible_emoji_status.down.sql +++ b/deploy/migrations/0114_collectible_emoji_status.down.sql @@ -23,7 +23,7 @@ ALTER TABLE public.user_update_events ADD CONSTRAINT user_update_events_type_che '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', 'delete_messages', 'dialog_filter', - 'dialog_filter_order', 'dialog_filters', 'folder_peers', 'channel_available_messages', + 'dialog_filter_order', 'dialog_filters', 'folder_peers', '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', diff --git a/deploy/migrations/0114_collectible_emoji_status.up.sql b/deploy/migrations/0114_collectible_emoji_status.up.sql index 58a7927e..cd17f6db 100644 --- a/deploy/migrations/0114_collectible_emoji_status.up.sql +++ b/deploy/migrations/0114_collectible_emoji_status.up.sql @@ -50,7 +50,7 @@ ALTER TABLE public.user_update_events ADD CONSTRAINT user_update_events_type_che 'contacts_reset', 'dialog_pinned', 'pinned_dialogs', 'pinned_messages', 'dialog_unread_mark', 'peer_settings', 'peer_story_blocked', 'user_phone', 'user_emoji_status', 'delete_messages', 'dialog_filter', 'dialog_filter_order', 'dialog_filters', 'folder_peers', - 'channel_available_messages', 'channel_view_forum_as_messages', 'channel_state', + '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' diff --git a/deploy/migrations/0147_user_moderation_profile_events.down.sql b/deploy/migrations/0147_user_moderation_profile_events.down.sql index dfe77ed5..d10fbfe0 100644 --- a/deploy/migrations/0147_user_moderation_profile_events.down.sql +++ b/deploy/migrations/0147_user_moderation_profile_events.down.sql @@ -9,7 +9,7 @@ ALTER TABLE public.user_update_events ADD CONSTRAINT user_update_events_type_che 'contacts_reset', 'dialog_pinned', 'pinned_dialogs', 'pinned_messages', 'dialog_unread_mark', 'peer_settings', 'peer_story_blocked', 'user_phone', 'user_emoji_status', 'delete_messages', 'dialog_filter', 'dialog_filter_order', 'dialog_filters', 'folder_peers', - 'channel_available_messages', 'channel_view_forum_as_messages', 'channel_state', + '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' diff --git a/deploy/migrations/0147_user_moderation_profile_events.up.sql b/deploy/migrations/0147_user_moderation_profile_events.up.sql index d141788e..2166a8d1 100644 --- a/deploy/migrations/0147_user_moderation_profile_events.up.sql +++ b/deploy/migrations/0147_user_moderation_profile_events.up.sql @@ -12,7 +12,7 @@ ALTER TABLE public.user_update_events ADD CONSTRAINT user_update_events_type_che 'contacts_reset', 'dialog_pinned', 'pinned_dialogs', 'pinned_messages', 'dialog_unread_mark', 'peer_settings', 'peer_story_blocked', 'user_phone', 'user_emoji_status', 'delete_messages', 'dialog_filter', 'dialog_filter_order', 'dialog_filters', 'folder_peers', - 'channel_available_messages', 'channel_view_forum_as_messages', 'channel_state', + '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' diff --git a/deploy/migrations/0161_channel_history_clear_anchor.down.sql b/deploy/migrations/0161_channel_history_clear_anchor.down.sql new file mode 100644 index 00000000..42bd5e1e --- /dev/null +++ b/deploy/migrations/0161_channel_history_clear_anchor.down.sql @@ -0,0 +1,4 @@ +ALTER TABLE channel_members + DROP CONSTRAINT IF EXISTS channel_members_history_clear_anchor_check, + DROP COLUMN IF EXISTS history_clear_anchor_date, + DROP COLUMN IF EXISTS history_clear_anchor_id; diff --git a/deploy/migrations/0161_channel_history_clear_anchor.up.sql b/deploy/migrations/0161_channel_history_clear_anchor.up.sql new file mode 100644 index 00000000..b82534a6 --- /dev/null +++ b/deploy/migrations/0161_channel_history_clear_anchor.up.sql @@ -0,0 +1,14 @@ +ALTER TABLE channel_members + ADD COLUMN history_clear_anchor_id integer DEFAULT 0 NOT NULL, + ADD COLUMN history_clear_anchor_date integer DEFAULT 0 NOT NULL, + ADD CONSTRAINT channel_members_history_clear_anchor_check CHECK ( + ( + history_clear_anchor_id = 0 + AND history_clear_anchor_date = 0 + ) + OR ( + history_clear_anchor_id > 0 + AND history_clear_anchor_date > 0 + AND history_clear_anchor_id <= available_min_id + ) + ); diff --git a/deploy/migrations/0162_channel_history_clear_recovery.down.sql b/deploy/migrations/0162_channel_history_clear_recovery.down.sql new file mode 100644 index 00000000..d096648f --- /dev/null +++ b/deploy/migrations/0162_channel_history_clear_recovery.down.sql @@ -0,0 +1,7 @@ +DROP INDEX IF EXISTS user_channel_member_index_history_clear_idx; + +ALTER TABLE user_channel_member_index + DROP CONSTRAINT IF EXISTS user_channel_member_index_history_clear_check, + DROP COLUMN IF EXISTS history_clear_updated_at, + DROP COLUMN IF EXISTS history_clear_anchor_id, + DROP COLUMN IF EXISTS available_min_id; diff --git a/deploy/migrations/0162_channel_history_clear_recovery.up.sql b/deploy/migrations/0162_channel_history_clear_recovery.up.sql new file mode 100644 index 00000000..7da43c7d --- /dev/null +++ b/deploy/migrations/0162_channel_history_clear_recovery.up.sql @@ -0,0 +1,22 @@ +ALTER TABLE user_channel_member_index + ADD COLUMN available_min_id integer DEFAULT 0 NOT NULL, + ADD COLUMN history_clear_anchor_id integer DEFAULT 0 NOT NULL, + ADD COLUMN history_clear_updated_at integer DEFAULT 0 NOT NULL, + ADD CONSTRAINT user_channel_member_index_history_clear_check CHECK ( + ( + history_clear_anchor_id = 0 + AND history_clear_updated_at = 0 + ) + OR ( + history_clear_anchor_id > 0 + AND history_clear_updated_at > 0 + AND history_clear_anchor_id <= available_min_id + ) + ); + +CREATE INDEX user_channel_member_index_history_clear_idx + ON user_channel_member_index (user_id, channel_id) + INCLUDE (available_min_id, history_clear_updated_at) + WHERE status = 'active' + AND NOT deleted + AND history_clear_anchor_id > 0; diff --git a/internal/app/channels/service.go b/internal/app/channels/service.go index f79d1ccf..6d527ff6 100644 --- a/internal/app/channels/service.go +++ b/internal/app/channels/service.go @@ -2301,7 +2301,16 @@ func (s *Service) GetDifference(ctx context.Context, userID int64, req domain.Ch if err != nil { return domain.ChannelDifference{}, err } - return s.filterBotChannelDifference(ctx, userID, diff), nil + diff = s.filterBotChannelDifference(ctx, userID, diff) + if diff.Self.HistoryClearAnchorID > 0 && + diff.Self.HistoryClearAnchorID == diff.Self.AvailableMinID && + !diff.TooLong { + // updateChannelAvailableMessages is an absolute no-PTS update. Keep it + // outside ChannelUpdateEvent so channel difference continuity remains + // defined solely by real channel events. + diff.AvailableMinID = diff.Self.AvailableMinID + } + return diff, nil } // ClearDanglingPinnedMessage 清除指向已删除消息的悬挂置顶值（unpinAll 自愈）。 diff --git a/internal/app/dialogs/service.go b/internal/app/dialogs/service.go index b28b8c2d..53f5d2ae 100644 --- a/internal/app/dialogs/service.go +++ b/internal/app/dialogs/service.go @@ -408,22 +408,24 @@ func isChannelPreviewAccessError(err error) bool { func dialogFromChannelView(view domain.ChannelView) domain.Dialog { dialog := view.Dialog return domain.Dialog{ - Peer: domain.Peer{Type: domain.PeerTypeChannel, ID: dialog.ChannelID}, - ChannelLeft: view.Self.Status == domain.ChannelMemberLeft, - FolderID: dialog.FolderID, - TopMessage: dialog.TopMessageID, - TopMessageDate: dialog.TopMessageDate, - ReadInboxMaxID: dialog.ReadInboxMaxID, - ReadOutboxMaxID: dialog.ReadOutboxMaxID, - UnreadCount: dialog.UnreadCount, - UnreadMentions: dialog.UnreadMentions, - UnreadReactions: dialog.UnreadReactions, - Pinned: dialog.Pinned, - PinnedOrder: dialog.PinnedOrder, - UnreadMark: dialog.UnreadMark, - ViewForumAsMessages: dialog.ViewForumAsMessages, - HasScheduled: dialog.HasScheduled, - Pts: view.Channel.Pts, + Peer: domain.Peer{Type: domain.PeerTypeChannel, ID: dialog.ChannelID}, + ChannelLeft: view.Self.Status == domain.ChannelMemberLeft, + FolderID: dialog.FolderID, + TopMessage: dialog.TopMessageID, + TopMessageDate: dialog.TopMessageDate, + HistoryClearAnchorID: dialog.HistoryClearAnchorID, + HistoryClearAnchorDate: dialog.HistoryClearAnchorDate, + ReadInboxMaxID: dialog.ReadInboxMaxID, + ReadOutboxMaxID: dialog.ReadOutboxMaxID, + UnreadCount: dialog.UnreadCount, + UnreadMentions: dialog.UnreadMentions, + UnreadReactions: dialog.UnreadReactions, + Pinned: dialog.Pinned, + PinnedOrder: dialog.PinnedOrder, + UnreadMark: dialog.UnreadMark, + ViewForumAsMessages: dialog.ViewForumAsMessages, + HasScheduled: dialog.HasScheduled, + Pts: view.Channel.Pts, } } diff --git a/internal/app/updates/service.go b/internal/app/updates/service.go index 3817a34b..66c06d37 100644 --- a/internal/app/updates/service.go +++ b/internal/app/updates/service.go @@ -731,16 +731,6 @@ func (s *Service) RecordFolderPeers(ctx context.Context, stateAuthKeyID [8]byte, }, true, excludeSessionID) } -// RecordChannelAvailableMessages records a local channel history clear for multi-device sync. -func (s *Service) RecordChannelAvailableMessages(ctx context.Context, stateAuthKeyID [8]byte, userID, channelID int64, availableMinID int, excludeAuthKeyID [8]byte, excludeSessionID int64) (domain.UpdateEvent, domain.UpdateState, error) { - return s.recordEvent(ctx, stateAuthKeyID, excludeAuthKeyID, userID, domain.UpdateEvent{ - Type: domain.UpdateEventChannelAvailable, - Peer: domain.Peer{Type: domain.PeerTypeChannel, ID: channelID}, - MaxID: availableMinID, - PtsCount: 1, - }, true, excludeSessionID) -} - func (s *Service) recordEvent(ctx context.Context, stateAuthKeyID, excludeAuthKeyID [8]byte, userID int64, event domain.UpdateEvent, dispatch bool, excludeSessionID int64) (domain.UpdateEvent, domain.UpdateState, error) { return s.recordEventCore(ctx, stateAuthKeyID, excludeAuthKeyID, userID, event, dispatch, excludeSessionID, true) } diff --git a/internal/domain/channel.go b/internal/domain/channel.go index 328dc3a0..36734100 100644 --- a/internal/domain/channel.go +++ b/internal/domain/channel.go @@ -485,23 +485,29 @@ func (c Channel) MembersListAdminOnly() bool { // ChannelMember is one user's channel membership and read state. type ChannelMember struct { - ChannelID int64 - UserID int64 - InviterUserID int64 - Role ChannelMemberRole - Status ChannelMemberStatus - JoinedAt int - LeftAt int - AdminRights ChannelAdminRights - BannedRights ChannelBannedRights - Rank string - AvailableMinID int - AvailableMinPts int - ReadInboxMaxID int - ReadInboxDate int - ReadOutboxMaxID int - UnreadMark bool - SlowmodeLastSendDate int + ChannelID int64 + UserID int64 + InviterUserID int64 + Role ChannelMemberRole + Status ChannelMemberStatus + JoinedAt int + LeftAt int + AdminRights ChannelAdminRights + BannedRights ChannelBannedRights + Rank string + AvailableMinID int + AvailableMinPts int + // HistoryClearAnchorID/Date identify an owner-local channels.deleteHistory + // boundary. They are deliberately separate from AvailableMinID: joining a + // hidden-prehistory channel also advances AvailableMinID but must not + // manufacture a "History cleared" service message. + HistoryClearAnchorID int + HistoryClearAnchorDate int + ReadInboxMaxID int + ReadInboxDate int + ReadOutboxMaxID int + UnreadMark bool + SlowmodeLastSendDate int // Guest is a computed, non-persisted view used for subscribers accessing a // private linked discussion group without joining it. Guest must never be // written to channel_members and is still projected to clients as left. @@ -545,22 +551,24 @@ func (m ChannelMember) CanPostChannelMessages() bool { // ChannelDialog is the current user's owner-view dialog state for a channel. type ChannelDialog struct { - UserID int64 - ChannelID int64 - FolderID int - TopMessageID int - TopMessageDate int - ReadInboxMaxID int - ReadOutboxMaxID int - UnreadCount int - UnreadMentions int - UnreadReactions int - Pinned bool - PinnedOrder int - UnreadMark bool - ViewForumAsMessages bool - HasScheduled bool - DefaultSendAs *Peer + UserID int64 + ChannelID int64 + FolderID int + TopMessageID int + TopMessageDate int + HistoryClearAnchorID int + HistoryClearAnchorDate int + ReadInboxMaxID int + ReadOutboxMaxID int + UnreadCount int + UnreadMentions int + UnreadReactions int + Pinned bool + PinnedOrder int + UnreadMark bool + ViewForumAsMessages bool + HasScheduled bool + DefaultSendAs *Peer } // ChannelMessageActionType identifies service messages generated by channel operations. @@ -573,7 +581,11 @@ const ( ChannelActionChatDelete ChannelMessageActionType = "chat_delete_user" ChannelActionChatEditPhoto ChannelMessageActionType = "chat_edit_photo" ChannelActionChatDeletePhoto ChannelMessageActionType = "chat_delete_photo" - ChannelActionChatJoined ChannelMessageActionType = "chat_joined" + // ChannelActionHistoryClear is an owner-local projection at the member's + // channels.deleteHistory boundary. It is never persisted into the shared + // channel_messages row and never produces channel PTS. + ChannelActionHistoryClear ChannelMessageActionType = "history_clear" + ChannelActionChatJoined ChannelMessageActionType = "chat_joined" // ChannelActionChatJoinedByLink 是经邀请链接加入的服务消息， // 渲染为 "X joined the group via invite link"。 ChannelActionChatJoinedByLink ChannelMessageActionType = "chat_joined_by_link" @@ -710,6 +722,39 @@ type ChannelMessage struct { Deleted bool } +// ProjectChannelHistoryClearMessage returns the owner-local service-message +// projection for one channel history boundary. Identity fields from the shared +// source are retained when available, while all user payload, media, reply, +// reaction, pin, TTL and edit state is removed. A zero source is valid for a +// retained anchor whose shared row has already been physically pruned. +func ProjectChannelHistoryClearMessage(source ChannelMessage, channelID int64, messageID, messageDate int) ChannelMessage { + if source.ChannelID == 0 { + source.ChannelID = channelID + } + if source.ID == 0 { + source.ID = messageID + } + if source.Date == 0 { + source.Date = messageDate + } + return ChannelMessage{ + ChannelID: source.ChannelID, + ID: source.ID, + SenderUserID: source.SenderUserID, + From: source.From, + Date: source.Date, + Post: source.Post, + Silent: source.Silent, + Action: &ChannelMessageAction{Type: ChannelActionHistoryClear}, + } +} + +// IsChannelHistoryClearMessage reports whether msg is an owner-local history +// clear projection rather than a shared channel service message. +func IsChannelHistoryClearMessage(msg ChannelMessage) bool { + return msg.Action != nil && msg.Action.Type == ChannelActionHistoryClear +} + // MessageReactionType identifies one stored reaction constructor without depending on TL types. type MessageReactionType string @@ -1290,19 +1335,27 @@ type ChannelDifference struct { Events []ChannelUpdateEvent NewMessages []ChannelMessage OtherUpdates []ChannelUpdateEvent - Users []User - Channels []Channel - Pts int - Final bool - TooLong bool - Dialog ChannelDialog - Timeout int + // AvailableMinID is an owner-local, absolute no-PTS boundary appended to a + // normal channel difference. It must never be represented as a synthetic + // ChannelUpdateEvent because updateChannelAvailableMessages has no pts. + AvailableMinID int + Users []User + Channels []Channel + Pts int + Final bool + TooLong bool + Dialog ChannelDialog + Timeout int } -// DirtyChannel identifies an active channel with channel-scoped updates after an account difference date. +// DirtyChannel identifies an active channel with shared channel updates and/or +// an owner-local no-PTS history-clear boundary after an account difference date. type DirtyChannel struct { - ChannelID int64 - Pts int + ChannelID int64 + Pts int + ChannelUpdatesDirty bool + AvailableMinID int + HistoryClearDate int } // ChannelUpdateRetentionCheckpoint is the durable recovery boundary for one channel. @@ -1980,6 +2033,9 @@ type DeleteChannelHistoryResult struct { Recipients []int64 Offset int AvailableMinID int + // AvailableMinChanged is true only when this owner-local request advanced + // the member boundary and installed a new history-clear anchor. + AvailableMinChanged bool } // UpdateChannelPinnedMessageRequest pins or unpins one channel/supergroup message. @@ -2192,6 +2248,10 @@ type ChannelHistoryFilter struct { SenderUserID int64 PinnedOnly bool MusicOnly bool + // IncludeHistoryClearAnchor is set only by messages.getHistory. Search, + // media and topic projections must not count the owner-local service marker + // as shared channel content. + IncludeHistoryClearAnchor bool // NeedTotalCount requests the exact number of messages matching the static // filters before offset/add_offset pagination. Ordinary history pages leave // this false and keep the bounded len(page)+has-more hint. diff --git a/internal/domain/dialog.go b/internal/domain/dialog.go index 96cc5a73..cc6607c9 100644 --- a/internal/domain/dialog.go +++ b/internal/domain/dialog.go @@ -69,24 +69,29 @@ func (p Peer) IsSelfUser(userID int64) bool { // Dialog 是账号的一条会话摘要。 type Dialog struct { - Peer Peer - ChannelLeft bool - FolderID int - TopMessage int - TopMessageDate int - ReadInboxMaxID int - ReadOutboxMaxID int - UnreadCount int - UnreadMentions int - UnreadReactions int - TTLPeriod int - ThemeEmoticon string - HasScheduled bool - Pinned bool - PinnedOrder int - UnreadMark bool - ViewForumAsMessages bool - PeerSettingsBarHidden bool + Peer Peer + ChannelLeft bool + FolderID int + TopMessage int + TopMessageDate int + // HistoryClearAnchorID/Date are internal owner-view projection metadata. + // They are not TL dialog fields; dialog response assembly uses them to + // materialize the matching messageActionHistoryClear top payload. + HistoryClearAnchorID int + HistoryClearAnchorDate int + ReadInboxMaxID int + ReadOutboxMaxID int + UnreadCount int + UnreadMentions int + UnreadReactions int + TTLPeriod int + ThemeEmoticon string + HasScheduled bool + Pinned bool + PinnedOrder int + UnreadMark bool + ViewForumAsMessages bool + PeerSettingsBarHidden bool // Pts 是 channel peer 当前 channel pts；客户端用 dialog.pts 初始化本地 // channel 序列并决定 getChannelDifference 起点，channel dialog 必填。 Pts int diff --git a/internal/domain/update_event.go b/internal/domain/update_event.go index 9dab6286..5398e809 100644 --- a/internal/domain/update_event.go +++ b/internal/domain/update_event.go @@ -44,7 +44,6 @@ const ( UpdateEventDialogFilterOrder UpdateEventType = "dialog_filter_order" UpdateEventDialogFilters UpdateEventType = "dialog_filters" UpdateEventFolderPeers UpdateEventType = "folder_peers" - UpdateEventChannelAvailable UpdateEventType = "channel_available_messages" UpdateEventChannelViewForum UpdateEventType = "channel_view_forum_as_messages" UpdateEventStory UpdateEventType = "story" UpdateEventReadStories UpdateEventType = "read_stories" @@ -141,7 +140,6 @@ func (e UpdateEvent) LacksWirePts() bool { UpdateEventDialogFilter, UpdateEventDialogFilterOrder, UpdateEventDialogFilters, - UpdateEventChannelAvailable, UpdateEventChannelViewForum, UpdateEventStory, UpdateEventReadStories, @@ -178,7 +176,10 @@ type UpdateDifference struct { // ChannelDifferenceNudge is a computed account-level hint that a channel diff is dirty. type ChannelDifferenceNudge struct { - ChannelID int64 - Pts int - Channel *ChannelView + ChannelID int64 + Pts int + ChannelUpdatesDirty bool + AvailableMinID int + HistoryClearDate int + Channel *ChannelView } diff --git a/internal/rpc/channels_messages.go b/internal/rpc/channels_messages.go index a5703fb6..67c10c45 100644 --- a/internal/rpc/channels_messages.go +++ b/internal/rpc/channels_messages.go @@ -394,10 +394,14 @@ func (r *Router) onChannelsDeleteHistory(ctx context.Context, req *tg.ChannelsDe return nil, channelDeleteErr(err) } if res.Event.Pts == 0 { - event := r.recordChannelAvailableMessages(ctx, userID, res.Channel.ID, res.AvailableMinID) - updates := r.channelAvailableMessagesUpdates(userID, res.Channel, event.MaxID) - updates.Updates = appendAuxPtsBookkeeping(updates.Updates, event) - r.pushUserUpdates(ctx, userID, updates) + updates := r.channelAvailableMessagesUpdates(userID, res.Channel, res.AvailableMinID) + if res.AvailableMinChanged { + // updateChannelAvailableMessages is an absolute owner-local boundary + // with no account/channel pts. Other online sessions consume it + // immediately; future cold/offline sessions discover the same + // boundary through account/channel difference recovery. + r.pushUserUpdates(ctx, userID, updates) + } return updates, nil } pushBatch := func(batch domain.DeleteChannelHistoryResult) *tg.Updates { diff --git a/internal/rpc/channels_messages_rpc_test.go b/internal/rpc/channels_messages_rpc_test.go index 9eb3e394..f36adbb6 100644 --- a/internal/rpc/channels_messages_rpc_test.go +++ b/internal/rpc/channels_messages_rpc_test.go @@ -174,6 +174,7 @@ func TestChannelsDeleteHistoryLocalClearEmitsAvailableMessagesUpdate(t *testing. r := New(Config{}, Deps{ Users: appusers.NewService(userStore), Channels: appchannels.NewService(channelStore), + Dialogs: appdialogs.NewService(memory.NewDialogStore(), channelStore), Updates: updateSvc, Sessions: sessions, }, zaptest.NewLogger(t), clock.System) @@ -193,7 +194,14 @@ func TestChannelsDeleteHistoryLocalClearEmitsAvailableMessagesUpdate(t *testing. if err != nil { t.Fatalf("send channel message: %v", err) } - msg := sent.(*tg.Updates).Updates[1].(*tg.UpdateNewChannelMessage).Message.(*tg.Message) + newChannelUpdate := sent.(*tg.Updates).Updates[1].(*tg.UpdateNewChannelMessage) + msg := newChannelUpdate.Message.(*tg.Message) + clearSince := int(time.Now().Unix()) - 1 + stateBefore, err := updateSvc.CurrentState(ctx, owner.ID) + if err != nil { + t.Fatalf("account state before clear: %v", err) + } + pushesBefore := len(sessions.pushedUserIDs()) cleared, err := r.onChannelsDeleteHistory(WithUserID(ctx, owner.ID), &tg.ChannelsDeleteHistoryRequest{ Channel: &tg.InputChannel{ChannelID: channel.ID, AccessHash: channel.AccessHash}, MaxID: msg.ID, @@ -202,42 +210,144 @@ func TestChannelsDeleteHistoryLocalClearEmitsAvailableMessagesUpdate(t *testing. t.Fatalf("delete channel history local: %v", err) } updates, ok := cleared.(*tg.Updates) - if !ok || len(updates.Updates) != 2 { - t.Fatalf("clear response = %T %+v, want available update plus pts bookkeeping", cleared, cleared) + if !ok || len(updates.Updates) != 1 { + t.Fatalf("clear response = %T %+v, want one no-pts available update", cleared, cleared) } available, ok := updates.Updates[0].(*tg.UpdateChannelAvailableMessages) if !ok || available.ChannelID != channel.ID || available.AvailableMinID != msg.ID { t.Fatalf("clear update = %#v, want updateChannelAvailableMessages channel=%d min=%d", updates.Updates[0], channel.ID, msg.ID) } - // updateChannelAvailableMessages 不带账号 pts，事件占用的 pts 槽位 - // 必须用空 updateDeleteMessages 显式同步给客户端。 - bookkeeping, ok := updates.Updates[1].(*tg.UpdateDeleteMessages) - if !ok || len(bookkeeping.Messages) != 0 || bookkeeping.Pts <= 0 || bookkeeping.PtsCount != 1 { - t.Fatalf("clear bookkeeping = %#v, want empty updateDeleteMessages carrying the account pts step", updates.Updates[1]) + stateAfter, err := updateSvc.CurrentState(ctx, owner.ID) + if err != nil { + t.Fatalf("account state after clear: %v", err) + } + if stateAfter.Pts != stateBefore.Pts { + t.Fatalf("account pts advanced on local channel clear: before=%d after=%d", stateBefore.Pts, stateAfter.Pts) } pushed := sessions.snapshot() if pushed.userID != owner.ID { t.Fatalf("pushed user = %d, want owner %d", pushed.userID, owner.ID) } pushedUpdates, ok := pushed.message.(*tg.Updates) - if !ok || len(pushedUpdates.Updates) != 2 { - t.Fatalf("pushed clear update = %T %+v, want available update plus pts bookkeeping", pushed.message, pushed.message) + if !ok || len(pushedUpdates.Updates) != 1 { + t.Fatalf("pushed clear update = %T %+v, want one no-pts available update", pushed.message, pushed.message) } if _, ok := pushedUpdates.Updates[0].(*tg.UpdateChannelAvailableMessages); !ok { t.Fatalf("pushed update[0] = %T, want updateChannelAvailableMessages", pushedUpdates.Updates[0]) } - diff, err := r.onUpdatesGetDifference(WithUserID(ctx, owner.ID), &tg.UpdatesGetDifferenceRequest{Pts: 0}) + if got := len(sessions.pushedUserIDs()) - pushesBefore; got != 1 { + t.Fatalf("clear online pushes = %d, want exactly one owner-session fanout", got) + } + diff, err := updateSvc.GetDifference(ctx, [8]byte{}, owner.ID, stateBefore) if err != nil { t.Fatalf("get difference: %v", err) } - full, ok := diff.(*tg.UpdatesDifference) - if !ok || len(full.OtherUpdates) != 1 { - t.Fatalf("difference = %T %+v, want one other update", diff, diff) + if len(diff.Events) != 0 || diff.State.Pts != stateBefore.Pts { + t.Fatalf("difference after no-pts clear = %+v, want no durable account event", diff) } - if diffUpdate, ok := full.OtherUpdates[0].(*tg.UpdateChannelAvailableMessages); !ok || diffUpdate.ChannelID != channel.ID || diffUpdate.AvailableMinID != msg.ID { - t.Fatalf("difference update = %#v, want updateChannelAvailableMessages", full.OtherUpdates[0]) + offline, err := r.onUpdatesGetDifference(WithUserID(ctx, owner.ID), &tg.UpdatesGetDifferenceRequest{ + Pts: stateBefore.Pts, + Date: clearSince, + }) + if err != nil { + t.Fatalf("account difference after offline local clear: %v", err) + } + offlineDiff, ok := offline.(*tg.UpdatesDifference) + if !ok { + t.Fatalf("offline account difference = %T %+v, want updates.difference", offline, offline) + } + var accountAvailable *tg.UpdateChannelAvailableMessages + for _, update := range offlineDiff.OtherUpdates { + if value, ok := update.(*tg.UpdateChannelAvailableMessages); ok { + accountAvailable = value + break + } + } + if accountAvailable == nil || + accountAvailable.ChannelID != channel.ID || + accountAvailable.AvailableMinID != msg.ID { + t.Fatalf("offline account updates = %+v, want updateChannelAvailableMessages channel=%d min=%d", + offlineDiff.OtherUpdates, channel.ID, msg.ID) + } + if offlineDiff.State.Pts != stateBefore.Pts || offlineDiff.State.Date < clearSince { + t.Fatalf("offline account state = %+v, want unchanged pts=%d and non-regressing date", offlineDiff.State, stateBefore.Pts) } + channelOffline, err := r.onUpdatesGetChannelDifference(WithUserID(ctx, owner.ID), &tg.UpdatesGetChannelDifferenceRequest{ + Channel: &tg.InputChannel{ChannelID: channel.ID, AccessHash: channel.AccessHash}, + Filter: &tg.ChannelMessagesFilterEmpty{}, + Pts: newChannelUpdate.Pts, + Limit: 100, + }) + if err != nil { + t.Fatalf("channel difference after offline local clear: %v", err) + } + channelDiff, ok := channelOffline.(*tg.UpdatesChannelDifference) + if !ok { + t.Fatalf("offline channel difference = %T %+v, want non-empty difference with absolute update", channelOffline, channelOffline) + } + var channelAvailable *tg.UpdateChannelAvailableMessages + for _, update := range channelDiff.OtherUpdates { + if value, ok := update.(*tg.UpdateChannelAvailableMessages); ok { + channelAvailable = value + break + } + } + if channelAvailable == nil || + channelAvailable.ChannelID != channel.ID || + channelAvailable.AvailableMinID != msg.ID || + channelDiff.Pts != newChannelUpdate.Pts { + t.Fatalf("offline channel updates = %+v pts=%d, want available boundary %d with unchanged channel pts=%d", + channelDiff.OtherUpdates, channelDiff.Pts, msg.ID, newChannelUpdate.Pts) + } + + req := &tg.MessagesGetDialogsRequest{OffsetPeer: &tg.InputPeerEmpty{}, Limit: 20} + var b bin.Buffer + if err := req.Encode(&b); err != nil { + t.Fatalf("encode get dialogs after clear: %v", err) + } + enc, err := r.Dispatch(WithUserID(ctx, owner.ID), [8]byte{}, 0, &b) + if err != nil { + t.Fatalf("dispatch get dialogs after clear: %v", err) + } + dialogs, ok := enc.(*tg.MessagesDialogs) + if !ok || len(dialogs.Dialogs) != 1 || len(dialogs.Messages) != 1 { + t.Fatalf("dialogs after cold projection = %T %+v, want one anchored dialog/message", enc, enc) + } + dialog := dialogs.Dialogs[0].(*tg.Dialog) + service, ok := dialogs.Messages[0].(*tg.MessageService) + if !ok || dialog.TopMessage != msg.ID || service.ID != msg.ID { + t.Fatalf("cold dialog projection = dialog=%+v message=%T %+v, want history-clear top %d", dialog, dialogs.Messages[0], dialogs.Messages[0], msg.ID) + } + if _, ok := service.Action.(*tg.MessageActionHistoryClear); !ok { + t.Fatalf("cold dialog service action = %T, want messageActionHistoryClear", service.Action) + } + + historyReq := &tg.MessagesGetHistoryRequest{ + Peer: &tg.InputPeerChannel{ChannelID: channel.ID, AccessHash: channel.AccessHash}, + Limit: 20, + } + b.Reset() + if err := historyReq.Encode(&b); err != nil { + t.Fatalf("encode get history after clear: %v", err) + } + historyEnc, err := r.Dispatch(WithUserID(ctx, owner.ID), [8]byte{}, 0, &b) + if err != nil { + t.Fatalf("dispatch get history after clear: %v", err) + } + history, ok := historyEnc.(*tg.MessagesChannelMessages) + if !ok || len(history.Messages) != 1 { + t.Fatalf("history after cold projection = %T %+v, want one history-clear marker", historyEnc, historyEnc) + } + historyService, ok := history.Messages[0].(*tg.MessageService) + if !ok || historyService.ID != msg.ID { + t.Fatalf("cold history projection = %T %+v, want service marker %d", history.Messages[0], history.Messages[0], msg.ID) + } + if _, ok := historyService.Action.(*tg.MessageActionHistoryClear); !ok { + t.Fatalf("cold history service action = %T, want messageActionHistoryClear", historyService.Action) + } + + stalePushesBefore := len(sessions.pushedUserIDs()) stale, err := r.onChannelsDeleteHistory(WithUserID(ctx, owner.ID), &tg.ChannelsDeleteHistoryRequest{ Channel: &tg.InputChannel{ChannelID: channel.ID, AccessHash: channel.AccessHash}, MaxID: msg.ID - 1, @@ -246,13 +356,16 @@ func TestChannelsDeleteHistoryLocalClearEmitsAvailableMessagesUpdate(t *testing. t.Fatalf("stale delete channel history local: %v", err) } staleUpdates, ok := stale.(*tg.Updates) - if !ok || len(staleUpdates.Updates) != 2 { - t.Fatalf("stale clear response = %T %+v, want monotonic update plus pts bookkeeping", stale, stale) + if !ok || len(staleUpdates.Updates) != 1 { + t.Fatalf("stale clear response = %T %+v, want monotonic absolute update", stale, stale) } staleAvailable, ok := staleUpdates.Updates[0].(*tg.UpdateChannelAvailableMessages) if !ok || staleAvailable.ChannelID != channel.ID || staleAvailable.AvailableMinID != msg.ID { t.Fatalf("stale clear update = %#v, want monotonic updateChannelAvailableMessages channel=%d min=%d", staleUpdates.Updates[0], channel.ID, msg.ID) } + if got := len(sessions.pushedUserIDs()); got != stalePushesBefore { + t.Fatalf("stale clear pushed another online update: before=%d after=%d", stalePushesBefore, got) + } } func TestChannelDeleteRejectsInvalidMessageIDsRPC(t *testing.T) { diff --git a/internal/rpc/channels_stubs.go b/internal/rpc/channels_stubs.go index a9135368..ead8cf50 100644 --- a/internal/rpc/channels_stubs.go +++ b/internal/rpc/channels_stubs.go @@ -290,27 +290,6 @@ func peerIDsExcept(ids []int64, skipIDs ...int64) []int64 { type channelFanoutScope int -func (r *Router) recordChannelAvailableMessages(ctx context.Context, userID, channelID int64, availableMinID int) domain.UpdateEvent { - event := domain.UpdateEvent{ - UserID: userID, - Type: domain.UpdateEventChannelAvailable, - Date: int(r.clock.Now().Unix()), - Peer: domain.Peer{Type: domain.PeerTypeChannel, ID: channelID}, - MaxID: availableMinID, - PtsCount: 1, - } - if r.deps.Updates == nil || userID == 0 || channelID == 0 || availableMinID <= 0 { - return event - } - authKeyID, _ := AuthKeyIDFrom(ctx) - sessionID, _ := SessionIDFrom(ctx) - recorded, _, err := r.deps.Updates.RecordChannelAvailableMessages(ctx, authKeyID, userID, channelID, availableMinID, rawAuthKeyIDForOrigin(ctx), sessionID) - if err != nil { - return event - } - return recorded -} - func (r *Router) recordChannelReadInbox(ctx context.Context, userID int64, read domain.ReadChannelHistoryResult) (domain.UpdateEvent, error) { if !read.Changed || read.ChannelID == 0 { return domain.UpdateEvent{}, nil diff --git a/internal/rpc/convert_channels_core.go b/internal/rpc/convert_channels_core.go index 695c8e93..10bee38a 100644 --- a/internal/rpc/convert_channels_core.go +++ b/internal/rpc/convert_channels_core.go @@ -222,6 +222,8 @@ func tgChannelMessageAction(action domain.ChannelMessageAction) tg.MessageAction switch action.Type { case domain.ChannelActionCreate: return &tg.MessageActionChannelCreate{Title: action.Title} + case domain.ChannelActionHistoryClear: + return &tg.MessageActionHistoryClear{} case domain.ChannelActionChatAddUser, domain.ChannelActionChatJoined: return &tg.MessageActionChatAddUser{Users: append([]int64(nil), action.UserIDs...)} case domain.ChannelActionChatJoinedByLink: diff --git a/internal/rpc/convert_updates.go b/internal/rpc/convert_updates.go index 7d3f7bd8..a12a48e7 100644 --- a/internal/rpc/convert_updates.go +++ b/internal/rpc/convert_updates.go @@ -55,11 +55,23 @@ func tgUpdatesDifference(viewerUserID int64, diff domain.UpdateDifference) tg.Up if nudge.ChannelID == 0 { continue } - update := &tg.UpdateChannelTooLong{ChannelID: nudge.ChannelID} - if nudge.Pts > 0 { - update.SetPts(nudge.Pts) + if nudge.AvailableMinID > 0 { + out.OtherUpdates = append(out.OtherUpdates, &tg.UpdateChannelAvailableMessages{ + ChannelID: nudge.ChannelID, + AvailableMinID: nudge.AvailableMinID, + }) + } + // Preserve compatibility for older domain callers that only supplied + // Pts: a nudge without an owner-local boundary is a shared channel + // update nudge. New store results set ChannelUpdatesDirty explicitly so + // a channel can carry both absolute clear and too-long updates. + if nudge.ChannelUpdatesDirty || nudge.AvailableMinID == 0 { + update := &tg.UpdateChannelTooLong{ChannelID: nudge.ChannelID} + if nudge.Pts > 0 { + update.SetPts(nudge.Pts) + } + out.OtherUpdates = append(out.OtherUpdates, update) } - out.OtherUpdates = append(out.OtherUpdates, update) if nudge.Channel != nil && nudge.Channel.Channel.ID != 0 { addChannelNudgeChat(out, seenChats, tgChannelChatForView(viewerUserID, *nudge.Channel)) } @@ -110,7 +122,7 @@ func tgChannelDifference(viewerUserID int64, diff domain.ChannelDifference) tg.U Users: tgUsersForViewer(viewerUserID, diff.Users), } } - if len(diff.Events) == 0 && len(diff.NewMessages) == 0 && len(diff.OtherUpdates) == 0 { + if len(diff.Events) == 0 && len(diff.NewMessages) == 0 && len(diff.OtherUpdates) == 0 && diff.AvailableMinID == 0 { return &tg.UpdatesChannelDifferenceEmpty{ Final: diff.Final, Pts: diff.Pts, @@ -136,6 +148,12 @@ func tgChannelDifference(viewerUserID int64, diff domain.ChannelDifference) tg.U updates = append(updates, update) } } + if diff.AvailableMinID > 0 { + updates = append(updates, &tg.UpdateChannelAvailableMessages{ + ChannelID: diff.Channel.ID, + AvailableMinID: diff.AvailableMinID, + }) + } chats := tgChannelDifferenceChats(viewerUserID, diff) users := tgUsersForViewer(viewerUserID, diff.Users) return &tg.UpdatesChannelDifference{ @@ -483,14 +501,6 @@ func tgOtherUpdateFromEvent(event domain.UpdateEvent) tg.UpdateClass { Pts: event.Pts, PtsCount: event.PtsCount, } - case domain.UpdateEventChannelAvailable: - if event.Peer.Type != domain.PeerTypeChannel || event.Peer.ID == 0 || event.MaxID <= 0 { - return nil - } - return &tg.UpdateChannelAvailableMessages{ - ChannelID: event.Peer.ID, - AvailableMinID: event.MaxID, - } default: return nil } diff --git a/internal/rpc/deps.go b/internal/rpc/deps.go index 3be483f3..186b8f31 100644 --- a/internal/rpc/deps.go +++ b/internal/rpc/deps.go @@ -496,7 +496,6 @@ type UpdatesService interface { RecordDialogFilterOrder(ctx context.Context, stateAuthKeyID [8]byte, userID int64, order []int, excludeAuthKeyID [8]byte, excludeSessionID int64) (domain.UpdateEvent, domain.UpdateState, error) RecordDialogFiltersReload(ctx context.Context, stateAuthKeyID [8]byte, userID int64, excludeAuthKeyID [8]byte, excludeSessionID int64) (domain.UpdateEvent, domain.UpdateState, error) RecordFolderPeers(ctx context.Context, stateAuthKeyID [8]byte, userID int64, peers []domain.FolderPeerUpdate, excludeAuthKeyID [8]byte, excludeSessionID int64) (domain.UpdateEvent, domain.UpdateState, error) - RecordChannelAvailableMessages(ctx context.Context, stateAuthKeyID [8]byte, userID, channelID int64, availableMinID int, excludeAuthKeyID [8]byte, excludeSessionID int64) (domain.UpdateEvent, domain.UpdateState, error) RecordChannelViewForumAsMessages(ctx context.Context, stateAuthKeyID [8]byte, userID, channelID int64, enabled bool, excludeAuthKeyID [8]byte, excludeSessionID int64) (domain.UpdateEvent, domain.UpdateState, error) RecordChannelDiscussionInbox(ctx context.Context, stateAuthKeyID [8]byte, userID, channelID int64, topicID, maxID int, excludeAuthKeyID [8]byte, excludeSessionID int64) (domain.UpdateEvent, domain.UpdateState, error) RecordDraftMessage(ctx context.Context, stateAuthKeyID [8]byte, userID int64, peer domain.Peer, topMsgID int, excludeAuthKeyID [8]byte, excludeSessionID int64) (domain.UpdateEvent, domain.UpdateState, error) diff --git a/internal/rpc/messages_delete.go b/internal/rpc/messages_delete.go index 2228b730..b3b6c966 100644 --- a/internal/rpc/messages_delete.go +++ b/internal/rpc/messages_delete.go @@ -74,16 +74,13 @@ func (r *Router) onMessagesDeleteHistory(ctx context.Context, req *tg.MessagesDe }) return &tg.MessagesAffectedHistory{Pts: res.Event.Pts, PtsCount: res.Event.PtsCount, Offset: res.Offset}, nil } - if res.AvailableMinID > 0 { - event := r.recordChannelAvailableMessages(ctx, userID, res.Channel.ID, res.AvailableMinID) - updates := r.channelAvailableMessagesUpdates(userID, res.Channel, event.MaxID) - updates.Updates = appendAuxPtsBookkeeping(updates.Updates, event) + if res.AvailableMinChanged && res.AvailableMinID > 0 { + updates := r.channelAvailableMessagesUpdates(userID, res.Channel, res.AvailableMinID) r.pushUserUpdates(ctx, userID, updates) - if event.Pts != 0 { - return &tg.MessagesAffectedHistory{Pts: event.Pts, PtsCount: event.PtsCount, Offset: res.Offset}, nil - } } - return &tg.MessagesAffectedHistory{Pts: res.Channel.Pts, PtsCount: 0, Offset: res.Offset}, nil + // messages.affectedHistory.pts is the caller's account-state snapshot. + // The local channel clear itself consumes no account/channel pts. + return r.affectedHistory(ctx, authKeyID, userID, res.Offset) } if peer.Type != domain.PeerTypeUser { return nil, peerIDInvalidErr() diff --git a/internal/rpc/messages_delete_rpc_test.go b/internal/rpc/messages_delete_rpc_test.go index f76f40d7..a47dfa4b 100644 --- a/internal/rpc/messages_delete_rpc_test.go +++ b/internal/rpc/messages_delete_rpc_test.go @@ -7,11 +7,74 @@ import ( "github.com/iamxvbaba/td/tg" "go.uber.org/zap/zaptest" appchannels "telesrv/internal/app/channels" + appupdates "telesrv/internal/app/updates" "telesrv/internal/domain" "telesrv/internal/store/memory" "testing" ) +func TestMessagesDeleteHistoryChannelLocalClearReturnsAccountStateWithoutPTS(t *testing.T) { + ctx := context.Background() + channelStore := memory.NewChannelStore() + created, err := channelStore.CreateChannel(ctx, domain.CreateChannelRequest{ + CreatorUserID: 7, + Title: "local clear no pts", + Megagroup: true, + Date: 1_700_002_100, + }) + if err != nil { + t.Fatalf("create channel: %v", err) + } + sent, err := channelStore.SendChannelMessage(ctx, domain.SendChannelMessageRequest{ + UserID: 7, + ChannelID: created.Channel.ID, + RandomID: 210_001, + Message: "clear locally", + Date: 1_700_002_101, + }) + if err != nil { + t.Fatalf("send channel message: %v", err) + } + updateSvc := appupdates.NewService(memory.NewUpdateStateStore(), memory.NewUpdateEventStore()) + stateBefore, err := updateSvc.CurrentState(ctx, 7) + if err != nil { + t.Fatalf("account state before clear: %v", err) + } + sessions := &captureSessions{} + r := New(Config{}, Deps{ + Channels: appchannels.NewService(channelStore), + Updates: updateSvc, + Sessions: sessions, + }, zaptest.NewLogger(t), clock.System) + + affected, err := r.onMessagesDeleteHistory(WithUserID(ctx, 7), &tg.MessagesDeleteHistoryRequest{ + Peer: &tg.InputPeerChannel{ChannelID: created.Channel.ID, AccessHash: created.Channel.AccessHash}, + MaxID: sent.Message.ID, + }) + if err != nil { + t.Fatalf("messages.deleteHistory local channel: %v", err) + } + if affected.Pts != stateBefore.Pts || affected.PtsCount != 0 || affected.Offset != 0 { + t.Fatalf("affected history = %+v, want account pts=%d pts_count=0 offset=0", affected, stateBefore.Pts) + } + stateAfter, err := updateSvc.CurrentState(ctx, 7) + if err != nil { + t.Fatalf("account state after clear: %v", err) + } + if stateAfter.Pts != stateBefore.Pts { + t.Fatalf("local channel clear advanced account pts: before=%d after=%d", stateBefore.Pts, stateAfter.Pts) + } + pushed := sessions.snapshot() + updates, ok := pushed.message.(*tg.Updates) + if !ok || len(updates.Updates) != 1 { + t.Fatalf("pushed local clear = %T %+v, want one available update", pushed.message, pushed.message) + } + available, ok := updates.Updates[0].(*tg.UpdateChannelAvailableMessages) + if !ok || available.ChannelID != created.Channel.ID || available.AvailableMinID != sent.Message.ID { + t.Fatalf("available update = %#v, want channel=%d min=%d", updates.Updates[0], created.Channel.ID, sent.Message.ID) + } +} + func TestMessagesDeleteHistoryChannelReturnsOffsetForBoundedPage(t *testing.T) { ctx := context.Background() userStore := memory.NewUserStore() diff --git a/internal/rpc/messages_register.go b/internal/rpc/messages_register.go index 67c6e510..38520986 100644 --- a/internal/rpc/messages_register.go +++ b/internal/rpc/messages_register.go @@ -604,14 +604,15 @@ func (r *Router) registerMessages(d *tlprofile.Dispatcher) { return &tg.MessagesMessages{}, nil } history, err := r.deps.Channels.GetHistory(ctx, userID, domain.ChannelHistoryFilter{ - ChannelID: filter.Peer.ID, - OffsetID: filter.OffsetID, - OffsetDate: filter.OffsetDate, - AddOffset: filter.AddOffset, - Limit: filter.Limit, - MaxID: filter.MaxID, - MinID: filter.MinID, - Hash: filter.Hash, + ChannelID: filter.Peer.ID, + OffsetID: filter.OffsetID, + OffsetDate: filter.OffsetDate, + AddOffset: filter.AddOffset, + Limit: filter.Limit, + MaxID: filter.MaxID, + MinID: filter.MinID, + Hash: filter.Hash, + IncludeHistoryClearAnchor: true, }) if err != nil { return nil, channelInvalidErr(err) diff --git a/internal/rpc/rpc_testkit_updates_test.go b/internal/rpc/rpc_testkit_updates_test.go index 0477236a..d1aff8e4 100644 --- a/internal/rpc/rpc_testkit_updates_test.go +++ b/internal/rpc/rpc_testkit_updates_test.go @@ -284,15 +284,6 @@ func (s *captureUpdates) RecordFolderPeers(_ context.Context, authKeyID [8]byte, return s.recordCapturedEvent(authKeyID, userID, domain.UpdateEvent{Type: domain.UpdateEventFolderPeers, FolderPeers: append([]domain.FolderPeerUpdate(nil), peers...)}) } -func (s *captureUpdates) RecordChannelAvailableMessages(_ context.Context, authKeyID [8]byte, userID, channelID int64, availableMinID int, excludeAuthKeyID [8]byte, excludeSessionID int64) (domain.UpdateEvent, domain.UpdateState, error) { - s.captureExclude(excludeAuthKeyID, excludeSessionID) - return s.recordCapturedEvent(authKeyID, userID, domain.UpdateEvent{ - Type: domain.UpdateEventChannelAvailable, - Peer: domain.Peer{Type: domain.PeerTypeChannel, ID: channelID}, - MaxID: availableMinID, - }) -} - func (s *captureUpdates) RecordChannelViewForumAsMessages(_ context.Context, authKeyID [8]byte, userID, channelID int64, enabled bool, excludeAuthKeyID [8]byte, excludeSessionID int64) (domain.UpdateEvent, domain.UpdateState, error) { s.captureExclude(excludeAuthKeyID, excludeSessionID) return s.recordCapturedEvent(authKeyID, userID, domain.UpdateEvent{ diff --git a/internal/rpc/updates.go b/internal/rpc/updates.go index 49bc8940..eab8fd6a 100644 --- a/internal/rpc/updates.go +++ b/internal/rpc/updates.go @@ -102,6 +102,21 @@ func (r *Router) onUpdatesGetDifference(ctx context.Context, req *tg.UpdatesGetD return nil, internalErr() } st.ChannelNudges = r.accountChannelDifferenceNudges(ctx, userID, req.Date) + if !st.Partial { + for _, nudge := range st.ChannelNudges { + if nudge.AvailableMinID <= 0 { + continue + } + // No-PTS owner-local clear recovery is date-indexed. Advance only + // on the final account page and never beyond the server clock; the + // indexed query deliberately overlaps equality, so a same-second + // reconnect can receive an idempotent duplicate rather than miss. + if now := int(r.clock.Now().Unix()); now > st.State.Date { + st.State.Date = now + } + break + } + } // 密聊设备级 qts 消息（独立于账号级 pts 事件）：按当前设备 req.Qts 补回。 encMsgs, newQts := r.encryptedDifference(ctx, req.Qts) // 密聊握手/已读状态事件（无 qts）：按未投递标记补回 OtherUpdates。 @@ -154,10 +169,19 @@ func (r *Router) accountChannelDifferenceNudges(ctx context.Context, userID int6 } } for _, item := range dirty { + if len(out) >= maxNudges { + break + } if item.ChannelID == 0 { continue } - nudge := domain.ChannelDifferenceNudge{ChannelID: item.ChannelID, Pts: item.Pts} + nudge := domain.ChannelDifferenceNudge{ + ChannelID: item.ChannelID, + Pts: item.Pts, + ChannelUpdatesDirty: item.ChannelUpdatesDirty, + AvailableMinID: item.AvailableMinID, + HistoryClearDate: item.HistoryClearDate, + } if view, ok := viewsByID[item.ChannelID]; ok { nudge.Channel = &view } diff --git a/internal/store/memory/channel_dialogs.go b/internal/store/memory/channel_dialogs.go index dfbd19cc..c5f9f4d1 100644 --- a/internal/store/memory/channel_dialogs.go +++ b/internal/store/memory/channel_dialogs.go @@ -95,8 +95,8 @@ func (s *ChannelStore) ListChannelDialogs(_ context.Context, viewerUserID int64, out.Dialogs = append(out.Dialogs, dialog) channel := s.channels[dialog.Peer.ID] out.Channels = append(out.Channels, channel) - if msg, ok := s.findMessageLocked(dialog.Peer.ID, dialog.TopMessage); ok && !msg.Deleted { - out.Messages = append(out.Messages, cloneChannelMessage(msg)) + if msg, ok := s.channelMessageForMemberLocked(viewerUserID, dialog.Peer.ID, dialog.TopMessage); ok { + out.Messages = append(out.Messages, msg) } } // 与 PG 同因：getDialogs top message 按 viewer 补未读提及标志。 @@ -151,8 +151,8 @@ func (s *ChannelStore) GetChannelDialogs(_ context.Context, viewerUserID int64, out.Channels = append(out.Channels, cloneChannel(parent)) } } - if msg, ok := s.findMessageLocked(channelID, dialog.TopMessage); ok && !msg.Deleted { - out.Messages = append(out.Messages, cloneChannelMessage(msg)) + if msg, ok := s.channelMessageForMemberLocked(viewerUserID, channelID, dialog.TopMessage); ok { + out.Messages = append(out.Messages, msg) } } out.Count = len(out.Dialogs) @@ -470,36 +470,47 @@ func channelDialogToDialog(dialog domain.ChannelDialog, channelPts int, memberSt return domain.Dialog{ Peer: domain.Peer{Type: domain.PeerTypeChannel, ID: dialog.ChannelID}, // 非成员预览(publicPreviewMember/被踢)须标记 ChannelLeft,客户端据此把频道渲染为只读 left 预览。 - ChannelLeft: memberStatus == domain.ChannelMemberLeft, - FolderID: dialog.FolderID, - TopMessage: dialog.TopMessageID, - TopMessageDate: dialog.TopMessageDate, - ReadInboxMaxID: dialog.ReadInboxMaxID, - ReadOutboxMaxID: dialog.ReadOutboxMaxID, - UnreadCount: dialog.UnreadCount, - UnreadMentions: dialog.UnreadMentions, - UnreadReactions: dialog.UnreadReactions, - Pinned: dialog.Pinned, - PinnedOrder: dialog.PinnedOrder, - UnreadMark: dialog.UnreadMark, - ViewForumAsMessages: dialog.ViewForumAsMessages, - HasScheduled: dialog.HasScheduled, - Pts: channelPts, + ChannelLeft: memberStatus == domain.ChannelMemberLeft, + FolderID: dialog.FolderID, + TopMessage: dialog.TopMessageID, + TopMessageDate: dialog.TopMessageDate, + HistoryClearAnchorID: dialog.HistoryClearAnchorID, + HistoryClearAnchorDate: dialog.HistoryClearAnchorDate, + ReadInboxMaxID: dialog.ReadInboxMaxID, + ReadOutboxMaxID: dialog.ReadOutboxMaxID, + UnreadCount: dialog.UnreadCount, + UnreadMentions: dialog.UnreadMentions, + UnreadReactions: dialog.UnreadReactions, + Pinned: dialog.Pinned, + PinnedOrder: dialog.PinnedOrder, + UnreadMark: dialog.UnreadMark, + ViewForumAsMessages: dialog.ViewForumAsMessages, + HasScheduled: dialog.HasScheduled, + Pts: channelPts, } } func previewChannelDialog(userID int64, channel domain.Channel, member domain.ChannelMember) domain.ChannelDialog { topMessageID := channel.TopMessageID if topMessageID <= member.AvailableMinID { - topMessageID = 0 + topMessageID = member.HistoryClearAnchorID + if topMessageID != member.AvailableMinID { + topMessageID = 0 + } + } + topMessageDate := channel.Date + if topMessageID > 0 && topMessageID == member.HistoryClearAnchorID { + topMessageDate = member.HistoryClearAnchorDate } return domain.ChannelDialog{ - UserID: userID, - ChannelID: channel.ID, - TopMessageID: topMessageID, - TopMessageDate: channel.Date, - ReadInboxMaxID: maxInt(channel.TopMessageID, member.ReadInboxMaxID), - ReadOutboxMaxID: maxInt(channel.TopMessageID, member.ReadOutboxMaxID), + UserID: userID, + ChannelID: channel.ID, + TopMessageID: topMessageID, + TopMessageDate: topMessageDate, + HistoryClearAnchorID: member.HistoryClearAnchorID, + HistoryClearAnchorDate: member.HistoryClearAnchorDate, + ReadInboxMaxID: maxInt(channel.TopMessageID, member.ReadInboxMaxID), + ReadOutboxMaxID: maxInt(channel.TopMessageID, member.ReadOutboxMaxID), } } diff --git a/internal/store/memory/channel_helpers.go b/internal/store/memory/channel_helpers.go index 13eb2cd3..e746b78a 100644 --- a/internal/store/memory/channel_helpers.go +++ b/internal/store/memory/channel_helpers.go @@ -337,9 +337,19 @@ func (s *ChannelStore) ListDirtyActiveChannelsForUser(_ context.Context, userID if !ok || member.Status != domain.ChannelMemberActive { continue } + item := domain.DirtyChannel{ChannelID: channelID, Pts: channel.Pts} checkpoint := s.channelUpdateCheckpointLocked(channelID, channel) if checkpoint.LatestEventDate > sinceDate { - out = append(out, domain.DirtyChannel{ChannelID: channelID, Pts: channel.Pts}) + item.ChannelUpdatesDirty = true + } + if clearDate := s.historyClearDates[channelID][userID]; clearDate >= sinceDate && + member.HistoryClearAnchorID > 0 && + member.HistoryClearAnchorID == member.AvailableMinID { + item.AvailableMinID = member.AvailableMinID + item.HistoryClearDate = clearDate + } + if item.ChannelUpdatesDirty || item.AvailableMinID > 0 { + out = append(out, item) } } for channelID, channel := range s.channels { @@ -356,7 +366,11 @@ func (s *ChannelStore) ListDirtyActiveChannelsForUser(_ context.Context, userID } } if !found { - out = append(out, domain.DirtyChannel{ChannelID: channelID, Pts: channel.Pts}) + out = append(out, domain.DirtyChannel{ + ChannelID: channelID, + Pts: channel.Pts, + ChannelUpdatesDirty: true, + }) } } } @@ -453,12 +467,16 @@ func (s *ChannelStore) dialogForMemberLocked(userID int64, channel domain.Channe dialog.UserID = userID dialog.ChannelID = channel.ID dialog.TopMessageID = s.visibleTopMessageIDForMemberLocked(channel, member) + dialog.HistoryClearAnchorID = member.HistoryClearAnchorID + dialog.HistoryClearAnchorDate = member.HistoryClearAnchorDate // TopMessageDate 必须从可见 top 消息派生(不能继承空缓存的 0),否则会话排序/分页与预览 // dialog 的日期全错。与 postgres GetChannelDialogs 用 getChannelMessage 设 date 对齐。 dialog.TopMessageDate = 0 if dialog.TopMessageID > 0 { - if top, ok := s.findMessageLocked(channel.ID, dialog.TopMessageID); ok { + if top, ok := s.channelMessageForMemberLocked(userID, channel.ID, dialog.TopMessageID); ok { dialog.TopMessageDate = top.Date + } else if dialog.TopMessageID == member.HistoryClearAnchorID { + dialog.TopMessageDate = member.HistoryClearAnchorDate } } if member.ReadInboxMaxID > dialog.ReadInboxMaxID { diff --git a/internal/store/memory/channel_members.go b/internal/store/memory/channel_members.go index 96be9cc3..00daed7c 100644 --- a/internal/store/memory/channel_members.go +++ b/internal/store/memory/channel_members.go @@ -1107,6 +1107,8 @@ func syntheticMonoforumAdminMember(mono domain.Channel, parentMember domain.Chan } member.AvailableMinID = 0 member.AvailableMinPts = 0 + member.HistoryClearAnchorID = 0 + member.HistoryClearAnchorDate = 0 member.ReadInboxMaxID = mono.TopMessageID member.ReadOutboxMaxID = mono.TopMessageID member.UnreadMark = false diff --git a/internal/store/memory/channel_message_delete.go b/internal/store/memory/channel_message_delete.go index 06831b16..05acbd64 100644 --- a/internal/store/memory/channel_message_delete.go +++ b/internal/store/memory/channel_message_delete.go @@ -72,6 +72,12 @@ func (s *ChannelStore) DeleteChannelHistory(_ context.Context, req domain.Delete if err != nil { return domain.DeleteChannelHistoryResult{}, err } + if req.Date <= 0 { + req.Date = channel.Date + if req.Date <= 0 { + req.Date = 1 + } + } maxID := req.MaxID if maxID <= 0 || maxID > channel.TopMessageID { maxID = channel.TopMessageID @@ -79,6 +85,22 @@ func (s *ChannelStore) DeleteChannelHistory(_ context.Context, req domain.Delete member := s.members[req.ChannelID][req.UserID] if !req.ForEveryone { appliedMinID := maxInt(member.AvailableMinID, maxID) + changed := appliedMinID > member.AvailableMinID + if changed { + anchorDate := req.Date + if msg, ok := s.findMessageLocked(req.ChannelID, appliedMinID); ok && msg.Date > 0 { + anchorDate = msg.Date + } + if anchorDate <= 0 { + anchorDate = channel.Date + } + member.HistoryClearAnchorID = appliedMinID + member.HistoryClearAnchorDate = anchorDate + if s.historyClearDates[req.ChannelID] == nil { + s.historyClearDates[req.ChannelID] = make(map[int64]int) + } + s.historyClearDates[req.ChannelID][req.UserID] = req.Date + } member.AvailableMinID = appliedMinID member.ReadInboxMaxID = maxInt(member.ReadInboxMaxID, appliedMinID) member.UnreadMark = false @@ -92,7 +114,11 @@ func (s *ChannelStore) DeleteChannelHistory(_ context.Context, req domain.Delete s.dialogs[req.UserID] = make(map[int64]domain.ChannelDialog) } s.dialogs[req.UserID][req.ChannelID] = s.dialogForUserLocked(req.UserID, channel) - return domain.DeleteChannelHistoryResult{Channel: channel, AvailableMinID: appliedMinID}, nil + return domain.DeleteChannelHistoryResult{ + Channel: channel, + AvailableMinID: appliedMinID, + AvailableMinChanged: changed, + }, nil } if !canDeleteAnyChannelMessage(member) { return domain.DeleteChannelHistoryResult{}, domain.ErrChannelAdminRequired diff --git a/internal/store/memory/channel_message_history.go b/internal/store/memory/channel_message_history.go index c7993dd1..aff44adf 100644 --- a/internal/store/memory/channel_message_history.go +++ b/internal/store/memory/channel_message_history.go @@ -25,8 +25,16 @@ func (s *ChannelStore) ListChannelHistory(_ context.Context, viewerUserID int64, query := strings.ToLower(strings.TrimSpace(filter.Query)) matched := make([]domain.ChannelMessage, 0, len(items)) monoforumUserView := channel.Monoforum && !member.CanManageDirectMessages() + anchorID := 0 + if filter.IncludeHistoryClearAnchor && + member.HistoryClearAnchorID > 0 && + member.HistoryClearAnchorID == member.AvailableMinID { + anchorID = member.HistoryClearAnchorID + } + anchorSeen := false for _, msg := range items { - if msg.Deleted { + isAnchor := anchorID > 0 && msg.ID == anchorID + if msg.Deleted && !isAnchor { continue } if channel.Monoforum { @@ -37,9 +45,18 @@ func (s *ChannelStore) ListChannelHistory(_ context.Context, viewerUserID int64, continue } } - if msg.ID <= member.AvailableMinID { + if msg.ID < member.AvailableMinID || (msg.ID == member.AvailableMinID && !isAnchor) { continue } + if isAnchor { + msg = domain.ProjectChannelHistoryClearMessage( + msg, + filter.ChannelID, + member.HistoryClearAnchorID, + member.HistoryClearAnchorDate, + ) + anchorSeen = true + } if filter.PinnedOnly && !msg.Pinned { continue } @@ -66,6 +83,24 @@ func (s *ChannelStore) ListChannelHistory(_ context.Context, viewerUserID int64, } matched = append(matched, msg) } + if anchorID > 0 && !anchorSeen { + msg := domain.ProjectChannelHistoryClearMessage( + domain.ChannelMessage{}, + filter.ChannelID, + member.HistoryClearAnchorID, + member.HistoryClearAnchorDate, + ) + if (filter.MinDate <= 0 || msg.Date > filter.MinDate) && + (filter.MaxDate <= 0 || msg.Date < filter.MaxDate) && + (filter.MaxID <= 0 || msg.ID <= filter.MaxID) && + (filter.MinID <= 0 || msg.ID > filter.MinID) && + !filter.PinnedOnly && + !filter.MusicOnly && + query == "" && + filter.SenderUserID == 0 { + matched = append(matched, msg) + } + } extraChannels := []domain.Channel(nil) if channel.Monoforum && channel.LinkedMonoforumID != 0 { if parent, ok := s.channels[channel.LinkedMonoforumID]; ok && !parent.Deleted { @@ -179,6 +214,7 @@ func (s *ChannelStore) ListChannelHistory(_ context.Context, viewerUserID int64, } s.populateChannelMessageRepliesLocked(viewerUserID, filter.ChannelID, out) s.populateChannelMessageReactionsLocked(viewerUserID, channel, out) + projectMemoryChannelHistoryClearMessages(channel.ID, member, out) return domain.ChannelHistory{ Channel: channel, Self: member, @@ -318,17 +354,59 @@ func (s *ChannelStore) GetChannelMessages(_ context.Context, viewerUserID, chann if _, ok := wanted[msg.ID]; !ok { continue } + if msg.ID == member.HistoryClearAnchorID && + member.HistoryClearAnchorID == member.AvailableMinID { + messages = append(messages, domain.ProjectChannelHistoryClearMessage( + msg, + channelID, + member.HistoryClearAnchorID, + member.HistoryClearAnchorDate, + )) + delete(wanted, msg.ID) + continue + } if msg.Deleted || msg.ID <= member.AvailableMinID { continue } messages = append(messages, cloneChannelMessage(msg)) + delete(wanted, msg.ID) + } + if member.HistoryClearAnchorID > 0 && + member.HistoryClearAnchorID == member.AvailableMinID { + if _, ok := wanted[member.HistoryClearAnchorID]; ok { + messages = append(messages, domain.ProjectChannelHistoryClearMessage( + domain.ChannelMessage{}, + channelID, + member.HistoryClearAnchorID, + member.HistoryClearAnchorDate, + )) + } } sort.Slice(messages, func(i, j int) bool { return messages[i].ID > messages[j].ID }) s.populateChannelMessageRepliesLocked(viewerUserID, channelID, messages) s.populateChannelMessageReactionsLocked(viewerUserID, channel, messages) + projectMemoryChannelHistoryClearMessages(channelID, member, messages) return domain.ChannelHistory{Channel: channel, Self: member, Messages: messages, Count: len(messages)}, nil } +func projectMemoryChannelHistoryClearMessages(channelID int64, member domain.ChannelMember, messages []domain.ChannelMessage) { + if member.HistoryClearAnchorID <= 0 || + member.HistoryClearAnchorID != member.AvailableMinID { + return + } + for i := range messages { + if messages[i].ID != member.HistoryClearAnchorID { + continue + } + messages[i] = domain.ProjectChannelHistoryClearMessage( + messages[i], + channelID, + member.HistoryClearAnchorID, + member.HistoryClearAnchorDate, + ) + } +} + func (s *ChannelStore) ListStoryMessageForwards(_ context.Context, req domain.StoryMessageForwardListRequest) (domain.StoryMessageForwardList, error) { if req.ViewerUserID == 0 || req.Owner.ID == 0 || req.StoryID <= 0 || req.StoryID > domain.MaxStoryID { return domain.StoryMessageForwardList{}, domain.ErrStoryIDInvalid @@ -631,9 +709,40 @@ func (s *ChannelStore) visibleTopMessageIDForMemberLocked(channel domain.Channel return msg.ID } } + if member.HistoryClearAnchorID > 0 && + member.HistoryClearAnchorID == member.AvailableMinID { + return member.HistoryClearAnchorID + } return 0 } +func (s *ChannelStore) channelMessageForMemberLocked(userID, channelID int64, messageID int) (domain.ChannelMessage, bool) { + member, ok := s.members[channelID][userID] + if !ok { + msg, found := s.findMessageLocked(channelID, messageID) + if !found || msg.Deleted { + return domain.ChannelMessage{}, false + } + return cloneChannelMessage(msg), true + } + if member.HistoryClearAnchorID > 0 && + member.HistoryClearAnchorID == member.AvailableMinID && + messageID == member.HistoryClearAnchorID { + source, _ := s.findMessageLocked(channelID, messageID) + return domain.ProjectChannelHistoryClearMessage( + source, + channelID, + member.HistoryClearAnchorID, + member.HistoryClearAnchorDate, + ), true + } + msg, ok := s.findMessageLocked(channelID, messageID) + if !ok || msg.Deleted || msg.ID <= member.AvailableMinID { + return domain.ChannelMessage{}, false + } + return cloneChannelMessage(msg), true +} + func (s *ChannelStore) topicHasVisibleMessagesLocked(channelID int64, topicID int) bool { for _, msg := range s.messages[channelID] { if msg.Deleted { diff --git a/internal/store/memory/channel_store.go b/internal/store/memory/channel_store.go index 5978f3c3..3ea05e00 100644 --- a/internal/store/memory/channel_store.go +++ b/internal/store/memory/channel_store.go @@ -73,14 +73,19 @@ type ChannelStore struct { messages map[int64][]domain.ChannelMessage reactions map[int64]map[int]map[int64][]domain.ChannelMessagePeerReaction // paidReactions 是 per-(channel,message,user) 付费 reaction 累计星数 + 匿名标志。 - paidReactions map[int64]map[int]map[int64]memoryPaidReaction - top map[int64]map[string]domain.TopMessageReaction - recent map[int64]map[string]domain.RecentMessageReaction - mentions map[int64]map[int64]map[int]memoryMention - msgViews map[int64]map[int]int - msgViewers map[int64]map[int]map[int64]struct{} - events map[int64][]domain.ChannelUpdateEvent - retention map[int64]domain.ChannelUpdateRetentionCheckpoint + paidReactions map[int64]map[int]map[int64]memoryPaidReaction + top map[int64]map[string]domain.TopMessageReaction + recent map[int64]map[string]domain.RecentMessageReaction + mentions map[int64]map[int64]map[int]memoryMention + msgViews map[int64]map[int]int + msgViewers map[int64]map[int]map[int64]struct{} + events map[int64][]domain.ChannelUpdateEvent + retention map[int64]domain.ChannelUpdateRetentionCheckpoint + // historyClearDates is the no-PTS recovery timestamp for a future + // owner-local clear, keyed by channel then user. The member remains the + // absolute boundary authority; this map only makes account difference + // discovery bounded without scanning messages. + historyClearDates map[int64]map[int64]int adminLogs map[int64][]domain.ChannelAdminLogEvent invites map[string]domain.ChannelInvite importers map[int64]map[int64]domain.ChannelInviteImporter @@ -137,6 +142,7 @@ func NewChannelStore() *ChannelStore { msgViewers: make(map[int64]map[int]map[int64]struct{}), events: make(map[int64][]domain.ChannelUpdateEvent), retention: make(map[int64]domain.ChannelUpdateRetentionCheckpoint), + historyClearDates: make(map[int64]map[int64]int), adminLogs: make(map[int64][]domain.ChannelAdminLogEvent), invites: make(map[string]domain.ChannelInvite), importers: make(map[int64]map[int64]domain.ChannelInviteImporter), diff --git a/internal/store/memory/channel_test.go b/internal/store/memory/channel_test.go index f1455157..0e7a3580 100644 --- a/internal/store/memory/channel_test.go +++ b/internal/store/memory/channel_test.go @@ -873,6 +873,7 @@ func TestChannelDeleteHistoryLocalClearReturnsMonotonicAvailableMinID(t *testing CreatorUserID: 1, Title: "monotonic local clear", Megagroup: true, + MemberUserIDs: []int64{2}, Date: 1_700_000_250, }) if err != nil { @@ -911,6 +912,20 @@ func TestChannelDeleteHistoryLocalClearReturnsMonotonicAvailableMinID(t *testing if high.AvailableMinID != second.Message.ID { t.Fatalf("high available_min_id = %d, want %d", high.AvailableMinID, second.Message.ID) } + if !high.AvailableMinChanged { + t.Fatal("high clear did not report an advanced owner-local boundary") + } + dirtyAfterClear, err := store.ListDirtyActiveChannelsForUser(ctx, 1, 1_700_000_253, 0, 10) + if err != nil { + t.Fatalf("list dirty channels after owner-local clear: %v", err) + } + if len(dirtyAfterClear) != 1 || + dirtyAfterClear[0].ChannelID != created.Channel.ID || + dirtyAfterClear[0].AvailableMinID != second.Message.ID || + dirtyAfterClear[0].HistoryClearDate != 1_700_000_253 || + dirtyAfterClear[0].ChannelUpdatesDirty { + t.Fatalf("dirty owner-local clear = %+v, want only absolute boundary %d at date 1700000253", dirtyAfterClear, second.Message.ID) + } stale, err := store.DeleteChannelHistory(ctx, domain.DeleteChannelHistoryRequest{ UserID: 1, @@ -924,13 +939,38 @@ func TestChannelDeleteHistoryLocalClearReturnsMonotonicAvailableMinID(t *testing if stale.AvailableMinID != second.Message.ID { t.Fatalf("stale available_min_id = %d, want monotonic %d", stale.AvailableMinID, second.Message.ID) } + if stale.AvailableMinChanged { + t.Fatal("stale clear unexpectedly replaced the owner-local anchor") + } + dirtyAfterStale, err := store.ListDirtyActiveChannelsForUser(ctx, 1, 1_700_000_254, 0, 10) + if err != nil { + t.Fatalf("list dirty channels after stale owner-local clear: %v", err) + } + if len(dirtyAfterStale) != 0 { + t.Fatalf("stale clear refreshed recovery timestamp: %+v", dirtyAfterStale) + } history, err := store.ListChannelHistory(ctx, 1, domain.ChannelHistoryFilter{ChannelID: created.Channel.ID, Limit: 10}) if err != nil { t.Fatalf("list history: %v", err) } if len(history.Messages) != 0 { - t.Fatalf("history after stale clear = %+v, want no visible messages", history.Messages) + t.Fatalf("unprojected history after stale clear = %+v, want no shared messages", history.Messages) + } + history, err = store.ListChannelHistory(ctx, 1, domain.ChannelHistoryFilter{ + ChannelID: created.Channel.ID, + Limit: 10, + IncludeHistoryClearAnchor: true, + }) + if err != nil { + t.Fatalf("list projected history: %v", err) + } + if len(history.Messages) != 1 || + history.Messages[0].ID != second.Message.ID || + !domain.IsChannelHistoryClearMessage(history.Messages[0]) || + history.Messages[0].Body != "" || + history.Messages[0].Media != nil { + t.Fatalf("projected history after stale clear = %+v, want sanitized history-clear anchor %d", history.Messages, second.Message.ID) } dialogs, err := store.GetChannelDialogs(ctx, 1, []int64{created.Channel.ID}) if err != nil { @@ -939,8 +979,140 @@ func TestChannelDeleteHistoryLocalClearReturnsMonotonicAvailableMinID(t *testing if len(dialogs.Dialogs) != 1 { t.Fatalf("dialogs = %+v, want one dialog", dialogs.Dialogs) } - if dialogs.Dialogs[0].TopMessage != 0 || dialogs.Dialogs[0].ReadInboxMaxID != second.Message.ID || dialogs.Dialogs[0].UnreadCount != 0 { - t.Fatalf("dialog after stale clear = %+v, want top=0 read=%d unread=0", dialogs.Dialogs[0], second.Message.ID) + if dialogs.Dialogs[0].TopMessage != second.Message.ID || + dialogs.Dialogs[0].ReadInboxMaxID != second.Message.ID || + dialogs.Dialogs[0].UnreadCount != 0 || + len(dialogs.Messages) != 1 || + !domain.IsChannelHistoryClearMessage(dialogs.Messages[0]) { + t.Fatalf("dialog after stale clear = %+v messages=%+v, want anchored top=%d read=%d unread=0", dialogs.Dialogs[0], dialogs.Messages, second.Message.ID, second.Message.ID) + } + otherDialogs, err := store.GetChannelDialogs(ctx, 2, []int64{created.Channel.ID}) + if err != nil { + t.Fatalf("get other member dialog: %v", err) + } + if len(otherDialogs.Dialogs) != 1 || + otherDialogs.Dialogs[0].TopMessage != second.Message.ID || + len(otherDialogs.Messages) != 1 || + otherDialogs.Messages[0].Body != "second visible" || + otherDialogs.Messages[0].Action != nil { + t.Fatalf("other member projection changed by owner clear: dialogs=%+v messages=%+v", otherDialogs.Dialogs, otherDialogs.Messages) + } +} + +func TestChannelDeleteHistoryLocalClearKeepsMegagroupAndBroadcastDialogs(t *testing.T) { + for _, tc := range []struct { + name string + megagroup bool + broadcast bool + }{ + {name: "megagroup", megagroup: true}, + {name: "broadcast", broadcast: true}, + } { + t.Run(tc.name, func(t *testing.T) { + ctx := context.Background() + store := NewChannelStore() + created, err := store.CreateChannel(ctx, domain.CreateChannelRequest{ + CreatorUserID: 11, + Title: tc.name + " local clear", + Megagroup: tc.megagroup, + Broadcast: tc.broadcast, + Date: 1_700_000_270, + }) + if err != nil { + t.Fatalf("create channel: %v", err) + } + sent, err := store.SendChannelMessage(ctx, domain.SendChannelMessageRequest{ + UserID: 11, + ChannelID: created.Channel.ID, + RandomID: 31_001, + Message: "clear this", + Date: 1_700_000_271, + }) + if err != nil { + t.Fatalf("send channel message: %v", err) + } + channelPts := sent.Channel.Pts + cleared, err := store.DeleteChannelHistory(ctx, domain.DeleteChannelHistoryRequest{ + UserID: 11, + ChannelID: created.Channel.ID, + MaxID: sent.Message.ID, + Date: 1_700_000_272, + }) + if err != nil { + t.Fatalf("clear local history: %v", err) + } + if cleared.Channel.Pts != channelPts || cleared.Event.Pts != 0 { + t.Fatalf("local clear changed channel pts: before=%d result=%+v", channelPts, cleared) + } + dialogs, err := store.GetChannelDialogs(ctx, 11, []int64{created.Channel.ID}) + if err != nil { + t.Fatalf("get dialogs after clear: %v", err) + } + if len(dialogs.Dialogs) != 1 || + dialogs.Dialogs[0].TopMessage != sent.Message.ID || + len(dialogs.Messages) != 1 || + !domain.IsChannelHistoryClearMessage(dialogs.Messages[0]) { + t.Fatalf("dialog disappeared after %s clear: dialogs=%+v messages=%+v", tc.name, dialogs.Dialogs, dialogs.Messages) + } + }) + } +} + +func TestChannelPrehistoryBoundaryDoesNotCreateHistoryClearAnchor(t *testing.T) { + ctx := context.Background() + store := NewChannelStore() + created, err := store.CreateChannel(ctx, domain.CreateChannelRequest{ + CreatorUserID: 21, + Title: "hidden prehistory", + Megagroup: true, + Date: 1_700_000_280, + }) + if err != nil { + t.Fatalf("create channel: %v", err) + } + old, err := store.SendChannelMessage(ctx, domain.SendChannelMessageRequest{ + UserID: 21, + ChannelID: created.Channel.ID, + RandomID: 32_001, + Message: "prehistory", + Date: 1_700_000_281, + }) + if err != nil { + t.Fatalf("send prehistory message: %v", err) + } + if _, err := store.SetPreHistoryHidden(ctx, 21, created.Channel.ID, true); err != nil { + t.Fatalf("hide prehistory: %v", err) + } + invited, err := store.InviteToChannel(ctx, created.Channel.ID, 21, []int64{22}, 1_700_000_282) + if err != nil { + t.Fatalf("invite member: %v", err) + } + if len(invited.Members) != 1 || + invited.Members[0].AvailableMinID != old.Message.ID || + invited.Members[0].HistoryClearAnchorID != 0 || + invited.Members[0].HistoryClearAnchorDate != 0 { + t.Fatalf("invited member = %+v, want prehistory boundary without local-clear anchor", invited.Members) + } + + history, err := store.ListChannelHistory(ctx, 22, domain.ChannelHistoryFilter{ + ChannelID: created.Channel.ID, + Limit: 10, + IncludeHistoryClearAnchor: true, + }) + if err != nil { + t.Fatalf("list invited member history: %v", err) + } + for _, message := range history.Messages { + if message.ID == old.Message.ID || domain.IsChannelHistoryClearMessage(message) { + t.Fatalf("prehistory boundary leaked a local-clear marker: %+v", history.Messages) + } + } + byID, err := store.GetChannelMessages(ctx, 22, created.Channel.ID, []int{old.Message.ID}) + if err != nil { + t.Fatalf("get prehistory message by id: %v", err) + } + if len(byID.Messages) != 0 { + t.Fatalf("prehistory message by id = %+v, want hidden without fabricated marker", byID.Messages) } } diff --git a/internal/store/postgres/channel_core.go b/internal/store/postgres/channel_core.go index 5458e3e9..1369293d 100644 --- a/internal/store/postgres/channel_core.go +++ b/internal/store/postgres/channel_core.go @@ -317,6 +317,7 @@ func (s *ChannelStore) GetChannels(ctx context.Context, viewerUserID int64, chan SELECT `+channelColumns+`, m.channel_id, m.user_id, m.inviter_user_id, m.role, m.status, m.joined_at, m.left_at, m.admin_rights::text, m.banned_rights::text, m.rank, m.available_min_id, m.available_min_pts, + m.history_clear_anchor_id, m.history_clear_anchor_date, m.read_inbox_max_id, m.read_outbox_max_id, m.unread_mark, m.slowmode_last_send_date FROM channels c JOIN channel_members m ON m.channel_id = c.id AND m.user_id = $1 @@ -529,7 +530,8 @@ func refreshChannelCountsTx(ctx context.Context, tx pgx.Tx, channel domain.Chann var participants, admins, kicked, banned int rows, err := tx.Query(ctx, ` SELECT channel_id, user_id, inviter_user_id, role, status, joined_at, left_at, admin_rights::text, banned_rights::text, - rank, available_min_id, available_min_pts, read_inbox_max_id, read_outbox_max_id, unread_mark, slowmode_last_send_date + rank, available_min_id, available_min_pts, history_clear_anchor_id, history_clear_anchor_date, + read_inbox_max_id, read_outbox_max_id, unread_mark, slowmode_last_send_date FROM channel_members WHERE channel_id = $1`, channel.ID) if err != nil { diff --git a/internal/store/postgres/channel_dialogs.go b/internal/store/postgres/channel_dialogs.go index 50a5baa7..413eb2a9 100644 --- a/internal/store/postgres/channel_dialogs.go +++ b/internal/store/postgres/channel_dialogs.go @@ -18,6 +18,26 @@ type channelDialogListItem struct { defaultSendAs *domain.Peer } +func channelDialogVisibleTopIDSQL() string { + return `CASE + WHEN c.top_message_id > m.available_min_id THEN c.top_message_id + WHEN m.history_clear_anchor_id > 0 + AND m.history_clear_anchor_id = m.available_min_id + THEN m.history_clear_anchor_id + ELSE 0 +END` +} + +func channelDialogVisibleTopDateSQL(globalTopDate, emptyFallback string) string { + return `CASE + WHEN c.top_message_id > m.available_min_id THEN ` + globalTopDate + ` + WHEN m.history_clear_anchor_id > 0 + AND m.history_clear_anchor_id = m.available_min_id + THEN m.history_clear_anchor_date + ELSE ` + emptyFallback + ` +END` +} + func (s *ChannelStore) ListChannelDialogs(ctx context.Context, viewerUserID int64, filter domain.DialogFilter) (domain.ChannelDialogList, error) { if viewerUserID == 0 { return domain.ChannelDialogList{}, nil @@ -33,8 +53,8 @@ func (s *ChannelStore) ListChannelDialogs(ctx context.Context, viewerUserID int6 if len(channelIDs) == 0 { return domain.ChannelDialogList{}, nil } - visibleTopID := "CASE WHEN c.top_message_id > m.available_min_id THEN c.top_message_id ELSE 0 END" - visibleTopDate := "CASE WHEN c.top_message_id > m.available_min_id THEN COALESCE(top_msg.message_date, d.top_message_date, c.date) ELSE 0 END" + visibleTopID := channelDialogVisibleTopIDSQL() + visibleTopDate := channelDialogVisibleTopDateSQL("COALESCE(top_msg.message_date, d.top_message_date, c.date)", "0") visibleReadInbox := "GREATEST(COALESCE(d.read_inbox_max_id, 0), m.read_inbox_max_id)" visibleUnreadCount := channelDialogVisibleUnreadCountSQL(visibleReadInbox, visibleTopID) args := []any{viewerUserID, channelIDs} @@ -139,7 +159,9 @@ SELECT `+channelColumns+`, COALESCE(d.view_forum_as_messages, false), COALESCE(d.has_scheduled, false), d.default_send_as_peer_type, - d.default_send_as_peer_id + d.default_send_as_peer_id, + m.history_clear_anchor_id, + m.history_clear_anchor_date FROM channel_members m JOIN channels c ON c.id = m.channel_id AND c.id = ANY($2::bigint[]) AND NOT c.deleted LEFT JOIN channel_messages top_msg ON top_msg.channel_id = m.channel_id AND top_msg.channel_id = ANY($2::bigint[]) AND top_msg.id = c.top_message_id AND NOT top_msg.deleted @@ -232,6 +254,7 @@ LIMIT `+limitArg, args...) if err := s.populateChannelMessagesReactions(ctx, s.db, viewerUserID, out.Channels, out.Messages); err != nil { return domain.ChannelDialogList{}, err } + projectChannelDialogHistoryClearMessages(out.Dialogs, out.Messages) return out, nil } @@ -255,7 +278,9 @@ SELECT `+channelColumns+`, COALESCE(d.view_forum_as_messages, false), COALESCE(d.has_scheduled, false), d.default_send_as_peer_type, - d.default_send_as_peer_id + d.default_send_as_peer_id, + 0, + 0 FROM user_channel_member_index i JOIN channel_members pm ON pm.user_id = i.user_id AND pm.channel_id = i.channel_id JOIN channels parent ON parent.id = i.channel_id AND parent.id = ANY($2::bigint[]) AND parent.broadcast AND parent.linked_monoforum_id <> 0 AND NOT parent.deleted @@ -357,6 +382,20 @@ WHERE `+where.String(), args...) if err := rows.Err(); err != nil { return nil, fmt.Errorf("scan channel dialog top messages: %w", err) } + for _, dialog := range dialogs { + if dialog.Peer.Type != domain.PeerTypeChannel || + dialog.TopMessage <= 0 || + dialog.TopMessage != dialog.HistoryClearAnchorID { + continue + } + key := channelMessageLookupKey{channelID: dialog.Peer.ID, id: dialog.TopMessage} + out[key] = domain.ProjectChannelHistoryClearMessage( + out[key], + dialog.Peer.ID, + dialog.HistoryClearAnchorID, + dialog.HistoryClearAnchorDate, + ) + } return out, nil } @@ -413,6 +452,14 @@ func (s *ChannelStore) GetChannelDialogs(ctx context.Context, viewerUserID int64 } } msg, _ := s.getChannelMessage(ctx, s.db, channelID, dialog.TopMessageID) + if dialog.TopMessageID > 0 && dialog.TopMessageID == dialog.HistoryClearAnchorID { + msg = domain.ProjectChannelHistoryClearMessage( + msg, + channelID, + dialog.HistoryClearAnchorID, + dialog.HistoryClearAnchorDate, + ) + } if msg.ID != 0 { dialog.TopMessageDate = msg.Date out.Messages = append(out.Messages, msg) @@ -425,9 +472,34 @@ func (s *ChannelStore) GetChannelDialogs(ctx context.Context, viewerUserID int64 if err := s.populateChannelMessagesReactions(ctx, s.db, viewerUserID, out.Channels, out.Messages); err != nil { return domain.ChannelDialogList{}, err } + projectChannelDialogHistoryClearMessages(out.Dialogs, out.Messages) return out, nil } +func projectChannelDialogHistoryClearMessages(dialogs []domain.Dialog, messages []domain.ChannelMessage) { + anchors := make(map[channelMessageLookupKey]domain.Dialog) + for _, dialog := range dialogs { + if dialog.Peer.Type != domain.PeerTypeChannel || + dialog.TopMessage <= 0 || + dialog.TopMessage != dialog.HistoryClearAnchorID { + continue + } + anchors[channelMessageLookupKey{channelID: dialog.Peer.ID, id: dialog.TopMessage}] = dialog + } + for i := range messages { + dialog, ok := anchors[channelMessageLookupKey{channelID: messages[i].ChannelID, id: messages[i].ID}] + if !ok { + continue + } + messages[i] = domain.ProjectChannelHistoryClearMessage( + messages[i], + dialog.Peer.ID, + dialog.HistoryClearAnchorID, + dialog.HistoryClearAnchorDate, + ) + } +} + func (s *ChannelStore) ListCommonChannels(ctx context.Context, req domain.CommonChannelsRequest) (domain.CommonChannelsResult, error) { if req.UserID == 0 || req.TargetUserID == 0 || req.UserID == req.TargetUserID || req.MaxID < 0 { return domain.CommonChannelsResult{}, domain.ErrChannelInvalid @@ -553,6 +625,7 @@ func (s *ChannelStore) leftChannelsByIDs(ctx context.Context, userID int64, ids rows, err := s.db.Query(ctx, ` SELECT channel_id, user_id, inviter_user_id, role, status, joined_at, left_at, admin_rights::text, banned_rights::text, rank, available_min_id, available_min_pts, + history_clear_anchor_id, history_clear_anchor_date, read_inbox_max_id, read_outbox_max_id, unread_mark, slowmode_last_send_date FROM channel_members WHERE user_id = $1 @@ -603,8 +676,8 @@ func (s *ChannelStore) ListInactiveChannels(ctx context.Context, userID int64, l if len(channelIDs) == 0 { return domain.ChannelDialogList{}, nil } - visibleTopID := "CASE WHEN c.top_message_id > m.available_min_id THEN c.top_message_id ELSE 0 END" - visibleTopDate := "CASE WHEN c.top_message_id > m.available_min_id THEN COALESCE(top_msg.message_date, d.top_message_date, c.date) ELSE GREATEST(c.date, m.joined_at) END" + visibleTopID := channelDialogVisibleTopIDSQL() + visibleTopDate := channelDialogVisibleTopDateSQL("COALESCE(top_msg.message_date, d.top_message_date, c.date)", "GREATEST(c.date, m.joined_at)") visibleReadInbox := "GREATEST(COALESCE(d.read_inbox_max_id, 0), m.read_inbox_max_id)" visibleUnreadCount := channelDialogVisibleUnreadCountSQL(visibleReadInbox, visibleTopID) rows, err := s.db.Query(ctx, ` @@ -619,7 +692,9 @@ SELECT `+channelColumns+`, COALESCE(d.view_forum_as_messages, false), COALESCE(d.has_scheduled, false), d.default_send_as_peer_type, - d.default_send_as_peer_id + d.default_send_as_peer_id, + m.history_clear_anchor_id, + m.history_clear_anchor_date FROM channel_members m JOIN channels c ON c.id = m.channel_id AND c.id = ANY($2::bigint[]) AND NOT c.deleted LEFT JOIN channel_messages top_msg ON top_msg.channel_id = m.channel_id AND top_msg.channel_id = ANY($2::bigint[]) AND top_msg.id = c.top_message_id AND NOT top_msg.deleted @@ -856,6 +931,7 @@ func (s *ChannelStore) EditChannelPeerFolders(ctx context.Context, userID int64, // 归档必须 ensure-INSERT：从未读过/置顶过的频道还没有 dialog 行， // 只 UPDATE 会让归档静默丢失；新行同时带上 member 真实水位， // 避免 0 水位缓存行遮蔽未读/已读状态。 + visibleTopID := channelDialogVisibleTopIDSQL() if _, err := s.db.Exec(ctx, ` WITH requested AS ( SELECT ($2::text[])[i] AS peer_type, ($3::bigint[])[i] AS channel_id, ($4::int[])[i] AS folder_id @@ -870,7 +946,7 @@ deduped AS ( ) INSERT INTO channel_dialogs (user_id, channel_id, folder_id, top_message_id, read_inbox_max_id, read_outbox_max_id, unread_count, updated_at) SELECT $1, m.channel_id, deduped.folder_id, - CASE WHEN c.top_message_id > m.available_min_id THEN c.top_message_id ELSE 0 END, + `+visibleTopID+`, m.read_inbox_max_id, m.read_outbox_max_id, ( SELECT COUNT(*)::int @@ -904,7 +980,7 @@ func (s *ChannelStore) CountChannelArchiveUnread(ctx context.Context, userID int // JOIN active member：退群残留的 channel_dialogs 行不计入归档徽章。 // unread 读时动态派生（H4a）：不再消费 channel_dialogs.unread_count 缓存列；归档集合 // 有界（需显式归档建行），每行动态 COUNT 已被 cap 钳制。 - visibleTopID := "CASE WHEN c.top_message_id > m.available_min_id THEN c.top_message_id ELSE 0 END" + visibleTopID := channelDialogVisibleTopIDSQL() visibleReadInbox := "GREATEST(COALESCE(d.read_inbox_max_id, 0), m.read_inbox_max_id)" if err := s.db.QueryRow(ctx, ` SELECT @@ -937,7 +1013,7 @@ func (s *ChannelStore) getChannelDialogUncached(ctx context.Context, db sqlcgen. dialog := domain.ChannelDialog{UserID: userID, ChannelID: channel.ID, TopMessageID: channel.TopMessageID} var defaultSendAsType sql.NullString var defaultSendAsID sql.NullInt64 - visibleTopID := "CASE WHEN c.top_message_id > m.available_min_id THEN c.top_message_id ELSE 0 END" + visibleTopID := channelDialogVisibleTopIDSQL() visibleReadInbox := "GREATEST(COALESCE(d.read_inbox_max_id, 0), m.read_inbox_max_id)" visibleUnreadCount := channelDialogVisibleUnreadCountSQL(visibleReadInbox, visibleTopID) // 单频道 TopMessageDate 用 LEFT JOIN top_msg 直接取,替代此前对每个频道再单查一次 @@ -946,7 +1022,10 @@ func (s *ChannelStore) getChannelDialogUncached(ctx context.Context, db sqlcgen. // message_date(不过滤 deleted,与原 getChannelMessage 一致),否则回退 // COALESCE(d.top_message_date, c.date)。注意:与批量版 getChannelDialogs 的隐藏态 // (ELSE 0)刻意保持各自既有差异,本改动只去往返、不改单频道输出。 - visibleTopDate := "CASE WHEN c.top_message_id > m.available_min_id AND top_msg.id IS NOT NULL THEN top_msg.message_date ELSE COALESCE(d.top_message_date, c.date) END" + visibleTopDate := channelDialogVisibleTopDateSQL( + "CASE WHEN top_msg.id IS NOT NULL THEN top_msg.message_date ELSE COALESCE(d.top_message_date, c.date) END", + "COALESCE(d.top_message_date, c.date)", + ) err := db.QueryRow(ctx, ` SELECT `+visibleTopID+`, `+visibleTopDate+`, @@ -962,7 +1041,9 @@ SELECT `+visibleTopID+`, COALESCE(d.view_forum_as_messages, false), COALESCE(d.has_scheduled, false), d.default_send_as_peer_type, - d.default_send_as_peer_id + d.default_send_as_peer_id, + m.history_clear_anchor_id, + m.history_clear_anchor_date FROM channels c JOIN channel_members m ON m.channel_id = c.id AND m.user_id = $1 LEFT JOIN channel_messages top_msg ON top_msg.channel_id = c.id AND top_msg.id = c.top_message_id @@ -983,6 +1064,8 @@ WHERE c.id = $2`, userID, channel.ID).Scan( &dialog.HasScheduled, &defaultSendAsType, &defaultSendAsID, + &dialog.HistoryClearAnchorID, + &dialog.HistoryClearAnchorDate, ) if err != nil && !errors.Is(err, pgx.ErrNoRows) { return domain.ChannelDialog{}, fmt.Errorf("get channel dialog: %w", err) @@ -1037,8 +1120,8 @@ func (s *ChannelStore) getChannelDialogsUncached(ctx context.Context, db sqlcgen if userID == 0 || len(channelIDs) == 0 { return nil, nil } - visibleTopID := "CASE WHEN c.top_message_id > m.available_min_id THEN c.top_message_id ELSE 0 END" - visibleTopDate := "CASE WHEN c.top_message_id > m.available_min_id THEN COALESCE(top_msg.message_date, d.top_message_date, c.date) ELSE 0 END" + visibleTopID := channelDialogVisibleTopIDSQL() + visibleTopDate := channelDialogVisibleTopDateSQL("COALESCE(top_msg.message_date, d.top_message_date, c.date)", "0") visibleReadInbox := "GREATEST(COALESCE(d.read_inbox_max_id, 0), m.read_inbox_max_id)" visibleUnreadCount := channelDialogVisibleUnreadCountSQL(visibleReadInbox, visibleTopID) rows, err := db.Query(ctx, ` @@ -1057,7 +1140,9 @@ SELECT c.id, COALESCE(d.view_forum_as_messages, false), COALESCE(d.has_scheduled, false), d.default_send_as_peer_type, - d.default_send_as_peer_id + d.default_send_as_peer_id, + m.history_clear_anchor_id, + m.history_clear_anchor_date FROM channels c JOIN channel_members m ON m.channel_id = c.id AND m.user_id = $1 LEFT JOIN channel_messages top_msg ON top_msg.channel_id = c.id AND top_msg.id = c.top_message_id AND NOT top_msg.deleted @@ -1089,6 +1174,8 @@ WHERE c.id = ANY($2::bigint[])`, userID, channelIDs) &dialog.HasScheduled, &defaultSendAsType, &defaultSendAsID, + &dialog.HistoryClearAnchorID, + &dialog.HistoryClearAnchorDate, ); err != nil { return nil, err } @@ -1197,6 +1284,7 @@ func scanChannelDialogRow(row rowScanner, userID int64) (domain.Channel, domain. var rights, reactionPolicy string var wallpaper *string var topID, topDate, folderID, readInbox, readOutbox, unreadCount, pinnedOrder, unreadMentions, unreadReactions int + var historyClearAnchorID, historyClearAnchorDate int var pinned, unreadMark, viewForumAsMessages, hasScheduled bool var defaultSendAsType sql.NullString var defaultSendAsID sql.NullInt64 @@ -1204,27 +1292,30 @@ func scanChannelDialogRow(row rowScanner, userID int64) (domain.Channel, domain. &topID, &topDate, &folderID, &readInbox, &readOutbox, &unreadCount, &pinned, &pinnedOrder, &unreadMark, &unreadMentions, &unreadReactions, &viewForumAsMessages, &hasScheduled, &defaultSendAsType, &defaultSendAsID, + &historyClearAnchorID, &historyClearAnchorDate, ) if err := row.Scan(dest...); err != nil { return domain.Channel{}, domain.Dialog{}, nil, err } finishChannelScan(&ch, rights, reactionPolicy, wallpaper) dialog := domain.Dialog{ - Peer: domain.Peer{Type: domain.PeerTypeChannel, ID: ch.ID}, - FolderID: folderID, - TopMessage: topID, - TopMessageDate: topDate, - ReadInboxMaxID: readInbox, - ReadOutboxMaxID: readOutbox, - UnreadCount: unreadCount, - UnreadMentions: unreadMentions, - UnreadReactions: unreadReactions, - Pinned: pinned, - PinnedOrder: pinnedOrder, - UnreadMark: unreadMark, - ViewForumAsMessages: viewForumAsMessages, - HasScheduled: hasScheduled, - Pts: ch.Pts, + Peer: domain.Peer{Type: domain.PeerTypeChannel, ID: ch.ID}, + FolderID: folderID, + TopMessage: topID, + TopMessageDate: topDate, + HistoryClearAnchorID: historyClearAnchorID, + HistoryClearAnchorDate: historyClearAnchorDate, + ReadInboxMaxID: readInbox, + ReadOutboxMaxID: readOutbox, + UnreadCount: unreadCount, + UnreadMentions: unreadMentions, + UnreadReactions: unreadReactions, + Pinned: pinned, + PinnedOrder: pinnedOrder, + UnreadMark: unreadMark, + ViewForumAsMessages: viewForumAsMessages, + HasScheduled: hasScheduled, + Pts: ch.Pts, } var defaultSendAs *domain.Peer if defaultSendAsType.Valid && defaultSendAsID.Valid && defaultSendAsID.Int64 != 0 { @@ -1236,41 +1327,45 @@ func scanChannelDialogRow(row rowScanner, userID int64) (domain.Channel, domain. func channelDialogToDialog(dialog domain.ChannelDialog, channelPts int) domain.Dialog { return domain.Dialog{ - Peer: domain.Peer{Type: domain.PeerTypeChannel, ID: dialog.ChannelID}, - FolderID: dialog.FolderID, - TopMessage: dialog.TopMessageID, - TopMessageDate: dialog.TopMessageDate, - ReadInboxMaxID: dialog.ReadInboxMaxID, - ReadOutboxMaxID: dialog.ReadOutboxMaxID, - UnreadCount: dialog.UnreadCount, - UnreadMentions: dialog.UnreadMentions, - UnreadReactions: dialog.UnreadReactions, - Pinned: dialog.Pinned, - PinnedOrder: dialog.PinnedOrder, - UnreadMark: dialog.UnreadMark, - ViewForumAsMessages: dialog.ViewForumAsMessages, - HasScheduled: dialog.HasScheduled, - Pts: channelPts, + Peer: domain.Peer{Type: domain.PeerTypeChannel, ID: dialog.ChannelID}, + FolderID: dialog.FolderID, + TopMessage: dialog.TopMessageID, + TopMessageDate: dialog.TopMessageDate, + HistoryClearAnchorID: dialog.HistoryClearAnchorID, + HistoryClearAnchorDate: dialog.HistoryClearAnchorDate, + ReadInboxMaxID: dialog.ReadInboxMaxID, + ReadOutboxMaxID: dialog.ReadOutboxMaxID, + UnreadCount: dialog.UnreadCount, + UnreadMentions: dialog.UnreadMentions, + UnreadReactions: dialog.UnreadReactions, + Pinned: dialog.Pinned, + PinnedOrder: dialog.PinnedOrder, + UnreadMark: dialog.UnreadMark, + ViewForumAsMessages: dialog.ViewForumAsMessages, + HasScheduled: dialog.HasScheduled, + Pts: channelPts, } } func channelDialogFromDialog(userID int64, dialog domain.Dialog) domain.ChannelDialog { return domain.ChannelDialog{ - UserID: userID, - ChannelID: dialog.Peer.ID, - FolderID: dialog.FolderID, - TopMessageID: dialog.TopMessage, - TopMessageDate: dialog.TopMessageDate, - ReadInboxMaxID: dialog.ReadInboxMaxID, - ReadOutboxMaxID: dialog.ReadOutboxMaxID, - UnreadCount: dialog.UnreadCount, - UnreadMentions: dialog.UnreadMentions, - UnreadReactions: dialog.UnreadReactions, - Pinned: dialog.Pinned, - PinnedOrder: dialog.PinnedOrder, - UnreadMark: dialog.UnreadMark, - ViewForumAsMessages: dialog.ViewForumAsMessages, - HasScheduled: dialog.HasScheduled, + UserID: userID, + ChannelID: dialog.Peer.ID, + FolderID: dialog.FolderID, + TopMessageID: dialog.TopMessage, + TopMessageDate: dialog.TopMessageDate, + HistoryClearAnchorID: dialog.HistoryClearAnchorID, + HistoryClearAnchorDate: dialog.HistoryClearAnchorDate, + ReadInboxMaxID: dialog.ReadInboxMaxID, + ReadOutboxMaxID: dialog.ReadOutboxMaxID, + UnreadCount: dialog.UnreadCount, + UnreadMentions: dialog.UnreadMentions, + UnreadReactions: dialog.UnreadReactions, + Pinned: dialog.Pinned, + PinnedOrder: dialog.PinnedOrder, + UnreadMark: dialog.UnreadMark, + ViewForumAsMessages: dialog.ViewForumAsMessages, + HasScheduled: dialog.HasScheduled, } } @@ -1392,14 +1487,23 @@ func channelFolderPeerIDs(primary []domain.DialogFolderPeer, rest ...[]domain.Di func previewChannelDialog(userID int64, channel domain.Channel, member domain.ChannelMember) domain.ChannelDialog { topMessageID := channel.TopMessageID if topMessageID <= member.AvailableMinID { - topMessageID = 0 + topMessageID = member.HistoryClearAnchorID + if topMessageID != member.AvailableMinID { + topMessageID = 0 + } + } + topMessageDate := channel.Date + if topMessageID > 0 && topMessageID == member.HistoryClearAnchorID { + topMessageDate = member.HistoryClearAnchorDate } return domain.ChannelDialog{ - UserID: userID, - ChannelID: channel.ID, - TopMessageID: topMessageID, - TopMessageDate: channel.Date, - ReadInboxMaxID: maxInt(channel.TopMessageID, member.ReadInboxMaxID), - ReadOutboxMaxID: maxInt(channel.TopMessageID, member.ReadOutboxMaxID), + UserID: userID, + ChannelID: channel.ID, + TopMessageID: topMessageID, + TopMessageDate: topMessageDate, + HistoryClearAnchorID: member.HistoryClearAnchorID, + HistoryClearAnchorDate: member.HistoryClearAnchorDate, + ReadInboxMaxID: maxInt(channel.TopMessageID, member.ReadInboxMaxID), + ReadOutboxMaxID: maxInt(channel.TopMessageID, member.ReadOutboxMaxID), } } diff --git a/internal/store/postgres/channel_helpers.go b/internal/store/postgres/channel_helpers.go index f70f6ea6..55484d9d 100644 --- a/internal/store/postgres/channel_helpers.go +++ b/internal/store/postgres/channel_helpers.go @@ -31,7 +31,10 @@ func (s *ChannelStore) SaveChannelDefaultSendAs(ctx context.Context, req domain. } topMessageID := channel.TopMessageID if topMessageID <= member.AvailableMinID { - topMessageID = 0 + topMessageID = member.HistoryClearAnchorID + if topMessageID != member.AvailableMinID { + topMessageID = 0 + } } if _, err := s.db.Exec(ctx, ` INSERT INTO channel_dialogs ( @@ -393,15 +396,59 @@ LIMIT $4`, userID, sinceDate, afterChannelID, limit) return nil, fmt.Errorf("list dirty active channels for user: %w", err) } defer rows.Close() - out := make([]domain.DirtyChannel, 0, limit) + byChannelID := make(map[int64]domain.DirtyChannel, limit*2) for rows.Next() { var item domain.DirtyChannel if err := rows.Scan(&item.ChannelID, &item.Pts); err != nil { return nil, err } + item.ChannelUpdatesDirty = true + byChannelID[item.ChannelID] = item + } + if err := rows.Err(); err != nil { + return nil, err + } + + clearRows, err := s.db.Query(ctx, ` +SELECT i.channel_id, c.pts, i.available_min_id, i.history_clear_updated_at +FROM user_channel_member_index i +JOIN channels c ON c.id = i.channel_id AND NOT c.deleted +WHERE i.user_id = $1 + AND i.status = 'active' + AND NOT i.deleted + AND i.channel_id > $3 + AND i.history_clear_anchor_id > 0 + AND i.history_clear_anchor_id = i.available_min_id + AND i.history_clear_updated_at >= $2 +ORDER BY i.channel_id ASC +LIMIT $4`, userID, sinceDate, afterChannelID, limit) + if err != nil { + return nil, fmt.Errorf("list owner-local channel history clears for user: %w", err) + } + defer clearRows.Close() + for clearRows.Next() { + var item domain.DirtyChannel + if err := clearRows.Scan(&item.ChannelID, &item.Pts, &item.AvailableMinID, &item.HistoryClearDate); err != nil { + return nil, err + } + if existing, ok := byChannelID[item.ChannelID]; ok { + item.ChannelUpdatesDirty = existing.ChannelUpdatesDirty + } + byChannelID[item.ChannelID] = item + } + if err := clearRows.Err(); err != nil { + return nil, err + } + + out := make([]domain.DirtyChannel, 0, len(byChannelID)) + for _, item := range byChannelID { out = append(out, item) } - return out, rows.Err() + sort.Slice(out, func(i, j int) bool { return out[i].ChannelID < out[j].ChannelID }) + if len(out) > limit { + out = out[:limit] + } + return out, nil } type rowScanner interface { diff --git a/internal/store/postgres/channel_member_helpers.go b/internal/store/postgres/channel_member_helpers.go index 9d4f9360..1629fac1 100644 --- a/internal/store/postgres/channel_member_helpers.go +++ b/internal/store/postgres/channel_member_helpers.go @@ -189,6 +189,7 @@ func getChannelMemberByID(ctx context.Context, db sqlcgen.DBTX, channelID, userI row := db.QueryRow(ctx, ` SELECT channel_id, user_id, inviter_user_id, role, status, joined_at, left_at, admin_rights::text, banned_rights::text, rank, available_min_id, available_min_pts, + history_clear_anchor_id, history_clear_anchor_date, read_inbox_max_id, read_outbox_max_id, unread_mark, slowmode_last_send_date FROM channel_members WHERE channel_id = $1 AND user_id = $2`, channelID, userID) @@ -214,8 +215,9 @@ func upsertChannelMemberTx(ctx context.Context, tx pgx.Tx, channel domain.Channe if _, err := tx.Exec(ctx, ` INSERT INTO channel_members ( channel_id, user_id, inviter_user_id, role, status, joined_at, left_at, admin_rights, banned_rights, - rank, available_min_id, available_min_pts, read_inbox_max_id, read_outbox_max_id, unread_mark, slowmode_last_send_date -) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16) + rank, available_min_id, available_min_pts, history_clear_anchor_id, history_clear_anchor_date, + read_inbox_max_id, read_outbox_max_id, unread_mark, slowmode_last_send_date +) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18) ON CONFLICT (channel_id, user_id) DO UPDATE SET inviter_user_id = EXCLUDED.inviter_user_id, role = EXCLUDED.role, @@ -231,7 +233,8 @@ ON CONFLICT (channel_id, user_id) DO UPDATE SET updated_at = now()`, member.ChannelID, member.UserID, member.InviterUserID, string(member.Role), string(member.Status), member.JoinedAt, member.LeftAt, adminRights, bannedRights, member.Rank, member.AvailableMinID, - member.AvailableMinPts, member.ReadInboxMaxID, member.ReadOutboxMaxID, member.UnreadMark, member.SlowmodeLastSendDate); err != nil { + member.AvailableMinPts, member.HistoryClearAnchorID, member.HistoryClearAnchorDate, + member.ReadInboxMaxID, member.ReadOutboxMaxID, member.UnreadMark, member.SlowmodeLastSendDate); err != nil { return fmt.Errorf("upsert channel member: %w", err) } return upsertUserChannelMemberIndexTx(ctx, tx, channel, member) @@ -316,7 +319,8 @@ func scanChannelWithMember(row rowScanner) (domain.Channel, domain.ChannelMember dest := append(channelScanDest(&ch, &defaultRights, &reactionPolicy, &wallpaper), &member.ChannelID, &member.UserID, &member.InviterUserID, &role, &status, &member.JoinedAt, &member.LeftAt, &adminRights, &bannedRights, &member.Rank, - &member.AvailableMinID, &member.AvailableMinPts, &member.ReadInboxMaxID, &member.ReadOutboxMaxID, &member.UnreadMark, &member.SlowmodeLastSendDate, + &member.AvailableMinID, &member.AvailableMinPts, &member.HistoryClearAnchorID, &member.HistoryClearAnchorDate, + &member.ReadInboxMaxID, &member.ReadOutboxMaxID, &member.UnreadMark, &member.SlowmodeLastSendDate, ) if err := row.Scan(dest...); err != nil { return domain.Channel{}, domain.ChannelMember{}, err @@ -351,7 +355,8 @@ func scanChannelMember(row rowScanner) (domain.ChannelMember, error) { if err := row.Scan( &member.ChannelID, &member.UserID, &member.InviterUserID, &role, &status, &member.JoinedAt, &member.LeftAt, &adminRights, &bannedRights, &member.Rank, - &member.AvailableMinID, &member.AvailableMinPts, &member.ReadInboxMaxID, &member.ReadOutboxMaxID, &member.UnreadMark, &member.SlowmodeLastSendDate, + &member.AvailableMinID, &member.AvailableMinPts, &member.HistoryClearAnchorID, &member.HistoryClearAnchorDate, + &member.ReadInboxMaxID, &member.ReadOutboxMaxID, &member.UnreadMark, &member.SlowmodeLastSendDate, ); err != nil { return domain.ChannelMember{}, err } @@ -370,7 +375,8 @@ func scanChannelMemberWithCount(row rowScanner) (domain.ChannelMember, int, erro if err := row.Scan( &member.ChannelID, &member.UserID, &member.InviterUserID, &role, &status, &member.JoinedAt, &member.LeftAt, &adminRights, &bannedRights, &member.Rank, - &member.AvailableMinID, &member.AvailableMinPts, &member.ReadInboxMaxID, &member.ReadOutboxMaxID, &member.UnreadMark, &member.SlowmodeLastSendDate, + &member.AvailableMinID, &member.AvailableMinPts, &member.HistoryClearAnchorID, &member.HistoryClearAnchorDate, + &member.ReadInboxMaxID, &member.ReadOutboxMaxID, &member.UnreadMark, &member.SlowmodeLastSendDate, &count, ); err != nil { return domain.ChannelMember{}, 0, err @@ -471,6 +477,8 @@ func syntheticMonoforumAdminMember(mono domain.Channel, parentMember domain.Chan } member.AvailableMinID = 0 member.AvailableMinPts = 0 + member.HistoryClearAnchorID = 0 + member.HistoryClearAnchorDate = 0 member.ReadInboxMaxID = mono.TopMessageID member.ReadOutboxMaxID = mono.TopMessageID member.UnreadMark = false diff --git a/internal/store/postgres/channel_member_join.go b/internal/store/postgres/channel_member_join.go index a948248a..5a77bd45 100644 --- a/internal/store/postgres/channel_member_join.go +++ b/internal/store/postgres/channel_member_join.go @@ -267,6 +267,7 @@ func (s *ChannelStore) futureCreatorAfterLeave(ctx context.Context, db sqlcgen.D member, err := scanChannelMember(db.QueryRow(ctx, ` SELECT channel_id, user_id, inviter_user_id, role, status, joined_at, left_at, admin_rights::text, banned_rights::text, rank, available_min_id, available_min_pts, + history_clear_anchor_id, history_clear_anchor_date, read_inbox_max_id, read_outbox_max_id, unread_mark, slowmode_last_send_date FROM channel_members WHERE channel_id = $1 diff --git a/internal/store/postgres/channel_member_list.go b/internal/store/postgres/channel_member_list.go index c2c4e280..ec7a943c 100644 --- a/internal/store/postgres/channel_member_list.go +++ b/internal/store/postgres/channel_member_list.go @@ -128,7 +128,8 @@ func (s *ChannelStore) GetParticipants(ctx context.Context, viewerUserID, channe } rows, err := s.db.Query(ctx, ` SELECT channel_id, user_id, inviter_user_id, role, status, joined_at, left_at, admin_rights::text, banned_rights::text, - rank, available_min_id, available_min_pts, read_inbox_max_id, read_outbox_max_id, unread_mark, slowmode_last_send_date + rank, available_min_id, available_min_pts, history_clear_anchor_id, history_clear_anchor_date, + read_inbox_max_id, read_outbox_max_id, unread_mark, slowmode_last_send_date `+from+` WHERE `+strings.Join(where, " AND ")+` ORDER BY CASE role WHEN 'creator' THEN 0 WHEN 'admin' THEN 1 ELSE 2 END, user_id @@ -221,7 +222,8 @@ func (s *ChannelStore) ListActiveChannelMembers(ctx context.Context, viewerUserI } rows, err := s.db.Query(ctx, ` SELECT channel_id, user_id, inviter_user_id, role, status, joined_at, left_at, admin_rights::text, banned_rights::text, - rank, available_min_id, available_min_pts, read_inbox_max_id, read_outbox_max_id, unread_mark, slowmode_last_send_date + rank, available_min_id, available_min_pts, history_clear_anchor_id, history_clear_anchor_date, + read_inbox_max_id, read_outbox_max_id, unread_mark, slowmode_last_send_date FROM channel_members WHERE channel_id = $1 AND status = 'active' ORDER BY user_id @@ -264,6 +266,7 @@ func (s *ChannelStore) ListActiveChannelBotMembers(ctx context.Context, viewerUs rows, err := s.db.Query(ctx, ` SELECT m.channel_id, m.user_id, m.inviter_user_id, m.role, m.status, m.joined_at, m.left_at, m.admin_rights::text, m.banned_rights::text, m.rank, m.available_min_id, m.available_min_pts, + m.history_clear_anchor_id, m.history_clear_anchor_date, m.read_inbox_max_id, m.read_outbox_max_id, m.unread_mark, m.slowmode_last_send_date, COUNT(*) OVER()::int FROM bots b diff --git a/internal/store/postgres/channel_message_delete.go b/internal/store/postgres/channel_message_delete.go index 1e9b1080..faa6088a 100644 --- a/internal/store/postgres/channel_message_delete.go +++ b/internal/store/postgres/channel_message_delete.go @@ -2,6 +2,7 @@ package postgres import ( "context" + "errors" "fmt" "github.com/jackc/pgx/v5" "sort" @@ -223,19 +224,68 @@ func (s *ChannelStore) DeleteChannelHistory(ctx context.Context, req domain.Dele } if !req.ForEveryone { appliedMinID := maxInt(member.AvailableMinID, maxID) + changed := appliedMinID > member.AvailableMinID + anchorID := member.HistoryClearAnchorID + anchorDate := member.HistoryClearAnchorDate + if changed { + anchorID = appliedMinID + anchorDate = req.Date + var messageDate int + err := tx.QueryRow(ctx, ` +SELECT message_date +FROM channel_messages +WHERE channel_id = $1 AND id = $2`, req.ChannelID, appliedMinID).Scan(&messageDate) + switch { + case err == nil && messageDate > 0: + anchorDate = messageDate + case errors.Is(err, pgx.ErrNoRows): + // max_id may name an already-pruned/hole id. The owner-local + // marker is still a valid monotonic boundary; retain request + // time so its dialog projection remains loadable. + case err != nil: + return domain.DeleteChannelHistoryResult{}, fmt.Errorf("select channel clear anchor date: %w", err) + } + if anchorDate <= 0 { + anchorDate = channel.Date + } + } topID, topDate, err := visibleChannelTopAfter(ctx, tx, req.ChannelID, appliedMinID, channel.Date) if err != nil { return domain.DeleteChannelHistoryResult{}, err } + if topID == 0 && anchorID > 0 && anchorID == appliedMinID { + topID = anchorID + topDate = anchorDate + } if _, err := tx.Exec(ctx, ` UPDATE channel_members SET available_min_id = GREATEST(available_min_id, $3), + history_clear_anchor_id = CASE WHEN available_min_id < $3 THEN $3 ELSE history_clear_anchor_id END, + history_clear_anchor_date = CASE WHEN available_min_id < $3 THEN $4 ELSE history_clear_anchor_date END, read_inbox_max_id = GREATEST(read_inbox_max_id, $3), unread_mark = false, updated_at = now() -WHERE channel_id = $1 AND user_id = $2`, req.ChannelID, req.UserID, appliedMinID); err != nil { +WHERE channel_id = $1 AND user_id = $2`, req.ChannelID, req.UserID, appliedMinID, anchorDate); err != nil { return domain.DeleteChannelHistoryResult{}, fmt.Errorf("update channel local clear member: %w", err) } + if changed { + member.AvailableMinID = appliedMinID + member.HistoryClearAnchorID = anchorID + member.HistoryClearAnchorDate = anchorDate + if err := upsertUserChannelMemberIndexTx(ctx, tx, channel, member); err != nil { + return domain.DeleteChannelHistoryResult{}, err + } + if _, err := tx.Exec(ctx, ` +UPDATE user_channel_member_index +SET available_min_id = $3, + history_clear_anchor_id = $3, + history_clear_updated_at = $4, + updated_at = now() +WHERE user_id = $1 AND channel_id = $2`, + req.UserID, req.ChannelID, appliedMinID, req.Date); err != nil { + return domain.DeleteChannelHistoryResult{}, fmt.Errorf("update channel local clear recovery index: %w", err) + } + } if err := deleteChannelUnreadMentionsUpToTx(ctx, tx, req.UserID, req.ChannelID, appliedMinID); err != nil { return domain.DeleteChannelHistoryResult{}, err } @@ -259,7 +309,17 @@ ON CONFLICT (user_id, channel_id) DO UPDATE SET return domain.DeleteChannelHistoryResult{}, fmt.Errorf("commit local clear channel history: %w", err) } committed = true - return domain.DeleteChannelHistoryResult{Channel: channel, AvailableMinID: appliedMinID}, nil + if s.memberCacheActive(s.db) { + s.memberCache.delete(req.ChannelID, req.UserID) + } + if s.dialogCacheActive(s.db) { + s.dialogCache.delete(req.UserID, req.ChannelID) + } + return domain.DeleteChannelHistoryResult{ + Channel: channel, + AvailableMinID: appliedMinID, + AvailableMinChanged: changed, + }, nil } if !canDeleteAnyChannelMessage(member) { return domain.DeleteChannelHistoryResult{}, domain.ErrChannelAdminRequired diff --git a/internal/store/postgres/channel_message_history.go b/internal/store/postgres/channel_message_history.go index e0325301..62b5e2aa 100644 --- a/internal/store/postgres/channel_message_history.go +++ b/internal/store/postgres/channel_message_history.go @@ -79,6 +79,10 @@ AND EXISTS ( baseArgs = append(baseArgs, filter.MinID) base += fmt.Sprintf(" AND id > $%d", len(baseArgs)) } + historyClearAnchor, hasHistoryClearAnchor, err := s.channelHistoryClearAnchor(ctx, channel, member, filter) + if err != nil { + return domain.ChannelHistory{}, err + } out := domain.ChannelHistory{Channel: channel, Self: member, Channels: extraChannels} needExactTotal := filter.NeedTotalCount || filter.CountOnly exactTotal := 0 @@ -92,6 +96,9 @@ AND EXISTS ( ).Scan(&exactTotal); err != nil { return domain.ChannelHistory{}, fmt.Errorf("count channel history: %w", err) } + if hasHistoryClearAnchor { + exactTotal++ + } out.Count = exactTotal } if filter.CountOnly { @@ -134,6 +141,18 @@ AND EXISTS ( } return "false" } + anchorMatchesForward := func() bool { + if !hasHistoryClearAnchor { + return false + } + if filter.OffsetDate > 0 { + return historyClearAnchor.Date >= filter.OffsetDate + } + if filter.OffsetID > 0 { + return historyClearAnchor.ID > filter.OffsetID + } + return false + } aroundOlderCond := func(args *[]any) string { if filter.OffsetDate > 0 { *args = append(*args, filter.OffsetDate) @@ -145,6 +164,30 @@ AND EXISTS ( } return "true" } + anchorMatchesAroundOlder := func() bool { + if !hasHistoryClearAnchor { + return false + } + if filter.OffsetDate > 0 { + return historyClearAnchor.Date < filter.OffsetDate + } + if filter.OffsetID > 0 { + return historyClearAnchor.ID <= filter.OffsetID + } + return true + } + anchorMatchesBackward := func() bool { + if !hasHistoryClearAnchor { + return false + } + if filter.OffsetDate > 0 { + return historyClearAnchor.Date < filter.OffsetDate + } + if filter.OffsetID > 0 { + return historyClearAnchor.ID < filter.OffsetID + } + return true + } switch { case addOffset < 0 && addOffset+limit > 0: // around：以锚点为中心，向更新取 -add_offset 条 + 向更旧（含锚点）取 limit+add_offset 条 @@ -157,6 +200,12 @@ AND EXISTS ( if err != nil { return domain.ChannelHistory{}, err } + if anchorMatchesForward() { + newer = append([]domain.ChannelMessage{historyClearAnchor}, newer...) + if len(newer) > fwdLimit { + newer = newer[:fwdLimit] + } + } bwdArgs := append([]any{}, baseArgs...) bwdWhere := aroundOlderCond(&bwdArgs) bwdArgs = append(bwdArgs, bwdLimit+1) @@ -164,6 +213,9 @@ AND EXISTS ( if err != nil { return domain.ChannelHistory{}, err } + if anchorMatchesAroundOlder() { + older = append(older, historyClearAnchor) + } if len(older) > bwdLimit { older = older[:bwdLimit] hasMoreOlder = true @@ -181,6 +233,9 @@ AND EXISTS ( if err != nil { return domain.ChannelHistory{}, err } + if anchorMatchesForward() { + newer = append([]domain.ChannelMessage{historyClearAnchor}, newer...) + } if len(newer) > limit { newer = newer[:limit] } @@ -198,18 +253,26 @@ AND EXISTS ( args = append(args, filter.OffsetID) where += fmt.Sprintf(" AND id < $%d", len(args)) } - args = append(args, limit+1) + // Fetch the bounded add_offset window and slice it in memory. This keeps + // the shared-history branch on its ordered (channel_id,id) seek index; + // the owner-local anchor is one separate PK lookup and never adds an OR + // that would force BitmapOr + Sort for large channels. + args = append(args, addOffset+limit+1) limIdx := len(args) - sql := "SELECT " + channelMessageColumns + " FROM channel_messages WHERE " + where + " ORDER BY id DESC" - if addOffset > 0 { - args = append(args, addOffset) - sql += fmt.Sprintf(" OFFSET $%d", len(args)) - } - sql += fmt.Sprintf(" LIMIT $%d", limIdx) + sql := "SELECT " + channelMessageColumns + " FROM channel_messages WHERE " + where + + fmt.Sprintf(" ORDER BY id DESC LIMIT $%d", limIdx) older, err := scanList(sql, args) if err != nil { return domain.ChannelHistory{}, err } + if anchorMatchesBackward() { + older = append(older, historyClearAnchor) + } + if addOffset >= len(older) { + older = nil + } else if addOffset > 0 { + older = older[addOffset:] + } if len(older) > limit { older = older[:limit] hasMoreOlder = true @@ -228,9 +291,63 @@ AND EXISTS ( if err := s.populateChannelMessagesReactions(ctx, s.db, viewerUserID, []domain.Channel{channel}, out.Messages); err != nil { return domain.ChannelHistory{}, err } + if hasHistoryClearAnchor { + for i := range out.Messages { + if out.Messages[i].ID == historyClearAnchor.ID { + out.Messages[i] = domain.ProjectChannelHistoryClearMessage( + out.Messages[i], + channel.ID, + member.HistoryClearAnchorID, + member.HistoryClearAnchorDate, + ) + } + } + } return out, nil } +func (s *ChannelStore) channelHistoryClearAnchor( + ctx context.Context, + channel domain.Channel, + member domain.ChannelMember, + filter domain.ChannelHistoryFilter, +) (domain.ChannelMessage, bool, error) { + if !filter.IncludeHistoryClearAnchor || + member.HistoryClearAnchorID <= 0 || + member.HistoryClearAnchorID != member.AvailableMinID || + filter.PinnedOnly || + filter.MusicOnly || + filter.Query != "" { + return domain.ChannelMessage{}, false, nil + } + source, err := s.getChannelMessage(ctx, s.db, channel.ID, member.HistoryClearAnchorID) + if err != nil && !errors.Is(err, domain.ErrMessageIDInvalid) { + return domain.ChannelMessage{}, false, fmt.Errorf("load channel history-clear anchor: %w", err) + } + anchor := domain.ProjectChannelHistoryClearMessage( + source, + channel.ID, + member.HistoryClearAnchorID, + member.HistoryClearAnchorDate, + ) + if filter.SenderUserID != 0 && anchor.SenderUserID != filter.SenderUserID { + return domain.ChannelMessage{}, false, nil + } + if filter.MinDate > 0 && anchor.Date <= filter.MinDate { + return domain.ChannelMessage{}, false, nil + } + if filter.MaxDate > 0 && anchor.Date >= filter.MaxDate { + return domain.ChannelMessage{}, false, nil + } + if filter.MaxID > 0 && anchor.ID > filter.MaxID { + return domain.ChannelMessage{}, false, nil + } + if filter.MinID > 0 && anchor.ID <= filter.MinID { + return domain.ChannelMessage{}, false, nil + } + return anchor, true, nil +} + func (s *ChannelStore) SearchJoinedMessages(ctx context.Context, viewerUserID int64, req domain.ChannelGlobalSearchRequest) (domain.ChannelHistory, error) { query := strings.TrimSpace(req.Query) if viewerUserID == 0 || (query == "" && !req.MusicOnly) { @@ -390,8 +507,15 @@ func (s *ChannelStore) getChannelMessagesForMember(ctx context.Context, viewerUs // 执行不变。注意:这种"OR 哨兵"只对【非排序锚点】的残余过滤安全;ListChannelHistory // 的方向/anchor 条件若同样哨兵化会让规划器无法用索引顺序做 LIMIT、退化为全表扫+排序 // (实测 0.06ms→23ms),故那里【刻意保留】动态 SQL。 - args := []any{channel.ID, id32, member.AvailableMinID} - where := "channel_id = $1 AND id = ANY($2::int[]) AND NOT deleted AND ($3 <= 0 OR id > $3)" + anchorID := 0 + if member.HistoryClearAnchorID > 0 && member.HistoryClearAnchorID == member.AvailableMinID { + anchorID = member.HistoryClearAnchorID + } + args := []any{channel.ID, id32, member.AvailableMinID, anchorID} + where := `channel_id = $1 +AND id = ANY($2::int[]) +AND (NOT deleted OR ($4 > 0 AND id = $4)) +AND (($3 <= 0 OR id > $3) OR ($4 > 0 AND id = $4))` rows, err := s.db.Query(ctx, ` SELECT `+channelMessageColumns+` FROM channel_messages @@ -419,6 +543,36 @@ ORDER BY id DESC`, args...) if err := s.populateChannelMessagesReactions(ctx, s.db, viewerUserID, []domain.Channel{channel}, out.Messages); err != nil { return domain.ChannelHistory{}, err } + if anchorID > 0 { + anchorFound := false + for i := range out.Messages { + if out.Messages[i].ID != anchorID { + continue + } + out.Messages[i] = domain.ProjectChannelHistoryClearMessage( + out.Messages[i], + channel.ID, + member.HistoryClearAnchorID, + member.HistoryClearAnchorDate, + ) + anchorFound = true + } + if !anchorFound { + for _, id := range ids { + if id != anchorID { + continue + } + out.Messages = append(out.Messages, domain.ProjectChannelHistoryClearMessage( + domain.ChannelMessage{}, + channel.ID, + member.HistoryClearAnchorID, + member.HistoryClearAnchorDate, + )) + out.Count = len(out.Messages) + break + } + } + } return out, nil } diff --git a/internal/store/postgres/channel_message_integration_test.go b/internal/store/postgres/channel_message_integration_test.go index 4112cd14..f9cb88f4 100644 --- a/internal/store/postgres/channel_message_integration_test.go +++ b/internal/store/postgres/channel_message_integration_test.go @@ -189,6 +189,17 @@ WHERE channel_id = $1 AND user_id = $2`, channelID, owner.ID).Scan(&ownerMemberR t.Fatalf("read participants = %+v, want friend read date", readers.Participants) } + var channelPtsBeforeClear, channelEventsBeforeClear int + if err := pool.QueryRow(ctx, ` +SELECT c.pts, ( + SELECT count(*)::int + FROM channel_update_events e + WHERE e.channel_id = c.id +) +FROM channels c +WHERE c.id = $1`, channelID).Scan(&channelPtsBeforeClear, &channelEventsBeforeClear); err != nil { + t.Fatalf("query channel state before local clear: %v", err) + } cleared, err := channels.DeleteChannelHistory(ctx, domain.DeleteChannelHistoryRequest{ UserID: friend.ID, ChannelID: channelID, @@ -201,6 +212,70 @@ WHERE channel_id = $1 AND user_id = $2`, channelID, owner.ID).Scan(&ownerMemberR if cleared.AvailableMinID != sent.Message.ID { t.Fatalf("local clear available_min_id = %d, want %d", cleared.AvailableMinID, sent.Message.ID) } + if !cleared.AvailableMinChanged { + t.Fatal("local clear did not report an advanced owner-local boundary") + } + var channelPtsAfterClear, channelEventsAfterClear int + if err := pool.QueryRow(ctx, ` +SELECT c.pts, ( + SELECT count(*)::int + FROM channel_update_events e + WHERE e.channel_id = c.id +) +FROM channels c +WHERE c.id = $1`, channelID).Scan(&channelPtsAfterClear, &channelEventsAfterClear); err != nil { + t.Fatalf("query channel state after local clear: %v", err) + } + if channelPtsAfterClear != channelPtsBeforeClear || channelEventsAfterClear != channelEventsBeforeClear { + t.Fatalf("local clear changed channel sequence: pts %d->%d events %d->%d", + channelPtsBeforeClear, channelPtsAfterClear, channelEventsBeforeClear, channelEventsAfterClear) + } + var recoveryMinID, recoveryAnchorID, recoveryDate int + if err := pool.QueryRow(ctx, ` +SELECT available_min_id, history_clear_anchor_id, history_clear_updated_at +FROM user_channel_member_index +WHERE user_id = $1 AND channel_id = $2`, friend.ID, channelID).Scan( + &recoveryMinID, &recoveryAnchorID, &recoveryDate, + ); err != nil { + t.Fatalf("query owner-local clear recovery index: %v", err) + } + if recoveryMinID != sent.Message.ID || recoveryAnchorID != sent.Message.ID || recoveryDate != 1700000302 { + t.Fatalf("recovery index = min %d anchor %d date %d, want %d/%d/1700000302", + recoveryMinID, recoveryAnchorID, recoveryDate, sent.Message.ID, sent.Message.ID) + } + dirtyAfterClear, err := channels.ListDirtyActiveChannelsForUser(ctx, friend.ID, 1700000302, 0, 10) + if err != nil { + t.Fatalf("list dirty channels after local clear: %v", err) + } + if len(dirtyAfterClear) != 1 || + dirtyAfterClear[0].ChannelID != channelID || + dirtyAfterClear[0].AvailableMinID != sent.Message.ID || + dirtyAfterClear[0].HistoryClearDate != 1700000302 { + t.Fatalf("dirty channel recovery = %+v, want channel %d boundary %d", dirtyAfterClear, channelID, sent.Message.ID) + } + planTx, err := pool.Begin(ctx) + if err != nil { + t.Fatalf("begin history-clear recovery plan transaction: %v", err) + } + defer func() { _ = planTx.Rollback(ctx) }() + if _, err := planTx.Exec(ctx, "SET LOCAL enable_seqscan = off"); err != nil { + t.Fatalf("disable seqscan for history-clear recovery plan: %v", err) + } + clearPlan := explainText(t, ctx, planTx, ` +SELECT i.channel_id, c.pts, i.available_min_id, i.history_clear_updated_at +FROM user_channel_member_index i +JOIN channels c ON c.id = i.channel_id AND NOT c.deleted +WHERE i.user_id = $1 + AND i.status = 'active' + AND NOT i.deleted + AND i.channel_id > $3 + AND i.history_clear_anchor_id > 0 + AND i.history_clear_anchor_id = i.available_min_id + AND i.history_clear_updated_at >= $2 +ORDER BY i.channel_id ASC +LIMIT $4`, friend.ID, 1700000302, int64(0), 10) + requirePlanContains(t, clearPlan, "user_channel_member_index_history_clear_idx") + _ = planTx.Rollback(ctx) staleClear, err := channels.DeleteChannelHistory(ctx, domain.DeleteChannelHistoryRequest{ UserID: friend.ID, ChannelID: channelID, @@ -213,12 +288,60 @@ WHERE channel_id = $1 AND user_id = $2`, channelID, owner.ID).Scan(&ownerMemberR if staleClear.AvailableMinID != sent.Message.ID { t.Fatalf("stale local clear available_min_id = %d, want monotonic %d", staleClear.AvailableMinID, sent.Message.ID) } + if staleClear.AvailableMinChanged { + t.Fatal("stale local clear unexpectedly replaced the owner-local anchor") + } + if err := pool.QueryRow(ctx, ` +SELECT history_clear_updated_at +FROM user_channel_member_index +WHERE user_id = $1 AND channel_id = $2`, friend.ID, channelID).Scan(&recoveryDate); err != nil { + t.Fatalf("query recovery timestamp after stale clear: %v", err) + } + if recoveryDate != 1700000302 { + t.Fatalf("stale clear recovery date = %d, want unchanged 1700000302", recoveryDate) + } afterClear, err := channels.GetChannel(ctx, friend.ID, channelID) if err != nil { t.Fatalf("get channel after clear: %v", err) } - if afterClear.Dialog.TopMessageID != 0 { - t.Fatalf("dialog after clear = %+v, want no visible top", afterClear.Dialog) + if afterClear.Dialog.TopMessageID != sent.Message.ID || + afterClear.Dialog.HistoryClearAnchorID != sent.Message.ID || + afterClear.Dialog.UnreadCount != 0 { + t.Fatalf("dialog after clear = %+v, want owner-local anchored top %d", afterClear.Dialog, sent.Message.ID) + } + dialogsAfterClear, err := channels.GetChannelDialogs(ctx, friend.ID, []int64{channelID}) + if err != nil { + t.Fatalf("get channel dialogs after clear: %v", err) + } + if len(dialogsAfterClear.Dialogs) != 1 || + len(dialogsAfterClear.Messages) != 1 || + !domain.IsChannelHistoryClearMessage(dialogsAfterClear.Messages[0]) || + dialogsAfterClear.Messages[0].ID != sent.Message.ID || + dialogsAfterClear.Messages[0].Body != "" { + t.Fatalf("dialog projection after clear = dialogs=%+v messages=%+v", dialogsAfterClear.Dialogs, dialogsAfterClear.Messages) + } + projectedHistory, err := channels.ListChannelHistory(ctx, friend.ID, domain.ChannelHistoryFilter{ + ChannelID: channelID, + Limit: 10, + IncludeHistoryClearAnchor: true, + }) + if err != nil { + t.Fatalf("project history-clear anchor: %v", err) + } + if len(projectedHistory.Messages) != 1 || + !domain.IsChannelHistoryClearMessage(projectedHistory.Messages[0]) || + projectedHistory.Messages[0].ID != sent.Message.ID { + t.Fatalf("projected history after clear = %+v, want anchor %d", projectedHistory.Messages, sent.Message.ID) + } + ownerDialogsAfterClear, err := channels.GetChannelDialogs(ctx, owner.ID, []int64{channelID}) + if err != nil { + t.Fatalf("get unaffected owner dialog after friend clear: %v", err) + } + if len(ownerDialogsAfterClear.Messages) != 1 || + ownerDialogsAfterClear.Messages[0].ID != sent.Message.ID || + ownerDialogsAfterClear.Messages[0].Body != "first visible channel text" || + ownerDialogsAfterClear.Messages[0].Action != nil { + t.Fatalf("friend clear changed shared owner projection: %+v", ownerDialogsAfterClear.Messages) } next, err := channels.SendChannelMessage(ctx, domain.SendChannelMessageRequest{ @@ -238,6 +361,59 @@ WHERE channel_id = $1 AND user_id = $2`, channelID, owner.ID).Scan(&ownerMemberR if afterNext.Dialog.TopMessageID != next.Message.ID || afterNext.Dialog.UnreadCount != 1 { t.Fatalf("dialog after next = %+v, want top %d unread 1", afterNext.Dialog, next.Message.ID) } + firstPage, err := channels.ListChannelHistory(ctx, friend.ID, domain.ChannelHistoryFilter{ + ChannelID: channelID, + Limit: 1, + IncludeHistoryClearAnchor: true, + }) + if err != nil { + t.Fatalf("list first page after next message: %v", err) + } + if len(firstPage.Messages) != 1 || + firstPage.Messages[0].ID != next.Message.ID || + firstPage.Count != 2 { + t.Fatalf("first page after next = %+v count=%d, want next message and bounded has-more count", firstPage.Messages, firstPage.Count) + } + offsetPage, err := channels.ListChannelHistory(ctx, friend.ID, domain.ChannelHistoryFilter{ + ChannelID: channelID, + AddOffset: 1, + Limit: 1, + IncludeHistoryClearAnchor: true, + }) + if err != nil { + t.Fatalf("list add-offset page after next message: %v", err) + } + if len(offsetPage.Messages) != 1 || + offsetPage.Messages[0].ID != sent.Message.ID || + !domain.IsChannelHistoryClearMessage(offsetPage.Messages[0]) { + t.Fatalf("add-offset page after next = %+v, want retained clear anchor %d", offsetPage.Messages, sent.Message.ID) + } + exactCount, err := channels.ListChannelHistory(ctx, friend.ID, domain.ChannelHistoryFilter{ + ChannelID: channelID, + CountOnly: true, + NeedTotalCount: true, + IncludeHistoryClearAnchor: true, + }) + if err != nil { + t.Fatalf("count history after next message: %v", err) + } + if exactCount.Count != 2 { + t.Fatalf("exact history count after next = %d, want shared message plus clear anchor", exactCount.Count) + } + olderPage, err := channels.ListChannelHistory(ctx, friend.ID, domain.ChannelHistoryFilter{ + ChannelID: channelID, + OffsetID: next.Message.ID, + Limit: 10, + IncludeHistoryClearAnchor: true, + }) + if err != nil { + t.Fatalf("list older page after next message: %v", err) + } + if len(olderPage.Messages) != 1 || + olderPage.Messages[0].ID != sent.Message.ID || + !domain.IsChannelHistoryClearMessage(olderPage.Messages[0]) { + t.Fatalf("older page after next = %+v, want retained clear anchor %d", olderPage.Messages, sent.Message.ID) + } } func TestChannelStoreStoryMessageForwardsPublicOnlyAndDeleteRollbackPostgres(t *testing.T) { diff --git a/internal/store/postgres/channel_read_integration_test.go b/internal/store/postgres/channel_read_integration_test.go index ea49ac96..a006de7b 100644 --- a/internal/store/postgres/channel_read_integration_test.go +++ b/internal/store/postgres/channel_read_integration_test.go @@ -574,8 +574,10 @@ WHERE channel_id = $1 AND user_id = $2`, channelID, member.ID).Scan(&storedTop, if err != nil { t.Fatalf("get large channel after local clear: %v", err) } - if afterClear.Dialog.TopMessageID != 0 || afterClear.Dialog.UnreadCount != 0 { - t.Fatalf("large dialog after local clear = %+v, want no visible unread top", afterClear.Dialog) + if afterClear.Dialog.TopMessageID != sent.Message.ID || + afterClear.Dialog.HistoryClearAnchorID != sent.Message.ID || + afterClear.Dialog.UnreadCount != 0 { + t.Fatalf("large dialog after local clear = %+v, want anchored top %d with no unread", afterClear.Dialog, sent.Message.ID) } } 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 645c8685..bed2fb9c 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 != 158 { - t.Fatalf("migration status = %+v, want clean version 158", status) + if status.Dirty || status.Empty || status.Version != 162 { + t.Fatalf("migration status = %+v, want clean version 162", status) } }