pay/internal/webhook/notifier_test.go

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)
	}
}