From 543380138001f2975e71f1fb03eb1f07c85ce8d8 Mon Sep 17 00:00:00 2001 From: iamxvbaba <28732408+iamxvbaba@users.noreply.github.com> Date: Mon, 27 Jul 2026 22:18:31 +0800 Subject: [PATCH] fix(stargifts): sync correct channel gift notifications --- ...0_channel_star_gift_notifications.down.sql | 9 + ...150_channel_star_gift_notifications.up.sql | 87 +++++++ internal/app/stargifts/service.go | 19 ++ internal/rpc/payments_star_gifts.go | 23 +- internal/rpc/payments_star_gifts_rpc_test.go | 73 ++++++ internal/rpc/update_peer_refs.go | 50 +++- internal/rpc/update_peer_refs_test.go | 93 +++++++ internal/store/memory/star_gift.go | 4 + internal/store/postgres/star_gift.go | 55 ++++ .../star_gift_channel_notifications.go | 234 ++++++++++++++++++ .../store/postgres/star_gift_craft_auction.go | 3 +- .../store/postgres/star_gift_entitlements.go | 8 + .../store/postgres/star_gift_lifecycle.go | 21 ++ .../star_gift_lifecycle_integration_test.go | 144 ++++++++++- ...ft_lifecycle_migration_integration_test.go | 4 +- internal/store/postgres/star_gift_purchase.go | 6 + internal/store/postgres/star_gift_upgrade.go | 18 +- .../postgres/star_gift_user_message_ref.go | 63 ++++- internal/store/star_gift.go | 5 + 19 files changed, 893 insertions(+), 26 deletions(-) create mode 100644 deploy/migrations/0150_channel_star_gift_notifications.down.sql create mode 100644 deploy/migrations/0150_channel_star_gift_notifications.up.sql create mode 100644 internal/store/postgres/star_gift_channel_notifications.go diff --git a/deploy/migrations/0150_channel_star_gift_notifications.down.sql b/deploy/migrations/0150_channel_star_gift_notifications.down.sql new file mode 100644 index 00000000..49fa0090 --- /dev/null +++ b/deploy/migrations/0150_channel_star_gift_notifications.down.sql @@ -0,0 +1,9 @@ +DELETE FROM star_gift_user_message_refs refs +USING peer_star_gifts gift +WHERE gift.id = refs.saved_gift_id + AND gift.owner_peer_type = 'channel'; + +COMMENT ON TABLE star_gift_user_message_refs IS + 'Owner-local private service-message aliases to user-owned saved gift aggregates.'; + +DROP TABLE IF EXISTS star_gift_channel_notification_jobs; diff --git a/deploy/migrations/0150_channel_star_gift_notifications.up.sql b/deploy/migrations/0150_channel_star_gift_notifications.up.sql new file mode 100644 index 00000000..d40a58ca --- /dev/null +++ b/deploy/migrations/0150_channel_star_gift_notifications.up.sql @@ -0,0 +1,87 @@ +-- Channel gift purchases commit a durable recipient snapshot before any +-- private notification is attempted. Delivery is idempotent and restart-safe: +-- the deterministic private-send replay is authoritative if a process stops +-- after writing the message but before marking this job delivered. +CREATE TABLE star_gift_channel_notification_jobs ( + saved_gift_id bigint NOT NULL REFERENCES peer_star_gifts(id) ON DELETE CASCADE, + target_user_id bigint NOT NULL REFERENCES users(id) ON DELETE CASCADE, + gift_date integer NOT NULL, + action jsonb NOT NULL, + attempts integer NOT NULL DEFAULT 0, + next_attempt_at integer NOT NULL, + lease_until integer NOT NULL DEFAULT 0, + delivered_at integer NOT NULL DEFAULT 0, + message_id integer NOT NULL DEFAULT 0, + last_error text NOT NULL DEFAULT '', + created_at timestamptz NOT NULL DEFAULT now(), + updated_at timestamptz NOT NULL DEFAULT now(), + PRIMARY KEY (saved_gift_id, target_user_id), + CONSTRAINT star_gift_channel_notification_jobs_state_check CHECK ( + saved_gift_id > 0 + AND target_user_id > 0 + AND gift_date > 0 + AND attempts >= 0 + AND next_attempt_at > 0 + AND lease_until >= 0 + AND delivered_at >= 0 + AND message_id >= 0 + AND ((delivered_at = 0 AND message_id = 0) OR (delivered_at > 0 AND message_id > 0)) + ) +); + +CREATE INDEX star_gift_channel_notification_jobs_pending_idx + ON star_gift_channel_notification_jobs(next_attempt_at, lease_until, saved_gift_id, target_user_id) + WHERE delivered_at = 0; + +COMMENT ON TABLE star_gift_channel_notification_jobs IS + 'Purchase-time snapshot of per-admin channel gift notification intents; delivery uses deterministic private-message replay.'; + +COMMENT ON TABLE star_gift_user_message_refs IS + 'Viewer-local private service-message aliases to saved gift aggregates; the saved gift owner may be that user or an authorized channel.'; + +-- Existing testserver/users already have channel prepaid/upgrade notifications. +-- Backfill only aliases that are cryptographically unnecessary to guess: the +-- persisted action itself names a channel and saved_id, and that exact pair +-- must resolve to one saved gift. This changes no message, PTS or outbox row. +WITH ordinary AS ( + SELECT box.owner_user_id, + box.box_id, + gift.id AS saved_gift_id + FROM message_boxes box + JOIN peer_star_gifts gift + ON gift.owner_peer_type = 'channel' + AND (box.media #>> '{service_action,star_gift,peer_channel_id}') ~ '^[1-9][0-9]*$' + AND gift.owner_peer_id = (box.media #>> '{service_action,star_gift,peer_channel_id}')::bigint + AND (box.media #>> '{service_action,star_gift,saved_id}') ~ '^[1-9][0-9]*$' + AND gift.saved_id = (box.media #>> '{service_action,star_gift,saved_id}')::bigint + WHERE NOT box.deleted + AND box.media #>> '{service_action,kind}' = 'star_gift' + AND gift.lifecycle_status = 'active' +), collectible AS ( + SELECT box.owner_user_id, + box.box_id, + gift.id AS saved_gift_id + FROM message_boxes box + JOIN peer_star_gifts gift + ON gift.owner_peer_type = 'channel' + AND box.media #>> '{service_action,star_gift_unique,peer,Type}' = 'channel' + AND (box.media #>> '{service_action,star_gift_unique,peer,ID}') ~ '^[1-9][0-9]*$' + AND gift.owner_peer_id = (box.media #>> '{service_action,star_gift_unique,peer,ID}')::bigint + AND (box.media #>> '{service_action,star_gift_unique,saved_id}') ~ '^[1-9][0-9]*$' + AND gift.saved_id = (box.media #>> '{service_action,star_gift_unique,saved_id}')::bigint + AND (box.media #>> '{service_action,star_gift_unique,gift,ID}') ~ '^[1-9][0-9]*$' + AND gift.unique_gift_id = (box.media #>> '{service_action,star_gift_unique,gift,ID}')::bigint + WHERE NOT box.deleted + AND box.media #>> '{service_action,kind}' = 'star_gift_unique' + AND gift.lifecycle_status = 'active' +), aliases AS ( + SELECT * FROM ordinary + UNION + SELECT * FROM collectible +) +INSERT INTO star_gift_user_message_refs(owner_user_id, msg_id, saved_gift_id) +SELECT owner_user_id, box_id, saved_gift_id +FROM aliases +ON CONFLICT(owner_user_id, msg_id) DO UPDATE +SET saved_gift_id = EXCLUDED.saved_gift_id +WHERE star_gift_user_message_refs.saved_gift_id = EXCLUDED.saved_gift_id; diff --git a/internal/app/stargifts/service.go b/internal/app/stargifts/service.go index e1b01923..4db8cdb4 100644 --- a/internal/app/stargifts/service.go +++ b/internal/app/stargifts/service.go @@ -740,6 +740,25 @@ func (s *Service) SetNotifications(ctx context.Context, userID, channelID int64, return s.lifecycle.SetStarGiftNotifications(ctx, userID, channelID, enabled) } +func (s *Service) NotificationsEnabled(ctx context.Context, userID, channelID int64) (bool, error) { + if s == nil { + return false, domain.ErrStarGiftUnavailable + } + if s.lifecycle == nil { + // Isolated memory/RPC adapters have no settings table; production's + // persisted default is enabled, so preserve that wire behavior. + return true, nil + } + return s.lifecycle.StarGiftNotificationsEnabled(ctx, userID, channelID) +} + +func (s *Service) ResolveUserMessageRef(ctx context.Context, viewerUserID int64, msgID int) (domain.SavedStarGiftRef, bool, error) { + if s == nil || s.store == nil { + return domain.SavedStarGiftRef{}, false, nil + } + return s.store.ResolveUserMessageRef(ctx, viewerUserID, msgID) +} + func (s *Service) Withdraw(ctx context.Context, req domain.StarGiftWithdrawalRequest) (domain.StarGiftWithdrawal, error) { if s == nil || s.lifecycle == nil || s.withdrawal == nil { return domain.StarGiftWithdrawal{}, domain.ErrStarGiftWithdrawalUnavailable diff --git a/internal/rpc/payments_star_gifts.go b/internal/rpc/payments_star_gifts.go index fa9bdb9f..c8faae39 100644 --- a/internal/rpc/payments_star_gifts.go +++ b/internal/rpc/payments_star_gifts.go @@ -17,7 +17,8 @@ import ( ) // Star gift（payments.* 礼物 RPC）：目录 / 购买表单 / 发送 / 收礼列表 / 展示切换 / 转换回 Stars。 -// 扣费经 r.deps.Stars 账本；用户礼物走私聊服务消息，频道礼物只落 saved gifts + admin log。 +// 扣费经原子礼物聚合账本；用户礼物走私聊服务消息，频道礼物落 saved gift/admin log， +// 并由持久 notification job 向启用通知的礼物管理员投递私聊 service message。 func starGiftInvalidErr() error { return tgerr.New(400, "STARGIFT_INVALID") } @@ -560,6 +561,15 @@ func (r *Router) onPaymentsGetSavedStarGifts(ctx context.Context, req *tg.Paymen if err != nil { return nil, internalErr() } + if settings, ok := r.deps.Gifts.(interface { + NotificationsEnabled(context.Context, int64, int64) (bool, error) + }); ok && owner.Type == domain.PeerTypeChannel && r.ensureCanManageStarGiftOwner(ctx, userID, owner) == nil { + enabled, settingsErr := settings.NotificationsEnabled(ctx, userID, owner.ID) + if settingsErr != nil { + return nil, internalErr() + } + response.SetChatNotificationsEnabled(enabled) + } return response, nil } @@ -753,6 +763,17 @@ func (r *Router) starGiftRefFromInput(ctx context.Context, userID int64, ref tg. if v == nil || v.MsgID <= 0 { return domain.SavedStarGiftRef{}, false, nil } + if resolver, ok := r.deps.Gifts.(interface { + ResolveUserMessageRef(context.Context, int64, int) (domain.SavedStarGiftRef, bool, error) + }); ok { + resolved, found, err := resolver.ResolveUserMessageRef(ctx, userID, v.MsgID) + if err != nil { + return domain.SavedStarGiftRef{}, false, internalErr() + } + if found { + return resolved, resolved.Valid(), nil + } + } return domain.SavedStarGiftRef{ Owner: domain.Peer{Type: domain.PeerTypeUser, ID: userID}, MsgID: v.MsgID, diff --git a/internal/rpc/payments_star_gifts_rpc_test.go b/internal/rpc/payments_star_gifts_rpc_test.go index e62ea857..2b6047d0 100644 --- a/internal/rpc/payments_star_gifts_rpc_test.go +++ b/internal/rpc/payments_star_gifts_rpc_test.go @@ -93,6 +93,48 @@ func (s *uniqueGiftRPCService) UniqueBySlug(_ context.Context, slug string) (dom return s.unique, slug == s.unique.Slug, nil } +type starGiftMessageAliasRPCService struct { + GiftsService + viewerUserID int64 + msgID int + ref domain.SavedStarGiftRef +} + +func (s *starGiftMessageAliasRPCService) ResolveUserMessageRef(_ context.Context, viewerUserID int64, msgID int) (domain.SavedStarGiftRef, bool, error) { + if viewerUserID == s.viewerUserID && msgID == s.msgID { + return s.ref, true, nil + } + return domain.SavedStarGiftRef{}, false, nil +} + +func TestStarGiftUserInputResolvesOnlyExplicitChannelNotificationAlias(t *testing.T) { + channelRef := domain.SavedStarGiftRef{ + Owner: domain.Peer{Type: domain.PeerTypeChannel, ID: 8801}, + SavedID: 91, + } + service := &starGiftMessageAliasRPCService{viewerUserID: 7102, msgID: 44, ref: channelRef} + r := New(Config{DC: 2}, Deps{Gifts: service}, zaptest.NewLogger(t), clock.System) + + resolved, ok, err := r.starGiftRefFromInput(context.Background(), 7102, &tg.InputSavedStarGiftUser{MsgID: 44}) + if err != nil || !ok || resolved != channelRef { + t.Fatalf("channel notification alias = %+v ok=%v err=%v", resolved, ok, err) + } + fallback, ok, err := r.starGiftRefFromInput(context.Background(), 7102, &tg.InputSavedStarGiftUser{MsgID: 45}) + wantFallback := domain.SavedStarGiftRef{Owner: domain.Peer{Type: domain.PeerTypeUser, ID: 7102}, MsgID: 45} + if err != nil || !ok || fallback != wantFallback { + t.Fatalf("unknown notification alias = %+v ok=%v err=%v, want user fallback %+v", fallback, ok, err, wantFallback) + } +} + +type starGiftNotificationSettingsRPCService struct { + GiftsService + enabled bool +} + +func (s *starGiftNotificationSettingsRPCService) NotificationsEnabled(context.Context, int64, int64) (bool, error) { + return s.enabled, nil +} + type craftStarGiftRPCService struct { GiftsService uniques map[string]domain.UniqueStarGift @@ -1154,10 +1196,41 @@ func TestStarGiftChannelSaga(t *testing.T) { if !savedRes.Gifts[0].CanUpgrade { t.Fatal("channel saved gift must advertise upgrade when a collectible pool is available") } + ownerSavedRes, err := r.onPaymentsGetSavedStarGifts(ownerCtx, &tg.PaymentsGetSavedStarGiftsRequest{Peer: channelPeer}) + if err != nil { + t.Fatalf("getSavedStarGifts(channel owner): %v", err) + } + if enabled, ok := ownerSavedRes.GetChatNotificationsEnabled(); !ok || !enabled { + t.Fatalf("channel owner notification setting enabled=%v ok=%v, want default true", enabled, ok) + } + baseGifts := r.deps.Gifts + r.deps.Gifts = &starGiftNotificationSettingsRPCService{GiftsService: baseGifts, enabled: false} + disabledSavedRes, err := r.onPaymentsGetSavedStarGifts(ownerCtx, &tg.PaymentsGetSavedStarGiftsRequest{Peer: channelPeer}) + r.deps.Gifts = baseGifts + if err != nil { + t.Fatalf("getSavedStarGifts(channel notifications disabled): %v", err) + } + if enabled, ok := disabledSavedRes.GetChatNotificationsEnabled(); !ok || enabled { + t.Fatalf("disabled channel notification setting enabled=%v ok=%v, want false/present", enabled, ok) + } savedID, ok := savedRes.Gifts[0].GetSavedID() if !ok || savedID <= 0 { t.Fatalf("saved gift saved_id = %d ok %v, want positive", savedID, ok) } + baseGifts = r.deps.Gifts + r.deps.Gifts = &starGiftMessageAliasRPCService{ + GiftsService: baseGifts, + viewerUserID: sender.ID, + msgID: 777, + ref: domain.SavedStarGiftRef{Owner: domain.Peer{Type: domain.PeerTypeChannel, ID: channel.ID}, SavedID: savedID}, + } + _, aliasPermissionErr := r.onPaymentsSaveStarGift(senderCtx, &tg.PaymentsSaveStarGiftRequest{ + Stargift: &tg.InputSavedStarGiftUser{MsgID: 777}, + }) + r.deps.Gifts = baseGifts + if !tgerr.Is(aliasPermissionErr, "CHAT_ADMIN_REQUIRED") { + t.Fatalf("non-admin channel notification alias err=%v, want CHAT_ADMIN_REQUIRED", aliasPermissionErr) + } if _, ok := savedRes.Gifts[0].GetMsgID(); ok { t.Fatalf("channel saved gift should not expose inputSavedStarGiftUser.msg_id") } diff --git a/internal/rpc/update_peer_refs.go b/internal/rpc/update_peer_refs.go index 6e594e8e..b5a18a7a 100644 --- a/internal/rpc/update_peer_refs.go +++ b/internal/rpc/update_peer_refs.go @@ -234,11 +234,7 @@ func collectMessagePeerRefs(msg domain.Message, currentChannelID int64, userIDs, if msg.Media != nil && msg.Media.Contact != nil && msg.Media.Contact.UserID != 0 { userIDs[msg.Media.Contact.UserID] = struct{}{} } - if msg.Media != nil && msg.Media.ServiceAction != nil && msg.Media.ServiceAction.RequestedPeer != nil { - for _, peer := range msg.Media.ServiceAction.RequestedPeer.Peers { - addDomainPeerRef(peer, currentChannelID, userIDs, channelIDs) - } - } + collectServiceActionPeerRefs(msg.Media, currentChannelID, userIDs, channelIDs) collectPollMediaUserRefs(msg.Media, userIDs) collectTodoMediaUserRefs(msg.Media, userIDs) if msg.Reactions != nil { @@ -332,6 +328,7 @@ func collectChannelMessagePeerRefs(msg domain.ChannelMessage, currentChannelID i channelIDs[id] = struct{}{} } } + collectStarGiftUniquePeerRefs(msg.Action.StarGiftUnique, currentChannelID, userIDs, channelIDs) } if msg.Reactions != nil { for _, reaction := range msg.Reactions.Recent { @@ -342,6 +339,49 @@ func collectChannelMessagePeerRefs(msg domain.ChannelMessage, currentChannelID i } } +func collectServiceActionPeerRefs(media *domain.MessageMedia, currentChannelID int64, userIDs, channelIDs map[int64]struct{}) { + if media == nil || media.ServiceAction == nil { + return + } + action := media.ServiceAction + if action.RequestedPeer != nil { + for _, peer := range action.RequestedPeer.Peers { + addDomainPeerRef(peer, currentChannelID, userIDs, channelIDs) + } + } + if gift := action.StarGift; gift != nil { + if gift.FromUserID != 0 && !gift.NameHidden { + userIDs[gift.FromUserID] = struct{}{} + } + if gift.PeerUserID != 0 { + userIDs[gift.PeerUserID] = struct{}{} + } + if gift.PeerChannelID != 0 && gift.PeerChannelID != currentChannelID { + channelIDs[gift.PeerChannelID] = struct{}{} + } + addDomainPeerRef(gift.To, currentChannelID, userIDs, channelIDs) + } + collectStarGiftUniquePeerRefs(action.StarGiftUnique, currentChannelID, userIDs, channelIDs) +} + +func collectStarGiftUniquePeerRefs(action *domain.MessageStarGiftUniqueAction, currentChannelID int64, userIDs, channelIDs map[int64]struct{}) { + if action == nil { + return + } + if action.FromUserID != 0 { + userIDs[action.FromUserID] = struct{}{} + } + addDomainPeerRef(action.Peer, currentChannelID, userIDs, channelIDs) + addDomainPeerRef(action.Gift.Owner, currentChannelID, userIDs, channelIDs) + addDomainPeerRef(action.Gift.OriginalOwner, currentChannelID, userIDs, channelIDs) + addDomainPeerRef(action.Gift.ReleasedBy, currentChannelID, userIDs, channelIDs) + addDomainPeerRef(action.Gift.ThemePeer, currentChannelID, userIDs, channelIDs) + addDomainPeerRef(action.Gift.Host, currentChannelID, userIDs, channelIDs) + if action.Gift.OriginalFromUserID != 0 && !action.Gift.OriginalNameHidden { + userIDs[action.Gift.OriginalFromUserID] = struct{}{} + } +} + func addDomainPeerRef(peer domain.Peer, currentChannelID int64, userIDs, channelIDs map[int64]struct{}) { switch peer.Type { case domain.PeerTypeUser: diff --git a/internal/rpc/update_peer_refs_test.go b/internal/rpc/update_peer_refs_test.go index f43c44b2..a2e4891e 100644 --- a/internal/rpc/update_peer_refs_test.go +++ b/internal/rpc/update_peer_refs_test.go @@ -28,3 +28,96 @@ func TestRemoveKnownChannelRefs(t *testing.T) { } } } + +func TestCollectMessagePeerRefsIncludesStarGiftServiceActions(t *testing.T) { + users := map[int64]struct{}{} + channels := map[int64]struct{}{} + collectMessagePeerRefs(domain.Message{Media: &domain.MessageMedia{ + Kind: domain.MessageMediaKindService, + ServiceAction: &domain.MessageServiceAction{ + Kind: domain.MessageServiceActionStarGift, + StarGift: &domain.MessageStarGiftAction{ + FromUserID: 1001, PeerChannelID: 55, + }, + }, + }}, 0, users, channels) + if _, ok := users[1001]; !ok { + t.Fatalf("ordinary star-gift user refs=%v, missing sender", users) + } + if _, ok := channels[55]; !ok { + t.Fatalf("ordinary star-gift channel refs=%v", channels) + } + collectMessagePeerRefs(domain.Message{Media: &domain.MessageMedia{ + Kind: domain.MessageMediaKindService, + ServiceAction: &domain.MessageServiceAction{ + Kind: domain.MessageServiceActionStarGift, + StarGift: &domain.MessageStarGiftAction{PeerUserID: 1002}, + }, + }}, 0, users, channels) + if _, ok := users[1002]; !ok { + t.Fatalf("ordinary star-gift recipient refs=%v", users) + } + + collectMessagePeerRefs(domain.Message{Media: &domain.MessageMedia{ + Kind: domain.MessageMediaKindService, + ServiceAction: &domain.MessageServiceAction{ + Kind: domain.MessageServiceActionStarGiftUnique, + StarGiftUnique: &domain.MessageStarGiftUniqueAction{ + FromUserID: 2001, + Peer: domain.Peer{Type: domain.PeerTypeChannel, ID: 56}, + Gift: domain.UniqueStarGift{ + Owner: domain.Peer{Type: domain.PeerTypeChannel, ID: 56}, + OriginalFromUserID: 2002, + OriginalOwner: domain.Peer{Type: domain.PeerTypeUser, ID: 2003}, + ReleasedBy: domain.Peer{Type: domain.PeerTypeUser, ID: 2004}, + ThemePeer: domain.Peer{Type: domain.PeerTypeChannel, ID: 57}, + Host: domain.Peer{Type: domain.PeerTypeUser, ID: 2005}, + }, + }, + }, + }}, 0, users, channels) + for _, id := range []int64{2001, 2002, 2003, 2004, 2005} { + if _, ok := users[id]; !ok { + t.Fatalf("unique star-gift user refs=%v, missing %d", users, id) + } + } + for _, id := range []int64{56, 57} { + if _, ok := channels[id]; !ok { + t.Fatalf("unique star-gift channel refs=%v, missing %d", channels, id) + } + } +} + +func TestCollectMessagePeerRefsHidesStarGiftSenderDetails(t *testing.T) { + users := map[int64]struct{}{} + channels := map[int64]struct{}{} + collectMessagePeerRefs(domain.Message{Media: &domain.MessageMedia{ + Kind: domain.MessageMediaKindService, + ServiceAction: &domain.MessageServiceAction{ + Kind: domain.MessageServiceActionStarGift, + StarGift: &domain.MessageStarGiftAction{ + FromUserID: 1001, NameHidden: true, PeerChannelID: 55, + }, + }, + }}, 0, users, channels) + collectMessagePeerRefs(domain.Message{Media: &domain.MessageMedia{ + Kind: domain.MessageMediaKindService, + ServiceAction: &domain.MessageServiceAction{ + Kind: domain.MessageServiceActionStarGiftUnique, + StarGiftUnique: &domain.MessageStarGiftUniqueAction{ + Gift: domain.UniqueStarGift{ + OriginalFromUserID: 2001, + OriginalNameHidden: true, + }, + }, + }, + }}, 0, users, channels) + for _, id := range []int64{1001, 2001} { + if _, ok := users[id]; ok { + t.Fatalf("hidden star-gift sender %d leaked into refs=%v", id, users) + } + } + if _, ok := channels[55]; !ok { + t.Fatalf("hidden ordinary gift lost recipient channel ref=%v", channels) + } +} diff --git a/internal/store/memory/star_gift.go b/internal/store/memory/star_gift.go index 63369feb..82d97893 100644 --- a/internal/store/memory/star_gift.go +++ b/internal/store/memory/star_gift.go @@ -506,6 +506,10 @@ func (s *StarGiftStore) GetByRef(_ context.Context, ref domain.SavedStarGiftRef) return domain.SavedStarGift{}, false, nil } +func (s *StarGiftStore) ResolveUserMessageRef(_ context.Context, _ int64, _ int) (domain.SavedStarGiftRef, bool, error) { + return domain.SavedStarGiftRef{}, false, nil +} + func (s *StarGiftStore) CountByOwner(_ context.Context, owner domain.Peer) (int, error) { if !validStarGiftOwner(owner) { return 0, nil diff --git a/internal/store/postgres/star_gift.go b/internal/store/postgres/star_gift.go index 87e69b61..2e305f00 100644 --- a/internal/store/postgres/star_gift.go +++ b/internal/store/postgres/star_gift.go @@ -735,6 +735,61 @@ WHERE `+where, args...) return g, true, nil } +func (s *StarGiftStore) ResolveUserMessageRef(ctx context.Context, viewerUserID int64, msgID int) (domain.SavedStarGiftRef, bool, error) { + if s == nil || s.db == nil || viewerUserID <= 0 || msgID <= 0 { + return domain.SavedStarGiftRef{}, false, nil + } + var ownerType string + var ownerID, savedID int64 + err := s.db.QueryRow(ctx, ` +SELECT gift.owner_peer_type,gift.owner_peer_id,gift.saved_id +FROM star_gift_user_message_refs ref +JOIN peer_star_gifts gift ON gift.id=ref.saved_gift_id +JOIN message_boxes box + ON box.owner_user_id=ref.owner_user_id AND box.box_id=ref.msg_id +WHERE ref.owner_user_id=$1 AND ref.msg_id=$2 + AND NOT box.deleted + AND gift.lifecycle_status='active' + AND ( + (gift.owner_peer_type='user' AND gift.owner_peer_id=$1) + OR ( + gift.owner_peer_type='channel' + AND ( + ( + box.media #>> '{service_action,kind}'='star_gift' + AND box.media #>> '{service_action,star_gift,peer_channel_id}'=gift.owner_peer_id::text + AND box.media #>> '{service_action,star_gift,saved_id}'=gift.saved_id::text + ) + OR ( + box.media #>> '{service_action,kind}'='star_gift_unique' + AND box.media #>> '{service_action,star_gift_unique,peer,Type}'='channel' + AND box.media #>> '{service_action,star_gift_unique,peer,ID}'=gift.owner_peer_id::text + AND box.media #>> '{service_action,star_gift_unique,saved_id}'=gift.saved_id::text + ) + ) + ) + )`, + viewerUserID, msgID).Scan(&ownerType, &ownerID, &savedID) + if errors.Is(err, pgx.ErrNoRows) { + return domain.SavedStarGiftRef{}, false, nil + } + if err != nil { + return domain.SavedStarGiftRef{}, false, fmt.Errorf("resolve star gift user message ref: %w", err) + } + owner := domain.Peer{Type: domain.PeerType(ownerType), ID: ownerID} + switch owner.Type { + case domain.PeerTypeUser: + return domain.SavedStarGiftRef{Owner: owner, MsgID: msgID}, true, nil + case domain.PeerTypeChannel: + if savedID <= 0 { + return domain.SavedStarGiftRef{}, false, domain.ErrStarGiftOwnerInvalid + } + return domain.SavedStarGiftRef{Owner: owner, SavedID: savedID}, true, nil + default: + return domain.SavedStarGiftRef{}, false, domain.ErrStarGiftOwnerInvalid + } +} + func (s *StarGiftStore) CountByOwner(ctx context.Context, owner domain.Peer) (int, error) { if !validStarGiftOwner(owner) { return 0, nil diff --git a/internal/store/postgres/star_gift_channel_notifications.go b/internal/store/postgres/star_gift_channel_notifications.go new file mode 100644 index 00000000..8c94290f --- /dev/null +++ b/internal/store/postgres/star_gift_channel_notifications.go @@ -0,0 +1,234 @@ +package postgres + +import ( + "context" + "crypto/sha256" + "encoding/json" + "fmt" + "strings" + + "github.com/jackc/pgx/v5" + + "telesrv/internal/domain" +) + +const ( + maxChannelStarGiftNotificationRecipients = 256 + channelStarGiftNotificationLeaseSeconds = 60 +) + +type channelStarGiftNotificationJob struct { + SavedGiftID int64 + TargetUserID int64 + GiftDate int + Action domain.MessageStarGiftAction + Attempts int +} + +func enqueueChannelStarGiftNotifications( + ctx context.Context, + tx pgx.Tx, + savedGiftID int64, + channelID int64, + giftDate int, + action *domain.MessageStarGiftAction, +) error { + if savedGiftID <= 0 || channelID <= 0 || giftDate <= 0 || action == nil || + action.PeerChannelID != channelID || action.SavedID <= 0 { + return fmt.Errorf("enqueue channel star gift notifications: invalid intent") + } + actionJSON, err := json.Marshal(action) + if err != nil { + return fmt.Errorf("encode channel star gift notification: %w", err) + } + _, err = tx.Exec(ctx, ` +WITH candidates AS ( + SELECT creator_user_id AS user_id + FROM channels + WHERE id=$2 AND NOT deleted + UNION + SELECT user_id + FROM channel_members + WHERE channel_id=$2 AND status='active' + AND (role='creator' OR ( + role='admin' + AND COALESCE((admin_rights->>'PostMessages')::boolean,false) + )) +), bounded AS ( + SELECT user_id FROM candidates + WHERE user_id>0 + ORDER BY user_id + LIMIT $5 +) +INSERT INTO star_gift_channel_notification_jobs + (saved_gift_id,target_user_id,gift_date,action,next_attempt_at) +SELECT $1,bounded.user_id,$3,$4::jsonb,$3 +FROM bounded +LEFT JOIN star_gift_notification_settings settings + ON settings.user_id=bounded.user_id AND settings.channel_id=$2 +WHERE COALESCE(settings.enabled,TRUE) +ON CONFLICT(saved_gift_id,target_user_id) DO NOTHING`, + savedGiftID, channelID, giftDate, string(actionJSON), maxChannelStarGiftNotificationRecipients) + if err != nil { + return fmt.Errorf("enqueue channel star gift notifications: %w", err) + } + return nil +} + +func (s *StarGiftLifecycleStore) dispatchChannelStarGiftNotifications( + ctx context.Context, + now int, + limit int, + savedGiftID int64, +) (int, error) { + if s == nil || s.db == nil || s.messages == nil || now <= 0 || limit <= 0 { + return 0, domain.ErrStarGiftUnavailable + } + if limit > maxChannelStarGiftNotificationRecipients { + limit = maxChannelStarGiftNotificationRecipients + } + jobs, err := s.claimChannelStarGiftNotificationJobs(ctx, now, limit, savedGiftID) + if err != nil { + return 0, err + } + var firstErr error + delivered := 0 + for _, job := range jobs { + messageID, sendErr := s.deliverChannelStarGiftNotification(ctx, job) + if sendErr == nil { + tag, markErr := s.db.Exec(ctx, `UPDATE star_gift_channel_notification_jobs +SET delivered_at=$3,message_id=$4,lease_until=0,last_error='',updated_at=now() +WHERE saved_gift_id=$1 AND target_user_id=$2 AND delivered_at=0`, + job.SavedGiftID, job.TargetUserID, now, messageID) + if markErr == nil && tag.RowsAffected() == 1 { + delivered++ + continue + } + if markErr == nil { + markErr = fmt.Errorf("channel star gift notification job disappeared") + } + sendErr = markErr + } + if firstErr == nil { + firstErr = sendErr + } + retryAt := now + channelStarGiftNotificationRetrySeconds(job.Attempts) + _, _ = s.db.Exec(ctx, `UPDATE star_gift_channel_notification_jobs +SET next_attempt_at=$3,lease_until=0,last_error=$4,updated_at=now() +WHERE saved_gift_id=$1 AND target_user_id=$2 AND delivered_at=0`, + job.SavedGiftID, job.TargetUserID, retryAt, truncateStarGiftNotificationError(sendErr)) + } + return delivered, firstErr +} + +func (s *StarGiftLifecycleStore) claimChannelStarGiftNotificationJobs( + ctx context.Context, + now int, + limit int, + savedGiftID int64, +) ([]channelStarGiftNotificationJob, error) { + jobs := make([]channelStarGiftNotificationJob, 0, limit) + err := withTx(ctx, s.db, "claim channel star gift notifications", func(tx pgx.Tx) error { + rows, err := tx.Query(ctx, ` +WITH picked AS ( + SELECT saved_gift_id,target_user_id + FROM star_gift_channel_notification_jobs + WHERE delivered_at=0 AND next_attempt_at<=$1 AND lease_until<$1 + AND ($3::bigint=0 OR saved_gift_id=$3) + ORDER BY next_attempt_at,saved_gift_id,target_user_id + FOR UPDATE SKIP LOCKED + LIMIT $2 +) +UPDATE star_gift_channel_notification_jobs job +SET attempts=job.attempts+1, + lease_until=$1+$4, + updated_at=now() +FROM picked +WHERE job.saved_gift_id=picked.saved_gift_id + AND job.target_user_id=picked.target_user_id +RETURNING job.saved_gift_id,job.target_user_id,job.gift_date,job.action,job.attempts`, + now, limit, savedGiftID, channelStarGiftNotificationLeaseSeconds) + if err != nil { + return err + } + defer rows.Close() + for rows.Next() { + var job channelStarGiftNotificationJob + var actionJSON []byte + if err := rows.Scan(&job.SavedGiftID, &job.TargetUserID, &job.GiftDate, &actionJSON, &job.Attempts); err != nil { + return err + } + if err := json.Unmarshal(actionJSON, &job.Action); err != nil { + return fmt.Errorf("decode channel star gift notification: %w", err) + } + if job.SavedGiftID <= 0 || job.TargetUserID <= 0 || job.GiftDate <= 0 || + job.Action.PeerChannelID <= 0 || job.Action.SavedID <= 0 { + return fmt.Errorf("decode channel star gift notification: invalid intent") + } + jobs = append(jobs, job) + } + return rows.Err() + }) + return jobs, err +} + +func (s *StarGiftLifecycleStore) deliverChannelStarGiftNotification( + ctx context.Context, + job channelStarGiftNotificationJob, +) (int, error) { + fingerprint := sha256.Sum256([]byte(fmt.Sprintf( + "telesrv:channel-star-gift-notification:v1:%d:%d", + job.SavedGiftID, job.TargetUserID, + ))) + action := job.Action + request := domain.SendPrivateTextRequest{ + SenderUserID: domain.OfficialSystemUserID, + RecipientUserID: job.TargetUserID, + RandomID: lifecycleCommandRandomID("channel-star-gift-notification", job.SavedGiftID, job.TargetUserID), + Date: job.GiftDate, + IdempotencyFingerprint: fingerprint[:], + Media: &domain.MessageMedia{Kind: domain.MessageMediaKindService, ServiceAction: &domain.MessageServiceAction{ + Kind: domain.MessageServiceActionStarGift, + StarGift: &action, + }}, + } + sent, err := s.messages.sendPrivateTextWithHooks(ctx, request, privateSendTxHooks{ + after: func(ctx context.Context, tx pgx.Tx, sent domain.SendPrivateTextResult) error { + if sent.RecipientMessage.ID <= 0 { + return fmt.Errorf("channel star gift notification missing recipient box") + } + return registerChannelNotificationMessageRef(ctx, tx, job.TargetUserID, + sent.RecipientMessage.ID, job.SavedGiftID) + }, + }) + if err != nil { + return 0, err + } + if sent.RecipientMessage.ID <= 0 { + return 0, fmt.Errorf("channel star gift notification replay missing recipient box") + } + return sent.RecipientMessage.ID, nil +} + +func channelStarGiftNotificationRetrySeconds(attempt int) int { + if attempt < 1 { + attempt = 1 + } + delay := attempt * attempt * 5 + if delay > 3600 { + return 3600 + } + return delay +} + +func truncateStarGiftNotificationError(err error) string { + if err == nil { + return "" + } + value := strings.TrimSpace(err.Error()) + runes := []rune(value) + if len(runes) > 1000 { + value = string(runes[:1000]) + } + return value +} diff --git a/internal/store/postgres/star_gift_craft_auction.go b/internal/store/postgres/star_gift_craft_auction.go index e55e9779..355f12ed 100644 --- a/internal/store/postgres/star_gift_craft_auction.go +++ b/internal/store/postgres/star_gift_craft_auction.go @@ -181,7 +181,8 @@ WHERE saved_gift_id IS NULL ORDER BY gift_id LIMIT $1`, minAuctionInt(remaining, } } } - return nil + _, err := s.dispatchChannelStarGiftNotifications(ctx, now, minAuctionInt(limit, 100), 0) + return err } func (s *StarGiftLifecycleStore) ListCraftStarGifts(ctx context.Context, userID, giftID int64, offset string, limit int) (domain.SavedStarGiftPage, error) { diff --git a/internal/store/postgres/star_gift_entitlements.go b/internal/store/postgres/star_gift_entitlements.go index 3d951201..2c5f239e 100644 --- a/internal/store/postgres/star_gift_entitlements.go +++ b/internal/store/postgres/star_gift_entitlements.go @@ -147,6 +147,14 @@ VALUES($1,$2,$3,$4,$5,$6,$7)`, req.PayerUserID, req.CommandKey, locked.ID, req.F } return registerUserStarGiftMessageRef(ctx, tx, req.Owner.ID, ownerMessageID, result.Saved.ID, 0) } + notificationMessageID := sent.RecipientMessage.ID + if notificationMessageID <= 0 { + return fmt.Errorf("prepaid channel gift notification missing recipient box") + } + if err := registerViewerStarGiftMessageRef(ctx, tx, req.PayerUserID, notificationMessageID, + result.Saved.ID, req.Owner, 0); err != nil { + return err + } action := messageReq.Media.ServiceAction.StarGift return NewChannelStore(tx).appendStarGiftAdminLogTx(ctx, tx, req.Owner.ID, req.PayerUserID, result.Saved.SavedID, req.Date, domain.ChannelMessageAction{Type: domain.ChannelActionStarGift, StarGift: action}) diff --git a/internal/store/postgres/star_gift_lifecycle.go b/internal/store/postgres/star_gift_lifecycle.go index ce9f0c4e..24714488 100644 --- a/internal/store/postgres/star_gift_lifecycle.go +++ b/internal/store/postgres/star_gift_lifecycle.go @@ -598,6 +598,15 @@ func (s *StarGiftLifecycleStore) PurchaseResaleStarGift(ctx context.Context, req if err := registerUserStarGiftMessageRef(ctx, tx, req.To.ID, msgID, result.Saved.ID, result.Unique.ID); err != nil { return err } + } else { + notificationMessageID := sent.RecipientMessage.ID + if notificationMessageID <= 0 { + return fmt.Errorf("channel resale notification missing buyer box") + } + if err := registerViewerStarGiftMessageRef(ctx, tx, req.BuyerUserID, notificationMessageID, + result.Saved.ID, req.To, result.Unique.ID); err != nil { + return err + } } if _, err := tx.Exec(ctx, `INSERT INTO star_gift_sales(unique_gift_id,seller_peer_type,seller_peer_id, buyer_peer_type,buyer_peer_id,currency,amount,commission_amount,sold_at,command_key) @@ -1399,6 +1408,18 @@ ON CONFLICT(user_id,channel_id) DO UPDATE SET enabled=EXCLUDED.enabled,updated_a return err } +func (s *StarGiftLifecycleStore) StarGiftNotificationsEnabled(ctx context.Context, userID, channelID int64) (bool, error) { + if s == nil || s.db == nil || userID <= 0 || channelID <= 0 { + return false, domain.ErrStarGiftOwnerInvalid + } + var enabled bool + err := s.db.QueryRow(ctx, `SELECT COALESCE(( +SELECT enabled FROM star_gift_notification_settings +WHERE user_id=$1 AND channel_id=$2 +),TRUE)`, userID, channelID).Scan(&enabled) + return enabled, err +} + func (s *StarGiftLifecycleStore) RecordStarGiftWithdrawal(ctx context.Context, req domain.StarGiftWithdrawalRequest, provider, providerRequestID, url string, expiresAt int) (domain.StarGiftWithdrawal, error) { if req.UserID <= 0 || !req.Ref.Valid() || req.Date <= 0 || expiresAt <= req.Date || strings.TrimSpace(provider) == "" || strings.TrimSpace(providerRequestID) == "" || strings.TrimSpace(url) == "" { return domain.StarGiftWithdrawal{}, domain.ErrStarGiftWithdrawalUnavailable diff --git a/internal/store/postgres/star_gift_lifecycle_integration_test.go b/internal/store/postgres/star_gift_lifecycle_integration_test.go index 62437833..68e02cfc 100644 --- a/internal/store/postgres/star_gift_lifecycle_integration_test.go +++ b/internal/store/postgres/star_gift_lifecycle_integration_test.go @@ -811,17 +811,37 @@ func TestStarGiftChannelLifecycleAtomicPostgres(t *testing.T) { now := int(time.Now().Unix()) users := NewUserStore(pool) actor := createTestUser(t, ctx, users, "+1882"+suffix+"01", "ChannelGiftActor", "") + notifyAdmin := createTestUser(t, ctx, users, "+1882"+suffix+"02", "ChannelGiftNotifyAdmin", "") + mutedAdmin := createTestUser(t, ctx, users, "+1882"+suffix+"03", "ChannelGiftMutedAdmin", "") + noPostAdmin := createTestUser(t, ctx, users, "+1882"+suffix+"04", "ChannelGiftNoPostAdmin", "") + ordinaryMember := createTestUser(t, ctx, users, "+1882"+suffix+"05", "ChannelGiftMember", "") if _, _, err := NewStarsStore(pool).EnsureGrant(ctx, actor.ID, 10000, now); err != nil { t.Fatalf("grant actor stars: %v", err) } - created, err := NewChannelStore(pool).CreateChannel(ctx, domain.CreateChannelRequest{ + channelStore := NewChannelStore(pool) + created, err := channelStore.CreateChannel(ctx, domain.CreateChannelRequest{ CreatorUserID: actor.ID, Title: "Gift Channel " + suffix, Megagroup: true, Date: now, + MemberUserIDs: []int64{notifyAdmin.ID, mutedAdmin.ID, noPostAdmin.ID, ordinaryMember.ID}, }) if err != nil { t.Fatalf("create gift channel: %v", err) } + for _, admin := range []domain.User{notifyAdmin, mutedAdmin} { + if _, err := channelStore.EditChannelAdmin(ctx, domain.EditChannelAdminRequest{ + UserID: actor.ID, ChannelID: created.Channel.ID, MemberID: admin.ID, + AdminRights: domain.ChannelAdminRights{PostMessages: true}, Date: now, + }); err != nil { + t.Fatalf("grant channel gift PostMessages admin %d: %v", admin.ID, err) + } + } + if _, err := channelStore.EditChannelAdmin(ctx, domain.EditChannelAdminRequest{ + UserID: actor.ID, ChannelID: created.Channel.ID, MemberID: noPostAdmin.ID, + AdminRights: domain.ChannelAdminRights{ChangeInfo: true}, Date: now, + }); err != nil { + t.Fatalf("grant non-posting channel admin: %v", err) + } channelPeer := domain.Peer{Type: domain.PeerTypeChannel, ID: created.Channel.ID} - createdTarget, err := NewChannelStore(pool).CreateChannel(ctx, domain.CreateChannelRequest{ + createdTarget, err := channelStore.CreateChannel(ctx, domain.CreateChannelRequest{ CreatorUserID: actor.ID, Title: "Gift Target Channel " + suffix, Megagroup: true, Date: now, }) if err != nil { @@ -867,9 +887,16 @@ func TestStarGiftChannelLifecycleAtomicPostgres(t *testing.T) { lifecycle := NewStarGiftLifecycleStore(pool, messages, 1_000_000, WithStarGiftMarketPolicy(domain.StarGiftMarketPolicy{ StarsProceedsPermille: 900, TONProceedsPermille: 900, })) + if err := lifecycle.SetStarGiftNotifications(ctx, mutedAdmin.ID, created.Channel.ID, false); err != nil { + t.Fatalf("disable muted admin channel gift notifications: %v", err) + } upgrades := NewStarGiftUpgradeStore(pool, messages, WithStarGiftLifecyclePolicy(domain.StarGiftLifecyclePolicy{ TransferStars: 25, DropOriginalDetailsStars: 25, OfferMinStars: 1, CraftChancePermille: 500, })) + var channelPtsBeforePurchase int + if err := pool.QueryRow(ctx, `SELECT pts FROM channels WHERE id=$1`, created.Channel.ID).Scan(&channelPtsBeforePurchase); err != nil { + t.Fatalf("load channel pts before gift purchase: %v", err) + } channelPurchaseReq := issueLifecyclePurchaseForm(t, ctx, lifecycle, domain.StarGiftPurchaseRequest{BuyerUserID: actor.ID, To: channelPeer, GiftID: entry.Gift.ID, CommandKey: "channel-purchase-" + suffix, Date: now + 1}) purchased, err := lifecycle.PurchaseStarGift(ctx, channelPurchaseReq) @@ -881,6 +908,107 @@ func TestStarGiftChannelLifecycleAtomicPostgres(t *testing.T) { WHERE channel_id=$1 AND event_type='send_message' AND message::text LIKE '%star_gift%'`, created.Channel.ID).Scan(®ularLogs); err != nil || regularLogs != 1 { t.Fatalf("channel purchase admin logs = %d err %v", regularLogs, err) } + var notificationRecipients []int64 + rows, err := pool.Query(ctx, `SELECT target_user_id FROM star_gift_channel_notification_jobs +WHERE saved_gift_id=$1 ORDER BY target_user_id`, purchased.Saved.ID) + if err != nil { + t.Fatalf("list channel gift notification recipients: %v", err) + } + for rows.Next() { + var userID int64 + if err := rows.Scan(&userID); err != nil { + rows.Close() + t.Fatalf("scan channel gift notification recipient: %v", err) + } + notificationRecipients = append(notificationRecipients, userID) + } + if err := rows.Err(); err != nil { + rows.Close() + t.Fatalf("iterate channel gift notification recipients: %v", err) + } + rows.Close() + wantRecipients := []int64{actor.ID, notifyAdmin.ID} + if fmt.Sprint(notificationRecipients) != fmt.Sprint(wantRecipients) { + t.Fatalf("channel gift notification recipients=%v want=%v (muted/no-post/member excluded)", + notificationRecipients, wantRecipients) + } + var notificationMessageID, notificationDeliveredAt, notificationAttempts int + if err := pool.QueryRow(ctx, `SELECT message_id,delivered_at,attempts +FROM star_gift_channel_notification_jobs +WHERE saved_gift_id=$1 AND target_user_id=$2`, purchased.Saved.ID, actor.ID). + Scan(¬ificationMessageID, ¬ificationDeliveredAt, ¬ificationAttempts); err != nil || + notificationMessageID <= 0 || notificationDeliveredAt <= 0 || notificationAttempts != 1 { + t.Fatalf("channel gift notification job message=%d delivered=%d attempts=%d err=%v", + notificationMessageID, notificationDeliveredAt, notificationAttempts, err) + } + notificationRef, found, err := gifts.ResolveUserMessageRef(ctx, actor.ID, notificationMessageID) + if err != nil || !found || notificationRef.Owner != channelPeer || + notificationRef.SavedID != purchased.Saved.SavedID { + t.Fatalf("channel gift notification alias = %+v found=%v err=%v", notificationRef, found, err) + } + var notificationPts, notificationEvents, notificationOutbox, channelPtsAfterPurchase int + var notificationMediaJSON string + if err := pool.QueryRow(ctx, `SELECT pts,media::text FROM message_boxes +WHERE owner_user_id=$1 AND box_id=$2`, actor.ID, notificationMessageID). + Scan(¬ificationPts, ¬ificationMediaJSON); err != nil { + t.Fatalf("load channel gift notification message: %v", err) + } + notificationMedia, err := decodeMessageMedia(notificationMediaJSON) + if err != nil || notificationMedia == nil || notificationMedia.ServiceAction == nil || + notificationMedia.ServiceAction.StarGift == nil || + notificationMedia.ServiceAction.StarGift.PeerChannelID != created.Channel.ID || + notificationMedia.ServiceAction.StarGift.SavedID != purchased.Saved.SavedID { + t.Fatalf("channel gift notification media = %+v err=%v", notificationMedia, err) + } + if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM user_update_events +WHERE user_id=$1 AND pts=$2 AND event_type='new_message'`, actor.ID, notificationPts).Scan(¬ificationEvents); err != nil || + notificationEvents != 1 { + t.Fatalf("channel gift notification events=%d err=%v", notificationEvents, err) + } + if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM dispatch_outbox +WHERE target_user_id=$1 AND pts=$2`, actor.ID, notificationPts).Scan(¬ificationOutbox); err != nil || + notificationOutbox != 1 { + t.Fatalf("channel gift notification outbox=%d err=%v", notificationOutbox, err) + } + if err := pool.QueryRow(ctx, `SELECT pts FROM channels WHERE id=$1`, created.Channel.ID).Scan(&channelPtsAfterPurchase); err != nil || + channelPtsAfterPurchase != channelPtsBeforePurchase { + t.Fatalf("channel gift notification changed channel pts: before=%d after=%d err=%v", + channelPtsBeforePurchase, channelPtsAfterPurchase, err) + } + // Simulate a process stopping after the private message committed but before + // the job completion update. The next claim must replay the same message and + // must not allocate a second account PTS/event/outbox row. + if _, err := pool.Exec(ctx, `UPDATE star_gift_channel_notification_jobs +SET delivered_at=0,message_id=0,next_attempt_at=$3,lease_until=0 +WHERE saved_gift_id=$1 AND target_user_id=$2`, purchased.Saved.ID, actor.ID, now+1); err != nil { + t.Fatalf("reset notification job for replay probe: %v", err) + } + if delivered, err := lifecycle.dispatchChannelStarGiftNotifications(ctx, now+2, 1, purchased.Saved.ID); err != nil || delivered != 1 { + t.Fatalf("replay channel gift notification delivered=%d err=%v", delivered, err) + } + var replayMessageID, replayEventCount int + if err := pool.QueryRow(ctx, `SELECT message_id FROM star_gift_channel_notification_jobs +WHERE saved_gift_id=$1 AND target_user_id=$2`, purchased.Saved.ID, actor.ID).Scan(&replayMessageID); err != nil || + replayMessageID != notificationMessageID { + t.Fatalf("channel gift notification replay message=%d want=%d err=%v", replayMessageID, notificationMessageID, err) + } + if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM user_update_events +WHERE user_id=$1 AND pts=$2 AND event_type='new_message'`, actor.ID, notificationPts).Scan(&replayEventCount); err != nil || + replayEventCount != 1 { + t.Fatalf("channel gift notification replay events=%d err=%v", replayEventCount, err) + } + if enabled, err := lifecycle.StarGiftNotificationsEnabled(ctx, actor.ID, created.Channel.ID); err != nil || !enabled { + t.Fatalf("default channel gift notification setting enabled=%v err=%v", enabled, err) + } + if err := lifecycle.SetStarGiftNotifications(ctx, actor.ID, created.Channel.ID, false); err != nil { + t.Fatalf("disable channel gift notifications: %v", err) + } + if enabled, err := lifecycle.StarGiftNotificationsEnabled(ctx, actor.ID, created.Channel.ID); err != nil || enabled { + t.Fatalf("disabled channel gift notification setting enabled=%v err=%v", enabled, err) + } + if err := lifecycle.SetStarGiftNotifications(ctx, actor.ID, created.Channel.ID, true); err != nil { + t.Fatalf("re-enable channel gift notifications: %v", err) + } var channelPrice string var channelPrepaidAmount any if err := pool.QueryRow(ctx, `SELECT message #>> '{Action,StarGift,upgrade_price_stars}', message #> '{Action,StarGift,upgrade_stars}' @@ -949,6 +1077,11 @@ WHERE channel_id=$1 AND event_type='send_message' AND message::text LIKE '%star_ channelPrepay.Send.RecipientMessage.OwnerUserID != actor.ID { t.Fatalf("channel prepaid entitlement = %+v err %v", channelPrepay, err) } + prepayAlias, found, err := gifts.ResolveUserMessageRef(ctx, actor.ID, channelPrepay.Send.RecipientMessage.ID) + if err != nil || !found || prepayAlias.Owner != channelPeer || + prepayAlias.SavedID != channelPrepay.Saved.SavedID { + t.Fatalf("channel prepaid notification alias = %+v found=%v err=%v", prepayAlias, found, err) + } var prepayLogs int if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM channel_admin_log_events WHERE channel_id=$1 AND message::text LIKE '%prepaid_upgrade%'`, created.Channel.ID).Scan(&prepayLogs); err != nil || prepayLogs != 2 { @@ -983,12 +1116,17 @@ WHERE channel_id=$1 AND message::text LIKE '%prepaid_upgrade%'`, created.Channel t.Fatalf("channel prepaid upgrade = %+v err %v", upgraded, err) } action := upgraded.Send.RecipientMessage.Media.ServiceAction.StarGiftUnique - if action == nil || action.FromUserID != domain.OfficialSystemUserID || action.Peer != channelPeer || + if action == nil || action.FromUserID != actor.ID || action.Peer != channelPeer || action.SavedID != prepaidPurchase.Saved.SavedID || !action.Upgrade || !action.PrepaidUpgrade || action.TransferStars != 25 || action.CanCraftAt != 0 || action.Gift.CraftChancePermille != 500 || upgraded.Saved.CanCraftAt != now+5 || upgraded.Unique.CraftChancePermille != 500 { t.Fatalf("channel upgrade service action = %+v", action) } + upgradeAlias, found, err := gifts.ResolveUserMessageRef(ctx, actor.ID, upgraded.Send.RecipientMessage.ID) + if err != nil || !found || upgradeAlias.Owner != channelPeer || + upgradeAlias.SavedID != upgraded.Saved.SavedID { + t.Fatalf("channel upgrade notification alias = %+v found=%v err=%v", upgradeAlias, found, err) + } var ptsAfterUpgrade int if err := pool.QueryRow(ctx, `SELECT pts FROM channels WHERE id=$1`, created.Channel.ID).Scan(&ptsAfterUpgrade); err != nil || ptsAfterUpgrade != ptsBeforeUpgrade { t.Fatalf("channel pts after profile gift upgrade = %d want %d err %v", ptsAfterUpgrade, ptsBeforeUpgrade, err) diff --git a/internal/store/postgres/star_gift_lifecycle_migration_integration_test.go b/internal/store/postgres/star_gift_lifecycle_migration_integration_test.go index 0114de3b..b87b34ce 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 != 149 { - t.Fatalf("migration status = %+v, want clean version 149", status) + if status.Dirty || status.Empty || status.Version != 150 { + t.Fatalf("migration status = %+v, want clean version 150", status) } } diff --git a/internal/store/postgres/star_gift_purchase.go b/internal/store/postgres/star_gift_purchase.go index 60d3e4c4..b3da2d65 100644 --- a/internal/store/postgres/star_gift_purchase.go +++ b/internal/store/postgres/star_gift_purchase.go @@ -188,6 +188,9 @@ func (s *StarGiftLifecycleStore) purchaseStarGiftToChannel(ctx context.Context, if err := NewChannelStore(tx).appendStarGiftAdminLogTx(ctx, tx, req.To.ID, req.BuyerUserID, id, req.Date, action); err != nil { return err } + if err := enqueueChannelStarGiftNotifications(ctx, tx, id, req.To.ID, req.Date, action.StarGift); err != nil { + return err + } if err := s.insertStarGiftPurchaseCommand(ctx, tx, req, id, gift.Stars+saved.PrepaidUpgradeStars, balance.Balance); err != nil { return err } @@ -202,6 +205,9 @@ func (s *StarGiftLifecycleStore) purchaseStarGiftToChannel(ctx context.Context, } return domain.StarGiftPurchaseResult{}, err } + // The purchase remains successful once its transaction has committed. Any + // immediate delivery failure leaves a durable job for the lifecycle sweeper. + _, _ = s.dispatchChannelStarGiftNotifications(ctx, req.Date, maxChannelStarGiftNotificationRecipients, result.Saved.ID) return result, nil } diff --git a/internal/store/postgres/star_gift_upgrade.go b/internal/store/postgres/star_gift_upgrade.go index 1d5f1c92..ec6503be 100644 --- a/internal/store/postgres/star_gift_upgrade.go +++ b/internal/store/postgres/star_gift_upgrade.go @@ -518,7 +518,7 @@ VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9)`, req.UserID, commandKey, locked.ID, req.For locked.CanCraftAt = canCraftAt locked.Unique = &unique result.Saved, result.Unique, result.Balance = locked, unique, balance - action := starGiftUpgradeUniqueAction(locked, unique, req, messageSenderID) + action := starGiftUpgradeUniqueAction(locked, unique, req) messageReq.Media = &domain.MessageMedia{ Kind: domain.MessageMediaKindService, ServiceAction: &domain.MessageServiceAction{ @@ -530,7 +530,7 @@ VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9)`, req.UserID, commandKey, locked.ID, req.For }, after: func(ctx context.Context, tx pgx.Tx, sent domain.SendPrivateTextResult) error { ownerMessageID := sent.RecipientMessage.ID - if saved.FromUserID == req.UserID { + if result.Saved.Owner.Type == domain.PeerTypeUser && saved.FromUserID == req.UserID { ownerMessageID = sent.SenderMessage.ID } if ownerMessageID <= 0 { @@ -548,6 +548,9 @@ VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9)`, req.UserID, commandKey, locked.ID, req.For result.Saved.ID, result.Unique.ID); err != nil { return err } + } else if err := registerViewerStarGiftMessageRef(ctx, tx, req.UserID, ownerMessageID, + result.Saved.ID, result.Saved.Owner, result.Unique.ID); err != nil { + return err } result.Saved.UpgradeMsgID = ownerMessageID if result.Saved.Owner.Type == domain.PeerTypeUser { @@ -576,7 +579,7 @@ WHERE user_id=$1 AND command_key=$2`, req.UserID, commandKey, ownerEditPts) return fmt.Errorf("save star gift source edit pts lost command row") } } else { - action := starGiftUpgradeUniqueAction(result.Saved, result.Unique, req, messageSenderID) + action := starGiftUpgradeUniqueAction(result.Saved, result.Unique, req) if err := NewChannelStore(tx).appendStarGiftAdminLogTx(ctx, tx, result.Saved.Owner.ID, req.UserID, result.Saved.SavedID, req.Date, domain.ChannelMessageAction{ Type: domain.ChannelActionStarGiftUnique, StarGiftUnique: action, @@ -599,15 +602,16 @@ WHERE user_id=$1 AND command_key=$2`, req.UserID, commandKey, ownerEditPts) return result, nil } -func starGiftUpgradeUniqueAction(saved domain.SavedStarGift, unique domain.UniqueStarGift, req domain.StarGiftUpgradeRequest, messageSenderID int64) *domain.MessageStarGiftUniqueAction { +func starGiftUpgradeUniqueAction(saved domain.SavedStarGift, unique domain.UniqueStarGift, req domain.StarGiftUpgradeRequest) *domain.MessageStarGiftUniqueAction { fromUserID := saved.FromUserID if saved.NameHidden { fromUserID = 0 } if saved.Owner.Type == domain.PeerTypeChannel { - // TDesktop recognizes a channel-owned upgrade from the official service - // peer plus action.peer=channel and action.saved_id. - fromUserID = messageSenderID + // The private envelope is sent by 777000, while action.from_id identifies + // the administrator who performed the upgrade. TDesktop uses that + // distinction to render "upgraded" instead of an unknown transfer. + fromUserID = req.UserID } peer := saved.Owner savedID := saved.SavedID diff --git a/internal/store/postgres/star_gift_user_message_ref.go b/internal/store/postgres/star_gift_user_message_ref.go index e69898ec..2b1aa448 100644 --- a/internal/store/postgres/star_gift_user_message_ref.go +++ b/internal/store/postgres/star_gift_user_message_ref.go @@ -5,6 +5,8 @@ import ( "fmt" "github.com/jackc/pgx/v5" + + "telesrv/internal/domain" ) // registerUserStarGiftMessageRef records an owner-scoped service-message alias @@ -22,24 +24,71 @@ func registerUserStarGiftMessageRef( savedGiftID int64, uniqueGiftID int64, ) error { - if ownerUserID <= 0 || msgID <= 0 || savedGiftID <= 0 || uniqueGiftID < 0 { - return fmt.Errorf("register user star gift message ref: invalid identity") + return registerViewerStarGiftMessageRef(ctx, tx, ownerUserID, msgID, savedGiftID, + domain.Peer{Type: domain.PeerTypeUser, ID: ownerUserID}, uniqueGiftID) +} + +// registerViewerStarGiftMessageRef binds one viewer-local private message to +// the aggregate owner explicitly named by the action. The alias does not grant +// ownership: RPC callers resolve the real owner and authorize it again. +func registerViewerStarGiftMessageRef( + ctx context.Context, + tx pgx.Tx, + viewerUserID int64, + msgID int, + savedGiftID int64, + expectedOwner domain.Peer, + uniqueGiftID int64, +) error { + if viewerUserID <= 0 || msgID <= 0 || savedGiftID <= 0 || !validLifecyclePeer(expectedOwner) || uniqueGiftID < 0 { + return fmt.Errorf("register star gift message ref: invalid identity") } tag, err := tx.Exec(ctx, ` INSERT INTO star_gift_user_message_refs(owner_user_id,msg_id,saved_gift_id) SELECT $1,$2,p.id FROM peer_star_gifts p -WHERE p.id=$3 AND p.owner_peer_type='user' AND p.owner_peer_id=$1 - AND (($4::bigint=0 AND p.unique_gift_id IS NULL) OR ($4::bigint>0 AND p.unique_gift_id=$4::bigint)) +WHERE p.id=$3 AND p.owner_peer_type=$4 AND p.owner_peer_id=$5 + AND (($6::bigint=0 AND p.unique_gift_id IS NULL) OR ($6::bigint>0 AND p.unique_gift_id=$6::bigint)) AND p.lifecycle_status='active' ON CONFLICT(owner_user_id,msg_id) DO UPDATE SET saved_gift_id=EXCLUDED.saved_gift_id -WHERE star_gift_user_message_refs.saved_gift_id=EXCLUDED.saved_gift_id`, ownerUserID, msgID, savedGiftID, uniqueGiftID) +WHERE star_gift_user_message_refs.saved_gift_id=EXCLUDED.saved_gift_id`, + viewerUserID, msgID, savedGiftID, string(expectedOwner.Type), expectedOwner.ID, uniqueGiftID) if err != nil { - return fmt.Errorf("register user star gift message ref: %w", err) + return fmt.Errorf("register star gift message ref: %w", err) } if tag.RowsAffected() != 1 { - return fmt.Errorf("register user star gift message ref: identity collision") + return fmt.Errorf("register star gift message ref: identity collision") + } + return nil +} + +func registerChannelNotificationMessageRef( + ctx context.Context, + tx pgx.Tx, + viewerUserID int64, + msgID int, + savedGiftID int64, +) error { + if viewerUserID <= 0 || msgID <= 0 || savedGiftID <= 0 { + return fmt.Errorf("register channel notification star gift message ref: invalid identity") + } + tag, err := tx.Exec(ctx, ` +INSERT INTO star_gift_user_message_refs(owner_user_id,msg_id,saved_gift_id) +SELECT $1,$2,gift.id +FROM star_gift_channel_notification_jobs job +JOIN peer_star_gifts gift ON gift.id=job.saved_gift_id +WHERE job.saved_gift_id=$3 AND job.target_user_id=$1 + AND gift.owner_peer_type='channel' AND gift.lifecycle_status='active' +ON CONFLICT(owner_user_id,msg_id) DO UPDATE +SET saved_gift_id=EXCLUDED.saved_gift_id +WHERE star_gift_user_message_refs.saved_gift_id=EXCLUDED.saved_gift_id`, + viewerUserID, msgID, savedGiftID) + if err != nil { + return fmt.Errorf("register channel notification star gift message ref: %w", err) + } + if tag.RowsAffected() != 1 { + return fmt.Errorf("register channel notification star gift message ref: identity collision") } return nil } diff --git a/internal/store/star_gift.go b/internal/store/star_gift.go index 4d97b740..916b45b2 100644 --- a/internal/store/star_gift.go +++ b/internal/store/star_gift.go @@ -42,6 +42,10 @@ type StarGiftStore interface { ListByOwnerFiltered(ctx context.Context, filter domain.SavedStarGiftFilter) (domain.SavedStarGiftPage, error) // GetByRef 按协议引用取礼物实例：用户用 msg_id，频道用 saved_id。 GetByRef(ctx context.Context, ref domain.SavedStarGiftRef) (domain.SavedStarGift, bool, error) + // ResolveUserMessageRef resolves only an explicit viewer-local service-message + // alias. The returned ref retains the saved gift's authoritative user/channel + // owner; callers must still authorize that owner. + ResolveUserMessageRef(ctx context.Context, viewerUserID int64, msgID int) (domain.SavedStarGiftRef, bool, error) // ResolveSavedIDs resolves an ordered batch of protocol references without // per-gift round trips. Every ref must belong to owner and resolve to a live gift. ResolveSavedIDs(ctx context.Context, owner domain.Peer, refs []domain.SavedStarGiftRef) ([]int64, error) @@ -95,6 +99,7 @@ type StarGiftLifecycleStore interface { PrepayStarGiftUpgrade(ctx context.Context, req domain.StarGiftPrepaidUpgradeRequest) (domain.StarGiftPrepaidUpgradeResult, error) DropStarGiftOriginalDetails(ctx context.Context, req domain.StarGiftDropOriginalDetailsRequest) (domain.StarGiftDropOriginalDetailsResult, error) SetStarGiftNotifications(ctx context.Context, userID, channelID int64, enabled bool) error + StarGiftNotificationsEnabled(ctx context.Context, userID, channelID int64) (bool, error) RecordStarGiftWithdrawal(ctx context.Context, req domain.StarGiftWithdrawalRequest, provider, providerRequestID, url string, expiresAt int) (domain.StarGiftWithdrawal, error) ResolveStarGiftWithdrawal(ctx context.Context, providerRequestID string) (domain.StarGiftWithdrawal, bool, error) CompleteStarGiftWithdrawal(ctx context.Context, providerRequestID string, date int) (domain.StarGiftWithdrawal, error)