2025-09-18 21:51:17 +08:00
6 changed files with 217 additions and 126 deletions
--- a/internal/app/service/task/analysis_plan_task_manager.go
+++ b/internal/app/service/task/analysis_plan_task_manager.go
@@ -0,0 +1,82 @@
+package task
+
+import (
+	"context"
+
+	"git.huangwc.com/pig/pig-farm-controller/internal/infra/logs"
+	"git.huangwc.com/pig/pig-farm-controller/internal/infra/models"
+	"git.huangwc.com/pig/pig-farm-controller/internal/infra/repository"
+	"git.huangwc.com/pig/pig-farm-controller/internal/infra/utils"
+)
+
+// AnalysisPlanTaskManager 封装了创建和更新计划分析任务（即触发器）的逻辑。
+// 这是一个可被 Scheduler 和其他应用服务（如 PlanService）共享的无状态组件。
+type AnalysisPlanTaskManager struct {
+	planRepo         repository.PlanRepository
+	pendingTaskRepo  repository.PendingTaskRepository
+	executionLogRepo repository.ExecutionLogRepository
+	logger           *logs.Logger
+}
+
+// NewAnalysisPlanTaskManager 是 AnalysisPlanTaskManager 的构造函数。
+func NewAnalysisPlanTaskManager(
+	planRepo repository.PlanRepository,
+	pendingTaskRepo repository.PendingTaskRepository,
+	executionLogRepo repository.ExecutionLogRepository,
+	logger *logs.Logger,
+) *AnalysisPlanTaskManager {
+	return &AnalysisPlanTaskManager{
+		planRepo:         planRepo,
+		pendingTaskRepo:  pendingTaskRepo,
+		executionLogRepo: executionLogRepo,
+		logger:           logger,
+	}
+}
+
+// CreateOrUpdateTrigger 为给定的 planID 创建或更新其关联的下一次触发任务。
+// 这个方法是幂等的，可以安全地被多次调用。
+func (m *AnalysisPlanTaskManager) CreateOrUpdateTrigger(ctx context.Context, planID uint) error {
+	// 获取计划信息
+	plan, err := m.planRepo.GetBasicPlanByID(planID)
+	if err != nil {
+		m.logger.Errorf("[严重] 获取计划失败, 错误: %v", err)
+		return err
+	}
+
+	// 获取触发任务
+	task, err := m.planRepo.FindPlanAnalysisTaskByParamsPlanID(planID)
+	if err != nil {
+		m.logger.Errorf("[严重] 获取计划解析任务失败, 错误: %v", err)
+		return 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
+	}
+
+	// 写入待执行队列
+	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)
+	return nil
+}
--- a/internal/app/service/task/plan_analysis_task.go
+++ b/internal/app/service/task/plan_analysis_task.go
--- a/internal/app/service/task/scheduler.go
+++ b/internal/app/service/task/scheduler.go
@@ -9,7 +9,6 @@ import (
 	"git.huangwc.com/pig/pig-farm-controller/internal/infra/logs"
 	"git.huangwc.com/pig/pig-farm-controller/internal/infra/models"
 	"git.huangwc.com/pig/pig-farm-controller/internal/infra/repository"
-	"git.huangwc.com/pig/pig-farm-controller/internal/infra/utils"
 	"github.com/panjf2000/ants/v2"
 	"gorm.io/gorm"
 )
@@ -109,6 +108,7 @@ type Scheduler struct {
 	pendingTaskRepo         repository.PendingTaskRepository
 	executionLogRepo        repository.ExecutionLogRepository
 	planRepo                repository.PlanRepository
+	analysisPlanTaskManager *AnalysisPlanTaskManager // <--- 注入共享的 Manager
 	progressTracker         *ProgressTracker
 	taskFactory             func(taskType models.TaskType) Task // 调度器需要注入一个任务工厂，用于创建任务实例

@@ -123,6 +123,7 @@ func NewScheduler(
 	pendingTaskRepo repository.PendingTaskRepository,
 	executionLogRepo repository.ExecutionLogRepository,
 	planRepo repository.PlanRepository,
+	analysisPlanTaskManager *AnalysisPlanTaskManager, // <--- 注入 Manager
 	taskFactory func(taskType models.TaskType) Task,
 	logger *logs.Logger,
 	interval time.Duration,
@@ -133,6 +134,7 @@ func NewScheduler(
 		pendingTaskRepo:         pendingTaskRepo,
 		executionLogRepo:        executionLogRepo,
 		planRepo:                planRepo,
+		analysisPlanTaskManager: analysisPlanTaskManager, // <--- 注入 Manager
 		logger:                  logger,
 		pollingInterval:         interval,
 		workers:                 numWorkers,
@@ -254,7 +256,8 @@ func (s *Scheduler) processTask(claimedLog *models.TaskExecutionLog) {

 	// 任务计数器校验, Plan的任务全部执行完成后需要插入一个新的PlanAnalysisTask用于触发下一次Plan的执行
 	if s.progressTracker.IsPlanOver(claimedLog.PlanExecutionLogID) {
-		err = s.createNewAnalysisPlanTask(claimedLog.Task.PlanID)
+		// 调用共享的 Manager 来处理触发器更新逻辑
+		err = s.analysisPlanTaskManager.CreateOrUpdateTrigger(s.ctx, claimedLog.Task.PlanID)
 		if err != nil {
 			s.logger.Errorf("[严重] 创建计划分析任务失败, 当前Plan(%v)将无法进行下次触发, 错误: %v", claimedLog.Task.PlanID, err)
 		}
@@ -365,48 +368,3 @@ func (s *Scheduler) updateTaskExecutionLogStatus(claimedLog *models.TaskExecutio

 	return nil
 }
-
-// createNewAnalysisPlanTask 创建一个新的Plan解析任务用于下次触发plan执行
-func (s *Scheduler) createNewAnalysisPlanTask(planID uint) error {
-	// 获取计划信息
-	plan, err := s.planRepo.GetBasicPlanByID(planID)
-	if err != nil {
-		s.logger.Errorf("[严重] 获取计划失败, 错误: %v", err)
-		return err
-	}
-
-	// 获取触发任务
-	task, err := s.planRepo.FindPlanAnalysisTaskByParamsPlanID(planID)
-	if err != nil {
-		s.logger.Errorf("[严重] 获取计划解析任务失败, 错误: %v", err)
-		return err
-	}
-
-	// 写入执行日志
-	taskLog := &models.TaskExecutionLog{
-		TaskID: task.ID,
-		Status: models.ExecutionStatusWaiting,
-	}
-	if err := s.executionLogRepo.CreateTaskExecutionLogsInBatch([]*models.TaskExecutionLog{taskLog}); err != nil {
-		s.logger.Errorf("[严重] 创建任务执行日志失败, 错误: %v", err)
-		return err
-	}
-
-	// 写入待执行队列
-	next, err := utils.GetNextCronTime(plan.CronExpression)
-	if err != nil {
-		s.logger.Errorf("[严重] 执行时间解析失败, 错误: %v", err)
-		return err
-	}
-	pendingTask := &models.PendingTask{
-		TaskID:             task.ID,
-		ExecuteAt:          next,
-		TaskExecutionLogID: taskLog.ID,
-	}
-	err = s.pendingTaskRepo.CreatePendingTasksInBatch([]*models.PendingTask{pendingTask})
-	if err != nil {
-		s.logger.Errorf("[严重] 创建待执行任务失败, 错误: %v", err)
-		return err
-	}
-	return nil
-}
--- a/internal/app/service/task/task.go
+++ b/internal/app/service/task/task.go
@@ -1,12 +1,29 @@
 package task

 import (
+	"git.huangwc.com/pig/pig-farm-controller/internal/infra/logs"
 	"git.huangwc.com/pig/pig-farm-controller/internal/infra/models"
-	"git.huangwc.com/pig/pig-farm-controller/internal/infra/task"
 )

+// Task 定义了所有可被调度器执行的任务必须实现的接口。
+type Task interface {
+	// Execute 是任务的核心执行逻辑。
+	// ctx: 用于控制任务的超时或取消。
+	// log: 包含了当前任务执行的完整上下文信息，包括从数据库中加载的任务参数等。
+	// 返回的 error 表示任务是否执行成功。调度器会根据返回的 error 是否为 nil 来决定任务状态。
+	Execute() error
+
+	// ParseParams 解析及校验参数
+	ParseParams(logger *logs.Logger, claimedLog *models.TaskExecutionLog) error
+
+	// OnFailure 定义了当 Execute 方法返回错误时，需要执行的回滚或清理逻辑。
+	// log: 任务执行的上下文。
+	// executeErr: 从 Execute 方法返回的原始错误。
+	OnFailure(executeErr error)
+}
+
 // TaskFactory 是一个任务组装工厂, 可以根据Task类型获取到对应的初始化函数
-var TaskFactory = func(tt models.TaskType) task.Task {
+var TaskFactory = func(tt models.TaskType) Task {
 	switch tt {
 	case models.TaskTypeWaiting:
 		return &DelayTask{}
--- a/internal/infra/repository/plan_repository.go
+++ b/internal/infra/repository/plan_repository.go
@@ -1,10 +1,12 @@
 package repository

 import (
+	"encoding/json"
 	"errors"
 	"fmt"

 	"git.huangwc.com/pig/pig-farm-controller/internal/infra/models"
+	"gorm.io/datatypes"
 	"gorm.io/gorm"
 )

@@ -170,6 +172,11 @@ func (r *gormPlanRepository) CreatePlan(plan *models.Plan) error {
 		if err := tx.Create(plan).Error; err != nil {
 			return err
 		}
+
+		// 3. 创建触发器Task
+		if err := r.createPlanAnalysisTask(tx, plan); err != nil {
+			return err
+		}
 		return nil
 	})
 }
@@ -186,7 +193,12 @@ func (r *gormPlanRepository) updatePlanTx(tx *gorm.DB, plan *models.Plan) error
 	if err := r.validatePlanTree(tx, plan); err != nil {
 		return err
 	}
-	return r.reconcilePlanNode(tx, plan)
+	if err := r.reconcilePlanNode(tx, plan); err != nil {
+		return err
+	}
+
+	// 更新Plan触发器
+	return r.updatePlanAnalysisTask(tx, plan)
 }

 // validatePlanTree 对整个计划树进行全面的只读健康检查
@@ -461,6 +473,12 @@ func (r *gormPlanRepository) flattenPlanTasksRecursive(plan *models.Plan) ([]mod
 func (r *gormPlanRepository) DeleteTask(id int) error {
 	// 使用事务确保操作的原子性
 	return r.db.Transaction(func(tx *gorm.DB) error {
+		return r.deleteTask(tx, id)
+	})
+}
+
+// deleteTask 根据ID删除任务
+func (r *gormPlanRepository) deleteTask(tx *gorm.DB, id int) error {
 	// 1. 检查是否有待执行任务引用了这个任务
 	var pendingTaskCount int64
 	if err := tx.Model(&models.PendingTask{}).Where("task_id = ?", id).Count(&pendingTaskCount).Error; err != nil {
@@ -495,17 +513,21 @@ func (r *gormPlanRepository) DeleteTask(id int) error {
 	}

 	return nil
-	})
 }

 // FindPlanAnalysisTaskByParamsPlanID 根据Parameters中的ParamsPlanID字段值查找TaskPlanAnalysis类型的Task
 func (r *gormPlanRepository) FindPlanAnalysisTaskByParamsPlanID(paramsPlanID uint) (*models.Task, error) {
+	return r.findPlanAnalysisTaskByParamsPlanID(r.db, paramsPlanID)
+}
+
+// findPlanAnalysisTaskByParamsPlanID 使用指定db根据Parameters中的ParamsPlanID字段值查找TaskPlanAnalysis类型的Task
+func (r *gormPlanRepository) findPlanAnalysisTaskByParamsPlanID(tx *gorm.DB, paramsPlanID uint) (*models.Task, error) {
 	var task models.Task

 	// 构造JSON查询条件，查找Parameters中包含指定ParamsPlanID且Type为TaskPlanAnalysis的任务
 	// TODO 在JSON字段中查找特定键值的语法取决于数据库类型，这里使用PostgreSQL的语法
 	// TODO 如果使用的是MySQL，则需要相应调整查询条件
-	result := r.db.Where(
+	result := tx.Where(
 		"type = ? AND parameters->>'plan_id' = ?",
 		models.TaskPlanAnalysis,
 		fmt.Sprintf("%d", paramsPlanID),
@@ -520,3 +542,38 @@ func (r *gormPlanRepository) FindPlanAnalysisTaskByParamsPlanID(paramsPlanID uin

 	return &task, nil
 }
+
+// createPlanAnalysisTask 用于创建一个TaskPlanAnalysis类型的Task
+func (r *gormPlanRepository) createPlanAnalysisTask(tx *gorm.DB, plan *models.Plan) error {
+	m := map[string]interface{}{
+		models.ParamsPlanID: plan.ID,
+	}
+	parameters, err := json.Marshal(m)
+	if err != nil {
+		return err
+	}
+
+	task := &models.Task{
+		PlanID:         plan.ID,
+		Name:           fmt.Sprintf("'%v'计划触发器", plan.Name),
+		Description:    fmt.Sprintf("计划名: %v, 计划ID: %v", plan.Name, plan.ID),
+		ExecutionOrder: 0,
+		Type:           models.TaskPlanAnalysis,
+		Parameters:     datatypes.JSON(parameters),
+	}
+
+	return tx.Create(task).Error
+}
+
+// updatePlanAnalysisTask 使用简单粗暴的删除再创建方式实现更新, 以控制AnalysisPlanTask的定义全部在createPlanAnalysisTask方法中
+func (r *gormPlanRepository) updatePlanAnalysisTask(tx *gorm.DB, plan *models.Plan) error {
+	task, err := r.findPlanAnalysisTaskByParamsPlanID(tx, plan.ID)
+	if err != nil {
+		return err
+	}
+	err = r.deleteTask(tx, task.ID)
+	if err != nil {
+		return err
+	}
+	return r.createPlanAnalysisTask(tx, plan)
+}
--- a/internal/infra/task/task.go
+++ b/internal/infra/task/task.go
@@ -1,23 +0,0 @@
-package task
-
-import (
-	"git.huangwc.com/pig/pig-farm-controller/internal/infra/logs"
-	"git.huangwc.com/pig/pig-farm-controller/internal/infra/models"
-)
-
-// Task 定义了所有可被调度器执行的任务必须实现的接口。
-type Task interface {
-	// Execute 是任务的核心执行逻辑。
-	// ctx: 用于控制任务的超时或取消。
-	// log: 包含了当前任务执行的完整上下文信息，包括从数据库中加载的任务参数等。
-	// 返回的 error 表示任务是否执行成功。调度器会根据返回的 error 是否为 nil 来决定任务状态。
-	Execute() error
-
-	// ParseParams 解析及校验参数
-	ParseParams(logger *logs.Logger, claimedLog *models.TaskExecutionLog) error
-
-	// OnFailure 定义了当 Execute 方法返回错误时，需要执行的回滚或清理逻辑。
-	// log: 任务执行的上下文。
-	// executeErr: 从 Execute 方法返回的原始错误。
-	OnFailure(executeErr error)
-}