wangjia/pay
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(v2): webhook v2 outbox + Notifier(event_type + HMAC 双向 + 重试兜底)

Browse Source
This commit is contained in:
wangjia
2026-07-10 11:25:14 +08:00
parent 0877b01874
commit 3f5e44581f
6 changed files with 394 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/model/testdb.go
+1 -1
View File
@@ -26,7 +26,7 @@ func OpenTestDB(t *testing.T) *gorm.DB {
if err != nil {
t.Fatalf("open test db: %v", err)
}
if err := db.AutoMigrate(&OrderV2{}, &Attempt{}, &Account{}, &Refund{}); err != nil {
if err := db.AutoMigrate(&OrderV2{}, &Attempt{}, &Account{}, &Refund{}, &WebhookDelivery{}); err != nil {
t.Fatalf("migrate: %v", err)
}
sqlDB, _ := db.DB()
internal/model/webhook_delivery.go
+14
View File
@@ -0,0 +1,14 @@
package model
// WebhookDelivery 是 pay→业务方 webhook 的 outbox(v2)。unique(out_trade_no,event_type)
// 保证同一订单同一事件只入队一次(幂等);后台 Notifier 扫 Delivered=false 重试兜底。
type WebhookDelivery struct {
Base
OutTradeNo string `gorm:"size:64;not null;uniqueIndex:uq_delivery" json:"out_trade_no"`
EventType string `gorm:"size:32;not null;uniqueIndex:uq_delivery" json:"event_type"`
BizSystem string `gorm:"index;size:32" json:"biz_system"`
Payload string `gorm:"type:text" json:"payload"` // 已序列化的领域 JSON(含 event_type)
Delivered bool `gorm:"index;default:false" json:"delivered"`
Attempts int `json:"attempts"`
LastError string `gorm:"size:255" json:"last_error,omitempty"`
}
internal/store/webhook.go
+68
View File
@@ -0,0 +1,68 @@
package store
import (
"fmt"
"gorm.io/gorm"
"gorm.io/gorm/clause"
"github.com/wangjia/pay/internal/model"
)
// WebhookDeliveryRow is the row type returned by WebhookStore (alias to the model).
type WebhookDeliveryRow = model.WebhookDelivery
type WebhookStore struct{ db *gorm.DB }
func NewWebhookStore(db *gorm.DB) *WebhookStore { return &WebhookStore{db: db} }
// EnqueueDelivery inserts an outbox row; a duplicate (out_trade_no,event_type)
// is a no-op (idempotent enqueue) via ON CONFLICT DO NOTHING.
func (s *WebhookStore) EnqueueDelivery(outTradeNo, bizSystem, eventType, payload string) error {
row := model.WebhookDelivery{
OutTradeNo: outTradeNo, BizSystem: bizSystem, EventType: eventType, Payload: payload,
}
err := s.db.Clauses(clause.OnConflict{
Columns: []clause.Column{{Name: "out_trade_no"}, {Name: "event_type"}},
DoNothing: true,
}).Create(&row).Error
if err != nil {
return fmt.Errorf("store.EnqueueDelivery: %w", err)
}
return nil
}
func (s *WebhookStore) ListUndelivered(limit int) ([]WebhookDeliveryRow, error) {
if limit <= 0 || limit > 200 {
limit = 50
}
var out []WebhookDeliveryRow
if err := s.db.Where("delivered = ?", false).Order("id ASC").Limit(limit).Find(&out).Error; err != nil {
return nil, fmt.Errorf("store.ListUndelivered: %w", err)
}
return out, nil
}
func (s *WebhookStore) MarkDelivered(id uint64) error {
if err := s.db.Model(&model.WebhookDelivery{}).Where("id = ?", id).
Update("delivered", true).Error; err != nil {
return fmt.Errorf("store.MarkDelivered: %w", err)
}
return nil
}
// MarkFailed increments attempts and records the last error, leaving the row
// undelivered for the next retry sweep.
func (s *WebhookStore) MarkFailed(id uint64, errMsg string) error {
if len(errMsg) > 255 {
errMsg = errMsg[:255]
}
if err := s.db.Model(&model.WebhookDelivery{}).Where("id = ?", id).
Updates(map[string]any{
"attempts": gorm.Expr("attempts + 1"),
"last_error": errMsg,
}).Error; err != nil {
return fmt.Errorf("store.MarkFailed: %w", err)
}
return nil
}
internal/store/webhook_test.go
+44
View File
@@ -0,0 +1,44 @@
package store_test
import (
"testing"
"github.com/wangjia/pay/internal/model"
"github.com/wangjia/pay/internal/store"
)
func TestWebhookOutboxEnqueueIdempotent(t *testing.T) {
ws := store.NewWebhookStore(model.OpenTestDB(t))
if err := ws.EnqueueDelivery("PAY-1", "pangolin", "payment.succeeded", `{"a":1}`); err != nil {
t.Fatalf("enqueue#1: %v", err)
}
// 幂等:同 (out_trade_no,event_type) 再入队不新增行、不报错。
if err := ws.EnqueueDelivery("PAY-1", "pangolin", "payment.succeeded", `{"a":1}`); err != nil {
t.Fatalf("enqueue#2: %v", err)
}
list, _ := ws.ListUndelivered(10)
if len(list) != 1 {
t.Fatalf("应恰 1 行待投递, got %d", len(list))
}
if err := ws.MarkDelivered(list[0].ID); err != nil {
t.Fatalf("markDelivered: %v", err)
}
if again, _ := ws.ListUndelivered(10); len(again) != 0 {
t.Fatalf("投递后应 0 待投递, got %d", len(again))
}
}
func TestWebhookMarkFailed(t *testing.T) {
ws := store.NewWebhookStore(model.OpenTestDB(t))
_ = ws.EnqueueDelivery("PAY-2", "jiu", "payment.succeeded", `{}`)
list, _ := ws.ListUndelivered(10)
if err := ws.MarkFailed(list[0].ID, "boom"); err != nil {
t.Fatalf("markFailed: %v", err)
}
again, _ := ws.ListUndelivered(10)
if len(again) != 1 || again[0].Attempts != 1 || again[0].LastError != "boom" {
t.Fatalf("失败后应仍待投递且 attempts=1, got %+v", again)
}
}
internal/webhook/notifier.go
+135
View File
@@ -0,0 +1,135 @@
// Package webhook delivers pay→business webhooks (v2, with event_type) from an
// outbox, signed with the shared HMAC scheme (双向验签,与 v1 notifyBiz 一致),
// with a background retry sweep as the backstop for lost/failed deliveries.
package webhook
import (
"bytes"
"encoding/json"
"fmt"
"io"
"log"
"net/http"
"strconv"
"strings"
"time"
"github.com/google/uuid"
"github.com/wangjia/pay/config"
"github.com/wangjia/pay/internal/store"
"github.com/wangjia/pay/internal/util"
)
// BizConfigFunc resolves a business system's callback URL + HMAC secret.
type BizConfigFunc func(system string) (config.BizSystemConfig, bool)
// OrderPaidFunc reports whether an order is settled (paid). Delivery gate:
// settle 是"先入队后翻转",崩溃窗口里 outbox 可能存在"未付单"的行——投递前必须
// 门禁,否则会把 payment.succeeded 发给业务方、白给权益。
type OrderPaidFunc func(outTradeNo string) (bool, error)
type Notifier struct {
deliveries *store.WebhookStore
bizConfig BizConfigFunc
orderPaid OrderPaidFunc
client *http.Client
}
func NewNotifier(ws *store.WebhookStore, bizConfig BizConfigFunc, orderPaid OrderPaidFunc) *Notifier {
return &Notifier{deliveries: ws, bizConfig: bizConfig, orderPaid: orderPaid,
client: &http.Client{Timeout: 10 * time.Second}}
}
// Enqueue implements gateway.WebhookEnqueuer: serialize the domain payload and
// idempotently persist it to the outbox (delivery happens async).
func (n *Notifier) Enqueue(outTradeNo, bizSystem, eventType string, data map[string]any) error {
body, err := json.Marshal(data)
if err != nil {
return fmt.Errorf("webhook.Enqueue marshal: %w", err)
}
return n.deliveries.EnqueueDelivery(outTradeNo, bizSystem, eventType, string(body))
}
// DeliverPending flushes undelivered rows; returns how many succeeded this pass.
func (n *Notifier) DeliverPending(limit int) (int, error) {
rows, err := n.deliveries.ListUndelivered(limit)
if err != nil {
return 0, err
}
ok := 0
for i := range rows {
if n.deliverOne(&rows[i]) {
ok++
}
}
return ok, nil
}
func (n *Notifier) deliverOne(d *store.WebhookDeliveryRow) bool {
// 门禁:订单未付不投递(不计失败,等 settle 翻转后自然放行)。
paid, err := n.orderPaid(d.OutTradeNo)
if err != nil || !paid {
return false
}
cfg, found := n.bizConfig(d.BizSystem)
if !found || cfg.CallbackURL == "" {
_ = n.deliveries.MarkFailed(d.ID, "biz system not configured")
return false
}
ts := strconv.FormatInt(time.Now().Unix(), 10)
nonce := uuid.NewString()
sign := util.HMACSign(cfg.Secret, d.BizSystem, ts, nonce, d.Payload)
req, err := http.NewRequest(http.MethodPost, cfg.CallbackURL, bytes.NewReader([]byte(d.Payload)))
if err != nil {
_ = n.deliveries.MarkFailed(d.ID, err.Error())
return false
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("X-Pay-System", d.BizSystem)
req.Header.Set("X-Pay-Event", d.EventType)
req.Header.Set("X-Pay-Timestamp", ts)
req.Header.Set("X-Pay-Nonce", nonce)
req.Header.Set("X-Pay-Sign", sign)
resp, err := n.client.Do(req)
if err != nil {
_ = n.deliveries.MarkFailed(d.ID, err.Error())
return false
}
rb, _ := io.ReadAll(io.LimitReader(resp.Body, 4096))
resp.Body.Close()
// 约定:业务方返回 HTTP 200 且响应含 SUCCESS 视为受理(与 v1 一致)。
if resp.StatusCode == http.StatusOK && strings.Contains(strings.ToUpper(string(rb)), "SUCCESS") {
_ = n.deliveries.MarkDelivered(d.ID)
return true
}
_ = n.deliveries.MarkFailed(d.ID, fmt.Sprintf("http %d: %s", resp.StatusCode, truncate(string(rb), 120)))
return false
}
func truncate(s string, n int) string {
if len(s) > n {
return s[:n]
}
return s
}
// Start runs a background retry sweep (backstop for lost/failed webhooks).
func (n *Notifier) Start(interval time.Duration) {
go func() {
defer func() {
if r := recover(); r != nil {
log.Printf("[webhook] retry sweep panic recovered: %v", r)
}
}()
t := time.NewTicker(interval)
defer t.Stop()
for range t.C {
if _, err := n.DeliverPending(50); err != nil {
log.Printf("[webhook] DeliverPending: %v", err)
}
}
}()
}
internal/webhook/notifier_test.go
+132
View File
@@ -0,0 +1,132 @@
package webhook_test
import (
"encoding/json"
"io"
"net/http"
"net/http/httptest"
"testing"
"github.com/wangjia/pay/config"
"github.com/wangjia/pay/internal/model"
"github.com/wangjia/pay/internal/store"
"github.com/wangjia/pay/internal/util"
"github.com/wangjia/pay/internal/webhook"
)
func TestNotifierDeliversSignedEvent(t *testing.T) {
const secret = "shh-secret"
var gotBody []byte
var gotHeaders http.Header
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
gotBody, _ = io.ReadAll(r.Body)
gotHeaders = r.Header.Clone()
w.WriteHeader(http.StatusOK)
_, _ = w.Write([]byte("SUCCESS"))
}))
defer srv.Close()
ws := store.NewWebhookStore(model.OpenTestDB(t))
bizCfg := func(system string) (config.BizSystemConfig, bool) {
if system == "pangolin" {
return config.BizSystemConfig{CallbackURL: srv.URL, Secret: secret}, true
}
return config.BizSystemConfig{}, false
}
alwaysPaid := func(string) (bool, error) { return true, nil }
n := webhook.NewNotifier(ws, bizCfg, alwaysPaid)
// 经 Enqueuer 接口入队(gateway 就是这么调的)。
err := n.Enqueue("PAY-1", "pangolin", "payment.succeeded", map[string]any{
"event_type": "payment.succeeded", "out_trade_no": "PAY-1", "amount_minor": 29990000, "currency": "USDT",
})
if err != nil {
t.Fatalf("enqueue: %v", err)
}
sent, err := n.DeliverPending(10)
if err != nil || sent != 1 {
t.Fatalf("DeliverPending = %d, %v", sent, err)
}
// 校验签名头(pay→业务方,双向 HMAC,业务方可同法验签)。
sys := gotHeaders.Get("X-Pay-System")
ts := gotHeaders.Get("X-Pay-Timestamp")
nonce := gotHeaders.Get("X-Pay-Nonce")
sign := gotHeaders.Get("X-Pay-Sign")
if gotHeaders.Get("X-Pay-Event") != "payment.succeeded" {
t.Fatalf("缺 X-Pay-Event 头")
}
if !util.HMACVerify(secret, sign, sys, ts, nonce, string(gotBody)) {
t.Fatalf("签名校验失败")
}
// body 带 event_type
var m map[string]any
_ = json.Unmarshal(gotBody, &m)
if m["event_type"] != "payment.succeeded" || m["out_trade_no"] != "PAY-1" {
t.Fatalf("body = %s", gotBody)
}
// 已标投递:再投不重发
if again, _ := n.DeliverPending(10); again != 0 {
t.Fatalf("已投递不应重发, got %d", again)
}
}
func TestNotifierRetriesOnFailure(t *testing.T) {
var hits int
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
hits++
w.WriteHeader(http.StatusInternalServerError) // 业务方暂时挂
}))
defer srv.Close()
ws := store.NewWebhookStore(model.OpenTestDB(t))
n := webhook.NewNotifier(ws, func(string) (config.BizSystemConfig, bool) {
return config.BizSystemConfig{CallbackURL: srv.URL, Secret: "x"}, true
}, func(string) (bool, error) { return true, nil })
_ = n.Enqueue("PAY-3", "pangolin", "payment.succeeded", map[string]any{"event_type": "payment.succeeded"})
if sent, _ := n.DeliverPending(10); sent != 0 {
t.Fatalf("失败不应算投递成功, got %d", sent)
}
// 仍待投递,可被下一轮重试兜底
pend, _ := ws.ListUndelivered(10)
if len(pend) != 1 || pend[0].Attempts != 1 {
t.Fatalf("失败后应留队重试, got %+v", pend)
}
if _, _ = n.DeliverPending(10); hits < 2 {
t.Fatalf("应重试第二次, hits=%d", hits)
}
}
// 投递门禁:订单未付(settle 崩在"入队后、翻转前"的窗口)绝不把 payment.succeeded
// 发给业务方;也不计失败次数,等订单翻转后自然放行。
func TestNotifierGateSkipsUnpaidOrder(t *testing.T) {
var hits int
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
hits++
w.WriteHeader(http.StatusOK)
_, _ = w.Write([]byte("SUCCESS"))
}))
defer srv.Close()
ws := store.NewWebhookStore(model.OpenTestDB(t))
paid := false
n := webhook.NewNotifier(ws, func(string) (config.BizSystemConfig, bool) {
return config.BizSystemConfig{CallbackURL: srv.URL, Secret: "x"}, true
}, func(string) (bool, error) { return paid, nil })
_ = n.Enqueue("PAY-4", "pangolin", "payment.succeeded", map[string]any{"event_type": "payment.succeeded"})
if sent, _ := n.DeliverPending(10); sent != 0 || hits != 0 {
t.Fatalf("未付单不应投递, sent=%d hits=%d", sent, hits)
}
pend, _ := ws.ListUndelivered(10)
if len(pend) != 1 || pend[0].Attempts != 0 {
t.Fatalf("门禁跳过不应计失败, got %+v", pend)
}
paid = true // 订单翻转后放行
if sent, _ := n.DeliverPending(10); sent != 1 || hits != 1 {
t.Fatalf("翻转后应投递, sent=%d hits=%d", sent, hits)
}
}