From 3f5e44581f787b1ae8f5727f6a1b6dbd2c2de986 Mon Sep 17 00:00:00 2001 From: wangjia <809946525@qq.com> Date: Fri, 10 Jul 2026 11:25:14 +0800 Subject: [PATCH] =?UTF-8?q?feat(v2):=20webhook=20v2=20outbox=20+=20Notifie?= =?UTF-8?q?r(event=5Ftype=20+=20HMAC=20=E5=8F=8C=E5=90=91=20+=20=E9=87=8D?= =?UTF-8?q?=E8=AF=95=E5=85=9C=E5=BA=95)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/model/testdb.go | 2 +- internal/model/webhook_delivery.go | 14 +++ internal/store/webhook.go | 68 +++++++++++++++ internal/store/webhook_test.go | 44 ++++++++++ internal/webhook/notifier.go | 135 +++++++++++++++++++++++++++++ internal/webhook/notifier_test.go | 132 ++++++++++++++++++++++++++++ 6 files changed, 394 insertions(+), 1 deletion(-) create mode 100644 internal/model/webhook_delivery.go create mode 100644 internal/store/webhook.go create mode 100644 internal/store/webhook_test.go create mode 100644 internal/webhook/notifier.go create mode 100644 internal/webhook/notifier_test.go diff --git a/internal/model/testdb.go b/internal/model/testdb.go index 61d8ad0..7be7bdc 100644 --- a/internal/model/testdb.go +++ b/internal/model/testdb.go @@ -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() diff --git a/internal/model/webhook_delivery.go b/internal/model/webhook_delivery.go new file mode 100644 index 0000000..ba1c1c4 --- /dev/null +++ b/internal/model/webhook_delivery.go @@ -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"` +} diff --git a/internal/store/webhook.go b/internal/store/webhook.go new file mode 100644 index 0000000..891ce5f --- /dev/null +++ b/internal/store/webhook.go @@ -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 +} diff --git a/internal/store/webhook_test.go b/internal/store/webhook_test.go new file mode 100644 index 0000000..6f51d53 --- /dev/null +++ b/internal/store/webhook_test.go @@ -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) + } +} diff --git a/internal/webhook/notifier.go b/internal/webhook/notifier.go new file mode 100644 index 0000000..d2a9540 --- /dev/null +++ b/internal/webhook/notifier.go @@ -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) + } + } + }() +} diff --git a/internal/webhook/notifier_test.go b/internal/webhook/notifier_test.go new file mode 100644 index 0000000..3a743c7 --- /dev/null +++ b/internal/webhook/notifier_test.go @@ -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) + } +}