diff --git a/internal/repository/workflow.go b/internal/repository/workflow.go index 60d60899..baa5e21b 100644 --- a/internal/repository/workflow.go +++ b/internal/repository/workflow.go @@ -24,7 +24,7 @@ func (r *WorkflowRepository) ListEnabledAuto(ctx context.Context) ([]*domain.Wor "enabled={:enabled} && trigger={:trigger}", "-created", 0, 0, - dbx.Params{"enabled": true, "trigger": domain.WorkflowTriggerTypeAuto}, + dbx.Params{"enabled": true, "trigger": string(domain.WorkflowTriggerTypeAuto)}, ) if err != nil { return nil, err diff --git a/internal/scheduler/certificate.go b/internal/scheduler/certificate.go index 26c7311f..43887cb0 100644 --- a/internal/scheduler/certificate.go +++ b/internal/scheduler/certificate.go @@ -6,6 +6,6 @@ type certificateService interface { InitSchedule(ctx context.Context) error } -func NewCertificateScheduler(service certificateService) error { +func InitCertificateScheduler(service certificateService) error { return service.InitSchedule(context.Background()) } diff --git a/internal/scheduler/scheduler.go b/internal/scheduler/scheduler.go index f5029599..91dbf115 100644 --- a/internal/scheduler/scheduler.go +++ b/internal/scheduler/scheduler.go @@ -1,6 +1,7 @@ package scheduler import ( + "github.com/usual2970/certimate/internal/app" "github.com/usual2970/certimate/internal/certificate" "github.com/usual2970/certimate/internal/repository" "github.com/usual2970/certimate/internal/workflow" @@ -14,7 +15,11 @@ func Register() { certificateRepo := repository.NewCertificateRepository() certificateSvc := certificate.NewCertificateService(certificateRepo) - NewCertificateScheduler(certificateSvc) + if err := InitWorkflowScheduler(workflowSvc); err != nil { + app.GetLogger().Error("failed to init workflow scheduler", "err", err) + } - NewWorkflowScheduler(workflowSvc) + if err := InitCertificateScheduler(certificateSvc); err != nil { + app.GetLogger().Error("failed to init certificate scheduler", "err", err) + } } diff --git a/internal/scheduler/workflow.go b/internal/scheduler/workflow.go index 7cb4dfa8..cef4adce 100644 --- a/internal/scheduler/workflow.go +++ b/internal/scheduler/workflow.go @@ -6,6 +6,6 @@ type workflowService interface { InitSchedule(ctx context.Context) error } -func NewWorkflowScheduler(service workflowService) error { +func InitWorkflowScheduler(service workflowService) error { return service.InitSchedule(context.Background()) } diff --git a/internal/workflow/dispatcher/dispatcher.go b/internal/workflow/dispatcher/dispatcher.go index 0ecc0828..0b504c25 100644 --- a/internal/workflow/dispatcher/dispatcher.go +++ b/internal/workflow/dispatcher/dispatcher.go @@ -92,6 +92,7 @@ func (w *WorkflowDispatcher) Dispatch(data *WorkflowWorkerData) { } w.enqueueWorker(data) + select { case w.chWork <- data: default: @@ -138,6 +139,11 @@ func (w *WorkflowDispatcher) Shutdown() { w.queueMutex.Unlock() // 等待所有正在执行的 WorkflowRun 完成 + w.workerMutex.Lock() + for _, worker := range w.workers { + worker.Cancel() + } + w.workerMutex.Unlock() w.wg.Wait() w.workers = make(map[string]*workflowWorker) w.workerIdMap = make(map[string]string) diff --git a/internal/workflow/event.go b/internal/workflow/event.go index fa6b4b1a..0fedd67b 100644 --- a/internal/workflow/event.go +++ b/internal/workflow/event.go @@ -72,7 +72,6 @@ func onWorkflowRecordCreateOrUpdate(ctx context.Context, record *core.Record) er }) }) if err != nil { - app.GetLogger().Error("add cron job failed", "err", err) return fmt.Errorf("add cron job failed: %w", err) } diff --git a/internal/workflow/service.go b/internal/workflow/service.go index 2d0a224d..d2236a8d 100644 --- a/internal/workflow/service.go +++ b/internal/workflow/service.go @@ -48,6 +48,8 @@ func (s *WorkflowService) InitSchedule(ctx context.Context) error { scheduler := app.GetScheduler() for _, workflow := range workflows { + var errs []error + err := scheduler.Add(fmt.Sprintf("workflow#%s", workflow.Id), workflow.TriggerCron, func() { s.StartRun(ctx, &dtos.WorkflowStartRunReq{ WorkflowId: workflow.Id, @@ -55,7 +57,11 @@ func (s *WorkflowService) InitSchedule(ctx context.Context) error { }) }) if err != nil { - return err + errs = append(errs, err) + } + + if len(errs) > 0 { + return errors.Join(errs...) } } diff --git a/ui/src/i18n/locales/en/nls.workflow.nodes.json b/ui/src/i18n/locales/en/nls.workflow.nodes.json index 964633e0..84f0f48e 100644 --- a/ui/src/i18n/locales/en/nls.workflow.nodes.json +++ b/ui/src/i18n/locales/en/nls.workflow.nodes.json @@ -18,7 +18,7 @@ "workflow_node.start.form.trigger_cron.label": "Cron expression", "workflow_node.start.form.trigger_cron.placeholder": "Please enter cron expression", "workflow_node.start.form.trigger_cron.errmsg.invalid": "Please enter a valid cron expression", - "workflow_node.start.form.trigger_cron.tooltip": "Time zone is based on the server.", + "workflow_node.start.form.trigger_cron.tooltip": "Exactly 5 space separated segments. Time zone is based on the server.", "workflow_node.start.form.trigger_cron.extra": "Expected execution time for the last 5 times:", "workflow_node.start.form.trigger_cron.guide": "Tips: If you have multiple workflows, it is recommended to set them to run at multiple times of the day instead of always running at specific times.

Reference links:
1. Let’s Encrypt rate limits
2. Why should my Let’s Encrypt (ACME) client run at a random time?", diff --git a/ui/src/i18n/locales/zh/nls.workflow.nodes.json b/ui/src/i18n/locales/zh/nls.workflow.nodes.json index d132f935..40860e1e 100644 --- a/ui/src/i18n/locales/zh/nls.workflow.nodes.json +++ b/ui/src/i18n/locales/zh/nls.workflow.nodes.json @@ -18,7 +18,7 @@ "workflow_node.start.form.trigger_cron.label": "Cron 表达式", "workflow_node.start.form.trigger_cron.placeholder": "请输入 Cron 表达式", "workflow_node.start.form.trigger_cron.errmsg.invalid": "请输入正确的 Cron 表达式", - "workflow_node.start.form.trigger_cron.tooltip": "支持使用任意值（即 *）、值列表分隔符（即 ,）、值的范围（即 -）、步骤值（即 /）等四种表达式，时区以服务器设置为准。", + "workflow_node.start.form.trigger_cron.tooltip": "五段式表达式，支持使用任意值（即 *）、值列表分隔符（即 ,）、值的范围（即 -）、步骤值（即 /）等四种表达式。时区以服务器设置为准。", "workflow_node.start.form.trigger_cron.extra": "预计最近 5 次执行时间：", "workflow_node.start.form.trigger_cron.guide": "小贴士：如果你有多个工作流，建议将它们设置为在一天中的多个时间段运行，而非总是在相同的特定时间。

参考链接：
1. Let’s Encrypt 速率限制
2. 为什么我的 Let’s Encrypt (ACME) 客户端启动时间应当随机？", diff --git a/ui/src/utils/cron.ts b/ui/src/utils/cron.ts index c2ca22e2..79a94ddc 100644 --- a/ui/src/utils/cron.ts +++ b/ui/src/utils/cron.ts @@ -3,6 +3,8 @@ export const validCronExpression = (expr: string): boolean => { try { parseExpression(expr); + + if (expr.trim().split(" ").length !== 5) return false; // pocketbase 后端仅支持五段式的表达式 return true; } catch { return false; @@ -10,19 +12,15 @@ export const validCronExpression = (expr: string): boolean => { }; export const getNextCronExecutions = (expr: string, times = 1): Date[] => { - if (!expr) return []; + if (!validCronExpression(expr)) return []; - try { - const now = new Date(); - const cron = parseExpression(expr, { currentDate: now, iterator: true }); + const now = new Date(); + const cron = parseExpression(expr, { currentDate: now, iterator: true }); - const result: Date[] = []; - for (let i = 0; i < times; i++) { - const next = cron.next(); - result.push(next.value.toDate()); - } - return result; - } catch { - return []; + const result: Date[] = []; + for (let i = 0; i < times; i++) { + const next = cron.next(); + result.push(next.value.toDate()); } + return result; };