diff --git a/internal/certificate/service.go b/internal/certificate/service.go index b8f5fa89..406e183c 100644 --- a/internal/certificate/service.go +++ b/internal/certificate/service.go @@ -30,18 +30,18 @@ type certificateRepository interface { } type CertificateService struct { - repo certificateRepository + certRepo certificateRepository } -func NewCertificateService(repo certificateRepository) *CertificateService { +func NewCertificateService(certRepo certificateRepository) *CertificateService { return &CertificateService{ - repo: repo, + certRepo: certRepo, } } func (s *CertificateService) InitSchedule(ctx context.Context) error { app.GetScheduler().MustAdd("certificateExpireSoonNotify", "0 0 * * *", func() { - certs, err := s.repo.ListExpireSoon(context.Background()) + certs, err := s.certRepo.ListExpireSoon(context.Background()) if err != nil { app.GetLogger().Error("failed to get certificates which expire soon", "err", err) return @@ -60,7 +60,7 @@ func (s *CertificateService) InitSchedule(ctx context.Context) error { } func (s *CertificateService) ArchiveFile(ctx context.Context, req *dtos.CertificateArchiveFileReq) ([]byte, error) { - certificate, err := s.repo.GetById(ctx, req.CertificateId) + certificate, err := s.certRepo.GetById(ctx, req.CertificateId) if err != nil { return nil, err } diff --git a/internal/domain/dtos/workflow.go b/internal/domain/dtos/workflow.go index 9d1f5781..3988220b 100644 --- a/internal/domain/dtos/workflow.go +++ b/internal/domain/dtos/workflow.go @@ -4,7 +4,7 @@ import "github.com/usual2970/certimate/internal/domain" type WorkflowStartRunReq struct { WorkflowId string `json:"-"` - Trigger domain.WorkflowTriggerType `json:"trigger"` + RunTrigger domain.WorkflowTriggerType `json:"trigger"` } type WorkflowCancelRunReq struct { diff --git a/internal/repository/workflow.go b/internal/repository/workflow.go index 738e898e..60d60899 100644 --- a/internal/repository/workflow.go +++ b/internal/repository/workflow.go @@ -95,65 +95,6 @@ func (r *WorkflowRepository) Save(ctx context.Context, workflow *domain.Workflow return workflow, nil } -func (r *WorkflowRepository) SaveRun(ctx context.Context, run *domain.WorkflowRun) (*domain.WorkflowRun, error) { - collection, err := app.GetApp().FindCollectionByNameOrId(domain.CollectionNameWorkflowRun) - if err != nil { - return run, err - } - - var runRecord *core.Record - if run.Id == "" { - runRecord = core.NewRecord(collection) - } else { - runRecord, err = app.GetApp().FindRecordById(collection, run.Id) - if err != nil { - if errors.Is(err, sql.ErrNoRows) { - return run, err - } - runRecord = core.NewRecord(collection) - } - } - - err = app.GetApp().RunInTransaction(func(txApp core.App) error { - runRecord.Set("workflowId", run.WorkflowId) - runRecord.Set("trigger", string(run.Trigger)) - runRecord.Set("status", string(run.Status)) - runRecord.Set("startedAt", run.StartedAt) - runRecord.Set("endedAt", run.EndedAt) - runRecord.Set("logs", run.Logs) - runRecord.Set("error", run.Error) - err = txApp.Save(runRecord) - if err != nil { - return err - } - - workflowRecord, err := txApp.FindRecordById(domain.CollectionNameWorkflow, run.WorkflowId) - if err != nil { - return err - } - - workflowRecord.IgnoreUnchangedFields(true) - workflowRecord.Set("lastRunId", runRecord.Id) - workflowRecord.Set("lastRunStatus", runRecord.GetString("status")) - workflowRecord.Set("lastRunTime", runRecord.GetString("startedAt")) - err = txApp.Save(workflowRecord) - if err != nil { - return err - } - - run.Id = runRecord.Id - run.CreatedAt = runRecord.GetDateTime("created").Time() - run.UpdatedAt = runRecord.GetDateTime("updated").Time() - - return nil - }) - if err != nil { - return run, err - } - - return run, nil -} - func (r *WorkflowRepository) castRecordToModel(record *core.Record) (*domain.Workflow, error) { if record == nil { return nil, fmt.Errorf("record is nil") diff --git a/internal/repository/workflow_output.go b/internal/repository/workflow_output.go index 1adf0c8c..e75b2cb7 100644 --- a/internal/repository/workflow_output.go +++ b/internal/repository/workflow_output.go @@ -94,12 +94,12 @@ func (r *WorkflowOutputRepository) castRecordToModel(record *core.Record) (*doma node := &domain.WorkflowNode{} if err := record.UnmarshalJSONField("node", node); err != nil { - return nil, errors.New("failed to unmarshal node") + return nil, err } outputs := make([]domain.WorkflowNodeIO, 0) if err := record.UnmarshalJSONField("outputs", &outputs); err != nil { - return nil, errors.New("failed to unmarshal output") + return nil, err } workflowOutput := &domain.WorkflowOutput{ @@ -118,27 +118,27 @@ func (r *WorkflowOutputRepository) castRecordToModel(record *core.Record) (*doma return workflowOutput, nil } -func (r *WorkflowOutputRepository) saveRecord(output *domain.WorkflowOutput) (*core.Record, error) { +func (r *WorkflowOutputRepository) saveRecord(workflowOutput *domain.WorkflowOutput) (*core.Record, error) { collection, err := app.GetApp().FindCollectionByNameOrId(domain.CollectionNameWorkflowOutput) if err != nil { return nil, err } var record *core.Record - if output.Id == "" { + if workflowOutput.Id == "" { record = core.NewRecord(collection) } else { - record, err = app.GetApp().FindRecordById(collection, output.Id) + record, err = app.GetApp().FindRecordById(collection, workflowOutput.Id) if err != nil { return record, err } } - record.Set("workflowId", output.WorkflowId) - record.Set("runId", output.RunId) - record.Set("nodeId", output.NodeId) - record.Set("node", output.Node) - record.Set("outputs", output.Outputs) - record.Set("succeeded", output.Succeeded) + record.Set("workflowId", workflowOutput.WorkflowId) + record.Set("runId", workflowOutput.RunId) + record.Set("nodeId", workflowOutput.NodeId) + record.Set("node", workflowOutput.Node) + record.Set("outputs", workflowOutput.Outputs) + record.Set("succeeded", workflowOutput.Succeeded) if err := app.GetApp().Save(record); err != nil { return record, err } diff --git a/internal/repository/workflow_run.go b/internal/repository/workflow_run.go new file mode 100644 index 00000000..01185a45 --- /dev/null +++ b/internal/repository/workflow_run.go @@ -0,0 +1,117 @@ +package repository + +import ( + "context" + "database/sql" + "errors" + "fmt" + + "github.com/pocketbase/pocketbase/core" + "github.com/usual2970/certimate/internal/app" + "github.com/usual2970/certimate/internal/domain" +) + +type WorkflowRunRepository struct{} + +func NewWorkflowRunRepository() *WorkflowRunRepository { + return &WorkflowRunRepository{} +} + +func (r *WorkflowRunRepository) GetById(ctx context.Context, id string) (*domain.WorkflowRun, error) { + record, err := app.GetApp().FindRecordById(domain.CollectionNameWorkflowRun, id) + if err != nil { + if errors.Is(err, sql.ErrNoRows) { + return nil, domain.ErrRecordNotFound + } + return nil, err + } + + return r.castRecordToModel(record) +} + +func (r *WorkflowRunRepository) Save(ctx context.Context, workflowRun *domain.WorkflowRun) (*domain.WorkflowRun, error) { + collection, err := app.GetApp().FindCollectionByNameOrId(domain.CollectionNameWorkflowRun) + if err != nil { + return workflowRun, err + } + + var record *core.Record + if workflowRun.Id == "" { + record = core.NewRecord(collection) + } else { + record, err = app.GetApp().FindRecordById(collection, workflowRun.Id) + if err != nil { + if errors.Is(err, sql.ErrNoRows) { + return workflowRun, err + } + record = core.NewRecord(collection) + } + } + + err = app.GetApp().RunInTransaction(func(txApp core.App) error { + record.Set("workflowId", workflowRun.WorkflowId) + record.Set("trigger", string(workflowRun.Trigger)) + record.Set("status", string(workflowRun.Status)) + record.Set("startedAt", workflowRun.StartedAt) + record.Set("endedAt", workflowRun.EndedAt) + record.Set("logs", workflowRun.Logs) + record.Set("error", workflowRun.Error) + err = txApp.Save(record) + if err != nil { + return err + } + + workflowRun.Id = record.Id + workflowRun.CreatedAt = record.GetDateTime("created").Time() + workflowRun.UpdatedAt = record.GetDateTime("updated").Time() + + // 事务级联更新所属工作流的最后运行记录 + workflowRecord, err := txApp.FindRecordById(domain.CollectionNameWorkflow, workflowRun.WorkflowId) + if err != nil { + return err + } else if workflowRecord.GetDateTime("lastRunTime").Time().IsZero() || workflowRun.StartedAt.After(workflowRecord.GetDateTime("lastRunTime").Time()) { + workflowRecord.IgnoreUnchangedFields(true) + workflowRecord.Set("lastRunId", record.Id) + workflowRecord.Set("lastRunStatus", record.GetString("status")) + workflowRecord.Set("lastRunTime", record.GetString("startedAt")) + err = txApp.Save(workflowRecord) + if err != nil { + return err + } + } + + return nil + }) + if err != nil { + return workflowRun, err + } + + return workflowRun, nil +} + +func (r *WorkflowRunRepository) castRecordToModel(record *core.Record) (*domain.WorkflowRun, error) { + if record == nil { + return nil, fmt.Errorf("record is nil") + } + + logs := make([]domain.WorkflowRunLog, 0) + if err := record.UnmarshalJSONField("logs", &logs); err != nil { + return nil, err + } + + workflowRun := &domain.WorkflowRun{ + Meta: domain.Meta{ + Id: record.Id, + CreatedAt: record.GetDateTime("created").Time(), + UpdatedAt: record.GetDateTime("updated").Time(), + }, + WorkflowId: record.GetString("workflowId"), + Status: domain.WorkflowRunStatusType(record.GetString("status")), + Trigger: domain.WorkflowTriggerType(record.GetString("trigger")), + StartedAt: record.GetDateTime("startedAt").Time(), + EndedAt: record.GetDateTime("endedAt").Time(), + Logs: logs, + Error: record.GetString("error"), + } + return workflowRun, nil +} diff --git a/internal/rest/routes/routes.go b/internal/rest/routes/routes.go index 58b4f0f1..756760da 100644 --- a/internal/rest/routes/routes.go +++ b/internal/rest/routes/routes.go @@ -27,13 +27,14 @@ func Register(router *router.Router[*core.RequestEvent]) { certificateSvc = certificate.NewCertificateService(certificateRepo) workflowRepo := repository.NewWorkflowRepository() - workflowSvc = workflow.NewWorkflowService(workflowRepo) + workflowRunRepo := repository.NewWorkflowRunRepository() + workflowSvc = workflow.NewWorkflowService(workflowRepo, workflowRunRepo) statisticsRepo := repository.NewStatisticsRepository() statisticsSvc = statistics.NewStatisticsService(statisticsRepo) - notifyRepo := repository.NewSettingsRepository() - notifySvc = notify.NewNotifyService(notifyRepo) + settingsRepo := repository.NewSettingsRepository() + notifySvc = notify.NewNotifyService(settingsRepo) group := router.Group("/api") group.Bind(apis.RequireSuperuserAuth()) diff --git a/internal/scheduler/scheduler.go b/internal/scheduler/scheduler.go index 055beddb..f5029599 100644 --- a/internal/scheduler/scheduler.go +++ b/internal/scheduler/scheduler.go @@ -8,7 +8,8 @@ import ( func Register() { workflowRepo := repository.NewWorkflowRepository() - workflowSvc := workflow.NewWorkflowService(workflowRepo) + workflowRunRepo := repository.NewWorkflowRunRepository() + workflowSvc := workflow.NewWorkflowService(workflowRepo, workflowRunRepo) certificateRepo := repository.NewCertificateRepository() certificateSvc := certificate.NewCertificateService(certificateRepo) diff --git a/internal/statistics/service.go b/internal/statistics/service.go index 3b1f5876..44388ba9 100644 --- a/internal/statistics/service.go +++ b/internal/statistics/service.go @@ -11,15 +11,15 @@ type statisticsRepository interface { } type StatisticsService struct { - repo statisticsRepository + statRepo statisticsRepository } -func NewStatisticsService(repo statisticsRepository) *StatisticsService { +func NewStatisticsService(statRepo statisticsRepository) *StatisticsService { return &StatisticsService{ - repo: repo, + statRepo: statRepo, } } func (s *StatisticsService) Get(ctx context.Context) (*domain.Statistics, error) { - return s.repo.Get(ctx) + return s.statRepo.Get(ctx) } diff --git a/internal/workflow/event.go b/internal/workflow/event.go index f8117dbc..fa6b4b1a 100644 --- a/internal/workflow/event.go +++ b/internal/workflow/event.go @@ -65,9 +65,10 @@ func onWorkflowRecordCreateOrUpdate(ctx context.Context, record *core.Record) er // 反之，重新添加定时任务 err := scheduler.Add(fmt.Sprintf("workflow#%s", workflowId), record.GetString("triggerCron"), func() { - NewWorkflowService(repository.NewWorkflowRepository()).StartRun(ctx, &dtos.WorkflowStartRunReq{ + workflowSrv := NewWorkflowService(repository.NewWorkflowRepository(), repository.NewWorkflowRunRepository()) + workflowSrv.StartRun(ctx, &dtos.WorkflowStartRunReq{ WorkflowId: workflowId, - Trigger: domain.WorkflowTriggerTypeAuto, + RunTrigger: domain.WorkflowTriggerTypeAuto, }) }) if err != nil { diff --git a/internal/workflow/node-processor/apply_node.go b/internal/workflow/node-processor/apply_node.go index 6b509d73..39a0167f 100644 --- a/internal/workflow/node-processor/apply_node.go +++ b/internal/workflow/node-processor/apply_node.go @@ -31,7 +31,7 @@ func NewApplyNode(node *domain.WorkflowNode) *applyNode { } } -func (n *applyNode) Run(ctx context.Context) error { +func (n *applyNode) Process(ctx context.Context) error { n.AddOutput(ctx, n.node.Name, "开始执行") // 查询上次执行结果 diff --git a/internal/workflow/node-processor/condition_node.go b/internal/workflow/node-processor/condition_node.go index a511ce20..994965ba 100644 --- a/internal/workflow/node-processor/condition_node.go +++ b/internal/workflow/node-processor/condition_node.go @@ -18,7 +18,7 @@ func NewConditionNode(node *domain.WorkflowNode) *conditionNode { } } -func (n *conditionNode) Run(ctx context.Context) error { +func (n *conditionNode) Process(ctx context.Context) error { // 此类型节点不需要执行任何操作，直接返回 n.AddOutput(ctx, n.node.Name, "完成") diff --git a/internal/workflow/node-processor/deploy_node.go b/internal/workflow/node-processor/deploy_node.go index 72665419..dccdf0e8 100644 --- a/internal/workflow/node-processor/deploy_node.go +++ b/internal/workflow/node-processor/deploy_node.go @@ -29,7 +29,7 @@ func NewDeployNode(node *domain.WorkflowNode) *deployNode { } } -func (n *deployNode) Run(ctx context.Context) error { +func (n *deployNode) Process(ctx context.Context) error { n.AddOutput(ctx, n.node.Name, "开始执行") // 查询上次执行结果 diff --git a/internal/workflow/node-processor/execute_failure_node.go b/internal/workflow/node-processor/execute_failure_node.go index 84042a4b..a64019bb 100644 --- a/internal/workflow/node-processor/execute_failure_node.go +++ b/internal/workflow/node-processor/execute_failure_node.go @@ -18,7 +18,7 @@ func NewExecuteFailureNode(node *domain.WorkflowNode) *executeFailureNode { } } -func (n *executeFailureNode) Run(ctx context.Context) error { +func (n *executeFailureNode) Process(ctx context.Context) error { // 此类型节点不需要执行任何操作，直接返回 n.AddOutput(ctx, n.node.Name, "进入执行失败分支") diff --git a/internal/workflow/node-processor/execute_success_node.go b/internal/workflow/node-processor/execute_success_node.go index ef058b06..e0cfea1e 100644 --- a/internal/workflow/node-processor/execute_success_node.go +++ b/internal/workflow/node-processor/execute_success_node.go @@ -18,7 +18,7 @@ func NewExecuteSuccessNode(node *domain.WorkflowNode) *executeSuccessNode { } } -func (n *executeSuccessNode) Run(ctx context.Context) error { +func (n *executeSuccessNode) Process(ctx context.Context) error { // 此类型节点不需要执行任何操作，直接返回 n.AddOutput(ctx, n.node.Name, "进入执行成功分支") diff --git a/internal/workflow/node-processor/notify_node.go b/internal/workflow/node-processor/notify_node.go index 052ebda7..21e6ac15 100644 --- a/internal/workflow/node-processor/notify_node.go +++ b/internal/workflow/node-processor/notify_node.go @@ -24,7 +24,7 @@ func NewNotifyNode(node *domain.WorkflowNode) *notifyNode { } } -func (n *notifyNode) Run(ctx context.Context) error { +func (n *notifyNode) Process(ctx context.Context) error { n.AddOutput(ctx, n.node.Name, "开始执行") nodeConfig := n.node.GetConfigForNotify() diff --git a/internal/workflow/node-processor/processor.go b/internal/workflow/node-processor/processor.go index 33e82e3b..61155892 100644 --- a/internal/workflow/node-processor/processor.go +++ b/internal/workflow/node-processor/processor.go @@ -9,7 +9,7 @@ import ( ) type NodeProcessor interface { - Run(ctx context.Context) error + Process(ctx context.Context) error GetLog(ctx context.Context) *domain.WorkflowRunLog AddOutput(ctx context.Context, title, content string, err ...string) } diff --git a/internal/workflow/node-processor/start_node.go b/internal/workflow/node-processor/start_node.go index 2f1026ad..99e15af2 100644 --- a/internal/workflow/node-processor/start_node.go +++ b/internal/workflow/node-processor/start_node.go @@ -18,7 +18,7 @@ func NewStartNode(node *domain.WorkflowNode) *startNode { } } -func (n *startNode) Run(ctx context.Context) error { +func (n *startNode) Process(ctx context.Context) error { // 此类型节点不需要执行任何操作，直接返回 n.AddOutput(ctx, n.node.Name, "完成") diff --git a/internal/workflow/node-processor/upload_node.go b/internal/workflow/node-processor/upload_node.go index 09d86a5d..9e316501 100644 --- a/internal/workflow/node-processor/upload_node.go +++ b/internal/workflow/node-processor/upload_node.go @@ -29,9 +29,7 @@ func NewUploadNode(node *domain.WorkflowNode) *uploadNode { } } -// Run 上传证书节点执行 -// 包含上传证书的工作流，理论上应该手动执行，如果每天定时执行，也只是重新保存一下 -func (n *uploadNode) Run(ctx context.Context) error { +func (n *uploadNode) Process(ctx context.Context) error { n.AddOutput(ctx, n.node.Name, "进入上传证书节点") nodeConfig := n.node.GetConfigForUpload() diff --git a/internal/workflow/processor/processor.go b/internal/workflow/processor/processor.go index 0923a8b2..39486419 100644 --- a/internal/workflow/processor/processor.go +++ b/internal/workflow/processor/processor.go @@ -8,27 +8,29 @@ import ( ) type workflowProcessor struct { - workflow *domain.Workflow - workflowRun *domain.WorkflowRun - workflorRunLogs []domain.WorkflowRunLog + workflowId string + workflowContent *domain.WorkflowNode + runId string + runLogs []domain.WorkflowRunLog } -func NewWorkflowProcessor(workflow *domain.Workflow, workflowRun *domain.WorkflowRun) *workflowProcessor { +func NewWorkflowProcessor(workflowId string, workflowContent *domain.WorkflowNode, workflowRunId string) *workflowProcessor { return &workflowProcessor{ - workflow: workflow, - workflowRun: workflowRun, - workflorRunLogs: make([]domain.WorkflowRunLog, 0), + workflowId: workflowId, + workflowContent: workflowContent, + runId: workflowRunId, + runLogs: make([]domain.WorkflowRunLog, 0), } } -func (w *workflowProcessor) Run(ctx context.Context) error { - ctx = context.WithValue(ctx, "workflow_id", w.workflow.Id) - ctx = context.WithValue(ctx, "workflow_run_id", w.workflowRun.Id) - return w.processNode(ctx, w.workflow.Content) +func (w *workflowProcessor) Process(ctx context.Context) error { + ctx = context.WithValue(ctx, "workflow_id", w.workflowId) + ctx = context.WithValue(ctx, "workflow_run_id", w.runId) + return w.processNode(ctx, w.workflowContent) } -func (w *workflowProcessor) GetRunLogs() []domain.WorkflowRunLog { - return w.workflorRunLogs +func (w *workflowProcessor) GetLogs() []domain.WorkflowRunLog { + return w.runLogs } func (w *workflowProcessor) processNode(ctx context.Context, node *domain.WorkflowNode) error { @@ -51,10 +53,10 @@ func (w *workflowProcessor) processNode(ctx context.Context, node *domain.Workfl break } - runErr = processor.Run(ctx) + runErr = processor.Process(ctx) log := processor.GetLog(ctx) if log != nil { - w.workflorRunLogs = append(w.workflorRunLogs, *log) + w.runLogs = append(w.runLogs, *log) } if runErr != nil { break @@ -67,9 +69,9 @@ func (w *workflowProcessor) processNode(ctx context.Context, node *domain.Workfl if runErr != nil && current.Next != nil && current.Next.Type != domain.WorkflowNodeTypeExecuteResultBranch { return runErr } else if runErr != nil && current.Next != nil && current.Next.Type == domain.WorkflowNodeTypeExecuteResultBranch { - current = getBranchByType(current.Next.Branches, domain.WorkflowNodeTypeExecuteFailure) + current = w.getBranchByType(current.Next.Branches, domain.WorkflowNodeTypeExecuteFailure) } else if runErr == nil && current.Next != nil && current.Next.Type == domain.WorkflowNodeTypeExecuteResultBranch { - current = getBranchByType(current.Next.Branches, domain.WorkflowNodeTypeExecuteSuccess) + current = w.getBranchByType(current.Next.Branches, domain.WorkflowNodeTypeExecuteSuccess) } else { current = current.Next } @@ -78,7 +80,7 @@ func (w *workflowProcessor) processNode(ctx context.Context, node *domain.Workfl return nil } -func getBranchByType(branches []domain.WorkflowNode, nodeType domain.WorkflowNodeType) *domain.WorkflowNode { +func (w *workflowProcessor) getBranchByType(branches []domain.WorkflowNode, nodeType domain.WorkflowNodeType) *domain.WorkflowNode { for _, branch := range branches { if branch.Type == nodeType { return &branch diff --git a/internal/workflow/service.go b/internal/workflow/service.go index c3b4a3e0..8181c969 100644 --- a/internal/workflow/service.go +++ b/internal/workflow/service.go @@ -13,46 +13,55 @@ import ( processor "github.com/usual2970/certimate/internal/workflow/processor" ) -const defaultRoutines = 10 +const defaultRoutines = 16 type workflowRunData struct { - Workflow *domain.Workflow - RunTrigger domain.WorkflowTriggerType + WorkflowId string + WorkflowContent *domain.WorkflowNode + RunTrigger domain.WorkflowTriggerType } type workflowRepository interface { ListEnabledAuto(ctx context.Context) ([]*domain.Workflow, error) GetById(ctx context.Context, id string) (*domain.Workflow, error) Save(ctx context.Context, workflow *domain.Workflow) (*domain.Workflow, error) - SaveRun(ctx context.Context, workflowRun *domain.WorkflowRun) (*domain.WorkflowRun, error) +} + +type workflowRunRepository interface { + GetById(ctx context.Context, id string) (*domain.WorkflowRun, error) + Save(ctx context.Context, workflowRun *domain.WorkflowRun) (*domain.WorkflowRun, error) } type WorkflowService struct { ch chan *workflowRunData - repo workflowRepository wg sync.WaitGroup cancel context.CancelFunc + + workflowRepo workflowRepository + workflowRunRepo workflowRunRepository } -func NewWorkflowService(repo workflowRepository) *WorkflowService { - srv := &WorkflowService{ - repo: repo, - ch: make(chan *workflowRunData, 1), - } - +func NewWorkflowService(workflowRepo workflowRepository, workflowRunRepo workflowRunRepository) *WorkflowService { ctx, cancel := context.WithCancel(context.Background()) - srv.cancel = cancel + + srv := &WorkflowService{ + ch: make(chan *workflowRunData, 1), + cancel: cancel, + + workflowRepo: workflowRepo, + workflowRunRepo: workflowRunRepo, + } srv.wg.Add(defaultRoutines) for i := 0; i < defaultRoutines; i++ { - go srv.run(ctx) + go srv.startRun(ctx) } return srv } func (s *WorkflowService) InitSchedule(ctx context.Context) error { - workflows, err := s.repo.ListEnabledAuto(ctx) + workflows, err := s.workflowRepo.ListEnabledAuto(ctx) if err != nil { return err } @@ -62,7 +71,7 @@ func (s *WorkflowService) InitSchedule(ctx context.Context) error { err := scheduler.Add(fmt.Sprintf("workflow#%s", workflow.Id), workflow.TriggerCron, func() { s.StartRun(ctx, &dtos.WorkflowStartRunReq{ WorkflowId: workflow.Id, - Trigger: domain.WorkflowTriggerTypeAuto, + RunTrigger: domain.WorkflowTriggerTypeAuto, }) }) if err != nil { @@ -75,35 +84,50 @@ func (s *WorkflowService) InitSchedule(ctx context.Context) error { } func (s *WorkflowService) StartRun(ctx context.Context, req *dtos.WorkflowStartRunReq) error { - workflow, err := s.repo.GetById(ctx, req.WorkflowId) + workflow, err := s.workflowRepo.GetById(ctx, req.WorkflowId) if err != nil { app.GetLogger().Error("failed to get workflow", "id", req.WorkflowId, "err", err) return err } - if workflow.LastRunStatus == domain.WorkflowRunStatusTypeRunning { - return errors.New("workflow is running") - } - - workflow.LastRunTime = time.Now() - workflow.LastRunStatus = domain.WorkflowRunStatusTypePending - workflow.LastRunId = "" - if resp, err := s.repo.Save(ctx, workflow); err != nil { - return err - } else { - workflow = resp + if workflow.LastRunStatus == domain.WorkflowRunStatusTypePending || workflow.LastRunStatus == domain.WorkflowRunStatusTypeRunning { + return errors.New("workflow is already pending or running") } s.ch <- &workflowRunData{ - Workflow: workflow, - RunTrigger: req.Trigger, + WorkflowId: workflow.Id, + WorkflowContent: workflow.Content, + RunTrigger: req.RunTrigger, } return nil } func (s *WorkflowService) CancelRun(ctx context.Context, req *dtos.WorkflowCancelRunReq) error { + workflow, err := s.workflowRepo.GetById(ctx, req.WorkflowId) + if err != nil { + app.GetLogger().Error("failed to get workflow", "id", req.WorkflowId, "err", err) + return err + } + + workflowRun, err := s.workflowRunRepo.GetById(ctx, req.RunId) + if err != nil { + app.GetLogger().Error("failed to get workflow run", "id", req.RunId, "err", err) + return err + } else if workflowRun.WorkflowId != workflow.Id { + return errors.New("workflow run not found") + } else if workflowRun.Status != domain.WorkflowRunStatusTypePending && workflowRun.Status != domain.WorkflowRunStatusTypeRunning { + return errors.New("workflow run is not pending or running") + } + // TODO: 取消运行，防止因为某些原因意外挂起（如进程被杀死）导致工作流一直处于 running 状态无法重新运行 + // workflowRun.Status = domain.WorkflowRunStatusTypeCanceled + // workflowRun.EndedAt = time.Now() + // if _, err := s.workflowRunRepo.Save(ctx, workflowRun); err != nil { + // return err + // } + + // return nil return errors.New("TODO: 尚未实现") } @@ -113,13 +137,14 @@ func (s *WorkflowService) Stop(ctx context.Context) { s.wg.Wait() } -func (s *WorkflowService) run(ctx context.Context) { +func (s *WorkflowService) startRun(ctx context.Context) { defer s.wg.Done() + for { select { case data := <-s.ch: - if err := s.runWithData(ctx, data); err != nil { - app.GetLogger().Error("failed to run workflow", "id", data.Workflow.Id, "err", err) + if err := s.startRunWithData(ctx, data); err != nil { + app.GetLogger().Error("failed to run workflow", "id", data.WorkflowId, "err", err) } case <-ctx.Done(): return @@ -127,27 +152,26 @@ func (s *WorkflowService) run(ctx context.Context) { } } -func (s *WorkflowService) runWithData(ctx context.Context, runData *workflowRunData) error { - workflow := runData.Workflow +func (s *WorkflowService) startRunWithData(ctx context.Context, data *workflowRunData) error { run := &domain.WorkflowRun{ - WorkflowId: workflow.Id, + WorkflowId: data.WorkflowId, Status: domain.WorkflowRunStatusTypeRunning, - Trigger: runData.RunTrigger, + Trigger: data.RunTrigger, StartedAt: time.Now(), } - if resp, err := s.repo.SaveRun(ctx, run); err != nil { + if resp, err := s.workflowRunRepo.Save(ctx, run); err != nil { return err } else { run = resp } - processor := processor.NewWorkflowProcessor(workflow, run) - if runErr := processor.Run(ctx); runErr != nil { + processor := processor.NewWorkflowProcessor(data.WorkflowId, data.WorkflowContent, run.Id) + if runErr := processor.Process(ctx); runErr != nil { run.Status = domain.WorkflowRunStatusTypeFailed run.EndedAt = time.Now() - run.Logs = processor.GetRunLogs() + run.Logs = processor.GetLogs() run.Error = runErr.Error() - if _, err := s.repo.SaveRun(ctx, run); err != nil { + if _, err := s.workflowRunRepo.Save(ctx, run); err != nil { app.GetLogger().Error("failed to save workflow run", "err", err) } @@ -155,14 +179,14 @@ func (s *WorkflowService) runWithData(ctx context.Context, runData *workflowRunD } run.EndedAt = time.Now() - run.Logs = processor.GetRunLogs() + run.Logs = processor.GetLogs() run.Error = domain.WorkflowRunLogs(run.Logs).ErrorString() if run.Error == "" { run.Status = domain.WorkflowRunStatusTypeSucceeded } else { run.Status = domain.WorkflowRunStatusTypeFailed } - if _, err := s.repo.SaveRun(ctx, run); err != nil { + if _, err := s.workflowRunRepo.Save(ctx, run); err != nil { app.GetLogger().Error("failed to save workflow run", "err", err) return err } diff --git a/ui/src/components/workflow/WorkflowRunDetail.tsx b/ui/src/components/workflow/WorkflowRunDetail.tsx new file mode 100644 index 00000000..db9610f7 --- /dev/null +++ b/ui/src/components/workflow/WorkflowRunDetail.tsx @@ -0,0 +1,52 @@ +import { useTranslation } from "react-i18next"; +import { Alert, Typography } from "antd"; +import dayjs from "dayjs"; + +import Show from "@/components/Show"; +import { WORKFLOW_RUN_STATUSES, type WorkflowRunModel } from "@/domain/workflowRun"; + +export type WorkflowRunDetailProps = { + className?: string; + style?: React.CSSProperties; + data: WorkflowRunModel; +}; + +const WorkflowRunDetail = ({ data, ...props }: WorkflowRunDetailProps) => { + const { t } = useTranslation(); + + return ( +
+ + {t("workflow_run.props.status.succeeded")}} /> + + + + {t("workflow_run.props.status.failed")}} /> + + +
+
+ {data.logs?.map((item, i) => { + return ( +
+
{item.nodeName}
+
+ {item.outputs?.map((output, j) => { + return ( +
+
[{dayjs(output.time).format("YYYY-MM-DD HH:mm:ss")}]
+ {output.error ?
{output.error}
:
{output.content}
} +
+ ); + })} +
+
+ ); + })} +
+
+
+ ); +}; + +export default WorkflowRunDetail; diff --git a/ui/src/components/workflow/WorkflowRunDetailDrawer.tsx b/ui/src/components/workflow/WorkflowRunDetailDrawer.tsx index 11e2847c..44ee8be8 100644 --- a/ui/src/components/workflow/WorkflowRunDetailDrawer.tsx +++ b/ui/src/components/workflow/WorkflowRunDetailDrawer.tsx @@ -1,12 +1,12 @@ -import { useTranslation } from "react-i18next"; import { useControllableValue } from "ahooks"; -import { Alert, Drawer, Typography } from "antd"; -import dayjs from "dayjs"; +import { Drawer } from "antd"; import Show from "@/components/Show"; -import { WORKFLOW_RUN_STATUSES, type WorkflowRunModel } from "@/domain/workflowRun"; +import { type WorkflowRunModel } from "@/domain/workflowRun"; import { useTriggerElement } from "@/hooks"; +import WorkflowRunDetail from "./WorkflowRunDetail"; + export type WorkflowRunDetailDrawerProps = { data?: WorkflowRunModel; loading?: boolean; @@ -16,8 +16,6 @@ export type WorkflowRunDetailDrawerProps = { }; const WorkflowRunDetailDrawer = ({ data, loading, trigger, ...props }: WorkflowRunDetailDrawerProps) => { - const { t } = useTranslation(); - const [open, setOpen] = useControllableValue(props, { valuePropName: "open", defaultValuePropName: "defaultOpen", @@ -30,37 +28,19 @@ const WorkflowRunDetailDrawer = ({ data, loading, trigger, ...props }: WorkflowR <> {triggerEl} - setOpen(false)}> + setOpen(false)} + > - - {t("workflow_run.props.status.succeeded")}} /> - - - - {t("workflow_run.props.status.failed")}} /> - - -
-
- {data!.logs?.map((item, i) => { - return ( -
-
{item.nodeName}
-
- {item.outputs?.map((output, j) => { - return ( -
-
[{dayjs(output.time).format("YYYY-MM-DD HH:mm:ss")}]
- {output.error ?
{output.error}
:
{output.content}
} -
- ); - })} -
-
- ); - })} -
-
+