execute workflows asynchronously

internal/domain/workflow.go

@@ -40,6 +40,10 @@ type Workflow struct {
 	LastRunTime   time.Time             `json:"lastRunTime" db:"lastRunTime"`
 }
 
+func (w *Workflow) Table() string {
+	return "workflow"
+}
+
 type WorkflowNode struct {
 	Id   string           `json:"id"`
 	Type WorkflowNodeType `json:"type"`

internal/repository/workflow.go

@@ -40,6 +40,39 @@ func (w *WorkflowRepository) ListEnabledAuto(ctx context.Context) ([]domain.Work
 	return rs, nil
 }
 
+func (w *WorkflowRepository) Save(ctx context.Context, workflow *domain.Workflow) error {
+	collection, err := app.GetApp().Dao().FindCollectionByNameOrId(workflow.Table())
+	if err != nil {
+		return err
+	}
+	var record *models.Record
+	if workflow.Id == "" {
+		record = models.NewRecord(collection)
+	} else {
+		record, err = app.GetApp().Dao().FindRecordById(workflow.Table(), workflow.Id)
+		if err != nil {
+			if errors.Is(err, sql.ErrNoRows) {
+				return domain.ErrRecordNotFound
+			}
+			return err
+		}
+	}
+
+	record.Set("name", workflow.Name)
+	record.Set("description", workflow.Description)
+	record.Set("trigger", string(workflow.Trigger))
+	record.Set("triggerCron", workflow.TriggerCron)
+	record.Set("enabled", workflow.Enabled)
+	record.Set("content", workflow.Content)
+	record.Set("draft", workflow.Draft)
+	record.Set("hasDraft", workflow.HasDraft)
+	record.Set("lastRunId", workflow.LastRunId)
+	record.Set("lastRunStatus", string(workflow.LastRunStatus))
+	record.Set("lastRunTime", workflow.LastRunTime)
+
+	return app.GetApp().Dao().SaveRecord(record)
+}
+
 func (w *WorkflowRepository) SaveRun(ctx context.Context, run *domain.WorkflowRun) error {
 	collection, err := app.GetApp().Dao().FindCollectionByNameOrId("workflow_run")
 	if err != nil {
@@ -60,20 +93,17 @@ func (w *WorkflowRepository) SaveRun(ctx context.Context, run *domain.WorkflowRu
 			return err
 		}
 
-		_, err = txDao.DB().Update(
-			"workflow",
-			dbx.Params{
-				"lastRunId":     record.GetId(),
-				"lastRunStatus": record.GetString("status"),
-				"lastRunTime":   record.GetString("startedAt"),
-			},
-			dbx.NewExp("id={:id}", dbx.Params{"id": run.WorkflowId}),
-		).Execute()
+		// unable trigger sse using DB()
+		wordflowRecord, err := txDao.FindRecordById("workflow", run.WorkflowId)
 		if err != nil {
 			return err
 		}
 
-		return nil
+		wordflowRecord.Set("lastRunId", record.GetId())
+		wordflowRecord.Set("lastRunStatus", record.GetString("status"))
+		wordflowRecord.Set("lastRunTime", record.GetString("startedAt"))
+
+		return txDao.SaveRecord(wordflowRecord)
 	})
 	if err != nil {
 		return err

internal/workflow/service.go

@@ -2,7 +2,9 @@ package workflow
 
 import (
 	"context"
+	"errors"
 	"fmt"
+	"sync"
 	"time"
 
 	"github.com/usual2970/certimate/internal/app"
@@ -10,19 +12,56 @@ import (
 	nodeprocessor "github.com/usual2970/certimate/internal/workflow/node-processor"
 )
 
+const defaultRoutines = 10
+
+type workflowRunData struct {
+	Workflow *domain.Workflow
+	Options  *domain.WorkflowRunReq
+}
+
 type WorkflowRepository interface {
 	GetById(ctx context.Context, id string) (*domain.Workflow, error)
 	SaveRun(ctx context.Context, run *domain.WorkflowRun) error
+	Save(ctx context.Context, workflow *domain.Workflow) error
 	ListEnabledAuto(ctx context.Context) ([]domain.Workflow, error)
 }
 
 type WorkflowService struct {
-	repo WorkflowRepository
+	ch     chan *workflowRunData
+	repo   WorkflowRepository
+	wg     sync.WaitGroup
+	cancel context.CancelFunc
 }
 
 func NewWorkflowService(repo WorkflowRepository) *WorkflowService {
-	return &WorkflowService{
+	rs := &WorkflowService{
 		repo: repo,
+		ch:   make(chan *workflowRunData, 1),
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	rs.cancel = cancel
+
+	rs.wg.Add(defaultRoutines)
+	for i := 0; i < defaultRoutines; i++ {
+		go rs.process(ctx)
+	}
+
+	return rs
+}
+
+func (s *WorkflowService) process(ctx context.Context) {
+	defer s.wg.Done()
+	for {
+		select {
+		case data := <-s.ch:
+			// 执行
+			if err := s.run(ctx, data); err != nil {
+				app.GetLogger().Error("failed to run workflow", "id", data.Workflow.Id, "err", err)
+			}
+		case <-ctx.Done():
+			return
+		}
 	}
 }
 
@@ -60,7 +99,32 @@ func (s *WorkflowService) Run(ctx context.Context, options *domain.WorkflowRunRe
 		return err
 	}
 
+	if workflow.LastRunStatus == domain.WorkflowRunStatusTypeRunning {
+		return errors.New("workflow is running")
+	}
+
+	// set last run
+	workflow.LastRunTime = time.Now()
+	workflow.LastRunStatus = domain.WorkflowRunStatusTypeRunning
+	workflow.LastRunId = ""
+
+	if err := s.repo.Save(ctx, workflow); err != nil {
+		return err
+	}
+
+	s.ch <- &workflowRunData{
+		Workflow: workflow,
+		Options:  options,
+	}
+
+	return nil
+}
+
+func (s *WorkflowService) run(ctx context.Context, runData *workflowRunData) error {
 	// 执行
+	workflow := runData.Workflow
+	options := runData.Options
+
 	run := &domain.WorkflowRun{
 		WorkflowId: workflow.Id,
 		Status:     domain.WorkflowRunStatusTypeRunning,
@@ -100,3 +164,8 @@ func (s *WorkflowService) Run(ctx context.Context, options *domain.WorkflowRunRe
 
 	return nil
 }
+
+func (s *WorkflowService) Stop() {
+	s.cancel()
+	s.wg.Wait()
+}