pangolin/server/internal/pay/store.go

package pay

import (
	"context"
	"database/sql"
	"fmt"
	"time"

	dbx "github.com/wangjia/pangolin/server/internal/db"
)

// PurchaseRow 是 pay_purchases 一行:biz_ref↔out_trade_no 映射 + webhook 幂等台账。
type PurchaseRow struct {
	ID          int64
	UserID      int64
	BizRef      string
	SKU         string
	OutTradeNo  string
	Method      string
	Status      string // created | paid | canceled
	AmountMinor int64
	Currency    string
	Channel     string
	SubID       sql.NullInt64
	PaidAt      sql.NullTime
}

type Store struct {
	db      *sql.DB
	dialect dbx.Dialect
}

func NewStore(db *sql.DB) *Store {
	return &Store{db: db, dialect: dbx.DialectForDB(db)}
}

func (s *Store) BeginTx(ctx context.Context) (*sql.Tx, error) {
	return s.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted})
}

// Insert 下单成功后落台账(status=created)。
func (s *Store) Insert(ctx context.Context, userID int64, bizRef, sku, outTradeNo, method string) error {
	now := time.Now().UTC()
	_, err := s.db.ExecContext(ctx,
		`INSERT INTO pay_purchases (user_id, biz_ref, sku, out_trade_no, method, status, created_at, updated_at)
		 VALUES (?, ?, ?, ?, ?, 'created', ?, ?)`,
		userID, bizRef, sku, outTradeNo, method, now, now)
	if err != nil {
		return fmt.Errorf("pay.Store.Insert: %w", err)
	}
	return nil
}

const purchaseCols = `id, user_id, biz_ref, sku, out_trade_no, method, status,
	amount_minor, currency, channel, sub_id, paid_at`

func scanPurchase(row *sql.Row) (*PurchaseRow, error) {
	var p PurchaseRow
	if err := row.Scan(&p.ID, &p.UserID, &p.BizRef, &p.SKU, &p.OutTradeNo, &p.Method,
		&p.Status, &p.AmountMinor, &p.Currency, &p.Channel, &p.SubID, &p.PaidAt); err != nil {
		return nil, err
	}
	return &p, nil
}

// GetForUser 按 (userID, outTradeNo) 取行——所有权校验由查询本身完成。
func (s *Store) GetForUser(ctx context.Context, userID int64, outTradeNo string) (*PurchaseRow, error) {
	return scanPurchase(s.db.QueryRowContext(ctx,
		`SELECT `+purchaseCols+` FROM pay_purchases WHERE user_id = ? AND out_trade_no = ?`,
		userID, outTradeNo))
}

// LockByOutTradeNoTx 事务内锁行(mysql FOR UPDATE;sqlite 空后缀,靠
// _txlock=immediate 串行化——与 codes 兑换同一套悲观语义)。
func (s *Store) LockByOutTradeNoTx(ctx context.Context, tx *sql.Tx, outTradeNo string) (*PurchaseRow, error) {
	q := `SELECT ` + purchaseCols + ` FROM pay_purchases WHERE out_trade_no = ? ` + s.dialect.LockForUpdate()
	return scanPurchase(tx.QueryRowContext(ctx, q, outTradeNo))
}

// InsertFromWebhookTx 兜底补台账(下单后本地写失败的孤儿单,webhook 按 biz_ref 修复)。
func (s *Store) InsertFromWebhookTx(ctx context.Context, tx *sql.Tx, userID int64, bizRef, sku, outTradeNo, channel string) (int64, error) {
	now := time.Now().UTC()
	res, err := tx.ExecContext(ctx,
		`INSERT INTO pay_purchases (user_id, biz_ref, sku, out_trade_no, method, status, channel, created_at, updated_at)
		 VALUES (?, ?, ?, ?, ?, 'created', ?, ?, ?)`,
		userID, bizRef, sku, outTradeNo, channel, channel, now, now)
	if err != nil {
		return 0, fmt.Errorf("pay.Store.InsertFromWebhookTx: %w", err)
	}
	id, _ := res.LastInsertId()
	return id, nil
}

// MarkPaidTx 台账翻转 →paid 并回填结算信息(幂等判定已在锁内完成,直写)。
func (s *Store) MarkPaidTx(ctx context.Context, tx *sql.Tx, id int64, amountMinor int64, currency, channel string, subID int64, paidAt time.Time) error {
	_, err := tx.ExecContext(ctx,
		`UPDATE pay_purchases SET status = 'paid', amount_minor = ?, currency = ?,
		        channel = ?, sub_id = ?, paid_at = ?, updated_at = ?
		 WHERE id = ?`,
		amountMinor, currency, channel, subID, paidAt, time.Now().UTC(), id)
	if err != nil {
		return fmt.Errorf("pay.Store.MarkPaidTx: %w", err)
	}
	return nil
}

// UpdateMethod retry 换渠道成功后同步台账(仅未支付单)。
func (s *Store) UpdateMethod(ctx context.Context, userID int64, outTradeNo, method string) error {
	_, err := s.db.ExecContext(ctx,
		`UPDATE pay_purchases SET method = ?, updated_at = ?
		 WHERE user_id = ? AND out_trade_no = ? AND status = 'created'`,
		method, time.Now().UTC(), userID, outTradeNo)
	if err != nil {
		return fmt.Errorf("pay.Store.UpdateMethod: %w", err)
	}
	return nil
}

// MarkCanceled 仅未支付单可取消(paid 行不动——钱已收,开通不回退)。
func (s *Store) MarkCanceled(ctx context.Context, userID int64, outTradeNo string) error {
	_, err := s.db.ExecContext(ctx,
		`UPDATE pay_purchases SET status = 'canceled', updated_at = ?
		 WHERE user_id = ? AND out_trade_no = ? AND status = 'created'`,
		time.Now().UTC(), userID, outTradeNo)
	if err != nil {
		return fmt.Errorf("pay.Store.MarkCanceled: %w", err)
	}
	return nil
}

// SubscriptionExpiry 查开通行的到期时间(查单响应回带给客户端)。
func (s *Store) SubscriptionExpiry(ctx context.Context, subID int64) (time.Time, error) {
	var exp time.Time
	err := s.db.QueryRowContext(ctx,
		`SELECT expires_at FROM subscriptions WHERE id = ?`, subID).Scan(&exp)
	return exp, err
}