diff --git a/internal/applicant/applicant.go b/internal/applicant/applicant.go index 49a56d0b6..2240df4f0 100644 --- a/internal/applicant/applicant.go +++ b/internal/applicant/applicant.go @@ -54,7 +54,7 @@ func NewWithWorkflowNode(config ApplicantWithWorkflowNodeConfig) (Applicant, err return nil, fmt.Errorf("the node type is '%s', expected '%s'", string(config.Node.Type), string(domain.WorkflowNodeTypeBizApply)) } - nodeCfg := config.Node.GetConfigForBizApply() + nodeCfg := config.Node.Data.Config.AsBizApply() options := &applicantProviderOptions{ Domains: lo.Filter(strings.Split(nodeCfg.Domains, ";"), func(s string, _ int) bool { return s != "" }), ContactEmail: nodeCfg.ContactEmail, diff --git a/internal/deployer/deployer.go b/internal/deployer/deployer.go index fc1463cbc..cec8e38f0 100644 --- a/internal/deployer/deployer.go +++ b/internal/deployer/deployer.go @@ -29,7 +29,7 @@ func NewWithWorkflowNode(config DeployerWithWorkflowNodeConfig) (Deployer, error return nil, fmt.Errorf("the node type is '%s', expected '%s'", string(config.Node.Type), string(domain.WorkflowNodeTypeBizDeploy)) } - nodeCfg := config.Node.GetConfigForBizDeploy() + nodeCfg := config.Node.Data.Config.AsBizDeploy() options := &deployerProviderOptions{ Provider: domain.DeploymentProviderType(nodeCfg.Provider), ProviderAccessConfig: make(map[string]any), diff --git a/internal/domain/workflow.go b/internal/domain/workflow.go index 008ba59c7..877c29349 100644 --- a/internal/domain/workflow.go +++ b/internal/domain/workflow.go @@ -95,8 +95,89 @@ const ( ) type WorkflowNodeData struct { - Name string `json:"name"` - Config map[string]any `json:"config"` + Name string `json:"name"` + Config WorkflowNodeConfig `json:"config"` +} + +type WorkflowNodeConfig map[string]any + +func (c WorkflowNodeConfig) AsBizApply() WorkflowNodeConfigForBizApply { + return WorkflowNodeConfigForBizApply{ + Domains: xmaps.GetString(c, "domains"), + ContactEmail: xmaps.GetString(c, "contactEmail"), + ChallengeType: xmaps.GetString(c, "challengeType"), + Provider: xmaps.GetString(c, "provider"), + ProviderAccessId: xmaps.GetString(c, "providerAccessId"), + ProviderConfig: xmaps.GetKVMapAny(c, "providerConfig"), + KeyAlgorithm: xmaps.GetOrDefaultString(c, "keyAlgorithm", string(CertificateKeyAlgorithmTypeRSA2048)), + CAProvider: xmaps.GetString(c, "caProvider"), + CAProviderAccessId: xmaps.GetString(c, "caProviderAccessId"), + CAProviderConfig: xmaps.GetKVMapAny(c, "caProviderConfig"), + ACMEProfile: xmaps.GetString(c, "acmeProfile"), + Nameservers: xmaps.GetString(c, "nameservers"), + DnsPropagationWait: xmaps.GetInt32(c, "dnsPropagationWait"), + DnsPropagationTimeout: xmaps.GetInt32(c, "dnsPropagationTimeout"), + DnsTTL: xmaps.GetInt32(c, "dnsTTL"), + DisableFollowCNAME: xmaps.GetBool(c, "disableFollowCNAME"), + DisableARI: xmaps.GetBool(c, "disableARI"), + SkipBeforeExpiryDays: xmaps.GetInt32(c, "skipBeforeExpiryDays"), + } +} + +func (c WorkflowNodeConfig) AsBizUpload() WorkflowNodeConfigForBizUpload { + return WorkflowNodeConfigForBizUpload{ + Certificate: xmaps.GetString(c, "certificate"), + PrivateKey: xmaps.GetString(c, "privateKey"), + Domains: xmaps.GetString(c, "domains"), + } +} + +func (c WorkflowNodeConfig) AsBizMonitor() WorkflowNodeConfigForBizMonitor { + host := xmaps.GetString(c, "host") + return WorkflowNodeConfigForBizMonitor{ + Host: host, + Port: xmaps.GetOrDefaultInt32(c, "port", 443), + Domain: xmaps.GetOrDefaultString(c, "domain", host), + RequestPath: xmaps.GetString(c, "path"), + } +} + +func (c WorkflowNodeConfig) AsBizDeploy() WorkflowNodeConfigForBizDeploy { + return WorkflowNodeConfigForBizDeploy{ + CertificateOutputNodeId: xmaps.GetString(c, "certificateOutputNodeId"), + Provider: xmaps.GetString(c, "provider"), + ProviderAccessId: xmaps.GetString(c, "providerAccessId"), + ProviderConfig: xmaps.GetKVMapAny(c, "providerConfig"), + SkipOnLastSucceeded: xmaps.GetBool(c, "skipOnLastSucceeded"), + } +} + +func (c WorkflowNodeConfig) AsBizNotify() WorkflowNodeConfigForBizNotify { + return WorkflowNodeConfigForBizNotify{ + Provider: xmaps.GetString(c, "provider"), + ProviderAccessId: xmaps.GetString(c, "providerAccessId"), + ProviderConfig: xmaps.GetKVMapAny(c, "providerConfig"), + Subject: xmaps.GetString(c, "subject"), + Message: xmaps.GetString(c, "message"), + SkipOnAllPrevSkipped: xmaps.GetBool(c, "skipOnAllPrevSkipped"), + } +} + +func (c WorkflowNodeConfig) AsBranchBlock() WorkflowNodeConfigForBranchBlock { + expression := c["expression"] + if expression == nil { + return WorkflowNodeConfigForBranchBlock{} + } + + exprRaw, _ := json.Marshal(expression) + expr, err := expr.UnmarshalExpr([]byte(exprRaw)) + if err != nil { + return WorkflowNodeConfigForBranchBlock{} + } + + return WorkflowNodeConfigForBranchBlock{ + Expression: expr, + } } type WorkflowNodeConfigForBizApply struct { @@ -153,82 +234,3 @@ type WorkflowNodeConfigForBizNotify struct { type WorkflowNodeConfigForBranchBlock struct { Expression expr.Expr `json:"expression"` // 条件表达式 } - -func (n *WorkflowNode) GetConfigForBizApply() WorkflowNodeConfigForBizApply { - return WorkflowNodeConfigForBizApply{ - Domains: xmaps.GetString(n.Data.Config, "domains"), - ContactEmail: xmaps.GetString(n.Data.Config, "contactEmail"), - ChallengeType: xmaps.GetString(n.Data.Config, "challengeType"), - Provider: xmaps.GetString(n.Data.Config, "provider"), - ProviderAccessId: xmaps.GetString(n.Data.Config, "providerAccessId"), - ProviderConfig: xmaps.GetKVMapAny(n.Data.Config, "providerConfig"), - KeyAlgorithm: xmaps.GetOrDefaultString(n.Data.Config, "keyAlgorithm", string(CertificateKeyAlgorithmTypeRSA2048)), - CAProvider: xmaps.GetString(n.Data.Config, "caProvider"), - CAProviderAccessId: xmaps.GetString(n.Data.Config, "caProviderAccessId"), - CAProviderConfig: xmaps.GetKVMapAny(n.Data.Config, "caProviderConfig"), - ACMEProfile: xmaps.GetString(n.Data.Config, "acmeProfile"), - Nameservers: xmaps.GetString(n.Data.Config, "nameservers"), - DnsPropagationWait: xmaps.GetInt32(n.Data.Config, "dnsPropagationWait"), - DnsPropagationTimeout: xmaps.GetInt32(n.Data.Config, "dnsPropagationTimeout"), - DnsTTL: xmaps.GetInt32(n.Data.Config, "dnsTTL"), - DisableFollowCNAME: xmaps.GetBool(n.Data.Config, "disableFollowCNAME"), - DisableARI: xmaps.GetBool(n.Data.Config, "disableARI"), - SkipBeforeExpiryDays: xmaps.GetInt32(n.Data.Config, "skipBeforeExpiryDays"), - } -} - -func (n *WorkflowNode) GetConfigForBizUpload() WorkflowNodeConfigForBizUpload { - return WorkflowNodeConfigForBizUpload{ - Certificate: xmaps.GetString(n.Data.Config, "certificate"), - PrivateKey: xmaps.GetString(n.Data.Config, "privateKey"), - Domains: xmaps.GetString(n.Data.Config, "domains"), - } -} - -func (n *WorkflowNode) GetConfigForBizMonitor() WorkflowNodeConfigForBizMonitor { - host := xmaps.GetString(n.Data.Config, "host") - return WorkflowNodeConfigForBizMonitor{ - Host: host, - Port: xmaps.GetOrDefaultInt32(n.Data.Config, "port", 443), - Domain: xmaps.GetOrDefaultString(n.Data.Config, "domain", host), - RequestPath: xmaps.GetString(n.Data.Config, "path"), - } -} - -func (n *WorkflowNode) GetConfigForBizDeploy() WorkflowNodeConfigForBizDeploy { - return WorkflowNodeConfigForBizDeploy{ - CertificateOutputNodeId: xmaps.GetString(n.Data.Config, "certificateOutputNodeId"), - Provider: xmaps.GetString(n.Data.Config, "provider"), - ProviderAccessId: xmaps.GetString(n.Data.Config, "providerAccessId"), - ProviderConfig: xmaps.GetKVMapAny(n.Data.Config, "providerConfig"), - SkipOnLastSucceeded: xmaps.GetBool(n.Data.Config, "skipOnLastSucceeded"), - } -} - -func (n *WorkflowNode) GetConfigForBizNotify() WorkflowNodeConfigForBizNotify { - return WorkflowNodeConfigForBizNotify{ - Provider: xmaps.GetString(n.Data.Config, "provider"), - ProviderAccessId: xmaps.GetString(n.Data.Config, "providerAccessId"), - ProviderConfig: xmaps.GetKVMapAny(n.Data.Config, "providerConfig"), - Subject: xmaps.GetString(n.Data.Config, "subject"), - Message: xmaps.GetString(n.Data.Config, "message"), - SkipOnAllPrevSkipped: xmaps.GetBool(n.Data.Config, "skipOnAllPrevSkipped"), - } -} - -func (n *WorkflowNode) GetConfigForBranchBlock() WorkflowNodeConfigForBranchBlock { - expression := n.Data.Config["expression"] - if expression == nil { - return WorkflowNodeConfigForBranchBlock{} - } - - exprRaw, _ := json.Marshal(expression) - expr, err := expr.UnmarshalExpr([]byte(exprRaw)) - if err != nil { - return WorkflowNodeConfigForBranchBlock{} - } - - return WorkflowNodeConfigForBranchBlock{ - Expression: expr, - } -} diff --git a/internal/domain/workflow_output.go b/internal/domain/workflow_output.go index 2108f73b6..28933832e 100644 --- a/internal/domain/workflow_output.go +++ b/internal/domain/workflow_output.go @@ -7,6 +7,7 @@ type WorkflowOutput struct { WorkflowId string `json:"workflowId" db:"workflowRef"` RunId string `json:"runId" db:"runRef"` NodeId string `json:"nodeId" db:"nodeId"` + NodeConfig WorkflowNodeConfig `json:"nodeConfig" db:"nodeConfig"` Outputs []*WorkflowOutputEntry `json:"outputs" db:"outputs"` Succeeded bool `json:"succeeded" db:"succeeded"` } diff --git a/internal/notify/notifier.go b/internal/notify/notifier.go index 533432d6f..8220ac9c0 100644 --- a/internal/notify/notifier.go +++ b/internal/notify/notifier.go @@ -29,7 +29,7 @@ func NewWithWorkflowNode(config NotifierWithWorkflowNodeConfig) (Notifier, error return nil, fmt.Errorf("the node type is '%s', expected '%s'", string(config.Node.Type), string(domain.WorkflowNodeTypeBizNotify)) } - nodeCfg := config.Node.GetConfigForBizNotify() + nodeCfg := config.Node.Data.Config.AsBizNotify() options := ¬ifierProviderOptions{ Provider: domain.NotificationProviderType(nodeCfg.Provider), ProviderAccessConfig: make(map[string]any), diff --git a/internal/repository/workflow_output.go b/internal/repository/workflow_output.go index 3330bf49b..f1b4b0f24 100644 --- a/internal/repository/workflow_output.go +++ b/internal/repository/workflow_output.go @@ -107,6 +107,11 @@ func (r *WorkflowOutputRepository) castRecordToModel(record *core.Record) (*doma return nil, fmt.Errorf("the record is nil") } + nodeConfig := make(domain.WorkflowNodeConfig) + if err := record.UnmarshalJSONField("nodeConfig", &nodeConfig); err != nil { + return nil, fmt.Errorf("field 'nodeConfig' is malformed") + } + outputs := make([]*domain.WorkflowOutputEntry, 0) if err := record.UnmarshalJSONField("outputs", &outputs); err != nil { return nil, fmt.Errorf("field 'outputs' is malformed") @@ -121,6 +126,7 @@ func (r *WorkflowOutputRepository) castRecordToModel(record *core.Record) (*doma WorkflowId: record.GetString("workflowRef"), RunId: record.GetString("runRef"), NodeId: record.GetString("nodeId"), + NodeConfig: nodeConfig, Outputs: outputs, Succeeded: record.GetBool("succeeded"), } @@ -145,6 +151,7 @@ func (r *WorkflowOutputRepository) saveRecord(workflowOutput *domain.WorkflowOut record.Set("workflowRef", workflowOutput.WorkflowId) record.Set("runRef", workflowOutput.RunId) record.Set("nodeId", workflowOutput.NodeId) + record.Set("nodeConfig", workflowOutput.NodeConfig) record.Set("outputs", workflowOutput.Outputs) record.Set("succeeded", workflowOutput.Succeeded) if err := app.GetApp().Save(record); err != nil { diff --git a/internal/workflow/engine/engine.go b/internal/workflow/engine/engine.go index 660e1ff91..ab3bbad16 100644 --- a/internal/workflow/engine/engine.go +++ b/internal/workflow/engine/engine.go @@ -24,6 +24,8 @@ type WorkflowEngine interface { } type workflowEngine struct { + logger *slog.Logger + executorRegistry map[NodeType]NodeExecutor hooksMtx sync.RWMutex @@ -184,7 +186,7 @@ func (we *workflowEngine) fireOnStartHooks(ctx context.Context) { defer we.hooksMtx.RUnlock() for _, cb := range we.onStartHooks { if cbErr := cb(ctx); cbErr != nil { - app.GetLogger().Error("workflow engine: error in onStart hook", slog.Any("error", cbErr)) + we.logger.Error("workflow engine: error in onStart hook", slog.Any("error", cbErr)) } } } @@ -194,7 +196,7 @@ func (we *workflowEngine) fireOnEndHooks(ctx context.Context) { defer we.hooksMtx.RUnlock() for _, cb := range we.onEndHooks { if cbErr := cb(ctx); cbErr != nil { - app.GetLogger().Error("workflow engine: error in onEnd hook", slog.Any("error", cbErr)) + we.logger.Error("workflow engine: error in onEnd hook", slog.Any("error", cbErr)) } } } @@ -204,7 +206,7 @@ func (we *workflowEngine) fireOnErrorHooks(ctx context.Context, err error) { defer we.hooksMtx.RUnlock() for _, cb := range we.onErrorHooks { if cbErr := cb(ctx, err); cbErr != nil { - app.GetLogger().Error("workflow engine: error in onError hook", slog.Any("error", cbErr)) + we.logger.Error("workflow engine: error in onError hook", slog.Any("error", cbErr)) } } } @@ -214,7 +216,7 @@ func (we *workflowEngine) fireOnNodeStartHooks(ctx context.Context, node *Node) defer we.hooksMtx.RUnlock() for _, cb := range we.onNodeStartHooks { if cbErr := cb(ctx, node); cbErr != nil { - app.GetLogger().Error("workflow engine: error in onNodeStart hook", slog.Any("error", cbErr)) + we.logger.Error("workflow engine: error in onNodeStart hook", slog.Any("error", cbErr)) } } } @@ -224,7 +226,7 @@ func (we *workflowEngine) fireOnNodeEndHooks(ctx context.Context, node *Node, re defer we.hooksMtx.RUnlock() for _, cb := range we.onNodeEndHooks { if cbErr := cb(ctx, node, result); cbErr != nil { - app.GetLogger().Error("workflow engine: error in onNodeEnd hook", slog.Any("error", cbErr)) + we.logger.Error("workflow engine: error in onNodeEnd hook", slog.Any("error", cbErr)) } } } @@ -234,7 +236,7 @@ func (we *workflowEngine) fireOnNodeErrorHooks(ctx context.Context, node *Node, defer we.hooksMtx.RUnlock() for _, cb := range we.onNodeErrorHooks { if cbErr := cb(ctx, node, err); cbErr != nil { - app.GetLogger().Error("workflow engine: error in onNodeError hook", slog.Any("error", cbErr)) + we.logger.Error("workflow engine: error in onNodeError hook", slog.Any("error", cbErr)) } } } @@ -244,7 +246,7 @@ func (we *workflowEngine) fireOnNodeLoggingHooks(ctx context.Context, node *Node defer we.hooksMtx.RUnlock() for _, cb := range we.onNodeLoggingHooks { if cbErr := cb(ctx, node, log); cbErr != nil { - app.GetLogger().Error("workflow engine: error in onNodeLogging hook", slog.Any("error", cbErr)) + we.logger.Error("workflow engine: error in onNodeLogging hook", slog.Any("error", cbErr)) } } } @@ -252,6 +254,7 @@ func (we *workflowEngine) fireOnNodeLoggingHooks(ctx context.Context, node *Node func NewWorkflowEngine() WorkflowEngine { engine := &workflowEngine{ executorRegistry: make(map[NodeType]NodeExecutor), + logger: app.GetLogger(), } engine.executorRegistry[NodeTypeStart] = newStartNodeExecutor() engine.executorRegistry[NodeTypeEnd] = newEndNodeExecutor() diff --git a/internal/workflow/engine/executor_bizapply.go b/internal/workflow/engine/executor_bizapply.go index 27ecf433d..4e8010e85 100644 --- a/internal/workflow/engine/executor_bizapply.go +++ b/internal/workflow/engine/executor_bizapply.go @@ -16,14 +16,13 @@ type bizApplyNodeExecutor struct { nodeExecutor certificateRepo certificateRepository - wfrunRepo workflowRunRepository wfoutputRepo workflowOutputRepository } func (ne *bizApplyNodeExecutor) Execute(execCtx *NodeExecutionContext) (*NodeExecutionResult, error) { execRes := &NodeExecutionResult{} - nodeCfg := execCtx.Node.GetConfigForBizApply() + nodeCfg := execCtx.Node.Data.Config.AsBizApply() ne.logger.Info("ready to request certificate ...", slog.Any("config", nodeCfg)) // 查询上次执行结果 @@ -87,6 +86,7 @@ func (ne *bizApplyNodeExecutor) Execute(execCtx *NodeExecutionContext) (*NodeExe WorkflowId: execCtx.WorkflowId, RunId: execCtx.RunId, NodeId: execCtx.Node.Id, + NodeConfig: execCtx.Node.Data.Config, Succeeded: true, Outputs: []*domain.WorkflowOutputEntry{ { @@ -136,20 +136,11 @@ func (ne *bizApplyNodeExecutor) getLastOutputArtifacts(execCtx *NodeExecutionCon } func (ne *bizApplyNodeExecutor) checkCanSkip(execCtx *NodeExecutionContext, lastOutput *domain.WorkflowOutput, lastCertificate *domain.Certificate) (_skip bool, _reason string) { - thisNodeCfg := execCtx.Node.GetConfigForBizApply() + thisNodeCfg := execCtx.Node.Data.Config.AsBizApply() if lastOutput != nil && lastOutput.Succeeded { - lastRun, err := ne.wfrunRepo.GetById(execCtx.ctx, lastOutput.RunId) - if err != nil { - return true, "failed to get last run" - } - lastNode, lastNodeExists := lastRun.Graph.GetNodeById(lastOutput.NodeId) - if !lastNodeExists { - return true, "failed to get last run node" - } - // 比较和上次申请时的关键配置（即影响证书签发的）参数是否一致 - lastNodeCfg := lastNode.GetConfigForBizApply() + lastNodeCfg := lastOutput.NodeConfig.AsBizApply() if thisNodeCfg.Domains != lastNodeCfg.Domains { return false, "the configuration item 'Domains' changed" @@ -201,7 +192,6 @@ func newBizApplyNodeExecutor() NodeExecutor { return &bizApplyNodeExecutor{ nodeExecutor: nodeExecutor{logger: slog.Default()}, certificateRepo: repository.NewCertificateRepository(), - wfrunRepo: repository.NewWorkflowRunRepository(), wfoutputRepo: repository.NewWorkflowOutputRepository(), } } diff --git a/internal/workflow/engine/executor_bizdeploy.go b/internal/workflow/engine/executor_bizdeploy.go index 30274e239..048e9a34d 100644 --- a/internal/workflow/engine/executor_bizdeploy.go +++ b/internal/workflow/engine/executor_bizdeploy.go @@ -14,14 +14,13 @@ type bizDeployNodeExecutor struct { nodeExecutor certificateRepo certificateRepository - wfrunRepo workflowRunRepository wfoutputRepo workflowOutputRepository } func (ne *bizDeployNodeExecutor) Execute(execCtx *NodeExecutionContext) (*NodeExecutionResult, error) { execRes := &NodeExecutionResult{} - nodeCfg := execCtx.Node.GetConfigForBizDeploy() + nodeCfg := execCtx.Node.Data.Config.AsBizDeploy() ne.logger.Info("ready to deploy certificate ...", slog.Any("config", nodeCfg)) // 查询上次执行结果 @@ -74,6 +73,7 @@ func (ne *bizDeployNodeExecutor) Execute(execCtx *NodeExecutionContext) (*NodeEx WorkflowId: execCtx.WorkflowId, RunId: execCtx.RunId, NodeId: execCtx.Node.Id, + NodeConfig: execCtx.Node.Data.Config, Succeeded: true, } if _, err := ne.wfoutputRepo.Save(execCtx.ctx, output); err != nil { @@ -98,20 +98,11 @@ func (ne *bizDeployNodeExecutor) getLastOutputArtifacts(execCtx *NodeExecutionCo } func (ne *bizDeployNodeExecutor) checkCanSkip(execCtx *NodeExecutionContext, lastOutput *domain.WorkflowOutput) (_skip bool, _reason string) { - thisNodeCfg := execCtx.Node.GetConfigForBizDeploy() + thisNodeCfg := execCtx.Node.Data.Config.AsBizDeploy() if lastOutput != nil && lastOutput.Succeeded { - lastRun, err := ne.wfrunRepo.GetById(execCtx.ctx, lastOutput.RunId) - if err != nil { - return true, "failed to get last run" - } - lastNode, lastNodeExists := lastRun.Graph.GetNodeById(lastOutput.NodeId) - if !lastNodeExists { - return true, "failed to get last run node" - } - // 比较和上次部署时的关键配置（即影响证书部署的）参数是否一致 - lastNodeCfg := lastNode.GetConfigForBizDeploy() + lastNodeCfg := lastOutput.NodeConfig.AsBizDeploy() if thisNodeCfg.ProviderAccessId != lastNodeCfg.ProviderAccessId { return false, "the configuration item 'ProviderAccessId' changed" @@ -132,7 +123,6 @@ func newBizDeployNodeExecutor() NodeExecutor { return &bizDeployNodeExecutor{ nodeExecutor: nodeExecutor{logger: slog.Default()}, certificateRepo: repository.NewCertificateRepository(), - wfrunRepo: repository.NewWorkflowRunRepository(), wfoutputRepo: repository.NewWorkflowOutputRepository(), } } diff --git a/internal/workflow/engine/executor_bizmonitor.go b/internal/workflow/engine/executor_bizmonitor.go index 78de2d954..2e96aa08c 100644 --- a/internal/workflow/engine/executor_bizmonitor.go +++ b/internal/workflow/engine/executor_bizmonitor.go @@ -25,7 +25,7 @@ type bizMonitorNodeExecutor struct { func (ne *bizMonitorNodeExecutor) Execute(execCtx *NodeExecutionContext) (*NodeExecutionResult, error) { execRes := &NodeExecutionResult{} - nodeCfg := execCtx.Node.GetConfigForBizMonitor() + nodeCfg := execCtx.Node.Data.Config.AsBizMonitor() ne.logger.Info("ready to monitor certificate ...", slog.Any("config", nodeCfg)) targetAddr := net.JoinHostPort(nodeCfg.Host, strconv.Itoa(int(nodeCfg.Port))) diff --git a/internal/workflow/engine/executor_biznotify.go b/internal/workflow/engine/executor_biznotify.go index 194831f23..af629f30b 100644 --- a/internal/workflow/engine/executor_biznotify.go +++ b/internal/workflow/engine/executor_biznotify.go @@ -17,7 +17,7 @@ type bizNotifyNodeExecutor struct { func (ne *bizNotifyNodeExecutor) Execute(execCtx *NodeExecutionContext) (*NodeExecutionResult, error) { execRes := &NodeExecutionResult{} - nodeCfg := execCtx.Node.GetConfigForBizNotify() + nodeCfg := execCtx.Node.Data.Config.AsBizNotify() ne.logger.Info("ready to send notification ...", slog.Any("config", nodeCfg)) // 检测是否可以跳过本次执行 @@ -49,7 +49,7 @@ func (ne *bizNotifyNodeExecutor) Execute(execCtx *NodeExecutionContext) (*NodeEx } func (ne *bizNotifyNodeExecutor) checkCanSkip(execCtx *NodeExecutionContext) (_skip bool) { - thisNodeCfg := execCtx.Node.GetConfigForBizNotify() + thisNodeCfg := execCtx.Node.Data.Config.AsBizNotify() if !thisNodeCfg.SkipOnAllPrevSkipped { return false } diff --git a/internal/workflow/engine/executor_bizupload.go b/internal/workflow/engine/executor_bizupload.go index d6def222d..e3afdc77f 100644 --- a/internal/workflow/engine/executor_bizupload.go +++ b/internal/workflow/engine/executor_bizupload.go @@ -14,14 +14,13 @@ type bizUploadNodeExecutor struct { nodeExecutor certificateRepo certificateRepository - wfrunRepo workflowRunRepository wfoutputRepo workflowOutputRepository } func (ne *bizUploadNodeExecutor) Execute(execCtx *NodeExecutionContext) (*NodeExecutionResult, error) { execRes := &NodeExecutionResult{} - nodeCfg := execCtx.Node.GetConfigForBizUpload() + nodeCfg := execCtx.Node.Data.Config.AsBizUpload() ne.logger.Info("ready to upload certiticate ...", slog.Any("config", nodeCfg)) // 查询上次执行结果 @@ -54,6 +53,7 @@ func (ne *bizUploadNodeExecutor) Execute(execCtx *NodeExecutionContext) (*NodeEx WorkflowId: execCtx.WorkflowId, RunId: execCtx.RunId, NodeId: execCtx.Node.Id, + NodeConfig: execCtx.Node.Data.Config, Succeeded: true, Outputs: []*domain.WorkflowOutputEntry{ { @@ -97,20 +97,11 @@ func (ne *bizUploadNodeExecutor) getLastOutputArtifacts(execCtx *NodeExecutionCo } func (ne *bizUploadNodeExecutor) checkCanSkip(execCtx *NodeExecutionContext, lastOutput *domain.WorkflowOutput, lastCertificate *domain.Certificate) (_skip bool, _reason string) { - thisNodeCfg := execCtx.Node.GetConfigForBizUpload() + thisNodeCfg := execCtx.Node.Data.Config.AsBizUpload() if lastOutput != nil && lastOutput.Succeeded { - lastRun, err := ne.wfrunRepo.GetById(execCtx.ctx, lastOutput.RunId) - if err != nil { - return true, "failed to get last run" - } - lastNode, lastNodeExists := lastRun.Graph.GetNodeById(lastOutput.NodeId) - if !lastNodeExists { - return true, "failed to get last run node" - } - // 比较和上次上传时的关键配置（即影响证书上传的）参数是否一致 - lastNodeCfg := lastNode.GetConfigForBizUpload() + lastNodeCfg := lastOutput.NodeConfig.AsBizUpload() if strings.TrimSpace(thisNodeCfg.Certificate) != strings.TrimSpace(lastNodeCfg.Certificate) { return false, "the configuration item 'Certificate' changed" @@ -135,7 +126,6 @@ func newBizUploadNodeExecutor() NodeExecutor { return &bizUploadNodeExecutor{ nodeExecutor: nodeExecutor{logger: slog.Default()}, certificateRepo: repository.NewCertificateRepository(), - wfrunRepo: repository.NewWorkflowRunRepository(), wfoutputRepo: repository.NewWorkflowOutputRepository(), } } diff --git a/internal/workflow/engine/executor_condition.go b/internal/workflow/engine/executor_condition.go index adcb3e7d4..f03de4c92 100644 --- a/internal/workflow/engine/executor_condition.go +++ b/internal/workflow/engine/executor_condition.go @@ -42,7 +42,7 @@ func (ne *conditionNodeExecutor) Execute(execCtx *NodeExecutionContext) (*NodeEx } if len(errs) > 0 { - return execRes, errors.Join(errs...) + return execRes, fmt.Errorf("error occurred when executing child nodes: %w", errors.Join(errs...)) } return execRes, nil @@ -61,7 +61,7 @@ type branchBlockNodeExecutor struct { func (ne *branchBlockNodeExecutor) Execute(execCtx *NodeExecutionContext) (*NodeExecutionResult, error) { execRes := &NodeExecutionResult{} - nodeCfg := execCtx.Node.GetConfigForBranchBlock() + nodeCfg := execCtx.Node.Data.Config.AsBranchBlock() if nodeCfg.Expression == nil { ne.logger.Info("enter this branch without any conditions") } else { diff --git a/internal/workflow/engine/executor_trycatch.go b/internal/workflow/engine/executor_trycatch.go index f562981eb..be0753b6b 100644 --- a/internal/workflow/engine/executor_trycatch.go +++ b/internal/workflow/engine/executor_trycatch.go @@ -2,6 +2,7 @@ import ( "errors" + "fmt" "log/slog" "github.com/samber/lo" @@ -59,10 +60,10 @@ func (ne *tryCatchNodeExecutor) Execute(execCtx *NodeExecutionContext) (*NodeExe } if len(catchErrs) > 0 { - return execRes, errors.Join(append(tryErrs, catchErrs...)...) + return execRes, fmt.Errorf("error occurred when executing child nodes: %w", errors.Join(append(tryErrs, catchErrs...)...)) } - return execRes, errors.Join(tryErrs...) + return execRes, fmt.Errorf("error occurred when executing child nodes: %w", errors.Join(tryErrs...)) } return execRes, nil diff --git a/migrations/1755187200_v0.4.0_migrate.go b/migrations/1755187200_v0.4.0_migrate.go index f62e91ba4..b87ebdfbc 100644 --- a/migrations/1755187200_v0.4.0_migrate.go +++ b/migrations/1755187200_v0.4.0_migrate.go @@ -556,7 +556,7 @@ func init() { // update collection `workflow_output` // - rename field `workflowId` to `workflowRef` // - rename field `runId` to `runRef` - // - delete field `node` + // - rename field `node` to `nodeConfig` { collection, err := app.FindCollectionByNameOrId("bqnxb95f2cooowp") if err != nil { @@ -604,8 +604,17 @@ func init() { return err } - if field := collection.Fields.GetByName("node"); field != nil { - collection.Fields.RemoveById(field.GetId()) + if err := collection.Fields.AddMarshaledJSONAt(4, []byte(`{ + "hidden": false, + "id": "json2239752261", + "maxSize": 5000000, + "name": "nodeConfig", + "presentable": false, + "required": false, + "system": false, + "type": "json" + }`)); err != nil { + return err } if err := app.Save(collection); err != nil { @@ -1049,6 +1058,7 @@ func init() { } // update collection `workflow_output` + // - migrate field `nodeConfig` // - migrate field `outputs` { collection, err := app.FindCollectionByNameOrId("bqnxb95f2cooowp") @@ -1062,6 +1072,18 @@ func init() { for _, record := range records { changed := false + nodeConfig := make(map[string]any) + if err := record.UnmarshalJSONField("nodeConfig", &nodeConfig); err == nil { + if _, ok := nodeConfig["id"]; ok { + if _, ok := nodeConfig["type"]; ok { + if _, ok := nodeConfig["config"]; ok { + record.Set("nodeConfig", nodeConfig["config"]) + changed = true + } + } + } + } + outputs := make([]map[string]any, 0) if err := record.UnmarshalJSONField("outputs", &outputs); err == nil { for i, output := range outputs { @@ -1077,8 +1099,6 @@ func init() { record.Set("outputs", outputs) changed = true } - } else { - println(err.Error()) } if changed {