pay/internal/webhook/notifier.go

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

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 {
		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.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
}

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