refactor: clean code

This commit is contained in:
Fu Diwei
2025-02-06 23:11:16 +08:00
parent 3f9fda8a2d
commit 886f166e66
22 changed files with 311 additions and 194 deletions

View File

@@ -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
}