From 5d5883a3d19053154eca0433e68e0d0a754766bc Mon Sep 17 00:00:00 2001 From: iamxvbaba <28732408+iamxvbaba@users.noreply.github.com> Date: Fri, 24 Jul 2026 11:56:59 +0800 Subject: [PATCH] feat: sync privacy read-model rules --- cmd/telesrv/main.go | 1 + .../0146_account_settings_read_model.down.sql | 2 + .../0146_account_settings_read_model.up.sql | 25 +++ internal/app/privacy/service.go | 15 ++ internal/app/privacy/service_test.go | 141 +++++++++++++++ internal/rpc/account_privacy_rpc_test.go | 122 +++++++++++++ internal/rpc/account_settings_cache.go | 39 +++- internal/rpc/messages_create_chat_test.go | 94 ++++++++++ internal/rpc/messages_forward.go | 8 + internal/rpc/messages_helpers.go | 2 +- internal/rpc/messages_send.go | 6 +- internal/rpc/messages_send_rpc_test.go | 33 +++- internal/rpc/payments_star_gifts_rpc_test.go | 67 +++++++ internal/rpc/privacy_paid_messages.go | 137 +++++++++++++++ internal/rpc/privacy_paid_messages_test.go | 166 ++++++++++++++++++ internal/rpc/send_media.go | 25 ++- internal/rpc/send_media_test.go | 72 ++++++++ internal/rpc/users.go | 7 + .../postgres/channel_member_cache_test.go | 39 ++++ .../store/postgres/read_model_listener.go | 30 +++- 20 files changed, 1011 insertions(+), 20 deletions(-) create mode 100644 deploy/migrations/0146_account_settings_read_model.down.sql create mode 100644 deploy/migrations/0146_account_settings_read_model.up.sql create mode 100644 internal/rpc/account_privacy_rpc_test.go create mode 100644 internal/rpc/privacy_paid_messages.go create mode 100644 internal/rpc/privacy_paid_messages_test.go diff --git a/cmd/telesrv/main.go b/cmd/telesrv/main.go index 73405588..c518471c 100644 --- a/cmd/telesrv/main.go +++ b/cmd/telesrv/main.go @@ -937,6 +937,7 @@ func run(logger *zap.Logger) error { BaseUsers: userCache, BotProfiles: botsService, StarGifts: giftsService, + AccountSettings: router, }, logger.Named("store").Named("read-model-listener")) go readModelListener.Run(ctx) activeSessions.SetLifecycleObserver(router) diff --git a/deploy/migrations/0146_account_settings_read_model.down.sql b/deploy/migrations/0146_account_settings_read_model.down.sql new file mode 100644 index 00000000..d57e6c93 --- /dev/null +++ b/deploy/migrations/0146_account_settings_read_model.down.sql @@ -0,0 +1,2 @@ +DROP TRIGGER IF EXISTS account_settings_read_model_changed ON account_settings; +DROP FUNCTION IF EXISTS telesrv_notify_account_settings_read_model(); diff --git a/deploy/migrations/0146_account_settings_read_model.up.sql b/deploy/migrations/0146_account_settings_read_model.up.sql new file mode 100644 index 00000000..d20b1e1e --- /dev/null +++ b/deploy/migrations/0146_account_settings_read_model.up.sql @@ -0,0 +1,25 @@ +CREATE OR REPLACE FUNCTION telesrv_notify_account_settings_read_model() +RETURNS trigger +LANGUAGE plpgsql +AS $$ +DECLARE + owner_id bigint; +BEGIN + owner_id := CASE WHEN TG_OP = 'DELETE' THEN OLD.user_id ELSE NEW.user_id END; + IF owner_id IS NOT NULL AND owner_id > 0 THEN + PERFORM telesrv_bump_read_model_version( + 'account_settings', + owner_id, + 'user', + owner_id + ); + END IF; + RETURN CASE WHEN TG_OP = 'DELETE' THEN OLD ELSE NEW END; +END; +$$; + +DROP TRIGGER IF EXISTS account_settings_read_model_changed ON account_settings; +CREATE TRIGGER account_settings_read_model_changed +AFTER INSERT OR UPDATE OR DELETE ON account_settings +FOR EACH ROW +EXECUTE FUNCTION telesrv_notify_account_settings_read_model(); diff --git a/internal/app/privacy/service.go b/internal/app/privacy/service.go index ac86349c..5ac0ee87 100644 --- a/internal/app/privacy/service.go +++ b/internal/app/privacy/service.go @@ -363,6 +363,21 @@ func (s *Service) CanContactForFreeBatch(ctx context.Context, ownerUserIDs []int return out, nil } +// ViewerIsPremium reads the same bounded viewer-facts read model used by +// AllowPremium privacy rules. Contact permission checks must not bypass that +// cache with a per-send users-table query. +func (s *Service) ViewerIsPremium(ctx context.Context, viewerUserID int64) (bool, error) { + if viewerUserID == 0 { + return false, nil + } + facts, err := s.loadViewerFacts(ctx, []int64{viewerUserID}) + if err != nil { + return false, err + } + fact := facts[viewerUserID] + return fact.Found && !fact.Bot && fact.PremiumUntil > s.now().Unix(), nil +} + // CanSeeMatrix 批量评估 owners × viewers × keys 的可见性矩阵，结果等价于逐 (owner,viewer,key) // 调 CanSee，但只用一次 ListPrivacyRules + 每 owner 一次 GetMany(owner,viewers) + 内存 Evaluate // （把 fan-out 投影从 O(viewer) 次 privacy 查询降到 O(owner)）。返回 map[owner]map[viewer]map[key]bool。 diff --git a/internal/app/privacy/service_test.go b/internal/app/privacy/service_test.go index da1fff74..b74d62c3 100644 --- a/internal/app/privacy/service_test.go +++ b/internal/app/privacy/service_test.go @@ -3,11 +3,44 @@ package privacy import ( "context" "testing" + "time" "telesrv/internal/domain" "telesrv/internal/store/memory" ) +type countingBaseUsers struct { + calls int + users map[int64]domain.User +} + +func (p *countingBaseUsers) PrivacyBaseUsers(_ context.Context, userIDs []int64) ([]domain.User, error) { + p.calls++ + out := make([]domain.User, 0, len(userIDs)) + for _, userID := range userIDs { + if user, ok := p.users[userID]; ok { + out = append(out, user) + } + } + return out, nil +} + +type countingMemberships struct { + calls int + active map[int64]map[int64]bool +} + +func (p *countingMemberships) FilterActiveChannelMemberIDs(_ context.Context, channelID int64, userIDs []int64) ([]int64, error) { + p.calls++ + out := make([]int64, 0, len(userIDs)) + for _, userID := range userIDs { + if p.active[channelID][userID] { + out = append(out, userID) + } + } + return out, nil +} + func TestDefaultPrivacyRules(t *testing.T) { ctx := context.Background() svc := NewService(memory.NewPrivacyStore(), memory.NewContactStore()) @@ -192,3 +225,111 @@ func TestCanSeeMatrixEquivalentToCanSee(t *testing.T) { } } } + +func TestViewerFactsReadModelBatchesCachesAndInvalidates(t *testing.T) { + ctx := context.Background() + rules := memory.NewPrivacyStore() + users := &countingBaseUsers{users: map[int64]domain.User{ + 2001: {ID: 2001, PremiumUntil: 2000}, + 2002: {ID: 2002, Bot: true}, + }} + svc := NewService(rules, memory.NewContactStore()).ConfigureReadModels(users, nil) + svc.now = func() time.Time { return time.Unix(1000, 0) } + + if _, err := svc.SetRules(ctx, 1001, domain.PrivacyKeyNoPaidMessages, []domain.PrivacyRule{ + {Kind: domain.PrivacyRuleAllowPremium}, + {Kind: domain.PrivacyRuleDisallowAll}, + }); err != nil { + t.Fatalf("set premium rules: %v", err) + } + if _, err := svc.SetRules(ctx, 1002, domain.PrivacyKeyNoPaidMessages, []domain.PrivacyRule{ + {Kind: domain.PrivacyRuleAllowBots}, + {Kind: domain.PrivacyRuleDisallowAll}, + }); err != nil { + t.Fatalf("set bot rules: %v", err) + } + + got, err := svc.CanSeeMatrix( + ctx, + []int64{1001, 1002}, + []int64{2001, 2002}, + []domain.PrivacyKey{domain.PrivacyKeyNoPaidMessages}, + ) + if err != nil { + t.Fatalf("CanSeeMatrix: %v", err) + } + if !got[1001][2001][domain.PrivacyKeyNoPaidMessages] || + got[1001][2002][domain.PrivacyKeyNoPaidMessages] || + got[1002][2001][domain.PrivacyKeyNoPaidMessages] || + !got[1002][2002][domain.PrivacyKeyNoPaidMessages] { + t.Fatalf("unexpected premium/bot visibility matrix: %+v", got) + } + if users.calls != 1 { + t.Fatalf("base user cold loads = %d, want one batched load", users.calls) + } + + if premium, err := svc.ViewerIsPremium(ctx, 2001); err != nil || !premium { + t.Fatalf("warm ViewerIsPremium = %v, err=%v; want true", premium, err) + } + if users.calls != 1 { + t.Fatalf("warm viewer facts hit called backend: calls=%d", users.calls) + } + + users.users[2001] = domain.User{ID: 2001} + svc.InvalidateViewerFacts(2001) + if premium, err := svc.ViewerIsPremium(ctx, 2001); err != nil || premium { + t.Fatalf("invalidated ViewerIsPremium = %v, err=%v; want false", premium, err) + } + if users.calls != 2 { + t.Fatalf("invalidated viewer facts cold loads = %d, want 2", users.calls) + } +} + +func TestMembershipReadModelCachesNegativeFactsAndInvalidatesPair(t *testing.T) { + ctx := context.Background() + rules := memory.NewPrivacyStore() + memberships := &countingMemberships{active: map[int64]map[int64]bool{ + 9001: {2001: true}, + 9002: {}, + }} + svc := NewService(rules, memory.NewContactStore()).ConfigureReadModels(nil, memberships) + if _, err := svc.SetRules(ctx, 1001, domain.PrivacyKeyChatInvite, []domain.PrivacyRule{ + {Kind: domain.PrivacyRuleAllowChatParticipants, ChatIDs: []int64{9001, 9002}}, + {Kind: domain.PrivacyRuleDisallowAll}, + }); err != nil { + t.Fatalf("set participant rules: %v", err) + } + + got, err := svc.CanSeeMatrix( + ctx, + []int64{1001}, + []int64{2001, 2002}, + []domain.PrivacyKey{domain.PrivacyKeyChatInvite}, + ) + if err != nil { + t.Fatalf("CanSeeMatrix: %v", err) + } + if !got[1001][2001][domain.PrivacyKeyChatInvite] || + got[1001][2002][domain.PrivacyKeyChatInvite] { + t.Fatalf("unexpected membership visibility matrix: %+v", got) + } + if memberships.calls != 2 { + t.Fatalf("membership cold loads = %d, want one batch per referenced chat", memberships.calls) + } + + if allowed, err := svc.CanSee(ctx, 1001, 2002, domain.PrivacyKeyChatInvite); err != nil || allowed { + t.Fatalf("warm negative membership = %v, err=%v; want false", allowed, err) + } + if memberships.calls != 2 { + t.Fatalf("negative cache missed: calls=%d", memberships.calls) + } + + memberships.active[9002][2002] = true + svc.InvalidateMembership(9002, 2002) + if allowed, err := svc.CanSee(ctx, 1001, 2002, domain.PrivacyKeyChatInvite); err != nil || !allowed { + t.Fatalf("invalidated membership = %v, err=%v; want true", allowed, err) + } + if memberships.calls != 3 { + t.Fatalf("pair invalidation reloads = %d, want 3", memberships.calls) + } +} diff --git a/internal/rpc/account_privacy_rpc_test.go b/internal/rpc/account_privacy_rpc_test.go new file mode 100644 index 00000000..a97e3eed --- /dev/null +++ b/internal/rpc/account_privacy_rpc_test.go @@ -0,0 +1,122 @@ +package rpc + +import ( + "context" + "testing" + + "github.com/iamxvbaba/td/clock" + "github.com/iamxvbaba/td/tg" + "go.uber.org/zap/zaptest" + + appprivacy "telesrv/internal/app/privacy" + appupdates "telesrv/internal/app/updates" + "telesrv/internal/domain" + "telesrv/internal/store/memory" +) + +func TestAccountPrivacyAllKeysRoundTripAndRecordDifferenceEvents(t *testing.T) { + ctx := context.Background() + const userID int64 = 8101 + authKeyID := [8]byte{8, 1} + sessionID := int64(81) + privacy := appprivacy.NewService(memory.NewPrivacyStore(), memory.NewContactStore()) + events := memory.NewUpdateEventStore() + updates := appupdates.NewService(memory.NewUpdateStateStore(), events) + router := New(Config{}, Deps{ + Privacy: privacy, + Updates: updates, + Sessions: &captureSessions{}, + }, zaptest.NewLogger(t), clock.System) + requestCtx := WithSessionID(WithAuthKeyID(WithUserID(ctx, userID), authKeyID), sessionID) + + keys := []struct { + name string + input tg.InputPrivacyKeyClass + domain domain.PrivacyKey + wire func(tg.PrivacyKeyClass) bool + }{ + {"status_timestamp", &tg.InputPrivacyKeyStatusTimestamp{}, domain.PrivacyKeyStatusTimestamp, func(v tg.PrivacyKeyClass) bool { _, ok := v.(*tg.PrivacyKeyStatusTimestamp); return ok }}, + {"chat_invite", &tg.InputPrivacyKeyChatInvite{}, domain.PrivacyKeyChatInvite, func(v tg.PrivacyKeyClass) bool { _, ok := v.(*tg.PrivacyKeyChatInvite); return ok }}, + {"phone_call", &tg.InputPrivacyKeyPhoneCall{}, domain.PrivacyKeyPhoneCall, func(v tg.PrivacyKeyClass) bool { _, ok := v.(*tg.PrivacyKeyPhoneCall); return ok }}, + {"phone_p2p", &tg.InputPrivacyKeyPhoneP2P{}, domain.PrivacyKeyPhoneP2P, func(v tg.PrivacyKeyClass) bool { _, ok := v.(*tg.PrivacyKeyPhoneP2P); return ok }}, + {"forwards", &tg.InputPrivacyKeyForwards{}, domain.PrivacyKeyForwards, func(v tg.PrivacyKeyClass) bool { _, ok := v.(*tg.PrivacyKeyForwards); return ok }}, + {"profile_photo", &tg.InputPrivacyKeyProfilePhoto{}, domain.PrivacyKeyProfilePhoto, func(v tg.PrivacyKeyClass) bool { _, ok := v.(*tg.PrivacyKeyProfilePhoto); return ok }}, + {"phone_number", &tg.InputPrivacyKeyPhoneNumber{}, domain.PrivacyKeyPhoneNumber, func(v tg.PrivacyKeyClass) bool { _, ok := v.(*tg.PrivacyKeyPhoneNumber); return ok }}, + {"added_by_phone", &tg.InputPrivacyKeyAddedByPhone{}, domain.PrivacyKeyAddedByPhone, func(v tg.PrivacyKeyClass) bool { _, ok := v.(*tg.PrivacyKeyAddedByPhone); return ok }}, + {"voice_messages", &tg.InputPrivacyKeyVoiceMessages{}, domain.PrivacyKeyVoiceMessages, func(v tg.PrivacyKeyClass) bool { _, ok := v.(*tg.PrivacyKeyVoiceMessages); return ok }}, + {"about", &tg.InputPrivacyKeyAbout{}, domain.PrivacyKeyAbout, func(v tg.PrivacyKeyClass) bool { _, ok := v.(*tg.PrivacyKeyAbout); return ok }}, + {"birthday", &tg.InputPrivacyKeyBirthday{}, domain.PrivacyKeyBirthday, func(v tg.PrivacyKeyClass) bool { _, ok := v.(*tg.PrivacyKeyBirthday); return ok }}, + {"star_gifts_auto_save", &tg.InputPrivacyKeyStarGiftsAutoSave{}, domain.PrivacyKeyStarGiftsAutoSave, func(v tg.PrivacyKeyClass) bool { _, ok := v.(*tg.PrivacyKeyStarGiftsAutoSave); return ok }}, + {"no_paid_messages", &tg.InputPrivacyKeyNoPaidMessages{}, domain.PrivacyKeyNoPaidMessages, func(v tg.PrivacyKeyClass) bool { _, ok := v.(*tg.PrivacyKeyNoPaidMessages); return ok }}, + {"saved_music", &tg.InputPrivacyKeySavedMusic{}, domain.PrivacyKeySavedMusic, func(v tg.PrivacyKeyClass) bool { _, ok := v.(*tg.PrivacyKeySavedMusic); return ok }}, + } + + for _, test := range keys { + t.Run(test.name, func(t *testing.T) { + gotKey, ok := domainPrivacyKeyFromInput(test.input) + if !ok || gotKey != test.domain { + t.Fatalf("input key maps to %q/%v, want %q/true", gotKey, ok, test.domain) + } + if !test.wire(tgPrivacyKey(test.domain)) { + t.Fatalf("domain key %q projected as %T", test.domain, tgPrivacyKey(test.domain)) + } + set, err := router.onAccountSetPrivacy(requestCtx, &tg.AccountSetPrivacyRequest{ + Key: test.input, + Rules: []tg.InputPrivacyRuleClass{&tg.InputPrivacyValueDisallowAll{}}, + }) + if err != nil { + t.Fatalf("setPrivacy: %v", err) + } + if len(set.Rules) != 1 { + t.Fatalf("setPrivacy rules=%d, want 1", len(set.Rules)) + } + if _, ok := set.Rules[0].(*tg.PrivacyValueDisallowAll); !ok { + t.Fatalf("setPrivacy rule=%T, want disallowAll", set.Rules[0]) + } + get, err := router.onAccountGetPrivacy(requestCtx, test.input) + if err != nil { + t.Fatalf("getPrivacy: %v", err) + } + if len(get.Rules) != 1 { + t.Fatalf("getPrivacy rules=%d, want 1", len(get.Rules)) + } + if _, ok := get.Rules[0].(*tg.PrivacyValueDisallowAll); !ok { + t.Fatalf("getPrivacy rule=%T, want disallowAll", get.Rules[0]) + } + }) + } + + recorded, err := events.ListAfter(ctx, userID, 0, 100) + if err != nil { + t.Fatalf("list privacy events: %v", err) + } + if len(recorded) != len(keys) { + t.Fatalf("privacy events=%d, want %d", len(recorded), len(keys)) + } + 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) + } + } + + 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)) + } + 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) + } + } +} diff --git a/internal/rpc/account_settings_cache.go b/internal/rpc/account_settings_cache.go index 9dee2c27..87b22bab 100644 --- a/internal/rpc/account_settings_cache.go +++ b/internal/rpc/account_settings_cache.go @@ -10,9 +10,10 @@ import ( const ( accountSettingsCacheMaxEntries = 4096 - // accountSettingsCacheTTL 兜底跨实例失效；同实例 Set 即时失效。设置页连续调 - // getGlobalPrivacy/getAccountTTL/getContentSettings/getContactSignUp 时只查一次 PG。 - accountSettingsCacheTTL = 60 * time.Second + // accountSettingsCacheTTL is only the lost-notification safety net. Normal + // consistency comes from account_settings read-model notifications; local + // writes update the cached value directly. + accountSettingsCacheTTL = 24 * time.Hour ) // accountSettingsCache 缓存 userID→AccountSettings，避免设置页 4 个 get handler 各查 @@ -52,6 +53,38 @@ func (c *accountSettingsCache) Store(userID int64, settings domain.AccountSettin c.cache.Store(userID, settings) } +func (c *accountSettingsCache) Flush() { + if c == nil { + return + } + c.cache.Flush() +} + +// InvalidateAccountSettingsReadModel is called by the shared PostgreSQL +// read-model listener. Router owns this cache, so exposing the invalidation on +// Router keeps store/postgres independent from the RPC package. +func (r *Router) InvalidateAccountSettingsReadModel(userID int64) { + if r == nil || r.accountSettings == nil { + return + } + r.accountSettings.Delete(userID) +} + +func (r *Router) FlushAccountSettingsReadModel() { + if r == nil || r.accountSettings == nil { + return + } + r.accountSettings.Flush() +} + +func (r *Router) WarmAccountSettingsReadModel(ctx context.Context, userID int64) error { + if r == nil || userID == 0 { + return nil + } + _, err := r.cachedAccountSettings(ctx, userID) + return err +} + type accountSettingsBatchReader interface { GetAccountSettingsBatch(ctx context.Context, userIDs []int64) (map[int64]domain.AccountSettings, error) } diff --git a/internal/rpc/messages_create_chat_test.go b/internal/rpc/messages_create_chat_test.go index 53108baa..2dd92af5 100644 --- a/internal/rpc/messages_create_chat_test.go +++ b/internal/rpc/messages_create_chat_test.go @@ -8,6 +8,7 @@ import ( "go.uber.org/zap/zaptest" appchannels "telesrv/internal/app/channels" appdialogs "telesrv/internal/app/dialogs" + appprivacy "telesrv/internal/app/privacy" appusers "telesrv/internal/app/users" "telesrv/internal/domain" "telesrv/internal/store/memory" @@ -119,6 +120,99 @@ func TestMessagesCreateChatCreatesMegagroupAndDialogsRPC(t *testing.T) { } } +func TestMessagesCreateChatFiltersPrivacyBeforeMembershipWritesRPC(t *testing.T) { + ctx := context.Background() + userStore := memory.NewUserStore() + owner, err := userStore.Create(ctx, domain.User{AccessHash: 31, Phone: "15550001031", FirstName: "Owner"}) + if err != nil { + t.Fatalf("create owner: %v", err) + } + allowed, err := userStore.Create(ctx, domain.User{AccessHash: 32, Phone: "15550001032", FirstName: "Allowed"}) + if err != nil { + t.Fatalf("create allowed user: %v", err) + } + denied, err := userStore.Create(ctx, domain.User{AccessHash: 33, Phone: "15550001033", FirstName: "Denied"}) + if err != nil { + t.Fatalf("create denied user: %v", err) + } + privacy := appprivacy.NewService(memory.NewPrivacyStore(), memory.NewContactStore()) + if _, err := privacy.SetRules(ctx, denied.ID, domain.PrivacyKeyChatInvite, []domain.PrivacyRule{ + {Kind: domain.PrivacyRuleDisallowAll}, + }); err != nil { + t.Fatalf("set denied invite privacy: %v", err) + } + channelStore := memory.NewChannelStore() + r := New(Config{}, Deps{ + Users: appusers.NewService(userStore), + Channels: appchannels.NewService(channelStore), + Dialogs: appdialogs.NewService(memory.NewDialogStore(), channelStore), + Privacy: privacy, + }, zaptest.NewLogger(t), clock.System) + + invited, err := r.onMessagesCreateChat(WithUserID(ctx, owner.ID), &tg.MessagesCreateChatRequest{ + Users: []tg.InputUserClass{ + &tg.InputUser{UserID: allowed.ID, AccessHash: allowed.AccessHash}, + &tg.InputUser{UserID: denied.ID, AccessHash: denied.AccessHash}, + }, + Title: "Privacy Group", + }) + if err != nil { + t.Fatalf("create chat: %v", err) + } + if len(invited.MissingInvitees) != 1 || invited.MissingInvitees[0].UserID != denied.ID { + t.Fatalf("missing invitees = %+v, want denied user %d", invited.MissingInvitees, denied.ID) + } + updates, ok := invited.Updates.(*tg.Updates) + if !ok { + t.Fatalf("updates = %T, want *tg.Updates", invited.Updates) + } + var channel *tg.Channel + for _, chat := range updates.Chats { + if candidate, ok := chat.(*tg.Channel); ok { + channel = candidate + break + } + } + if channel == nil || channel.ParticipantsCount != 2 { + t.Fatalf("channel = %#v, want owner + allowed only", channel) + } + for _, update := range updates.Updates { + newMessage, ok := update.(*tg.UpdateNewChannelMessage) + if !ok { + continue + } + service, ok := newMessage.Message.(*tg.MessageService) + if !ok { + continue + } + add, ok := service.Action.(*tg.MessageActionChatAddUser) + if !ok { + continue + } + if len(add.Users) != 1 || add.Users[0] != allowed.ID { + t.Fatalf("invite action users = %v, want allowed user %d only", add.Users, allowed.ID) + } + } + + participants, err := r.onChannelsGetParticipants(WithUserID(ctx, owner.ID), &tg.ChannelsGetParticipantsRequest{ + Channel: &tg.InputChannel{ChannelID: channel.ID, AccessHash: channel.AccessHash}, + Filter: &tg.ChannelParticipantsRecent{}, + Limit: 10, + }) + if err != nil { + t.Fatalf("get participants: %v", err) + } + list := participants.(*tg.ChannelsChannelParticipants) + if list.Count != 2 { + t.Fatalf("participant count = %d, want 2", list.Count) + } + for _, user := range list.Users { + if got, ok := user.(*tg.User); ok && got.ID == denied.ID { + t.Fatalf("denied user %d was written as a member", denied.ID) + } + } +} + func TestMessagesCreateChatCreatesOwnerOnlyMegagroupRPC(t *testing.T) { tests := []struct { name string diff --git a/internal/rpc/messages_forward.go b/internal/rpc/messages_forward.go index 2111b3d6..8254630a 100644 --- a/internal/rpc/messages_forward.go +++ b/internal/rpc/messages_forward.go @@ -110,6 +110,14 @@ func (r *Router) onMessagesForwardMessages(ctx context.Context, req *tg.Messages if err != nil { return nil, err } + if toPeer.Type == domain.PeerTypeUser { + if req.AllowPaidFloodskip { + return nil, paymentUnsupportedErr() + } + if err := r.ensurePrivateContactAllowed(ctx, userID, toPeer.ID, req.AllowPaidStars, len(absentIndexes)); err != nil { + return nil, err + } + } absentIDs := make([]int, len(absentIndexes)) absentRandomIDs := make([]int64, len(absentIndexes)) for i, originalIndex := range absentIndexes { diff --git a/internal/rpc/messages_helpers.go b/internal/rpc/messages_helpers.go index dc181842..e7f46406 100644 --- a/internal/rpc/messages_helpers.go +++ b/internal/rpc/messages_helpers.go @@ -412,7 +412,7 @@ func forwardMessagesUnsupportedOptionErr(req *tg.MessagesForwardMessagesRequest) return mediaInvalidErr() case req.AllowPaidStars < 0: return starsAmountInvalidErr() - case req.AllowPaidStars > 0 || req.AllowPaidFloodskip: + case req.AllowPaidFloodskip: return paymentUnsupportedErr() case !req.SuggestedPost.Zero(): return suggestedPostPeerInvalidErr() diff --git a/internal/rpc/messages_send.go b/internal/rpc/messages_send.go index be2e03cc..e253df37 100644 --- a/internal/rpc/messages_send.go +++ b/internal/rpc/messages_send.go @@ -141,10 +141,6 @@ func (r *Router) onMessagesSendMessage(ctx context.Context, req *tg.MessagesSend } return updates, nil } - if req.AllowPaidStars > 0 { - sendErr = paymentUnsupportedErr() - return nil, sendErr - } replay, err := r.lookupOutgoingReplay(ctx, userID, peer, req.RandomID, idempotencyFingerprint) if err != nil { sendErr = err @@ -227,6 +223,7 @@ func (r *Router) onMessagesSendMessage(ctx context.Context, req *tg.MessagesSend sendAsInput: req.SendAs, clearDraft: req.ClearDraft, richMessage: richMessage, + allowPaidStars: req.AllowPaidStars, }, req.ScheduleDate, req.ScheduleRepeatPeriod) if err != nil { sendErr = err @@ -249,6 +246,7 @@ func (r *Router) onMessagesSendMessage(ctx context.Context, req *tg.MessagesSend replyMarkup: replyMarkup, richMessage: richMessage, effect: req.Effect, + allowPaidStars: req.AllowPaidStars, }) duplicate = dup if err != nil { diff --git a/internal/rpc/messages_send_rpc_test.go b/internal/rpc/messages_send_rpc_test.go index 6876ada0..675e270b 100644 --- a/internal/rpc/messages_send_rpc_test.go +++ b/internal/rpc/messages_send_rpc_test.go @@ -392,15 +392,6 @@ func TestMessagesSendMessageUnsupportedOptionErrors(t *testing.T) { }(), want: "STARS_AMOUNT_INVALID", }, - { - name: "paid stars", - req: func() *tg.MessagesSendMessageRequest { - req := base() - req.SetAllowPaidStars(1) - return req - }(), - want: "PAYMENT_UNSUPPORTED", - }, { name: "paid floodskip", req: func() *tg.MessagesSendMessageRequest { @@ -428,3 +419,27 @@ func TestMessagesSendMessageUnsupportedOptionErrors(t *testing.T) { }) } } + +func TestMessagesSendMessageAllowsUnusedPaidAuthorizationForFreeRecipient(t *testing.T) { + const ( + senderID = int64(1000000001) + recipientID = int64(1000000002) + ) + messages := &captureMessages{} + r := New(Config{}, Deps{ + Messages: messages, + Users: mapUsersService{users: map[int64]domain.User{ + senderID: {ID: senderID, FirstName: "Sender"}, + recipientID: {ID: recipientID, FirstName: "Recipient"}, + }}, + }, zaptest.NewLogger(t), clock.System) + req := &tg.MessagesSendMessageRequest{ + Peer: &tg.InputPeerUser{UserID: recipientID}, + Message: "free", + RandomID: 457, + } + req.SetAllowPaidStars(10) + if _, err := r.onMessagesSendMessage(WithUserID(context.Background(), senderID), req); err != nil { + t.Fatalf("free recipient with unused authorization: %v", err) + } +} diff --git a/internal/rpc/payments_star_gifts_rpc_test.go b/internal/rpc/payments_star_gifts_rpc_test.go index 8468e396..e62ea857 100644 --- a/internal/rpc/payments_star_gifts_rpc_test.go +++ b/internal/rpc/payments_star_gifts_rpc_test.go @@ -13,6 +13,7 @@ import ( "github.com/iamxvbaba/td/tlprofile" appchannels "telesrv/internal/app/channels" appmessages "telesrv/internal/app/messages" + appprivacy "telesrv/internal/app/privacy" appstargifts "telesrv/internal/app/stargifts" appstars "telesrv/internal/app/stars" appusers "telesrv/internal/app/users" @@ -983,6 +984,72 @@ func TestStarGiftSaga(t *testing.T) { } } +func TestStarGiftAutoSavePrivacyCreatesPendingGiftUntilRecipientApproves(t *testing.T) { + r, sender, recipient, gift := starGiftTestRouter(t) + ctx := context.Background() + privacy := appprivacy.NewService(memory.NewPrivacyStore(), memory.NewContactStore()) + if _, err := privacy.SetRules(ctx, recipient.ID, domain.PrivacyKeyStarGiftsAutoSave, []domain.PrivacyRule{ + {Kind: domain.PrivacyRuleDisallowAll}, + }); err != nil { + t.Fatalf("set gift auto-save privacy: %v", err) + } + r.deps.Privacy = privacy + senderCtx := WithUserID(ctx, sender.ID) + recipientCtx := WithUserID(ctx, recipient.ID) + invoice := &tg.InputInvoiceStarGift{ + Peer: &tg.InputPeerUser{UserID: recipient.ID, AccessHash: recipient.AccessHash}, + GiftID: gift.ID, + } + formClass, err := r.onPaymentsGetPaymentForm(senderCtx, &tg.PaymentsGetPaymentFormRequest{Invoice: invoice}) + if err != nil { + t.Fatalf("get gift form: %v", err) + } + form := formClass.(*tg.PaymentsPaymentFormStarGift) + if _, err := r.onPaymentsSendStarsForm(senderCtx, &tg.PaymentsSendStarsFormRequest{ + FormID: form.FormID, + Invoice: invoice, + }); err != nil { + t.Fatalf("send gift: %v", err) + } + + all, err := r.onPaymentsGetSavedStarGifts(recipientCtx, &tg.PaymentsGetSavedStarGiftsRequest{ + Peer: &tg.InputPeerSelf{}, + Limit: 10, + }) + if err != nil || all.Count != 1 || len(all.Gifts) != 1 { + t.Fatalf("all saved gifts count=%d len=%d err=%v, want pending gift", all.Count, len(all.Gifts), err) + } + if !all.Gifts[0].Unsaved { + t.Fatal("privacy-denied incoming gift must be pending (unsaved=true), not rejected") + } + msgID, ok := all.Gifts[0].GetMsgID() + if !ok || msgID == 0 { + t.Fatalf("pending gift msg_id=%d present=%v", msgID, ok) + } + visible, err := r.onPaymentsGetSavedStarGifts(recipientCtx, &tg.PaymentsGetSavedStarGiftsRequest{ + Peer: &tg.InputPeerSelf{}, + ExcludeUnsaved: true, + Limit: 10, + }) + if err != nil || visible.Count != 0 || len(visible.Gifts) != 0 { + t.Fatalf("exclude-unsaved count=%d len=%d err=%v, want hidden pending gift", visible.Count, len(visible.Gifts), err) + } + + if ok, err := r.onPaymentsSaveStarGift(recipientCtx, &tg.PaymentsSaveStarGiftRequest{ + Stargift: &tg.InputSavedStarGiftUser{MsgID: msgID}, + }); err != nil || !ok { + t.Fatalf("recipient approve gift = %v err=%v", ok, err) + } + visible, err = r.onPaymentsGetSavedStarGifts(recipientCtx, &tg.PaymentsGetSavedStarGiftsRequest{ + Peer: &tg.InputPeerSelf{}, + ExcludeUnsaved: true, + Limit: 10, + }) + if err != nil || visible.Count != 1 || len(visible.Gifts) != 1 || visible.Gifts[0].Unsaved { + t.Fatalf("approved visible gifts=%+v err=%v, want one saved gift", visible, err) + } +} + // 频道 star gift saga：channel peer 能付款发送，但不生成频道历史消息； // saved gift 用 inputSavedStarGiftChat.saved_id 定位，Recent Actions 用 admin log 快照承载。 func TestStarGiftChannelSaga(t *testing.T) { diff --git a/internal/rpc/privacy_paid_messages.go b/internal/rpc/privacy_paid_messages.go new file mode 100644 index 00000000..450426ab --- /dev/null +++ b/internal/rpc/privacy_paid_messages.go @@ -0,0 +1,137 @@ +package rpc + +import ( + "context" + "math" + + "github.com/iamxvbaba/td/tg" +) + +type privateContactRequirement struct { + paidStars int64 + requirePremium bool +} + +type privacyContactFreeEvaluator interface { + CanContactForFreeBatch(ctx context.Context, ownerUserIDs []int64, viewerUserID int64) (map[int64]bool, error) +} + +type privacyViewerPremiumEvaluator interface { + ViewerIsPremium(ctx context.Context, viewerUserID int64) (bool, error) +} + +func applyPrivateContactRestrictionToUser(user *tg.User, restriction privateContactRequirement) { + if user == nil { + return + } + user.SetContactRequirePremium(restriction.requirePremium) + if restriction.paidStars > 0 { + user.SetSendPaidMessagesStars(restriction.paidStars) + } else { + user.Flags2.Unset(15) + user.SendPaidMessagesStars = 0 + } +} + +func applyPrivateContactRestrictionToUserFull(full *tg.UserFull, restriction privateContactRequirement) { + if full == nil { + return + } + full.SetContactRequirePremium(restriction.requirePremium) + if restriction.paidStars > 0 { + full.SetSendPaidMessagesStars(restriction.paidStars) + } else { + full.Flags2.Unset(14) + full.SendPaidMessagesStars = 0 + } +} + +// privateContactRequirementFor returns the recipient's current restriction +// from two in-memory read models: +// - account settings: base premium/paid requirement; +// - privacy/contact facts: contacts and PrivacyKeyNoPaidMessages exceptions. +// +// PostgreSQL is only a bounded cache-miss loader. Local writes are +// write-through and cross-instance changes invalidate + prewarm both models. +func (r *Router) privateContactRestrictionFor( + ctx context.Context, + senderUserID, recipientUserID int64, +) (privateContactRequirement, error) { + if r == nil || senderUserID == 0 || recipientUserID == 0 || senderUserID == recipientUserID { + return privateContactRequirement{}, nil + } + if evaluator, ok := r.deps.Privacy.(privacyContactFreeEvaluator); ok { + free, err := evaluator.CanContactForFreeBatch(ctx, []int64{recipientUserID}, senderUserID) + if err != nil { + return privateContactRequirement{}, internalErr() + } + if free[recipientUserID] { + return privateContactRequirement{}, nil + } + } + settings, err := r.cachedAccountSettings(ctx, recipientUserID) + if err != nil { + return privateContactRequirement{}, internalErr() + } + global := settings.GlobalPrivacy + if global.NoncontactPeersPaidStars > 0 { + return privateContactRequirement{paidStars: global.NoncontactPeersPaidStars}, nil + } + return privateContactRequirement{requirePremium: global.NewNoncontactPeersRequirePremium}, nil +} + +func (r *Router) viewerIsPremiumForPrivacy(ctx context.Context, viewerUserID int64) (bool, error) { + var err error + premium := false + if evaluator, ok := r.deps.Privacy.(privacyViewerPremiumEvaluator); ok { + premium, err = evaluator.ViewerIsPremium(ctx, viewerUserID) + if err != nil { + return false, internalErr() + } + } else if r.deps.Users != nil { + user, found, loadErr := r.deps.Users.ByID(ctx, viewerUserID, viewerUserID) + if loadErr != nil { + return false, internalErr() + } + premium = found && user.PremiumActiveAt(r.clock.Now().Unix()) + } + return premium, nil +} + +func (r *Router) ensurePrivateContactAllowed( + ctx context.Context, + senderUserID, recipientUserID, allowPaidStars int64, + messageCount int, +) error { + if allowPaidStars < 0 || messageCount < 1 { + return starsAmountInvalidErr() + } + requirement, err := r.privateContactRestrictionFor(ctx, senderUserID, recipientUserID) + if err != nil { + return err + } + if requirement.requirePremium { + premium, err := r.viewerIsPremiumForPrivacy(ctx, senderUserID) + if err != nil { + return err + } + if !premium { + return premiumAccountRequiredErr() + } + return nil + } + if requirement.paidStars <= 0 { + return nil + } + if requirement.paidStars > math.MaxInt64/int64(messageCount) { + return starsAmountInvalidErr() + } + required := requirement.paidStars * int64(messageCount) + if allowPaidStars < required { + return allowPaymentRequiredErr(required) + } + // The privacy gate and no-paid exception are complete here. The separate + // private paid-message ledger is not part of the current message store yet; + // never accept an authorization without an atomic debit. + return paymentUnsupportedErr() +} diff --git a/internal/rpc/privacy_paid_messages_test.go b/internal/rpc/privacy_paid_messages_test.go new file mode 100644 index 00000000..6d11f483 --- /dev/null +++ b/internal/rpc/privacy_paid_messages_test.go @@ -0,0 +1,166 @@ +package rpc + +import ( + "context" + "strings" + "testing" + "time" + + "github.com/iamxvbaba/td/clock" + "github.com/iamxvbaba/td/tg" + "go.uber.org/zap/zaptest" + + appaccount "telesrv/internal/app/account" + appprivacy "telesrv/internal/app/privacy" + appusers "telesrv/internal/app/users" + "telesrv/internal/domain" + "telesrv/internal/store/memory" +) + +func newPrivateRequirementRouter(t *testing.T, senderPremium bool) (*Router, *appaccount.Service, *appprivacy.Service, *memory.ContactStore, domain.User, domain.User) { + t.Helper() + ctx := context.Background() + usersStore := memory.NewUserStore() + aliceInput := domain.User{AccessHash: 11, Phone: "15550008101", FirstName: "Alice"} + if senderPremium { + aliceInput.PremiumUntil = int(time.Now().Add(time.Hour).Unix()) + } + alice, err := usersStore.Create(ctx, aliceInput) + if err != nil { + t.Fatalf("create Alice: %v", err) + } + bob, err := usersStore.Create(ctx, domain.User{AccessHash: 22, Phone: "15550008102", FirstName: "Bob"}) + if err != nil { + t.Fatalf("create Bob: %v", err) + } + users := appusers.NewService(usersStore) + contacts := memory.NewContactStore() + privacy := appprivacy.NewService(memory.NewPrivacyStore(), contacts).ConfigureReadModels(users, nil) + settingsStore := memory.NewPasswordStore() + account := appaccount.NewService(settingsStore, appaccount.WithAccountSettings(settingsStore)) + router := New(Config{}, Deps{ + Account: account, + Privacy: privacy, + Users: users, + Messages: &captureMessages{}, + }, zaptest.NewLogger(t), clock.System) + return router, account, privacy, contacts, alice, bob +} + +func TestPrivateContactRequirementUsesNoPaidMessagesReadModel(t *testing.T) { + ctx := context.Background() + r, account, privacy, contacts, alice, bob := newPrivateRequirementRouter(t, false) + if _, err := account.SetGlobalPrivacy(ctx, bob.ID, domain.GlobalPrivacy{NoncontactPeersPaidStars: 7}); err != nil { + t.Fatalf("set Bob paid requirement: %v", err) + } + full, err := r.onUsersGetFullUser(WithUserID(ctx, alice.ID), &tg.InputUser{ + UserID: bob.ID, AccessHash: bob.AccessHash, + }) + if err != nil { + t.Fatalf("get Bob full user: %v", err) + } + if stars, ok := full.FullUser.GetSendPaidMessagesStars(); !ok || stars != 7 { + t.Fatalf("full user paid stars = %d, %v; want 7, true", stars, ok) + } + projected, ok := full.Users[0].(*tg.User) + if !ok { + t.Fatalf("projected user = %T, want *tg.User", full.Users[0]) + } + if stars, ok := projected.GetSendPaidMessagesStars(); !ok || stars != 7 { + t.Fatalf("user paid stars = %d, %v; want 7, true", stars, ok) + } + requirements, err := r.onUsersGetRequirementsToContact(WithUserID(ctx, alice.ID), []tg.InputUserClass{ + &tg.InputUser{UserID: bob.ID, AccessHash: bob.AccessHash}, + }) + if err != nil { + t.Fatalf("get requirements: %v", err) + } + if paid, ok := requirements[0].(*tg.RequirementToContactPaidMessages); !ok || paid.StarsAmount != 7 { + t.Fatalf("requirement = %#v, want paid 7", requirements[0]) + } + + send := func(randomID int64, allow int64) error { + req := &tg.MessagesSendMessageRequest{ + Peer: &tg.InputPeerUser{UserID: bob.ID, AccessHash: bob.AccessHash}, + Message: "hello", + RandomID: randomID, + } + if allow != 0 { + req.SetAllowPaidStars(allow) + } + _, err := r.onMessagesSendMessage(WithUserID(ctx, alice.ID), req) + return err + } + + if err := send(81001, 0); err == nil || !strings.Contains(err.Error(), "ALLOW_PAYMENT_REQUIRED") || !strings.Contains(err.Error(), "(7)") { + t.Fatalf("non-exempt send err = %v, want ALLOW_PAYMENT_REQUIRED 7", err) + } + if err := send(81002, 7); err == nil || !strings.Contains(err.Error(), "PAYMENT_UNSUPPORTED") { + t.Fatalf("authorized paid send err = %v, want explicit unsupported ledger", err) + } + + if _, err := privacy.SetRules(ctx, bob.ID, domain.PrivacyKeyNoPaidMessages, []domain.PrivacyRule{{ + Kind: domain.PrivacyRuleAllowUsers, + UserIDs: []int64{alice.ID}, + }, { + Kind: domain.PrivacyRuleDisallowAll, + }}); err != nil { + t.Fatalf("allow Alice in NoPaidMessages: %v", err) + } + if err := send(81003, 0); err != nil { + t.Fatalf("explicit no-paid exception send: %v", err) + } + full, err = r.onUsersGetFullUser(WithUserID(ctx, alice.ID), &tg.InputUser{ + UserID: bob.ID, AccessHash: bob.AccessHash, + }) + if err != nil { + t.Fatalf("get exempt Bob full user: %v", err) + } + if _, ok := full.FullUser.GetSendPaidMessagesStars(); ok || full.FullUser.ContactRequirePremium { + t.Fatalf("exempt full user still carries contact restriction: %+v", full.FullUser) + } + + if _, err := privacy.SetRules(ctx, bob.ID, domain.PrivacyKeyNoPaidMessages, []domain.PrivacyRule{{Kind: domain.PrivacyRuleDisallowAll}}); err != nil { + t.Fatalf("clear explicit exception: %v", err) + } + if _, err := contacts.Upsert(ctx, bob.ID, domain.ContactInput{ + ContactUserID: alice.ID, + FirstName: "Alice", + }); err != nil { + t.Fatalf("Bob add Alice contact: %v", err) + } + if err := send(81004, 0); err != nil { + t.Fatalf("recipient contact must be free: %v", err) + } +} + +func TestPrivateContactRequirementPremiumGateUsesViewerFactsReadModel(t *testing.T) { + ctx := context.Background() + for _, tc := range []struct { + name string + senderPremium bool + wantErr bool + }{ + {name: "non-premium blocked", wantErr: true}, + {name: "premium allowed", senderPremium: true}, + } { + t.Run(tc.name, func(t *testing.T) { + r, account, _, _, alice, bob := newPrivateRequirementRouter(t, tc.senderPremium) + if _, err := account.SetGlobalPrivacy(ctx, bob.ID, domain.GlobalPrivacy{NewNoncontactPeersRequirePremium: true}); err != nil { + t.Fatalf("set premium requirement: %v", err) + } + _, err := r.onMessagesSendMessage(WithUserID(ctx, alice.ID), &tg.MessagesSendMessageRequest{ + Peer: &tg.InputPeerUser{UserID: bob.ID, AccessHash: bob.AccessHash}, + Message: "hello", + RandomID: 82001, + }) + if tc.wantErr { + if err == nil || !strings.Contains(err.Error(), "PREMIUM_ACCOUNT_REQUIRED") { + t.Fatalf("send err = %v, want PREMIUM_ACCOUNT_REQUIRED", err) + } + } else if err != nil { + t.Fatalf("premium sender should pass: %v", err) + } + }) + } +} diff --git a/internal/rpc/send_media.go b/internal/rpc/send_media.go index 1913ab3e..a1caf325 100644 --- a/internal/rpc/send_media.go +++ b/internal/rpc/send_media.go @@ -47,7 +47,8 @@ type outgoingSend struct { groupedID int64 // effect 是消息特效 id（私聊专属，0 表无特效）。调用方已对 catalog 校验合法性； // 频道侧忽略（官方群/频道不渲染特效）。 - effect int64 + effect int64 + allowPaidStars int64 } // sendOutgoing 把一条出站消息落地到私聊或频道，返回 *tg.Updates、是否重复、错误。 @@ -127,6 +128,9 @@ func (r *Router) sendOutgoing(ctx context.Context, userID int64, peer domain.Pee if r.deps.Messages == nil { return nil, false, peerIDInvalidErr() } + if err := r.ensurePrivateContactAllowed(ctx, userID, peer.ID, p.allowPaidStars, 1); err != nil { + return nil, false, err + } if err := r.ensureVoiceMessagesAllowed(ctx, userID, peer, p.media != nil && p.media.HasUnreadPayload()); err != nil { return nil, false, err } @@ -357,7 +361,7 @@ func (r *Router) onMessagesSendMedia(ctx context.Context, req *tg.MessagesSendMe ClearDraft: req.ClearDraft, }) } - if req.AllowPaidStars > 0 || req.AllowPaidFloodskip { + if req.AllowPaidFloodskip { return nil, paymentUnsupportedErr() } replay, err := r.lookupOutgoingReplay(ctx, userID, peer, req.RandomID, idempotencyFingerprint) @@ -417,6 +421,7 @@ func (r *Router) onMessagesSendMedia(ctx context.Context, req *tg.MessagesSendMe replyToInput: req.ReplyTo, sendAsInput: req.SendAs, clearDraft: req.ClearDraft, + allowPaidStars: req.AllowPaidStars, }, req.ScheduleDate, req.ScheduleRepeatPeriod) } updates, _, err := r.sendOutgoing(ctx, userID, peer, outgoingSend{ @@ -433,6 +438,7 @@ func (r *Router) onMessagesSendMedia(ctx context.Context, req *tg.MessagesSendMe clearDraft: req.ClearDraft, replyMarkup: replyMarkup, effect: req.Effect, + allowPaidStars: req.AllowPaidStars, }) if err != nil { return nil, err @@ -452,6 +458,12 @@ func (r *Router) onMessagesSendMultiMedia(ctx context.Context, req *tg.MessagesS if len(req.MultiMedia) == 0 || len(req.MultiMedia) > maxSendMultiMediaItems { return nil, limitInvalidErr() } + if req.AllowPaidStars < 0 { + return nil, starsAmountInvalidErr() + } + if req.AllowPaidFloodskip { + return nil, paymentUnsupportedErr() + } peer, ok := r.domainPeerFromInputPeer(userID, req.Peer) if !ok || peer.ID == 0 { return nil, peerIDInvalidErr() @@ -511,6 +523,14 @@ func (r *Router) onMessagesSendMultiMedia(ctx context.Context, req *tg.MessagesS if err != nil { return nil, err } + if peer.Type == domain.PeerTypeUser { + if req.AllowPaidFloodskip { + return nil, paymentUnsupportedErr() + } + if err := r.ensurePrivateContactAllowed(ctx, userID, peer.ID, req.AllowPaidStars, absentCount); err != nil { + return nil, err + } + } pendingMedia := make([]tg.InputMediaClass, 0, absentCount) for i, item := range req.MultiMedia { if !replays[i].found { @@ -571,6 +591,7 @@ func (r *Router) onMessagesSendMultiMedia(ctx context.Context, req *tg.MessagesS sendAsInput: req.SendAs, clearDraft: clearDraftPending, groupedID: groupedID, + allowPaidStars: req.AllowPaidStars, } var result tg.UpdatesClass duplicate := false diff --git a/internal/rpc/send_media_test.go b/internal/rpc/send_media_test.go index 720ecb89..2de6dc30 100644 --- a/internal/rpc/send_media_test.go +++ b/internal/rpc/send_media_test.go @@ -14,6 +14,7 @@ import ( appmessages "telesrv/internal/app/messages" apppolls "telesrv/internal/app/polls" + appprivacy "telesrv/internal/app/privacy" appstories "telesrv/internal/app/stories" appusers "telesrv/internal/app/users" "telesrv/internal/domain" @@ -36,6 +37,8 @@ type fakeFiles struct { resolveWebPageFn func(string) (domain.MessageWebPage, error) lookupWebPageFn func(string) (domain.MessageWebPage, bool) webPagePreviewOn bool + getDocumentsCalls int + createUploadCalls int } type fakeProfilePhotoKey struct { @@ -78,6 +81,7 @@ func (f *fakeFiles) AvailableEffects(context.Context) ([]domain.AvailableEffect, return append([]domain.AvailableEffect(nil), f.effects...), hash & 0x7fffffff, nil } func (f *fakeFiles) GetDocuments(_ context.Context, ids []int64) ([]domain.Document, error) { + f.getDocumentsCalls++ out := make([]domain.Document, 0, len(ids)) for _, id := range ids { if d, ok := f.docs[id]; ok { @@ -564,6 +568,7 @@ func fakeAvatarStaticSizes() []domain.PhotoSize { } } func (f *fakeFiles) CreateDocumentFromUpload(_ context.Context, _ domain.UploadedFileRef, spec domain.DocumentSpec) (domain.Document, error) { + f.createUploadCalls++ return domain.Document{ID: 888, AccessHash: 8, DCID: 2, MimeType: spec.MimeType, Attributes: spec.Attributes}, nil } func (f *fakeFiles) CreateDocumentFromBytes(_ context.Context, data []byte, spec domain.DocumentSpec) (domain.Document, error) { @@ -837,6 +842,73 @@ func TestSendMediaPrivateSticker(t *testing.T) { } } +func TestSendMediaVoicePrivacyPreflightsBeforeUploadMaterialization(t *testing.T) { + ctx := context.Background() + r, owner, friend := newMediaTestRouter(t) + files := r.deps.Files.(*fakeFiles) + privacy := appprivacy.NewService(memory.NewPrivacyStore(), memory.NewContactStore()) + if _, err := privacy.SetRules(ctx, friend.ID, domain.PrivacyKeyVoiceMessages, []domain.PrivacyRule{ + {Kind: domain.PrivacyRuleDisallowAll}, + }); err != nil { + t.Fatalf("set voice privacy: %v", err) + } + r.deps.Privacy = privacy + + _, err := r.onMessagesSendMedia(WithUserID(ctx, owner.ID), &tg.MessagesSendMediaRequest{ + Peer: &tg.InputPeerUser{UserID: friend.ID, AccessHash: friend.AccessHash}, + Media: &tg.InputMediaUploadedDocument{ + File: &tg.InputFile{ID: 7001, Parts: 1, Name: "voice.ogg"}, + MimeType: "audio/ogg", + Attributes: []tg.DocumentAttributeClass{&tg.DocumentAttributeAudio{Voice: true, Duration: 1}}, + }, + RandomID: 7001, + }) + if err == nil || !tgerr.Is(err, "CHAT_SEND_VOICES_FORBIDDEN") { + t.Fatalf("send uploaded voice err=%v, want CHAT_SEND_VOICES_FORBIDDEN", err) + } + if files.createUploadCalls != 0 { + t.Fatalf("uploaded voice materialized %d documents before privacy rejection", files.createUploadCalls) + } + if files.getDocumentsCalls != 0 { + t.Fatalf("uploaded voice unexpectedly loaded documents: calls=%d", files.getDocumentsCalls) + } +} + +func TestSendMediaVoicePrivacyBatchesReferencedDocumentPreflight(t *testing.T) { + ctx := context.Background() + r, owner, friend := newMediaTestRouter(t) + files := r.deps.Files.(*fakeFiles) + files.docs[7011] = domain.Document{ + ID: 7011, + AccessHash: 71, + Attributes: []domain.DocumentAttribute{{Kind: domain.DocAttrVideo, RoundMessage: true}}, + } + privacy := appprivacy.NewService(memory.NewPrivacyStore(), memory.NewContactStore()) + if _, err := privacy.SetRules(ctx, friend.ID, domain.PrivacyKeyVoiceMessages, []domain.PrivacyRule{ + {Kind: domain.PrivacyRuleDisallowAll}, + }); err != nil { + t.Fatalf("set voice privacy: %v", err) + } + r.deps.Privacy = privacy + + _, err := r.onMessagesSendMedia(WithUserID(ctx, owner.ID), &tg.MessagesSendMediaRequest{ + Peer: &tg.InputPeerUser{UserID: friend.ID, AccessHash: friend.AccessHash}, + Media: &tg.InputMediaDocument{ID: &tg.InputDocument{ + ID: 7011, AccessHash: 71, + }}, + RandomID: 7011, + }) + if err == nil || !tgerr.Is(err, "CHAT_SEND_VOICES_FORBIDDEN") { + t.Fatalf("send referenced round video err=%v, want CHAT_SEND_VOICES_FORBIDDEN", err) + } + if files.getDocumentsCalls != 1 { + t.Fatalf("referenced document preflight loads=%d, want one bounded batch", files.getDocumentsCalls) + } + if files.createUploadCalls != 0 { + t.Fatalf("referenced document path unexpectedly materialized upload: calls=%d", files.createUploadCalls) + } +} + func TestSendMultiMediaPartialFailureSubsetRetryKeepsReservedGroupedID(t *testing.T) { ctx := context.Background() r, owner, friend := newMediaTestRouter(t) diff --git a/internal/rpc/users.go b/internal/rpc/users.go index ef576444..59d26855 100644 --- a/internal/rpc/users.go +++ b/internal/rpc/users.go @@ -154,6 +154,11 @@ func (r *Router) onUsersGetFullUser(ctx context.Context, id tg.InputUserClass) ( if err := r.applyBotCanEditToUser(ctx, currentUserID, u, user); err != nil { return nil, err } + contactRestriction, err := r.privateContactRestrictionFor(ctx, currentUserID, u.ID) + if err != nil { + return nil, err + } + applyPrivateContactRestrictionToUser(user, contactRestriction) r.applyStoryMaxIDsToPeerObjects(ctx, currentUserID, []tg.UserClass{user}, nil) loadEpoch := r.userFullProjectionCache.LoadEpoch() if full, ok := r.userFullProjectionCache.Lookup(currentUserID, u.ID); ok { @@ -165,6 +170,7 @@ func (r *Router) onUsersGetFullUser(ctx context.Context, id tg.InputUserClass) ( } r.applyStoriesPinnedAvailableToUserFull(ctx, currentUserID, u.ID, &full) r.applyNotifySettingsToUserFull(ctx, currentUserID, u.ID, &full) + applyPrivateContactRestrictionToUserFull(&full, contactRestriction) chats := r.applyPersonalChannelToUserFull(ctx, currentUserID, u.PersonalChannelID, &full) return &tg.UsersUserFull{ FullUser: full, @@ -185,6 +191,7 @@ func (r *Router) onUsersGetFullUser(ctx context.Context, id tg.InputUserClass) ( } r.applyStoriesPinnedAvailableToUserFull(ctx, currentUserID, u.ID, &full) r.applyNotifySettingsToUserFull(ctx, currentUserID, u.ID, &full) + applyPrivateContactRestrictionToUserFull(&full, contactRestriction) chats := r.applyPersonalChannelToUserFull(ctx, currentUserID, u.PersonalChannelID, &full) return &tg.UsersUserFull{ FullUser: full, diff --git a/internal/store/postgres/channel_member_cache_test.go b/internal/store/postgres/channel_member_cache_test.go index d0d0fcb3..53bb687b 100644 --- a/internal/store/postgres/channel_member_cache_test.go +++ b/internal/store/postgres/channel_member_cache_test.go @@ -207,6 +207,45 @@ func TestReadModelChangeListenerInvalidatesPrivateMediaCountCache(t *testing.T) } } +func TestReadModelChangeListenerInvalidatesAccountSettingsCache(t *testing.T) { + settings := &fakeAccountSettingsReadModelCache{} + listener := NewReadModelChangeListener("", ReadModelCacheSet{ + AccountSettings: settings, + }, nil) + + listener.handlePayload(`{"model":"account_settings","owner_user_id":100,"peer_type":"user","peer_id":100,"version":2}`) + if len(settings.invalidated) != 1 || settings.invalidated[0] != 100 { + t.Fatalf("account_settings invalidations = %v, want [100]", settings.invalidated) + } + if len(settings.warmed) != 1 || settings.warmed[0] != 100 { + t.Fatalf("account_settings warmups = %v, want [100]", settings.warmed) + } + + listener.flush("test") + if settings.flushes != 1 { + t.Fatalf("account_settings flushes = %d, want 1", settings.flushes) + } +} + +type fakeAccountSettingsReadModelCache struct { + invalidated []int64 + warmed []int64 + flushes int +} + +func (f *fakeAccountSettingsReadModelCache) InvalidateAccountSettingsReadModel(userID int64) { + f.invalidated = append(f.invalidated, userID) +} + +func (f *fakeAccountSettingsReadModelCache) FlushAccountSettingsReadModel() { + f.flushes++ +} + +func (f *fakeAccountSettingsReadModelCache) WarmAccountSettingsReadModel(_ context.Context, userID int64) error { + f.warmed = append(f.warmed, userID) + return nil +} + // TestReadModelChangeListenerBotFullFlushesChannelFullBots 回归: bot 改资料(bot_full 事件, // 迁移 0013)须 flush channelFullBotInfoCache(否则群信息页 bot 简介/命令跨实例陈旧至 TTL); // 普通用户的 user_base 事件不得 flush(否则该缓存形同虚设)。 diff --git a/internal/store/postgres/read_model_listener.go b/internal/store/postgres/read_model_listener.go index f63211c2..7317dede 100644 --- a/internal/store/postgres/read_model_listener.go +++ b/internal/store/postgres/read_model_listener.go @@ -37,6 +37,7 @@ type ReadModelCacheSet struct { BaseUsers BaseUserCache BotProfiles BotProfileReadModelCache StarGifts StarGiftCatalogCache + AccountSettings AccountSettingsReadModelCache } type StarGiftCatalogCache interface { @@ -44,6 +45,15 @@ type StarGiftCatalogCache interface { FlushStarGiftCatalog() } +type AccountSettingsReadModelCache interface { + InvalidateAccountSettingsReadModel(userID int64) + FlushAccountSettingsReadModel() +} + +type AccountSettingsReadModelWarmer interface { + WarmAccountSettingsReadModel(context.Context, int64) error +} + // BaseUserCache 是跨进程共享的 user:base 缓存(Redis)。user_base read-model 事件必须删除 // 对应 user 键，否则 RPC 投影失效后会从陈旧的 user:base 重建(失效被未失效的源自我抵消)。 // 不参与重连 flush：Redis 是跨实例共享的，整库清空是错的，漏掉的通知靠其自身 TTL 兜底。 @@ -241,7 +251,8 @@ func (l *ReadModelChangeListener) empty() bool { l.caches.RPCProjections == nil && l.caches.BaseUsers == nil && l.caches.BotProfiles == nil && - l.caches.StarGifts == nil + l.caches.StarGifts == nil && + l.caches.AccountSettings == nil } func (l *ReadModelChangeListener) flush(reasons ...string) { @@ -318,6 +329,10 @@ func (l *ReadModelChangeListener) flush(reasons ...string) { l.caches.StarGifts.FlushStarGiftCatalog() flushed = append(flushed, "star_gifts") } + if l.caches.AccountSettings != nil { + l.caches.AccountSettings.FlushAccountSettingsReadModel() + flushed = append(flushed, "account_settings") + } // 注意：BaseUsers(Redis) 刻意不在重连时 flush——它是跨实例共享缓存，整库清空会误伤 // 其它实例；漏掉的通知由其 5min TTL 兜底。 l.log.Info("read model caches flushed", @@ -346,6 +361,19 @@ func (l *ReadModelChangeListener) handlePayload(payload string) { } } switch evt.Model { + case "account_settings": + if evt.OwnerUserID != 0 && l.caches.AccountSettings != nil { + l.caches.AccountSettings.InvalidateAccountSettingsReadModel(evt.OwnerUserID) + if warmer, ok := l.caches.AccountSettings.(AccountSettingsReadModelWarmer); ok { + ctx, cancel := context.WithTimeout(context.Background(), privacyReadModelWarmTimeout) + err := warmer.WarmAccountSettingsReadModel(ctx, evt.OwnerUserID) + cancel() + if err != nil { + l.log.Warn("warm account settings read model after change", + zap.Int64("owner_user_id", evt.OwnerUserID), zap.Error(err)) + } + } + } case "star_gift_catalog": if l.caches.StarGifts != nil { l.caches.StarGifts.InvalidateStarGiftCatalog()