yml2213
cf2ee96baa
- 回调重试改为微信式固定退避序列（15s/15s/30s/3m/10m/20m/30m/30m/30m/1h/3h/3h/3h/6h/6h），共 16 次尝试 - 重试次数/推送超时/退避序列可通过环境变量配置（CALLBACK_MAX_ATTEMPTS/CALLBACK_PUSH_TIMEOUT_SECONDS/CALLBACK_RETRY_SCHEDULE） - 每个商户仅允许 1 个 active 回调订阅，创建前校验，先停用后新建 - 前端回调页提示推送策略与单地址限制 - docker-compose 透传回调相关配置
406 lines
13 KiB
Go
406 lines
13 KiB
Go
package service
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"crypto/hmac"
|
|
"crypto/sha256"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"log"
|
|
"net/http"
|
|
"net/url"
|
|
"strings"
|
|
"time"
|
|
|
|
"affiliate_dash/internal/model"
|
|
"affiliate_dash/internal/pkg/timeutil"
|
|
|
|
"github.com/google/uuid"
|
|
"gorm.io/gorm"
|
|
)
|
|
|
|
// 回调推送策略（参考微信/支付宝通知机制）：
|
|
// 首次推送失败后按固定序列退避重试，默认共 16 次尝试，之后标记 failed 不再推送。
|
|
// 序列与总次数均可通过环境变量覆盖（CALLBACK_RETRY_SCHEDULE / CALLBACK_MAX_ATTEMPTS）。
|
|
var defaultCallbackRetrySchedule = []time.Duration{
|
|
15 * time.Second, // 第 2 次
|
|
15 * time.Second, // 第 3 次
|
|
30 * time.Second, // 第 4 次
|
|
3 * time.Minute, // 第 5 次
|
|
10 * time.Minute, // 第 6 次
|
|
20 * time.Minute, // 第 7 次
|
|
30 * time.Minute, // 第 8 次
|
|
30 * time.Minute, // 第 9 次
|
|
30 * time.Minute, // 第 10 次
|
|
60 * time.Minute, // 第 11 次
|
|
3 * time.Hour, // 第 12 次
|
|
3 * time.Hour, // 第 13 次
|
|
3 * time.Hour, // 第 14 次
|
|
6 * time.Hour, // 第 15 次
|
|
6 * time.Hour, // 第 16 次
|
|
}
|
|
|
|
// CallbackConfig 回调推送策略配置。
|
|
type CallbackConfig struct {
|
|
MaxAttempts int
|
|
PushTimeout time.Duration
|
|
// RetrySchedule 第 N 次重试前的等待间隔；不足时复用最后一个间隔。
|
|
RetrySchedule []time.Duration
|
|
}
|
|
|
|
// CallbackService 以数据库 outbox 方式管理回调，进程重启不会丢失待发送事件。
|
|
type CallbackService struct {
|
|
db *gorm.DB
|
|
codec *SecretCodec
|
|
httpClient *http.Client
|
|
maxAttempts int
|
|
retrySchedule []time.Duration
|
|
}
|
|
|
|
func NewCallbackService(db *gorm.DB, codec *SecretCodec, opts ...func(*CallbackConfig)) *CallbackService {
|
|
cfg := CallbackConfig{
|
|
MaxAttempts: 16,
|
|
PushTimeout: 15 * time.Second,
|
|
RetrySchedule: defaultCallbackRetrySchedule,
|
|
}
|
|
for _, opt := range opts {
|
|
opt(&cfg)
|
|
}
|
|
if cfg.MaxAttempts < 1 {
|
|
cfg.MaxAttempts = 16
|
|
}
|
|
if cfg.PushTimeout <= 0 {
|
|
cfg.PushTimeout = 15 * time.Second
|
|
}
|
|
if len(cfg.RetrySchedule) == 0 {
|
|
cfg.RetrySchedule = defaultCallbackRetrySchedule
|
|
}
|
|
return &CallbackService{
|
|
db: db,
|
|
codec: codec,
|
|
httpClient: &http.Client{Timeout: cfg.PushTimeout},
|
|
maxAttempts: cfg.MaxAttempts,
|
|
retrySchedule: cfg.RetrySchedule,
|
|
}
|
|
}
|
|
|
|
// WithCallbackConfig 便捷的配置选项，供 main 装配。
|
|
func WithCallbackConfig(cfg CallbackConfig) func(*CallbackConfig) {
|
|
return func(target *CallbackConfig) { *target = cfg }
|
|
}
|
|
|
|
func (s *CallbackService) SetHTTPClient(client *http.Client) {
|
|
if client != nil {
|
|
s.httpClient = client
|
|
}
|
|
}
|
|
|
|
type CreateCallbackInput struct {
|
|
Name string
|
|
URL string
|
|
Events string
|
|
}
|
|
|
|
type CallbackCredential struct {
|
|
Subscription *model.CallbackSubscription `json:"subscription"`
|
|
Secret string `json:"secret"`
|
|
}
|
|
|
|
func (s *CallbackService) CreateSubscription(merchantID uint, in CreateCallbackInput, actorUserID uint) (*CallbackCredential, error) {
|
|
in.Name = strings.TrimSpace(in.Name)
|
|
in.URL = strings.TrimSpace(in.URL)
|
|
if in.Name == "" {
|
|
return nil, errors.New("回调名称不能为空")
|
|
}
|
|
if err := validateCallbackURL(in.URL); err != nil {
|
|
return nil, err
|
|
}
|
|
events := scopeList(in.Events)
|
|
if len(events) == 0 {
|
|
return nil, errors.New("至少订阅一个事件")
|
|
}
|
|
// 每个商户仅允许一个回调地址：存在 active 订阅时拒绝创建，先停用旧的再新建。
|
|
var activeCount int64
|
|
if err := s.db.Model(&model.CallbackSubscription{}).
|
|
Where("merchant_id = ? AND status = ?", merchantID, model.CallbackStatusActive).
|
|
Count(&activeCount).Error; err != nil {
|
|
return nil, err
|
|
}
|
|
if activeCount > 0 {
|
|
return nil, errors.New("每个商户仅支持一个回调地址，请先停用当前订阅再创建")
|
|
}
|
|
secret, err := randomToken("cb_", 32)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
ciphertext, err := s.codec.Encrypt(secret)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
subscription := &model.CallbackSubscription{
|
|
MerchantID: merchantID,
|
|
Name: in.Name,
|
|
URL: in.URL,
|
|
Events: strings.Join(events, ","),
|
|
SecretCiphertext: ciphertext,
|
|
Status: model.CallbackStatusActive,
|
|
}
|
|
err = s.db.Transaction(func(tx *gorm.DB) error {
|
|
var merchant model.Merchant
|
|
if err := tx.Where("id = ? AND status = ?", merchantID, model.MerchantStatusActive).First(&merchant).Error; err != nil {
|
|
return errors.New("商户不存在或已禁用")
|
|
}
|
|
if err := tx.Create(subscription).Error; err != nil {
|
|
return err
|
|
}
|
|
return writeAudit(tx, &merchantID, &actorUserID, nil, "callback_subscription.create", "callback_subscription", fmt.Sprint(subscription.ID), map[string]string{"url": subscription.URL})
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return &CallbackCredential{Subscription: subscription, Secret: secret}, nil
|
|
}
|
|
|
|
func (s *CallbackService) ListSubscriptions(merchantID uint) ([]model.CallbackSubscription, error) {
|
|
var subscriptions []model.CallbackSubscription
|
|
err := s.db.Where("merchant_id = ?", merchantID).Order("id DESC").Find(&subscriptions).Error
|
|
return subscriptions, err
|
|
}
|
|
|
|
func (s *CallbackService) UpdateSubscriptionStatus(merchantID, id uint, status string, actorUserID uint) error {
|
|
if status != model.CallbackStatusActive && status != model.CallbackStatusDisabled {
|
|
return errors.New("无效的回调状态")
|
|
}
|
|
return s.db.Transaction(func(tx *gorm.DB) error {
|
|
result := tx.Model(&model.CallbackSubscription{}).
|
|
Where("id = ? AND merchant_id = ?", id, merchantID).
|
|
Update("status", status)
|
|
if result.Error != nil {
|
|
return result.Error
|
|
}
|
|
if result.RowsAffected == 0 {
|
|
return errors.New("回调订阅不存在")
|
|
}
|
|
return writeAudit(tx, &merchantID, &actorUserID, nil, "callback_subscription.status.update", "callback_subscription", fmt.Sprint(id), map[string]string{"status": status})
|
|
})
|
|
}
|
|
|
|
// Enqueue 在调用方事务中写入回调 outbox，只有订单事务成功才会发送事件。
|
|
func (s *CallbackService) Enqueue(tx *gorm.DB, merchantID uint, event string, data interface{}) error {
|
|
var subscriptions []model.CallbackSubscription
|
|
if err := tx.Where("merchant_id = ? AND status = ?", merchantID, model.CallbackStatusActive).Find(&subscriptions).Error; err != nil {
|
|
return err
|
|
}
|
|
now := time.Now()
|
|
for _, subscription := range subscriptions {
|
|
if !subscribesTo(subscription.Events, event) {
|
|
continue
|
|
}
|
|
eventID := uuid.NewString()
|
|
payload, err := json.Marshal(map[string]interface{}{
|
|
"event_id": eventID,
|
|
"event": event,
|
|
"occurred_at": timeutil.FormatAPITime(now),
|
|
"data": data,
|
|
})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
delivery := model.CallbackDelivery{
|
|
MerchantID: merchantID,
|
|
CallbackSubscriptionID: subscription.ID,
|
|
EventID: eventID,
|
|
Event: event,
|
|
Payload: string(payload),
|
|
Status: model.CallbackDeliveryPending,
|
|
NextAttemptAt: now,
|
|
}
|
|
if err := tx.Create(&delivery).Error; err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// DispatchDue 执行一批可发送的 outbox 记录。返回成功/失败尝试数，便于日志监控。
|
|
func (s *CallbackService) DispatchDue(ctx context.Context, limit int) (int, int, error) {
|
|
if limit <= 0 || limit > 100 {
|
|
limit = 50
|
|
}
|
|
now := time.Now()
|
|
var deliveries []model.CallbackDelivery
|
|
err := s.db.Preload("CallbackSubscription").
|
|
Where("(status = ? AND next_attempt_at <= ?) OR (status = ? AND updated_at <= ?)",
|
|
model.CallbackDeliveryPending, now,
|
|
model.CallbackDeliverySending, now.Add(-5*time.Minute),
|
|
).
|
|
Order("next_attempt_at ASC, id ASC").
|
|
Limit(limit).
|
|
Find(&deliveries).Error
|
|
if err != nil {
|
|
return 0, 0, err
|
|
}
|
|
successes, failures := 0, 0
|
|
for _, delivery := range deliveries {
|
|
ok, attempted, err := s.dispatchOne(ctx, delivery.ID)
|
|
if err != nil {
|
|
return successes, failures, err
|
|
}
|
|
if !attempted {
|
|
continue
|
|
}
|
|
if ok {
|
|
successes++
|
|
} else {
|
|
failures++
|
|
}
|
|
}
|
|
return successes, failures, nil
|
|
}
|
|
|
|
func (s *CallbackService) dispatchOne(ctx context.Context, id uint) (bool, bool, error) {
|
|
now := time.Now()
|
|
claimed := s.db.Model(&model.CallbackDelivery{}).
|
|
Where("id = ? AND ((status = ? AND next_attempt_at <= ?) OR (status = ? AND updated_at <= ?))",
|
|
id, model.CallbackDeliveryPending, now,
|
|
model.CallbackDeliverySending, now.Add(-5*time.Minute),
|
|
).
|
|
Update("status", model.CallbackDeliverySending)
|
|
if claimed.Error != nil {
|
|
return false, false, claimed.Error
|
|
}
|
|
if claimed.RowsAffected == 0 {
|
|
return false, false, nil
|
|
}
|
|
|
|
var delivery model.CallbackDelivery
|
|
if err := s.db.Preload("CallbackSubscription").First(&delivery, id).Error; err != nil {
|
|
return false, true, err
|
|
}
|
|
if delivery.CallbackSubscription == nil || delivery.CallbackSubscription.Status != model.CallbackStatusActive {
|
|
return false, true, s.recordCallbackFailure(delivery, 0, "回调订阅已禁用")
|
|
}
|
|
secret, err := s.codec.Decrypt(delivery.CallbackSubscription.SecretCiphertext)
|
|
if err != nil {
|
|
return false, true, s.recordCallbackFailure(delivery, 0, "回调密钥不可用")
|
|
}
|
|
|
|
timestamp := fmt.Sprint(time.Now().Unix())
|
|
req, err := http.NewRequestWithContext(ctx, http.MethodPost, delivery.CallbackSubscription.URL, bytes.NewBufferString(delivery.Payload))
|
|
if err != nil {
|
|
return false, true, s.recordCallbackFailure(delivery, 0, "创建回调请求失败")
|
|
}
|
|
req.Header.Set("Content-Type", "application/json")
|
|
req.Header.Set("X-Event-ID", delivery.EventID)
|
|
req.Header.Set("X-Timestamp", timestamp)
|
|
req.Header.Set("X-Sign", BuildCallbackSign(secret, timestamp, delivery.Payload))
|
|
|
|
resp, err := s.httpClient.Do(req)
|
|
if err != nil {
|
|
log.Printf("[callback] push fail event=%s event_id=%s merchant_id=%d url=%s err=%s", delivery.Event, delivery.EventID, delivery.MerchantID, delivery.CallbackSubscription.URL, truncateCallbackError(err.Error()))
|
|
return false, true, s.recordCallbackFailure(delivery, 0, truncateCallbackError(err.Error()))
|
|
}
|
|
defer resp.Body.Close()
|
|
body, _ := io.ReadAll(io.LimitReader(resp.Body, 4096))
|
|
if resp.StatusCode >= http.StatusOK && resp.StatusCode < http.StatusMultipleChoices {
|
|
log.Printf("[callback] push ok event=%s event_id=%s merchant_id=%d url=%s status=%d", delivery.Event, delivery.EventID, delivery.MerchantID, delivery.CallbackSubscription.URL, resp.StatusCode)
|
|
return true, true, s.db.Model(&model.CallbackDelivery{}).Where("id = ?", delivery.ID).Updates(map[string]interface{}{
|
|
"status": model.CallbackDeliveryDelivered,
|
|
"attempts": delivery.Attempts + 1,
|
|
"last_status_code": resp.StatusCode,
|
|
"last_response": string(body),
|
|
"delivered_at": time.Now(),
|
|
}).Error
|
|
}
|
|
log.Printf("[callback] push fail event=%s event_id=%s merchant_id=%d url=%s status=%d body=%s", delivery.Event, delivery.EventID, delivery.MerchantID, delivery.CallbackSubscription.URL, resp.StatusCode, truncateCallbackError(string(body)))
|
|
return false, true, s.recordCallbackFailure(delivery, resp.StatusCode, string(body))
|
|
}
|
|
|
|
func (s *CallbackService) recordCallbackFailure(delivery model.CallbackDelivery, statusCode int, response string) error {
|
|
attempts := delivery.Attempts + 1
|
|
status := model.CallbackDeliveryPending
|
|
nextAttempt := time.Now().Add(s.callbackRetryDelay(attempts))
|
|
if attempts >= s.maxAttempts {
|
|
status = model.CallbackDeliveryFailed
|
|
nextAttempt = time.Now()
|
|
}
|
|
return s.db.Model(&model.CallbackDelivery{}).Where("id = ?", delivery.ID).Updates(map[string]interface{}{
|
|
"status": status,
|
|
"attempts": attempts,
|
|
"next_attempt_at": nextAttempt,
|
|
"last_status_code": statusCode,
|
|
"last_response": truncateCallbackError(response),
|
|
}).Error
|
|
}
|
|
|
|
func (s *CallbackService) Run(ctx context.Context) {
|
|
ticker := time.NewTicker(3 * time.Second)
|
|
defer ticker.Stop()
|
|
for {
|
|
successes, failures, err := s.DispatchDue(ctx, 50)
|
|
if err != nil {
|
|
log.Printf("[callback] dispatch error: %v", err)
|
|
} else if successes > 0 || failures > 0 {
|
|
log.Printf("[callback] dispatch done successes=%d failures=%d", successes, failures)
|
|
}
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-ticker.C:
|
|
}
|
|
}
|
|
}
|
|
|
|
// BuildCallbackSign 供接收方校验：HMAC-SHA256(secret, timestamp + "\n" + sha256(body))。
|
|
func BuildCallbackSign(secret, timestamp, body string) string {
|
|
bodyHash := sha256.Sum256([]byte(body))
|
|
content := timestamp + "\n" + hex.EncodeToString(bodyHash[:])
|
|
mac := hmac.New(sha256.New, []byte(secret))
|
|
_, _ = mac.Write([]byte(content))
|
|
return hex.EncodeToString(mac.Sum(nil))
|
|
}
|
|
|
|
func subscribesTo(events, event string) bool {
|
|
for subscribed := range ParseScopes(events) {
|
|
if subscribed == "*" || subscribed == event {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
func validateCallbackURL(rawURL string) error {
|
|
parsed, err := url.Parse(rawURL)
|
|
if err != nil || parsed.Scheme == "" || parsed.Host == "" {
|
|
return errors.New("回调地址格式错误")
|
|
}
|
|
if parsed.Scheme != "https" && parsed.Scheme != "http" {
|
|
return errors.New("回调地址仅支持 http 或 https")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// callbackRetryDelay 返回第 attempts 次失败后的重试间隔（微信式固定序列）。
|
|
// 序列用尽后复用最后一个间隔。
|
|
func (s *CallbackService) callbackRetryDelay(attempts int) time.Duration {
|
|
if attempts < 1 {
|
|
attempts = 1
|
|
}
|
|
idx := attempts - 1
|
|
if idx >= len(s.retrySchedule) {
|
|
idx = len(s.retrySchedule) - 1
|
|
}
|
|
return s.retrySchedule[idx]
|
|
}
|
|
|
|
func truncateCallbackError(value string) string {
|
|
if len(value) <= 4096 {
|
|
return value
|
|
}
|
|
return value[:4096]
|
|
}
|