重构AnalysisPlanTaskManager

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 封装了创建和更新计划分析任务（即触发器）的逻辑。
+// AnalysisPlanTaskManager 负责管理分析计划的触发器任务。
-// 这是一个可被 Scheduler 和其他应用服务（如 PlanService）共享的无状态组件。
+// 它确保数据库中可执行的计划在待执行队列中有对应的触发器，并移除无效的触发器。
+// 这是一个有状态的组件，包含一个互斥锁以确保并发安全。
 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 创建或更新其关联的下一次触发任务。
+// Refresh 同步数据库中的计划状态和待执行队列中的触发器任务。
-// 这个方法是幂等的，可以安全地被多次调用。
+// 这是一个编排方法，将复杂的逻辑分解到多个内部方法中。
-func (m *AnalysisPlanTaskManager) CreateOrUpdateTrigger(ctx context.Context, planID uint) error {
+func (m *AnalysisPlanTaskManager) Refresh() error {
-	// 获取计划信息
+	m.mu.Lock()
-	plan, err := m.planRepo.GetBasicPlanByID(planID)
+	defer m.mu.Unlock()
+
+	m.logger.Info("开始同步计划任务管理器...")
+
+	// 1. 一次性获取所有需要的数据
+	runnablePlans, invalidPlanIDs, pendingTasks, err := m.getRefreshData()
 	if err != nil {
-		m.logger.Errorf("[严重] 获取计划失败, 错误: %v", err)
+		return fmt.Errorf("获取刷新数据失败: %w", err)
-		return err
 	}
 
-	// 获取触发任务
+	// 2. 清理所有与失效计划相关的待执行任务
-	task, err := m.planRepo.FindPlanAnalysisTaskByParamsPlanID(planID)
+	if err := m.cleanupInvalidTasks(invalidPlanIDs, pendingTasks); err != nil {
-	if err != nil {
+		// 仅记录错误，清理失败不应阻止新任务的添加
-		m.logger.Errorf("[严重] 获取计划解析任务失败, 错误: %v", err)
+		m.logger.Errorf("清理无效任务时出错: %v", err)
-		return err
 	}
 
-	// 写入执行日志
+	// 3. 为应执行但缺失的计划添加新触发器
-	taskLog := &models.TaskExecutionLog{
+	if err := m.addNewTriggers(runnablePlans, pendingTasks); err != nil {
-		TaskID: task.ID,
+		return fmt.Errorf("添加新触发器时出错: %w", err)
-		Status: models.ExecutionStatusWaiting,
-	}
-	if err := m.executionLogRepo.CreateTaskExecutionLogsInBatch([]*models.TaskExecutionLog{taskLog}); err != nil {
-		m.logger.Errorf("[严重] 创建任务执行日志失败, 错误: %v", err)
-		return err
 	}
 
-	// 写入待执行队列
+	m.logger.Info("计划任务管理器同步完成.")
-	next, err := utils.GetNextCronTime(plan.CronExpression)
+	return nil
-	if err != nil {
+}
-		m.logger.Errorf("[严重] 执行时间解析失败, 错误: %v", err)
+
-		return err
+// CreateOrUpdateTrigger 为给定的 planID 创建其关联的触发任务。
-	}
+// 这个方法是幂等的：如果一个有效的触发器已存在，它将不会重复创建。
-	pendingTask := &models.PendingTask{
+func (m *AnalysisPlanTaskManager) CreateOrUpdateTrigger(planID uint) error {
-		TaskID:             task.ID,
+	m.mu.Lock()
-		ExecuteAt:          next,
+	defer m.mu.Unlock()
-		TaskExecutionLogID: taskLog.ID,
+
-	}
+	// 检查计划是否可执行
-	err = m.pendingTaskRepo.CreatePendingTasksInBatch([]*models.PendingTask{pendingTask})
+	plan, err := m.planRepo.GetBasicPlanByID(planID)
-	if err != nil {
+	if err != nil {
-		m.logger.Errorf("[严重] 创建待执行任务失败, 错误: %v", err)
+		return fmt.Errorf("获取计划基本信息失败: %w", err)
-		return err
+	}
-	}
+	if plan.Status != models.PlanStatusEnabled {
-
+		return fmt.Errorf("计划 #%d 当前状态为 '%d'，无法创建触发器", planID, plan.Status)
-	m.logger.Infof("成功为 Plan %d 创建/更新了下一次的触发任务，执行时间: %v", planID, next)
+	}
+
+	// 幂等性检查：如果触发器已存在，则直接返回
+	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
 }

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;"`

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
 }

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
 }
 

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)
+	task, err := r.findPlanAnalysisTask(tx, plan.ID)
-	if err != nil {
+	if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
-		return err
+		return fmt.Errorf("查找现有计划分析任务失败: %w", err)
-	}
-	err = r.deleteTask(tx, task.ID)
-	if err != nil {
-		return err
 	}
+
+	// 如果触发器任务不存在，则创建一个
+	if task == nil {
 		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)
+}