// 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" "unicode/utf8" "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 now func() time.Time maxAttempts int baseBackoff time.Duration maxBackoff time.Duration alert func(d *store.WebhookDeliveryRow, reason string) } // Option customizes a Notifier's retry/backoff/alerting behavior (functional options). type Option func(*Notifier) // WithClock overrides the time source (tests inject a fake clock to drive backoff windows). func WithClock(f func() time.Time) Option { return func(n *Notifier) { n.now = f } } // WithMaxAttempts sets how many failed attempts before a delivery is marked dead. func WithMaxAttempts(m int) Option { return func(n *Notifier) { n.maxAttempts = m } } // WithBaseBackoff sets the base duration for exponential backoff (attempt 1). func WithBaseBackoff(d time.Duration) Option { return func(n *Notifier) { n.baseBackoff = d } } // WithMaxBackoff caps the exponential backoff duration. func WithMaxBackoff(d time.Duration) Option { return func(n *Notifier) { n.maxBackoff = d } } // WithAlerter overrides the dead-letter alert hook (default logs). func WithAlerter(a func(d *store.WebhookDeliveryRow, reason string)) Option { return func(n *Notifier) { n.alert = a } } // NewNotifier builds a Notifier. The original 3-arg call form keeps compiling // (opts is variadic); pass Option values to customize clock/backoff/alerting. func NewNotifier(ws *store.WebhookStore, bizConfig BizConfigFunc, orderPaid OrderPaidFunc, opts ...Option) *Notifier { n := &Notifier{ deliveries: ws, bizConfig: bizConfig, orderPaid: orderPaid, client: &http.Client{Timeout: 10 * time.Second}, now: time.Now, maxAttempts: 12, // ~ 覆盖数小时退避后放弃(见 backoffFor 封顶) baseBackoff: 30 * time.Second, // 首次失败退避基 maxBackoff: time.Hour, // 单次退避封顶 } for _, o := range opts { o(n) } if n.alert == nil { n.alert = func(d *store.WebhookDeliveryRow, reason string) { log.Printf("[webhook][死信] out_trade_no=%s biz=%s event=%s attempts=%d 放弃投递: %s", d.OutTradeNo, d.BizSystem, d.EventType, d.Attempts, reason) } } return n } // backoffFor 计算第 attempts 次失败后的退避:base·2^(attempts-1),封顶 maxBackoff。 // attempts≥1;移位用循环倍增且封顶,防溢出。 func (n *Notifier) backoffFor(attempts int) time.Duration { d := n.baseBackoff for i := 1; i < attempts; i++ { d *= 2 if d >= n.maxBackoff { return n.maxBackoff } } if d > n.maxBackoff { return n.maxBackoff } return d } // fail 统一失败分流:未达上限→退避重排;达上限→死信+告警。 func (n *Notifier) fail(d *store.WebhookDeliveryRow, msg string) { attempts := d.Attempts + 1 // 本次即将记的失败次数 if attempts >= n.maxAttempts { _ = n.deliveries.MarkDead(d.ID, msg) n.alert(d, msg) return } _ = n.deliveries.ScheduleRetry(d.ID, msg, n.now().Add(n.backoffFor(attempts))) } // Enqueue implements gateway.WebhookEnqueuer: serialize the domain payload and // idempotently persist it to the outbox (delivery happens async). refundID is // the refund's idempotency dimension (payment events pass ""). func (n *Notifier) Enqueue(outTradeNo, bizSystem, eventType, refundID 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, refundID, string(body)) } // DeliverPending flushes deliverable rows (未投递、未死信、退避到点); returns how // many succeeded this pass. func (n *Notifier) DeliverPending(limit int) (int, error) { rows, err := n.deliveries.ListDeliverable(n.now(), 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 { log.Printf("[webhook] 投递门禁查单失败 %s: %v", d.OutTradeNo, err) return false } if !paid { return false } cfg, found := n.bizConfig(d.BizSystem) if !found || cfg.CallbackURL == "" { n.fail(d, "biz system not configured") return false } if !eventSupported(cfg.SupportedEvents, d.EventType) { _ = n.deliveries.MarkDelivered(d.ID) // 业务方未订阅该事件:视为已受理,不投递、不重试 return true } 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.fail(d, 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.fail(d, 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.fail(d, fmt.Sprintf("http %d: %s", resp.StatusCode, truncate(string(rb), 120))) return false } // eventSupported 判断业务方是否声明支持该 event_type(设计 §「接入方显式声明支持事件 // 集」):list 为空 → 仅 payment.succeeded 为真(v1/P2 接入方向后兼容,不会突然收到 // subscription.*/chargeback.received 等新事件把它们搞崩);非空则按声明集精确匹配。 func eventSupported(list []string, ev string) bool { if len(list) == 0 { return ev == "payment.succeeded" } for _, e := range list { if e == ev { return true } } return false } // truncate safely truncates a string to n bytes without splitting UTF-8 runes. func truncate(s string, n int) string { if len(s) <= n { return s } s = s[:n] for len(s) > 0 && !utf8.ValidString(s) { s = s[:len(s)-1] } 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) } } }() }