yml2213
ef8b8a7213
- 签名串由 timestamp+换行+bodyhash 改为 body_sha256=<hash>×tamp=<unix秒>（ASCII 字典序 & 拼接） - 同步更新对接文档
434 lines
14 KiB
Go
434 lines
14 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"
|
|
"gorm.io/gorm/clause"
|
|
)
|
|
|
|
// 回调推送策略（参考微信/支付宝通知机制）：
|
|
// 首次推送失败后按固定序列退避重试，默认共 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
|
|
Status string
|
|
// RotateSecret 重置回调密钥：重新生成 secret（旧密钥立即失效），返回新 secret 仅此一次。
|
|
RotateSecret bool
|
|
}
|
|
|
|
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 == "" {
|
|
in.Name = "默认回调"
|
|
}
|
|
if err := validateCallbackURL(in.URL); err != nil {
|
|
return nil, err
|
|
}
|
|
events := scopeList(in.Events)
|
|
if len(events) == 0 {
|
|
return nil, errors.New("至少订阅一个事件")
|
|
}
|
|
if in.Status == "" {
|
|
in.Status = model.CallbackStatusActive
|
|
}
|
|
if in.Status != model.CallbackStatusActive && in.Status != model.CallbackStatusDisabled {
|
|
return nil, errors.New("无效的回调状态")
|
|
}
|
|
|
|
subscription := &model.CallbackSubscription{}
|
|
secret := ""
|
|
err := s.db.Transaction(func(tx *gorm.DB) error {
|
|
var merchant model.Merchant
|
|
if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).
|
|
Where("id = ? AND status = ?", merchantID, model.MerchantStatusActive).
|
|
First(&merchant).Error; err != nil {
|
|
return errors.New("商户不存在或已禁用")
|
|
}
|
|
|
|
err := tx.Where("merchant_id = ?", merchantID).Order("id DESC").First(subscription).Error
|
|
if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
|
|
return err
|
|
}
|
|
action := "callback_subscription.update"
|
|
rotate := in.RotateSecret
|
|
if errors.Is(err, gorm.ErrRecordNotFound) {
|
|
rotate = true // 新建时必然生成新密钥
|
|
action = "callback_subscription.create"
|
|
}
|
|
if rotate {
|
|
var tokenErr error
|
|
secret, tokenErr = randomToken("cb_", 32)
|
|
if tokenErr != nil {
|
|
return tokenErr
|
|
}
|
|
ciphertext, encryptErr := s.codec.Encrypt(secret)
|
|
if encryptErr != nil {
|
|
return encryptErr
|
|
}
|
|
if subscription.ID == 0 {
|
|
*subscription = model.CallbackSubscription{
|
|
MerchantID: merchantID,
|
|
SecretCiphertext: ciphertext,
|
|
}
|
|
} else {
|
|
subscription.SecretCiphertext = ciphertext
|
|
}
|
|
}
|
|
subscription.Name = in.Name
|
|
subscription.URL = in.URL
|
|
subscription.Events = strings.Join(events, ",")
|
|
subscription.Status = in.Status
|
|
if subscription.ID == 0 {
|
|
if err := tx.Create(subscription).Error; err != nil {
|
|
return err
|
|
}
|
|
} else if err := tx.Save(subscription).Error; err != nil {
|
|
return err
|
|
}
|
|
if err := tx.Model(&model.CallbackSubscription{}).
|
|
Where("merchant_id = ? AND id <> ? AND status = ?", merchantID, subscription.ID, model.CallbackStatusActive).
|
|
Update("status", model.CallbackStatusDisabled).Error; err != nil {
|
|
return err
|
|
}
|
|
return writeAudit(tx, &merchantID, &actorUserID, nil, action, "callback_subscription", fmt.Sprint(subscription.ID), map[string]string{"url": subscription.URL, "status": subscription.Status})
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return &CallbackCredential{Subscription: subscription, Secret: secret}, nil
|
|
}
|
|
|
|
func (s *CallbackService) GetSubscription(merchantID uint) (*model.CallbackSubscription, error) {
|
|
var subscription model.CallbackSubscription
|
|
err := s.db.Where("merchant_id = ?", merchantID).Order("id DESC").First(&subscription).Error
|
|
if errors.Is(err, gorm.ErrRecordNotFound) {
|
|
return nil, nil
|
|
}
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return &subscription, nil
|
|
}
|
|
|
|
// Enqueue 在调用方事务中写入回调 outbox，只有订单事务成功才会发送事件。
|
|
func (s *CallbackService) Enqueue(tx *gorm.DB, merchantID uint, event string, data interface{}) error {
|
|
var subscription model.CallbackSubscription
|
|
err := tx.Where("merchant_id = ?", merchantID).
|
|
Order("id DESC").
|
|
First(&subscription).Error
|
|
if errors.Is(err, gorm.ErrRecordNotFound) {
|
|
return nil
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if subscription.Status != model.CallbackStatusActive {
|
|
return nil
|
|
}
|
|
if !subscribesTo(subscription.Events, event) {
|
|
return nil
|
|
}
|
|
now := time.Now()
|
|
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,
|
|
}
|
|
return tx.Create(&delivery).Error
|
|
}
|
|
|
|
// 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 供接收方校验。签名风格与开放 API 一致：
|
|
// 参数按 ASCII 字典序 + "&" 拼接，body 以 SHA256 摘要参与签名。
|
|
//
|
|
// content = body_sha256=<sha256hex(body)>×tamp=<unix秒>
|
|
// X-Sign = hex(HMAC-SHA256(secret, content))
|
|
func BuildCallbackSign(secret, timestamp, body string) string {
|
|
bodyHash := sha256.Sum256([]byte(body))
|
|
content := "body_sha256=" + hex.EncodeToString(bodyHash[:]) + "×tamp=" + timestamp
|
|
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]
|
|
}
|