From cf2ee96baaaa6828522ac1039652ea88c41dc281 Mon Sep 17 00:00:00 2001 From: yml2213 Date: Mon, 3 Aug 2026 10:57:21 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BC=98=E5=8C=96=E5=9B=9E=E8=B0=83=E6=8E=A8?= =?UTF-8?q?=E9=80=81=EF=BC=9A=E5=8F=82=E8=80=83=E5=BE=AE=E4=BF=A1/?= =?UTF-8?q?=E6=94=AF=E4=BB=98=E5=AE=9D=E6=9C=BA=E5=88=B6=EF=BC=8C=E9=99=90?= =?UTF-8?q?=E5=88=B6=E6=AF=8F=E5=95=86=E6=88=B7=E5=8D=95=E4=B8=80=E5=9B=9E?= =?UTF-8?q?=E8=B0=83=E5=9C=B0=E5=9D=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 回调重试改为微信式固定退避序列（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 透传回调相关配置 --- backend/cmd/server/main.go | 18 +++++- backend/internal/config/config.go | 30 +++++++++ backend/internal/service/callback.go | 91 +++++++++++++++++++++++---- docker-compose.yml | 3 + frontend/src/pages/MerchantCenter.tsx | 4 +- 5 files changed, 131 insertions(+), 15 deletions(-) diff --git a/backend/cmd/server/main.go b/backend/cmd/server/main.go index c12d87b..87778c2 100644 --- a/backend/cmd/server/main.go +++ b/backend/cmd/server/main.go @@ -20,6 +20,18 @@ import ( "gorm.io/gorm" ) +// secondsToDurations 把秒数列表转成 time.Duration 列表（回调重试序列）。 +func secondsToDurations(seconds []int) []time.Duration { + if len(seconds) == 0 { + return nil + } + out := make([]time.Duration, 0, len(seconds)) + for _, s := range seconds { + out = append(out, time.Duration(s)*time.Second) + } + return out +} + func main() { cfg := config.Load() time.Local = timeutil.Location() @@ -54,7 +66,11 @@ func main() { tenantSvc := service.NewTenantService(db) authSvc := service.NewAuthService(db, jm, tenantSvc) userSvc := service.NewUserService(db, tenantSvc) - callbackSvc := service.NewCallbackService(db, codec) + callbackSvc := service.NewCallbackService(db, codec, service.WithCallbackConfig(service.CallbackConfig{ + MaxAttempts: cfg.CallbackMaxAttempts, + PushTimeout: time.Duration(cfg.CallbackPushTimeoutSeconds) * time.Second, + RetrySchedule: secondsToDurations(cfg.CallbackRetryScheduleSeconds), + })) fulfillmentSvc := service.NewFulfillmentService(db, callbackSvc) merchantSvc := service.NewMerchantService(db, codec, tenantSvc) deliverySvc := service.NewDeliveryService( diff --git a/backend/internal/config/config.go b/backend/internal/config/config.go index e3ec8d0..b6d2625 100644 --- a/backend/internal/config/config.go +++ b/backend/internal/config/config.go @@ -40,6 +40,13 @@ type Config struct { FulfillmentProcessingTimeoutMinutes int // FulfillmentTimeoutScanIntervalSeconds 发货超时巡检间隔秒数。 FulfillmentTimeoutScanIntervalSeconds int + // CallbackMaxAttempts 回调推送总尝试次数（首次 + 重试），默认 16（微信式）。 + CallbackMaxAttempts int + // CallbackPushTimeoutSeconds 单次回调推送 HTTP 超时秒数。 + CallbackPushTimeoutSeconds int + // CallbackRetryScheduleSeconds 回调失败重试间隔序列（秒，逗号分隔，微信/支付宝式固定退避）。 + // 为空时使用默认微信式序列。 + CallbackRetryScheduleSeconds []int } // Load 加载配置：先尝试读取 .env，再读系统环境变量（已存在的系统环境变量优先级更高） @@ -66,9 +73,32 @@ func Load() *Config { DeliveryLinkTTLMinutes: getEnvInt("DELIVERY_LINK_TTL_MINUTES", 120), FulfillmentProcessingTimeoutMinutes: getEnvInt("FULFILLMENT_PROCESSING_TIMEOUT_MINUTES", 30), FulfillmentTimeoutScanIntervalSeconds: getEnvInt("FULFILLMENT_TIMEOUT_SCAN_INTERVAL_SECONDS", 60), + CallbackMaxAttempts: getEnvInt("CALLBACK_MAX_ATTEMPTS", 16), + CallbackPushTimeoutSeconds: getEnvInt("CALLBACK_PUSH_TIMEOUT_SECONDS", 15), + CallbackRetryScheduleSeconds: getEnvIntList("CALLBACK_RETRY_SCHEDULE", nil), } } +// getEnvIntList 解析逗号分隔的整数列表（如 "15,15,30,180"）；空串返回 nil。 +func getEnvIntList(key string, def []int) []int { + v := os.Getenv(key) + if v == "" { + return def + } + var out []int + for _, part := range strings.Split(v, ",") { + n, err := strconv.Atoi(strings.TrimSpace(part)) + if err != nil || n < 1 { + continue + } + out = append(out, n) + } + if len(out) == 0 { + return def + } + return out +} + func getEnvBool(key string, def bool) bool { v := os.Getenv(key) if v == "" { diff --git a/backend/internal/service/callback.go b/backend/internal/service/callback.go index f584da3..e6f63b8 100644 --- a/backend/internal/service/callback.go +++ b/backend/internal/service/callback.go @@ -23,21 +23,74 @@ import ( "gorm.io/gorm" ) -const maxCallbackAttempts = 8 +// 回调推送策略（参考微信/支付宝通知机制）： +// 首次推送失败后按固定序列退避重试，默认共 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) *CallbackService { - return &CallbackService{ - db: db, - codec: codec, - httpClient: &http.Client{Timeout: 10 * time.Second}, +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) { @@ -70,6 +123,16 @@ func (s *CallbackService) CreateSubscription(merchantID uint, in CreateCallbackI 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 @@ -260,8 +323,8 @@ func (s *CallbackService) dispatchOne(ctx context.Context, id uint) (bool, bool, 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 { + nextAttempt := time.Now().Add(s.callbackRetryDelay(attempts)) + if attempts >= s.maxAttempts { status = model.CallbackDeliveryFailed nextAttempt = time.Now() } @@ -321,15 +384,17 @@ func validateCallbackURL(rawURL string) error { return nil } -func callbackRetryDelay(attempts int) time.Duration { +// callbackRetryDelay 返回第 attempts 次失败后的重试间隔（微信式固定序列）。 +// 序列用尽后复用最后一个间隔。 +func (s *CallbackService) 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 + idx := attempts - 1 + if idx >= len(s.retrySchedule) { + idx = len(s.retrySchedule) - 1 } - return delay + return s.retrySchedule[idx] } func truncateCallbackError(value string) string { diff --git a/docker-compose.yml b/docker-compose.yml index d56672c..10f7e17 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -52,6 +52,9 @@ services: - DELIVERY_LINK_TTL_MINUTES=${DELIVERY_LINK_TTL_MINUTES:-120} - FULFILLMENT_PROCESSING_TIMEOUT_MINUTES=${FULFILLMENT_PROCESSING_TIMEOUT_MINUTES:-30} - FULFILLMENT_TIMEOUT_SCAN_INTERVAL_SECONDS=${FULFILLMENT_TIMEOUT_SCAN_INTERVAL_SECONDS:-60} + - CALLBACK_MAX_ATTEMPTS=${CALLBACK_MAX_ATTEMPTS:-16} + - CALLBACK_PUSH_TIMEOUT_SECONDS=${CALLBACK_PUSH_TIMEOUT_SECONDS:-15} + - CALLBACK_RETRY_SCHEDULE=${CALLBACK_RETRY_SCHEDULE:-} volumes: - backend-data:/data - ./logs:/app/logs diff --git a/frontend/src/pages/MerchantCenter.tsx b/frontend/src/pages/MerchantCenter.tsx index b084159..a6c2c9b 100644 --- a/frontend/src/pages/MerchantCenter.tsx +++ b/frontend/src/pages/MerchantCenter.tsx @@ -730,7 +730,9 @@ export default function MerchantCenter({ fixedTab, title = '商户中心' }: Mer const callbackContent = ( - 回调事件通过 outbox 持久化发送。 + + 回调事件通过 outbox 持久化推送，失败按固定间隔退避重试（最多 16 次）后标记失败。每个商户仅支持一个回调地址，变更地址请先停用当前订阅。 + {canManage &&