astra/owpengram-server
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

fix(stars): sync filter transaction history by direction

Browse source
This commit is contained in:
iamxvbaba 2026-08-02 01:51:02 +08:00
parent 94f3843d23
commit 3a123f38db
19 changed files with 559 additions and 135 deletions
Download patch file Download diff file Expand all files Collapse all files

8
deploy/migrations/0164_star_transaction_direction_indexes.down.sql Normal file
View file

@ -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;

22
deploy/migrations/0164_star_transaction_direction_indexes.up.sql Normal file
View file

@ -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;

36
internal/app/stargifts/service.go
View file

@ -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 {

14
internal/app/stars/service.go
View file

@ -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)
}

80
internal/app/stars/service_test.go
View file

@ -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)
}
}
}

53
internal/domain/stars.go
View file

@ -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

4
internal/rpc/deps.go
View file

@ -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）。

2
internal/rpc/errors.go
View file

@ -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") }

52
internal/rpc/payments.go
View file

@ -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()

2
internal/rpc/payments_star_gifts_rpc_test.go
View file

@ -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)
}

75
internal/rpc/payments_stars_rpc_test.go
View file

@ -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
}

46
internal/store/memory/stars.go
View file

@ -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

68
internal/store/postgres/star_gift_lifecycle.go
View file

@ -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

6
internal/store/postgres/star_gift_lifecycle_integration_test.go
View file

@ -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)

117
internal/store/postgres/star_transaction_query_integration_test.go Normal file
View file

@ -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)
}
}
}

58
internal/store/postgres/stars.go
View file

@ -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, `

41
internal/store/postgres/stars_integration_test.go
View file

@ -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)
}
}
}

6
internal/store/star_gift.go
View file

@ -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

4
internal/store/stars.go
View file

@ -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)
}