feat(server): pay webhook 接收器(验签/时间窗/nonce + out_trade_no 幂等开通,回 SUCCESS)
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013nMthbVEmQquxBRKb9Fj8u
This commit is contained in:
wangjia
@@ -302,12 +302,15 @@ func mountV1(r chi.Router, sqlDB *sql.DB, rdb *redis.Client, nodeSvc *nodes.Serv
|
||||
|
||||
// ── Pay(pay v2 统一支付网关;PAY_BASE_URL 未配则整组不挂载)──────────────
|
||||
var payHandler *pay.Handler
|
||||
var payWebhook *pay.WebhookHandler
|
||||
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)
|
||||
payWebhook = pay.NewWebhookHandler(payStore, codesSvc, sqlDB, rdb,
|
||||
paySystem, paySecret, 5*time.Minute, 15*time.Minute)
|
||||
} else {
|
||||
log.Printf("PAY_BASE_URL 未配置 — /v1/pay 支付端点不挂载")
|
||||
}
|
||||
@@ -374,6 +377,9 @@ func mountV1(r chi.Router, sqlDB *sql.DB, rdb *redis.Client, nodeSvc *nodes.Serv
|
||||
|
||||
// Webhook: HMAC-authenticated, no JWT.
|
||||
v1.Post("/webhook/store/codes", webhookHandler.ServeHTTP)
|
||||
if payWebhook != nil {
|
||||
v1.Post("/webhook/pay", payWebhook.ServeHTTP)
|
||||
}
|
||||
|
||||
// Protected: all routes that require a valid Bearer JWT.
|
||||
if tm != nil {
|
||||
|
||||
@@ -0,0 +1,173 @@
|
||||
package pay
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"time"
|
||||
|
||||
"github.com/redis/go-redis/v9"
|
||||
|
||||
"github.com/wangjia/pangolin/server/internal/codes"
|
||||
)
|
||||
|
||||
// Granter 抽象 codes.Service 的支付授予入口(测试可替身;生产传 *codes.Service)。
|
||||
type Granter interface {
|
||||
GrantPaidSubscriptionTx(ctx context.Context, tx *sql.Tx, userID int64, plan codes.PlanCode, days int, ref string) (int64, time.Time, error)
|
||||
}
|
||||
|
||||
// WebhookHandler 接收 pay 的 payment.succeeded 出站 webhook。
|
||||
// 验签与 pay verifyBizSign 对称:同 secret,parts=[system, ts, nonce, rawBody],
|
||||
// ±tolerance 时间窗。幂等三层:nonce SETNX(传输重放)→ out_trade_no 锁内
|
||||
// CAS(业务幂等,重投唯一可靠键)→ biz_ref 兜底(台账缺行自修复)。
|
||||
type WebhookHandler struct {
|
||||
store *Store
|
||||
granter Granter
|
||||
db *sql.DB
|
||||
rdb *redis.Client // 可为 nil:跳过 nonce 层,业务幂等仍成立
|
||||
system string
|
||||
secret string
|
||||
tolerance time.Duration
|
||||
nonceTTL time.Duration
|
||||
now func() time.Time // 测试注入
|
||||
}
|
||||
|
||||
func NewWebhookHandler(store *Store, granter Granter, db *sql.DB, rdb *redis.Client,
|
||||
system, secret string, tolerance, nonceTTL time.Duration) *WebhookHandler {
|
||||
return &WebhookHandler{store: store, granter: granter, db: db, rdb: rdb,
|
||||
system: system, secret: secret, tolerance: tolerance, nonceTTL: nonceTTL, now: time.Now}
|
||||
}
|
||||
|
||||
// webhookEvent 对应 pay settle.go::enqueuePaymentSucceeded 的 payload
|
||||
// (注意:payment.succeeded 无 refund_id 字段)。
|
||||
type webhookEvent struct {
|
||||
EventType string `json:"event_type"`
|
||||
OutTradeNo string `json:"out_trade_no"`
|
||||
BizSystem string `json:"biz_system"`
|
||||
BizRef string `json:"biz_ref"`
|
||||
ProductBizCode string `json:"product_biz_code"`
|
||||
AmountMinor int64 `json:"amount_minor"`
|
||||
Currency string `json:"currency"`
|
||||
Channel string `json:"channel"`
|
||||
PaidAt string `json:"paid_at"` // RFC3339
|
||||
}
|
||||
|
||||
func (h *WebhookHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
body, err := io.ReadAll(io.LimitReader(r.Body, 64<<10))
|
||||
if err != nil {
|
||||
http.Error(w, "read body", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
if !h.verify(r, body) {
|
||||
http.Error(w, "signature verification failed", http.StatusUnauthorized)
|
||||
return
|
||||
}
|
||||
// nonce 防重放(仅传输层;pay 每次重投换新 nonce,业务幂等靠 out_trade_no)。
|
||||
if h.rdb != nil {
|
||||
if nonce := r.Header.Get("X-Pay-Nonce"); nonce != "" {
|
||||
ok, err := h.rdb.SetNX(r.Context(), "pay:webhook:nonce:"+nonce, 1, h.nonceTTL).Result()
|
||||
if err == nil && !ok {
|
||||
writeSuccess(w) // 同 nonce 重放:已处理过,直接确认
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
var ev webhookEvent
|
||||
if err := json.Unmarshal(body, &ev); err != nil {
|
||||
http.Error(w, "bad payload", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
if ev.EventType != "payment.succeeded" {
|
||||
// 事件白名单外(pay 侧只应配 payment.succeeded):确认不处理,免重投。
|
||||
writeSuccess(w)
|
||||
return
|
||||
}
|
||||
if err := h.settle(r.Context(), &ev); err != nil {
|
||||
slog.Error("pay webhook 开通失败(pay 将退避重投)", "order_no", ev.OutTradeNo, "err", err)
|
||||
http.Error(w, "settle failed", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
writeSuccess(w)
|
||||
}
|
||||
|
||||
// writeSuccess:pay 的 ACK 判据是 HTTP 200 且 body 含 "SUCCESS"(大写包含)。
|
||||
func writeSuccess(w http.ResponseWriter) {
|
||||
w.WriteHeader(http.StatusOK)
|
||||
_, _ = w.Write([]byte("SUCCESS"))
|
||||
}
|
||||
|
||||
func (h *WebhookHandler) verify(r *http.Request, body []byte) bool {
|
||||
if r.Header.Get("X-Pay-System") != h.system {
|
||||
return false
|
||||
}
|
||||
ts := r.Header.Get("X-Pay-Timestamp")
|
||||
nonce := r.Header.Get("X-Pay-Nonce")
|
||||
sign := r.Header.Get("X-Pay-Sign")
|
||||
if ts == "" || nonce == "" || sign == "" {
|
||||
return false
|
||||
}
|
||||
tsi, err := strconv.ParseInt(ts, 10, 64)
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
tol := int64(h.tolerance.Seconds())
|
||||
if d := h.now().Unix() - tsi; d > tol || d < -tol {
|
||||
return false
|
||||
}
|
||||
return hmacVerify(h.secret, sign, h.system, ts, nonce, string(body))
|
||||
}
|
||||
|
||||
// settle 幂等开通:锁台账行 → created→paid 翻转 + 同事务 grant(叠加语义
|
||||
// 复用 codes.applySubscription)。已 paid 直接返回 nil(重投/并发输家)。
|
||||
// canceled 行也照常开通——钱已实收,本地 cancel 只是未支付单的整理。
|
||||
func (h *WebhookHandler) settle(ctx context.Context, ev *webhookEvent) error {
|
||||
item, ok := CatalogBySKU(ev.ProductBizCode)
|
||||
if !ok {
|
||||
return fmt.Errorf("未知 product_biz_code %q(与 pay 种子漂移?)", ev.ProductBizCode)
|
||||
}
|
||||
tx, err := h.store.BeginTx(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer func() { _ = tx.Rollback() }()
|
||||
|
||||
var purchaseID, userID int64
|
||||
row, err := h.store.LockByOutTradeNoTx(ctx, tx, ev.OutTradeNo)
|
||||
switch {
|
||||
case err == sql.ErrNoRows:
|
||||
// 台账缺行(下单后本地写失败)→ 按 biz_ref=用户 uuid 兜底定位补建。
|
||||
if err := h.db.QueryRowContext(ctx,
|
||||
`SELECT id FROM users WHERE uuid = ? AND status = 'active'`, ev.BizRef).Scan(&userID); err != nil {
|
||||
return fmt.Errorf("biz_ref %q 定位用户失败: %w", ev.BizRef, err)
|
||||
}
|
||||
purchaseID, err = h.store.InsertFromWebhookTx(ctx, tx, userID, ev.BizRef, ev.ProductBizCode, ev.OutTradeNo, ev.Channel)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
case err != nil:
|
||||
return err
|
||||
case row.Status == "paid":
|
||||
return nil // 幂等:已消费,直接 SUCCESS
|
||||
default:
|
||||
purchaseID, userID = row.ID, row.UserID
|
||||
}
|
||||
|
||||
subID, _, err := h.granter.GrantPaidSubscriptionTx(ctx, tx, userID,
|
||||
codes.PlanCode(item.Plan), item.Days, "pay:"+ev.OutTradeNo)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
paidAt, perr := time.Parse(time.RFC3339, ev.PaidAt)
|
||||
if perr != nil {
|
||||
paidAt = h.now().UTC()
|
||||
}
|
||||
if err := h.store.MarkPaidTx(ctx, tx, purchaseID, ev.AmountMinor, ev.Currency, ev.Channel, subID, paidAt); err != nil {
|
||||
return err
|
||||
}
|
||||
return tx.Commit()
|
||||
}
|
||||
@@ -0,0 +1,219 @@
|
||||
package pay
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strconv"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/wangjia/pangolin/server/internal/codes"
|
||||
)
|
||||
|
||||
func newWebhookRig(t *testing.T) (*WebhookHandler, *sql.DB, *Store) {
|
||||
t.Helper()
|
||||
db := openMigratedSQLite(t)
|
||||
seedUser(t, db, 1, "uuid-1")
|
||||
st := NewStore(db)
|
||||
codesSvc := codes.NewService(codes.NewStore(db), nil, 5, time.Hour)
|
||||
h := NewWebhookHandler(st, codesSvc, db, nil, "pangolin", testSecret, 5*time.Minute, 15*time.Minute)
|
||||
return h, db, st
|
||||
}
|
||||
|
||||
// deliver 按 pay notifier.go 出站语义构造签名请求(每次新 nonce,模拟重投)。
|
||||
func deliver(t *testing.T, h *WebhookHandler, payload map[string]any) *httptest.ResponseRecorder {
|
||||
t.Helper()
|
||||
body, _ := json.Marshal(payload)
|
||||
ts := strconv.FormatInt(time.Now().Unix(), 10)
|
||||
nonce := newNonce()
|
||||
r := httptest.NewRequest(http.MethodPost, "/v1/webhook/pay", bytes.NewReader(body))
|
||||
r.Header.Set("Content-Type", "application/json")
|
||||
r.Header.Set("X-Pay-System", "pangolin")
|
||||
r.Header.Set("X-Pay-Event", "payment.succeeded")
|
||||
r.Header.Set("X-Pay-Timestamp", ts)
|
||||
r.Header.Set("X-Pay-Nonce", nonce)
|
||||
r.Header.Set("X-Pay-Sign", hmacSign(testSecret, "pangolin", ts, nonce, string(body)))
|
||||
w := httptest.NewRecorder()
|
||||
h.ServeHTTP(w, r)
|
||||
return w
|
||||
}
|
||||
|
||||
func succeededPayload(orderNo, sku string) map[string]any {
|
||||
return map[string]any{
|
||||
"event_type": "payment.succeeded",
|
||||
"out_trade_no": orderNo,
|
||||
"biz_system": "pangolin",
|
||||
"biz_ref": "uuid-1",
|
||||
"product_biz_code": sku,
|
||||
"amount_minor": int64(4201234),
|
||||
"currency": "USDT",
|
||||
"channel": "crypto",
|
||||
"paid_at": time.Now().UTC().Format(time.RFC3339),
|
||||
}
|
||||
}
|
||||
|
||||
func TestWebhook_HappyPath(t *testing.T) {
|
||||
h, db, st := newWebhookRig(t)
|
||||
ctx := context.Background()
|
||||
if err := st.Insert(ctx, 1, "uuid-1", "pro_month", "pay001", "crypto"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
w := deliver(t, h, succeededPayload("pay001", "pro_month"))
|
||||
if w.Code != http.StatusOK || !strings.Contains(w.Body.String(), "SUCCESS") {
|
||||
t.Fatalf("ACK 不符: %d %q(pay 要求 200+body 含 SUCCESS)", w.Code, w.Body.String())
|
||||
}
|
||||
row, err := st.GetForUser(ctx, 1, "pay001")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if row.Status != "paid" || !row.SubID.Valid || row.AmountMinor != 4201234 || row.Channel != "crypto" {
|
||||
t.Fatalf("台账未正确消费: %+v", row)
|
||||
}
|
||||
var source string
|
||||
var expires time.Time
|
||||
if err := db.QueryRow(`SELECT source, expires_at FROM subscriptions WHERE id = ?`, row.SubID.Int64).
|
||||
Scan(&source, &expires); err != nil {
|
||||
t.Fatalf("订阅未开通: %v", err)
|
||||
}
|
||||
if source != "pay" {
|
||||
t.Errorf("source = %q, want pay", source)
|
||||
}
|
||||
want := time.Now().UTC().AddDate(0, 0, 31)
|
||||
if d := expires.Sub(want); d > time.Minute || d < -time.Minute {
|
||||
t.Errorf("expires = %v, want ≈ %v(pro_month=31 天)", expires, want)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWebhook_RedeliveryIdempotent(t *testing.T) {
|
||||
h, db, st := newWebhookRig(t)
|
||||
ctx := context.Background()
|
||||
_ = st.Insert(ctx, 1, "uuid-1", "pro_month", "pay001", "crypto")
|
||||
deliver(t, h, succeededPayload("pay001", "pro_month"))
|
||||
var exp1 time.Time
|
||||
_ = db.QueryRow(`SELECT expires_at FROM subscriptions WHERE user_id = 1`).Scan(&exp1)
|
||||
|
||||
// 重投(新 nonce)必须:200+SUCCESS、订阅行数不变、到期不变。
|
||||
w := deliver(t, h, succeededPayload("pay001", "pro_month"))
|
||||
if w.Code != http.StatusOK || !strings.Contains(w.Body.String(), "SUCCESS") {
|
||||
t.Fatalf("重投未确认: %d %q", w.Code, w.Body.String())
|
||||
}
|
||||
var n int
|
||||
_ = db.QueryRow(`SELECT COUNT(*) FROM subscriptions WHERE user_id = 1`).Scan(&n)
|
||||
if n != 1 {
|
||||
t.Fatalf("重投多开了订阅: rows = %d", n)
|
||||
}
|
||||
var exp2 time.Time
|
||||
_ = db.QueryRow(`SELECT expires_at FROM subscriptions WHERE user_id = 1`).Scan(&exp2)
|
||||
if !exp1.Equal(exp2) {
|
||||
t.Errorf("重投改了到期: %v → %v", exp1, exp2)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWebhook_TwoOrdersStack(t *testing.T) {
|
||||
h, db, st := newWebhookRig(t)
|
||||
ctx := context.Background()
|
||||
_ = st.Insert(ctx, 1, "uuid-1", "pro_month", "pay001", "crypto")
|
||||
_ = st.Insert(ctx, 1, "uuid-1", "pro_quarter", "pay002", "crypto")
|
||||
deliver(t, h, succeededPayload("pay001", "pro_month"))
|
||||
deliver(t, h, succeededPayload("pay002", "pro_quarter"))
|
||||
var n int
|
||||
_ = db.QueryRow(`SELECT COUNT(*) FROM subscriptions WHERE user_id = 1`).Scan(&n)
|
||||
if n != 1 {
|
||||
t.Fatalf("同 plan 应原地叠加: rows = %d", n)
|
||||
}
|
||||
var expires time.Time
|
||||
_ = db.QueryRow(`SELECT expires_at FROM subscriptions WHERE user_id = 1`).Scan(&expires)
|
||||
want := time.Now().UTC().AddDate(0, 0, 31+92)
|
||||
if d := expires.Sub(want); d > time.Minute || d < -time.Minute {
|
||||
t.Errorf("expires = %v, want ≈ %v(31+92 天叠加)", expires, want)
|
||||
}
|
||||
}
|
||||
|
||||
// 台账缺行(下单后本地写失败):按 biz_ref 兜底定位用户、补台账、照常开通。
|
||||
func TestWebhook_MissingLedgerFallsBackToBizRef(t *testing.T) {
|
||||
h, db, st := newWebhookRig(t)
|
||||
w := deliver(t, h, succeededPayload("pay-orphan", "pro_year"))
|
||||
if w.Code != http.StatusOK {
|
||||
t.Fatalf("code = %d body = %s", w.Code, w.Body.String())
|
||||
}
|
||||
row, err := st.GetForUser(context.Background(), 1, "pay-orphan")
|
||||
if err != nil {
|
||||
t.Fatalf("兜底台账未建: %v", err)
|
||||
}
|
||||
if row.Status != "paid" || row.SKU != "pro_year" {
|
||||
t.Fatalf("兜底行不符: %+v", row)
|
||||
}
|
||||
var n int
|
||||
_ = db.QueryRow(`SELECT COUNT(*) FROM subscriptions WHERE user_id = 1 AND source = 'pay'`).Scan(&n)
|
||||
if n != 1 {
|
||||
t.Errorf("订阅未开通: %d", n)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWebhook_RejectsBadSignatureAndStaleTimestamp(t *testing.T) {
|
||||
h, db, _ := newWebhookRig(t)
|
||||
body, _ := json.Marshal(succeededPayload("pay001", "pro_month"))
|
||||
mk := func(mutate func(r *http.Request)) int {
|
||||
ts := strconv.FormatInt(time.Now().Unix(), 10)
|
||||
nonce := newNonce()
|
||||
r := httptest.NewRequest(http.MethodPost, "/v1/webhook/pay", bytes.NewReader(body))
|
||||
r.Header.Set("X-Pay-System", "pangolin")
|
||||
r.Header.Set("X-Pay-Timestamp", ts)
|
||||
r.Header.Set("X-Pay-Nonce", nonce)
|
||||
r.Header.Set("X-Pay-Sign", hmacSign(testSecret, "pangolin", ts, nonce, string(body)))
|
||||
mutate(r)
|
||||
w := httptest.NewRecorder()
|
||||
h.ServeHTTP(w, r)
|
||||
return w.Code
|
||||
}
|
||||
if c := mk(func(r *http.Request) { r.Header.Set("X-Pay-Sign", "AAAA") }); c != http.StatusUnauthorized {
|
||||
t.Errorf("坏签名 code = %d, want 401", c)
|
||||
}
|
||||
if c := mk(func(r *http.Request) { r.Header.Set("X-Pay-System", "jiu") }); c != http.StatusUnauthorized {
|
||||
t.Errorf("错 system code = %d, want 401", c)
|
||||
}
|
||||
if c := mk(func(r *http.Request) {
|
||||
stale := strconv.FormatInt(time.Now().Add(-10*time.Minute).Unix(), 10)
|
||||
r.Header.Set("X-Pay-Timestamp", stale)
|
||||
// 注意:重签,否则先挂在签名而非时间窗上
|
||||
nonce := r.Header.Get("X-Pay-Nonce")
|
||||
r.Header.Set("X-Pay-Sign", hmacSign(testSecret, "pangolin", stale, nonce, string(body)))
|
||||
}); c != http.StatusUnauthorized {
|
||||
t.Errorf("过期 ts code = %d, want 401", c)
|
||||
}
|
||||
var n int
|
||||
_ = db.QueryRow(`SELECT COUNT(*) FROM subscriptions`).Scan(&n)
|
||||
if n != 0 {
|
||||
t.Errorf("拒绝路径不得开通: %d", n)
|
||||
}
|
||||
}
|
||||
|
||||
// 未知 sku(catalog 漂移)→ 500,pay 会重投,期间可修 catalog 后自愈。
|
||||
func TestWebhook_UnknownSKU500(t *testing.T) {
|
||||
h, _, _ := newWebhookRig(t)
|
||||
w := deliver(t, h, succeededPayload("pay001", "pro_lifetime"))
|
||||
if w.Code != http.StatusInternalServerError {
|
||||
t.Fatalf("code = %d, want 500", w.Code)
|
||||
}
|
||||
}
|
||||
|
||||
// 白名单外事件(如未来误配 refund.succeeded):确认不处理,避免无谓重投 12 次。
|
||||
func TestWebhook_IgnoredEventAcked(t *testing.T) {
|
||||
h, db, _ := newWebhookRig(t)
|
||||
p := succeededPayload("pay001", "pro_month")
|
||||
p["event_type"] = "refund.succeeded"
|
||||
w := deliver(t, h, p)
|
||||
if w.Code != http.StatusOK || !strings.Contains(w.Body.String(), "SUCCESS") {
|
||||
t.Fatalf("白名单外事件应直接确认: %d", w.Code)
|
||||
}
|
||||
var n int
|
||||
_ = db.QueryRow(`SELECT COUNT(*) FROM subscriptions`).Scan(&n)
|
||||
if n != 0 {
|
||||
t.Errorf("不得开通: %d", n)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user