diff --git a/deploy/migrations/0164_star_transaction_direction_indexes.down.sql b/deploy/migrations/0164_star_transaction_direction_indexes.down.sql new file mode 100644 index 00000000..3a1fcffc --- /dev/null +++ b/deploy/migrations/0164_star_transaction_direction_indexes.down.sql @@ -0,0 +1,8 @@ +DROP INDEX IF EXISTS public.channel_ton_transactions_outgoing_idx; +DROP INDEX IF EXISTS public.channel_ton_transactions_incoming_idx; +DROP INDEX IF EXISTS public.channel_stars_transactions_outgoing_idx; +DROP INDEX IF EXISTS public.channel_stars_transactions_incoming_idx; +DROP INDEX IF EXISTS public.ton_transactions_user_outgoing_idx; +DROP INDEX IF EXISTS public.ton_transactions_user_incoming_idx; +DROP INDEX IF EXISTS public.stars_transactions_user_outgoing_idx; +DROP INDEX IF EXISTS public.stars_transactions_user_incoming_idx; diff --git a/deploy/migrations/0164_star_transaction_direction_indexes.up.sql b/deploy/migrations/0164_star_transaction_direction_indexes.up.sql new file mode 100644 index 00000000..402edc39 --- /dev/null +++ b/deploy/migrations/0164_star_transaction_direction_indexes.up.sql @@ -0,0 +1,22 @@ +-- payments.getStarsTransactions applies the sign predicate before keyset LIMIT. +-- Partial owner/id indexes keep sparse inbound/outbound histories bounded even +-- when one account or channel has a long run of transactions in the other direction. +CREATE INDEX stars_transactions_user_incoming_idx + ON public.stars_transactions(user_id, id DESC) WHERE amount > 0; +CREATE INDEX stars_transactions_user_outgoing_idx + ON public.stars_transactions(user_id, id DESC) WHERE amount < 0; + +CREATE INDEX ton_transactions_user_incoming_idx + ON public.ton_transactions(user_id, id DESC) WHERE amount_nanoton > 0; +CREATE INDEX ton_transactions_user_outgoing_idx + ON public.ton_transactions(user_id, id DESC) WHERE amount_nanoton < 0; + +CREATE INDEX channel_stars_transactions_incoming_idx + ON public.channel_stars_transactions(channel_id, id DESC) WHERE amount > 0; +CREATE INDEX channel_stars_transactions_outgoing_idx + ON public.channel_stars_transactions(channel_id, id DESC) WHERE amount < 0; + +CREATE INDEX channel_ton_transactions_incoming_idx + ON public.channel_ton_transactions(channel_id, id DESC) WHERE amount_nanoton > 0; +CREATE INDEX channel_ton_transactions_outgoing_idx + ON public.channel_ton_transactions(channel_id, id DESC) WHERE amount_nanoton < 0; diff --git a/internal/app/stargifts/service.go b/internal/app/stargifts/service.go index 9a20e100..6df89537 100644 --- a/internal/app/stargifts/service.go +++ b/internal/app/stargifts/service.go @@ -825,17 +825,15 @@ func (s *Service) TonBalance(ctx context.Context, userID int64) (int64, error) { return s.lifecycle.TonBalance(ctx, userID) } -func (s *Service) TonTransactions(ctx context.Context, userID int64, offset string, limit int) (domain.TonTransactionPage, error) { +func (s *Service) TonTransactions(ctx context.Context, userID int64, query domain.StarsTransactionQuery) (domain.TonTransactionPage, error) { if s == nil || s.lifecycle == nil { return domain.TonTransactionPage{}, nil } - if len(offset) > domain.MaxStarsTransactionsOffsetBytes { - offset = "" + query, err := domain.NormalizeStarsTransactionQuery(query) + if err != nil { + return domain.TonTransactionPage{}, err } - if limit <= 0 || limit > domain.MaxStarsTransactionsLimit { - limit = domain.MaxStarsTransactionsLimit - } - return s.lifecycle.TonTransactions(ctx, userID, offset, limit) + return s.lifecycle.TonTransactions(ctx, userID, query) } func (s *Service) ChannelStarsBalance(ctx context.Context, channelID int64) (int64, error) { @@ -845,17 +843,15 @@ func (s *Service) ChannelStarsBalance(ctx context.Context, channelID int64) (int return s.lifecycle.ChannelStarsBalance(ctx, channelID) } -func (s *Service) ChannelStarsTransactions(ctx context.Context, channelID int64, offset string, limit int) (domain.StarsTransactionPage, error) { +func (s *Service) ChannelStarsTransactions(ctx context.Context, channelID int64, query domain.StarsTransactionQuery) (domain.StarsTransactionPage, error) { if s == nil || s.lifecycle == nil { return domain.StarsTransactionPage{}, nil } - if len(offset) > domain.MaxStarsTransactionsOffsetBytes { - offset = "" + query, err := domain.NormalizeStarsTransactionQuery(query) + if err != nil { + return domain.StarsTransactionPage{}, err } - if limit <= 0 || limit > domain.MaxStarsTransactionsLimit { - limit = domain.MaxStarsTransactionsLimit - } - return s.lifecycle.ChannelStarsTransactions(ctx, channelID, offset, limit) + return s.lifecycle.ChannelStarsTransactions(ctx, channelID, query) } func (s *Service) ChannelTonBalance(ctx context.Context, channelID int64) (int64, error) { @@ -865,17 +861,15 @@ func (s *Service) ChannelTonBalance(ctx context.Context, channelID int64) (int64 return s.lifecycle.ChannelTonBalance(ctx, channelID) } -func (s *Service) ChannelTonTransactions(ctx context.Context, channelID int64, offset string, limit int) (domain.TonTransactionPage, error) { +func (s *Service) ChannelTonTransactions(ctx context.Context, channelID int64, query domain.StarsTransactionQuery) (domain.TonTransactionPage, error) { if s == nil || s.lifecycle == nil { return domain.TonTransactionPage{}, nil } - if len(offset) > domain.MaxStarsTransactionsOffsetBytes { - offset = "" + query, err := domain.NormalizeStarsTransactionQuery(query) + if err != nil { + return domain.TonTransactionPage{}, err } - if limit <= 0 || limit > domain.MaxStarsTransactionsLimit { - limit = domain.MaxStarsTransactionsLimit - } - return s.lifecycle.ChannelTonTransactions(ctx, channelID, offset, limit) + return s.lifecycle.ChannelTonTransactions(ctx, channelID, query) } func (s *Service) SweepLifecycle(ctx context.Context, now, limit int) error { diff --git a/internal/app/stars/service.go b/internal/app/stars/service.go index fc35df63..5aa5686c 100644 --- a/internal/app/stars/service.go +++ b/internal/app/stars/service.go @@ -78,16 +78,14 @@ func (s *Service) Debit(ctx context.Context, userID, amount int64, reason domain return s.store.Debit(ctx, userID, amount, reason, peer, int(s.now().Unix()), title, desc) } -// ListTransactions 按 keyset 分页返回流水 + 当前余额，首读时惰性授予。 -func (s *Service) ListTransactions(ctx context.Context, userID int64, offset string, limit int) (domain.StarsTransactionPage, error) { - if len(offset) > domain.MaxStarsTransactionsOffsetBytes { - offset = "" - } - if limit <= 0 || limit > domain.MaxStarsTransactionsLimit { - limit = domain.MaxStarsTransactionsLimit +// ListTransactions 按方向与顺序做 keyset 分页，首读时惰性授予。 +func (s *Service) ListTransactions(ctx context.Context, userID int64, query domain.StarsTransactionQuery) (domain.StarsTransactionPage, error) { + query, err := domain.NormalizeStarsTransactionQuery(query) + if err != nil { + return domain.StarsTransactionPage{}, err } if _, err := s.ensureGranted(ctx, userID); err != nil { return domain.StarsTransactionPage{}, err } - return s.store.ListTransactions(ctx, userID, offset, limit) + return s.store.ListTransactions(ctx, userID, query) } diff --git a/internal/app/stars/service_test.go b/internal/app/stars/service_test.go index f8df4e03..04caa845 100644 --- a/internal/app/stars/service_test.go +++ b/internal/app/stars/service_test.go @@ -33,7 +33,7 @@ func TestStartingGrantOnce(t *testing.T) { t.Fatalf("second balance = %d, want 1000 (no double grant)", bal2.Balance) } // 流水里应恰有一条 grant。 - page, err := svc.ListTransactions(ctx, 7, "", 100) + page, err := svc.ListTransactions(ctx, 7, domain.StarsTransactionQuery{Limit: 100}) if err != nil { t.Fatalf("ListTransactions: %v", err) } @@ -106,7 +106,7 @@ func TestListTransactionsPagination(t *testing.T) { t.Fatalf("Credit#%d: %v", i, err) } } - page1, err := svc.ListTransactions(ctx, 7, "", 2) + page1, err := svc.ListTransactions(ctx, 7, domain.StarsTransactionQuery{Limit: 2}) if err != nil { t.Fatalf("page1: %v", err) } @@ -117,14 +117,14 @@ func TestListTransactionsPagination(t *testing.T) { if page1.Transactions[0].Amount != 14 { t.Fatalf("page1[0].Amount = %d, want 14 (newest first)", page1.Transactions[0].Amount) } - page2, err := svc.ListTransactions(ctx, 7, page1.NextOffset, 2) + page2, err := svc.ListTransactions(ctx, 7, domain.StarsTransactionQuery{Offset: page1.NextOffset, Limit: 2}) if err != nil { t.Fatalf("page2: %v", err) } if len(page2.Transactions) != 2 { t.Fatalf("page2 = %d txns, want 2", len(page2.Transactions)) } - page3, err := svc.ListTransactions(ctx, 7, page2.NextOffset, 2) + page3, err := svc.ListTransactions(ctx, 7, domain.StarsTransactionQuery{Offset: page2.NextOffset, Limit: 2}) if err != nil { t.Fatalf("page3: %v", err) } @@ -135,3 +135,75 @@ func TestListTransactionsPagination(t *testing.T) { t.Fatalf("last page NextOffset = %q, want empty (no infinite paging)", page3.NextOffset) } } + +func TestListTransactionsDirectionAndAscending(t *testing.T) { + svc := newTestService(0) + ctx := context.Background() + if _, err := svc.Credit(ctx, 7, 100, domain.StarsReasonTopup, domain.Peer{}, "", ""); err != nil { + t.Fatalf("credit 100: %v", err) + } + if _, err := svc.Debit(ctx, 7, 40, domain.StarsReasonGift, domain.Peer{}, "", ""); err != nil { + t.Fatalf("debit 40: %v", err) + } + if _, err := svc.Credit(ctx, 7, 20, domain.StarsReasonGift, domain.Peer{}, "", ""); err != nil { + t.Fatalf("credit 20: %v", err) + } + if _, err := svc.Debit(ctx, 7, 10, domain.StarsReasonReaction, domain.Peer{}, "", ""); err != nil { + t.Fatalf("debit 10: %v", err) + } + + all, err := svc.ListTransactions(ctx, 7, domain.StarsTransactionQuery{Limit: 10}) + if err != nil { + t.Fatalf("all transactions: %v", err) + } + assertStarsAmounts(t, all.Transactions, []int64{-10, 20, -40, 100}) + if all.Balance != 70 { + t.Fatalf("all balance = %d, want 70", all.Balance) + } + + incoming1, err := svc.ListTransactions(ctx, 7, domain.StarsTransactionQuery{ + Limit: 1, Direction: domain.StarsTransactionDirectionIncoming, + }) + if err != nil { + t.Fatalf("incoming page1: %v", err) + } + assertStarsAmounts(t, incoming1.Transactions, []int64{20}) + if incoming1.NextOffset == "" { + t.Fatal("incoming page1 missing next offset") + } + incoming2, err := svc.ListTransactions(ctx, 7, domain.StarsTransactionQuery{ + Offset: incoming1.NextOffset, Limit: 1, Direction: domain.StarsTransactionDirectionIncoming, + }) + if err != nil { + t.Fatalf("incoming page2: %v", err) + } + assertStarsAmounts(t, incoming2.Transactions, []int64{100}) + if incoming2.NextOffset != "" { + t.Fatalf("terminal incoming next offset = %q", incoming2.NextOffset) + } + + outgoing, err := svc.ListTransactions(ctx, 7, domain.StarsTransactionQuery{ + Limit: 10, Direction: domain.StarsTransactionDirectionOutgoing, Ascending: true, + }) + if err != nil { + t.Fatalf("ascending outgoing: %v", err) + } + assertStarsAmounts(t, outgoing.Transactions, []int64{-40, -10}) + + _, err = svc.ListTransactions(ctx, 7, domain.StarsTransactionQuery{Direction: 99}) + if !errors.Is(err, domain.ErrStarsTransactionQueryInvalid) { + t.Fatalf("invalid direction error = %v", err) + } +} + +func assertStarsAmounts(t *testing.T, transactions []domain.StarsTransaction, want []int64) { + t.Helper() + if len(transactions) != len(want) { + t.Fatalf("transaction count = %d, want %d: %+v", len(transactions), len(want), transactions) + } + for i, amount := range want { + if transactions[i].Amount != amount { + t.Fatalf("transaction[%d].amount = %d, want %d", i, transactions[i].Amount, amount) + } + } +} diff --git a/internal/domain/stars.go b/internal/domain/stars.go index 0d2f45f3..2fbdb19e 100644 --- a/internal/domain/stars.go +++ b/internal/domain/stars.go @@ -53,6 +53,57 @@ type StarsTransaction struct { // IsCredit 报告该流水是否为入账（贷记），投影到 tg.StarsTransaction.Refund。 func (t StarsTransaction) IsCredit() bool { return t.Amount > 0 } +// StarsTransactionDirection scopes one payments.getStarsTransactions view. +// The zero value intentionally means the combined inbound/outbound history. +type StarsTransactionDirection uint8 + +const ( + StarsTransactionDirectionAll StarsTransactionDirection = iota + StarsTransactionDirectionIncoming + StarsTransactionDirectionOutgoing +) + +func (d StarsTransactionDirection) Valid() bool { + return d <= StarsTransactionDirectionOutgoing +} + +func (d StarsTransactionDirection) IncludesAmount(amount int64) bool { + switch d { + case StarsTransactionDirectionAll: + return true + case StarsTransactionDirectionIncoming: + return amount > 0 + case StarsTransactionDirectionOutgoing: + return amount < 0 + default: + return false + } +} + +// StarsTransactionQuery keeps direction, ordering and the opaque keyset cursor +// together so filtering is applied before LIMIT in every ledger backend. +type StarsTransactionQuery struct { + Offset string + Limit int + Direction StarsTransactionDirection + Ascending bool +} + +// NormalizeStarsTransactionQuery preserves the existing bounded limit/offset +// behavior while rejecting impossible internal direction values. +func NormalizeStarsTransactionQuery(query StarsTransactionQuery) (StarsTransactionQuery, error) { + if !query.Direction.Valid() { + return StarsTransactionQuery{}, ErrStarsTransactionQueryInvalid + } + if len(query.Offset) > MaxStarsTransactionsOffsetBytes { + query.Offset = "" + } + if query.Limit <= 0 || query.Limit > MaxStarsTransactionsLimit { + query.Limit = MaxStarsTransactionsLimit + } + return query, nil +} + // StarsTransactionPage 是一页账本流水 + 当前余额 + 分页游标 + 对手方用户富化集合。 type StarsTransactionPage struct { Balance int64 @@ -99,6 +150,8 @@ var ( ErrStarsInsufficient = errors.New("stars: insufficient balance") // ErrStarsInvalidAmount 表示金额非法（<=0）。 ErrStarsInvalidAmount = errors.New("stars: invalid amount") + // ErrStarsTransactionQueryInvalid 表示内部构造了不可能的流水方向。 + ErrStarsTransactionQueryInvalid = errors.New("stars: invalid transaction query") ) // StarsPaymentRequiredError reports the minimum paid-message authorization the diff --git a/internal/rpc/deps.go b/internal/rpc/deps.go index 032a2581..6cda9794 100644 --- a/internal/rpc/deps.go +++ b/internal/rpc/deps.go @@ -1159,7 +1159,7 @@ type GiftsService interface { SetNotifications(ctx context.Context, userID, channelID int64, enabled bool) error Withdraw(ctx context.Context, req domain.StarGiftWithdrawalRequest) (domain.StarGiftWithdrawal, error) TonBalance(ctx context.Context, userID int64) (int64, error) - TonTransactions(ctx context.Context, userID int64, offset string, limit int) (domain.TonTransactionPage, error) + TonTransactions(ctx context.Context, userID int64, query domain.StarsTransactionQuery) (domain.TonTransactionPage, error) IssuePurchaseForm(ctx context.Context, form domain.StarGiftPurchaseForm) (domain.StarGiftPurchaseForm, error) ValidatePurchaseForm(ctx context.Context, req domain.StarGiftPurchaseRequest) error Purchase(ctx context.Context, req domain.StarGiftPurchaseRequest) (domain.StarGiftPurchaseResult, error) @@ -1172,7 +1172,7 @@ type StarsService interface { GetBalance(ctx context.Context, userID int64) (domain.StarsBalance, error) Credit(ctx context.Context, userID, amount int64, reason domain.StarsTransactionReason, peer domain.Peer, title, desc string) (domain.StarsBalance, error) Debit(ctx context.Context, userID, amount int64, reason domain.StarsTransactionReason, peer domain.Peer, title, desc string) (domain.StarsBalance, error) - ListTransactions(ctx context.Context, userID int64, offset string, limit int) (domain.StarsTransactionPage, error) + ListTransactions(ctx context.Context, userID int64, query domain.StarsTransactionQuery) (domain.StarsTransactionPage, error) } // SecretChatService 抽象私聊端对端加密（Secret Chat）握手状态机（app/secretchat）。 diff --git a/internal/rpc/errors.go b/internal/rpc/errors.go index 194b58ae..a552faeb 100644 --- a/internal/rpc/errors.go +++ b/internal/rpc/errors.go @@ -152,6 +152,8 @@ func balanceTooLowErr() error { return tgerr.New(400, "BALANCE_TOO_LOW") } func starsAmountInvalidErr() error { return tgerr.New(400, "STARS_AMOUNT_INVALID") } +func subscriptionIDInvalidErr() error { return tgerr.New(400, "SUBSCRIPTION_ID_INVALID") } + func starsFormAmountMismatchErr() error { return tgerr.New(406, "STARS_FORM_AMOUNT_MISMATCH") } func formIDEmptyErr() error { return tgerr.New(400, "FORM_ID_EMPTY") } diff --git a/internal/rpc/payments.go b/internal/rpc/payments.go index 56c3ff2b..f0e93de4 100644 --- a/internal/rpc/payments.go +++ b/internal/rpc/payments.go @@ -203,9 +203,9 @@ func (r *Router) onPaymentsGetStarsRevenueStats(ctx context.Context, req *tg.Pay type channelGiftLedgerReader interface { ChannelStarsBalance(ctx context.Context, channelID int64) (int64, error) - ChannelStarsTransactions(ctx context.Context, channelID int64, offset string, limit int) (domain.StarsTransactionPage, error) + ChannelStarsTransactions(ctx context.Context, channelID int64, query domain.StarsTransactionQuery) (domain.StarsTransactionPage, error) ChannelTonBalance(ctx context.Context, channelID int64) (int64, error) - ChannelTonTransactions(ctx context.Context, channelID int64, offset string, limit int) (domain.TonTransactionPage, error) + ChannelTonTransactions(ctx context.Context, channelID int64, query domain.StarsTransactionQuery) (domain.TonTransactionPage, error) } // onPaymentsGetStarsStatus 返回请求 peer 的 Stars/本地 TON 余额。个人与频道账本 @@ -270,12 +270,9 @@ func (r *Router) onPaymentsGetStarsTransactions(ctx context.Context, req *tg.Pay if err != nil { return nil, err } - offset, limit := "", domain.MaxStarsTransactionsLimit - if req != nil { - offset = req.Offset - if req.Limit > 0 { - limit = req.Limit - } + query, err := starsTransactionQuery(req) + if err != nil { + return nil, err } ton := req != nil && req.GetTon() if owner.Type == domain.PeerTypeChannel { @@ -287,7 +284,7 @@ func (r *Router) onPaymentsGetStarsTransactions(ctx context.Context, req *tg.Pay return emptyStarsStatus(&tg.StarsAmount{}), nil } if ton { - page, err := ledger.ChannelTonTransactions(ctx, owner.ID, offset, limit) + page, err := ledger.ChannelTonTransactions(ctx, owner.ID, query) if err != nil { return nil, internalErr() } @@ -301,7 +298,7 @@ func (r *Router) onPaymentsGetStarsTransactions(ctx context.Context, req *tg.Pay r.enrichChannelTonLedgerStatus(ctx, userID, owner.ID, page.Transactions, out) return out, nil } - page, err := ledger.ChannelStarsTransactions(ctx, owner.ID, offset, limit) + page, err := ledger.ChannelStarsTransactions(ctx, owner.ID, query) if err != nil { return nil, internalErr() } @@ -319,7 +316,7 @@ func (r *Router) onPaymentsGetStarsTransactions(ctx context.Context, req *tg.Pay if r.deps.Gifts == nil { return emptyStarsStatus(&tg.StarsTonAmount{}), nil } - page, err := r.deps.Gifts.TonTransactions(ctx, userID, offset, limit) + page, err := r.deps.Gifts.TonTransactions(ctx, userID, query) if err != nil { return nil, internalErr() } @@ -342,7 +339,7 @@ func (r *Router) onPaymentsGetStarsTransactions(ctx context.Context, req *tg.Pay if r.deps.Stars == nil { return emptyStarsStatus(&tg.StarsAmount{}), nil } - page, err := r.deps.Stars.ListTransactions(ctx, userID, offset, limit) + page, err := r.deps.Stars.ListTransactions(ctx, userID, query) if err != nil { return nil, starsErr(err) } @@ -360,6 +357,37 @@ func (r *Router) onPaymentsGetStarsTransactions(ctx context.Context, req *tg.Pay return out, nil } +func starsTransactionQuery(req *tg.PaymentsGetStarsTransactionsRequest) (domain.StarsTransactionQuery, error) { + if req == nil { + return domain.StarsTransactionQuery{}, inputRequestInvalidErr() + } + inbound, outbound := req.GetInbound(), req.GetOutbound() + if inbound && outbound { + return domain.StarsTransactionQuery{}, inputRequestInvalidErr() + } + if _, ok := req.GetSubscriptionID(); ok { + // Stars subscriptions are not part of the current business model. Do not + // silently return the unfiltered ledger for a requested subscription. + return domain.StarsTransactionQuery{}, subscriptionIDInvalidErr() + } + direction := domain.StarsTransactionDirectionAll + if inbound { + direction = domain.StarsTransactionDirectionIncoming + } else if outbound { + direction = domain.StarsTransactionDirectionOutgoing + } + limit := req.Limit + if limit <= 0 || limit > domain.MaxStarsTransactionsLimit { + limit = domain.MaxStarsTransactionsLimit + } + return domain.StarsTransactionQuery{ + Offset: req.Offset, + Limit: limit, + Direction: direction, + Ascending: req.GetAscending(), + }, nil +} + func (r *Router) starGiftLedgerOwner(ctx context.Context, req *tg.PaymentsGetStarsStatusRequest) (int64, domain.Peer, error) { if req == nil { return 0, domain.Peer{}, peerIDInvalidErr() diff --git a/internal/rpc/payments_star_gifts_rpc_test.go b/internal/rpc/payments_star_gifts_rpc_test.go index e5e1c1d5..0c4d7692 100644 --- a/internal/rpc/payments_star_gifts_rpc_test.go +++ b/internal/rpc/payments_star_gifts_rpc_test.go @@ -1598,7 +1598,7 @@ func TestStarsTopupInvoiceFallbackCreditsBalance(t *testing.T) { if bal, _ := r.deps.Stars.GetBalance(ctx, sender.ID); bal.Balance != 3500 { t.Fatalf("balance after topup = %d, want 3500", bal.Balance) } - page, err := r.deps.Stars.ListTransactions(ctx, sender.ID, "", 10) + page, err := r.deps.Stars.ListTransactions(ctx, sender.ID, domain.StarsTransactionQuery{Limit: 10}) if err != nil { t.Fatalf("list transactions: %v", err) } diff --git a/internal/rpc/payments_stars_rpc_test.go b/internal/rpc/payments_stars_rpc_test.go index 76702ca1..62e1c01a 100644 --- a/internal/rpc/payments_stars_rpc_test.go +++ b/internal/rpc/payments_stars_rpc_test.go @@ -85,6 +85,77 @@ func TestOnPaymentsGetStarsTransactions(t *testing.T) { } } +func TestOnPaymentsGetStarsTransactionsDirections(t *testing.T) { + const userID int64 = 1000000001 + svc := appstars.NewService(memory.NewStarsStore(), appstars.WithStartingGrant(0)) + ctx := WithUserID(context.Background(), userID) + if _, err := svc.Credit(ctx, userID, 100, domain.StarsReasonTopup, domain.Peer{}, "", ""); err != nil { + t.Fatalf("credit 100: %v", err) + } + if _, err := svc.Debit(ctx, userID, 40, domain.StarsReasonGift, domain.Peer{}, "", ""); err != nil { + t.Fatalf("debit 40: %v", err) + } + if _, err := svc.Credit(ctx, userID, 20, domain.StarsReasonGift, domain.Peer{}, "", ""); err != nil { + t.Fatalf("credit 20: %v", err) + } + if _, err := svc.Debit(ctx, userID, 10, domain.StarsReasonReaction, domain.Peer{}, "", ""); err != nil { + t.Fatalf("debit 10: %v", err) + } + r := New(Config{}, Deps{Stars: svc}, zaptest.NewLogger(t), clock.System) + + all := &tg.PaymentsGetStarsTransactionsRequest{Peer: &tg.InputPeerSelf{}, Limit: 50} + assertRPCStarsAmounts(t, r, ctx, all, []int64{-10, 20, -40, 100}) + + incoming := &tg.PaymentsGetStarsTransactionsRequest{Peer: &tg.InputPeerSelf{}, Limit: 50} + incoming.SetInbound(true) + assertRPCStarsAmounts(t, r, ctx, incoming, []int64{20, 100}) + + outgoing := &tg.PaymentsGetStarsTransactionsRequest{Peer: &tg.InputPeerSelf{}, Limit: 50} + outgoing.SetOutbound(true) + assertRPCStarsAmounts(t, r, ctx, outgoing, []int64{-10, -40}) + + ascending := &tg.PaymentsGetStarsTransactionsRequest{Peer: &tg.InputPeerSelf{}, Limit: 50} + ascending.SetInbound(true) + ascending.SetAscending(true) + assertRPCStarsAmounts(t, r, ctx, ascending, []int64{100, 20}) +} + +func TestOnPaymentsGetStarsTransactionsRejectsInvalidFilters(t *testing.T) { + r := starsRouter(t, 1000) + ctx := WithUserID(context.Background(), 1000000001) + + both := &tg.PaymentsGetStarsTransactionsRequest{Peer: &tg.InputPeerSelf{}} + both.SetInbound(true) + both.SetOutbound(true) + if _, err := r.onPaymentsGetStarsTransactions(ctx, both); err == nil { + t.Fatal("mutually exclusive inbound/outbound unexpectedly succeeded") + } + + subscription := &tg.PaymentsGetStarsTransactionsRequest{Peer: &tg.InputPeerSelf{}} + subscription.SetSubscriptionID("subscription-1") + if _, err := r.onPaymentsGetStarsTransactions(ctx, subscription); err == nil { + t.Fatal("unsupported subscription filter unexpectedly returned the unfiltered ledger") + } +} + +func assertRPCStarsAmounts(t *testing.T, r *Router, ctx context.Context, req *tg.PaymentsGetStarsTransactionsRequest, want []int64) { + t.Helper() + status, err := r.onPaymentsGetStarsTransactions(ctx, req) + if err != nil { + t.Fatalf("getStarsTransactions: %v", err) + } + history, _ := status.GetHistory() + if len(history) != len(want) { + t.Fatalf("history count = %d, want %d: %+v", len(history), len(want), history) + } + for i, amount := range want { + stars, ok := history[i].Amount.(*tg.StarsAmount) + if !ok || stars.Amount != amount { + t.Fatalf("history[%d].amount = %#v, want %d", i, history[i].Amount, amount) + } + } +} + func TestTGStarsTransactionsPaidMessage(t *testing.T) { out := tgStarsTransactions([]domain.StarsTransaction{{ ID: 1, UserID: 42, Peer: domain.Peer{Type: domain.PeerTypeChannel, ID: 50}, @@ -127,7 +198,7 @@ func (s *channelLedgerGifts) ChannelStarsBalance(context.Context, int64) (int64, return s.starsBalance, nil } -func (s *channelLedgerGifts) ChannelStarsTransactions(context.Context, int64, string, int) (domain.StarsTransactionPage, error) { +func (s *channelLedgerGifts) ChannelStarsTransactions(context.Context, int64, domain.StarsTransactionQuery) (domain.StarsTransactionPage, error) { return s.starsPage, nil } @@ -135,7 +206,7 @@ func (s *channelLedgerGifts) ChannelTonBalance(context.Context, int64) (int64, e return s.tonBalance, nil } -func (s *channelLedgerGifts) ChannelTonTransactions(context.Context, int64, string, int) (domain.TonTransactionPage, error) { +func (s *channelLedgerGifts) ChannelTonTransactions(context.Context, int64, domain.StarsTransactionQuery) (domain.TonTransactionPage, error) { return s.tonPage, nil } diff --git a/internal/store/memory/stars.go b/internal/store/memory/stars.go index 2cb001de..441b9cb8 100644 --- a/internal/store/memory/stars.go +++ b/internal/store/memory/stars.go @@ -93,12 +93,13 @@ func (s *StarsStore) Debit(_ context.Context, userID, amount int64, reason domai return domain.StarsBalance{UserID: userID, Balance: st.balance, Granted: st.granted}, nil } -func (s *StarsStore) ListTransactions(_ context.Context, userID int64, offset string, limit int) (domain.StarsTransactionPage, error) { +func (s *StarsStore) ListTransactions(_ context.Context, userID int64, query domain.StarsTransactionQuery) (domain.StarsTransactionPage, error) { if userID == 0 { return domain.StarsTransactionPage{}, nil } - if limit <= 0 || limit > domain.MaxStarsTransactionsLimit { - limit = domain.MaxStarsTransactionsLimit + query, err := domain.NormalizeStarsTransactionQuery(query) + if err != nil { + return domain.StarsTransactionPage{}, err } s.mu.Lock() defer s.mu.Unlock() @@ -107,22 +108,35 @@ func (s *StarsStore) ListTransactions(_ context.Context, userID int64, offset st return domain.StarsTransactionPage{}, nil } page := domain.StarsTransactionPage{Balance: st.balance} - cursor, hasCursor := domain.DecodeStarsCursor(offset) - // 倒序遍历（id DESC）。 - out := make([]domain.StarsTransaction, 0, limit) - for i := len(st.txns) - 1; i >= 0; i-- { - t := st.txns[i] - if hasCursor && t.ID >= cursor { - continue + cursor, hasCursor := domain.DecodeStarsCursor(query.Offset) + out := make([]domain.StarsTransaction, 0, query.Limit+1) + appendMatch := func(t domain.StarsTransaction) bool { + if hasCursor { + if query.Ascending && t.ID <= cursor { + return false + } + if !query.Ascending && t.ID >= cursor { + return false + } + } + if !query.Direction.IncludesAmount(t.Amount) { + return false } out = append(out, t) - if len(out) == limit { - // 还有更早的流水则给出下一页游标。 - if i-1 >= 0 { - page.NextOffset = domain.EncodeStarsCursor(t.ID) - } - break + return len(out) > query.Limit + } + if query.Ascending { + for i := 0; i < len(st.txns) && len(out) <= query.Limit; i++ { + appendMatch(st.txns[i]) } + } else { + for i := len(st.txns) - 1; i >= 0 && len(out) <= query.Limit; i-- { + appendMatch(st.txns[i]) + } + } + if len(out) > query.Limit { + out = out[:query.Limit] + page.NextOffset = domain.EncodeStarsCursor(out[len(out)-1].ID) } page.Transactions = out return page, nil diff --git a/internal/store/postgres/star_gift_lifecycle.go b/internal/store/postgres/star_gift_lifecycle.go index 24714488..bd9981a0 100644 --- a/internal/store/postgres/star_gift_lifecycle.go +++ b/internal/store/postgres/star_gift_lifecycle.go @@ -1611,27 +1611,25 @@ VALUES($1,$2,$3,$4)`, userID, s.tonStartingGrant, string(domain.StarsReasonGrant return balance, nil } -func (s *StarGiftLifecycleStore) TonTransactions(ctx context.Context, userID int64, offset string, limit int) (domain.TonTransactionPage, error) { - if userID <= 0 || limit <= 0 || limit > domain.MaxStarsTransactionsLimit || len(offset) > domain.MaxStarsTransactionsOffsetBytes { +func (s *StarGiftLifecycleStore) TonTransactions(ctx context.Context, userID int64, query domain.StarsTransactionQuery) (domain.TonTransactionPage, error) { + if userID <= 0 { return domain.TonTransactionPage{}, domain.ErrStarGiftOwnerInvalid } + query, err := domain.NormalizeStarsTransactionQuery(query) + if err != nil { + return domain.TonTransactionPage{}, err + } if _, err := s.TonBalance(ctx, userID); err != nil { return domain.TonTransactionPage{}, err } - cursor, hasCursor := domain.DecodeStarsCursor(offset) - args := []any{userID, limit + 1} - where := "user_id=$1" - if hasCursor { - where += " AND id<$3" - args = append(args, cursor) - } + where, order, args := starsTransactionQueryParts("user_id", "amount_nanoton", userID, query) rows, err := s.db.Query(ctx, `SELECT id,user_id,COALESCE(peer_type,''),COALESCE(peer_id,0),COALESCE(gift_id,0), -amount_nanoton,date,reason FROM ton_transactions WHERE `+where+` ORDER BY id DESC LIMIT $2`, args...) +amount_nanoton,date,reason FROM ton_transactions WHERE `+where+` ORDER BY id `+order+` LIMIT $2`, args...) if err != nil { return domain.TonTransactionPage{}, err } defer rows.Close() - items := make([]domain.TonTransaction, 0, limit+1) + items := make([]domain.TonTransaction, 0, query.Limit+1) for rows.Next() { var item domain.TonTransaction var peerType string @@ -1645,8 +1643,8 @@ amount_nanoton,date,reason FROM ton_transactions WHERE `+where+` ORDER BY id DES return domain.TonTransactionPage{}, err } page := domain.TonTransactionPage{} - if len(items) > limit { - items = items[:limit] + if len(items) > query.Limit { + items = items[:query.Limit] page.NextOffset = domain.EncodeStarsCursor(items[len(items)-1].ID) } page.Transactions = items @@ -1668,24 +1666,22 @@ func (s *StarGiftLifecycleStore) ChannelStarsBalance(ctx context.Context, channe return balance, err } -func (s *StarGiftLifecycleStore) ChannelStarsTransactions(ctx context.Context, channelID int64, offset string, limit int) (domain.StarsTransactionPage, error) { - if channelID <= 0 || limit <= 0 || limit > domain.MaxStarsTransactionsLimit || len(offset) > domain.MaxStarsTransactionsOffsetBytes { +func (s *StarGiftLifecycleStore) ChannelStarsTransactions(ctx context.Context, channelID int64, query domain.StarsTransactionQuery) (domain.StarsTransactionPage, error) { + if channelID <= 0 { return domain.StarsTransactionPage{}, domain.ErrStarGiftOwnerInvalid } - cursor, hasCursor := domain.DecodeStarsCursor(offset) - args := []any{channelID, limit + 1} - where := "channel_id=$1" - if hasCursor { - where += " AND id<$3" - args = append(args, cursor) + query, err := domain.NormalizeStarsTransactionQuery(query) + if err != nil { + return domain.StarsTransactionPage{}, err } + where, order, args := starsTransactionQueryParts("channel_id", "amount", channelID, query) rows, err := s.db.Query(ctx, `SELECT id,COALESCE(peer_type,''),COALESCE(peer_id,0),amount,date,reason -FROM channel_stars_transactions WHERE `+where+` ORDER BY id DESC LIMIT $2`, args...) +FROM channel_stars_transactions WHERE `+where+` ORDER BY id `+order+` LIMIT $2`, args...) if err != nil { return domain.StarsTransactionPage{}, err } defer rows.Close() - items := make([]domain.StarsTransaction, 0, limit+1) + items := make([]domain.StarsTransaction, 0, query.Limit+1) for rows.Next() { var item domain.StarsTransaction var peerType string @@ -1699,8 +1695,8 @@ FROM channel_stars_transactions WHERE `+where+` ORDER BY id DESC LIMIT $2`, args return domain.StarsTransactionPage{}, err } page := domain.StarsTransactionPage{} - if len(items) > limit { - items = items[:limit] + if len(items) > query.Limit { + items = items[:query.Limit] page.NextOffset = domain.EncodeStarsCursor(items[len(items)-1].ID) } page.Transactions = items @@ -1717,24 +1713,22 @@ func (s *StarGiftLifecycleStore) ChannelTonBalance(ctx context.Context, channelI return balance, err } -func (s *StarGiftLifecycleStore) ChannelTonTransactions(ctx context.Context, channelID int64, offset string, limit int) (domain.TonTransactionPage, error) { - if channelID <= 0 || limit <= 0 || limit > domain.MaxStarsTransactionsLimit || len(offset) > domain.MaxStarsTransactionsOffsetBytes { +func (s *StarGiftLifecycleStore) ChannelTonTransactions(ctx context.Context, channelID int64, query domain.StarsTransactionQuery) (domain.TonTransactionPage, error) { + if channelID <= 0 { return domain.TonTransactionPage{}, domain.ErrStarGiftOwnerInvalid } - cursor, hasCursor := domain.DecodeStarsCursor(offset) - args := []any{channelID, limit + 1} - where := "channel_id=$1" - if hasCursor { - where += " AND id<$3" - args = append(args, cursor) + query, err := domain.NormalizeStarsTransactionQuery(query) + if err != nil { + return domain.TonTransactionPage{}, err } + where, order, args := starsTransactionQueryParts("channel_id", "amount_nanoton", channelID, query) rows, err := s.db.Query(ctx, `SELECT id,COALESCE(peer_type,''),COALESCE(peer_id,0),COALESCE(gift_id,0),amount_nanoton,date,reason -FROM channel_ton_transactions WHERE `+where+` ORDER BY id DESC LIMIT $2`, args...) +FROM channel_ton_transactions WHERE `+where+` ORDER BY id `+order+` LIMIT $2`, args...) if err != nil { return domain.TonTransactionPage{}, err } defer rows.Close() - items := make([]domain.TonTransaction, 0, limit+1) + items := make([]domain.TonTransaction, 0, query.Limit+1) for rows.Next() { var item domain.TonTransaction var peerType string @@ -1748,8 +1742,8 @@ FROM channel_ton_transactions WHERE `+where+` ORDER BY id DESC LIMIT $2`, args.. return domain.TonTransactionPage{}, err } page := domain.TonTransactionPage{} - if len(items) > limit { - items = items[:limit] + if len(items) > query.Limit { + items = items[:query.Limit] page.NextOffset = domain.EncodeStarsCursor(items[len(items)-1].ID) } page.Transactions = items diff --git a/internal/store/postgres/star_gift_lifecycle_integration_test.go b/internal/store/postgres/star_gift_lifecycle_integration_test.go index 66921fe6..0f358526 100644 --- a/internal/store/postgres/star_gift_lifecycle_integration_test.go +++ b/internal/store/postgres/star_gift_lifecycle_integration_test.go @@ -426,7 +426,7 @@ WHERE b.owner_user_id=$1 AND b.box_id=$2`, owner.ID, upgraded.Send.RecipientMess Scan(&resaleCommission); err != nil || resaleCommission != 100 { t.Fatalf("TON resale commission = %d err %v", resaleCommission, err) } - tonPage, err := lifecycle.TonTransactions(ctx, resaleBuyer.ID, "", 20) + tonPage, err := lifecycle.TonTransactions(ctx, resaleBuyer.ID, domain.StarsTransactionQuery{Limit: 20}) if err != nil || tonPage.Balance != 999000 || len(tonPage.Transactions) < 2 { t.Fatalf("TON ledger page = %+v err %v", tonPage, err) } @@ -1050,7 +1050,7 @@ WHERE channel_id=$1 AND event_type='send_message' AND message::text LIKE '%star_ if balance, err := lifecycle.ChannelStarsBalance(ctx, created.Channel.ID); err != nil || balance != 20 { t.Fatalf("channel stars balance projection = %d err %v", balance, err) } - starsPage, err := lifecycle.ChannelStarsTransactions(ctx, created.Channel.ID, "", 20) + starsPage, err := lifecycle.ChannelStarsTransactions(ctx, created.Channel.ID, domain.StarsTransactionQuery{Limit: 20}) if err != nil || starsPage.Balance != 20 || len(starsPage.Transactions) != 1 || starsPage.Transactions[0].Amount != 20 || starsPage.Transactions[0].Reason != domain.StarsReasonGift { t.Fatalf("channel stars transaction projection = %+v err %v", starsPage, err) @@ -1188,7 +1188,7 @@ WHERE channel_id=$1 AND message::text LIKE '%star_gift_unique%'`, created.Channe if balance, err := lifecycle.ChannelTonBalance(ctx, created.Channel.ID); err != nil || balance != 900 { t.Fatalf("channel ton balance projection = %d err %v", balance, err) } - tonPage, err := lifecycle.ChannelTonTransactions(ctx, created.Channel.ID, "", 20) + tonPage, err := lifecycle.ChannelTonTransactions(ctx, created.Channel.ID, domain.StarsTransactionQuery{Limit: 20}) if err != nil || tonPage.Balance != 900 || len(tonPage.Transactions) != 1 || tonPage.Transactions[0].Amount != 900 || tonPage.Transactions[0].Reason != domain.StarsReasonGiftResale { t.Fatalf("channel ton transaction projection = %+v err %v", tonPage, err) diff --git a/internal/store/postgres/star_transaction_query_integration_test.go b/internal/store/postgres/star_transaction_query_integration_test.go new file mode 100644 index 00000000..11dafba5 --- /dev/null +++ b/internal/store/postgres/star_transaction_query_integration_test.go @@ -0,0 +1,117 @@ +package postgres + +import ( + "context" + "testing" + "time" + + "telesrv/internal/domain" +) + +func TestStarGiftLedgerTransactionDirectionsPostgres(t *testing.T) { + pool := testPool(t) + ctx := context.Background() + ownerID := (time.Now().UnixNano() & 0x1fffffffffffffff) + 3_000_000_000 + channelID := ownerID + 1 + lifecycle := NewStarGiftLifecycleStore(pool, nil, 0) + + t.Cleanup(func() { + _, _ = pool.Exec(ctx, `DELETE FROM ton_transactions WHERE user_id=$1`, ownerID) + _, _ = pool.Exec(ctx, `DELETE FROM ton_balances WHERE user_id=$1`, ownerID) + _, _ = pool.Exec(ctx, `DELETE FROM channel_stars_transactions WHERE channel_id=$1`, channelID) + _, _ = pool.Exec(ctx, `DELETE FROM channel_stars_balances WHERE channel_id=$1`, channelID) + _, _ = pool.Exec(ctx, `DELETE FROM channel_ton_transactions WHERE channel_id=$1`, channelID) + _, _ = pool.Exec(ctx, `DELETE FROM channel_ton_balances WHERE channel_id=$1`, channelID) + }) + + if _, err := pool.Exec(ctx, `INSERT INTO ton_balances(user_id,balance_nanoton,granted) VALUES($1,70,true)`, ownerID); err != nil { + t.Fatalf("insert ton balance: %v", err) + } + if _, err := pool.Exec(ctx, `INSERT INTO channel_stars_balances(channel_id,balance) VALUES($1,70)`, channelID); err != nil { + t.Fatalf("insert channel stars balance: %v", err) + } + if _, err := pool.Exec(ctx, `INSERT INTO channel_ton_balances(channel_id,balance_nanoton) VALUES($1,70)`, channelID); err != nil { + t.Fatalf("insert channel ton balance: %v", err) + } + for i, amount := range []int64{100, -40, 20, -10} { + date := 1_800_000_000 + i + if _, err := pool.Exec(ctx, `INSERT INTO ton_transactions(user_id,amount_nanoton,reason,date) VALUES($1,$2,'adjust',$3)`, ownerID, amount, date); err != nil { + t.Fatalf("insert ton transaction %d: %v", i, err) + } + if _, err := pool.Exec(ctx, `INSERT INTO channel_stars_transactions(channel_id,actor_user_id,amount,reason,date) VALUES($1,$2,$3,'adjust',$4)`, channelID, ownerID, amount, date); err != nil { + t.Fatalf("insert channel stars transaction %d: %v", i, err) + } + if _, err := pool.Exec(ctx, `INSERT INTO channel_ton_transactions(channel_id,actor_user_id,amount_nanoton,reason,date) VALUES($1,$2,$3,'adjust',$4)`, channelID, ownerID, amount, date); err != nil { + t.Fatalf("insert channel ton transaction %d: %v", i, err) + } + } + + tonIncoming, err := lifecycle.TonTransactions(ctx, ownerID, domain.StarsTransactionQuery{ + Limit: 10, Direction: domain.StarsTransactionDirectionIncoming, + }) + if err != nil { + t.Fatalf("personal ton incoming: %v", err) + } + assertTonTransactionAmounts(t, tonIncoming.Transactions, []int64{20, 100}) + + tonOutgoing, err := lifecycle.TonTransactions(ctx, ownerID, domain.StarsTransactionQuery{ + Limit: 10, Direction: domain.StarsTransactionDirectionOutgoing, Ascending: true, + }) + if err != nil { + t.Fatalf("personal ton outgoing: %v", err) + } + assertTonTransactionAmounts(t, tonOutgoing.Transactions, []int64{-40, -10}) + + channelIncoming1, err := lifecycle.ChannelStarsTransactions(ctx, channelID, domain.StarsTransactionQuery{ + Limit: 1, Direction: domain.StarsTransactionDirectionIncoming, + }) + if err != nil { + t.Fatalf("channel stars incoming page1: %v", err) + } + assertPostgresStarsAmounts(t, channelIncoming1.Transactions, []int64{20}) + if channelIncoming1.NextOffset == "" { + t.Fatal("channel stars incoming page1 missing next offset") + } + channelIncoming2, err := lifecycle.ChannelStarsTransactions(ctx, channelID, domain.StarsTransactionQuery{ + Offset: channelIncoming1.NextOffset, Limit: 1, Direction: domain.StarsTransactionDirectionIncoming, + }) + if err != nil { + t.Fatalf("channel stars incoming page2: %v", err) + } + assertPostgresStarsAmounts(t, channelIncoming2.Transactions, []int64{100}) + if channelIncoming2.NextOffset != "" { + t.Fatalf("channel stars terminal next offset = %q", channelIncoming2.NextOffset) + } + + channelTonOutgoing, err := lifecycle.ChannelTonTransactions(ctx, channelID, domain.StarsTransactionQuery{ + Limit: 10, Direction: domain.StarsTransactionDirectionOutgoing, + }) + if err != nil { + t.Fatalf("channel ton outgoing: %v", err) + } + assertTonTransactionAmounts(t, channelTonOutgoing.Transactions, []int64{-10, -40}) +} + +func assertPostgresStarsAmounts(t *testing.T, transactions []domain.StarsTransaction, want []int64) { + t.Helper() + if len(transactions) != len(want) { + t.Fatalf("stars transaction count = %d, want %d: %+v", len(transactions), len(want), transactions) + } + for i, amount := range want { + if transactions[i].Amount != amount { + t.Fatalf("stars transaction[%d].amount = %d, want %d", i, transactions[i].Amount, amount) + } + } +} + +func assertTonTransactionAmounts(t *testing.T, transactions []domain.TonTransaction, want []int64) { + t.Helper() + if len(transactions) != len(want) { + t.Fatalf("ton transaction count = %d, want %d: %+v", len(transactions), len(want), transactions) + } + for i, amount := range want { + if transactions[i].Amount != amount { + t.Fatalf("ton transaction[%d].amount = %d, want %d", i, transactions[i].Amount, amount) + } + } +} diff --git a/internal/store/postgres/stars.go b/internal/store/postgres/stars.go index beee06d4..5443599f 100644 --- a/internal/store/postgres/stars.go +++ b/internal/store/postgres/stars.go @@ -130,12 +130,13 @@ func (s *StarsStore) Debit(ctx context.Context, userID, amount int64, reason dom return out, nil } -func (s *StarsStore) ListTransactions(ctx context.Context, userID int64, offset string, limit int) (domain.StarsTransactionPage, error) { +func (s *StarsStore) ListTransactions(ctx context.Context, userID int64, query domain.StarsTransactionQuery) (domain.StarsTransactionPage, error) { if userID == 0 { return domain.StarsTransactionPage{}, nil } - if limit <= 0 || limit > domain.MaxStarsTransactionsLimit { - limit = domain.MaxStarsTransactionsLimit + query, err := domain.NormalizeStarsTransactionQuery(query) + if err != nil { + return domain.StarsTransactionPage{}, err } bal, err := s.GetBalance(ctx, userID) if err != nil { @@ -143,29 +144,19 @@ func (s *StarsStore) ListTransactions(ctx context.Context, userID int64, offset } page := domain.StarsTransactionPage{Balance: bal.Balance} - // keyset：多取一条以探测是否还有下一页。 - args := []any{userID, limit + 1} - query := ` + // keyset：方向过滤先于 LIMIT，多取一条以探测同一视图是否还有下一页。 + where, order, args := starsTransactionQueryParts("user_id", "amount", userID, query) + rows, err := s.db.Query(ctx, ` SELECT id, peer_type, peer_id, amount, reason, title, description, date FROM stars_transactions -WHERE user_id = $1 -ORDER BY id DESC -LIMIT $2` - if cursor, ok := domain.DecodeStarsCursor(offset); ok { - query = ` -SELECT id, peer_type, peer_id, amount, reason, title, description, date -FROM stars_transactions -WHERE user_id = $1 AND id < $3 -ORDER BY id DESC -LIMIT $2` - args = append(args, cursor) - } - rows, err := s.db.Query(ctx, query, args...) +WHERE `+where+` +ORDER BY id `+order+` +LIMIT $2`, args...) if err != nil { return domain.StarsTransactionPage{}, fmt.Errorf("list stars transactions: %w", err) } defer rows.Close() - txns := make([]domain.StarsTransaction, 0, limit) + txns := make([]domain.StarsTransaction, 0, query.Limit+1) for rows.Next() { var ( t domain.StarsTransaction @@ -186,14 +177,37 @@ LIMIT $2` if err := rows.Err(); err != nil { return domain.StarsTransactionPage{}, fmt.Errorf("iterate stars transactions: %w", err) } - if len(txns) > limit { - txns = txns[:limit] + if len(txns) > query.Limit { + txns = txns[:query.Limit] page.NextOffset = domain.EncodeStarsCursor(txns[len(txns)-1].ID) } page.Transactions = txns return page, nil } +// starsTransactionQueryParts centralizes the sign predicate and keyset +// direction for personal/channel Stars and TON ledgers. Column names are only +// package-owned constants; client values remain bind parameters. +func starsTransactionQueryParts(ownerColumn, amountColumn string, ownerID int64, query domain.StarsTransactionQuery) (string, string, []any) { + where := ownerColumn + "=$1" + switch query.Direction { + case domain.StarsTransactionDirectionIncoming: + where += " AND " + amountColumn + ">0" + case domain.StarsTransactionDirectionOutgoing: + where += " AND " + amountColumn + "<0" + } + order, comparator := "DESC", "<" + if query.Ascending { + order, comparator = "ASC", ">" + } + args := []any{ownerID, query.Limit + 1} + if cursor, ok := domain.DecodeStarsCursor(query.Offset); ok { + where += " AND id" + comparator + "$3" + args = append(args, cursor) + } + return where, order, args +} + // insertStarsTxn 在事务内写一条流水（amount 带符号）。 func insertStarsTxn(ctx context.Context, tx pgx.Tx, userID, amount int64, reason domain.StarsTransactionReason, peer domain.Peer, date int, title, desc string) error { if _, err := tx.Exec(ctx, ` diff --git a/internal/store/postgres/stars_integration_test.go b/internal/store/postgres/stars_integration_test.go index 42a6f483..bcdc1f6c 100644 --- a/internal/store/postgres/stars_integration_test.go +++ b/internal/store/postgres/stars_integration_test.go @@ -65,7 +65,7 @@ func TestStarsLedgerPostgres(t *testing.T) { } // 流水：grant(+1000) / debit(-300) / credit(+50) 共 3 条，倒序最新在前。 - page, err := st.ListTransactions(ctx, u.ID, "", 2) + page, err := st.ListTransactions(ctx, u.ID, domain.StarsTransactionQuery{Limit: 2}) if err != nil { t.Fatalf("list page1: %v", err) } @@ -78,7 +78,7 @@ func TestStarsLedgerPostgres(t *testing.T) { if page.Balance != 750 { t.Fatalf("page balance = %d, want 750", page.Balance) } - page2, err := st.ListTransactions(ctx, u.ID, page.NextOffset, 2) + page2, err := st.ListTransactions(ctx, u.ID, domain.StarsTransactionQuery{Offset: page.NextOffset, Limit: 2}) if err != nil { t.Fatalf("list page2: %v", err) } @@ -88,4 +88,41 @@ func TestStarsLedgerPostgres(t *testing.T) { if page2.Transactions[0].Reason != domain.StarsReasonGrant || page2.Transactions[0].Amount != 1000 { t.Fatalf("page2[0] = %+v, want +1000 grant (oldest)", page2.Transactions[0]) } + + incoming1, err := st.ListTransactions(ctx, u.ID, domain.StarsTransactionQuery{ + Limit: 1, Direction: domain.StarsTransactionDirectionIncoming, + }) + if err != nil { + t.Fatalf("incoming page1: %v", err) + } + if len(incoming1.Transactions) != 1 || incoming1.Transactions[0].Amount != 50 || incoming1.NextOffset == "" { + t.Fatalf("incoming page1 = %+v next=%q, want +50 and next", incoming1.Transactions, incoming1.NextOffset) + } + incoming2, err := st.ListTransactions(ctx, u.ID, domain.StarsTransactionQuery{ + Offset: incoming1.NextOffset, Limit: 1, Direction: domain.StarsTransactionDirectionIncoming, + }) + if err != nil { + t.Fatalf("incoming page2: %v", err) + } + if len(incoming2.Transactions) != 1 || incoming2.Transactions[0].Amount != 1000 || incoming2.NextOffset != "" { + t.Fatalf("incoming page2 = %+v next=%q, want +1000 terminal", incoming2.Transactions, incoming2.NextOffset) + } + + outgoing, err := st.ListTransactions(ctx, u.ID, domain.StarsTransactionQuery{ + Limit: 10, Direction: domain.StarsTransactionDirectionOutgoing, + }) + if err != nil || len(outgoing.Transactions) != 1 || outgoing.Transactions[0].Amount != -300 { + t.Fatalf("outgoing = %+v err=%v, want only -300", outgoing.Transactions, err) + } + + ascending, err := st.ListTransactions(ctx, u.ID, domain.StarsTransactionQuery{Limit: 10, Ascending: true}) + if err != nil || len(ascending.Transactions) != 3 { + t.Fatalf("ascending = %+v err=%v", ascending.Transactions, err) + } + wantAscending := []int64{1000, -300, 50} + for i, amount := range wantAscending { + if ascending.Transactions[i].Amount != amount { + t.Fatalf("ascending[%d].amount = %d, want %d", i, ascending.Transactions[i].Amount, amount) + } + } } diff --git a/internal/store/star_gift.go b/internal/store/star_gift.go index 7d81d2d6..17e7ef6b 100644 --- a/internal/store/star_gift.go +++ b/internal/store/star_gift.go @@ -108,11 +108,11 @@ type StarGiftLifecycleStore interface { ResolveStarGiftWithdrawal(ctx context.Context, providerRequestID string) (domain.StarGiftWithdrawal, bool, error) CompleteStarGiftWithdrawal(ctx context.Context, providerRequestID string, date int) (domain.StarGiftWithdrawal, error) TonBalance(ctx context.Context, userID int64) (int64, error) - TonTransactions(ctx context.Context, userID int64, offset string, limit int) (domain.TonTransactionPage, error) + TonTransactions(ctx context.Context, userID int64, query domain.StarsTransactionQuery) (domain.TonTransactionPage, error) ChannelStarsBalance(ctx context.Context, channelID int64) (int64, error) - ChannelStarsTransactions(ctx context.Context, channelID int64, offset string, limit int) (domain.StarsTransactionPage, error) + ChannelStarsTransactions(ctx context.Context, channelID int64, query domain.StarsTransactionQuery) (domain.StarsTransactionPage, error) ChannelTonBalance(ctx context.Context, channelID int64) (int64, error) - ChannelTonTransactions(ctx context.Context, channelID int64, offset string, limit int) (domain.TonTransactionPage, error) + ChannelTonTransactions(ctx context.Context, channelID int64, query domain.StarsTransactionQuery) (domain.TonTransactionPage, error) // SweepStarGiftLifecycle advances time-driven offer/auction aggregates and // drains their durable notification/delivery outboxes in bounded batches. SweepStarGiftLifecycle(ctx context.Context, now, limit int) error diff --git a/internal/store/stars.go b/internal/store/stars.go index 48cb8669..e92ee98d 100644 --- a/internal/store/stars.go +++ b/internal/store/stars.go @@ -19,6 +19,6 @@ type StarsStore interface { // Debit 在单事务内做 SELECT ... FOR UPDATE 充足性检查后扣款（amount>0），写流水（amount=-x）。 // 余额不足返回 domain.ErrStarsInsufficient。 Debit(ctx context.Context, userID, amount int64, reason domain.StarsTransactionReason, peer domain.Peer, date int, title, desc string) (domain.StarsBalance, error) - // ListTransactions 按 id DESC keyset 分页返回一页流水 + 当前余额。 - ListTransactions(ctx context.Context, userID int64, offset string, limit int) (domain.StarsTransactionPage, error) + // ListTransactions 按方向与顺序做 keyset 分页，返回一页流水 + 当前余额。 + ListTransactions(ctx context.Context, userID int64, query domain.StarsTransactionQuery) (domain.StarsTransactionPage, error) }