diff --git a/internal/app/service/task/analysis_plan_task_manager.go b/internal/app/service/task/analysis_plan_task_manager.go index 1d4dee6..093490b 100644 --- a/internal/app/service/task/analysis_plan_task_manager.go +++ b/internal/app/service/task/analysis_plan_task_manager.go @@ -1,7 +1,9 @@ package task import ( - "context" + "fmt" + "sync" + "time" "git.huangwc.com/pig/pig-farm-controller/internal/infra/logs" "git.huangwc.com/pig/pig-farm-controller/internal/infra/models" @@ -9,13 +11,15 @@ import ( "git.huangwc.com/pig/pig-farm-controller/internal/infra/utils" ) -// AnalysisPlanTaskManager 封装了创建和更新计划分析任务（即触发器）的逻辑。 -// 这是一个可被 Scheduler 和其他应用服务（如 PlanService）共享的无状态组件。 +// AnalysisPlanTaskManager 负责管理分析计划的触发器任务。 +// 它确保数据库中可执行的计划在待执行队列中有对应的触发器，并移除无效的触发器。 +// 这是一个有状态的组件，包含一个互斥锁以确保并发安全。 type AnalysisPlanTaskManager struct { planRepo repository.PlanRepository pendingTaskRepo repository.PendingTaskRepository executionLogRepo repository.ExecutionLogRepository logger *logs.Logger + mu sync.Mutex } // NewAnalysisPlanTaskManager 是 AnalysisPlanTaskManager 的构造函数。 @@ -33,50 +37,196 @@ func NewAnalysisPlanTaskManager( } } -// CreateOrUpdateTrigger 为给定的 planID 创建或更新其关联的下一次触发任务。 -// 这个方法是幂等的，可以安全地被多次调用。 -func (m *AnalysisPlanTaskManager) CreateOrUpdateTrigger(ctx context.Context, planID uint) error { - // 获取计划信息 - plan, err := m.planRepo.GetBasicPlanByID(planID) +// Refresh 同步数据库中的计划状态和待执行队列中的触发器任务。 +// 这是一个编排方法，将复杂的逻辑分解到多个内部方法中。 +func (m *AnalysisPlanTaskManager) Refresh() error { + m.mu.Lock() + defer m.mu.Unlock() + + m.logger.Info("开始同步计划任务管理器...") + + // 1. 一次性获取所有需要的数据 + runnablePlans, invalidPlanIDs, pendingTasks, err := m.getRefreshData() if err != nil { - m.logger.Errorf("[严重] 获取计划失败, 错误: %v", err) - return err + return fmt.Errorf("获取刷新数据失败: %w", err) } - // 获取触发任务 - task, err := m.planRepo.FindPlanAnalysisTaskByParamsPlanID(planID) - if err != nil { - m.logger.Errorf("[严重] 获取计划解析任务失败, 错误: %v", err) - return err + // 2. 清理所有与失效计划相关的待执行任务 + if err := m.cleanupInvalidTasks(invalidPlanIDs, pendingTasks); err != nil { + // 仅记录错误，清理失败不应阻止新任务的添加 + m.logger.Errorf("清理无效任务时出错: %v", err) } - // 写入执行日志 - taskLog := &models.TaskExecutionLog{ - TaskID: task.ID, - Status: models.ExecutionStatusWaiting, - } - if err := m.executionLogRepo.CreateTaskExecutionLogsInBatch([]*models.TaskExecutionLog{taskLog}); err != nil { - m.logger.Errorf("[严重] 创建任务执行日志失败, 错误: %v", err) - return err + // 3. 为应执行但缺失的计划添加新触发器 + if err := m.addNewTriggers(runnablePlans, pendingTasks); err != nil { + return fmt.Errorf("添加新触发器时出错: %w", err) } - // 写入待执行队列 - next, err := utils.GetNextCronTime(plan.CronExpression) - if err != nil { - m.logger.Errorf("[严重] 执行时间解析失败, 错误: %v", err) - return err - } - pendingTask := &models.PendingTask{ - TaskID: task.ID, - ExecuteAt: next, - TaskExecutionLogID: taskLog.ID, - } - err = m.pendingTaskRepo.CreatePendingTasksInBatch([]*models.PendingTask{pendingTask}) - if err != nil { - m.logger.Errorf("[严重] 创建待执行任务失败, 错误: %v", err) - return err - } - - m.logger.Infof("成功为 Plan %d 创建/更新了下一次的触发任务，执行时间: %v", planID, next) + m.logger.Info("计划任务管理器同步完成.") + return nil +} + +// CreateOrUpdateTrigger 为给定的 planID 创建其关联的触发任务。 +// 这个方法是幂等的：如果一个有效的触发器已存在，它将不会重复创建。 +func (m *AnalysisPlanTaskManager) CreateOrUpdateTrigger(planID uint) error { + m.mu.Lock() + defer m.mu.Unlock() + + // 检查计划是否可执行 + plan, err := m.planRepo.GetBasicPlanByID(planID) + if err != nil { + return fmt.Errorf("获取计划基本信息失败: %w", err) + } + if plan.Status != models.PlanStatusEnabled { + return fmt.Errorf("计划 #%d 当前状态为 '%d'，无法创建触发器", planID, plan.Status) + } + + // 幂等性检查：如果触发器已存在，则直接返回 + existingTrigger, err := m.pendingTaskRepo.FindPendingTriggerByPlanID(planID) + if err != nil { + return fmt.Errorf("查找现有触发器失败: %w", err) + } + if existingTrigger != nil { + m.logger.Infof("计划 #%d 的触发器已存在于待执行队列中，无需重复创建。", planID) + return nil + } + + m.logger.Infof("为计划 #%d 创建新的触发器...", planID) + return m.createTriggerTask(plan) +} + +// --- 内部私有方法 --- + +// getRefreshData 从数据库获取刷新所需的所有数据。 +func (m *AnalysisPlanTaskManager) getRefreshData() (runnablePlans []*models.Plan, invalidPlanIDs []uint, pendingTasks []models.PendingTask, err error) { + runnablePlans, err = m.planRepo.FindRunnablePlans() + if err != nil { + m.logger.Errorf("获取可执行计划列表失败: %v", err) + return + } + + invalidPlans, err := m.planRepo.FindDisabledAndStoppedPlans() + if err != nil { + m.logger.Errorf("获取失效计划列表失败: %v", err) + return + } + invalidPlanIDs = make([]uint, len(invalidPlans)) + for i, p := range invalidPlans { + invalidPlanIDs[i] = p.ID + } + + pendingTasks, err = m.pendingTaskRepo.FindAllPendingTasks() + if err != nil { + m.logger.Errorf("获取所有待执行任务失败: %v", err) + return + } + return +} + +// cleanupInvalidTasks 清理所有与失效计划相关的待执行任务。 +func (m *AnalysisPlanTaskManager) cleanupInvalidTasks(invalidPlanIDs []uint, allPendingTasks []models.PendingTask) error { + if len(invalidPlanIDs) == 0 { + return nil // 没有需要清理的计划 + } + + invalidPlanIDSet := make(map[uint]struct{}, len(invalidPlanIDs)) + for _, id := range invalidPlanIDs { + invalidPlanIDSet[id] = struct{}{} + } + + var tasksToDeleteIDs []uint + var logsToCancelIDs []uint + + for _, pt := range allPendingTasks { + if pt.Task == nil { // 防御性编程，确保 Task 被预加载 + continue + } + if _, isInvalid := invalidPlanIDSet[pt.Task.PlanID]; isInvalid { + tasksToDeleteIDs = append(tasksToDeleteIDs, pt.ID) + logsToCancelIDs = append(logsToCancelIDs, pt.TaskExecutionLogID) + } + } + + if len(tasksToDeleteIDs) == 0 { + return nil // 没有找到需要清理的任务 + } + + m.logger.Infof("准备从待执行队列中清理 %d 个与失效计划相关的任务...", len(tasksToDeleteIDs)) + + // 批量删除待执行任务 + if err := m.pendingTaskRepo.DeletePendingTasksByIDs(tasksToDeleteIDs); err != nil { + return fmt.Errorf("批量删除待执行任务失败: %w", err) + } + + // 批量更新相关执行日志状态为“已取消” + if err := m.executionLogRepo.UpdateLogStatusByIDs(logsToCancelIDs, models.ExecutionStatusCancelled); err != nil { + // 这是一个非关键性错误，只记录日志 + m.logger.Warnf("批量更新日志状态为 'Cancelled' 失败: %v", err) + } + + return nil +} + +// addNewTriggers 检查并为应执行但缺失的计划添加新触发器。 +func (m *AnalysisPlanTaskManager) addNewTriggers(runnablePlans []*models.Plan, allPendingTasks []models.PendingTask) error { + // 创建一个集合，存放所有已在队列中的计划触发器 + pendingTriggerPlanIDs := make(map[uint]struct{}) + for _, pt := range allPendingTasks { + if pt.Task != nil && pt.Task.Type == models.TaskPlanAnalysis { + pendingTriggerPlanIDs[pt.Task.PlanID] = struct{}{} + } + } + + for _, plan := range runnablePlans { + if _, exists := pendingTriggerPlanIDs[plan.ID]; !exists { + m.logger.Infof("发现应执行但队列中缺失的计划 #%d，正在为其创建触发器...", plan.ID) + if err := m.createTriggerTask(plan); err != nil { + m.logger.Errorf("为计划 #%d 创建触发器失败: %v", plan.ID, err) + // 继续处理下一个，不因单点失败而中断 + } + } + } + return nil +} + +// createTriggerTask 是创建触发器任务的内部核心逻辑。 +func (m *AnalysisPlanTaskManager) createTriggerTask(plan *models.Plan) error { + analysisTask, err := m.planRepo.FindPlanAnalysisTaskByPlanID(plan.ID) + if err != nil { + return fmt.Errorf("查找计划分析任务失败: %w", err) + } + if analysisTask == nil { + return fmt.Errorf("未找到计划 #%d 关联的 'plan_analysis' 任务", plan.ID) + } + + var executeAt time.Time + if plan.ExecutionType == models.PlanExecutionTypeManual { + executeAt = time.Now() + } else { + next, err := utils.GetNextCronTime(plan.CronExpression) + if err != nil { + return fmt.Errorf("解析 Cron 表达式 '%s' 失败: %w", plan.CronExpression, err) + } + executeAt = next + } + + taskLog := &models.TaskExecutionLog{ + TaskID: analysisTask.ID, + Status: models.ExecutionStatusWaiting, + } + if err := m.executionLogRepo.CreateTaskExecutionLog(taskLog); err != nil { + return fmt.Errorf("创建任务执行日志失败: %w", err) + } + + pendingTask := &models.PendingTask{ + TaskID: analysisTask.ID, + ExecuteAt: executeAt, + TaskExecutionLogID: taskLog.ID, + } + if err := m.pendingTaskRepo.CreatePendingTask(pendingTask); err != nil { + return fmt.Errorf("创建待执行任务失败: %w", err) + } + + m.logger.Infof("成功为计划 #%d 创建触发器 (任务ID: %d)，执行时间: %v", plan.ID, analysisTask.ID, executeAt) return nil } diff --git a/internal/infra/models/schedule.go b/internal/infra/models/schedule.go index a3f5f4d..40e43b1 100644 --- a/internal/infra/models/schedule.go +++ b/internal/infra/models/schedule.go @@ -16,11 +16,13 @@ type PendingTask struct { // TaskID 使用 int 类型以容纳特殊的负数ID，代表系统任务 TaskID int `gorm:"index"` + // Task 字段，用于在代码中访问关联的任务详情 + // GORM 会根据 TaskID 字段自动填充此关联 + Task *Task `gorm:"foreignKey:TaskID"` ExecuteAt time.Time `gorm:"index"` // 任务执行时间 TaskExecutionLogID uint `gorm:"unique;not null"` // 对应的执行历史记录ID - // 关联关系定义 // 通过 TaskExecutionLogID 关联到唯一的 TaskExecutionLog 记录 // ON DELETE CASCADE 确保如果日志被删除，这个待办任务也会被自动清理 TaskExecutionLog TaskExecutionLog `gorm:"foreignKey:TaskExecutionLogID;references:ID;constraint:OnUpdate:CASCADE,OnDelete:CASCADE;"` diff --git a/internal/infra/repository/execution_log_repository.go b/internal/infra/repository/execution_log_repository.go index 4503cb0..d0e4f81 100644 --- a/internal/infra/repository/execution_log_repository.go +++ b/internal/infra/repository/execution_log_repository.go @@ -8,6 +8,9 @@ import ( // ExecutionLogRepository 定义了与执行日志交互的接口。 // 这为服务层提供了一个清晰的契约，并允许在测试中轻松地进行模拟。 type ExecutionLogRepository interface { + UpdateLogStatusByIDs(logIDs []uint, status models.ExecutionStatus) error + UpdateLogStatus(logID uint, status models.ExecutionStatus) error + CreateTaskExecutionLog(log *models.TaskExecutionLog) error CreatePlanExecutionLog(log *models.PlanExecutionLog) error UpdatePlanExecutionLog(log *models.PlanExecutionLog) error CreateTaskExecutionLogsInBatch(logs []*models.TaskExecutionLog) error @@ -26,6 +29,23 @@ func NewGormExecutionLogRepository(db *gorm.DB) ExecutionLogRepository { return &gormExecutionLogRepository{db: db} } +func (r *gormExecutionLogRepository) UpdateLogStatusByIDs(logIDs []uint, status models.ExecutionStatus) error { + if len(logIDs) == 0 { + return nil + } + return r.db.Model(&models.TaskExecutionLog{}). + Where("id IN ?", logIDs). + Update("status", status).Error +} + +func (r *gormExecutionLogRepository) UpdateLogStatus(logID uint, status models.ExecutionStatus) error { + return r.db.Model(&models.TaskExecutionLog{}).Where("id = ?", logID).Update("status", status).Error +} + +func (r *gormExecutionLogRepository) CreateTaskExecutionLog(log *models.TaskExecutionLog) error { + return r.db.Create(log).Error +} + // CreatePlanExecutionLog 为一次计划执行创建一条新的日志条目。 func (r *gormExecutionLogRepository) CreatePlanExecutionLog(log *models.PlanExecutionLog) error { return r.db.Create(log).Error @@ -41,6 +61,9 @@ func (r *gormExecutionLogRepository) UpdatePlanExecutionLog(log *models.PlanExec // CreateTaskExecutionLogsInBatch 在一次数据库调用中创建多个任务执行日志条目。 // 这是“预写日志”步骤的关键。 func (r *gormExecutionLogRepository) CreateTaskExecutionLogsInBatch(logs []*models.TaskExecutionLog) error { + if len(logs) == 0 { + return nil + } // GORM 的 Create 传入一个切片指针会执行批量插入。 return r.db.Create(&logs).Error } diff --git a/internal/infra/repository/pending_task_repository.go b/internal/infra/repository/pending_task_repository.go index 689197b..166496b 100644 --- a/internal/infra/repository/pending_task_repository.go +++ b/internal/infra/repository/pending_task_repository.go @@ -1,6 +1,7 @@ package repository import ( + "errors" "time" "git.huangwc.com/pig/pig-farm-controller/internal/infra/models" @@ -10,6 +11,10 @@ import ( // PendingTaskRepository 定义了与待执行任务队列交互的接口。 type PendingTaskRepository interface { + FindAllPendingTasks() ([]models.PendingTask, error) + FindPendingTriggerByPlanID(planID uint) (*models.PendingTask, error) + DeletePendingTasksByIDs(ids []uint) error + CreatePendingTask(task *models.PendingTask) error CreatePendingTasksInBatch(tasks []*models.PendingTask) error // ClaimNextAvailableTask 原子地认领下一个可用的任务。 // 它会同时返回被认领任务对应的日志对象，以及被删除的待办任务对象的内存副本。 @@ -28,8 +33,41 @@ func NewGormPendingTaskRepository(db *gorm.DB) PendingTaskRepository { return &gormPendingTaskRepository{db: db} } +func (r *gormPendingTaskRepository) FindAllPendingTasks() ([]models.PendingTask, error) { + var tasks []models.PendingTask + // 预加载 Task 以便后续访问 Task.PlanID + err := r.db.Preload("Task").Find(&tasks).Error + return tasks, err +} + +func (r *gormPendingTaskRepository) FindPendingTriggerByPlanID(planID uint) (*models.PendingTask, error) { + var pendingTask models.PendingTask + err := r.db. + Joins("JOIN tasks ON tasks.id = pending_tasks.task_id"). + Where("tasks.plan_id = ? AND tasks.type = ?", planID, models.TaskPlanAnalysis). + First(&pendingTask).Error + if errors.Is(err, gorm.ErrRecordNotFound) { + return nil, nil // 未找到不是错误 + } + return &pendingTask, err +} + +func (r *gormPendingTaskRepository) DeletePendingTasksByIDs(ids []uint) error { + if len(ids) == 0 { + return nil + } + return r.db.Where("id IN ?", ids).Delete(&models.PendingTask{}).Error +} + +func (r *gormPendingTaskRepository) CreatePendingTask(task *models.PendingTask) error { + return r.db.Create(task).Error +} + // CreatePendingTasksInBatch 在一次数据库调用中创建多个待执行任务条目。 func (r *gormPendingTaskRepository) CreatePendingTasksInBatch(tasks []*models.PendingTask) error { + if len(tasks) == 0 { + return nil + } return r.db.Create(&tasks).Error } diff --git a/internal/infra/repository/plan_repository.go b/internal/infra/repository/plan_repository.go index e32236d..66ebd8d 100644 --- a/internal/infra/repository/plan_repository.go +++ b/internal/infra/repository/plan_repository.go @@ -42,6 +42,12 @@ type PlanRepository interface { DeleteTask(id int) error // FindPlanAnalysisTaskByParamsPlanID 根据Parameters中的ParamsPlanID字段值查找TaskPlanAnalysis类型的Task FindPlanAnalysisTaskByParamsPlanID(paramsPlanID uint) (*models.Task, error) + // FindRunnablePlans 获取所有应执行的计划 + FindRunnablePlans() ([]*models.Plan, error) + // FindDisabledAndStoppedPlans 获取所有已禁用或已停止的计划 + FindDisabledAndStoppedPlans() ([]*models.Plan, error) + // FindPlanAnalysisTaskByPlanID 根据 PlanID 找到其关联的 'plan_analysis' 任务 + FindPlanAnalysisTaskByPlanID(planID uint) (*models.Task, error) } // gormPlanRepository 是 PlanRepository 的 GORM 实现 @@ -565,15 +571,56 @@ func (r *gormPlanRepository) createPlanAnalysisTask(tx *gorm.DB, plan *models.Pl return tx.Create(task).Error } -// updatePlanAnalysisTask 使用简单粗暴的删除再创建方式实现更新, 以控制AnalysisPlanTask的定义全部在createPlanAnalysisTask方法中 +// updatePlanAnalysisTask 使用更安全的方式更新触发器任务 func (r *gormPlanRepository) updatePlanAnalysisTask(tx *gorm.DB, plan *models.Plan) error { - task, err := r.findPlanAnalysisTaskByParamsPlanID(tx, plan.ID) - if err != nil { - return err + task, err := r.findPlanAnalysisTask(tx, plan.ID) + if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) { + return fmt.Errorf("查找现有计划分析任务失败: %w", err) } - err = r.deleteTask(tx, task.ID) - if err != nil { - return err + + // 如果触发器任务不存在，则创建一个 + if task == nil { + return r.createPlanAnalysisTask(tx, plan) } - return r.createPlanAnalysisTask(tx, plan) + + // 如果存在，则更新它的名称和描述以反映计划的最新信息 + task.Name = fmt.Sprintf("'%s'计划触发器", plan.Name) + task.Description = fmt.Sprintf("计划名: %s, 计划ID: %d", plan.Name, plan.ID) + + return tx.Save(task).Error +} + +func (r *gormPlanRepository) FindRunnablePlans() ([]*models.Plan, error) { + var plans []*models.Plan + err := r.db. + Where("status = ?", models.PlanStatusEnabled). + Where( + r.db.Where("execution_type = ?", models.PlanExecutionTypeManual). + Or("execution_type = ? AND (execute_num = 0 OR execute_count < execute_num)", models.PlanExecutionTypeAutomatic), + ). + Find(&plans).Error + return plans, err +} + +func (r *gormPlanRepository) FindDisabledAndStoppedPlans() ([]*models.Plan, error) { + var plans []*models.Plan + err := r.db. + Where("status = ? OR status = ?", models.PlanStatusDisabled, models.PlanStatusStopeed). + Find(&plans).Error + return plans, err +} + +// findPlanAnalysisTask 是一个内部使用的、更高效的查找方法 +func (r *gormPlanRepository) findPlanAnalysisTask(tx *gorm.DB, planID uint) (*models.Task, error) { + var task models.Task + err := tx.Where("plan_id = ? AND type = ?", planID, models.TaskPlanAnalysis).First(&task).Error + if errors.Is(err, gorm.ErrRecordNotFound) { + return nil, nil // 未找到不是错误，返回nil, nil + } + return &task, err +} + +// FindPlanAnalysisTaskByPlanID 是暴露给外部的公共方法 +func (r *gormPlanRepository) FindPlanAnalysisTaskByPlanID(planID uint) (*models.Task, error) { + return r.findPlanAnalysisTask(r.db, planID) }