package service import ( "context" "log/slog" "time" "git.lzcwy.cn/zhengchiliu/zt-closet-backend/global" "git.lzcwy.cn/zhengchiliu/zt-closet-backend/internal/mysqldao" "git.lzcwy.cn/zhengchiliu/zt-closet-backend/internal/mysqlmodel" "git.lzcwy.cn/zhengchiliu/zt-closet-backend/internal/redisdao" ) // StartBenefitCompensationScheduler 启动权益发放补偿定时任务:周期性重试发放失败的权益。 func StartBenefitCompensationScheduler() { go func() { slog.Info("benefit compensation scheduler started") for { func() { defer func() { if r := recover(); r != nil { slog.Error("benefit compensation scheduler panic", "recover", r) } }() runBenefitCompensation() }() time.Sleep(time.Minute) } }() } // runBenefitCompensation 扫描待发放补偿任务并重试,单实例通过 Redis 分布式锁防重。 func runBenefitCompensation() { ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute) defer cancel() ctx, span := global.Tracer.Start(ctx, "scheduler.BenefitCompensation") defer span.End() lockKey := "ztcloset:lock:benefit:compensation" lockVal := genUUID() ok, lerr := redisdao.TryLock(ctx, lockKey, lockVal, time.Minute) if lerr != nil || !ok { return } defer redisdao.Unlock(context.Background(), lockKey, lockVal) taskDao := mysqldao.NewBenefitGrantTaskDao() // 恢复超时的「发放中」任务(认领后崩溃的场景),使其回到待发放继续重试。 _, _ = taskDao.BenefitGrantTaskRecoverStale(ctx, nowUnix()-int64((10*time.Minute).Seconds())) tasks, err := taskDao.BenefitGrantTaskListPending(ctx, 100) if err != nil { slog.ErrorContext(ctx, "list pending benefit tasks failed", "error", err) return } if len(tasks) == 0 { return } s := NewOrderService() for i := range tasks { s.retryBenefitTask(ctx, &tasks[i]) } } // retryBenefitTask 重试单个补偿任务(幂等:先查发放流水避免重复发放)。 func (s *OrderService) retryBenefitTask(ctx context.Context, task *mysqlmodel.BenefitGrantTask) { taskDao := mysqldao.NewBenefitGrantTaskDao() // 重试次数超限,标记失败并告警,等待人工介入。 if task.RetryCount >= task.MaxRetry { _ = taskDao.BenefitGrantTaskUpdateByID(ctx, task.ID, map[string]any{"status": BenefitTaskFailed}) slog.ErrorContext(ctx, "benefit grant task exhausted retries", "order_no", task.OrderNo, "benefit_type", task.BenefitType, "retry_count", task.RetryCount) return } // 幂等判断:若发放流水已存在(上次发放成功但未置 success 的崩溃窗口),直接置成功。 if granted, err := s.benefitAlreadyGranted(ctx, task); err == nil && granted { _ = taskDao.BenefitGrantTaskUpdateByID(ctx, task.ID, map[string]any{"status": BenefitTaskSuccess}) return } else if err != nil { slog.ErrorContext(ctx, "check benefit granted failed", "order_no", task.OrderNo, "benefit_type", task.BenefitType, "error", err) return } // 执行发放(claim 防并发,内部处理 retry_count 原子自增与状态流转)。 s.executeGrant(ctx, task) } // benefitAlreadyGranted 判断该权益是否已发放(用于崩溃恢复幂等)。 // VIP 查 VIP 流水;额度类查额度凭证(quota_grants)。 func (s *OrderService) benefitAlreadyGranted(ctx context.Context, task *mysqlmodel.BenefitGrantTask) (bool, error) { grantDao := mysqldao.NewQuotaGrantDao() switch task.BenefitType { case BenefitTypeVIP: return s.vipDao.VIPFlowLogExistsByOrderNo(ctx, task.OrderNo) case BenefitTypeCos: return grantDao.QuotaGrantExistsByRefIDAndType(ctx, task.OrderNo, QuotaTypeCos) case BenefitTypeAI: return grantDao.QuotaGrantExistsByRefIDAndType(ctx, task.OrderNo, QuotaTypeAI) case BenefitTypeClosetShare: return grantDao.QuotaGrantExistsByRefIDAndType(ctx, task.OrderNo, QuotaTypeClosetShare) case BenefitTypeOutfitShare: return grantDao.QuotaGrantExistsByRefIDAndType(ctx, task.OrderNo, QuotaTypeOutfitShare) default: return false, nil } }