package webhook_test import ( "encoding/json" "fmt" "io" "net/http" "net/http/httptest" "testing" "time" "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) } } // 业务方持续 500:每次失败按指数退避重排;时钟推进后才重投;达上限标死信 + 告警。 func TestNotifierBackoffThenDeadWithAlert(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() clk := time.Date(2026, 7, 10, 12, 0, 0, 0, time.UTC) nowFn := func() time.Time { return clk } var alerted []string 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 }, webhook.WithClock(nowFn), webhook.WithBaseBackoff(time.Second), webhook.WithMaxBackoff(4*time.Second), webhook.WithMaxAttempts(3), webhook.WithAlerter(func(d *store.WebhookDeliveryRow, reason string) { alerted = append(alerted, d.OutTradeNo) }), ) _ = n.Enqueue("PAY-D", "pangolin", "payment.succeeded", "", map[string]any{"event_type": "payment.succeeded"}) // 尝试 1:失败 → attempts=1,退避到 +1s。 if sent, _ := n.DeliverPending(10); sent != 0 || hits != 1 { t.Fatalf("try1 sent=%d hits=%d", sent, hits) } // 退避窗内不投。 if sent, _ := n.DeliverPending(10); sent != 0 || hits != 1 { t.Fatalf("退避窗内不应再敲, hits=%d", hits) } // 推进越过退避;尝试 2 失败 → attempts=2,退避到 +2s。 clk = clk.Add(2 * time.Second) if sent, _ := n.DeliverPending(10); sent != 0 || hits != 2 { t.Fatalf("try2 hits=%d", hits) } // 推进;尝试 3 失败 → attempts 达 maxAttempts(3)→ 死信 + 告警。 clk = clk.Add(4 * time.Second) if sent, _ := n.DeliverPending(10); sent != 0 || hits != 3 { t.Fatalf("try3 hits=%d", hits) } if len(alerted) != 1 || alerted[0] != "PAY-D" { t.Fatalf("死信应触发告警一次, got %+v", alerted) } // 已死信:无论时钟怎么走都不再投。 clk = clk.Add(time.Hour) if sent, _ := n.DeliverPending(10); sent != 0 || hits != 3 { t.Fatalf("死信后不应再投, hits=%d", hits) } } 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() clk := time.Date(2026, 7, 10, 12, 0, 0, 0, time.UTC) 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 }, webhook.WithClock(func() time.Time { return clk }), webhook.WithBaseBackoff(time.Second)) _ = 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) } clk = clk.Add(2 * time.Second) // 越过退避窗 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) } } // 事件集过滤(P8 Task7):业务方未声明支持的事件不投递、直接标 delivered(视为已受理, // 不占重试),已声明的事件正常投。SupportedEvents 空 → 只收 payment.succeeded(向后兼容 // v1/P2 接入方,不会突然收到新事件把它们搞崩)。 func TestNotifierFiltersUnsupportedEvents(t *testing.T) { var hits []string srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { hits = append(hits, r.Header.Get("X-Pay-Event")) w.WriteHeader(http.StatusOK) _, _ = w.Write([]byte("SUCCESS")) })) defer srv.Close() ws := store.NewWebhookStore(model.OpenTestDB(t)) n := webhook.NewNotifier(ws, func(string) (config.BizSystemConfig, bool) { // 只声明支持 payment.succeeded(空集等价语义的显式形式,顺带覆盖非空单元素集)。 return config.BizSystemConfig{CallbackURL: srv.URL, Secret: "x", SupportedEvents: []string{"payment.succeeded"}}, true }, func(string) (bool, error) { return true, nil }) _ = n.Enqueue("PAY-5", "pangolin", "subscription.renewed", "", map[string]any{"event_type": "subscription.renewed"}) _ = n.Enqueue("PAY-6", "pangolin", "payment.succeeded", "", map[string]any{"event_type": "payment.succeeded"}) // deliverOne 对"未订阅事件"也返回 true(视为已受理、直接标 delivered,见 notifier.go // eventSupported 分支的注释),所以 sent 计两条;真正发出的 POST 只应有 1 次。 sent, err := n.DeliverPending(10) if err != nil || sent != 2 { t.Fatalf("DeliverPending = %d, %v, want 2(1 条真投 + 1 条未订阅直接标 delivered)", sent, err) } if len(hits) != 1 || hits[0] != "payment.succeeded" { t.Fatalf("只应收到 payment.succeeded 一次 POST, got %v", hits) } // 未订阅的事件已标 delivered(视为已受理),不占重试队列。 pend, _ := ws.ListUndelivered(10) if len(pend) != 0 { t.Fatalf("未订阅事件应标 delivered、不留队, got %+v", pend) } } // 声明了全部 P8 事件集的业务方:所有事件都应正常投递(不被过滤)。 func TestNotifierDeliversAllDeclaredEvents(t *testing.T) { var hits []string srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { hits = append(hits, r.Header.Get("X-Pay-Event")) w.WriteHeader(http.StatusOK) _, _ = w.Write([]byte("SUCCESS")) })) defer srv.Close() all := []string{ "payment.succeeded", "subscription.renewed", "subscription.created", "subscription.canceled", "subscription.past_due", "chargeback.received", } ws := store.NewWebhookStore(model.OpenTestDB(t)) n := webhook.NewNotifier(ws, func(string) (config.BizSystemConfig, bool) { return config.BizSystemConfig{CallbackURL: srv.URL, Secret: "x", SupportedEvents: all}, true }, func(string) (bool, error) { return true, nil }) for i, ev := range all { _ = n.Enqueue(fmt.Sprintf("PAY-ALL-%d", i), "pangolin", ev, "", map[string]any{"event_type": ev}) } sent, err := n.DeliverPending(10) if err != nil || sent != len(all) { t.Fatalf("DeliverPending = %d, %v, want %d(声明集内全投)", sent, err, len(all)) } if len(hits) != len(all) { t.Fatalf("应收到 %d 次 POST, got %v", len(all), hits) } }