diff --git a/deploy/migrations/0145_privacy_update_events.down.sql b/deploy/migrations/0145_privacy_update_events.down.sql index 977222ba..9a542784 100644 --- a/deploy/migrations/0145_privacy_update_events.down.sql +++ b/deploy/migrations/0145_privacy_update_events.down.sql @@ -1,17 +1 @@ -ALTER TABLE public.user_update_events DROP CONSTRAINT IF EXISTS user_update_events_type_check; -ALTER TABLE public.user_update_events ADD CONSTRAINT user_update_events_type_check CHECK ( - (event_type)::text = ANY (ARRAY[ - 'new_message', 'read_history_inbox', 'read_history_outbox', 'read_message_contents', - 'edit_message', 'web_page', 'message_reactions', 'message_poll', 'draft_message', 'quick_replies', - 'new_quick_reply', 'delete_quick_reply', 'quick_reply_message', 'delete_quick_reply_messages', - 'contacts_reset', 'dialog_pinned', 'pinned_dialogs', 'pinned_messages', 'dialog_unread_mark', - 'peer_settings', 'peer_story_blocked', 'user_phone', 'user_emoji_status', 'delete_messages', - 'dialog_filter', 'dialog_filter_order', 'dialog_filters', 'folder_peers', - 'channel_available_messages', 'channel_view_forum_as_messages', 'channel_state', - 'saved_dialog_pinned', 'pinned_saved_dialogs', 'story', 'read_stories', - 'sent_story_reaction', 'new_story_reaction', 'noop', - 'read_channel_discussion_inbox', 'read_channel_discussion_outbox' - ]::text[]) -); - -DROP TABLE IF EXISTS public.user_update_privacy_payloads; +-- Reserved development migration version; up is intentionally a no-op. diff --git a/deploy/migrations/0145_privacy_update_events.up.sql b/deploy/migrations/0145_privacy_update_events.up.sql index ee3c6f57..0acc1dd8 100644 --- a/deploy/migrations/0145_privacy_update_events.up.sql +++ b/deploy/migrations/0145_privacy_update_events.up.sql @@ -1,30 +1,5 @@ --- Privacy settings are absolute account state. Persist the immutable rule --- snapshot next to the account pts event so other online sessions and offline --- getDifference replay exactly the committed value without re-querying the --- mutable account_privacy_rules row. -CREATE TABLE public.user_update_privacy_payloads ( - user_id bigint NOT NULL, - pts integer NOT NULL CHECK (pts > 0), - payload jsonb NOT NULL CHECK (jsonb_typeof(payload) = 'object'), - PRIMARY KEY (user_id, pts), - CONSTRAINT user_update_privacy_payloads_event_fk - FOREIGN KEY (user_id, pts) - REFERENCES public.user_update_events(user_id, pts) - ON DELETE CASCADE -); - -ALTER TABLE public.user_update_events DROP CONSTRAINT IF EXISTS user_update_events_type_check; -ALTER TABLE public.user_update_events ADD CONSTRAINT user_update_events_type_check CHECK ( - (event_type)::text = ANY (ARRAY[ - 'new_message', 'read_history_inbox', 'read_history_outbox', 'read_message_contents', - 'edit_message', 'web_page', 'message_reactions', 'message_poll', 'draft_message', 'quick_replies', - 'new_quick_reply', 'delete_quick_reply', 'quick_reply_message', 'delete_quick_reply_messages', - 'contacts_reset', 'dialog_pinned', 'pinned_dialogs', 'pinned_messages', 'dialog_unread_mark', - 'peer_settings', 'peer_story_blocked', 'user_phone', 'user_emoji_status', 'privacy', 'delete_messages', - 'dialog_filter', 'dialog_filter_order', 'dialog_filters', 'folder_peers', - 'channel_available_messages', 'channel_view_forum_as_messages', 'channel_state', - 'saved_dialog_pinned', 'pinned_saved_dialogs', 'story', 'read_stories', - 'sent_story_reaction', 'new_story_reaction', 'noop', - 'read_channel_discussion_inbox', 'read_channel_discussion_outbox' - ]::text[]) -); +-- Reserved development migration version. +-- account privacy is authoritative absolute state: updatePrivacy has no +-- pts/pts_count, so this migration must not add a privacy event type or payload +-- table. The earlier development-only definition was corrected in place +-- because no user/production database can contain that unpublished shape. diff --git a/deploy/migrations/0147_user_moderation_profile_events.down.sql b/deploy/migrations/0147_user_moderation_profile_events.down.sql index 53f0a3dd..dfe77ed5 100644 --- a/deploy/migrations/0147_user_moderation_profile_events.down.sql +++ b/deploy/migrations/0147_user_moderation_profile_events.down.sql @@ -1,14 +1,5 @@ --- Preserve account pts continuity if this development migration is reversed: --- a profile refresh is absolute state, so replacing it with noop is safer than --- deleting business history and leaving a permanent difference gap. -UPDATE public.user_update_events -SET event_type = 'noop', peer_type = NULL, peer_id = NULL -WHERE event_type = 'user_profile'; - -UPDATE public.dispatch_outbox -SET event_type = 'noop' -WHERE event_type = 'user_profile'; - +-- The corrected up migration does not add schema, payloads or event types. +-- Keep the constraint normalized to the same non-moderation event set. ALTER TABLE public.user_update_events DROP CONSTRAINT IF EXISTS user_update_events_type_check; ALTER TABLE public.user_update_events ADD CONSTRAINT user_update_events_type_check CHECK ( (event_type)::text = ANY (ARRAY[ @@ -16,7 +7,7 @@ ALTER TABLE public.user_update_events ADD CONSTRAINT user_update_events_type_che 'edit_message', 'web_page', 'message_reactions', 'message_poll', 'draft_message', 'quick_replies', 'new_quick_reply', 'delete_quick_reply', 'quick_reply_message', 'delete_quick_reply_messages', 'contacts_reset', 'dialog_pinned', 'pinned_dialogs', 'pinned_messages', 'dialog_unread_mark', - 'peer_settings', 'peer_story_blocked', 'user_phone', 'user_emoji_status', 'privacy', 'delete_messages', + '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', 'saved_dialog_pinned', 'pinned_saved_dialogs', 'story', 'read_stories', diff --git a/deploy/migrations/0147_user_moderation_profile_events.up.sql b/deploy/migrations/0147_user_moderation_profile_events.up.sql index b8527dd2..a04a5d98 100644 --- a/deploy/migrations/0147_user_moderation_profile_events.up.sql +++ b/deploy/migrations/0147_user_moderation_profile_events.up.sql @@ -1,7 +1,8 @@ --- SCAM/FAKE is viewer-visible absolute profile state. Each affected direct --- contact/dialog viewer receives a durable account-pts refresh so online --- outbox delivery and offline getDifference converge even when TDesktop keeps --- its local contact/dialog cache across restarts. +-- Reserved development migration version. +-- SCAM/FAKE is authoritative peer profile state. updateUser/updateChannel have +-- no pts/pts_count, so this migration must not admit a moderation-only account +-- PTS event type. The earlier development definition was corrected in place +-- because no user/production database can contain that unpublished shape. ALTER TABLE public.user_update_events DROP CONSTRAINT IF EXISTS user_update_events_type_check; ALTER TABLE public.user_update_events ADD CONSTRAINT user_update_events_type_check CHECK ( (event_type)::text = ANY (ARRAY[ @@ -9,7 +10,7 @@ ALTER TABLE public.user_update_events ADD CONSTRAINT user_update_events_type_che 'edit_message', 'web_page', 'message_reactions', 'message_poll', 'draft_message', 'quick_replies', 'new_quick_reply', 'delete_quick_reply', 'quick_reply_message', 'delete_quick_reply_messages', 'contacts_reset', 'dialog_pinned', 'pinned_dialogs', 'pinned_messages', 'dialog_unread_mark', - 'peer_settings', 'peer_story_blocked', 'user_phone', 'user_emoji_status', 'user_profile', 'privacy', + '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', 'saved_dialog_pinned', 'pinned_saved_dialogs', 'story', 'read_stories', diff --git a/internal/app/privacy/cache.go b/internal/app/privacy/cache.go index b2ac8bc1..28d67180 100644 --- a/internal/app/privacy/cache.go +++ b/internal/app/privacy/cache.go @@ -87,34 +87,6 @@ func (c *CachedPrivacyStore) SetPrivacyRules(ctx context.Context, rules domain.P return nil } -func (c *CachedPrivacyStore) SetPrivacyRulesWithUpdate( - ctx context.Context, - rules domain.PrivacyRules, - event domain.UpdateEvent, - excludeAuthKeyID [8]byte, - excludeSessionID int64, -) (domain.UpdateEvent, error) { - writer, ok := c.inner.(store.PrivacyUpdateStore) - if !ok { - return domain.UpdateEvent{}, domain.ErrPrivacyRuleInvalid - } - recorded, err := writer.SetPrivacyRulesWithUpdate(ctx, rules, event, excludeAuthKeyID, excludeSessionID) - if err != nil { - return domain.UpdateEvent{}, err - } - c.InvalidateOwners(rules.OwnerUserID) - _ = c.WarmOwners(ctx, rules.OwnerUserID) - return recorded, nil -} - -func (c *CachedPrivacyStore) SupportsDurablePrivacyUpdates() bool { - if c == nil { - return false - } - capability, ok := c.inner.(interface{ SupportsDurablePrivacyUpdates() bool }) - return ok && capability.SupportsDurablePrivacyUpdates() -} - // WarmOwners 在低频写/变更通知路径一次性装入 owner 的完整规则集。调用方必须先 // InvalidateOwners；epoch 保证预热期间若又发生失效，不会把旧快照写回。 func (c *CachedPrivacyStore) WarmOwners(ctx context.Context, ownerUserIDs ...int64) error { diff --git a/internal/app/privacy/service.go b/internal/app/privacy/service.go index 5ac0ee87..7cd914f8 100644 --- a/internal/app/privacy/service.go +++ b/internal/app/privacy/service.go @@ -83,39 +83,6 @@ func (s *Service) SetRules(ctx context.Context, ownerUserID int64, key domain.Pr return out, nil } -// SetRulesWithUpdate uses the production atomic write boundary when available. -// durable=false means no write was attempted; the RPC layer may then use the -// ordinary SetRules + Updates.RecordPrivacy fallback used by memory tests. -func (s *Service) SetRulesWithUpdate( - ctx context.Context, - ownerUserID int64, - key domain.PrivacyKey, - rules []domain.PrivacyRule, - date int, - excludeAuthKeyID [8]byte, - excludeSessionID int64, -) (domain.PrivacyRules, domain.UpdateEvent, bool, error) { - out, err := normalizedRules(ownerUserID, key, rules) - if err != nil { - return domain.PrivacyRules{}, domain.UpdateEvent{}, false, err - } - capability, ok := s.rules.(interface{ SupportsDurablePrivacyUpdates() bool }) - if !ok || !capability.SupportsDurablePrivacyUpdates() { - return domain.PrivacyRules{}, domain.UpdateEvent{}, false, nil - } - writer := s.rules.(store.PrivacyUpdateStore) - event, err := writer.SetPrivacyRulesWithUpdate(ctx, out, domain.UpdateEvent{ - Type: domain.UpdateEventPrivacy, - Date: date, - Privacy: cloneRules(out), - PtsCount: 1, - }, excludeAuthKeyID, excludeSessionID) - if err != nil { - return domain.PrivacyRules{}, domain.UpdateEvent{}, false, err - } - return out, event, true, nil -} - func normalizedRules(ownerUserID int64, key domain.PrivacyKey, rules []domain.PrivacyRule) (domain.PrivacyRules, error) { if !ValidKey(key) { return domain.PrivacyRules{}, domain.ErrPrivacyKeyInvalid diff --git a/internal/app/updates/service.go b/internal/app/updates/service.go index 812e4dfe..bbae8f17 100644 --- a/internal/app/updates/service.go +++ b/internal/app/updates/service.go @@ -607,19 +607,6 @@ func (s *Service) RecordUserEmojiStatus(ctx context.Context, stateAuthKeyID [8]b }, true, excludeSessionID) } -// RecordPrivacy durably synchronizes the exact immutable privacy snapshot to -// the account's other sessions and offline difference stream. -func (s *Service) RecordPrivacy(ctx context.Context, stateAuthKeyID [8]byte, userID int64, rules domain.PrivacyRules, excludeAuthKeyID [8]byte, excludeSessionID int64) (domain.UpdateEvent, domain.UpdateState, error) { - if rules.OwnerUserID != userID || rules.Key == "" || len(rules.Rules) == 0 { - return domain.UpdateEvent{}, domain.UpdateState{}, domain.ErrPrivacyRuleInvalid - } - return s.recordEvent(ctx, stateAuthKeyID, excludeAuthKeyID, userID, domain.UpdateEvent{ - Type: domain.UpdateEventPrivacy, - Privacy: rules, - PtsCount: 1, - }, true, excludeSessionID) -} - // RecordDraftMessage 记录某会话云草稿变化（保存/清空都是同一事件——草稿是绝对 // 状态，重放时按 peer 重载当前值）。updateDraftMessage 无 pts 字段，走 LacksWirePts // aux 簿记；topMsgID 是 forum 话题草稿键（复用 MaxID 列持久化）。 diff --git a/internal/domain/update_event.go b/internal/domain/update_event.go index 65361dec..07ac5099 100644 --- a/internal/domain/update_event.go +++ b/internal/domain/update_event.go @@ -40,10 +40,7 @@ const ( // user's absolute profile flags (currently moderation SCAM/FAKE state). // The changed user is carried in Peer; hydration attaches the authoritative // viewer projection to Users before online dispatch or getDifference. - UpdateEventUserProfile UpdateEventType = "user_profile" - // UpdateEventPrivacy carries the immutable account privacy key/rule - // snapshot committed at this pts. updatePrivacy has no wire pts. - UpdateEventPrivacy UpdateEventType = "privacy" + UpdateEventUserProfile UpdateEventType = "user_profile" UpdateEventDeleteMessages UpdateEventType = "delete_messages" // UpdateEventPinnedMessages 映射 updatePinnedMessages（私聊置顶/取消 // 置顶；MessageIDs 是该 owner 自己视角的 box id，Bool 为 pinned）。 @@ -96,7 +93,6 @@ type UpdateEvent struct { Bool bool Phone string EmojiStatus UserEmojiStatus - Privacy PrivacyRules Settings PeerSettings MessageIDs []int MaxID int @@ -150,7 +146,6 @@ func (e UpdateEvent) LacksWirePts() bool { UpdateEventUserPhone, UpdateEventUserEmojiStatus, UpdateEventUserProfile, - UpdateEventPrivacy, UpdateEventDialogFilter, UpdateEventDialogFilterOrder, UpdateEventDialogFilters, diff --git a/internal/rpc/account.go b/internal/rpc/account.go index ac908c8e..53bb5aca 100644 --- a/internal/rpc/account.go +++ b/internal/rpc/account.go @@ -873,27 +873,7 @@ func (r *Router) onAccountSetPrivacy(ctx context.Context, req *tg.AccountSetPriv if r.deps.Privacy == nil { return &tg.AccountPrivacyRules{Rules: tgPrivacyRules(rules), Users: []tg.UserClass{}, Chats: []tg.ChatClass{}}, nil } - authKeyID, _ := AuthKeyIDFrom(ctx) - sessionID, _ := SessionIDFrom(ctx) - var ( - saved domain.PrivacyRules - event domain.UpdateEvent - durableWrite bool - ) - if durable, ok := r.deps.Privacy.(PrivacyDurableService); ok { - saved, event, durableWrite, err = durable.SetRulesWithUpdate( - ctx, - userID, - domainKey, - rules, - int(r.clock.Now().Unix()), - rawAuthKeyIDForOrigin(ctx), - sessionID, - ) - } - if err == nil && !durableWrite { - saved, err = r.deps.Privacy.SetRules(ctx, userID, domainKey, rules) - } + saved, err := r.deps.Privacy.SetRules(ctx, userID, domainKey, rules) if err != nil { return nil, privacyErr(err) } @@ -902,33 +882,20 @@ func (r *Router) onAccountSetPrivacy(ctx context.Context, req *tg.AccountSetPriv return nil, err } r.invalidateRPCProjectionForUser(userID) - if durableWrite { - if sessionID != 0 { - r.bookkeepAuxPtsForCurrentSession(ctx, event) - } - r.pushUserUpdatesIfNoReliableDispatch(ctx, userID, tgUpdateForOutboxEvent(event)) - } else if updates, ok := r.deps.Updates.(PrivacyUpdatesService); ok { - event, _, recordErr := updates.RecordPrivacy( - ctx, authKeyID, userID, saved, rawAuthKeyIDForOrigin(ctx), sessionID, - ) - if recordErr != nil { - return nil, internalErr() - } - if sessionID != 0 { - r.bookkeepAuxPtsForCurrentSession(ctx, event) - } - r.pushUserUpdatesIfNoReliableDispatch(ctx, userID, tgUpdateForOutboxEvent(event)) - } else { - r.pushUserUpdates(ctx, userID, &tg.Updates{ - Updates: []tg.UpdateClass{&tg.UpdatePrivacy{ - Key: tgPrivacyKey(saved.Key), - Rules: tgPrivacyRules(saved.Rules), - }}, - Users: []tg.UserClass{}, - Chats: []tg.ChatClass{}, - Date: int(r.clock.Now().Unix()), - }) - } + // updatePrivacy is an absolute, non-PTS account-state notification. The + // originating session applies account.setPrivacy's response; other online + // sessions receive this best-effort update, while offline sessions reload + // the authoritative rules through account.getPrivacy. + r.pushUserUpdates(ctx, userID, &tg.Updates{ + Updates: []tg.UpdateClass{&tg.UpdatePrivacy{ + Key: tgPrivacyKey(saved.Key), + Rules: tgPrivacyRules(saved.Rules), + }}, + Users: []tg.UserClass{}, + Chats: []tg.ChatClass{}, + Date: int(r.clock.Now().Unix()), + Seq: 0, + }) if domainKey == domain.PrivacyKeyStatusTimestamp { r.pushStatusPrivacyRefresh(ctx, userID) } diff --git a/internal/rpc/account_privacy_rpc_test.go b/internal/rpc/account_privacy_rpc_test.go index a97e3eed..9a66f11b 100644 --- a/internal/rpc/account_privacy_rpc_test.go +++ b/internal/rpc/account_privacy_rpc_test.go @@ -14,7 +14,7 @@ import ( "telesrv/internal/store/memory" ) -func TestAccountPrivacyAllKeysRoundTripAndRecordDifferenceEvents(t *testing.T) { +func TestAccountPrivacyAllKeysRoundTripWithoutAdvancingPts(t *testing.T) { ctx := context.Background() const userID int64 = 8101 authKeyID := [8]byte{8, 1} @@ -22,10 +22,11 @@ func TestAccountPrivacyAllKeysRoundTripAndRecordDifferenceEvents(t *testing.T) { privacy := appprivacy.NewService(memory.NewPrivacyStore(), memory.NewContactStore()) events := memory.NewUpdateEventStore() updates := appupdates.NewService(memory.NewUpdateStateStore(), events) + sessions := &captureSessions{} router := New(Config{}, Deps{ Privacy: privacy, Updates: updates, - Sessions: &captureSessions{}, + Sessions: sessions, }, zaptest.NewLogger(t), clock.System) requestCtx := WithSessionID(WithAuthKeyID(WithUserID(ctx, userID), authKeyID), sessionID) @@ -83,40 +84,78 @@ func TestAccountPrivacyAllKeysRoundTripAndRecordDifferenceEvents(t *testing.T) { if _, ok := get.Rules[0].(*tg.PrivacyValueDisallowAll); !ok { t.Fatalf("getPrivacy rule=%T, want disallowAll", get.Rules[0]) } + pushed, ok := sessions.lastUserPush().(*tg.Updates) + if !ok || len(pushed.Updates) != 1 { + t.Fatalf("online push=%T/%+v, want one updatePrivacy", sessions.lastUserPush(), pushed) + } + privacyUpdate, ok := pushed.Updates[0].(*tg.UpdatePrivacy) + if !ok { + t.Fatalf("online push update=%T, want updatePrivacy(%q)", pushed.Updates[0], test.domain) + } + if !test.wire(privacyUpdate.Key) { + t.Fatalf("online push key=%T, want %q", privacyUpdate.Key, test.domain) + } }) } + if pushedUserIDs := sessions.pushedUserIDs(); len(pushedUserIDs) != len(keys) { + t.Fatalf("online privacy pushes=%v, want exactly one per key", pushedUserIDs) + } else { + for i, pushedUserID := range pushedUserIDs { + if pushedUserID != userID { + t.Fatalf("online privacy push[%d] target=%d, want owner %d", i, pushedUserID, userID) + } + } + } + if snapshot := sessions.snapshot(); snapshot.sessionID != sessionID || snapshot.userID != userID { + t.Fatalf("online push exclusion/target=%+v, want current session %d excluded for user %d", snapshot, sessionID, userID) + } recorded, err := events.ListAfter(ctx, userID, 0, 100) if err != nil { - t.Fatalf("list privacy events: %v", err) + t.Fatalf("list account update events: %v", err) } - if len(recorded) != len(keys) { - t.Fatalf("privacy events=%d, want %d", len(recorded), len(keys)) + if len(recorded) != 0 { + t.Fatalf("account update events=%+v, want none for privacy changes", recorded) } - for i, event := range recorded { - if event.Type != domain.UpdateEventPrivacy || - event.Privacy.OwnerUserID != userID || - event.Privacy.Key != keys[i].domain || - event.PtsCount != 1 { - t.Fatalf("event[%d]=%+v, want durable privacy snapshot for %q", i, event, keys[i].domain) - } + state, err := updates.CurrentState(ctx, userID) + if err != nil { + t.Fatalf("current update state: %v", err) + } + if state.Pts != 0 { + t.Fatalf("privacy changes advanced pts to %d, want 0", state.Pts) } difference, err := updates.GetDifference(ctx, [8]byte{8, 2}, userID, domain.UpdateState{}) if err != nil { t.Fatalf("getDifference: %v", err) } - wireDifference, ok := tgUpdatesDifference(userID, difference).(*tg.UpdatesDifference) - if !ok || len(wireDifference.OtherUpdates) != len(keys) { - t.Fatalf("wire difference=%T updates=%d, want %d privacy updates", wireDifference, len(wireDifference.OtherUpdates), len(keys)) + if difference.State.Pts != 0 || len(difference.Events) != 0 { + t.Fatalf("difference after privacy changes=%+v, want empty pts=0", difference) } - for i, update := range wireDifference.OtherUpdates { - privacyUpdate, ok := update.(*tg.UpdatePrivacy) - if !ok { - t.Fatalf("difference update[%d]=%T, want updatePrivacy", i, update) - } - if !keys[i].wire(privacyUpdate.Key) { - t.Fatalf("difference update[%d] key=%T, want %q", i, privacyUpdate.Key, keys[i].domain) - } + + // A real message-box update immediately after privacy changes must still + // receive pts=1. This catches both hidden privacy allocations and gaps left + // behind by synthetic bookkeeping events. + message := domain.Message{ + ID: 1, + OwnerUserID: userID, + Peer: domain.Peer{Type: domain.PeerTypeUser, ID: 8102}, + From: domain.Peer{Type: domain.PeerTypeUser, ID: 8102}, + Date: 1700000000, + Body: "after privacy", + } + event, state, err := updates.RecordNewMessage(ctx, authKeyID, userID, message) + if err != nil { + t.Fatalf("record adjacent message update: %v", err) + } + if event.Pts != 1 || event.PtsCount != 1 || state.Pts != 1 { + t.Fatalf("adjacent message event/state=%+v/%+v, want first pts=1", event, state) + } + difference, err = updates.GetDifference(ctx, [8]byte{8, 2}, userID, domain.UpdateState{}) + if err != nil { + t.Fatalf("getDifference after message: %v", err) + } + if difference.State.Pts != 1 || len(difference.Events) != 1 || difference.Events[0].Type != domain.UpdateEventNewMessage { + t.Fatalf("difference after adjacent message=%+v, want one contiguous new_message at pts=1", difference) } } diff --git a/internal/rpc/convert_updates.go b/internal/rpc/convert_updates.go index 6b36eac1..6fef3f78 100644 --- a/internal/rpc/convert_updates.go +++ b/internal/rpc/convert_updates.go @@ -247,14 +247,6 @@ func tgOtherUpdateFromEvent(event domain.UpdateEvent) tg.UpdateClass { return nil } return &tg.UpdateUser{UserID: event.Peer.ID} - case domain.UpdateEventPrivacy: - if event.Privacy.OwnerUserID == 0 || event.Privacy.Key == "" || len(event.Privacy.Rules) == 0 { - return nil - } - return &tg.UpdatePrivacy{ - Key: tgPrivacyKey(event.Privacy.Key), - Rules: tgPrivacyRules(event.Privacy.Rules), - } case domain.UpdateEventChannelState: if event.Peer.Type != domain.PeerTypeChannel || event.Peer.ID == 0 { return nil diff --git a/internal/rpc/deps.go b/internal/rpc/deps.go index 73f754bc..888717a7 100644 --- a/internal/rpc/deps.go +++ b/internal/rpc/deps.go @@ -487,26 +487,6 @@ type UserEmojiStatusUpdatesService interface { RecordUserEmojiStatus(ctx context.Context, stateAuthKeyID [8]byte, userID int64, status domain.UserEmojiStatus, excludeAuthKeyID [8]byte, excludeSessionID int64) (domain.UpdateEvent, domain.UpdateState, error) } -// PrivacyUpdatesService is the fallback durable extension for stores that do -// not support the atomic privacy+event write boundary (mainly memory tests). -type PrivacyUpdatesService interface { - RecordPrivacy(ctx context.Context, stateAuthKeyID [8]byte, userID int64, rules domain.PrivacyRules, excludeAuthKeyID [8]byte, excludeSessionID int64) (domain.UpdateEvent, domain.UpdateState, error) -} - -// PrivacyDurableService is implemented by the production privacy service. Its -// successful path commits rules+pts+event+dispatch in one transaction. -type PrivacyDurableService interface { - SetRulesWithUpdate( - ctx context.Context, - ownerUserID int64, - key domain.PrivacyKey, - rules []domain.PrivacyRule, - date int, - excludeAuthKeyID [8]byte, - excludeSessionID int64, - ) (saved domain.PrivacyRules, event domain.UpdateEvent, durable bool, err error) -} - // ContactsService 抽象通讯录查询。 type ContactsService interface { GetContacts(ctx context.Context, userID int64, hash int64) (domain.ContactList, bool, error) diff --git a/internal/store/memory/updates.go b/internal/store/memory/updates.go index db86d26d..7124744a 100644 --- a/internal/store/memory/updates.go +++ b/internal/store/memory/updates.go @@ -105,20 +105,9 @@ func cloneUpdateEvent(event domain.UpdateEvent) domain.UpdateEvent { event.Reaction = cloneUpdateReaction(event.Reaction) event.QuickReplies = cloneUpdateQuickReplies(event.QuickReplies) event.QuickReplyMessage = cloneUpdateQuickReplyMessage(event.QuickReplyMessage) - event.Privacy.Rules = clonePrivacyRuleSlice(event.Privacy.Rules) return event } -func clonePrivacyRuleSlice(in []domain.PrivacyRule) []domain.PrivacyRule { - out := make([]domain.PrivacyRule, len(in)) - for i := range in { - out[i] = in[i] - out[i].UserIDs = append([]int64(nil), in[i].UserIDs...) - out[i].ChatIDs = append([]int64(nil), in[i].ChatIDs...) - } - return out -} - func cloneUpdateStory(story domain.Story) domain.Story { story.Entities = append([]domain.MessageEntity(nil), story.Entities...) story.Views.Reactions = append([]domain.ChannelMessageReactionCount(nil), story.Views.Reactions...) diff --git a/internal/store/postgres/privacy.go b/internal/store/postgres/privacy.go index 8cb7dcbd..2078b524 100644 --- a/internal/store/postgres/privacy.go +++ b/internal/store/postgres/privacy.go @@ -24,16 +24,6 @@ func NewPrivacyStore(db sqlcgen.DBTX) *PrivacyStore { return &PrivacyStore{db: db} } -func (s *PrivacyStore) SupportsDurablePrivacyUpdates() bool { - if s == nil { - return false - } - _, ok := s.db.(interface { - Begin(context.Context) (pgx.Tx, error) - }) - return ok -} - func (s *PrivacyStore) GetPrivacyRules(ctx context.Context, ownerUserID int64, key domain.PrivacyKey) (domain.PrivacyRules, bool, error) { row := s.db.QueryRow(ctx, ` SELECT rules::text @@ -77,56 +67,6 @@ ON CONFLICT (owner_user_id, privacy_key) DO UPDATE SET return nil } -// SetPrivacyRulesWithUpdate commits the mutable rule row and the immutable -// account update snapshot in one transaction. A privacy rule can therefore -// never become visible without a matching pts event/outbox item. -func (s *PrivacyStore) SetPrivacyRulesWithUpdate( - ctx context.Context, - rules domain.PrivacyRules, - event domain.UpdateEvent, - excludeAuthKeyID [8]byte, - excludeSessionID int64, -) (domain.UpdateEvent, error) { - beginner, ok := s.db.(interface { - Begin(context.Context) (pgx.Tx, error) - }) - if !ok { - return domain.UpdateEvent{}, fmt.Errorf("privacy update transaction unavailable") - } - tx, err := beginner.Begin(ctx) - if err != nil { - return domain.UpdateEvent{}, fmt.Errorf("begin privacy update: %w", err) - } - committed := false - defer func() { - if !committed { - _ = tx.Rollback(ctx) - } - }() - if err := setPrivacyRules(ctx, tx, rules); err != nil { - return domain.UpdateEvent{}, err - } - if event.Date == 0 { - return domain.UpdateEvent{}, fmt.Errorf("privacy update date is required") - } - event.Type = domain.UpdateEventPrivacy - event.Privacy = rules - event.PtsCount = 1 - qtx := sqlcgen.New(tx) - recorded, err := NewUpdateEventStore(tx).appendInTx( - ctx, tx, qtx, rules.OwnerUserID, event, true, - excludeAuthKeyID, excludeSessionID, true, - ) - if err != nil { - return domain.UpdateEvent{}, fmt.Errorf("append privacy update: %w", err) - } - if err := tx.Commit(ctx); err != nil { - return domain.UpdateEvent{}, fmt.Errorf("commit privacy update: %w", err) - } - committed = true - return recorded, nil -} - func (s *PrivacyStore) ListPrivacyRules(ctx context.Context, ownerUserIDs []int64, keys []domain.PrivacyKey) ([]domain.PrivacyRules, error) { if len(ownerUserIDs) == 0 || len(keys) == 0 { return nil, nil diff --git a/internal/store/postgres/privacy_integration_test.go b/internal/store/postgres/privacy_integration_test.go new file mode 100644 index 00000000..967f1270 --- /dev/null +++ b/internal/store/postgres/privacy_integration_test.go @@ -0,0 +1,103 @@ +package postgres + +import ( + "context" + "errors" + "testing" + + "github.com/jackc/pgerrcode" + "github.com/jackc/pgx/v5/pgconn" + + "telesrv/internal/domain" +) + +// TestPrivacyRulesDoNotAllocateAccountPts protects the protocol boundary: +// account privacy is authoritative absolute state, not a message-box event. +func TestPrivacyRulesDoNotAllocateAccountPts(t *testing.T) { + pool := testPool(t) + ctx := context.Background() + users := NewUserStore(pool) + suffix := randomSuffix(t) + user, err := users.Create(ctx, domain.User{ + AccessHash: 9201, + Phone: "+1665" + suffix + "01", + FirstName: "PrivacyPts", + }) + if err != nil { + t.Fatalf("create user: %v", err) + } + t.Cleanup(func() { + _, _ = pool.Exec(ctx, "DELETE FROM users WHERE id = $1", user.ID) + }) + var privacyPayloadTableAbsent bool + if err := pool.QueryRow(ctx, ` +SELECT to_regclass('public.user_update_privacy_payloads') IS NULL`).Scan(&privacyPayloadTableAbsent); err != nil { + t.Fatalf("inspect privacy payload schema: %v", err) + } + if !privacyPayloadTableAbsent { + t.Fatal("development-only user_update_privacy_payloads table still exists") + } + if _, err := pool.Exec(ctx, ` +INSERT INTO user_update_events (user_id, pts, pts_count, date, event_type) +VALUES ($1, 1, 1, 1700000000, 'privacy')`, user.ID); err == nil { + t.Fatal("development-only privacy update event type is still accepted") + } else { + var pgErr *pgconn.PgError + if !errors.As(err, &pgErr) || pgErr.Code != pgerrcode.CheckViolation { + t.Fatalf("insert privacy update event error=%v, want check violation", err) + } + } + + type updateFootprint struct { + eventCount int + maxPts int + outboxCount int + watermarkRow int + watermarkPts int + } + readFootprint := func() updateFootprint { + t.Helper() + var got updateFootprint + if err := pool.QueryRow(ctx, ` +SELECT count(*), COALESCE(max(pts), 0) +FROM user_update_events +WHERE user_id = $1`, user.ID).Scan(&got.eventCount, &got.maxPts); err != nil { + t.Fatalf("read update events footprint: %v", err) + } + if err := pool.QueryRow(ctx, ` +SELECT count(*) +FROM dispatch_outbox +WHERE target_user_id = $1`, user.ID).Scan(&got.outboxCount); err != nil { + t.Fatalf("read outbox footprint: %v", err) + } + if err := pool.QueryRow(ctx, ` +SELECT count(*), COALESCE(max(contiguous_pts), 0) +FROM user_update_watermarks +WHERE user_id = $1`, user.ID).Scan(&got.watermarkRow, &got.watermarkPts); err != nil { + t.Fatalf("read update watermark footprint: %v", err) + } + return got + } + + before := readFootprint() + store := NewPrivacyStore(pool) + want := domain.PrivacyRules{ + OwnerUserID: user.ID, + Key: domain.PrivacyKeyPhoneNumber, + Rules: []domain.PrivacyRule{{Kind: domain.PrivacyRuleDisallowAll}}, + } + if err := store.SetPrivacyRules(ctx, want); err != nil { + t.Fatalf("set privacy rules: %v", err) + } + got, found, err := store.GetPrivacyRules(ctx, user.ID, want.Key) + if err != nil || !found { + t.Fatalf("get privacy rules: found=%v err=%v", found, err) + } + if len(got.Rules) != 1 || got.Rules[0].Kind != domain.PrivacyRuleDisallowAll { + t.Fatalf("stored privacy rules=%+v, want disallow_all", got) + } + after := readFootprint() + if after != before { + t.Fatalf("privacy write changed PTS footprint: before=%+v after=%+v", before, after) + } +} diff --git a/internal/store/postgres/update_event.go b/internal/store/postgres/update_event.go index 76ef2530..364a831b 100644 --- a/internal/store/postgres/update_event.go +++ b/internal/store/postgres/update_event.go @@ -244,35 +244,12 @@ func appendUserUpdateEvent(ctx context.Context, db sqlcgen.DBTX, q *sqlcgen.Quer }); err != nil { return err } - if err := appendPrivacyPayload(ctx, db, userID, event); err != nil { - return err - } if err := appendQuickReplyPayload(ctx, db, userID, event); err != nil { return err } return nil } -func appendPrivacyPayload(ctx context.Context, db sqlcgen.DBTX, userID int64, event domain.UpdateEvent) error { - if event.Type != domain.UpdateEventPrivacy { - return nil - } - if event.Privacy.OwnerUserID != userID || event.Privacy.Key == "" || len(event.Privacy.Rules) == 0 { - return domain.ErrPrivacyRuleInvalid - } - raw, err := json.Marshal(event.Privacy) - if err != nil { - return fmt.Errorf("encode privacy update payload: %w", err) - } - if _, err := db.Exec(ctx, ` -INSERT INTO user_update_privacy_payloads (user_id, pts, payload) -VALUES ($1, $2, $3::jsonb) -`, userID, event.Pts, string(raw)); err != nil { - return fmt.Errorf("save privacy update payload: %w", err) - } - return nil -} - func appendQuickReplyPayload(ctx context.Context, db sqlcgen.DBTX, userID int64, event domain.UpdateEvent) error { switch event.Type { case domain.UpdateEventQuickReplies, @@ -489,9 +466,6 @@ func (s *UpdateEventStore) ListAfter(ctx context.Context, userID int64, pts, lim } out = append(out, event) } - if err := s.hydratePrivacyEvents(ctx, out); err != nil { - return nil, err - } return out, nil } @@ -691,77 +665,9 @@ func (s *UpdateEventStore) BatchByCursor(ctx context.Context, cursors []store.Ev } out = append(out, event) } - if err := s.hydratePrivacyEvents(ctx, out); err != nil { - return nil, err - } return out, nil } -type privacyEventCursor struct { - userID int64 - pts int -} - -// hydratePrivacyEvents fetches all immutable privacy payloads for one -// difference/outbox batch in one query. Ordinary event batches incur no extra -// query at all. -func (s *UpdateEventStore) hydratePrivacyEvents(ctx context.Context, events []domain.UpdateEvent) error { - indexes := make(map[privacyEventCursor]int) - userIDs := make([]int64, 0) - pts := make([]int32, 0) - for i := range events { - if events[i].Type != domain.UpdateEventPrivacy { - continue - } - key := privacyEventCursor{userID: events[i].UserID, pts: events[i].Pts} - indexes[key] = i - userIDs = append(userIDs, key.userID) - pts = append(pts, int32(key.pts)) - } - if len(indexes) == 0 { - return nil - } - rows, err := s.db.Query(ctx, ` -SELECT p.user_id, p.pts, p.payload::text -FROM unnest($1::bigint[], $2::int[]) AS requested(user_id, pts) -JOIN user_update_privacy_payloads p USING (user_id, pts) -`, userIDs, pts) - if err != nil { - return fmt.Errorf("list privacy update payloads: %w", err) - } - defer rows.Close() - found := 0 - for rows.Next() { - var userID int64 - var eventPts int - var raw string - if err := rows.Scan(&userID, &eventPts, &raw); err != nil { - return fmt.Errorf("scan privacy update payload: %w", err) - } - index, ok := indexes[privacyEventCursor{userID: userID, pts: eventPts}] - if !ok { - continue - } - var payload domain.PrivacyRules - if err := json.Unmarshal([]byte(raw), &payload); err != nil { - return fmt.Errorf("decode privacy update payload: %w", err) - } - if payload.OwnerUserID != userID || payload.Key == "" || len(payload.Rules) == 0 { - return fmt.Errorf("invalid privacy update payload for user %d pts %d", userID, eventPts) - } - events[index].Privacy = payload - delete(indexes, privacyEventCursor{userID: userID, pts: eventPts}) - found++ - } - if err := rows.Err(); err != nil { - return fmt.Errorf("list privacy update payloads rows: %w", err) - } - if found != len(userIDs) || len(indexes) != 0 { - return fmt.Errorf("privacy update payload missing") - } - return nil -} - func usersFromUpdateEventRow(row sqlcgen.ListUserUpdateEventsAfterRow) []domain.User { return mergeEventUsers( domain.User{ diff --git a/internal/store/privacy.go b/internal/store/privacy.go index 1f3c1860..e6577568 100644 --- a/internal/store/privacy.go +++ b/internal/store/privacy.go @@ -12,16 +12,3 @@ type PrivacyStore interface { SetPrivacyRules(ctx context.Context, rules domain.PrivacyRules) error ListPrivacyRules(ctx context.Context, ownerUserIDs []int64, keys []domain.PrivacyKey) ([]domain.PrivacyRules, error) } - -// PrivacyUpdateStore atomically commits an absolute privacy rule snapshot, -// allocates account pts, appends its durable event, and enqueues online -// dispatch. Implementations return the event with its final pts. -type PrivacyUpdateStore interface { - SetPrivacyRulesWithUpdate( - ctx context.Context, - rules domain.PrivacyRules, - event domain.UpdateEvent, - excludeAuthKeyID [8]byte, - excludeSessionID int64, - ) (domain.UpdateEvent, error) -}