From 932953bafeed5ee69ba8e2251c644505c9ebb05e Mon Sep 17 00:00:00 2001 From: wangjia <809946525@qq.com> Date: Fri, 10 Jul 2026 22:32:43 +0800 Subject: [PATCH] =?UTF-8?q?feat(server):=20/v1/pay=20=E4=B8=8B=E5=8D=95?= =?UTF-8?q?=E4=BB=A3=E7=90=86=E7=AB=AF=E7=82=B9=20+=20=E8=B4=AD=E4=B9=B0?= =?UTF-8?q?=E5=8F=B0=E8=B4=A6(JWT=20=E9=89=B4=E6=9D=83,user=E2=86=92biz=5F?= =?UTF-8?q?ref=20=E6=98=A0=E5=B0=84)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_013nMthbVEmQquxBRKb9Fj8u --- server/cmd/server/main.go | 20 +++ server/internal/pay/catalog.go | 30 ++++ server/internal/pay/handler.go | 241 +++++++++++++++++++++++++++ server/internal/pay/handler_test.go | 146 ++++++++++++++++ server/internal/pay/store.go | 137 +++++++++++++++ server/internal/pay/testutil_test.go | 40 +++++ 6 files changed, 614 insertions(+) create mode 100644 server/internal/pay/catalog.go create mode 100644 server/internal/pay/handler.go create mode 100644 server/internal/pay/handler_test.go create mode 100644 server/internal/pay/store.go create mode 100644 server/internal/pay/testutil_test.go diff --git a/server/cmd/server/main.go b/server/cmd/server/main.go index 9bf23c3..239ebf3 100644 --- a/server/cmd/server/main.go +++ b/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) diff --git a/server/internal/pay/catalog.go b/server/internal/pay/catalog.go new file mode 100644 index 0000000..75823f2 --- /dev/null +++ b/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 +} diff --git a/server/internal/pay/handler.go b/server/internal/pay/handler.go new file mode 100644 index 0000000..aba7369 --- /dev/null +++ b/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}) +} diff --git a/server/internal/pay/handler_test.go b/server/internal/pay/handler_test.go new file mode 100644 index 0000000..d2f5e2c --- /dev/null +++ b/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) + } +} diff --git a/server/internal/pay/store.go b/server/internal/pay/store.go new file mode 100644 index 0000000..99b4585 --- /dev/null +++ b/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 +} diff --git a/server/internal/pay/testutil_test.go b/server/internal/pay/testutil_test.go new file mode 100644 index 0000000..7d1ea24 --- /dev/null +++ b/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) + } +}