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" "gorm.io/gorm" ) // 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() s := NewOrderService() // 对账已支付但任务未完整落库的订单,覆盖支付状态提交后进程崩溃/数据库瞬时错误。 if orders, err := mysqldao.NewOrderDao().OrderListPaidMissingBenefitTasks(ctx, 100); err == nil { for i := range orders { s.grantOrderBenefits(ctx, &orders[i]) } } else { slog.ErrorContext(ctx, "reconcile paid order benefits failed", "error", err) } // 恢复超时的「发放中」任务(认领后崩溃的场景),使其回到待发放继续重试。 _, _ = 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 } for i := range tasks { s.retryBenefitTask(ctx, &tasks[i]) } revokeDao := mysqldao.NewBenefitRevokeTaskDao() _ = revokeDao.RecoverStale(ctx, nowUnix()-int64((10*time.Minute).Seconds())) if revokeTasks, err := revokeDao.ListPending(ctx, 100); err == nil { for i := range revokeTasks { s.executeRevokeTask(ctx, &revokeTasks[i]) } } else { slog.ErrorContext(ctx, "list pending benefit revoke tasks failed", "error", err) } } // 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 } } func (s *OrderService) executeRevokeTaskByOrder(ctx context.Context, orderNo string) { task, err := mysqldao.NewBenefitRevokeTaskDao().GetByOrderNo(ctx, orderNo) if err == nil { s.executeRevokeTask(ctx, task) } } func (s *OrderService) executeRevokeTask(ctx context.Context, task *mysqlmodel.BenefitRevokeTask) { dao := mysqldao.NewBenefitRevokeTaskDao() if task.Status == BenefitTaskSuccess { return } if task.RetryCount >= task.MaxRetry { slog.ErrorContext(ctx, "benefit revoke task exceeded normal retry threshold; continuing for eventual consistency", "order_no", task.OrderNo, "retry_count", task.RetryCount) } claimed, err := dao.Claim(ctx, task.ID) if err != nil || !claimed { return } order, err := s.orderDao.OrderGetByOrderNo(ctx, task.OrderNo) if err == nil { err = s.revokeOrderBenefits(ctx, order) } if err != nil { _ = dao.Update(ctx, task.ID, map[string]any{"status": BenefitTaskPending, "retry_count": gorm.Expr("retry_count + 1"), "last_error": err.Error()}) slog.ErrorContext(ctx, "revoke refunded order benefits failed", "order_no", task.OrderNo, "error", err) return } _ = dao.Update(ctx, task.ID, map[string]any{"status": BenefitTaskSuccess, "last_error": ""}) }