738 lines
25 KiB
Go
738 lines
25 KiB
Go
package repository
|
||
|
||
import (
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
|
||
"git.huangwc.com/pig/pig-farm-controller/internal/infra/models"
|
||
"gorm.io/datatypes"
|
||
"gorm.io/gorm"
|
||
)
|
||
|
||
// 定义仓库层可导出的公共错误
|
||
var (
|
||
ErrUpdateWithInvalidRoot = errors.New("更新操作的目标根计划无效或ID为0")
|
||
ErrNewSubPlanInUpdate = errors.New("计划树中包含一个ID为0的新子计划，更新操作只允许关联已存在的计划")
|
||
ErrNodeDoesNotExist = errors.New("计划树中包含一个或多个在数据库中不存在的计划")
|
||
ErrCreateWithNonZeroID = errors.New("创建操作的计划ID必须为0")
|
||
ErrSubPlanIDIsZeroOnCreate = errors.New("子计划ID为0，创建操作只允许关联已存在的计划")
|
||
ErrMixedContent = errors.New("计划不能同时包含任务和子计划")
|
||
ErrDeleteWithReferencedPlan = errors.New("禁止删除正在被引用的计划")
|
||
)
|
||
|
||
// PlanRepository 定义了与计划模型相关的数据库操作接口
|
||
// 这是为了让业务逻辑层依赖于抽象，而不是具体的数据库实现
|
||
type PlanRepository interface {
|
||
// ListBasicPlans 获取所有计划的基本信息，不包含子计划和任务详情
|
||
ListBasicPlans() ([]models.Plan, error)
|
||
// GetBasicPlanByID 根据ID获取计划的基本信息，不包含子计划和任务详情
|
||
GetBasicPlanByID(id uint) (*models.Plan, error)
|
||
// GetPlanByID 根据ID获取计划，包含子计划和任务详情
|
||
GetPlanByID(id uint) (*models.Plan, error)
|
||
// CreatePlan 创建一个新的计划
|
||
CreatePlan(plan *models.Plan) error
|
||
// UpdatePlan 更新计划，包括子计划和任务
|
||
UpdatePlan(plan *models.Plan) error
|
||
// UpdatePlanStatus 更新指定计划的状态
|
||
UpdatePlanStatus(id uint, status models.PlanStatus) error
|
||
// DeletePlan 根据ID删除计划，同时删除其关联的任务（非子任务）或子计划关联
|
||
DeletePlan(id uint) error
|
||
// FlattenPlanTasks 递归展开计划，返回按执行顺序排列的所有任务列表
|
||
FlattenPlanTasks(planID uint) ([]models.Task, error)
|
||
// DeleteTask 根据ID删除任务
|
||
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)
|
||
|
||
// CreatePlanAnalysisTask 创建一个 plan_analysis 类型的任务并返回它
|
||
CreatePlanAnalysisTask(plan *models.Plan) (*models.Task, error)
|
||
|
||
// FindPlansWithPendingTasks 查找所有正在执行的计划
|
||
FindPlansWithPendingTasks() ([]*models.Plan, error)
|
||
|
||
// DB 返回底层的数据库连接实例，用于服务层事务
|
||
DB() *gorm.DB
|
||
|
||
// StopPlanTransactionally 停止一个计划的执行，包括更新状态、移除待执行任务和更新执行日志
|
||
StopPlanTransactionally(planID uint) error
|
||
}
|
||
|
||
// gormPlanRepository 是 PlanRepository 的 GORM 实现
|
||
type gormPlanRepository struct {
|
||
db *gorm.DB
|
||
}
|
||
|
||
// NewGormPlanRepository 创建一个新的 PlanRepository GORM 实现实例
|
||
func NewGormPlanRepository(db *gorm.DB) PlanRepository {
|
||
return &gormPlanRepository{
|
||
db: db,
|
||
}
|
||
}
|
||
|
||
// ListBasicPlans 获取所有计划的基本信息，不包含子计划和任务详情
|
||
func (r *gormPlanRepository) ListBasicPlans() ([]models.Plan, error) {
|
||
var plans []models.Plan
|
||
// GORM 默认不会加载关联，除非使用 Preload，所以直接 Find 即可满足要求
|
||
result := r.db.Find(&plans)
|
||
if result.Error != nil {
|
||
return nil, result.Error
|
||
}
|
||
return plans, nil
|
||
}
|
||
|
||
// GetBasicPlanByID 根据ID获取计划的基本信息，不包含子计划和任务详情
|
||
func (r *gormPlanRepository) GetBasicPlanByID(id uint) (*models.Plan, error) {
|
||
var plan models.Plan
|
||
// GORM 默认不会加载关联，除非使用 Preload，所以直接 First 即可满足要求
|
||
result := r.db.First(&plan, id)
|
||
if result.Error != nil {
|
||
return nil, result.Error
|
||
}
|
||
return &plan, nil
|
||
}
|
||
|
||
// GetPlanByID 根据ID获取计划，包含子计划和任务详情
|
||
func (r *gormPlanRepository) GetPlanByID(id uint) (*models.Plan, error) {
|
||
var plan models.Plan
|
||
|
||
// 先获取基本计划信息
|
||
result := r.db.First(&plan, id)
|
||
if result.Error != nil {
|
||
return nil, result.Error
|
||
}
|
||
|
||
// 根据内容类型加载关联数据
|
||
switch plan.ContentType {
|
||
case models.PlanContentTypeSubPlans:
|
||
// 加载子计划引用
|
||
var subPlans []models.SubPlan
|
||
result = r.db.Where("parent_plan_id = ?", plan.ID).Order("execution_order").Find(&subPlans)
|
||
if result.Error != nil {
|
||
return nil, result.Error
|
||
}
|
||
|
||
// 递归加载每个子计划的完整信息
|
||
for i := range subPlans {
|
||
childPlan, err := r.GetPlanByID(subPlans[i].ChildPlanID)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
subPlans[i].ChildPlan = childPlan
|
||
}
|
||
|
||
plan.SubPlans = subPlans
|
||
case models.PlanContentTypeTasks:
|
||
// 加载任务
|
||
result = r.db.Preload("Tasks", func(taskDB *gorm.DB) *gorm.DB {
|
||
return taskDB.Order("execution_order")
|
||
}).First(&plan, id)
|
||
if result.Error != nil {
|
||
return nil, result.Error
|
||
}
|
||
default:
|
||
return nil, fmt.Errorf("未知的计划内容类型: %v; 计划ID: %v", plan.ContentType, plan.ID)
|
||
}
|
||
|
||
return &plan, nil
|
||
}
|
||
|
||
// CreatePlan 创建一个新的计划
|
||
func (r *gormPlanRepository) CreatePlan(plan *models.Plan) error {
|
||
return r.db.Transaction(func(tx *gorm.DB) error {
|
||
// 1. 前置校验
|
||
if plan.ID != 0 {
|
||
return ErrCreateWithNonZeroID
|
||
}
|
||
|
||
// 检查是否同时包含任务和子计划
|
||
if len(plan.Tasks) > 0 && len(plan.SubPlans) > 0 {
|
||
return ErrMixedContent
|
||
}
|
||
|
||
// 检查是否有重复的执行顺序
|
||
if err := plan.ValidateExecutionOrder(); err != nil {
|
||
return fmt.Errorf("计划 (ID: %d) 的执行顺序无效: %w", plan.ID, err)
|
||
}
|
||
|
||
// 如果是子计划类型，验证所有子计划是否存在且ID不为0
|
||
if plan.ContentType == models.PlanContentTypeSubPlans {
|
||
childIDsToValidate := make(map[uint]bool)
|
||
for _, subPlanLink := range plan.SubPlans {
|
||
if subPlanLink.ChildPlanID == 0 {
|
||
return ErrSubPlanIDIsZeroOnCreate
|
||
}
|
||
childIDsToValidate[subPlanLink.ChildPlanID] = true
|
||
}
|
||
|
||
var ids []uint
|
||
for id := range childIDsToValidate {
|
||
ids = append(ids, id)
|
||
}
|
||
|
||
if len(ids) > 0 {
|
||
var count int64
|
||
if err := tx.Model(&models.Plan{}).Where("id IN ?", ids).Count(&count).Error; err != nil {
|
||
return fmt.Errorf("验证子计划存在性失败: %w", err)
|
||
}
|
||
if int(count) != len(ids) {
|
||
return ErrNodeDoesNotExist
|
||
}
|
||
}
|
||
}
|
||
|
||
// 2. 创建根计划
|
||
// GORM 会自动处理关联的 Tasks (如果 ContentType 是 tasks 且 Task.ID 为 0)
|
||
if err := tx.Create(plan).Error; err != nil {
|
||
return err
|
||
}
|
||
|
||
// 3. 创建触发器Task
|
||
// 关键修改：调用 createPlanAnalysisTask 并处理其返回的 Task 对象
|
||
_, err := r.createPlanAnalysisTask(tx, plan)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
return nil
|
||
})
|
||
}
|
||
|
||
// UpdatePlan 是更新计划的公共入口点
|
||
func (r *gormPlanRepository) UpdatePlan(plan *models.Plan) error {
|
||
return r.db.Transaction(func(tx *gorm.DB) error {
|
||
return r.updatePlanTx(tx, plan)
|
||
})
|
||
}
|
||
|
||
// updatePlanTx 在事务中协调整个更新过程
|
||
func (r *gormPlanRepository) updatePlanTx(tx *gorm.DB, plan *models.Plan) error {
|
||
if err := r.validatePlanTree(tx, plan); err != nil {
|
||
return err
|
||
}
|
||
if err := r.reconcilePlanNode(tx, plan); err != nil {
|
||
return err
|
||
}
|
||
|
||
// 更新Plan触发器
|
||
return r.updatePlanAnalysisTask(tx, plan)
|
||
}
|
||
|
||
// validatePlanTree 对整个计划树进行全面的只读健康检查
|
||
func (r *gormPlanRepository) validatePlanTree(tx *gorm.DB, plan *models.Plan) error {
|
||
// 1. 检查根节点
|
||
if plan == nil || plan.ID == 0 {
|
||
return ErrUpdateWithInvalidRoot
|
||
}
|
||
if len(plan.Tasks) > 0 && len(plan.SubPlans) > 0 {
|
||
return fmt.Errorf("计划 (ID: %d) 不能同时包含任务和子计划", plan.ID)
|
||
}
|
||
|
||
// 2. 递归验证所有子节点，并检测循环引用
|
||
allIDs := make(map[uint]bool)
|
||
recursionStack := make(map[uint]bool)
|
||
if err := validateNodeAndDetectCycles(plan, allIDs, recursionStack); err != nil {
|
||
return err
|
||
}
|
||
|
||
// 3. 检查是否有重复的执行顺序
|
||
if err := plan.ValidateExecutionOrder(); err != nil {
|
||
return fmt.Errorf("计划 (ID: %d) 的执行顺序无效: %w", plan.ID, err)
|
||
}
|
||
|
||
// 4. 一次性数据库存在性校验
|
||
var idsToCheck []uint
|
||
for id := range allIDs {
|
||
idsToCheck = append(idsToCheck, id)
|
||
}
|
||
|
||
if len(idsToCheck) > 0 {
|
||
var count int64
|
||
if err := tx.Model(&models.Plan{}).Where("id IN ?", idsToCheck).Count(&count).Error; err != nil {
|
||
return fmt.Errorf("检查计划存在性时出错: %w", err)
|
||
}
|
||
if int(count) != len(idsToCheck) {
|
||
return ErrNodeDoesNotExist
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// validateNodeAndDetectCycles 递归地验证节点有效性并检测循环引用
|
||
func validateNodeAndDetectCycles(plan *models.Plan, allIDs, recursionStack map[uint]bool) error {
|
||
if plan == nil {
|
||
return nil
|
||
}
|
||
if plan.ID == 0 {
|
||
return ErrNewSubPlanInUpdate
|
||
}
|
||
if recursionStack[plan.ID] {
|
||
return fmt.Errorf("检测到循环引用：计划 (ID: %d) 是其自身的祖先", plan.ID)
|
||
}
|
||
if allIDs[plan.ID] {
|
||
return nil // 已经验证过这个节点及其子树，无需重复
|
||
}
|
||
|
||
recursionStack[plan.ID] = true
|
||
allIDs[plan.ID] = true
|
||
|
||
for _, subPlanLink := range plan.SubPlans {
|
||
if err := validateNodeAndDetectCycles(subPlanLink.ChildPlan, allIDs, recursionStack); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
|
||
delete(recursionStack, plan.ID) // 回溯，将节点移出当前递归路径
|
||
return nil
|
||
}
|
||
|
||
// reconcilePlanNode 递归地同步数据库状态以匹配给定的计划节点
|
||
func (r *gormPlanRepository) reconcilePlanNode(tx *gorm.DB, plan *models.Plan) error {
|
||
if plan == nil {
|
||
return nil
|
||
}
|
||
// 1. 更新节点本身的基础字段
|
||
if err := tx.Model(plan).Select("Name", "Description", "ExecutionType", "ExecuteNum", "CronExpression", "ContentType").Updates(plan).Error; err != nil {
|
||
return err
|
||
}
|
||
|
||
// 2. 根据内容类型协调子项
|
||
switch plan.ContentType {
|
||
case models.PlanContentTypeTasks:
|
||
// 清理旧的子计划关联
|
||
if err := tx.Where("parent_plan_id = ?", plan.ID).Delete(&models.SubPlan{}).Error; err != nil {
|
||
return fmt.Errorf("更新时清理旧的子计划关联失败: %w", err)
|
||
}
|
||
// 协调任务列表
|
||
return r.reconcileTasks(tx, plan)
|
||
case models.PlanContentTypeSubPlans:
|
||
// 清理旧的任务
|
||
if err := tx.Where("plan_id = ?", plan.ID).Delete(&models.Task{}).Error; err != nil {
|
||
return fmt.Errorf("更新时清理旧的任务失败: %w", err)
|
||
}
|
||
// 协调子计划关联
|
||
return r.reconcileSubPlans(tx, plan)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// reconcileTasks 精确同步任务列表
|
||
func (r *gormPlanRepository) reconcileTasks(tx *gorm.DB, plan *models.Plan) error {
|
||
var existingTasks []models.Task
|
||
if err := tx.Where("plan_id = ?", plan.ID).Find(&existingTasks).Error; err != nil {
|
||
return err
|
||
}
|
||
|
||
existingTaskMap := make(map[int]bool)
|
||
for _, task := range existingTasks {
|
||
existingTaskMap[task.ID] = true
|
||
}
|
||
|
||
for i := range plan.Tasks {
|
||
task := &plan.Tasks[i]
|
||
if task.ID == 0 {
|
||
task.PlanID = plan.ID
|
||
if err := tx.Create(task).Error; err != nil {
|
||
return err
|
||
}
|
||
} else {
|
||
delete(existingTaskMap, task.ID) // 从待删除map中移除
|
||
if err := tx.Model(task).Updates(task).Error; err != nil {
|
||
return err
|
||
}
|
||
}
|
||
}
|
||
|
||
var tasksToDelete []int
|
||
for id := range existingTaskMap {
|
||
tasksToDelete = append(tasksToDelete, id)
|
||
}
|
||
|
||
if len(tasksToDelete) > 0 {
|
||
if err := tx.Delete(&models.Task{}, tasksToDelete).Error; err != nil {
|
||
return err
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// reconcileSubPlans 精确同步子计划关联并递归
|
||
func (r *gormPlanRepository) reconcileSubPlans(tx *gorm.DB, plan *models.Plan) error {
|
||
var existingLinks []models.SubPlan
|
||
if err := tx.Where("parent_plan_id = ?", plan.ID).Find(&existingLinks).Error; err != nil {
|
||
return err
|
||
}
|
||
|
||
existingLinkMap := make(map[uint]bool)
|
||
for _, link := range existingLinks {
|
||
existingLinkMap[link.ID] = true
|
||
}
|
||
|
||
for i := range plan.SubPlans {
|
||
link := &plan.SubPlans[i]
|
||
link.ParentPlanID = plan.ID
|
||
if link.ID == 0 {
|
||
if err := tx.Create(link).Error; err != nil {
|
||
return err
|
||
}
|
||
} else {
|
||
delete(existingLinkMap, link.ID) // 从待删除map中移除
|
||
if err := tx.Model(link).Updates(link).Error; err != nil {
|
||
return err
|
||
}
|
||
}
|
||
}
|
||
|
||
var linksToDelete []uint
|
||
for id := range existingLinkMap {
|
||
linksToDelete = append(linksToDelete, id)
|
||
}
|
||
|
||
if len(linksToDelete) > 0 {
|
||
if err := tx.Delete(&models.SubPlan{}, linksToDelete).Error; err != nil {
|
||
return err
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// DeletePlan 根据ID删除计划，同时删除其关联的任务（非子任务）或子计划关联
|
||
func (r *gormPlanRepository) DeletePlan(id uint) error {
|
||
return r.db.Transaction(func(tx *gorm.DB) error {
|
||
// 1. 检查该计划是否是其他计划的子计划
|
||
var count int64
|
||
if err := tx.Model(&models.SubPlan{}).Where("child_plan_id = ?", id).Count(&count).Error; err != nil {
|
||
return fmt.Errorf("检查计划是否为子计划失败: %w", err)
|
||
}
|
||
if count > 0 {
|
||
return ErrDeleteWithReferencedPlan
|
||
}
|
||
|
||
var plan models.Plan
|
||
// 2. 获取计划以确定其内容类型
|
||
if err := tx.First(&plan, id).Error; err != nil {
|
||
return err
|
||
}
|
||
|
||
// 3. 根据内容类型删除关联数据
|
||
switch plan.ContentType {
|
||
case models.PlanContentTypeTasks:
|
||
// 删除与此计划关联的所有非子任务
|
||
if err := tx.Where("plan_id = ?", id).Delete(&models.Task{}).Error; err != nil {
|
||
return fmt.Errorf("删除计划ID %d 的任务失败: %w", id, err)
|
||
}
|
||
case models.PlanContentTypeSubPlans:
|
||
// 删除与此计划关联的所有子计划链接
|
||
if err := tx.Where("parent_plan_id = ?", id).Delete(&models.SubPlan{}).Error; err != nil {
|
||
return fmt.Errorf("删除计划ID %d 的子计划关联失败: %w", id, err)
|
||
}
|
||
}
|
||
|
||
// 4. 删除计划本身
|
||
if err := tx.Delete(&models.Plan{}, id).Error; err != nil {
|
||
return fmt.Errorf("删除计划ID %d 失败: %w", id, err)
|
||
}
|
||
|
||
return nil
|
||
})
|
||
}
|
||
|
||
// FlattenPlanTasks 递归展开计划，返回按执行顺序排列的所有任务列表
|
||
func (r *gormPlanRepository) FlattenPlanTasks(planID uint) ([]models.Task, error) {
|
||
plan, err := r.GetPlanByID(planID)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("获取计划(ID: %d)失败: %w", planID, err)
|
||
}
|
||
|
||
return r.flattenPlanTasksRecursive(plan)
|
||
}
|
||
|
||
// flattenPlanTasksRecursive 递归展开计划的内部实现
|
||
func (r *gormPlanRepository) flattenPlanTasksRecursive(plan *models.Plan) ([]models.Task, error) {
|
||
var tasks []models.Task
|
||
|
||
switch plan.ContentType {
|
||
case models.PlanContentTypeTasks:
|
||
// 如果计划直接包含任务，直接返回这些任务
|
||
// 由于GetPlanByID已经预加载并排序了任务，这里直接使用即可
|
||
tasks = append(tasks, plan.Tasks...)
|
||
|
||
case models.PlanContentTypeSubPlans:
|
||
// 如果计划包含子计划，则递归处理每个子计划
|
||
for _, subPlan := range plan.SubPlans {
|
||
// 获取子计划的任务列表
|
||
var subTasks []models.Task
|
||
var err error
|
||
|
||
// 确保子计划已经被加载
|
||
if subPlan.ChildPlan != nil {
|
||
subTasks, err = r.flattenPlanTasksRecursive(subPlan.ChildPlan)
|
||
} else {
|
||
// 如果子计划未加载，则从数据库获取并递归展开
|
||
subTasks, err = r.FlattenPlanTasks(subPlan.ChildPlanID)
|
||
}
|
||
|
||
if err != nil {
|
||
return nil, fmt.Errorf("展开子计划(ID: %d)失败: %w", subPlan.ChildPlanID, err)
|
||
}
|
||
|
||
// 将子计划的任务添加到结果中
|
||
tasks = append(tasks, subTasks...)
|
||
}
|
||
default:
|
||
return nil, fmt.Errorf("未知的计划内容类型: %v", plan.ContentType)
|
||
}
|
||
|
||
return tasks, nil
|
||
}
|
||
|
||
// DeleteTask 根据ID删除任务
|
||
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 {
|
||
return fmt.Errorf("检查待执行任务时出错: %w", err)
|
||
}
|
||
|
||
// 如果有待执行任务引用该任务，不能删除
|
||
if pendingTaskCount > 0 {
|
||
return fmt.Errorf("无法删除任务(ID: %d)，因为存在 %d 条待执行任务引用该任务", id, pendingTaskCount)
|
||
}
|
||
|
||
// 2. 检查是否有计划仍在使用这个任务
|
||
var planCount int64
|
||
if err := tx.Model(&models.Plan{}).Joins("JOIN tasks ON plans.id = tasks.plan_id").Where("tasks.id = ?", id).Count(&planCount).Error; err != nil {
|
||
return fmt.Errorf("检查计划引用任务时出错: %w", err)
|
||
}
|
||
|
||
// 如果有计划在使用该任务，不能删除
|
||
if planCount > 0 {
|
||
return fmt.Errorf("无法删除任务(ID: %d)，因为存在 %d 个计划仍在使用该任务", id, planCount)
|
||
}
|
||
|
||
// 3. 执行删除操作
|
||
result := tx.Delete(&models.Task{}, id)
|
||
if result.Error != nil {
|
||
return fmt.Errorf("删除任务失败: %w", result.Error)
|
||
}
|
||
|
||
// 检查是否实际删除了记录
|
||
if result.RowsAffected == 0 {
|
||
return gorm.ErrRecordNotFound
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
// FindPlanAnalysisTaskByParamsPlanID 根据Parameters中的ParamsPlanID字段值查找TaskPlanAnalysis类型的Task
|
||
func (r *gormPlanRepository) FindPlanAnalysisTaskByParamsPlanID(paramsPlanID uint) (*models.Task, error) {
|
||
return r.findPlanAnalysisTask(r.db, paramsPlanID)
|
||
}
|
||
|
||
// createPlanAnalysisTask 用于创建一个TaskPlanAnalysis类型的Task
|
||
// 关键修改：Task.PlanID 设置为 0，实际 PlanID 存储在 Parameters 中，并返回创建的 Task
|
||
func (r *gormPlanRepository) createPlanAnalysisTask(tx *gorm.DB, plan *models.Plan) (*models.Task, error) {
|
||
m := map[string]interface{}{
|
||
models.ParamsPlanID: plan.ID,
|
||
}
|
||
parameters, err := json.Marshal(m)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
task := &models.Task{
|
||
PlanID: 0, // 关键：设置为 0，避免被常规 PlanID 查询查到
|
||
Name: fmt.Sprintf("'%s'计划触发器", plan.Name),
|
||
Description: fmt.Sprintf("计划名: %s, 计划ID: %d", plan.Name, plan.ID),
|
||
ExecutionOrder: 0, // 触发器任务的执行顺序通常为0或不关心
|
||
Type: models.TaskPlanAnalysis,
|
||
Parameters: datatypes.JSON(parameters),
|
||
}
|
||
|
||
if err := tx.Create(task).Error; err != nil {
|
||
return nil, err
|
||
}
|
||
return task, nil
|
||
}
|
||
|
||
// updatePlanAnalysisTask 使用更安全的方式更新触发器任务
|
||
func (r *gormPlanRepository) updatePlanAnalysisTask(tx *gorm.DB, plan *models.Plan) error {
|
||
task, err := r.findPlanAnalysisTask(tx, plan.ID)
|
||
if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
|
||
return fmt.Errorf("查找现有计划分析任务失败: %w", err)
|
||
}
|
||
|
||
// 如果触发器任务不存在，则创建一个
|
||
if task == nil {
|
||
_, err := r.createPlanAnalysisTask(tx, plan)
|
||
return err
|
||
}
|
||
|
||
// 如果存在，则更新它的名称和描述以反映计划的最新信息
|
||
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 是一个内部使用的、更高效的查找方法
|
||
// 关键修改：通过查询 parameters JSON 字段来查找
|
||
func (r *gormPlanRepository) findPlanAnalysisTask(tx *gorm.DB, planID uint) (*models.Task, error) {
|
||
var task models.Task
|
||
err := tx.Where("type = ? AND parameters->>'plan_id' = ?", models.TaskPlanAnalysis, fmt.Sprintf("%d", planID)).First(&task).Error
|
||
if errors.Is(err, gorm.ErrRecordNotFound) {
|
||
return nil, nil // 未找到不是错误，返回nil, nil
|
||
}
|
||
return &task, err
|
||
}
|
||
|
||
// FindPlanAnalysisTaskByPlanID 是暴露给外部的公共方法
|
||
// 关键修改：通过查询 parameters JSON 字段来查找
|
||
func (r *gormPlanRepository) FindPlanAnalysisTaskByPlanID(planID uint) (*models.Task, error) {
|
||
return r.findPlanAnalysisTask(r.db, planID)
|
||
}
|
||
|
||
// CreatePlanAnalysisTask 创建一个 plan_analysis 类型的任务并返回它
|
||
// 这个方法是公开的，主要由 TaskManager 在发现触发器任务定义丢失时调用。
|
||
func (r *gormPlanRepository) CreatePlanAnalysisTask(plan *models.Plan) (*models.Task, error) {
|
||
var createdTask *models.Task
|
||
err := r.db.Transaction(func(tx *gorm.DB) error {
|
||
var err error
|
||
createdTask, err = r.createPlanAnalysisTask(tx, plan)
|
||
return err
|
||
})
|
||
return createdTask, err
|
||
}
|
||
|
||
// FindPlansWithPendingTasks 查找所有正在执行的计划
|
||
func (r *gormPlanRepository) FindPlansWithPendingTasks() ([]*models.Plan, error) {
|
||
var plans []*models.Plan
|
||
|
||
// 关联 pending_tasks, task_execution_logs, tasks 表来查找符合条件的计划
|
||
err := r.db.Table("plans").
|
||
Joins("JOIN tasks ON plans.id = tasks.plan_id").
|
||
Joins("JOIN task_execution_logs ON tasks.id = task_execution_logs.task_id").
|
||
Joins("JOIN pending_tasks ON task_execution_logs.id = pending_tasks.task_execution_log_id").
|
||
Group("plans.id"). // 避免重复，因为一个计划可能有多个待执行任务
|
||
Find(&plans).Error
|
||
|
||
return plans, err
|
||
}
|
||
|
||
// DB 返回底层的数据库连接实例
|
||
func (r *gormPlanRepository) DB() *gorm.DB {
|
||
return r.db
|
||
}
|
||
|
||
// StopPlanTransactionally 停止一个计划的执行，包括更新状态、移除待执行任务和更新执行日志。
|
||
func (r *gormPlanRepository) StopPlanTransactionally(planID uint) error {
|
||
return r.db.Transaction(func(tx *gorm.DB) error {
|
||
// 使用事务创建新的仓库实例，确保所有操作都在同一个事务中
|
||
planRepoTx := NewGormPlanRepository(tx)
|
||
executionLogRepoTx := NewGormExecutionLogRepository(tx)
|
||
pendingTaskRepoTx := NewGormPendingTaskRepository(tx)
|
||
|
||
// 1. 更新计划状态为“已停止”
|
||
if err := planRepoTx.UpdatePlanStatus(planID, models.PlanStatusDisabled); err != nil {
|
||
return fmt.Errorf("更新计划 #%d 状态为 '已停止' 失败: %w", planID, err)
|
||
}
|
||
|
||
// 2. 查找当前正在进行的计划执行日志
|
||
planLog, err := executionLogRepoTx.FindInProgressPlanExecutionLogByPlanID(planID)
|
||
if err != nil {
|
||
return fmt.Errorf("查找计划 #%d 正在进行的执行日志失败: %w", planID, err)
|
||
}
|
||
|
||
if planLog == nil {
|
||
// 没有正在进行的执行，直接返回成功
|
||
return nil
|
||
}
|
||
|
||
// 3. 查找所有需要被取消的任务执行日志
|
||
taskLogs, err := executionLogRepoTx.FindIncompleteTaskExecutionLogsByPlanLogID(planLog.ID)
|
||
if err != nil {
|
||
return fmt.Errorf("查找计划执行日志 #%d 下未完成的任务日志失败: %w", planLog.ID, err)
|
||
}
|
||
|
||
if len(taskLogs) > 0 {
|
||
var taskLogIDs []uint
|
||
for _, tl := range taskLogs {
|
||
taskLogIDs = append(taskLogIDs, tl.ID)
|
||
}
|
||
|
||
// 3.1 批量更新任务执行日志状态为“已取消”
|
||
if err := executionLogRepoTx.UpdateLogStatusByIDs(taskLogIDs, models.ExecutionStatusCancelled); err != nil {
|
||
return fmt.Errorf("批量更新任务执行日志状态为 '已取消' 失败: %w", err)
|
||
}
|
||
|
||
// 3.2 查找并删除待执行队列中对应的任务
|
||
pendingTasks, err := pendingTaskRepoTx.FindPendingTasksByTaskLogIDs(taskLogIDs)
|
||
if err != nil {
|
||
return fmt.Errorf("查找计划执行日志 #%d 下对应的待执行任务失败: %w", planLog.ID, err)
|
||
}
|
||
|
||
if len(pendingTasks) > 0 {
|
||
var pendingTaskIDs []uint
|
||
for _, pt := range pendingTasks {
|
||
pendingTaskIDs = append(pendingTaskIDs, pt.ID)
|
||
}
|
||
if err := pendingTaskRepoTx.DeletePendingTasksByIDs(pendingTaskIDs); err != nil {
|
||
return fmt.Errorf("批量删除待执行任务失败: %w", err)
|
||
}
|
||
}
|
||
}
|
||
|
||
// 4. 更新计划执行历史的总状态为“失败”
|
||
if err := executionLogRepoTx.UpdatePlanExecutionLogStatus(planLog.ID, models.ExecutionStatusFailed); err != nil {
|
||
return fmt.Errorf("更新计划执行日志 #%d 状态为 '失败' 失败: %w", planLog.ID, err)
|
||
}
|
||
|
||
return nil
|
||
})
|
||
}
|
||
|
||
// UpdatePlanStatus 更新指定计划的状态
|
||
func (r *gormPlanRepository) UpdatePlanStatus(id uint, status models.PlanStatus) error {
|
||
result := r.db.Model(&models.Plan{}).Where("id = ?", id).Update("status", status)
|
||
if result.Error != nil {
|
||
return result.Error
|
||
}
|
||
if result.RowsAffected == 0 {
|
||
return gorm.ErrRecordNotFound
|
||
}
|
||
return nil
|
||
}
|