affiliate_dash/backend/internal/service/callback.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 供接收方校验：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]
}