拆分task包

internal/domain/task/analysis_plan_task_manager.go

@@ -1,316 +0,0 @@
-package task
-
-import (
-	"fmt"
-	"sync"
-	"time"
-
-	"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 负责管理分析计划的触发器任务。
-// 它确保数据库中可执行的计划在待执行队列中有对应的触发器，并移除无效的触发器。
-// 这是一个有状态的组件，包含一个互斥锁以确保并发安全。
-type AnalysisPlanTaskManager struct {
-	planRepo         repository.PlanRepository
-	pendingTaskRepo  repository.PendingTaskRepository
-	executionLogRepo repository.ExecutionLogRepository
-	logger           *logs.Logger
-	mu               sync.Mutex
-}
-
-// 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,
-	}
-}
-
-// 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 {
-		return fmt.Errorf("获取刷新数据失败: %w", err)
-	}
-
-	// 2. 清理所有与失效计划相关的待执行任务
-	if err := m.cleanupInvalidTasks(invalidPlanIDs, pendingTasks); err != nil {
-		// 仅记录错误，清理失败不应阻止新任务的添加
-		m.logger.Errorf("清理无效任务时出错: %v", err)
-	}
-
-	// 3. 添加或更新触发器
-	if err := m.addOrUpdateTriggers(runnablePlans, pendingTasks); err != nil {
-		return fmt.Errorf("添加或更新触发器时出错: %w", err)
-	}
-
-	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 {
-		var expectedExecuteAt time.Time
-		if plan.ExecutionType == models.PlanExecutionTypeManual {
-			// 手动计划，如果再次触发，则立即执行
-			expectedExecuteAt = time.Now()
-		} else { // 自动计划
-			// 自动计划，根据 Cron 表达式计算下一次执行时间
-			next, err := utils.GetNextCronTime(plan.CronExpression)
-			if err != nil {
-				m.logger.Errorf("为计划 #%d 解析Cron表达式失败，无法更新触发器: %v", plan.ID, err)
-				return fmt.Errorf("解析 Cron 表达式失败: %w", err)
-			}
-			expectedExecuteAt = next
-		}
-
-		// 如果计算出的执行时间与当前待执行任务的时间不一致，则更新
-		if !existingTrigger.ExecuteAt.Equal(expectedExecuteAt) {
-			m.logger.Infof("计划 #%d 的执行时间已变更，正在更新触发器 #%d 的执行时间从 %v 到 %v...", plan.ID, existingTrigger.ID, existingTrigger.ExecuteAt, expectedExecuteAt)
-			if err := m.pendingTaskRepo.UpdatePendingTaskExecuteAt(existingTrigger.ID, expectedExecuteAt); err != nil {
-				m.logger.Errorf("更新触发器 #%d 的执行时间失败: %v", existingTrigger.ID, err)
-				return fmt.Errorf("更新触发器执行时间失败: %w", err)
-			}
-		} else {
-			m.logger.Infof("计划 #%d 的触发器已存在且执行时间无需更新。", plan.ID)
-		}
-		return nil // 触发器已存在且已处理更新，直接返回
-	}
-
-	// 如果触发器不存在，则创建新的触发器
-	m.logger.Infof("为计划 #%d 创建新的触发器...", planID)
-	return m.createTriggerTask(plan)
-}
-
-// EnsureAnalysisTaskDefinition 确保计划的分析任务定义存在于 tasks 表中。
-// 如果不存在，则会自动创建。此方法不涉及待执行队列。
-func (m *AnalysisPlanTaskManager) EnsureAnalysisTaskDefinition(planID uint) error {
-	m.mu.Lock()
-	defer m.mu.Unlock()
-
-	plan, err := m.planRepo.GetBasicPlanByID(planID)
-	if err != nil {
-		return fmt.Errorf("确保分析任务定义失败：获取计划 #%d 基本信息时出错: %w", planID, err)
-	}
-
-	analysisTask, err := m.planRepo.FindPlanAnalysisTaskByPlanID(plan.ID)
-	if err != nil {
-		return fmt.Errorf("确保分析任务定义失败：查找计划 #%d 的分析任务时出错: %w", plan.ID, err)
-	}
-
-	if analysisTask == nil {
-		m.logger.Infof("未找到计划 #%d 关联的 'plan_analysis' 任务定义，将自动创建...", plan.ID)
-		_, err := m.planRepo.CreatePlanAnalysisTask(plan) // CreatePlanAnalysisTask returns *models.Task, error
-		if err != nil {
-			return fmt.Errorf("自动创建 'plan_analysis' 任务定义失败: %w", err)
-		}
-		m.logger.Infof("已成功为计划 #%d 创建 'plan_analysis' 任务定义。", plan.ID)
-	} else {
-		m.logger.Infof("计划 #%d 的 'plan_analysis' 任务定义已存在。", plan.ID)
-	}
-
-	return nil
-}
-
-// --- 内部私有方法 ---
-
-// 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.FindInactivePlans()
-	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.UpdateTaskExecutionLogStatusByIDs(logsToCancelIDs, models.ExecutionStatusCancelled); err != nil {
-		// 这是一个非关键性错误，只记录日志
-		m.logger.Warnf("批量更新日志状态为 'Cancelled' 失败: %v", err)
-	}
-
-	return nil
-}
-
-// addOrUpdateTriggers 检查、更新或创建触发器。
-func (m *AnalysisPlanTaskManager) addOrUpdateTriggers(runnablePlans []*models.Plan, allPendingTasks []models.PendingTask) error {
-	// 创建一个映射，存放所有已在队列中的计划触发器
-	pendingTriggersMap := make(map[uint]models.PendingTask)
-	for _, pt := range allPendingTasks {
-		if pt.Task != nil && pt.Task.Type == models.TaskPlanAnalysis {
-			pendingTriggersMap[pt.Task.PlanID] = pt
-		}
-	}
-
-	for _, plan := range runnablePlans {
-		existingTrigger, exists := pendingTriggersMap[plan.ID]
-
-		if exists {
-			// --- 新增逻辑：检查并更新现有触发器 ---
-			// 只对自动计划检查时间更新
-			if plan.ExecutionType == models.PlanExecutionTypeAutomatic {
-				next, err := utils.GetNextCronTime(plan.CronExpression)
-				if err != nil {
-					m.logger.Errorf("为计划 #%d 解析Cron表达式失败，跳过更新: %v", plan.ID, err)
-					continue
-				}
-				// 如果数据库中记录的执行时间与根据当前Cron表达式计算出的下一次时间不一致，则更新
-				if !existingTrigger.ExecuteAt.Equal(next) {
-					m.logger.Infof("计划 #%d 的执行时间已变更，正在更新触发器 #%d 的执行时间从 %v 到 %v...", plan.ID, existingTrigger.ID, existingTrigger.ExecuteAt, next)
-					if err := m.pendingTaskRepo.UpdatePendingTaskExecuteAt(existingTrigger.ID, next); err != nil {
-						m.logger.Errorf("更新触发器 #%d 的执行时间失败: %v", existingTrigger.ID, err)
-					}
-				}
-			}
-		} else {
-			// --- 原有逻辑：为缺失的计划创建新触发器 ---
-			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 {
-		m.logger.Warnf("未找到计划 #%d 关联的 'plan_analysis' 任务定义，将自动创建...", plan.ID)
-		newAnalysisTask, err := m.planRepo.CreatePlanAnalysisTask(plan)
-		if err != nil {
-			return fmt.Errorf("自动创建 'plan_analysis' 任务定义失败: %w", err)
-		}
-		analysisTask = newAnalysisTask
-		m.logger.Infof("已成功为计划 #%d 创建 'plan_analysis' 任务定义 (ID: %d)", plan.ID, analysisTask.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/domain/task/delay_task.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"time"
 
+	"git.huangwc.com/pig/pig-farm-controller/internal/domain/scheduler"
 	"git.huangwc.com/pig/pig-farm-controller/internal/infra/logs"
 	"git.huangwc.com/pig/pig-farm-controller/internal/infra/models"
 )
@@ -19,7 +20,7 @@ type DelayTask struct {
 	logger        *logs.Logger
 }
 
-func NewDelayTask(logger *logs.Logger, executionTask *models.TaskExecutionLog) Task {
+func NewDelayTask(logger *logs.Logger, executionTask *models.TaskExecutionLog) scheduler.Task {
 	return &DelayTask{
 		executionTask: executionTask,
 		logger:        logger,

internal/domain/task/release_feed_weight_task.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"git.huangwc.com/pig/pig-farm-controller/internal/domain/device"
+	"git.huangwc.com/pig/pig-farm-controller/internal/domain/scheduler"
 	"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"
@@ -40,12 +41,12 @@ func NewReleaseFeedWeightTask(
 	deviceRepo repository.DeviceRepository,
 	deviceService device.Service,
 	logger *logs.Logger,
-) Task {
+) scheduler.Task {
 	return &ReleaseFeedWeightTask{
 		claimedLog:     claimedLog,
 		deviceRepo:     deviceRepo,
 		sensorDataRepo: sensorDataRepo,
-		feedPort:       deviceService, // 直接注入
+		feedPort:       deviceService,
 		logger:         logger,
 	}
 }

internal/domain/task/scheduler.go

@@ -1,462 +0,0 @@
-package task
-
-import (
-	"errors"
-	"sync"
-	"time"
-
-	"git.huangwc.com/pig/pig-farm-controller/internal/domain/device"
-	"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"
-	"github.com/panjf2000/ants/v2"
-	"gorm.io/gorm"
-)
-
-// ProgressTracker 仅用于在内存中提供计划执行的并发锁
-type ProgressTracker struct {
-	mu           sync.Mutex
-	cond         *sync.Cond    // 用于实现阻塞锁
-	runningPlans map[uint]bool // key: planExecutionLogID, value: true (用作内存锁)
-}
-
-// NewProgressTracker 创建一个新的进度跟踪器
-func NewProgressTracker() *ProgressTracker {
-	t := &ProgressTracker{
-		runningPlans: make(map[uint]bool),
-	}
-	t.cond = sync.NewCond(&t.mu)
-	return t
-}
-
-// TryLock (非阻塞) 尝试锁定一个计划。如果计划未被锁定，则锁定并返回 true。
-func (t *ProgressTracker) TryLock(planLogID uint) bool {
-	t.mu.Lock()
-	defer t.mu.Unlock()
-	if t.runningPlans[planLogID] {
-		return false // 已被锁定
-	}
-	t.runningPlans[planLogID] = true
-	return true
-}
-
-// Lock (阻塞) 获取一个计划的执行锁。如果锁已被占用，则会一直等待直到锁被释放。
-func (t *ProgressTracker) Lock(planLogID uint) {
-	t.mu.Lock()
-	// 当计划正在运行时，调用 t.cond.Wait() 会原子地解锁 mu 并挂起当前协程。
-	// 当被唤醒时，它会重新锁定 mu 并再次检查循环条件。
-	for t.runningPlans[planLogID] {
-		t.cond.Wait()
-	}
-	// 获取到锁
-	t.runningPlans[planLogID] = true
-	t.mu.Unlock()
-}
-
-// Unlock 解锁一个计划，并唤醒所有正在等待此锁的协程。
-func (t *ProgressTracker) Unlock(planLogID uint) {
-	t.mu.Lock()
-	defer t.mu.Unlock()
-	delete(t.runningPlans, planLogID)
-	// 唤醒所有在此条件上等待的协程
-	t.cond.Broadcast()
-}
-
-// GetRunningPlanIDs 获取当前所有正在执行的计划ID列表
-func (t *ProgressTracker) GetRunningPlanIDs() []uint {
-	t.mu.Lock()
-	defer t.mu.Unlock()
-	ids := make([]uint, 0, len(t.runningPlans))
-	for id := range t.runningPlans {
-		ids = append(ids, id)
-	}
-	return ids
-}
-
-// Scheduler 是核心的、持久化的任务调度器
-type Scheduler struct {
-	logger                  *logs.Logger
-	pollingInterval         time.Duration
-	workers                 int
-	pendingTaskRepo         repository.PendingTaskRepository
-	executionLogRepo        repository.ExecutionLogRepository
-	deviceRepo              repository.DeviceRepository
-	sensorDataRepo          repository.SensorDataRepository
-	planRepo                repository.PlanRepository
-	analysisPlanTaskManager *AnalysisPlanTaskManager
-	progressTracker         *ProgressTracker
-	deviceService           device.Service
-
-	pool     *ants.Pool // 使用 ants 协程池来管理并发
-	wg       sync.WaitGroup
-	stopChan chan struct{} // 用于停止主循环的信号通道
-}
-
-// NewScheduler 创建一个新的调度器实例
-func NewScheduler(
-	pendingTaskRepo repository.PendingTaskRepository,
-	executionLogRepo repository.ExecutionLogRepository,
-	deviceRepo repository.DeviceRepository,
-	sensorDataRepo repository.SensorDataRepository,
-	planRepo repository.PlanRepository,
-	analysisPlanTaskManager *AnalysisPlanTaskManager,
-	logger *logs.Logger,
-	deviceService device.Service,
-	interval time.Duration,
-	numWorkers int,
-) *Scheduler {
-	return &Scheduler{
-		pendingTaskRepo:         pendingTaskRepo,
-		executionLogRepo:        executionLogRepo,
-		deviceRepo:              deviceRepo,
-		sensorDataRepo:          sensorDataRepo,
-		planRepo:                planRepo,
-		analysisPlanTaskManager: analysisPlanTaskManager,
-		logger:                  logger,
-		deviceService:           deviceService,
-		pollingInterval:         interval,
-		workers:                 numWorkers,
-		progressTracker:         NewProgressTracker(),
-		stopChan:                make(chan struct{}), // 初始化停止信号通道
-	}
-}
-
-// Start 启动调度器，包括初始化协程池和启动主轮询循环
-func (s *Scheduler) Start() {
-	s.logger.Warnf("任务调度器正在启动，工作协程数: %d...", s.workers)
-	pool, err := ants.NewPool(s.workers, ants.WithPanicHandler(func(err interface{}) {
-		s.logger.Errorf("[严重] 任务执行时发生 panic: %v", err)
-	}))
-	if err != nil {
-		panic("初始化协程池失败: " + err.Error())
-	}
-	s.pool = pool
-
-	s.wg.Add(1)
-	go s.run()
-	s.logger.Warnf("任务调度器已成功启动")
-}
-
-// Stop 优雅地停止调度器
-func (s *Scheduler) Stop() {
-	s.logger.Warnf("正在停止任务调度器...")
-	close(s.stopChan) // 1. 发出停止信号，停止主循环
-	s.wg.Wait()       // 2. 等待主循环完成
-	s.pool.Release()  // 3. 释放 ants 池 (等待所有已提交的任务执行完毕)
-	s.logger.Warnf("任务调度器已安全停止")
-}
-
-// run 是主轮询循环，负责从数据库认领任务并提交到协程池
-func (s *Scheduler) run() {
-	defer s.wg.Done()
-	ticker := time.NewTicker(s.pollingInterval)
-	defer ticker.Stop()
-
-	for {
-		select {
-		case <-s.stopChan:
-			// 收到停止信号，退出循环
-			return
-		case <-ticker.C:
-			// 定时触发任务认领和提交
-			go s.claimAndSubmit()
-		}
-	}
-}
-
-// claimAndSubmit 实现了最终的“认领-锁定-执行 或 等待-放回”的健壮逻辑
-func (s *Scheduler) claimAndSubmit() {
-	runningPlanIDs := s.progressTracker.GetRunningPlanIDs()
-
-	claimedLog, pendingTask, err := s.pendingTaskRepo.ClaimNextAvailableTask(runningPlanIDs)
-	if err != nil {
-		if !errors.Is(err, gorm.ErrRecordNotFound) {
-			s.logger.Errorf("认领任务时发生错误: %v", err)
-		}
-		// gorm.ErrRecordNotFound 说明没任务要执行
-		return
-	}
-
-	// 尝试获取内存执行锁
-	if s.progressTracker.TryLock(claimedLog.PlanExecutionLogID) {
-		// 成功获取锁，正常派发任务
-		err = s.pool.Submit(func() {
-			defer s.progressTracker.Unlock(claimedLog.PlanExecutionLogID)
-			s.processTask(claimedLog)
-		})
-		if err != nil {
-			s.logger.Errorf("向协程池提交任务失败: %v", err)
-			// 提交失败，必须释放刚刚获取的锁
-			s.progressTracker.Unlock(claimedLog.PlanExecutionLogID)
-			// 同样需要将任务安全放回
-			s.handleRequeue(claimedLog.PlanExecutionLogID, pendingTask)
-		}
-	} else {
-		// 获取锁失败，说明有“兄弟”任务正在执行。执行“锁定并安全放回”逻辑。
-		s.handleRequeue(claimedLog.PlanExecutionLogID, pendingTask)
-	}
-}
-
-// handleRequeue 同步地、安全地将一个无法立即执行的任务放回队列。
-func (s *Scheduler) handleRequeue(planExecutionLogID uint, taskToRequeue *models.PendingTask) {
-	s.logger.Warnf("计划 %d 正在执行，任务 %d (TaskID: %d) 将等待并重新入队...", planExecutionLogID, taskToRequeue.ID, taskToRequeue.TaskID)
-
-	// 1. 阻塞式地等待，直到可以获取到该计划的锁。
-	s.progressTracker.Lock(planExecutionLogID)
-	defer s.progressTracker.Unlock(planExecutionLogID)
-
-	// 2. 在持有锁的情况下，将任务安全地放回队列。
-	if err := s.pendingTaskRepo.RequeueTask(taskToRequeue); err != nil {
-		s.logger.Errorf("[严重] 任务重新入队失败, 原始PendingTaskID: %d, 错误: %v", taskToRequeue.ID, err)
-		return
-	}
-
-	s.logger.Warnf("任务 (原始ID: %d) 已成功重新入队，并已释放计划 %d 的锁。", taskToRequeue.ID, planExecutionLogID)
-}
-
-// processTask 处理单个任务的逻辑
-func (s *Scheduler) processTask(claimedLog *models.TaskExecutionLog) {
-	s.logger.Warnf("开始处理任务, 日志ID: %d, 任务ID: %d, 任务名称: %s, 描述: %s",
-		claimedLog.ID, claimedLog.TaskID, claimedLog.Task.Name, claimedLog.Task.Description)
-
-	claimedLog.StartedAt = time.Now()
-	claimedLog.Status = models.ExecutionStatusCompleted // 先乐观假定任务成功, 后续失败了再改
-	defer s.updateTaskExecutionLogStatus(claimedLog)
-
-	// 执行任务
-	err := s.runTask(claimedLog)
-	if err != nil {
-		claimedLog.Status = models.ExecutionStatusFailed
-		claimedLog.Output = err.Error()
-
-		// 任务失败时，调用统一的终止服务
-		s.handlePlanTermination(claimedLog.PlanExecutionLogID, "子任务执行失败: "+err.Error())
-		return
-	}
-
-	// 如果是计划分析任务，它的职责是解析和分发任务，到此即完成，不参与后续的计划完成度检查。
-	if claimedLog.Task.Type == models.TaskPlanAnalysis {
-		s.logger.Warnf("完成计划分析任务, 日志ID: %d", claimedLog.ID)
-		return
-	}
-
-	// --- 以下是常规任务的完成逻辑 ---
-	s.logger.Warnf("完成任务, 日志ID: %d", claimedLog.ID)
-
-	// 检查是否是最后一个任务
-	incompleteCount, err := s.executionLogRepo.CountIncompleteTasksByPlanLogID(claimedLog.PlanExecutionLogID)
-	if err != nil {
-		s.logger.Errorf("检查计划 %d 的未完成任务数时出错: %v", claimedLog.PlanExecutionLogID, err)
-		return
-	}
-
-	// 如果此计划执行中，未完成的任务只剩下当前这一个（因为当前任务的状态此时在数据库中仍为 'started'），
-	// 则认为整个计划已完成。
-	if incompleteCount == 1 {
-		s.handlePlanCompletion(claimedLog.PlanExecutionLogID)
-	}
-}
-
-// runTask 用于执行具体任务
-func (s *Scheduler) runTask(claimedLog *models.TaskExecutionLog) error {
-	// 这是个特殊任务, 用于解析Plan并将解析出的任务队列添加到待执行队列中
-	if claimedLog.Task.Type == models.TaskPlanAnalysis {
-		// 解析plan
-		err := s.analysisPlan(claimedLog)
-		if err != nil {
-			// TODO 这里要处理一下, 比如再插一个新的触发器回去
-			s.logger.Errorf("[严重] 计划解析失败, 日志ID: %d, 错误: %v", claimedLog.ID, err)
-			return err
-		}
-
-	} else {
-		// 执行普通任务
-		task := s.taskFactory(claimedLog)
-
-		if err := task.Execute(); err != nil {
-			s.logger.Errorf("[严重] 任务执行失败, 日志ID: %d, 错误: %v", claimedLog.ID, err)
-			task.OnFailure(err)
-			return err
-		}
-
-	}
-	return nil
-}
-
-// taskFactory 会根据任务类型初始化对应任务
-func (s *Scheduler) taskFactory(claimedLog *models.TaskExecutionLog) Task {
-	switch claimedLog.Task.Type {
-	case models.TaskTypeWaiting:
-		return NewDelayTask(s.logger, claimedLog)
-	case models.TaskTypeReleaseFeedWeight:
-		return NewReleaseFeedWeightTask(claimedLog, s.sensorDataRepo, s.deviceRepo, s.deviceService, s.logger)
-
-	default:
-		// TODO 这里直接panic合适吗? 不过这个场景确实不该出现任何异常的任务类型
-		panic("不支持的任务类型")
-	}
-}
-
-// analysisPlan 解析Plan并将解析出的Task列表插入待执行队列中
-func (s *Scheduler) analysisPlan(claimedLog *models.TaskExecutionLog) error {
-	// 创建Plan执行记录
-	// 从任务的 Parameters 中解析出真实的 PlanID
-	var params struct {
-		PlanID uint `json:"plan_id"`
-	}
-	if err := claimedLog.Task.ParseParameters(&params); err != nil {
-		s.logger.Errorf("解析任务参数中的计划ID失败，日志ID: %d, 错误: %v", claimedLog.ID, err)
-		return err
-	}
-	realPlanID := params.PlanID
-
-	planLog := &models.PlanExecutionLog{
-		PlanID:    realPlanID, // 使用从参数中解析出的真实 PlanID
-		Status:    models.ExecutionStatusStarted,
-		StartedAt: time.Now(),
-	}
-	if err := s.executionLogRepo.CreatePlanExecutionLog(planLog); err != nil {
-		s.logger.Errorf("[严重] 创建计划执行日志失败, 日志ID: %d, 错误: %v", claimedLog.ID, err)
-		return err
-	}
-
-	// 解析出Task列表
-	tasks, err := s.planRepo.FlattenPlanTasks(realPlanID)
-	if err != nil {
-		s.logger.Errorf("[严重] 解析计划失败, 日志ID: %d, 错误: %v", claimedLog.ID, err)
-		return err
-	}
-
-	// 写入执行历史
-	taskLogs := make([]*models.TaskExecutionLog, len(tasks))
-	for i, task := range tasks {
-		taskLogs[i] = &models.TaskExecutionLog{
-			PlanExecutionLogID: planLog.ID,
-			TaskID:             task.ID,
-			Status:             models.ExecutionStatusWaiting,
-		}
-
-	}
-	err = s.executionLogRepo.CreateTaskExecutionLogsInBatch(taskLogs)
-	if err != nil {
-		s.logger.Errorf("[严重] 写入执行历史, 日志ID: %d, 错误: %v", claimedLog.ID, err)
-		return err
-	}
-
-	// 写入待执行队列
-	pendingTasks := make([]*models.PendingTask, len(tasks))
-	for i, task := range tasks {
-		pendingTasks[i] = &models.PendingTask{
-			TaskID:             task.ID,
-			TaskExecutionLogID: taskLogs[i].ID, // 使用正确的 TaskExecutionLogID
-
-			// 待执行队列是通过任务触发时间排序的, 且只要在调度器获取的时间点之前的都可以被触发
-			ExecuteAt: time.Now().Add(time.Duration(i) * time.Second),
-		}
-	}
-	err = s.pendingTaskRepo.CreatePendingTasksInBatch(pendingTasks)
-	if err != nil {
-		s.logger.Errorf("[严重] 写入待执行队列, 日志ID: %d, 错误: %v", claimedLog.ID, err)
-		return err
-	}
-
-	// --- 处理空计划的边缘情况 ---
-	// 如果一个计划被解析后，发现其任务列表为空，
-	// 那么它实际上已经“执行”完毕了，我们需要在这里手动为它创建下一次的触发器。
-	if len(tasks) == 0 {
-		s.handlePlanCompletion(planLog.ID)
-	}
-
-	return nil
-}
-
-// updateTaskExecutionLogStatus 修改任务历史中的执行状态
-func (s *Scheduler) updateTaskExecutionLogStatus(claimedLog *models.TaskExecutionLog) error {
-	claimedLog.EndedAt = time.Now()
-
-	if err := s.executionLogRepo.UpdateTaskExecutionLog(claimedLog); err != nil {
-		s.logger.Errorf("[严重] 更新任务执行日志失败, 日志ID: %d, 错误: %v", claimedLog.ID, err)
-		return err
-	}
-
-	return nil
-}
-
-// handlePlanTermination 集中处理计划的终止逻辑（失败或取消）
-func (s *Scheduler) handlePlanTermination(planLogID uint, reason string) {
-	// 1. 从待执行队列中删除所有相关的子任务
-	if err := s.pendingTaskRepo.DeletePendingTasksByPlanLogID(planLogID); err != nil {
-		s.logger.Errorf("从待执行队列中删除计划 %d 的后续任务时出错: %v", planLogID, err)
-	}
-
-	// 2. 将父计划的执行日志标记为失败
-	if err := s.executionLogRepo.FailPlanExecution(planLogID, reason); err != nil {
-		s.logger.Errorf("标记计划执行日志 %d 为失败时出错: %v", planLogID, err)
-	}
-
-	// 3. 将所有未完成的子任务日志标记为已取消
-	if err := s.executionLogRepo.CancelIncompleteTasksByPlanLogID(planLogID, "父计划失败或被取消"); err != nil {
-		s.logger.Errorf("取消计划 %d 的后续任务日志时出错: %v", planLogID, err)
-	}
-
-	// 4. 将计划本身的状态更新为失败
-	planLog, err := s.executionLogRepo.FindPlanExecutionLogByID(planLogID)
-	if err != nil {
-		s.logger.Errorf("无法找到计划执行日志 %d 以更新父计划状态: %v", planLogID, err)
-		return
-	}
-	if err := s.planRepo.UpdatePlanStatus(planLog.PlanID, models.PlanStatusFailed); err != nil {
-		s.logger.Errorf("更新计划 %d 状态为 '失败' 时出错: %v", planLog.PlanID, err)
-	}
-}
-
-// handlePlanCompletion 集中处理计划成功完成后的所有逻辑
-func (s *Scheduler) handlePlanCompletion(planLogID uint) {
-	s.logger.Infof("计划执行 %d 的所有任务已完成，开始处理计划完成逻辑...", planLogID)
-
-	// 1. 通过 PlanExecutionLog 反查正确的顶层 PlanID
-	planExecutionLog, err := s.executionLogRepo.FindPlanExecutionLogByID(planLogID)
-	if err != nil {
-		s.logger.Errorf("获取计划执行日志 %d 失败: %v", planLogID, err)
-		return
-	}
-	topLevelPlanID := planExecutionLog.PlanID // 这才是正确的顶层计划ID
-
-	// 2. 获取计划的最新数据，这里我们只需要基本信息来判断执行类型和次数
-	plan, err := s.planRepo.GetBasicPlanByID(topLevelPlanID)
-	if err != nil {
-		s.logger.Errorf("获取计划 %d 的基本信息失败: %v", topLevelPlanID, err)
-		return
-	}
-
-	// 3. 在内存中计算新的计数值和状态
-	newExecuteCount := plan.ExecuteCount + 1
-	newStatus := plan.Status // 默认为当前状态
-
-	// 如果是自动计划且达到执行次数上限，或计划是手动类型，则更新计划状态为已停止
-	if (plan.ExecutionType == models.PlanExecutionTypeAutomatic && plan.ExecuteNum > 0 && newExecuteCount >= plan.ExecuteNum) || plan.ExecutionType == models.PlanExecutionTypeManual {
-		newStatus = models.PlanStatusStopped
-		s.logger.Infof("计划 %d 已完成执行，状态更新为 '执行完毕'。", topLevelPlanID)
-	}
-
-	// 4. 使用专门的方法来原子性地更新计数值和状态
-	if err := s.planRepo.UpdatePlanStateAfterExecution(topLevelPlanID, newExecuteCount, newStatus); err != nil {
-		s.logger.Errorf("更新计划 %d 的执行后状态失败: %v", topLevelPlanID, err)
-		return
-	}
-
-	// 5. 更新计划执行日志状态为完成
-	if err := s.executionLogRepo.UpdatePlanExecutionLogStatus(planLogID, models.ExecutionStatusCompleted); err != nil {
-		s.logger.Errorf("更新计划执行日志 %d 状态为 '完成' 失败: %v", planLogID, err)
-	}
-
-	// 6. 调用共享的 Manager 来处理触发器更新逻辑
-	// 只有当计划在本次执行后仍然是 Enabled 状态时，才需要创建下一次的触发器。
-	if newStatus == models.PlanStatusEnabled {
-		if err := s.analysisPlanTaskManager.CreateOrUpdateTrigger(topLevelPlanID); err != nil {
-			s.logger.Errorf("为计划 %d 创建/更新触发器失败: %v", topLevelPlanID, err)
-		}
-	} else {
-		s.logger.Infof("计划 %d 状态为 '%d'，无需创建下一次触发器。", topLevelPlanID, newStatus)
-	}
-}

internal/domain/task/task.go

@@ -1,30 +1,43 @@
 package task
 
 import (
+	"git.huangwc.com/pig/pig-farm-controller/internal/domain/device"
+	"git.huangwc.com/pig/pig-farm-controller/internal/domain/scheduler"
+	"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"
 )
 
-// Task 定义了所有可被调度器执行的任务必须实现的接口。
-type Task interface {
-	// Execute 是任务的核心执行逻辑。
-	// ctx: 用于控制任务的超时或取消。
-	// log: 包含了当前任务执行的完整上下文信息，包括从数据库中加载的任务参数等。
-	// 返回的 error 表示任务是否执行成功。调度器会根据返回的 error 是否为 nil 来决定任务状态。
-	Execute() error
-
-	// OnFailure 定义了当 Execute 方法返回错误时，需要执行的回滚或清理逻辑。
-	// log: 任务执行的上下文。
-	// executeErr: 从 Execute 方法返回的原始错误。
-	OnFailure(executeErr error)
+type taskFactory struct {
+	logger         *logs.Logger
+	sensorDataRepo repository.SensorDataRepository
+	deviceRepo     repository.DeviceRepository
+	deviceService  device.Service
 }
 
-// TaskFactory 是一个任务组装工厂, 可以根据Task类型获取到对应的初始化函数
-var TaskFactory = func(tt models.TaskType) Task {
-	switch tt {
-	case models.TaskTypeWaiting:
-		return &DelayTask{}
-	default:
-		// 出现位置任务类型说明业务逻辑出现重大问题, 一个异常任务被创建了出来
-		panic("发现未知任务类型")
+func NewTaskFactory(
+	logger *logs.Logger,
+	sensorDataRepo repository.SensorDataRepository,
+	deviceRepo repository.DeviceRepository,
+	deviceService device.Service,
+) scheduler.TaskFactory {
+	return &taskFactory{
+		logger:         logger,
+		sensorDataRepo: sensorDataRepo,
+		deviceRepo:     deviceRepo,
+		deviceService:  deviceService,
+	}
+}
+
+func (t *taskFactory) Production(claimedLog *models.TaskExecutionLog) scheduler.Task {
+	switch claimedLog.Task.Type {
+	case models.TaskTypeWaiting:
+		return NewDelayTask(t.logger, claimedLog)
+	case models.TaskTypeReleaseFeedWeight:
+		return NewReleaseFeedWeightTask(claimedLog, t.sensorDataRepo, t.deviceRepo, t.deviceService, t.logger)
+	default:
+		// TODO 这里直接panic合适吗? 不过这个场景确实不该出现任何异常的任务类型
+		t.logger.Panicf("不支持的任务类型: %s", claimedLog.Task.Type)
+		panic("不支持的任务类型") // 显式panic防编译器报错
 	}
 }