feat(server): /v1/pay 下单代理端点 + 购买台账(JWT 鉴权,user→biz_ref 映射)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_013nMthbVEmQquxBRKb9Fj8u

server/cmd/server/main.go

@@ -33,6 +33,7 @@ import (
 	"github.com/wangjia/pangolin/server/internal/httpapi"
 	"github.com/wangjia/pangolin/server/internal/mtls"
 	"github.com/wangjia/pangolin/server/internal/nodes"
+	"github.com/wangjia/pangolin/server/internal/pay"
 	agentv1 "github.com/wangjia/pangolin/server/internal/pb/agentv1"
 	"github.com/wangjia/pangolin/server/internal/provision"
 	"github.com/wangjia/pangolin/server/internal/provision/providers"
@@ -299,6 +300,18 @@ func mountV1(r chi.Router, sqlDB *sql.DB, rdb *redis.Client, nodeSvc *nodes.Serv
 	webhookHandler := codes.NewWebhookHandler(codesStore, rdb,
 		os.Getenv("WEBHOOK_SECRET"), 5*time.Minute, 15*time.Minute)
 
+	// ── Pay(pay v2 统一支付网关;PAY_BASE_URL 未配则整组不挂载)──────────────
+	var payHandler *pay.Handler
+	if payBase := os.Getenv("PAY_BASE_URL"); payBase != "" {
+		paySystem := getenvDefault("PAY_BIZ_SYSTEM", "pangolin")
+		paySecret := os.Getenv("PAY_BIZ_SECRET")
+		payClient := pay.NewClient(payBase, paySystem, paySecret)
+		payStore := pay.NewStore(sqlDB)
+		payHandler = pay.NewHandler(payClient, payStore, sqlDB)
+	} else {
+		log.Printf("PAY_BASE_URL 未配置 — /v1/pay 支付端点不挂载")
+	}
+
 	// ── Usage ─────────────────────────────────────────────────────────────────
 	usageStore := usage.NewStore(sqlDB)
 	usageSvc := usage.NewService(usageStore, rdb, nil, time.Hour)
@@ -386,6 +399,13 @@ func mountV1(r chi.Router, sqlDB *sql.DB, rdb *redis.Client, nodeSvc *nodes.Serv
 				protected.Post("/ads/unlock", adsHandler.ServeHTTP)
 				protected.Get("/plans", accountAPI.ListPlans)
 				protected.Get("/notices", accountAPI.ListNotices)
+				if payHandler != nil {
+					protected.Get("/pay/catalog", payHandler.Catalog)
+					protected.Post("/pay/orders", payHandler.CreateOrder)
+					protected.Get("/pay/orders/{orderNo}", payHandler.GetOrder)
+					protected.Post("/pay/orders/{orderNo}/retry", payHandler.Retry)
+					protected.Post("/pay/orders/{orderNo}/cancel", payHandler.Cancel)
+				}
 
 				if nodeAPI != nil {
 					protected.Get("/nodes", nodeAPI.ListNodes)

server/internal/pay/catalog.go

@@ -0,0 +1,30 @@
+package pay
+
+import "github.com/wangjia/pangolin/server/internal/codes"
+
+// CatalogItem 是可购档位。SKU 与 pay 侧 products.biz_code 一一对应
+// (webhook product_biz_code 原样回带);PriceMinor 仅展示(CNY 分),
+// 实际扣款以 pay 侧 ProductPrice/price 为准——一致性列入联调 checklist。
+type CatalogItem struct {
+	SKU        string `json:"sku"`
+	Plan       string `json:"plan"`
+	Days       int    `json:"days"`
+	PriceMinor int64  `json:"price_minor"`
+	Currency   string `json:"currency"`
+}
+
+// Catalog 三档单源。时长取宽松口径(31/92/366 覆盖大月与最长季)。
+var Catalog = []CatalogItem{
+	{SKU: "pro_month", Plan: string(codes.PlanPro), Days: 31, PriceMinor: 2999, Currency: "CNY"},
+	{SKU: "pro_quarter", Plan: string(codes.PlanPro), Days: 92, PriceMinor: 6888, Currency: "CNY"},
+	{SKU: "pro_year", Plan: string(codes.PlanPro), Days: 366, PriceMinor: 19999, Currency: "CNY"},
+}
+
+func CatalogBySKU(sku string) (CatalogItem, bool) {
+	for _, it := range Catalog {
+		if it.SKU == sku {
+			return it, true
+		}
+	}
+	return CatalogItem{}, false
+}

server/internal/pay/handler.go

@@ -0,0 +1,241 @@
+package pay
+
+import (
+	"database/sql"
+	"encoding/json"
+	"errors"
+	"io"
+	"log/slog"
+	"net/http"
+	"time"
+
+	"github.com/go-chi/chi/v5"
+
+	"github.com/wangjia/pangolin/server/internal/apierr"
+	"github.com/wangjia/pangolin/server/internal/auth"
+)
+
+// Handler 是面向 App 的下单代理(JWT 保护;user→biz_ref 映射在 server 侧,
+// 客户端只传 sku+method+端型 metadata,永远不传金额)。
+type Handler struct {
+	client *Client
+	store  *Store
+	db     *sql.DB
+}
+
+func NewHandler(client *Client, store *Store, db *sql.DB) *Handler {
+	return &Handler{client: client, store: store, db: db}
+}
+
+// allowedMetadataKeys 与 pay gateway.go 白名单一致(is_mobile/render)。
+var allowedMetadataKeys = map[string]bool{"is_mobile": true, "render": true}
+
+func filterMetadata(in map[string]string) map[string]string {
+	if len(in) == 0 {
+		return nil
+	}
+	out := make(map[string]string, len(in))
+	for k, v := range in {
+		if allowedMetadataKeys[k] {
+			out[k] = v
+		}
+	}
+	if len(out) == 0 {
+		return nil
+	}
+	return out
+}
+
+func writeJSON(w http.ResponseWriter, v any) {
+	w.Header().Set("Content-Type", "application/json")
+	_ = json.NewEncoder(w).Encode(v)
+}
+
+// writePayErr 把 pay 网关错误映射为 apierr(脱敏双语;code 透传供客户端分辨)。
+func writePayErr(w http.ResponseWriter, err error) {
+	var pe *Error
+	if !errors.As(err, &pe) {
+		apierr.WriteJSON(w, http.StatusBadGateway,
+			apierr.New("PAY_UPSTREAM", "支付服务暂不可用，请稍后重试", "Payment service unavailable, please retry later"))
+		return
+	}
+	switch pe.Code {
+	case "currency_mismatch":
+		apierr.WriteJSON(w, http.StatusConflict,
+			apierr.New("CURRENCY_MISMATCH", "该支付方式结算币种与订单不符，请重新下单", "Settlement currency mismatch, please create a new order"))
+	case "order_not_pending":
+		apierr.WriteJSON(w, http.StatusConflict,
+			apierr.New("ORDER_NOT_PENDING", "订单状态已变化，请刷新后重试", "Order is no longer pending"))
+	case "order_not_found", "product_not_found":
+		apierr.WriteJSON(w, http.StatusNotFound, apierr.ErrNotFound)
+	case "rate_limited":
+		apierr.WriteJSON(w, http.StatusTooManyRequests, apierr.ErrRateLimited)
+	case "unknown_method", "bad_request", "method_not_recurring":
+		apierr.WriteJSON(w, http.StatusBadRequest, apierr.ErrBadRequest)
+	default: // no_account / no_settle_currency / create_failed / upstream_error…
+		apierr.WriteJSON(w, http.StatusBadGateway,
+			apierr.New("PAY_UPSTREAM", "支付服务暂不可用，请稍后重试", "Payment service unavailable, please retry later"))
+	}
+}
+
+// ─── GET /v1/pay/catalog ────────────────────────────────────────────────────
+
+func (h *Handler) Catalog(w http.ResponseWriter, _ *http.Request) {
+	writeJSON(w, map[string]any{"items": Catalog})
+}
+
+// ─── POST /v1/pay/orders ────────────────────────────────────────────────────
+
+type createOrderRequest struct {
+	SKU      string            `json:"sku"`
+	Method   string            `json:"method"`
+	Metadata map[string]string `json:"metadata,omitempty"`
+}
+
+type sessionResponse struct {
+	OrderNo string  `json:"order_no"`
+	Session Session `json:"session"`
+}
+
+func (h *Handler) CreateOrder(w http.ResponseWriter, r *http.Request) {
+	ctx := r.Context()
+	uid, ok := auth.UserIDFromContext(ctx)
+	if !ok {
+		apierr.WriteJSON(w, http.StatusUnauthorized, apierr.ErrUnauthorized)
+		return
+	}
+	var req createOrderRequest
+	if err := json.NewDecoder(io.LimitReader(r.Body, 16<<10)).Decode(&req); err != nil {
+		apierr.WriteJSON(w, http.StatusBadRequest, apierr.ErrBadRequest)
+		return
+	}
+	if _, ok := CatalogBySKU(req.SKU); !ok || req.Method == "" {
+		apierr.WriteJSON(w, http.StatusBadRequest, apierr.ErrBadRequest)
+		return
+	}
+	var bizRef string
+	err := h.db.QueryRowContext(ctx,
+		`SELECT uuid FROM users WHERE id = ? AND status = 'active'`, uid).Scan(&bizRef)
+	if err == sql.ErrNoRows {
+		apierr.WriteJSON(w, http.StatusNotFound, apierr.ErrNotFound)
+		return
+	} else if err != nil {
+		apierr.WriteJSON(w, http.StatusInternalServerError, apierr.ErrInternal)
+		return
+	}
+
+	res, err := h.client.CreateOrder(ctx, req.SKU, req.Method, bizRef, filterMetadata(req.Metadata))
+	if err != nil {
+		writePayErr(w, err)
+		return
+	}
+	if err := h.store.Insert(ctx, uid, bizRef, req.SKU, res.OrderNo, req.Method); err != nil {
+		// 不吞单:webhook 会按 biz_ref 兜底补台账,这里记日志便于追查。
+		slog.Error("pay: 台账写入失败(webhook 将按 biz_ref 兜底)", "order_no", res.OrderNo, "err", err)
+	}
+	writeJSON(w, sessionResponse{OrderNo: res.OrderNo, Session: res.Session})
+}
+
+// ─── GET /v1/pay/orders/{orderNo} ───────────────────────────────────────────
+
+type orderStatusResponse struct {
+	OrderNo   string  `json:"order_no"`
+	PayStatus string  `json:"pay_status"` // pay 侧状态词汇原样透传
+	Activated bool    `json:"activated"`  // 本地台账已消费(权益已开通)——客户端轮询以此为成功判据
+	ExpiresAt *string `json:"expires_at,omitempty"`
+}
+
+func (h *Handler) GetOrder(w http.ResponseWriter, r *http.Request) {
+	ctx := r.Context()
+	uid, ok := auth.UserIDFromContext(ctx)
+	if !ok {
+		apierr.WriteJSON(w, http.StatusUnauthorized, apierr.ErrUnauthorized)
+		return
+	}
+	orderNo := chi.URLParam(r, "orderNo")
+	row, err := h.store.GetForUser(ctx, uid, orderNo)
+	if err == sql.ErrNoRows {
+		apierr.WriteJSON(w, http.StatusNotFound, apierr.ErrNotFound)
+		return
+	} else if err != nil {
+		apierr.WriteJSON(w, http.StatusInternalServerError, apierr.ErrInternal)
+		return
+	}
+	st, err := h.client.GetOrder(ctx, orderNo)
+	if err != nil {
+		writePayErr(w, err)
+		return
+	}
+	resp := orderStatusResponse{OrderNo: orderNo, PayStatus: st.Status, Activated: row.Status == "paid"}
+	if row.SubID.Valid {
+		if exp, err := h.store.SubscriptionExpiry(ctx, row.SubID.Int64); err == nil {
+			s := exp.UTC().Format(time.RFC3339)
+			resp.ExpiresAt = &s
+		}
+	}
+	writeJSON(w, resp)
+}
+
+// ─── POST /v1/pay/orders/{orderNo}/retry ────────────────────────────────────
+
+type retryOrderRequest struct {
+	Method   string            `json:"method"`
+	Metadata map[string]string `json:"metadata,omitempty"`
+}
+
+func (h *Handler) Retry(w http.ResponseWriter, r *http.Request) {
+	ctx := r.Context()
+	uid, ok := auth.UserIDFromContext(ctx)
+	if !ok {
+		apierr.WriteJSON(w, http.StatusUnauthorized, apierr.ErrUnauthorized)
+		return
+	}
+	orderNo := chi.URLParam(r, "orderNo")
+	var req retryOrderRequest
+	if err := json.NewDecoder(io.LimitReader(r.Body, 16<<10)).Decode(&req); err != nil || req.Method == "" {
+		apierr.WriteJSON(w, http.StatusBadRequest, apierr.ErrBadRequest)
+		return
+	}
+	if _, err := h.store.GetForUser(ctx, uid, orderNo); err == sql.ErrNoRows {
+		apierr.WriteJSON(w, http.StatusNotFound, apierr.ErrNotFound)
+		return
+	} else if err != nil {
+		apierr.WriteJSON(w, http.StatusInternalServerError, apierr.ErrInternal)
+		return
+	}
+	res, err := h.client.Retry(ctx, orderNo, req.Method, filterMetadata(req.Metadata))
+	if err != nil {
+		writePayErr(w, err)
+		return
+	}
+	_ = h.store.UpdateMethod(ctx, uid, orderNo, req.Method)
+	writeJSON(w, sessionResponse{OrderNo: res.OrderNo, Session: res.Session})
+}
+
+// ─── POST /v1/pay/orders/{orderNo}/cancel ───────────────────────────────────
+
+func (h *Handler) Cancel(w http.ResponseWriter, r *http.Request) {
+	ctx := r.Context()
+	uid, ok := auth.UserIDFromContext(ctx)
+	if !ok {
+		apierr.WriteJSON(w, http.StatusUnauthorized, apierr.ErrUnauthorized)
+		return
+	}
+	orderNo := chi.URLParam(r, "orderNo")
+	if _, err := h.store.GetForUser(ctx, uid, orderNo); err == sql.ErrNoRows {
+		apierr.WriteJSON(w, http.StatusNotFound, apierr.ErrNotFound)
+		return
+	} else if err != nil {
+		apierr.WriteJSON(w, http.StatusInternalServerError, apierr.ErrInternal)
+		return
+	}
+	canceled, err := h.client.Cancel(ctx, orderNo)
+	if err != nil {
+		writePayErr(w, err)
+		return
+	}
+	if canceled {
+		_ = h.store.MarkCanceled(ctx, uid, orderNo)
+	}
+	writeJSON(w, map[string]bool{"canceled": canceled})
+}

server/internal/pay/handler_test.go

@@ -0,0 +1,146 @@
+package pay
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/go-chi/chi/v5"
+
+	"github.com/wangjia/pangolin/server/internal/codes"
+)
+
+// newHandlerRig: 假 pay + sqlite 台账 + chi 路由(URL 参数解析需要真实路由)。
+func newHandlerRig(t *testing.T, payFn http.HandlerFunc) (*chi.Mux, *Store) {
+	t.Helper()
+	db := openMigratedSQLite(t)
+	seedUser(t, db, 1, "uuid-1")
+	seedUser(t, db, 2, "uuid-2")
+	if payFn == nil {
+		payFn = func(http.ResponseWriter, *http.Request) {}
+	}
+	srv := fakePay(t, payFn)
+	st := NewStore(db)
+	h := NewHandler(NewClient(srv.URL, "pangolin", testSecret), st, db)
+	r := chi.NewRouter()
+	r.Get("/v1/pay/catalog", h.Catalog)
+	r.Post("/v1/pay/orders", h.CreateOrder)
+	r.Get("/v1/pay/orders/{orderNo}", h.GetOrder)
+	r.Post("/v1/pay/orders/{orderNo}/retry", h.Retry)
+	r.Post("/v1/pay/orders/{orderNo}/cancel", h.Cancel)
+	return r, st
+}
+
+// authed 注入 uid(auth.RequireAuth 注入的就是 codes.CtxKeyUserID)。
+func authed(r *http.Request, uid int64) *http.Request {
+	return r.WithContext(context.WithValue(r.Context(), codes.CtxKeyUserID, uid))
+}
+
+func TestCreateOrder_ProxiesAndRecords(t *testing.T) {
+	router, st := newHandlerRig(t, func(w http.ResponseWriter, r *http.Request) {
+		_, _ = w.Write([]byte(`{"data":{"order_no":"pay001","session":{
+			"render_type":"redirect","payload":{"url":"https://alipay.example/x"}}}}`))
+	})
+	body := []byte(`{"sku":"pro_month","method":"alipay","metadata":{"is_mobile":"1","evil":"x"}}`)
+	req := authed(httptest.NewRequest(http.MethodPost, "/v1/pay/orders", bytes.NewReader(body)), 1)
+	w := httptest.NewRecorder()
+	router.ServeHTTP(w, req)
+	if w.Code != http.StatusOK {
+		t.Fatalf("code = %d, body = %s", w.Code, w.Body)
+	}
+	var resp struct {
+		OrderNo string  `json:"order_no"`
+		Session Session `json:"session"`
+	}
+	_ = json.Unmarshal(w.Body.Bytes(), &resp)
+	if resp.OrderNo != "pay001" || resp.Session.RenderType != "redirect" {
+		t.Fatalf("resp = %+v", resp)
+	}
+	row, err := st.GetForUser(context.Background(), 1, "pay001")
+	if err != nil {
+		t.Fatalf("台账未落: %v", err)
+	}
+	if row.SKU != "pro_month" || row.BizRef != "uuid-1" || row.Status != "created" {
+		t.Fatalf("台账行不符: %+v", row)
+	}
+}
+
+func TestCreateOrder_UnknownSKU400(t *testing.T) {
+	router, _ := newHandlerRig(t, func(w http.ResponseWriter, r *http.Request) {
+		t.Fatal("不该打到 pay")
+	})
+	body := []byte(`{"sku":"pro_lifetime","method":"alipay"}`)
+	req := authed(httptest.NewRequest(http.MethodPost, "/v1/pay/orders", bytes.NewReader(body)), 1)
+	w := httptest.NewRecorder()
+	router.ServeHTTP(w, req)
+	if w.Code != http.StatusBadRequest {
+		t.Fatalf("code = %d", w.Code)
+	}
+}
+
+func TestGetOrder_OwnershipEnforced(t *testing.T) {
+	router, st := newHandlerRig(t, func(w http.ResponseWriter, r *http.Request) {
+		_, _ = w.Write([]byte(`{"data":{"order_no":"pay001","status":"pending",
+			"subject":"s","amount_minor":2999,"currency":"CNY"}}`))
+	})
+	if err := st.Insert(context.Background(), 1, "uuid-1", "pro_month", "pay001", "alipay"); err != nil {
+		t.Fatal(err)
+	}
+	// 属主可查
+	w := httptest.NewRecorder()
+	router.ServeHTTP(w, authed(httptest.NewRequest(http.MethodGet, "/v1/pay/orders/pay001", nil), 1))
+	if w.Code != http.StatusOK {
+		t.Fatalf("owner code = %d", w.Code)
+	}
+	var resp struct {
+		PayStatus string `json:"pay_status"`
+		Activated bool   `json:"activated"`
+	}
+	_ = json.Unmarshal(w.Body.Bytes(), &resp)
+	if resp.PayStatus != "pending" || resp.Activated {
+		t.Fatalf("resp = %+v", resp)
+	}
+	// 他人 404
+	w2 := httptest.NewRecorder()
+	router.ServeHTTP(w2, authed(httptest.NewRequest(http.MethodGet, "/v1/pay/orders/pay001", nil), 2))
+	if w2.Code != http.StatusNotFound {
+		t.Fatalf("other code = %d", w2.Code)
+	}
+}
+
+func TestRetry_CurrencyMismatchMapped409(t *testing.T) {
+	router, st := newHandlerRig(t, func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusConflict)
+		_, _ = w.Write([]byte(`{"code":"currency_mismatch","message":"换渠道需新单"}`))
+	})
+	_ = st.Insert(context.Background(), 1, "uuid-1", "pro_month", "pay001", "crypto")
+	body := []byte(`{"method":"alipay"}`)
+	w := httptest.NewRecorder()
+	router.ServeHTTP(w, authed(httptest.NewRequest(http.MethodPost, "/v1/pay/orders/pay001/retry", bytes.NewReader(body)), 1))
+	if w.Code != http.StatusConflict {
+		t.Fatalf("code = %d", w.Code)
+	}
+	var e struct {
+		Code string `json:"code"`
+	}
+	_ = json.Unmarshal(w.Body.Bytes(), &e)
+	if e.Code != "CURRENCY_MISMATCH" {
+		t.Fatalf("code = %q, want CURRENCY_MISMATCH", e.Code)
+	}
+}
+
+func TestCatalog(t *testing.T) {
+	router, _ := newHandlerRig(t, nil)
+	w := httptest.NewRecorder()
+	router.ServeHTTP(w, authed(httptest.NewRequest(http.MethodGet, "/v1/pay/catalog", nil), 1))
+	var resp struct {
+		Items []CatalogItem `json:"items"`
+	}
+	_ = json.Unmarshal(w.Body.Bytes(), &resp)
+	if len(resp.Items) != 3 || resp.Items[0].SKU != "pro_month" {
+		t.Fatalf("catalog = %+v", resp.Items)
+	}
+}

server/internal/pay/store.go

@@ -0,0 +1,137 @@
+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
+}

server/internal/pay/testutil_test.go

@@ -0,0 +1,40 @@
+package pay
+
+import (
+	"context"
+	"database/sql"
+	"testing"
+	"time"
+
+	"github.com/wangjia/pangolin/server/internal/config"
+	"github.com/wangjia/pangolin/server/internal/store"
+)
+
+func openMigratedSQLite(t *testing.T) *sql.DB {
+	t.Helper()
+	db, err := store.Open(&config.Config{Driver: "sqlite", DSN: ":memory:"})
+	if err != nil {
+		t.Fatalf("open: %v", err)
+	}
+	t.Cleanup(func() { _ = db.Close() })
+	if err := store.MigrateUp(db, "sqlite"); err != nil {
+		t.Fatalf("migrate: %v", err)
+	}
+	if err := store.ApplyCodesLibMigrations(context.Background(), db, "sqlite"); err != nil {
+		t.Fatalf("lib migrate: %v", err)
+	}
+	return db
+}
+
+// seedUser 造最小 users 行(列以 codes/sqlite_helper_test.go::seedUser 为准,
+// 实现时照抄那份 INSERT——含 uuid/email/pw_hash/dp_uuid 等 NOT NULL 列)。
+func seedUser(t *testing.T, db *sql.DB, id int64, uuid string) {
+	t.Helper()
+	_, err := db.Exec(
+		`INSERT INTO users (id, uuid, email, pw_hash, dp_uuid, status, created_at)
+		 VALUES (?, ?, ?, 'x', ?, 'active', ?)`,
+		id, uuid, uuid+"@t.local", uuid, time.Now().UTC())
+	if err != nil {
+		t.Fatalf("seedUser: %v", err)
+	}
+}