hfb_sys/backend/internal/jobs/refundretry/job.go

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("refund retry job acquire lock failed, run without lock", 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("refund retry job release lock failed", 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.Info("refund retry job stopped")
			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("refund retry job sync payments failed", zap.Error(err))
	}
	missing, rebuilt, err := j.warnMissingRefundOrders(ctx, now)
	if err != nil {
		j.logger.Warn("refund retry job scan missing refund orders failed", zap.Error(err))
	}
	if processed > 0 || missing > 0 {
		j.logger.Info("refund retry job finished",
			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("refund retry mark failure failed",
					zap.Uint64("payment_id", row.ID),
					zap.Uint64("order_id", row.OrderID),
					zap.Error(markErr),
				)
			}
			j.logger.Warn("refund retry sync failed",
				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("refund retry reset counter failed",
				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("refund retry max count scan failed", zap.Error(err))
		return
	}
	for _, row := range rows {
		j.logger.Warn("refund retry reached max count, need manual check",
			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),
		)
	}
}

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
	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 {
				j.logger.Warn("auto rebuild arbitration refund skipped, payment repository unavailable",
					zap.Uint64("order_id", row.ID),
					zap.Int64("refund_amount_cent", row.RefundAmountCent),
				)
				continue
			}
			if _, err := j.payments.StartRefund(ctx, row.ID, row.RefundAmountCent, "arbitration_refund", "仲裁退款补偿补建"); err != nil {
				j.logger.Warn("auto rebuild arbitration refund failed, need manual check",
					zap.Uint64("order_id", row.ID),
					zap.Int64("refund_amount_cent", row.RefundAmountCent),
					zap.Error(err),
				)
			} else {
				rebuilt++
				j.logger.Info("auto rebuild arbitration refund submitted",
					zap.Uint64("order_id", row.ID),
					zap.Int64("refund_amount_cent", row.RefundAmountCent),
				)
			}
			continue
		}
		j.logger.Warn("refund order missing payment record, need manual check",
			zap.Uint64("order_id", row.ID),
			zap.String("order_no", row.OrderNo),
			zap.String("refund_status", row.RefundStatus),
			zap.String("settlement_status", row.SettlementStatus),
			zap.Int64("refund_amount_cent", row.RefundAmountCent),
		)
	}
	return missing, rebuilt, nil
}