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
	CreatedAt   time.Time
}

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)。amountMinor/currency 是下单时的
// **展示预估**(按支付方式的结算币种,见 DisplayAmountMinor),webhook 到账后由
// MarkPaidTx 用实际结算金额覆盖——保证订单列表/详情在支付前也有金额可显。
func (s *Store) Insert(ctx context.Context, userID int64, bizRef, sku, outTradeNo, method string, amountMinor int64, currency 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, amount_minor, currency, created_at, updated_at)
		 VALUES (?, ?, ?, ?, ?, 'created', ?, ?, ?, ?)`,
		userID, bizRef, sku, outTradeNo, method, amountMinor, currency, 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, created_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, &p.CreatedAt); err != nil {
		return nil, err
	}
	return &p, nil
}

// ListByUser 按用户列历史订单(created_at 倒序,复用 idx_pay_user 索引)。
// 供「订单列表」页;limit 上限保护。
func (s *Store) ListByUser(ctx context.Context, userID int64, limit int) ([]PurchaseRow, error) {
	if limit <= 0 || limit > 200 {
		limit = 100
	}
	rows, err := s.db.QueryContext(ctx,
		`SELECT `+purchaseCols+` FROM pay_purchases WHERE user_id = ? ORDER BY created_at DESC, id DESC LIMIT ?`,
		userID, limit)
	if err != nil {
		return nil, fmt.Errorf("pay.Store.ListByUser: %w", err)
	}
	defer rows.Close()
	var out []PurchaseRow
	for rows.Next() {
		var p PurchaseRow
		if err := rows.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, &p.CreatedAt); err != nil {
			return nil, fmt.Errorf("pay.Store.ListByUser scan: %w", err)
		}
		out = append(out, p)
	}
	return out, rows.Err()
}

// 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 修复)。
// 注意:webhookEvent payload 无 method 字段(用户选的支付方式,如 alipay/wxpay)可复原,
// 只有 channel(结算渠道);method 留空,不能拿 channel 冒充——语义不同,避免台账观感误导。
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, 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
}

// HasPaidPurchase 该用户是否已有该 SKU 的 paid 台账——供优惠档(Promo SKU)限购一次校验用
// (CreateOrder 在调上游前查,已购则 409 拒单,不打上游)。
func (s *Store) HasPaidPurchase(ctx context.Context, userID int64, sku string) (bool, error) {
	var n int
	err := s.db.QueryRowContext(ctx,
		`SELECT COUNT(*) FROM pay_purchases WHERE user_id = ? AND sku = ? AND status = 'paid'`,
		userID, sku).Scan(&n)
	if err != nil {
		return false, fmt.Errorf("pay.Store.HasPaidPurchase: %w", err)
	}
	return n > 0, nil
}

// HasPaidPurchaseExcludingTx 是 HasPaidPurchase 的事务内(锁行后)复查版本,供
// webhook.go settle 在开通前对 Promo SKU 再查一次——CreateOrder 那次下单时的
// 检查是裸 SELECT 无锁,并发/多挂起单可绕过(TOCTOU)。excludeID 排除本单自己
// (本单尚未 MarkPaidTx,通常不会自匹配,但显式排除更稳妥、也便于未来复用)。
func (s *Store) HasPaidPurchaseExcludingTx(ctx context.Context, tx *sql.Tx, userID int64, sku string, excludeID int64) (bool, error) {
	var n int
	err := tx.QueryRowContext(ctx,
		`SELECT COUNT(*) FROM pay_purchases WHERE user_id = ? AND sku = ? AND status = 'paid' AND id <> ?`,
		userID, sku, excludeID).Scan(&n)
	if err != nil {
		return false, fmt.Errorf("pay.Store.HasPaidPurchaseExcludingTx: %w", err)
	}
	return n > 0, nil
}

// MarkDuplicatePromoTx 收口一笔在 TOCTOU 竞争中"输"掉的 Promo SKU 重复单——
// 另一笔同 user+SKU 的订单已先一步 settle 为 paid(见 webhook.go settle 的
// promo 复查)。刻意标记为 'canceled' 而非 'paid':(user_id, sku) WHERE
// status='paid' 是部分唯一索引(migration 000027,仅 sqlite)本就不允许同一
// user+promo-SKU 出现第二条 paid 行,这里若也写 paid 会在 sqlite 上直接撞
// 约束报错、把整个 webhook 打成 500 引发 pay 无限重投——与"吞掉重复单,不
// 再重投"的目标相反。仍落一次结算回执字段(amount/currency/channel/paid_at)
// 供人工核对"钱是否真收到过、为何没有二次开通",不同于用户主动取消未付
// 单的语义(MarkCanceled 的原生用途),但复用同一 status 取值。
func (s *Store) MarkDuplicatePromoTx(ctx context.Context, tx *sql.Tx, id int64, amountMinor int64, currency, channel string, paidAt time.Time) error {
	_, err := tx.ExecContext(ctx,
		`UPDATE pay_purchases SET status = 'canceled', amount_minor = ?, currency = ?,
		        channel = ?, paid_at = ?, updated_at = ?
		 WHERE id = ?`,
		amountMinor, currency, channel, paidAt, time.Now().UTC(), id)
	if err != nil {
		return fmt.Errorf("pay.Store.MarkDuplicatePromoTx: %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
}