feat(stars): sync implement unified purchase flow

This commit is contained in:
iamxvbaba 2026-08-02 01:51:03 +08:00
parent c13fbf8885
commit 7e0f9d1e62
26 changed files with 2087 additions and 282 deletions

View file

@ -12,6 +12,18 @@ import (
)
func (s *ChannelStore) SendChannelMessage(ctx context.Context, req domain.SendChannelMessageRequest) (domain.SendChannelMessageResult, error) {
return s.sendChannelMessageWithHooks(ctx, req, channelSendTxHooks{})
}
type channelSendTxHooks struct {
before func(context.Context, pgx.Tx, *domain.SendChannelMessageRequest) error
after func(context.Context, pgx.Tx, domain.SendChannelMessageResult) error
}
// sendChannelMessageWithHooks lets a tightly coupled domain command join the
// channel message/event/PTS transaction. It is deliberately package-private:
// ordinary callers must use SendChannelMessage and may not inject SQL work.
func (s *ChannelStore) sendChannelMessageWithHooks(ctx context.Context, req domain.SendChannelMessageRequest, hooks channelSendTxHooks) (domain.SendChannelMessageResult, error) {
if req.UserID == 0 || req.ChannelID == 0 || (strings.TrimSpace(req.Message) == "" && req.Action == nil && req.Media.IsZero() && req.RichMessage.IsZero()) {
return domain.SendChannelMessageResult{}, domain.ErrChannelInvalid
}
@ -27,7 +39,7 @@ func (s *ChannelStore) SendChannelMessage(ctx context.Context, req domain.SendCh
}
var lastErr error
for attempt := 0; attempt < retryableChannelTxAttempts; attempt++ {
res, err := s.sendChannelMessageOnce(ctx, req, requestFingerprint)
res, err := s.sendChannelMessageOnce(ctx, req, requestFingerprint, hooks)
if err == nil || !isRetryablePostgresTxError(err) || ctx.Err() != nil {
return res, err
}
@ -36,7 +48,7 @@ func (s *ChannelStore) SendChannelMessage(ctx context.Context, req domain.SendCh
return domain.SendChannelMessageResult{}, lastErr
}
func (s *ChannelStore) sendChannelMessageOnce(ctx context.Context, req domain.SendChannelMessageRequest, requestFingerprint []byte) (domain.SendChannelMessageResult, error) {
func (s *ChannelStore) sendChannelMessageOnce(ctx context.Context, req domain.SendChannelMessageRequest, requestFingerprint []byte, hooks channelSendTxHooks) (domain.SendChannelMessageResult, error) {
if req.RandomID != 0 && !req.IdempotencyPreflighted {
if dup, found, err := s.LookupChannelSendReplay(ctx, domain.ChannelSendReplayRequest{
ChannelID: req.ChannelID,
@ -125,6 +137,11 @@ func (s *ChannelStore) sendChannelMessageOnce(ctx context.Context, req domain.Se
p := *req.SendAs
sendAs = &p
}
if hooks.before != nil {
if err := hooks.before(ctx, tx, &req); err != nil {
return domain.SendChannelMessageResult{}, err
}
}
msgID, err := s.msgIDs.NextChannelMessageID(ctx, req.ChannelID)
if err != nil {
return domain.SendChannelMessageResult{}, fmt.Errorf("allocate channel message id: %w", err)
@ -135,7 +152,7 @@ func (s *ChannelStore) sendChannelMessageOnce(ctx context.Context, req domain.Se
}
var discussion *domain.SendChannelDiscussionResult
var discussionRef *domain.ChannelDiscussionRef
if channel.Broadcast && channel.LinkedChatID != 0 {
if channel.Broadcast && channel.LinkedChatID != 0 && req.Action == nil {
linked, err := getChannelByID(ctx, tx, channel.LinkedChatID)
if err == nil && !linked.Deleted && linked.Megagroup {
discussionMsgID, err := s.msgIDs.NextChannelMessageID(ctx, linked.ID)
@ -330,6 +347,16 @@ WHERE channel_id = $1 AND user_id = $2 AND unread_mark`, req.ChannelID, req.User
return domain.SendChannelMessageResult{}, err
}
}
txResult := domain.SendChannelMessageResult{
Channel: channel, Message: msg, Event: event, Discussion: discussion,
MentionUserIDs: append([]int64(nil), req.MentionUserIDs...),
SkipDeliveryUserIDs: append([]int64(nil), req.SkipDeliveryUserIDs...),
}
if hooks.after != nil {
if err := hooks.after(ctx, tx, txResult); err != nil {
return domain.SendChannelMessageResult{}, err
}
}
if err := tx.Commit(ctx); err != nil {
return domain.SendChannelMessageResult{}, fmt.Errorf("commit send channel: %w", err)
}
@ -342,7 +369,8 @@ WHERE channel_id = $1 AND user_id = $2 AND unread_mark`, req.ChannelID, req.User
discussion.Recipients, _ = s.ListActiveChannelMemberIDs(ctx, req.UserID, discussion.Channel.ID, 0)
}
}
return domain.SendChannelMessageResult{Channel: channel, Message: msg, Event: event, Recipients: recipients, Discussion: discussion, MentionUserIDs: append([]int64(nil), req.MentionUserIDs...), SkipDeliveryUserIDs: append([]int64(nil), req.SkipDeliveryUserIDs...)}, nil
txResult.Recipients = recipients
return txResult, nil
}
func channelDeliverySkipSet(ids []int64) map[int64]struct{} {

View file

@ -6,69 +6,96 @@ import (
"crypto/rand"
"crypto/sha256"
"encoding/binary"
"encoding/json"
"errors"
"fmt"
"math"
"strings"
"unicode/utf8"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgtype"
"telesrv/internal/domain"
"telesrv/internal/store"
"telesrv/internal/store/postgres/sqlcgen"
)
// StarsGiftPurchaseStore commits a fiat Stars gift as one aggregate with the
// private service message. No external provider is contacted by this local
// development checkout; form binding and settlement idempotency are still
// production-shaped so retries cannot mint twice.
type StarsGiftPurchaseStore struct {
// StarsPurchaseStore commits fiat Stars top-ups, friend gifts and giveaway
// launches. No external
// provider is contacted by this local development checkout; form binding and
// settlement idempotency are still production-shaped so retries cannot mint twice.
type StarsPurchaseStore struct {
db sqlcgen.DBTX
messages *MessageStore
channels *ChannelStore
}
func NewStarsGiftPurchaseStore(db sqlcgen.DBTX, messages *MessageStore) *StarsGiftPurchaseStore {
return &StarsGiftPurchaseStore{db: db, messages: messages}
func NewStarsPurchaseStore(db sqlcgen.DBTX, messages *MessageStore, channels ...*ChannelStore) *StarsPurchaseStore {
var channelStore *ChannelStore
if len(channels) > 0 {
channelStore = channels[0]
}
return &StarsPurchaseStore{db: db, messages: messages, channels: channelStore}
}
func (s *StarsGiftPurchaseStore) IssueStarsGiftPurchaseForm(ctx context.Context, form domain.StarsGiftPurchaseForm) (domain.StarsGiftPurchaseForm, error) {
if s == nil || s.db == nil || form.BuyerUserID <= 0 || form.RecipientUserID <= 0 ||
form.BuyerUserID == form.RecipientUserID || form.Stars <= 0 || form.Amount <= 0 ||
len(form.Currency) != 3 || form.IssuedAt <= 0 || form.ExpiresAt != form.IssuedAt+600 {
return domain.StarsGiftPurchaseForm{}, domain.ErrStarsGiftFormInvalid
func (s *StarsPurchaseStore) IssueStarsPurchaseForm(ctx context.Context, form domain.StarsPurchaseForm) (domain.StarsPurchaseForm, error) {
if s == nil || s.db == nil || !validStarsPurchaseForm(form) {
return domain.StarsPurchaseForm{}, domain.ErrStarsPurchaseFormInvalid
}
purposeJSON, err := starsPurchasePurposeJSON(form)
if err != nil {
return domain.StarsPurchaseForm{}, domain.ErrStarsPurchaseFormInvalid
}
for attempt := 0; attempt < 8; attempt++ {
formID, err := newStarsGiftFormID()
formID, err := newStarsPurchaseFormID()
if err != nil {
return domain.StarsGiftPurchaseForm{}, fmt.Errorf("generate stars gift form id: %w", err)
return domain.StarsPurchaseForm{}, fmt.Errorf("generate stars purchase form id: %w", err)
}
tag, err := s.db.Exec(ctx, `
INSERT INTO stars_gift_purchase_forms
(buyer_user_id,form_id,recipient_user_id,stars,currency,amount,issued_at,expires_at)
VALUES($1,$2,$3,$4,$5,$6,$7,$8)
ON CONFLICT DO NOTHING`, form.BuyerUserID, formID, form.RecipientUserID, form.Stars,
form.Currency, form.Amount, form.IssuedAt, form.ExpiresAt)
INSERT INTO stars_purchase_forms
(buyer_user_id,form_id,kind,recipient_user_id,spend_peer_type,spend_peer_id,purpose_json,stars,currency,amount,issued_at,expires_at)
VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12)
ON CONFLICT DO NOTHING`, form.BuyerUserID, formID, string(form.Kind), starsPurchaseRecipientValue(form.RecipientUserID),
starsPurchasePeerTypeValue(form.SpendPurposePeer), starsPurchasePeerIDValue(form.SpendPurposePeer),
purposeJSON, form.Stars, form.Currency, form.Amount, form.IssuedAt, form.ExpiresAt)
if err != nil {
return domain.StarsGiftPurchaseForm{}, fmt.Errorf("insert stars gift form: %w", err)
return domain.StarsPurchaseForm{}, fmt.Errorf("insert stars purchase form: %w", err)
}
if tag.RowsAffected() == 1 {
form.FormID = formID
return form, nil
}
}
return domain.StarsGiftPurchaseForm{}, domain.ErrStarsGiftUnavailable
return domain.StarsPurchaseForm{}, domain.ErrStarsGiftUnavailable
}
var errStarsGiftPurchaseReplay = errors.New("stars gift purchase replay")
var errStarsPurchaseReplay = errors.New("stars purchase replay")
func (s *StarsGiftPurchaseStore) PurchaseStarsGift(ctx context.Context, req domain.StarsGiftPurchaseRequest) (domain.StarsGiftPurchaseResult, error) {
if s == nil || s.db == nil || s.messages == nil || req.FormID == 0 ||
req.BuyerUserID <= 0 || req.RecipientUserID <= 0 || req.BuyerUserID == req.RecipientUserID ||
req.Stars <= 0 || req.Amount <= 0 || len(req.Currency) != 3 || req.Date <= 0 {
return domain.StarsGiftPurchaseResult{}, domain.ErrStarsGiftFormInvalid
func (s *StarsPurchaseStore) PurchaseStars(ctx context.Context, req domain.StarsPurchaseRequest) (domain.StarsPurchaseResult, error) {
if s == nil || s.db == nil || req.FormID == 0 || req.Date <= 0 || !validStarsPurchaseCommand(req.StarsPurchaseForm) {
return domain.StarsPurchaseResult{}, domain.ErrStarsPurchaseFormInvalid
}
fingerprint := starsGiftPurchaseFingerprint(req)
if replay, found, err := s.loadStarsGiftPurchaseReplay(ctx, req, fingerprint); err != nil || found {
fingerprint := starsPurchaseFingerprint(req)
if replay, found, err := s.loadStarsPurchaseReplay(ctx, req, fingerprint); err != nil || found {
return replay, err
}
// A committed command is replayable after both the checkout and giveaway
// deadlines: the provider may retry a successful submission after losing the
// response, and a terminal campaign must not turn that exact retry into a
// different outcome. The deadline only gates a first settlement.
if req.Kind == domain.StarsPurchaseGiveaway && req.Giveaway.UntilDate <= req.Date {
return domain.StarsPurchaseResult{}, domain.ErrStarsPurchaseFormExpired
}
switch req.Kind {
case domain.StarsPurchaseTopup:
return s.purchaseStarsTopup(ctx, req, fingerprint)
case domain.StarsPurchaseGiveaway:
return s.purchaseStarsGiveaway(ctx, req, fingerprint)
}
if s.messages == nil {
return domain.StarsPurchaseResult{}, domain.ErrStarsPurchaseFormInvalid
}
transactionID := fmt.Sprintf("stars-gift:%d:%d", req.BuyerUserID, req.FormID)
randomID := lifecycleCommandRandomID("stars-fiat-gift", req.BuyerUserID, req.FormID)
@ -83,16 +110,16 @@ func (s *StarsGiftPurchaseStore) PurchaseStarsGift(ctx context.Context, req doma
OriginUserID: req.BuyerUserID, OriginAuthKeyID: req.OriginAuthKeyID,
OriginSessionID: req.OriginSessionID, IdempotencyFingerprint: fingerprint[:],
}
result := domain.StarsGiftPurchaseResult{TransactionID: transactionID}
result := domain.StarsPurchaseResult{TransactionID: transactionID}
hooks := privateSendTxHooks{
before: func(ctx context.Context, tx pgx.Tx, send *domain.SendPrivateTextRequest) error {
if err := validateStarsGiftPurchaseForm(ctx, tx, req, true); err != nil {
if err := validateStarsPurchaseForm(ctx, tx, req, true); err != nil {
return err
}
if found, err := starsGiftPurchaseCommandExists(ctx, tx, req.BuyerUserID, req.FormID); err != nil {
if found, err := starsPurchaseCommandExists(ctx, tx, req.BuyerUserID, req.FormID); err != nil {
return err
} else if found {
return errStarsGiftPurchaseReplay
return errStarsPurchaseReplay
}
balance := domain.StarsBalance{UserID: req.RecipientUserID}
if err := tx.QueryRow(ctx, `
@ -107,21 +134,20 @@ RETURNING balance,granted`, req.RecipientUserID, req.Stars).Scan(&balance.Balanc
"Stars gift", fmt.Sprintf("%d Stars", req.Stars)); err != nil {
return err
}
result.RecipientBalance = balance
result.Balance = balance
if send.Media == nil || send.Media.ServiceAction == nil || send.Media.ServiceAction.GiftStars == nil {
return domain.ErrStarsGiftFormInvalid
return domain.ErrStarsPurchaseFormInvalid
}
send.Media.ServiceAction.GiftStars.BalanceAfter = balance.Balance
return nil
},
after: func(ctx context.Context, tx pgx.Tx, sent domain.SendPrivateTextResult) error {
_, err := tx.Exec(ctx, `
INSERT INTO stars_gift_purchase_commands
(buyer_user_id,form_id,request_fingerprint,recipient_user_id,stars,currency,amount,
recipient_balance_after,transaction_id,created_at)
VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10)`, req.BuyerUserID, req.FormID, fingerprint[:],
req.RecipientUserID, req.Stars, req.Currency, req.Amount,
result.RecipientBalance.Balance, transactionID, req.Date)
INSERT INTO stars_purchase_commands
(buyer_user_id,form_id,kind,request_fingerprint,recipient_user_id,spend_peer_type,spend_peer_id,purpose_json,stars,currency,amount,
balance_after,transaction_id,created_at)
VALUES($1,$2,$3,$4,$5,NULL,NULL,'{}'::jsonb,$6,$7,$8,$9,$10,$11)`, req.BuyerUserID, req.FormID, string(req.Kind), fingerprint[:],
req.RecipientUserID, req.Stars, req.Currency, req.Amount, result.Balance.Balance, transactionID, req.Date)
if err != nil {
return fmt.Errorf("insert stars gift purchase command: %w", err)
}
@ -131,61 +157,233 @@ VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10)`, req.BuyerUserID, req.FormID, fingerprin
}
sent, err := s.messages.sendPrivateTextWithHooks(ctx, messageReq, hooks)
if err != nil {
if errors.Is(err, errStarsGiftPurchaseReplay) {
if replay, found, replayErr := s.loadStarsGiftPurchaseReplay(ctx, req, fingerprint); replayErr != nil || found {
if errors.Is(err, errStarsPurchaseReplay) {
if replay, found, replayErr := s.loadStarsPurchaseReplay(ctx, req, fingerprint); replayErr != nil || found {
return replay, replayErr
}
}
return domain.StarsGiftPurchaseResult{}, err
return domain.StarsPurchaseResult{}, err
}
result.Send = sent
return result, nil
}
func validateStarsGiftPurchaseForm(ctx context.Context, db sqlcgen.DBTX, req domain.StarsGiftPurchaseRequest, lock bool) error {
query := `SELECT recipient_user_id,stars,currency,amount,issued_at,expires_at
FROM stars_gift_purchase_forms WHERE buyer_user_id=$1 AND form_id=$2`
func (s *StarsPurchaseStore) purchaseStarsTopup(ctx context.Context, req domain.StarsPurchaseRequest, fingerprint [32]byte) (domain.StarsPurchaseResult, error) {
transactionID := fmt.Sprintf("stars-topup:%d:%d", req.BuyerUserID, req.FormID)
result := domain.StarsPurchaseResult{TransactionID: transactionID}
err := withTx(ctx, s.db, "settle stars topup", func(tx pgx.Tx) error {
if err := validateStarsPurchaseForm(ctx, tx, req, true); err != nil {
return err
}
if found, err := starsPurchaseCommandExists(ctx, tx, req.BuyerUserID, req.FormID); err != nil {
return err
} else if found {
return errStarsPurchaseReplay
}
result.Balance = domain.StarsBalance{UserID: req.BuyerUserID}
if err := tx.QueryRow(ctx, `
INSERT INTO stars_balances (user_id,balance,updated_at) VALUES($1,$2,now())
ON CONFLICT (user_id) DO UPDATE
SET balance=stars_balances.balance+EXCLUDED.balance, updated_at=now()
RETURNING balance,granted`, req.BuyerUserID, req.Stars).Scan(&result.Balance.Balance, &result.Balance.Granted); err != nil {
return fmt.Errorf("credit stars topup buyer: %w", err)
}
if err := insertStarsTxn(ctx, tx, req.BuyerUserID, req.Stars, domain.StarsReasonTopup,
req.SpendPurposePeer, req.Date, "Stars top-up", "telesrv dev purchase"); err != nil {
return err
}
_, err := tx.Exec(ctx, `
INSERT INTO stars_purchase_commands
(buyer_user_id,form_id,kind,request_fingerprint,recipient_user_id,spend_peer_type,spend_peer_id,purpose_json,stars,currency,amount,
balance_after,transaction_id,created_at)
VALUES($1,$2,$3,$4,NULL,$5,$6,'{}'::jsonb,$7,$8,$9,$10,$11,$12)`, req.BuyerUserID, req.FormID, string(req.Kind), fingerprint[:],
starsPurchasePeerTypeValue(req.SpendPurposePeer), starsPurchasePeerIDValue(req.SpendPurposePeer),
req.Stars, req.Currency, req.Amount, result.Balance.Balance, transactionID, req.Date)
if err != nil {
return fmt.Errorf("insert stars topup command: %w", err)
}
return nil
})
if errors.Is(err, errStarsPurchaseReplay) {
if replay, found, replayErr := s.loadStarsPurchaseReplay(ctx, req, fingerprint); replayErr != nil || found {
return replay, replayErr
}
}
if err != nil {
return domain.StarsPurchaseResult{}, err
}
return result, nil
}
func (s *StarsPurchaseStore) purchaseStarsGiveaway(ctx context.Context, req domain.StarsPurchaseRequest, fingerprint [32]byte) (domain.StarsPurchaseResult, error) {
if s.channels == nil || req.Giveaway == nil {
return domain.StarsPurchaseResult{}, domain.ErrStarsPurchaseFormInvalid
}
purposeJSON, err := starsPurchasePurposeJSON(req.StarsPurchaseForm)
if err != nil {
return domain.StarsPurchaseResult{}, domain.ErrStarsPurchaseFormInvalid
}
giveaway := *req.Giveaway
transactionID := fmt.Sprintf("stars-giveaway:%d:%d", req.BuyerUserID, req.FormID)
result := domain.StarsPurchaseResult{TransactionID: transactionID}
sendReq := domain.SendChannelMessageRequest{
UserID: req.BuyerUserID, ChannelID: giveaway.BoostPeer.ID,
RandomID: giveaway.RandomID, Date: req.Date,
Media: &domain.MessageMedia{Kind: domain.MessageMediaKindGiveaway, Giveaway: &domain.MessageGiveaway{
OnlyNewSubscribers: giveaway.OnlyNewSubscribers, WinnersAreVisible: giveaway.WinnersAreVisible,
Channels: starsGiveawayChannelIDs(giveaway), CountriesISO2: append([]string(nil), giveaway.CountriesISO2...),
PrizeDescription: giveaway.PrizeDescription, Quantity: giveaway.Users, Stars: req.Stars, UntilDate: giveaway.UntilDate,
}},
IdempotencyFingerprint: fingerprint[:],
}
hooks := channelSendTxHooks{
before: func(ctx context.Context, tx pgx.Tx, _ *domain.SendChannelMessageRequest) error {
if err := validateStarsPurchaseForm(ctx, tx, req, true); err != nil {
return err
}
if found, err := starsPurchaseCommandExists(ctx, tx, req.BuyerUserID, req.FormID); err != nil {
return err
} else if found {
return errStarsPurchaseReplay
}
return nil
},
after: func(ctx context.Context, tx pgx.Tx, sent domain.SendChannelMessageResult) error {
if sent.Message.ID <= 0 || sent.Event.Pts <= 0 || sent.Event.PtsCount != 1 {
return fmt.Errorf("settle stars giveaway: invalid channel send receipt")
}
if _, err := tx.Exec(ctx, `
INSERT INTO stars_giveaways
(buyer_user_id,form_id,channel_id,launch_message_id,random_id,stars,users,per_user_stars,yearly_boosts,
until_date,purpose_json,state,created_at)
VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,'active',$12)`,
req.BuyerUserID, req.FormID, giveaway.BoostPeer.ID, sent.Message.ID, giveaway.RandomID,
req.Stars, giveaway.Users, giveaway.PerUserStars, giveaway.YearlyBoosts, giveaway.UntilDate, purposeJSON, req.Date); err != nil {
return fmt.Errorf("insert stars giveaway campaign: %w", err)
}
if _, err := tx.Exec(ctx, `
INSERT INTO stars_purchase_commands
(buyer_user_id,form_id,kind,request_fingerprint,recipient_user_id,spend_peer_type,spend_peer_id,purpose_json,stars,currency,amount,
balance_after,transaction_id,created_at)
VALUES($1,$2,$3,$4,NULL,NULL,NULL,$5,$6,$7,$8,0,$9,$10)`,
req.BuyerUserID, req.FormID, string(req.Kind), fingerprint[:], purposeJSON,
req.Stars, req.Currency, req.Amount, transactionID, req.Date); err != nil {
return fmt.Errorf("insert stars giveaway purchase command: %w", err)
}
result.ChannelSend = sent
return nil
},
}
sent, err := s.channels.sendChannelMessageWithHooks(ctx, sendReq, hooks)
if err != nil {
if errors.Is(err, errStarsPurchaseReplay) {
if replay, found, replayErr := s.loadStarsPurchaseReplay(ctx, req, fingerprint); replayErr != nil || found {
return replay, replayErr
}
}
return domain.StarsPurchaseResult{}, err
}
if sent.Duplicate {
if replay, found, replayErr := s.loadStarsPurchaseReplay(ctx, req, fingerprint); replayErr != nil || found {
return replay, replayErr
}
return domain.StarsPurchaseResult{}, domain.ErrStarsPurchaseFormInvalid
}
result.ChannelSend = sent
return result, nil
}
func starsGiveawayChannelIDs(giveaway domain.StarsGiveawayPurchase) []int64 {
ids := make([]int64, 0, 1+len(giveaway.AdditionalPeers))
ids = append(ids, giveaway.BoostPeer.ID)
for _, peer := range giveaway.AdditionalPeers {
ids = append(ids, peer.ID)
}
return ids
}
func validateStarsPurchaseForm(ctx context.Context, db sqlcgen.DBTX, req domain.StarsPurchaseRequest, lock bool) error {
query := `SELECT kind,recipient_user_id,spend_peer_type,spend_peer_id,purpose_json,stars,currency,amount,issued_at,expires_at
FROM stars_purchase_forms WHERE buyer_user_id=$1 AND form_id=$2`
if lock {
query += ` FOR UPDATE`
}
var recipientID, stars, amount int64
var kind string
var recipientID pgtype.Int8
var spendPeerType pgtype.Text
var spendPeerID pgtype.Int8
var purposeJSON []byte
var stars, amount int64
var currency string
var issuedAt, expiresAt int
err := db.QueryRow(ctx, query, req.BuyerUserID, req.FormID).
Scan(&recipientID, &stars, &currency, &amount, &issuedAt, &expiresAt)
Scan(&kind, &recipientID, &spendPeerType, &spendPeerID, &purposeJSON, &stars, &currency, &amount, &issuedAt, &expiresAt)
if errors.Is(err, pgx.ErrNoRows) {
return domain.ErrStarsGiftFormInvalid
return domain.ErrStarsPurchaseFormInvalid
}
if err != nil {
return fmt.Errorf("load stars gift form: %w", err)
}
if req.Date >= expiresAt {
return domain.ErrStarsGiftFormExpired
return domain.ErrStarsPurchaseFormExpired
}
if issuedAt <= 0 || recipientID != req.RecipientUserID || stars != req.Stars ||
currency != req.Currency || amount != req.Amount {
return domain.ErrStarsGiftFormInvalid
if issuedAt <= 0 || kind != string(req.Kind) || nullableStarsRecipient(recipientID) != req.RecipientUserID ||
nullableStarsPeer(spendPeerType, spendPeerID) != req.SpendPurposePeer || stars != req.Stars ||
currency != req.Currency || amount != req.Amount || !sameStarsPurchasePurpose(purposeJSON, req.StarsPurchaseForm) {
return domain.ErrStarsPurchaseFormInvalid
}
return nil
}
func (s *StarsGiftPurchaseStore) loadStarsGiftPurchaseReplay(ctx context.Context, req domain.StarsGiftPurchaseRequest, fingerprint [32]byte) (domain.StarsGiftPurchaseResult, bool, error) {
var recipientID, stars, amount, balance int64
var currency, transactionID string
var storedFingerprint []byte
func (s *StarsPurchaseStore) loadStarsPurchaseReplay(ctx context.Context, req domain.StarsPurchaseRequest, fingerprint [32]byte) (domain.StarsPurchaseResult, bool, error) {
var recipientID pgtype.Int8
var spendPeerType pgtype.Text
var spendPeerID pgtype.Int8
var stars, amount, balance int64
var kind, currency, transactionID string
var storedFingerprint, purposeJSON []byte
err := s.db.QueryRow(ctx, `
SELECT request_fingerprint,recipient_user_id,stars,currency,amount,recipient_balance_after,transaction_id
FROM stars_gift_purchase_commands WHERE buyer_user_id=$1 AND form_id=$2`, req.BuyerUserID, req.FormID).
Scan(&storedFingerprint, &recipientID, &stars, &currency, &amount, &balance, &transactionID)
SELECT kind,request_fingerprint,recipient_user_id,spend_peer_type,spend_peer_id,purpose_json,stars,currency,amount,balance_after,transaction_id
FROM stars_purchase_commands WHERE buyer_user_id=$1 AND form_id=$2`, req.BuyerUserID, req.FormID).
Scan(&kind, &storedFingerprint, &recipientID, &spendPeerType, &spendPeerID, &purposeJSON, &stars, &currency, &amount, &balance, &transactionID)
if errors.Is(err, pgx.ErrNoRows) {
return domain.StarsGiftPurchaseResult{}, false, nil
return domain.StarsPurchaseResult{}, false, nil
}
if err != nil {
return domain.StarsGiftPurchaseResult{}, false, fmt.Errorf("load stars gift purchase replay: %w", err)
return domain.StarsPurchaseResult{}, false, fmt.Errorf("load stars purchase replay: %w", err)
}
if !bytes.Equal(storedFingerprint, fingerprint[:]) || recipientID != req.RecipientUserID ||
stars != req.Stars || currency != req.Currency || amount != req.Amount || transactionID == "" {
return domain.StarsGiftPurchaseResult{}, false, domain.ErrStarsGiftFormInvalid
if kind != string(req.Kind) || !bytes.Equal(storedFingerprint, fingerprint[:]) || nullableStarsRecipient(recipientID) != req.RecipientUserID ||
nullableStarsPeer(spendPeerType, spendPeerID) != req.SpendPurposePeer ||
stars != req.Stars || currency != req.Currency || amount != req.Amount || transactionID == "" ||
!sameStarsPurchasePurpose(purposeJSON, req.StarsPurchaseForm) {
return domain.StarsPurchaseResult{}, false, domain.ErrStarsPurchaseFormInvalid
}
result := domain.StarsPurchaseResult{
Balance: domain.StarsBalance{UserID: req.BuyerUserID, Balance: balance},
TransactionID: transactionID, Duplicate: true,
}
if req.Kind == domain.StarsPurchaseTopup {
return result, true, nil
}
if req.Kind == domain.StarsPurchaseGiveaway {
if s.channels == nil || req.Giveaway == nil {
return domain.StarsPurchaseResult{}, false, domain.ErrStarsPurchaseFormInvalid
}
sent, found, err := s.channels.LookupChannelSendReplay(ctx, domain.ChannelSendReplayRequest{
ChannelID: req.Giveaway.BoostPeer.ID, SenderUserID: req.BuyerUserID,
RandomID: req.Giveaway.RandomID, IdempotencyFingerprint: fingerprint[:],
})
if err != nil {
return domain.StarsPurchaseResult{}, false, err
}
if !found {
return domain.StarsPurchaseResult{}, false, domain.ErrStarsPurchaseFormInvalid
}
result.ChannelSend = sent
return result, true, nil
}
if s.messages == nil {
return domain.StarsPurchaseResult{}, false, domain.ErrStarsPurchaseFormInvalid
}
sent, found, err := s.messages.LookupPrivateSendReplay(ctx, domain.PrivateSendReplayRequest{
SenderUserID: req.BuyerUserID, RecipientUserID: req.RecipientUserID,
@ -193,32 +391,33 @@ FROM stars_gift_purchase_commands WHERE buyer_user_id=$1 AND form_id=$2`, req.Bu
IdempotencyFingerprint: fingerprint[:],
})
if err != nil {
return domain.StarsGiftPurchaseResult{}, false, err
return domain.StarsPurchaseResult{}, false, err
}
if !found {
return domain.StarsGiftPurchaseResult{}, false, domain.ErrStarsGiftFormInvalid
return domain.StarsPurchaseResult{}, false, domain.ErrStarsPurchaseFormInvalid
}
return domain.StarsGiftPurchaseResult{
RecipientBalance: domain.StarsBalance{UserID: req.RecipientUserID, Balance: balance},
Send: sent, TransactionID: transactionID, Duplicate: true,
}, true, nil
result.Balance.UserID = req.RecipientUserID
result.Send = sent
return result, true, nil
}
func starsGiftPurchaseCommandExists(ctx context.Context, db sqlcgen.DBTX, buyerUserID, formID int64) (bool, error) {
func starsPurchaseCommandExists(ctx context.Context, db sqlcgen.DBTX, buyerUserID, formID int64) (bool, error) {
var exists bool
if err := db.QueryRow(ctx, `SELECT EXISTS(
SELECT 1 FROM stars_gift_purchase_commands WHERE buyer_user_id=$1 AND form_id=$2)`, buyerUserID, formID).Scan(&exists); err != nil {
return false, fmt.Errorf("check stars gift purchase command: %w", err)
SELECT 1 FROM stars_purchase_commands WHERE buyer_user_id=$1 AND form_id=$2)`, buyerUserID, formID).Scan(&exists); err != nil {
return false, fmt.Errorf("check stars purchase command: %w", err)
}
return exists, nil
}
func starsGiftPurchaseFingerprint(req domain.StarsGiftPurchaseRequest) [32]byte {
return sha256.Sum256([]byte(fmt.Sprintf("telesrv:stars-fiat-gift:v1:%d:%d:%d:%s:%d:%d",
req.BuyerUserID, req.RecipientUserID, req.Stars, req.Currency, req.Amount, req.FormID)))
func starsPurchaseFingerprint(req domain.StarsPurchaseRequest) [32]byte {
purposeJSON, _ := starsPurchasePurposeJSON(req.StarsPurchaseForm)
return sha256.Sum256([]byte(fmt.Sprintf("telesrv:stars-fiat-purchase:v2:%s:%d:%d:%s:%d:%d:%s:%d:%d:%s",
req.Kind, req.BuyerUserID, req.RecipientUserID, req.SpendPurposePeer.Type, req.SpendPurposePeer.ID,
req.Stars, req.Currency, req.Amount, req.FormID, purposeJSON)))
}
func newStarsGiftFormID() (int64, error) {
func newStarsPurchaseFormID() (int64, error) {
var raw [8]byte
if _, err := rand.Read(raw[:]); err != nil {
return 0, err
@ -230,4 +429,200 @@ func newStarsGiftFormID() (int64, error) {
return id, nil
}
var _ store.StarsGiftPurchaseStore = (*StarsGiftPurchaseStore)(nil)
func validStarsPurchaseForm(form domain.StarsPurchaseForm) bool {
if !validStarsPurchaseCommand(form) || form.IssuedAt <= 0 || form.ExpiresAt != form.IssuedAt+600 {
return false
}
return form.Kind != domain.StarsPurchaseGiveaway ||
(form.Giveaway.UntilDate > form.IssuedAt && form.Giveaway.UntilDate <= form.IssuedAt+7*24*60*60)
}
func validStarsPurchaseCommand(form domain.StarsPurchaseForm) bool {
if !form.Kind.Valid() || form.BuyerUserID <= 0 || form.Stars <= 0 || form.Amount <= 0 || len(form.Currency) != 3 {
return false
}
validSpendPeer := (form.SpendPurposePeer == domain.Peer{}) ||
((form.SpendPurposePeer.Type == domain.PeerTypeUser || form.SpendPurposePeer.Type == domain.PeerTypeChannel) && form.SpendPurposePeer.ID > 0)
switch form.Kind {
case domain.StarsPurchaseTopup:
return form.RecipientUserID == 0 && validSpendPeer && form.Giveaway == nil
case domain.StarsPurchaseGift:
return form.RecipientUserID > 0 && form.RecipientUserID != form.BuyerUserID && form.SpendPurposePeer == (domain.Peer{}) && form.Giveaway == nil
case domain.StarsPurchaseGiveaway:
return form.RecipientUserID == 0 && form.SpendPurposePeer == (domain.Peer{}) && validStarsGiveawayPurchase(form.Giveaway, form.Stars)
default:
return false
}
}
func validStarsGiveawayPurchase(g *domain.StarsGiveawayPurchase, stars int64) bool {
if g == nil || g.BoostPeer.Type != domain.PeerTypeChannel || g.BoostPeer.ID <= 0 || g.RandomID == 0 ||
g.UntilDate <= 0 || g.Users <= 0 || g.PerUserStars <= 0 || g.YearlyBoosts < 0 ||
int64(g.Users) > math.MaxInt64/g.PerUserStars || int64(g.Users)*g.PerUserStars != stars ||
len(g.AdditionalPeers) > 10 || len(g.CountriesISO2) > 10 || utf8.RuneCountInString(g.PrizeDescription) > 128 {
return false
}
seenPeers := map[int64]struct{}{g.BoostPeer.ID: struct{}{}}
for _, peer := range g.AdditionalPeers {
if peer.Type != domain.PeerTypeChannel || peer.ID <= 0 {
return false
}
if _, exists := seenPeers[peer.ID]; exists {
return false
}
seenPeers[peer.ID] = struct{}{}
}
seenCountries := make(map[string]struct{}, len(g.CountriesISO2))
for _, country := range g.CountriesISO2 {
if len(country) != 2 || country != strings.ToUpper(country) || country[0] < 'A' || country[0] > 'Z' || country[1] < 'A' || country[1] > 'Z' {
return false
}
if _, exists := seenCountries[country]; exists {
return false
}
seenCountries[country] = struct{}{}
}
return true
}
func starsPurchasePurposeJSON(form domain.StarsPurchaseForm) ([]byte, error) {
if form.Kind != domain.StarsPurchaseGiveaway {
return []byte(`{}`), nil
}
if form.Giveaway == nil {
return nil, domain.ErrStarsPurchaseFormInvalid
}
return json.Marshal(form.Giveaway)
}
func sameStarsPurchasePurpose(stored []byte, form domain.StarsPurchaseForm) bool {
want, err := starsPurchasePurposeJSON(form)
if err != nil {
return false
}
if form.Kind != domain.StarsPurchaseGiveaway {
var value map[string]any
return json.Unmarshal(stored, &value) == nil && len(value) == 0
}
var decoded domain.StarsGiveawayPurchase
if err := json.Unmarshal(stored, &decoded); err != nil {
return false
}
got, err := json.Marshal(&decoded)
return err == nil && bytes.Equal(got, want)
}
func starsPurchaseRecipientValue(recipientUserID int64) any {
if recipientUserID == 0 {
return nil
}
return recipientUserID
}
func nullableStarsRecipient(value pgtype.Int8) int64 {
if !value.Valid {
return 0
}
return value.Int64
}
func starsPurchasePeerTypeValue(peer domain.Peer) any {
if peer == (domain.Peer{}) {
return nil
}
return string(peer.Type)
}
func starsPurchasePeerIDValue(peer domain.Peer) any {
if peer == (domain.Peer{}) {
return nil
}
return peer.ID
}
func nullableStarsPeer(peerType pgtype.Text, peerID pgtype.Int8) domain.Peer {
if !peerType.Valid || !peerID.Valid {
return domain.Peer{}
}
return domain.Peer{Type: domain.PeerType(peerType.String), ID: peerID.Int64}
}
func (s *StarsPurchaseStore) GetStarsGiveawayInfo(ctx context.Context, viewerUserID, channelID int64, messageID, date int) (domain.StarsGiveawayInfo, error) {
if s == nil || s.db == nil || viewerUserID <= 0 || channelID <= 0 || messageID <= 0 || date <= 0 {
return domain.StarsGiveawayInfo{}, domain.ErrStarsPurchaseFormInvalid
}
var purposeJSON []byte
var state string
var startDate, untilDate int
err := s.db.QueryRow(ctx, `
SELECT purpose_json,state,created_at,until_date
FROM stars_giveaways WHERE channel_id=$1 AND launch_message_id=$2`, channelID, messageID).
Scan(&purposeJSON, &state, &startDate, &untilDate)
if errors.Is(err, pgx.ErrNoRows) {
return domain.StarsGiveawayInfo{}, domain.ErrMessageIDInvalid
}
if err != nil {
return domain.StarsGiveawayInfo{}, fmt.Errorf("load stars giveaway info: %w", err)
}
var purpose domain.StarsGiveawayPurchase
if err := json.Unmarshal(purposeJSON, &purpose); err != nil || purpose.BoostPeer.ID != channelID {
return domain.StarsGiveawayInfo{}, domain.ErrStarsPurchaseFormInvalid
}
info := domain.StarsGiveawayInfo{StartDate: startDate}
if state == "cancelled" {
return info, nil
}
if state != "active" || date >= untilDate {
info.PreparingResults = true
return info, nil
}
channels := starsGiveawayChannelIDs(purpose)
for _, requiredChannelID := range channels {
var role, status string
var joinedAt int
err := s.db.QueryRow(ctx, `
SELECT role,status,joined_at FROM channel_members WHERE channel_id=$1 AND user_id=$2`, requiredChannelID, viewerUserID).
Scan(&role, &status, &joinedAt)
if errors.Is(err, pgx.ErrNoRows) {
return info, nil
}
if err != nil {
return domain.StarsGiveawayInfo{}, fmt.Errorf("load giveaway participant membership: %w", err)
}
if status != string(domain.ChannelMemberActive) {
return info, nil
}
if role == string(domain.ChannelRoleCreator) || role == string(domain.ChannelRoleAdmin) {
info.AdminDisallowedChatID = requiredChannelID
return info, nil
}
if purpose.OnlyNewSubscribers && joinedAt > 0 && joinedAt <= startDate {
info.JoinedTooEarlyDate = joinedAt
return info, nil
}
}
if len(purpose.CountriesISO2) > 0 {
var country string
if err := s.db.QueryRow(ctx, `
SELECT COALESCE((SELECT cc.iso2 FROM country_codes cc WHERE cc.country_code=u.country_code ORDER BY cc.id LIMIT 1),'')
FROM users u WHERE u.id=$1`, viewerUserID).Scan(&country); err != nil {
return domain.StarsGiveawayInfo{}, fmt.Errorf("load giveaway participant country: %w", err)
}
allowed := false
for _, candidate := range purpose.CountriesISO2 {
if candidate == country {
allowed = true
break
}
}
if !allowed {
info.DisallowedCountry = country
return info, nil
}
}
info.Participating = true
return info, nil
}
var _ store.StarsPurchaseStore = (*StarsPurchaseStore)(nil)
var _ store.StarsGiveawayStore = (*StarsPurchaseStore)(nil)

View file

@ -8,6 +8,43 @@ import (
"telesrv/internal/domain"
)
type starsPurchaseAttempt struct {
result domain.StarsPurchaseResult
err error
}
func purchaseStarsTwiceConcurrently(t *testing.T, ctx context.Context, store *StarsPurchaseStore, req domain.StarsPurchaseRequest) (domain.StarsPurchaseResult, domain.StarsPurchaseResult) {
t.Helper()
start := make(chan struct{})
attempts := make(chan starsPurchaseAttempt, 2)
for range 2 {
go func() {
<-start
result, err := store.PurchaseStars(ctx, req)
attempts <- starsPurchaseAttempt{result: result, err: err}
}()
}
close(start)
var first, replay domain.StarsPurchaseResult
firstCount, replayCount := 0, 0
for range 2 {
attempt := <-attempts
if attempt.err != nil {
t.Fatalf("concurrent Stars purchase: %v", attempt.err)
}
if attempt.result.Duplicate {
replay, replayCount = attempt.result, replayCount+1
} else {
first, firstCount = attempt.result, firstCount+1
}
}
if firstCount != 1 || replayCount != 1 {
t.Fatalf("concurrent Stars purchase first/replay counts = %d/%d, want 1/1", firstCount, replayCount)
}
return first, replay
}
func TestStarsFriendGiftPurchaseAtomicReplayAndValidationPostgres(t *testing.T) {
pool := testPool(t)
ctx := context.Background()
@ -22,17 +59,17 @@ func TestStarsFriendGiftPurchaseAtomicReplayAndValidationPostgres(t *testing.T)
t.Fatalf("create recipient: %v", err)
}
t.Cleanup(func() {
_, _ = pool.Exec(ctx, "DELETE FROM stars_gift_purchase_commands WHERE buyer_user_id=$1", buyer.ID)
_, _ = pool.Exec(ctx, "DELETE FROM stars_gift_purchase_forms WHERE buyer_user_id=$1", buyer.ID)
_, _ = pool.Exec(ctx, "DELETE FROM stars_purchase_commands WHERE buyer_user_id=$1", buyer.ID)
_, _ = pool.Exec(ctx, "DELETE FROM stars_purchase_forms WHERE buyer_user_id=$1", buyer.ID)
_, _ = pool.Exec(ctx, "DELETE FROM stars_transactions WHERE user_id=$1", recipient.ID)
_, _ = pool.Exec(ctx, "DELETE FROM stars_balances WHERE user_id=$1", recipient.ID)
_, _ = pool.Exec(ctx, "DELETE FROM users WHERE id=ANY($1::bigint[])", []int64{buyer.ID, recipient.ID})
})
messages := NewMessageStore(pool)
store := NewStarsGiftPurchaseStore(pool, messages)
issued, err := store.IssueStarsGiftPurchaseForm(ctx, domain.StarsGiftPurchaseForm{
BuyerUserID: buyer.ID, RecipientUserID: recipient.ID,
store := NewStarsPurchaseStore(pool, messages)
issued, err := store.IssueStarsPurchaseForm(ctx, domain.StarsPurchaseForm{
Kind: domain.StarsPurchaseGift, BuyerUserID: buyer.ID, RecipientUserID: recipient.ID,
Stars: 2500, Currency: "USD", Amount: 199,
IssuedAt: 1_700_000_000, ExpiresAt: 1_700_000_600,
})
@ -41,18 +78,15 @@ func TestStarsFriendGiftPurchaseAtomicReplayAndValidationPostgres(t *testing.T)
}
var origin [8]byte
origin[0] = 9
req := domain.StarsGiftPurchaseRequest{
StarsGiftPurchaseForm: domain.StarsGiftPurchaseForm{
FormID: issued.FormID, BuyerUserID: buyer.ID, RecipientUserID: recipient.ID,
req := domain.StarsPurchaseRequest{
StarsPurchaseForm: domain.StarsPurchaseForm{
FormID: issued.FormID, Kind: domain.StarsPurchaseGift, BuyerUserID: buyer.ID, RecipientUserID: recipient.ID,
Stars: 2500, Currency: "USD", Amount: 199,
},
Date: 1_700_000_100, OriginAuthKeyID: origin, OriginSessionID: 77,
}
first, err := store.PurchaseStarsGift(ctx, req)
if err != nil {
t.Fatalf("purchase: %v", err)
}
if first.Duplicate || first.RecipientBalance.Balance != 2500 || first.TransactionID == "" ||
first, replay := purchaseStarsTwiceConcurrently(t, ctx, store, req)
if first.Duplicate || first.Balance.Balance != 2500 || first.TransactionID == "" ||
first.Send.SenderEvent.PtsCount != 1 || first.Send.RecipientEvent.PtsCount != 1 {
t.Fatalf("first purchase = %+v", first)
}
@ -66,10 +100,6 @@ func TestStarsFriendGiftPurchaseAtomicReplayAndValidationPostgres(t *testing.T)
t.Fatalf("recipient gift action = %+v", action)
}
replay, err := store.PurchaseStarsGift(ctx, req)
if err != nil {
t.Fatalf("replay: %v", err)
}
if !replay.Duplicate || replay.TransactionID != first.TransactionID ||
replay.Send.SenderMessage.ID != first.Send.SenderMessage.ID || replay.Send.SenderEvent.Pts != first.Send.SenderEvent.Pts {
t.Fatalf("replay = %+v, first=%+v", replay, first)
@ -82,7 +112,7 @@ func TestStarsFriendGiftPurchaseAtomicReplayAndValidationPostgres(t *testing.T)
if err := pool.QueryRow(ctx, "SELECT count(*) FROM stars_transactions WHERE user_id=$1 AND reason='gift'", recipient.ID).Scan(&txnCount); err != nil {
t.Fatal(err)
}
if err := pool.QueryRow(ctx, "SELECT count(*) FROM stars_gift_purchase_commands WHERE buyer_user_id=$1", buyer.ID).Scan(&commandCount); err != nil {
if err := pool.QueryRow(ctx, "SELECT count(*) FROM stars_purchase_commands WHERE buyer_user_id=$1", buyer.ID).Scan(&commandCount); err != nil {
t.Fatal(err)
}
if balance != 2500 || txnCount != 1 || commandCount != 1 {
@ -103,11 +133,11 @@ func TestStarsFriendGiftPurchaseAtomicReplayAndValidationPostgres(t *testing.T)
tampered := req
tampered.Amount++
if _, err := store.PurchaseStarsGift(ctx, tampered); !errors.Is(err, domain.ErrStarsGiftFormInvalid) {
if _, err := store.PurchaseStars(ctx, tampered); !errors.Is(err, domain.ErrStarsPurchaseFormInvalid) {
t.Fatalf("tampered replay err=%v", err)
}
expired, err := store.IssueStarsGiftPurchaseForm(ctx, domain.StarsGiftPurchaseForm{
BuyerUserID: buyer.ID, RecipientUserID: recipient.ID,
expired, err := store.IssueStarsPurchaseForm(ctx, domain.StarsPurchaseForm{
Kind: domain.StarsPurchaseGift, BuyerUserID: buyer.ID, RecipientUserID: recipient.ID,
Stars: 1000, Currency: "USD", Amount: 99,
IssuedAt: 1_699_999_000, ExpiresAt: 1_699_999_600,
})
@ -116,10 +146,221 @@ func TestStarsFriendGiftPurchaseAtomicReplayAndValidationPostgres(t *testing.T)
}
expiredReq := req
expiredReq.FormID, expiredReq.Stars, expiredReq.Amount = expired.FormID, 1000, 99
if _, err := store.PurchaseStarsGift(ctx, expiredReq); !errors.Is(err, domain.ErrStarsGiftFormExpired) {
if _, err := store.PurchaseStars(ctx, expiredReq); !errors.Is(err, domain.ErrStarsPurchaseFormExpired) {
t.Fatalf("expired form err=%v", err)
}
if err := pool.QueryRow(ctx, "SELECT balance FROM stars_balances WHERE user_id=$1", recipient.ID).Scan(&balance); err != nil || balance != 2500 {
t.Fatalf("balance after failures=%d err=%v", balance, err)
}
}
func TestStarsTopupPurchaseAtomicReplayAndPurposeBindingPostgres(t *testing.T) {
pool := testPool(t)
ctx := context.Background()
users := NewUserStore(pool)
suffix := randomSuffix(t)
buyer, err := users.Create(ctx, domain.User{AccessHash: 94201, Phone: "+1665942" + suffix + "01", FirstName: "TopupBuyer"})
if err != nil {
t.Fatalf("create buyer: %v", err)
}
t.Cleanup(func() {
_, _ = pool.Exec(ctx, "DELETE FROM stars_purchase_commands WHERE buyer_user_id=$1", buyer.ID)
_, _ = pool.Exec(ctx, "DELETE FROM stars_purchase_forms WHERE buyer_user_id=$1", buyer.ID)
_, _ = pool.Exec(ctx, "DELETE FROM stars_transactions WHERE user_id=$1", buyer.ID)
_, _ = pool.Exec(ctx, "DELETE FROM stars_balances WHERE user_id=$1", buyer.ID)
_, _ = pool.Exec(ctx, "DELETE FROM users WHERE id=$1", buyer.ID)
})
store := NewStarsPurchaseStore(pool, nil)
purposePeer := domain.Peer{Type: domain.PeerTypeUser, ID: buyer.ID + 100}
issued, err := store.IssueStarsPurchaseForm(ctx, domain.StarsPurchaseForm{
Kind: domain.StarsPurchaseTopup, BuyerUserID: buyer.ID, SpendPurposePeer: purposePeer,
Stars: 2500, Currency: "USD", Amount: 199,
IssuedAt: 1_700_000_000, ExpiresAt: 1_700_000_600,
})
if err != nil || issued.FormID == 0 {
t.Fatalf("issue form = %+v err=%v", issued, err)
}
req := domain.StarsPurchaseRequest{
StarsPurchaseForm: domain.StarsPurchaseForm{
FormID: issued.FormID, Kind: domain.StarsPurchaseTopup, BuyerUserID: buyer.ID,
SpendPurposePeer: purposePeer, Stars: 2500, Currency: "USD", Amount: 199,
},
Date: 1_700_000_100,
}
first, replay := purchaseStarsTwiceConcurrently(t, ctx, store, req)
if first.Duplicate || first.Balance.Balance != 2500 || first.TransactionID == "" {
t.Fatalf("first purchase = %+v", first)
}
if !replay.Duplicate || replay.Balance.Balance != 2500 || replay.TransactionID != first.TransactionID {
t.Fatalf("replay = %+v, first=%+v", replay, first)
}
var balance, txnCount, commandCount int64
if err := pool.QueryRow(ctx, "SELECT balance FROM stars_balances WHERE user_id=$1", buyer.ID).Scan(&balance); err != nil {
t.Fatal(err)
}
if err := pool.QueryRow(ctx, "SELECT count(*) FROM stars_transactions WHERE user_id=$1 AND reason='topup'", buyer.ID).Scan(&txnCount); err != nil {
t.Fatal(err)
}
if err := pool.QueryRow(ctx, "SELECT count(*) FROM stars_purchase_commands WHERE buyer_user_id=$1", buyer.ID).Scan(&commandCount); err != nil {
t.Fatal(err)
}
if balance != 2500 || txnCount != 1 || commandCount != 1 {
t.Fatalf("replay footprint balance=%d txns=%d commands=%d", balance, txnCount, commandCount)
}
tampered := req
tampered.SpendPurposePeer.ID++
if _, err := store.PurchaseStars(ctx, tampered); !errors.Is(err, domain.ErrStarsPurchaseFormInvalid) {
t.Fatalf("tampered purpose replay err=%v", err)
}
otherBuyer := req
otherBuyer.BuyerUserID++
if _, err := store.PurchaseStars(ctx, otherBuyer); !errors.Is(err, domain.ErrStarsPurchaseFormInvalid) {
t.Fatalf("cross-account form err=%v", err)
}
if err := pool.QueryRow(ctx, "SELECT balance FROM stars_balances WHERE user_id=$1", buyer.ID).Scan(&balance); err != nil || balance != 2500 {
t.Fatalf("balance after invalid submissions=%d err=%v", balance, err)
}
}
func TestStarsGiveawayPurchaseAtomicChannelPTSReplayAndInfoPostgres(t *testing.T) {
pool := testPool(t)
ctx := context.Background()
users := NewUserStore(pool)
suffix := randomSuffix(t)
owner, err := users.Create(ctx, domain.User{AccessHash: 94301, Phone: "+1665943" + suffix + "01", FirstName: "GiveawayOwner", CountryCode: "1"})
if err != nil {
t.Fatalf("create owner: %v", err)
}
member, err := users.Create(ctx, domain.User{AccessHash: 94302, Phone: "+1665943" + suffix + "02", FirstName: "GiveawayMember", CountryCode: "1"})
if err != nil {
t.Fatalf("create member: %v", err)
}
channels := NewChannelStore(pool)
created, err := channels.CreateChannel(ctx, domain.CreateChannelRequest{
CreatorUserID: owner.ID, Title: "Stars Giveaway " + suffix, Megagroup: true,
MemberUserIDs: []int64{member.ID}, Date: 1_700_000_000,
})
if err != nil {
t.Fatalf("create channel: %v", err)
}
channelID := created.Channel.ID
t.Cleanup(func() {
_, _ = pool.Exec(ctx, "DELETE FROM stars_giveaways WHERE buyer_user_id=$1", owner.ID)
_, _ = pool.Exec(ctx, "DELETE FROM stars_purchase_commands WHERE buyer_user_id=$1", owner.ID)
_, _ = pool.Exec(ctx, "DELETE FROM stars_purchase_forms WHERE buyer_user_id=$1", owner.ID)
_, _ = pool.Exec(ctx, "DELETE FROM channels WHERE id=$1", channelID)
_, _ = pool.Exec(ctx, "DELETE FROM users WHERE id=ANY($1::bigint[])", []int64{owner.ID, member.ID})
})
before, err := channels.GetChannelByID(ctx, channelID)
if err != nil {
t.Fatal(err)
}
purpose := &domain.StarsGiveawayPurchase{
BoostPeer: domain.Peer{Type: domain.PeerTypeChannel, ID: channelID},
CountriesISO2: []string{"US"}, RandomID: 9430001, UntilDate: 1_700_003_700,
Users: 2, PerUserStars: 500, YearlyBoosts: 4, WinnersAreVisible: true,
}
store := NewStarsPurchaseStore(pool, nil, channels)
issued, err := store.IssueStarsPurchaseForm(ctx, domain.StarsPurchaseForm{
Kind: domain.StarsPurchaseGiveaway, BuyerUserID: owner.ID, Giveaway: purpose,
Stars: 1000, Currency: "USD", Amount: 99, IssuedAt: 1_700_000_100, ExpiresAt: 1_700_000_700,
})
if err != nil || issued.FormID == 0 {
t.Fatalf("issue giveaway form=%+v err=%v", issued, err)
}
req := domain.StarsPurchaseRequest{StarsPurchaseForm: domain.StarsPurchaseForm{
FormID: issued.FormID, Kind: domain.StarsPurchaseGiveaway, BuyerUserID: owner.ID, Giveaway: purpose,
Stars: 1000, Currency: "USD", Amount: 99,
}, Date: 1_700_000_200}
first, replay := purchaseStarsTwiceConcurrently(t, ctx, store, req)
if first.Duplicate || first.TransactionID == "" || first.ChannelSend.Event.PtsCount != 1 ||
first.ChannelSend.Event.Pts != before.Pts+1 || first.ChannelSend.Message.Media == nil ||
first.ChannelSend.Message.Media.Giveaway == nil {
t.Fatalf("first giveaway result=%+v before_pts=%d", first, before.Pts)
}
media := first.ChannelSend.Message.Media.Giveaway
if media.Stars != 1000 || media.Quantity != 2 || len(media.Channels) != 1 || media.Channels[0] != channelID ||
media.UntilDate != purpose.UntilDate || !media.WinnersAreVisible {
t.Fatalf("giveaway media=%+v", media)
}
difference, err := channels.ListChannelDifference(ctx, domain.ChannelDifferenceRequest{
UserID: member.ID, ChannelID: channelID, Pts: before.Pts, Limit: 10,
})
if err != nil || difference.Pts != first.ChannelSend.Event.Pts || len(difference.Events) != 1 || len(difference.NewMessages) != 1 ||
difference.NewMessages[0].Media == nil || difference.NewMessages[0].Media.Giveaway == nil ||
difference.NewMessages[0].Media.Giveaway.Stars != 1000 {
t.Fatalf("giveaway channel difference=%+v err=%v", difference, err)
}
if !replay.Duplicate || replay.TransactionID != first.TransactionID ||
replay.ChannelSend.Message.ID != first.ChannelSend.Message.ID || replay.ChannelSend.Event.Pts != first.ChannelSend.Event.Pts {
t.Fatalf("giveaway replay=%+v first=%+v", replay, first)
}
lateReplayReq := req
lateReplayReq.Date = purpose.UntilDate
lateReplay, err := store.PurchaseStars(ctx, lateReplayReq)
if err != nil || !lateReplay.Duplicate || lateReplay.TransactionID != first.TransactionID ||
lateReplay.ChannelSend.Message.ID != first.ChannelSend.Message.ID || lateReplay.ChannelSend.Event.Pts != first.ChannelSend.Event.Pts {
t.Fatalf("giveaway replay after until_date=%+v err=%v first=%+v", lateReplay, err, first)
}
latePurpose := *purpose
latePurpose.RandomID++
lateForm, err := store.IssueStarsPurchaseForm(ctx, domain.StarsPurchaseForm{
Kind: domain.StarsPurchaseGiveaway, BuyerUserID: owner.ID, Giveaway: &latePurpose,
Stars: 1000, Currency: "USD", Amount: 99,
IssuedAt: purpose.UntilDate - 100, ExpiresAt: purpose.UntilDate + 500,
})
if err != nil {
t.Fatalf("issue giveaway form before until_date: %v", err)
}
lateFirstReq := req
lateFirstReq.FormID = lateForm.FormID
lateFirstReq.Giveaway = &latePurpose
lateFirstReq.Date = purpose.UntilDate
if _, err := store.PurchaseStars(ctx, lateFirstReq); !errors.Is(err, domain.ErrStarsPurchaseFormExpired) {
t.Fatalf("first giveaway settlement at until_date err=%v, want form expired", err)
}
var campaigns, commands, messages, events, balanceRows, txns int64
queries := []struct {
query string
args []any
target *int64
}{
{"SELECT count(*) FROM stars_giveaways WHERE buyer_user_id=$1", []any{owner.ID}, &campaigns},
{"SELECT count(*) FROM stars_purchase_commands WHERE buyer_user_id=$1", []any{owner.ID}, &commands},
{"SELECT count(*) FROM channel_messages WHERE channel_id=$1 AND id=$2", []any{channelID, first.ChannelSend.Message.ID}, &messages},
{"SELECT count(*) FROM channel_update_events WHERE channel_id=$1 AND pts=$2", []any{channelID, first.ChannelSend.Event.Pts}, &events},
{"SELECT count(*) FROM stars_balances WHERE user_id=$1", []any{owner.ID}, &balanceRows},
{"SELECT count(*) FROM stars_transactions WHERE user_id=$1", []any{owner.ID}, &txns},
}
for _, item := range queries {
if err := pool.QueryRow(ctx, item.query, item.args...).Scan(item.target); err != nil {
t.Fatalf("footprint query %q: %v", item.query, err)
}
}
if campaigns != 1 || commands != 1 || messages != 1 || events != 1 || balanceRows != 0 || txns != 0 {
t.Fatalf("footprint campaigns=%d commands=%d messages=%d events=%d balances=%d txns=%d", campaigns, commands, messages, events, balanceRows, txns)
}
ownerInfo, err := store.GetStarsGiveawayInfo(ctx, owner.ID, channelID, first.ChannelSend.Message.ID, 1_700_000_300)
if err != nil || ownerInfo.AdminDisallowedChatID != channelID || ownerInfo.Participating {
t.Fatalf("owner giveaway info=%+v err=%v", ownerInfo, err)
}
memberInfo, err := store.GetStarsGiveawayInfo(ctx, member.ID, channelID, first.ChannelSend.Message.ID, 1_700_000_300)
if err != nil || !memberInfo.Participating || memberInfo.StartDate != req.Date {
t.Fatalf("member giveaway info=%+v err=%v", memberInfo, err)
}
preparing, err := store.GetStarsGiveawayInfo(ctx, member.ID, channelID, first.ChannelSend.Message.ID, purpose.UntilDate)
if err != nil || !preparing.PreparingResults || preparing.Participating {
t.Fatalf("preparing giveaway info=%+v err=%v", preparing, err)
}
tampered := req
changed := *purpose
changed.Users, changed.PerUserStars = 1, 1000
tampered.Giveaway = &changed
if _, err := store.PurchaseStars(ctx, tampered); !errors.Is(err, domain.ErrStarsPurchaseFormInvalid) {
t.Fatalf("tampered giveaway replay err=%v", err)
}
}