331 lines
10 KiB
Go
331 lines
10 KiB
Go
package service
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"crypto/hmac"
|
|
"crypto/sha256"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"net/http"
|
|
"net/url"
|
|
"strings"
|
|
"time"
|
|
|
|
"affiliate_dash/internal/model"
|
|
|
|
"github.com/google/uuid"
|
|
"gorm.io/gorm"
|
|
)
|
|
|
|
const maxCallbackAttempts = 8
|
|
|
|
// CallbackService 以数据库 outbox 方式管理回调，进程重启不会丢失待发送事件。
|
|
type CallbackService struct {
|
|
db *gorm.DB
|
|
codec *SecretCodec
|
|
httpClient *http.Client
|
|
}
|
|
|
|
func NewCallbackService(db *gorm.DB, codec *SecretCodec) *CallbackService {
|
|
return &CallbackService{
|
|
db: db,
|
|
codec: codec,
|
|
httpClient: &http.Client{Timeout: 10 * time.Second},
|
|
}
|
|
}
|
|
|
|
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("至少订阅一个事件")
|
|
}
|
|
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": now.UTC().Format(time.RFC3339Nano),
|
|
"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 {
|
|
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 {
|
|
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
|
|
}
|
|
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(callbackRetryDelay(attempts))
|
|
if attempts >= maxCallbackAttempts {
|
|
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 {
|
|
_, _, _ = s.DispatchDue(ctx, 50)
|
|
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
|
|
}
|
|
|
|
func callbackRetryDelay(attempts int) time.Duration {
|
|
if attempts < 1 {
|
|
attempts = 1
|
|
}
|
|
delay := time.Second * time.Duration(1<<(attempts-1))
|
|
if delay > 15*time.Minute {
|
|
return 15 * time.Minute
|
|
}
|
|
return delay
|
|
}
|
|
|
|
func truncateCallbackError(value string) string {
|
|
if len(value) <= 4096 {
|
|
return value
|
|
}
|
|
return value[:4096]
|
|
}
|