refactor: re-impl workflow dispatcher

internal/certificate/service.go

@@ -5,6 +5,7 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"log/slog"
 	"strings"
 	"time"
 
@@ -193,7 +194,7 @@ func (s *CertificateService) ValidatePrivateKey(ctx context.Context, req *dtos.C
 func (s *CertificateService) cleanupExpiredCertificates(ctx context.Context) error {
 	settings, err := s.settingsRepo.GetByName(ctx, "persistence")
 	if err != nil {
-		app.GetLogger().Error("failed to get persistence settings", "err", err)
+		app.GetLogger().Error("failed to get persistence settings", slog.Any("error", err))
 		return err
 	}
 
@@ -204,7 +205,7 @@ func (s *CertificateService) cleanupExpiredCertificates(ctx context.Context) err
 			dbx.NewExp(fmt.Sprintf("validityNotAfter<DATETIME('now', '-%d days')", persistenceSettings.ExpiredCertificatesMaxDaysRetention)),
 		)
 		if err != nil {
-			app.GetLogger().Error("failed to delete expired certificates", "err", err)
+			app.GetLogger().Error("failed to delete expired certificates", slog.Any("error", err))
 			return err
 		}
 

internal/domain/dtos/workflow.go

@@ -7,7 +7,13 @@ type WorkflowStartRunReq struct {
 	RunTrigger domain.WorkflowTriggerType `json:"trigger"`
 }
 
+type WorkflowStartRunResp struct {
+	RunId string `json:"runId"`
+}
+
 type WorkflowCancelRunReq struct {
 	WorkflowId string `json:"-"`
 	RunId      string `json:"-"`
 }
+
+type WorkflowCancelRunResp struct{}

internal/repository/workflow_run.go

@@ -61,6 +61,9 @@ func (r *WorkflowRunRepository) Save(ctx context.Context, workflowRun *domain.Wo
 		return workflowRun, err
 	}
 
+	workflowRun.Id = record.Id
+	workflowRun.CreatedAt = record.GetDateTime("created").Time()
+	workflowRun.UpdatedAt = record.GetDateTime("updated").Time()
 	return workflowRun, nil
 }
 

internal/rest/handlers/certificate.go

@@ -38,11 +38,12 @@ func (handler *CertificateHandler) archiveFile(e *core.RequestEvent) error {
 		return resp.Err(e, err)
 	}
 
-	if res, err := handler.service.DownloadArchivedFile(e.Request.Context(), req); err != nil {
+	res, err := handler.service.DownloadArchivedFile(e.Request.Context(), req)
+	if err != nil {
 		return resp.Err(e, err)
-	} else {
-		return resp.Ok(e, res)
 	}
+
+	return resp.Ok(e, res)
 }
 
 func (handler *CertificateHandler) validateCertificate(e *core.RequestEvent) error {
@@ -51,11 +52,12 @@ func (handler *CertificateHandler) validateCertificate(e *core.RequestEvent) err
 		return resp.Err(e, err)
 	}
 
-	if res, err := handler.service.ValidateCertificate(e.Request.Context(), req); err != nil {
+	res, err := handler.service.ValidateCertificate(e.Request.Context(), req)
+	if err != nil {
 		return resp.Err(e, err)
-	} else {
-		return resp.Ok(e, res)
 	}
+
+	return resp.Ok(e, res)
 }
 
 func (handler *CertificateHandler) validatePrivateKey(e *core.RequestEvent) error {
@@ -64,9 +66,10 @@ func (handler *CertificateHandler) validatePrivateKey(e *core.RequestEvent) erro
 		return resp.Err(e, err)
 	}
 
-	if res, err := handler.service.ValidatePrivateKey(e.Request.Context(), req); err != nil {
+	res, err := handler.service.ValidatePrivateKey(e.Request.Context(), req)
+	if err != nil {
 		return resp.Err(e, err)
-	} else {
-		return resp.Ok(e, res)
 	}
+
+	return resp.Ok(e, res)
 }

internal/rest/handlers/statistics.go

@@ -28,9 +28,10 @@ func NewStatisticsHandler(router *router.RouterGroup[*core.RequestEvent], servic
 }
 
 func (handler *StatisticsHandler) get(e *core.RequestEvent) error {
-	if statistics, err := handler.service.Get(e.Request.Context()); err != nil {
+	res, err := handler.service.Get(e.Request.Context())
+	if err != nil {
 		return resp.Err(e, err)
-	} else {
-		return resp.Ok(e, statistics)
 	}
+
+	return resp.Ok(e, res)
 }

internal/rest/handlers/workflow.go

@@ -11,8 +11,8 @@ import (
 )
 
 type workflowService interface {
-	StartRun(ctx context.Context, req *dtos.WorkflowStartRunReq) error
-	CancelRun(ctx context.Context, req *dtos.WorkflowCancelRunReq) error
+	StartRun(ctx context.Context, req *dtos.WorkflowStartRunReq) (*dtos.WorkflowStartRunResp, error)
+	CancelRun(ctx context.Context, req *dtos.WorkflowCancelRunReq) (*dtos.WorkflowCancelRunResp, error)
 	Shutdown(ctx context.Context)
 }
 
@@ -37,11 +37,12 @@ func (handler *WorkflowHandler) run(e *core.RequestEvent) error {
 		return resp.Err(e, err)
 	}
 
-	if err := handler.service.StartRun(e.Request.Context(), req); err != nil {
+	res, err := handler.service.StartRun(e.Request.Context(), req)
+	if err != nil {
 		return resp.Err(e, err)
 	}
 
-	return resp.Ok(e, nil)
+	return resp.Ok(e, res)
 }
 
 func (handler *WorkflowHandler) cancel(e *core.RequestEvent) error {
@@ -49,9 +50,10 @@ func (handler *WorkflowHandler) cancel(e *core.RequestEvent) error {
 	req.WorkflowId = e.Request.PathValue("workflowId")
 	req.RunId = e.Request.PathValue("runId")
 
-	if err := handler.service.CancelRun(e.Request.Context(), req); err != nil {
+	res, err := handler.service.CancelRun(e.Request.Context(), req)
+	if err != nil {
 		return resp.Err(e, err)
 	}
 
-	return resp.Ok(e, nil)
+	return resp.Ok(e, res)
 }

internal/scheduler/scheduler.go

@@ -1,6 +1,8 @@
 package scheduler
 
 import (
+	"log/slog"
+
 	"github.com/certimate-go/certimate/internal/app"
 	"github.com/certimate-go/certimate/internal/certificate"
 	"github.com/certimate-go/certimate/internal/repository"
@@ -17,10 +19,10 @@ func Register() {
 	certificateSvc := certificate.NewCertificateService(certificateRepo, settingsRepo)
 
 	if err := InitWorkflowScheduler(workflowSvc); err != nil {
-		app.GetLogger().Error("failed to init workflow scheduler", "err", err)
+		app.GetLogger().Error("failed to init workflow scheduler", slog.Any("error", err))
 	}
 
 	if err := InitCertificateScheduler(certificateSvc); err != nil {
-		app.GetLogger().Error("failed to init certificate scheduler", "err", err)
+		app.GetLogger().Error("failed to init certificate scheduler", slog.Any("error", err))
 	}
 }

internal/workflow/dispatcher/deps.go

@@ -0,0 +1,22 @@
+package dispatcher
+
+import (
+	"context"
+
+	"github.com/certimate-go/certimate/internal/domain"
+)
+
+type workflowRepository interface {
+	GetById(ctx context.Context, id string) (*domain.Workflow, error)
+	Save(ctx context.Context, workflow *domain.Workflow) (*domain.Workflow, error)
+}
+
+type workflowRunRepository interface {
+	GetById(ctx context.Context, id string) (*domain.WorkflowRun, error)
+	Save(ctx context.Context, workflowRun *domain.WorkflowRun) (*domain.WorkflowRun, error)
+	SaveWithCascading(ctx context.Context, workflowRun *domain.WorkflowRun) (*domain.WorkflowRun, error)
+}
+
+type workflowLogRepository interface {
+	Save(ctx context.Context, workflowLog *domain.WorkflowLog) (*domain.WorkflowLog, error)
+}

internal/workflow/dispatcher/dispatcher.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"log"
+	"log/slog"
 	"os"
 	"runtime"
 	"runtime/debug"
@@ -12,10 +13,11 @@ import (
 	"sync"
 	"time"
 
-	"github.com/samber/lo"
-
 	"github.com/certimate-go/certimate/internal/app"
 	"github.com/certimate-go/certimate/internal/domain"
+	"github.com/certimate-go/certimate/internal/repository"
+	"github.com/certimate-go/certimate/internal/workflow/engine"
+	"github.com/certimate-go/certimate/pkg/logging"
 )
 
 var maxWorkers = 1
@@ -32,280 +34,288 @@ func init() {
 	}
 }
 
-type workflowWorker struct {
-	Data   *WorkflowWorkerData
-	Cancel context.CancelFunc
+type WorkflowDispatcher interface {
+	Bootup(ctx context.Context) error
+	Shutdown(ctx context.Context) error
+	Start(ctx context.Context, runId string) error
+	Cancel(ctx context.Context, runId string) error
 }
 
-type WorkflowWorkerData struct {
-	WorkflowId    string
-	WorkflowGraph *domain.WorkflowGraph
-	RunId         string
-}
+type workflowDispatcher struct {
+	booted      bool
+	concurrency int
 
-type WorkflowDispatcher struct {
-	semaphore chan struct{}
-
-	queue      []*WorkflowWorkerData
-	queueMutex sync.Mutex
-
-	workers     map[string]*workflowWorker // key: WorkflowId
-	workerIdMap map[string]string          // key: RunId, value: WorkflowId
-	workerMutex sync.Mutex
-
-	chWork  chan *WorkflowWorkerData
-	chCandi chan struct{}
-
-	wg sync.WaitGroup
+	taskMtx         sync.RWMutex
+	pendingRunQueue []string
+	processingTasks map[string]*taskInfo // Key: RunId
 
 	workflowRepo    workflowRepository
 	workflowRunRepo workflowRunRepository
 	workflowLogRepo workflowLogRepository
 }
 
-func newWorkflowDispatcher(
-	workflowRepo workflowRepository,
-	workflowRunRepo workflowRunRepository,
-	workflowLogRepo workflowLogRepository,
-) *WorkflowDispatcher {
-	dispatcher := &WorkflowDispatcher{
-		semaphore: make(chan struct{}, maxWorkers),
+var _ WorkflowDispatcher = (*workflowDispatcher)(nil)
 
-		queue:      make([]*WorkflowWorkerData, 0),
-		queueMutex: sync.Mutex{},
-
-		workers:     make(map[string]*workflowWorker),
-		workerIdMap: make(map[string]string),
-		workerMutex: sync.Mutex{},
-
-		chWork:  make(chan *WorkflowWorkerData),
-		chCandi: make(chan struct{}, 1),
-
-		workflowRepo:    workflowRepo,
-		workflowRunRepo: workflowRunRepo,
-		workflowLogRepo: workflowLogRepo,
+func (wd *workflowDispatcher) Bootup(ctx context.Context) error {
+	if wd.booted {
+		return errors.New("could not re-bootup")
 	}
 
-	go func() {
-		for {
-			select {
-			case <-dispatcher.chWork:
-				dispatcher.dequeueWorker()
+	wd.taskMtx.Lock()
+	defer wd.taskMtx.Unlock()
 
-			case <-dispatcher.chCandi:
-				dispatcher.dequeueWorker()
-			}
-		}
-	}()
+	if _, err := app.GetDB().NewQuery("UPDATE workflow SET lastRunStatus = 'canceled' WHERE lastRunStatus = 'pending' OR lastRunStatus = 'processing'").Execute(); err != nil {
+		return err
+	}
+	if _, err := app.GetDB().NewQuery("UPDATE workflow_run SET status = 'canceled' WHERE status = 'pending' OR status = 'processing'").Execute(); err != nil {
+		return err
+	}
 
-	return dispatcher
+	wd.booted = true
+	return nil
 }
 
-func (d *WorkflowDispatcher) Dispatch(data *WorkflowWorkerData) {
-	if data == nil {
-		panic("workflow dispatcher: the worker data is nil")
+func (wd *workflowDispatcher) Shutdown(ctx context.Context) error {
+	if !wd.booted {
+		return errors.New("could not re-shutdown")
 	}
 
-	d.enqueueWorker(data)
+	wd.taskMtx.Lock()
+	defer wd.taskMtx.Unlock()
 
-	select {
-	case d.chWork <- data:
-	default:
+	for runId, task := range wd.processingTasks {
+		task.cancel()
+		delete(wd.processingTasks, runId)
 	}
+
+	wd.booted = false
+	wd.pendingRunQueue = make([]string, 0)
+	wd.processingTasks = make(map[string]*taskInfo)
+	return nil
 }
 
-func (d *WorkflowDispatcher) Cancel(runId string) {
-	hasWorker := false
+func (wd *workflowDispatcher) Start(ctx context.Context, runId string) error {
+	wd.taskMtx.Lock()
+	defer wd.taskMtx.Unlock()
 
-	// 取消正在执行的 WorkflowRun
-	d.workerMutex.Lock()
-	if workflowId, ok := d.workerIdMap[runId]; ok {
-		if worker, ok := d.workers[workflowId]; ok {
-			hasWorker = true
-			worker.Cancel()
-			delete(d.workers, workflowId)
-			delete(d.workerIdMap, runId)
+	if _, exists := wd.processingTasks[runId]; exists {
+		return errors.New("workflow run is already processing")
+	}
+
+	for _, pendingRunId := range wd.pendingRunQueue {
+		if pendingRunId == runId {
+			return errors.New("workflow run is already in the queue")
 		}
 	}
-	d.workerMutex.Unlock()
 
-	// 移除排队中的 WorkflowRun
-	d.queueMutex.Lock()
-	d.queue = lo.Filter(d.queue, func(d *WorkflowWorkerData, _ int) bool {
-		return d.RunId != runId
-	})
-	d.queueMutex.Unlock()
+	wd.pendingRunQueue = append(wd.pendingRunQueue, runId)
+	go func() { wd.tryNextAsync() }()
 
-	// 已挂起，查询 WorkflowRun 并更新其状态为 Canceled
-	if !hasWorker {
-		if run, err := d.workflowRunRepo.GetById(context.Background(), runId); err == nil {
-			if run.Status == domain.WorkflowRunStatusTypePending || run.Status == domain.WorkflowRunStatusTypeProcessing {
-				run.Status = domain.WorkflowRunStatusTypeCanceled
-				d.workflowRunRepo.SaveWithCascading(context.Background(), run)
-			}
+	return nil
+}
+
+func (wd *workflowDispatcher) Cancel(ctx context.Context, runId string) error {
+	wd.taskMtx.Lock()
+	defer wd.taskMtx.Unlock()
+
+	workflowRun, err := wd.workflowRunRepo.GetById(ctx, runId)
+	if err != nil {
+		return err
+	} else if workflowRun.Status != domain.WorkflowRunStatusTypePending && workflowRun.Status != domain.WorkflowRunStatusTypeProcessing {
+		return errors.New("workflow run is already completed")
+	}
+
+	workflow, err := wd.workflowRepo.GetById(ctx, workflowRun.WorkflowId)
+	if err != nil {
+		return err
+	}
+
+	workflowRun.Status = domain.WorkflowRunStatusTypeCanceled
+	if workflow.LastRunId == workflowRun.Id {
+		_, err := wd.workflowRunRepo.SaveWithCascading(ctx, workflowRun)
+		if err != nil {
+			return err
+		}
+	} else {
+		_, err := wd.workflowRunRepo.Save(ctx, workflowRun)
+		if err != nil {
+			return err
 		}
 	}
-}
 
-func (d *WorkflowDispatcher) Shutdown() {
-	// 清空排队中的 WorkflowRun
-	d.queueMutex.Lock()
-	d.queue = make([]*WorkflowWorkerData, 0)
-	d.queueMutex.Unlock()
-
-	// 等待所有正在执行的 WorkflowRun 完成
-	d.workerMutex.Lock()
-	for _, worker := range d.workers {
-		worker.Cancel()
-		delete(d.workers, worker.Data.WorkflowId)
-		delete(d.workerIdMap, worker.Data.RunId)
+	if task, exists := wd.processingTasks[runId]; exists {
+		task.cancel()
+		delete(wd.processingTasks, runId)
 	}
-	d.workerMutex.Unlock()
-	d.wg.Wait()
-}
 
-func (d *WorkflowDispatcher) enqueueWorker(data *WorkflowWorkerData) {
-	d.queueMutex.Lock()
-	defer d.queueMutex.Unlock()
-	d.queue = append(d.queue, data)
-}
-
-func (d *WorkflowDispatcher) dequeueWorker() {
-	for {
-		select {
-		case d.semaphore <- struct{}{}:
-		default:
-			// 达到最大并发数
-			return
+	for i, pendingRunId := range wd.pendingRunQueue {
+		if pendingRunId == runId {
+			wd.pendingRunQueue = append(wd.pendingRunQueue[:i], wd.pendingRunQueue[i+1:]...)
+			break
 		}
-
-		d.queueMutex.Lock()
-		if len(d.queue) == 0 {
-			d.queueMutex.Unlock()
-			<-d.semaphore
-			return
-		}
-
-		// TODO: 优化队列
-		data := d.queue[0]
-		d.queue = d.queue[1:]
-		d.queueMutex.Unlock()
-
-		// 检查是否有相同 WorkflowId 的 WorkflowRun 正在执行
-		// 如果有，则重新排队，以保证同一个工作流同一时间内只有一个正在执行
-		// 即不同 WorkflowId 的任务并行化，相同 WorkflowId 的任务串行化
-		d.workerMutex.Lock()
-		if _, exists := d.workers[data.WorkflowId]; exists {
-			d.queueMutex.Lock()
-			d.queue = append(d.queue, data)
-			d.queueMutex.Unlock()
-			d.workerMutex.Unlock()
-
-			<-d.semaphore
-
-			continue
-		}
-
-		ctx, cancel := context.WithCancel(context.Background())
-		d.workers[data.WorkflowId] = &workflowWorker{data, cancel}
-		d.workerIdMap[data.RunId] = data.WorkflowId
-		d.workerMutex.Unlock()
-
-		d.wg.Add(1)
-		go d.work(ctx, data)
 	}
+
+	go func() { wd.tryNextAsync() }()
+
+	return nil
 }
 
-func (d *WorkflowDispatcher) work(ctx context.Context, data *WorkflowWorkerData) {
-	var run *domain.WorkflowRun
-	var err error
+func (wd *workflowDispatcher) tryExecuteAsync(task *taskInfo) {
+	var workflowRun *domain.WorkflowRun
 
+	// 捕获 panic
 	defer func() {
-		// 捕获 panic，避免影响其他工作流的执行
 		if r := recover(); r != nil {
-			log.Default().Println("WorkflowId:", data.WorkflowId, "RunId:", data.RunId)
-			log.Default().Println("Recovered from panic:", r)
-			log.Default().Println("Stack trace:", string(debug.Stack()))
-			if run != nil {
-				run.Status = domain.WorkflowRunStatusTypeFailed
-				run.EndedAt = time.Now()
-				run.Error = fmt.Sprintf("workflow run panic: %v", r)
-				if _, err := d.workflowRunRepo.SaveWithCascading(ctx, run); err != nil {
-					log.Default().Println("Failed to save workflow run after panic:", err)
+			slog.Default().Warn(fmt.Sprintf("workflow dispatcher panic: %v, stack trace: %s", r, string(debug.Stack())), slog.Any("workflowId", task.WorkflowId), slog.Any("runId", task.RunId))
+			app.GetLogger().Error(fmt.Sprintf("workflow dispatcher panic: %v", r), slog.Any("workflowId", task.WorkflowId), slog.Any("runId", task.RunId))
+
+			if workflowRun != nil {
+				workflowRun.Status = domain.WorkflowRunStatusTypeFailed
+				workflowRun.EndedAt = time.Now()
+				workflowRun.Error = fmt.Sprintf("workflow dispatcher panic: %v", r)
+				if _, err := wd.workflowRunRepo.SaveWithCascading(context.Background(), workflowRun); err != nil {
+					log.Default().Println("failed to save workflow run after panic", slog.Any("error", err))
 				}
 			}
 		}
-
-		<-d.semaphore
-		d.workerMutex.Lock()
-		delete(d.workers, data.WorkflowId)
-		delete(d.workerIdMap, data.RunId)
-		d.workerMutex.Unlock()
-
-		d.wg.Done()
-
-		// 尝试取出排队中的其他 WorkflowRun 继续执行
-		select {
-		case d.chCandi <- struct{}{}:
-		default:
-		}
 	}()
 
-	// 查询 WorkflowRun
-	run, err = d.workflowRunRepo.GetById(ctx, data.RunId)
-	if err != nil {
+	// 尝试继续执行等待队列中的任务
+	defer func() {
+		delete(wd.processingTasks, task.RunId)
+		wd.tryNextAsync()
+	}()
+
+	// 查询运行实体，并级联更新状态
+	if run, err := wd.workflowRunRepo.GetById(task.ctx, task.RunId); err != nil {
 		if !(errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded)) {
-			app.GetLogger().Error(fmt.Sprintf("failed to get workflow run #%s", data.RunId), "err", err)
+			app.GetLogger().Error(fmt.Sprintf("failed to get workflow run #%s record", task.RunId), slog.Any("error", err))
 		}
 		return
-	} else if run.Status != domain.WorkflowRunStatusTypePending {
-		return
-	} else if ctx.Err() != nil {
-		run.Status = domain.WorkflowRunStatusTypeCanceled
-		d.workflowRunRepo.SaveWithCascading(ctx, run)
-		return
+	} else {
+		workflowRun = run
+
+		if run.Status == domain.WorkflowRunStatusTypePending {
+			run.Status = domain.WorkflowRunStatusTypeProcessing
+			wd.workflowRunRepo.SaveWithCascading(task.ctx, run)
+		} else {
+			// WTF? That should be impossible!
+			return
+		}
 	}
 
-	// 更新 WorkflowRun 状态为 Processing
-	run.Status = domain.WorkflowRunStatusTypeProcessing
-	if _, err := d.workflowRunRepo.SaveWithCascading(ctx, run); err != nil {
-		if !(errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded)) {
-			panic(err)
+	// 初始化工作流引擎
+	logsBuf := make(domain.WorkflowLogs, 0)
+	we := engine.NewWorkflowEngine()
+	we.OnEnd(func(ctx context.Context) error {
+		if errmsg := logsBuf.ErrorString(); errmsg == "" {
+			workflowRun.Status = domain.WorkflowRunStatusTypeSucceeded
+			workflowRun.EndedAt = time.Now()
+		} else {
+			workflowRun.Status = domain.WorkflowRunStatusTypeFailed
+			workflowRun.EndedAt = time.Now()
+			workflowRun.Error = errmsg
 		}
-		return
-	}
+		wd.workflowRunRepo.SaveWithCascading(task.ctx, workflowRun)
+		return nil
+	})
+	we.OnError(func(ctx context.Context, err error) error {
+		if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
+			workflowRun.Status = domain.WorkflowRunStatusTypeCanceled
+			wd.workflowRunRepo.SaveWithCascading(context.Background(), workflowRun)
+		} else {
+			workflowRun.Status = domain.WorkflowRunStatusTypeFailed
+			workflowRun.EndedAt = time.Now()
+			workflowRun.Error = err.Error()
+			wd.workflowRunRepo.SaveWithCascading(task.ctx, workflowRun)
+		}
+		return nil
+	})
+	we.OnNodeError(func(ctx context.Context, node *engine.Node, err error) error {
+		log := domain.WorkflowLog{}
+		log.WorkflowId = task.WorkflowId
+		log.RunId = task.RunId
+		log.NodeId = node.Id
+		log.NodeName = node.Data.Name
+		log.Timestamp = time.Now().UnixMilli()
+		log.Level = int32(slog.LevelError)
+		log.Message = err.Error()
+		log.CreatedAt = time.Now()
+		if _, err := wd.workflowLogRepo.Save(ctx, &log); err != nil {
+			app.GetLogger().Error(err.Error())
+		}
+
+		logsBuf = append(logsBuf, log)
+		return nil
+	})
+	we.OnNodeLogging(func(ctx context.Context, node *engine.Node, record logging.Record) error {
+		log := domain.WorkflowLog{}
+		log.WorkflowId = task.WorkflowId
+		log.RunId = task.RunId
+		log.NodeId = node.Id
+		log.NodeName = node.Data.Name
+		log.Timestamp = record.Time.UnixMilli()
+		log.Level = int32(record.Level)
+		log.Message = record.Message
+		log.Data = record.Data
+		log.CreatedAt = record.Time
+		if _, err := wd.workflowLogRepo.Save(ctx, &log); err != nil {
+			app.GetLogger().Error(err.Error())
+		}
+
+		logsBuf = append(logsBuf, log)
+		return nil
+	})
 
 	// 执行工作流
-	invoker := newWorkflowInvokerWithData(d.workflowLogRepo, data)
-	if runErr := invoker.Invoke(ctx); runErr != nil {
-		if errors.Is(runErr, context.Canceled) {
-			run.Status = domain.WorkflowRunStatusTypeCanceled
-		} else {
-			run.Status = domain.WorkflowRunStatusTypeFailed
-			run.EndedAt = time.Now()
-			run.Error = runErr.Error()
+	app.GetLogger().Info(fmt.Sprintf("start to invoke workflow run #%s", task.RunId))
+	we.Invoke(task.ctx, workflowRun.WorkflowId, workflowRun.Id, &workflowRun.Graph.WorkflowGraph)
+}
+
+func (wd *workflowDispatcher) tryNextAsync() {
+	wd.taskMtx.RLock()
+
+	for i, pendingRunId := range wd.pendingRunQueue {
+		workflowRun, err := wd.workflowRunRepo.GetById(context.Background(), pendingRunId)
+		if err != nil {
+			app.GetLogger().Error(fmt.Sprintf("failed to get workflow run #%s record", pendingRunId), slog.Any("error", err))
+			continue
 		}
 
-		if _, err := d.workflowRunRepo.SaveWithCascading(ctx, run); err != nil {
-			if !(errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded)) {
-				panic(err)
+		var hasSameWorkflowTask bool // 相同 Workflow 的任务同一时间只能有一个 Run 在执行
+		for _, task := range wd.processingTasks {
+			if task.WorkflowId == workflowRun.WorkflowId {
+				hasSameWorkflowTask = true
+				break
 			}
 		}
 
-		return
-	}
+		if !hasSameWorkflowTask && len(wd.processingTasks) < wd.concurrency {
+			wd.taskMtx.RUnlock()
+			wd.taskMtx.Lock()
+			defer wd.taskMtx.Unlock()
 
-	// 更新 WorkflowRun 状态为 Succeeded/Failed
-	run.EndedAt = time.Now()
-	run.Error = invoker.GetLogs().ErrorString()
-	if run.Error == "" {
-		run.Status = domain.WorkflowRunStatusTypeSucceeded
-	} else {
-		run.Status = domain.WorkflowRunStatusTypeFailed
-	}
-	if _, err := d.workflowRunRepo.SaveWithCascading(ctx, run); err != nil {
-		if !(errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded)) {
-			panic(err)
+			ctxRun, ctxCancel := context.WithCancel(context.Background())
+			task := &taskInfo{WorkflowId: workflowRun.WorkflowId, RunId: workflowRun.Id, ctx: ctxRun, cancel: ctxCancel}
+			wd.pendingRunQueue = append(wd.pendingRunQueue[:i], wd.pendingRunQueue[i+1:]...)
+			wd.processingTasks[pendingRunId] = task
+			go func() { wd.tryExecuteAsync(task) }()
+			return
 		}
 	}
+
+	wd.taskMtx.RUnlock()
+}
+
+func newWorkflowDispatcher() WorkflowDispatcher {
+	return &workflowDispatcher{
+		concurrency: maxWorkers,
+
+		pendingRunQueue: make([]string, 0),
+		processingTasks: make(map[string]*taskInfo),
+
+		workflowRepo:    repository.NewWorkflowRepository(),
+		workflowRunRepo: repository.NewWorkflowRunRepository(),
+		workflowLogRepo: repository.NewWorkflowLogRepository(),
+	}
 }

internal/workflow/dispatcher/invoker.go

@@ -1,87 +0,0 @@
-package dispatcher
-
-import (
-	"context"
-	"log/slog"
-	"time"
-
-	"github.com/certimate-go/certimate/internal/app"
-	"github.com/certimate-go/certimate/internal/domain"
-	"github.com/certimate-go/certimate/internal/workflow/engine"
-	"github.com/certimate-go/certimate/pkg/logging"
-)
-
-type workflowInvoker struct {
-	workflowId    string
-	workflowGraph *domain.WorkflowGraph
-	runId         string
-	logs          []domain.WorkflowLog
-
-	workflowLogRepo workflowLogRepository
-}
-
-func newWorkflowInvokerWithData(workflowLogRepo workflowLogRepository, data *WorkflowWorkerData) *workflowInvoker {
-	if data == nil {
-		panic("workflow dispatcher: the worker data is nil")
-	}
-
-	return &workflowInvoker{
-		workflowId:    data.WorkflowId,
-		workflowGraph: data.WorkflowGraph,
-		runId:         data.RunId,
-		logs:          make([]domain.WorkflowLog, 0),
-
-		workflowLogRepo: workflowLogRepo,
-	}
-}
-
-func (w *workflowInvoker) Invoke(ctx context.Context) error {
-	we := engine.NewWorkflowEngine()
-	// TODO: 优化日志记录
-	we.OnNodeError(func(ctx context.Context, node *engine.Node, err error) error {
-		log := domain.WorkflowLog{}
-		log.WorkflowId = w.workflowId
-		log.RunId = w.runId
-		log.NodeId = node.Id
-		log.NodeName = node.Data.Name
-		log.Timestamp = time.Now().UnixMilli()
-		log.Level = int32(slog.LevelError)
-		log.Message = err.Error()
-		log.CreatedAt = time.Now()
-		if _, err := w.workflowLogRepo.Save(ctx, &log); err != nil {
-			app.GetLogger().Error(err.Error())
-		}
-
-		w.logs = append(w.logs, log)
-		return nil
-	})
-	we.OnNodeLogging(func(ctx context.Context, node *engine.Node, record logging.Record) error {
-		log := domain.WorkflowLog{}
-		log.WorkflowId = w.workflowId
-		log.RunId = w.runId
-		log.NodeId = node.Id
-		log.NodeName = node.Data.Name
-		log.Timestamp = record.Time.UnixMilli()
-		log.Level = int32(record.Level)
-		log.Message = record.Message
-		log.Data = record.Data
-		log.CreatedAt = record.Time
-		if _, err := w.workflowLogRepo.Save(ctx, &log); err != nil {
-			app.GetLogger().Error(err.Error())
-		}
-
-		w.logs = append(w.logs, log)
-		return nil
-	})
-
-	err := we.Invoke(ctx, w.workflowId, w.runId, w.workflowGraph)
-	if err != nil {
-		return err
-	}
-
-	return nil
-}
-
-func (w *workflowInvoker) GetLogs() domain.WorkflowLogs {
-	return w.logs
-}

internal/workflow/dispatcher/singleton.go

@@ -1,41 +1,17 @@
 package dispatcher
 
 import (
-	"context"
 	"sync"
-
-	"github.com/certimate-go/certimate/internal/domain"
-	"github.com/certimate-go/certimate/internal/repository"
 )
 
-type workflowRepository interface {
-	GetById(ctx context.Context, id string) (*domain.Workflow, error)
-	Save(ctx context.Context, workflow *domain.Workflow) (*domain.Workflow, error)
-}
-
-type workflowRunRepository interface {
-	GetById(ctx context.Context, id string) (*domain.WorkflowRun, error)
-	Save(ctx context.Context, workflowRun *domain.WorkflowRun) (*domain.WorkflowRun, error)
-	SaveWithCascading(ctx context.Context, workflowRun *domain.WorkflowRun) (*domain.WorkflowRun, error)
-}
-
-type workflowLogRepository interface {
-	Save(ctx context.Context, workflowLog *domain.WorkflowLog) (*domain.WorkflowLog, error)
-}
-
 var (
-	instance    *WorkflowDispatcher
+	instance    WorkflowDispatcher
 	intanceOnce sync.Once
 )
 
-func GetSingletonDispatcher() *WorkflowDispatcher {
+func GetSingletonDispatcher() WorkflowDispatcher {
 	intanceOnce.Do(func() {
-		instance = newWorkflowDispatcher(
-			repository.NewWorkflowRepository(),
-			repository.NewWorkflowRunRepository(),
-			repository.NewWorkflowLogRepository(),
-		)
+		instance = newWorkflowDispatcher()
 	})
-
 	return instance
 }

internal/workflow/dispatcher/task.go

@@ -0,0 +1,13 @@
+package dispatcher
+
+import (
+	"context"
+)
+
+type taskInfo struct {
+	WorkflowId string
+	RunId      string
+
+	ctx    context.Context
+	cancel context.CancelFunc
+}

internal/workflow/service.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"log/slog"
 	"time"
 
 	"github.com/pocketbase/dbx"
@@ -15,7 +16,7 @@ import (
 )
 
 type WorkflowService struct {
-	dispatcher *dispatcher.WorkflowDispatcher
+	dispatcher dispatcher.WorkflowDispatcher
 
 	workflowRepo    workflowRepository
 	workflowRunRepo workflowRunRepository
@@ -34,12 +35,17 @@ func NewWorkflowService(workflowRepo workflowRepository, workflowRunRepo workflo
 }
 
 func (s *WorkflowService) InitSchedule(ctx context.Context) error {
-	// 每日清理工作流执行历史
+	// 每日清理工作流运行历史
 	app.GetScheduler().MustAdd("cleanupWorkflowHistoryRuns", "0 0 * * *", func() {
 		s.cleanupHistoryRuns(context.Background())
 	})
 
-	// 工作流后台任务
+	// 初始化工作流调度器
+	if err := s.dispatcher.Bootup(ctx); err != nil {
+		panic(err)
+	}
+
+	// 注册工作流后台任务
 	{
 		workflows, err := s.workflowRepo.ListEnabledScheduled(ctx)
 		if err != nil {
@@ -56,7 +62,7 @@ func (s *WorkflowService) InitSchedule(ctx context.Context) error {
 				})
 			})
 			if err != nil {
-				app.GetLogger().Error(fmt.Sprintf("failed to add workflow #%s to scheduler: %w", workflow.Id, err))
+				app.GetLogger().Error(fmt.Sprintf("failed to add workflow #%s to scheduler: %w", workflow.Id), slog.Any("error", err))
 				errs = append(errs, err)
 			}
 
@@ -69,79 +75,79 @@ func (s *WorkflowService) InitSchedule(ctx context.Context) error {
 	return nil
 }
 
-func (s *WorkflowService) StartRun(ctx context.Context, req *dtos.WorkflowStartRunReq) error {
+func (s *WorkflowService) StartRun(ctx context.Context, req *dtos.WorkflowStartRunReq) (*dtos.WorkflowStartRunResp, error) {
 	workflow, err := s.workflowRepo.GetById(ctx, req.WorkflowId)
 	if err != nil {
-		return err
+		return nil, err
 	}
 
 	if workflow.LastRunStatus == domain.WorkflowRunStatusTypePending || workflow.LastRunStatus == domain.WorkflowRunStatusTypeProcessing {
-		return errors.New("workflow is already pending or processing")
+		return nil, errors.New("workflow is already pending or processing")
+	} else if workflow.GraphContent == nil {
+		return nil, errors.New("workflow graph content is empty")
+	} else if err := workflow.GraphContent.Verify(); err != nil {
+		return nil, err
 	}
 
-	run := &domain.WorkflowRun{
+	workflowRun := &domain.WorkflowRun{
 		WorkflowId: workflow.Id,
 		Status:     domain.WorkflowRunStatusTypePending,
 		Trigger:    req.RunTrigger,
 		StartedAt:  time.Now(),
 		Graph:      &domain.WorkflowGraphWithResult{WorkflowGraph: *workflow.GraphContent},
 	}
-	if resp, err := s.workflowRunRepo.SaveWithCascading(ctx, run); err != nil {
-		return err
+	if resp, err := s.workflowRunRepo.Save(ctx, workflowRun); err != nil {
+		return nil, err
 	} else {
-		run = resp
+		workflowRun = resp
 	}
 
-	s.dispatcher.Dispatch(&dispatcher.WorkflowWorkerData{
-		WorkflowId:    run.WorkflowId,
-		WorkflowGraph: &run.Graph.WorkflowGraph,
-		RunId:         run.Id,
-	})
+	s.dispatcher.Start(ctx, workflowRun.Id)
 
-	return nil
+	return &dtos.WorkflowStartRunResp{RunId: workflowRun.Id}, nil
 }
 
-func (s *WorkflowService) CancelRun(ctx context.Context, req *dtos.WorkflowCancelRunReq) error {
+func (s *WorkflowService) CancelRun(ctx context.Context, req *dtos.WorkflowCancelRunReq) (*dtos.WorkflowCancelRunResp, error) {
 	workflow, err := s.workflowRepo.GetById(ctx, req.WorkflowId)
 	if err != nil {
-		return err
+		return nil, err
 	}
 
 	workflowRun, err := s.workflowRunRepo.GetById(ctx, req.RunId)
 	if err != nil {
-		return err
+		return nil, err
 	} else if workflowRun.WorkflowId != workflow.Id {
-		return errors.New("workflow run not found")
+		return nil, errors.New("workflow run not found")
 	} else if workflowRun.Status != domain.WorkflowRunStatusTypePending && workflowRun.Status != domain.WorkflowRunStatusTypeProcessing {
-		return errors.New("workflow run is not pending or processing")
+		return nil, errors.New("workflow run is not pending or processing")
 	}
 
-	s.dispatcher.Cancel(workflowRun.Id)
+	s.dispatcher.Cancel(ctx, workflowRun.Id)
 
-	return nil
+	return &dtos.WorkflowCancelRunResp{}, nil
 }
 
 func (s *WorkflowService) Shutdown(ctx context.Context) {
-	s.dispatcher.Shutdown()
+	s.dispatcher.Shutdown(ctx)
 }
 
 func (s *WorkflowService) cleanupHistoryRuns(ctx context.Context) error {
 	settings, err := s.settingsRepo.GetByName(ctx, "persistence")
 	if err != nil {
-		app.GetLogger().Error(fmt.Sprintf("failed to get persistence settings: %w", err))
+		app.GetLogger().Error("failed to get persistence settings", slog.Any("error", err))
 		return err
 	}
 
 	persistenceSettings, _ := settings.UnmarshalContentAsPersistence()
 	if persistenceSettings != nil && persistenceSettings.WorkflowRunsMaxDaysRetention != 0 {
 		ret, err := s.workflowRunRepo.DeleteWhere(
-			context.Background(),
+			ctx,
 			dbx.NewExp(fmt.Sprintf("status!='%s'", string(domain.WorkflowRunStatusTypePending))),
 			dbx.NewExp(fmt.Sprintf("status!='%s'", string(domain.WorkflowRunStatusTypeProcessing))),
 			dbx.NewExp(fmt.Sprintf("endedAt<DATETIME('now', '-%d days')", persistenceSettings.WorkflowRunsMaxDaysRetention)),
 		)
 		if err != nil {
-			app.GetLogger().Error(fmt.Sprintf("failed to delete workflow history runs: %w", err))
+			app.GetLogger().Error("failed to delete workflow history runs", slog.Any("error", err))
 			return err
 		}
 

internal/workflow/service_deps.go

@@ -16,6 +16,7 @@ type workflowRepository interface {
 
 type workflowRunRepository interface {
 	GetById(ctx context.Context, id string) (*domain.WorkflowRun, error)
+	Save(ctx context.Context, workflowRun *domain.WorkflowRun) (*domain.WorkflowRun, error)
 	SaveWithCascading(ctx context.Context, workflowRun *domain.WorkflowRun) (*domain.WorkflowRun, error)
 	DeleteWhere(ctx context.Context, exprs ...dbx.Expression) (int, error)
 }

main.go

@@ -68,6 +68,6 @@ func main() {
 	})
 
 	if err := app.Start(); err != nil {
-		slog.Error("[CERTIMATE] Start failed.", "err", err)
+		slog.Error("[CERTIMATE] Start failed.", slog.Any("error", err))
 	}
 }

ui/src/components/workflow/designer/Designer.tsx

@@ -2,6 +2,7 @@ import { forwardRef, useEffect, useImperativeHandle, useMemo, useRef } from "rea
 import {
   ConstantKeys,
   EditorRenderer,
+  EditorState,
   FixedLayoutEditorProvider,
   type FixedLayoutPluginContext,
   type FixedLayoutProps,
@@ -119,6 +120,13 @@ const Designer = forwardRef<DesignerInstance, DesignerProps>(
           }),
         ],
 
+        onInit: (ctx) => {
+          const maybeMobile = ["android", "ios", "iphone", "ipad", "micromessenger"].some((s) => navigator.userAgent.includes(s));
+          if (maybeMobile) {
+            ctx.playground.editorState.changeState(EditorState.STATE_MOUSE_FRIENDLY_SELECT.id);
+          }
+        },
+
         onAllLayersRendered: (ctx) => {
           rendered.current = true;
 

ui/src/components/workflow/designer/Toolbar.tsx

@@ -29,9 +29,9 @@ const Toolbar = ({ className, style }: ToolbarProps) => {
     return () => d.dispose();
   }, [playground]);
 
-  const [isMinimapVisible, setIsMinimapVisible] = useState(window.screen.availWidth >= 1024);
+  const [isMinimapVisible, setIsMinimapVisible] = useState(() => window.screen.availWidth >= 1024);
 
-  const [isMouseFriendly, setIsMouseFriendly] = useState(playground.editorState.is(EditorState.STATE_MOUSE_FRIENDLY_SELECT.id));
+  const [isMouseFriendly, setIsMouseFriendly] = useState(() => playground.editorState.is(EditorState.STATE_MOUSE_FRIENDLY_SELECT.id));
 
   const handleToggleLayout = useCallback(() => {
     if (tools.isVertical) {

ui/src/i18n/locales/en/nls.workflow.runs.json

@@ -1,9 +1,9 @@
 {
   "workflow_run.action.view.menu": "View details",
-  "workflow_run.action.cancel.menu": "Cancel workflow run",
-  "workflow_run.action.cancel.modal.title": "Cancel run",
+  "workflow_run.action.cancel.menu": "Cancel run",
+  "workflow_run.action.cancel.modal.title": "Cancel workflow run",
   "workflow_run.action.cancel.modal.content": "Are you sure to cancel this run?",
-  "workflow_run.action.delete.menu": "Delete workflow run",
+  "workflow_run.action.delete.menu": "Delete run",
   "workflow_run.action.delete.modal.title": "Delete \"{{name}}\"",
   "workflow_run.action.delete.modal.content": "Are you sure want to delete this workflow run? <br>This action cannot be undone.",
   "workflow_run.action.batch_delete.modal.title": "Delete workflow runs",

ui/src/pages/ConsoleLayout.tsx

@@ -107,7 +107,7 @@ const ConsoleLayout = () => {
           </div>
           <div className="absolute top-1/2 right-0 translate-x-1/2 -translate-y-1/2 opacity-0 transition-opacity group-hover/sider:opacity-100">
             <Button
-              className="bg-background"
+              className="bg-background shadow"
               icon={
                 siderCollapsed ? (
                   <IconLayoutSidebarRightCollapse size="1.5em" stroke="1.25" color="#999" />

ui/src/pages/workflows/WorkflowDetailRuns.tsx

@@ -120,10 +120,10 @@ const WorkflowDetailRuns = () => {
                 },
                 {
                   key: "cancel",
-                  label: t("workflow_run.action.cancel.menu"),
+                  label: <span style={{ color: cancelDisabled ? void 0 : "var(--color-warning)" }}>{t("workflow_run.action.cancel.menu")}</span>,
                   icon: (
                     <span className="anticon scale-125">
-                      <IconPlayerPause size="1em" />
+                      <IconPlayerPause size="1em" color={cancelDisabled ? void 0 : "var(--color-warning)"} />
                     </span>
                   ),
                   disabled: cancelDisabled,