package refundretry import ( "context" "crypto/rand" "encoding/hex" "fmt" "time" "hfb_sys/backend/internal/model" "hfb_sys/backend/internal/modules/payment" "github.com/redis/go-redis/v9" "go.uber.org/zap" "gorm.io/gorm" ) const ( refundRetryLockKey = "hfb:job:refundretry:lock" maxRetryCount = 10 baseRetryBackoff = 5 * time.Minute maxRetryBackoff = time.Hour manualWarnThreshold = 24 * time.Hour ) type Job struct { db *gorm.DB redis *redis.Client logger *zap.Logger payments *payment.Repository interval time.Duration instanceID string } func New(db *gorm.DB, redisClient *redis.Client, logger *zap.Logger, payments *payment.Repository) *Job { return &Job{ db: db, redis: redisClient, logger: logger, payments: payments, interval: 2 * time.Minute, instanceID: newInstanceID(), } } // acquireLock 通过 Redis 分布式锁确保同一时刻只有一个实例执行退款补偿。 // 未配置 Redis 时直接执行;退款同步本身按退款单精确处理,可容忍短时间重复扫描。 func (j *Job) acquireLock(ctx context.Context) (func(), bool) { if j.redis == nil { return func() {}, true } ok, err := j.redis.SetNX(ctx, refundRetryLockKey, j.instanceID, j.lockTTL()).Result() if err != nil { j.logger.Warn("退款重试任务获取锁失败,将以单实例模式执行", zap.Error(err)) return func() {}, true } if !ok { return nil, false } return j.releaseLock, true } // releaseLock 仅在锁仍归本实例时释放,避免误删其他实例的锁。 func (j *Job) releaseLock() { if j.redis == nil { return } relCtx, cancel := context.WithTimeout(context.Background(), 3*time.Second) defer cancel() script := redis.NewScript(`if redis.call("get", KEYS[1]) == ARGV[1] then return redis.call("del", KEYS[1]) else return 0 end`) if err := script.Run(relCtx, j.redis, []string{refundRetryLockKey}, j.instanceID).Err(); err != nil { j.logger.Warn("退款重试任务释放锁失败", zap.Error(err)) } } func (j *Job) lockTTL() time.Duration { return 2 * j.interval } func newInstanceID() string { b := make([]byte, 8) if _, err := rand.Read(b); err != nil { return fmt.Sprintf("inst-%d", time.Now().UnixNano()) } return hex.EncodeToString(b) } func (j *Job) Start(ctx context.Context) { if j == nil || j.db == nil || j.payments == nil { return } go j.loop(ctx) } func (j *Job) loop(ctx context.Context) { j.run(ctx) ticker := time.NewTicker(j.interval) defer ticker.Stop() for { select { case <-ctx.Done(): j.logger.Debug("退款重试任务已停止") return case <-ticker.C: j.run(ctx) } } } func (j *Job) run(ctx context.Context) { release, ok := j.acquireLock(ctx) if !ok { return } defer release() now := time.Now() processed, err := j.syncRefundPayments(ctx, now) if err != nil { j.logger.Warn("退款重试任务同步失败", zap.Error(err)) } missing, rebuilt, err := j.warnMissingRefundOrders(ctx, now) if err != nil { j.logger.Warn("退款重试任务扫描缺失退款单失败", zap.Error(err)) } if processed > 0 || missing > 0 { j.logger.Info("退款重试任务处理完成", zap.Int("processed", processed), zap.Int("missing_refund_orders", missing), zap.Int("rebuilt_arbitration_refunds", rebuilt), ) } } func (j *Job) syncRefundPayments(ctx context.Context, now time.Time) (int, error) { var rows []model.PaymentOrder err := j.db.WithContext(ctx). Where("biz_type IN ? AND status IN ? AND updated_at <= ? AND retry_count < ? AND (next_retry_at IS NULL OR next_retry_at <= ?)", payment.RefundBizTypes(), []string{"refunding", "failed"}, now.Add(-5*time.Minute), maxRetryCount, now). Order("id ASC"). Limit(100). Find(&rows).Error if err != nil { return 0, err } processed := 0 for _, row := range rows { if _, err := j.payments.SyncRefundStatusByPaymentID(ctx, row.ID); err != nil { if markErr := j.markRetryFailed(ctx, row, now); markErr != nil { j.logger.Warn("退款重试失败状态保存失败", zap.Uint64("payment_id", row.ID), zap.Uint64("order_id", row.OrderID), zap.Error(markErr), ) } j.logger.Warn("退款状态同步失败", zap.Uint64("payment_id", row.ID), zap.Uint64("order_id", row.OrderID), zap.String("biz_type", row.BizType), zap.String("status", row.Status), zap.Int("retry_count", row.RetryCount+1), zap.Error(err), ) continue } if err := j.resetRetry(ctx, row.ID); err != nil { j.logger.Warn("退款重试计数重置失败", zap.Uint64("payment_id", row.ID), zap.Uint64("order_id", row.OrderID), zap.Error(err), ) } processed++ } j.warnMaxRetryRefunds(ctx, now) return processed, nil } func (j *Job) markRetryFailed(ctx context.Context, row model.PaymentOrder, now time.Time) error { nextCount := row.RetryCount + 1 backoff := retryBackoff(nextCount) nextRetryAt := now.Add(backoff) updates := map[string]any{ "retry_count": nextCount, "last_retry_at": now, "next_retry_at": nextRetryAt, } if nextCount >= maxRetryCount { updates["next_retry_at"] = nil } return j.db.WithContext(ctx).Model(&model.PaymentOrder{}).Where("id = ?", row.ID).Updates(updates).Error } func (j *Job) resetRetry(ctx context.Context, paymentID uint64) error { return j.db.WithContext(ctx).Model(&model.PaymentOrder{}).Where("id = ?", paymentID).Updates(map[string]any{ "retry_count": 0, "last_retry_at": nil, "next_retry_at": nil, }).Error } func retryBackoff(retryCount int) time.Duration { if retryCount <= 1 { return baseRetryBackoff } backoff := baseRetryBackoff for i := 1; i < retryCount; i++ { backoff *= 2 if backoff >= maxRetryBackoff { return maxRetryBackoff } } return backoff } func (j *Job) warnMaxRetryRefunds(ctx context.Context, now time.Time) { var rows []model.PaymentOrder err := j.db.WithContext(ctx). Where("biz_type IN ? AND status IN ? AND retry_count >= ? AND (last_retry_at IS NULL OR last_retry_at <= ?)", payment.RefundBizTypes(), []string{"refunding", "failed"}, maxRetryCount, now.Add(-manualWarnThreshold)). Order("id ASC"). Limit(50). Find(&rows).Error if err != nil { j.logger.Warn("退款最大重试次数扫描失败", zap.Error(err)) return } if len(rows) == 0 { return } // 汇总一条告警,避免每 2 分钟对同一批单刷屏;更新 last_retry_at 实现 24h 节流 ids := make([]uint64, 0, len(rows)) orderIDs := make([]uint64, 0, len(rows)) for _, row := range rows { ids = append(ids, row.ID) orderIDs = append(orderIDs, row.OrderID) } j.logger.Warn("退款已达最大重试次数,需要人工处理", zap.Int("count", len(rows)), zap.Uint64s("payment_ids", ids), zap.Uint64s("order_ids", orderIDs), ) if err := j.db.WithContext(ctx).Model(&model.PaymentOrder{}). Where("id IN ?", ids). Update("last_retry_at", now).Error; err != nil { j.logger.Warn("退款重试时间更新失败", zap.Error(err)) } } func (j *Job) warnMissingRefundOrders(ctx context.Context, now time.Time) (int, int, error) { var rows []model.RentalOrder err := j.db.WithContext(ctx). Where("refund_status IN ? AND refund_amount_cent > 0 AND updated_at <= ?", []string{"pending", "refunding"}, now.Add(-10*time.Minute)). Order("id ASC"). Limit(100). Find(&rows).Error if err != nil { return 0, 0, err } missing := 0 rebuilt := 0 var missingOrderIDs []uint64 var rebuildFailOrderIDs []uint64 for _, row := range rows { var count int64 if err := j.db.WithContext(ctx).Model(&model.PaymentOrder{}). Where("order_id = ? AND biz_type IN ?", row.ID, payment.RefundBizTypes()). Count(&count).Error; err != nil { return missing, rebuilt, err } if count > 0 { continue } missing++ // 仲裁孤儿:settlement_status="arbitrated" 是仲裁事务无条件写入的标志 // (arbitration.go:61),biz_type 可确定还原为 arbitration_refund,自动补建。 // 非仲裁孤儿 biz_type 无法从订单状态可靠区分,维持只告警。 if row.SettlementStatus == "arbitrated" { if j.payments == nil { rebuildFailOrderIDs = append(rebuildFailOrderIDs, row.ID) continue } if _, err := j.payments.StartRefund(ctx, row.ID, row.RefundAmountCent, "arbitration_refund", "仲裁退款补偿补建"); err != nil { rebuildFailOrderIDs = append(rebuildFailOrderIDs, row.ID) } else { rebuilt++ j.logger.Info("仲裁退款单已自动补建", zap.Uint64("order_id", row.ID), zap.Int64("refund_amount_cent", row.RefundAmountCent), ) } continue } missingOrderIDs = append(missingOrderIDs, row.ID) } // 汇总告警 + Redis 24h 去重,避免每轮扫描刷屏 if len(missingOrderIDs) > 0 && j.shouldWarn(ctx, "missing_payment", 0) { j.logger.Warn("退款订单缺少支付记录,需要人工处理", zap.Int("count", len(missingOrderIDs)), zap.Uint64s("order_ids", missingOrderIDs), ) } if len(rebuildFailOrderIDs) > 0 && j.shouldWarn(ctx, "rebuild_fail", 0) { j.logger.Warn("仲裁退款单自动补建失败,需要人工处理", zap.Int("count", len(rebuildFailOrderIDs)), zap.Uint64s("order_ids", rebuildFailOrderIDs), ) } return missing, rebuilt, nil } // shouldWarn 用 Redis 做 24h 节流;无 Redis 时每个 job 周期最多打一次同类汇总(由调用方聚合)。 func (j *Job) shouldWarn(ctx context.Context, kind string, id uint64) bool { if j.redis == nil { return true } key := fmt.Sprintf("hfb:job:refundretry:warn:%s", kind) if id > 0 { key = fmt.Sprintf("%s:%d", key, id) } ok, err := j.redis.SetNX(ctx, key, "1", manualWarnThreshold).Result() if err != nil { return true } return ok }