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 { slog.ErrorContext(ctx, "benefit compensation: acquire lock failed", "error", lerr) return } if !ok { slog.DebugContext(ctx, "benefit compensation: another instance running, skip") return } defer func() { if uerr := redisdao.Unlock(context.Background(), lockKey, lockVal); uerr != nil { slog.WarnContext(ctx, "benefit compensation: release lock failed", "error", uerr) } }() 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) } // 恢复超时的「发放中」任务(认领后崩溃的场景),使其回到待发放继续重试。 recovered, rerr := taskDao.BenefitGrantTaskRecoverStale(ctx, nowUnix()-int64((10*time.Minute).Seconds())) if rerr != nil { slog.ErrorContext(ctx, "benefit compensation: recover stale grant tasks failed", "error", rerr) } else if recovered > 0 { slog.InfoContext(ctx, "benefit compensation: recovered stale grant tasks", "count", recovered) } 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() if rerr := revokeDao.RecoverStale(ctx, nowUnix()-int64((10*time.Minute).Seconds())); rerr != nil { slog.ErrorContext(ctx, "benefit compensation: recover stale revoke tasks failed", "error", rerr) } 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 { if uerr := taskDao.BenefitGrantTaskUpdateByID(ctx, task.ID, map[string]any{"status": BenefitTaskFailed}); uerr != nil { slog.ErrorContext(ctx, "benefit compensation: mark exhausted task failed status failed", "order_no", task.OrderNo, "benefit_type", task.BenefitType, "error", uerr) } 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 { if uerr := taskDao.BenefitGrantTaskUpdateByID(ctx, task.ID, map[string]any{"status": BenefitTaskSuccess}); uerr != nil { slog.ErrorContext(ctx, "benefit compensation: mark already-granted task success failed", "order_no", task.OrderNo, "benefit_type", task.BenefitType, "error", uerr) } slog.InfoContext(ctx, "benefit already granted, task marked success", "order_no", task.OrderNo, "benefit_type", task.BenefitType) 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 { if isNotFound(err) { return // 无回收任务(如未退款订单),属正常路径。 } slog.ErrorContext(ctx, "get benefit revoke task failed", "order_no", orderNo, "error", err) return } 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 { slog.ErrorContext(ctx, "claim benefit revoke task failed", "order_no", task.OrderNo, "error", err) return } if !claimed { slog.DebugContext(ctx, "benefit revoke task already claimed, skip", "order_no", task.OrderNo) return } order, oerr := s.orderDao.OrderGetByOrderNo(ctx, task.OrderNo) if oerr != nil { _ = dao.Update(ctx, task.ID, map[string]any{"status": BenefitTaskPending, "retry_count": gorm.Expr("retry_count + 1"), "last_error": oerr.Error()}) slog.ErrorContext(ctx, "revoke refunded order benefits failed: get order failed", "order_no", task.OrderNo, "retry_count", task.RetryCount, "error", oerr) return } if rerr := s.revokeOrderBenefits(ctx, order); rerr != nil { _ = dao.Update(ctx, task.ID, map[string]any{"status": BenefitTaskPending, "retry_count": gorm.Expr("retry_count + 1"), "last_error": rerr.Error()}) slog.ErrorContext(ctx, "revoke refunded order benefits failed", "order_no", task.OrderNo, "retry_count", task.RetryCount, "error", rerr) return } if uerr := dao.Update(ctx, task.ID, map[string]any{"status": BenefitTaskSuccess, "last_error": ""}); uerr != nil { slog.ErrorContext(ctx, "mark benefit revoke task success failed", "order_no", task.OrderNo, "error", uerr) return } slog.InfoContext(ctx, "refunded order benefits revoked", "order_no", task.OrderNo, "user_id", order.UserID) }