From d587a91795907dceed17d2325845fe7ca8dd2142 Mon Sep 17 00:00:00 2001 From: Fu Diwei Date: Thu, 22 Jan 2026 21:21:16 +0800 Subject: [PATCH] refactor: clean code --- internal/workflow/engine/context.go | 4 ++++ internal/workflow/engine/engine.go | 2 +- internal/workflow/engine/executor_bizapply.go | 16 ++++++++-------- internal/workflow/engine/executor_bizdeploy.go | 8 ++++---- internal/workflow/engine/executor_bizmonitor.go | 7 ++++--- internal/workflow/engine/executor_biznotify.go | 4 ++-- internal/workflow/engine/executor_bizupload.go | 6 +++--- internal/workflow/engine/executor_condition.go | 5 +++-- internal/workflow/engine/executor_delay.go | 2 +- internal/workflow/engine/executor_trycatch.go | 9 +++++---- 10 files changed, 35 insertions(+), 28 deletions(-) diff --git a/internal/workflow/engine/context.go b/internal/workflow/engine/context.go index 64c43872e..0bb4e0e87 100644 --- a/internal/workflow/engine/context.go +++ b/internal/workflow/engine/context.go @@ -43,6 +43,10 @@ func (c *WorkflowContext) SetContext(ctx context.Context) *WorkflowContext { return c } +func (c *WorkflowContext) Context() context.Context { + return c.ctx +} + func (c *WorkflowContext) Clone() *WorkflowContext { return &WorkflowContext{ WorkflowId: c.WorkflowId, diff --git a/internal/workflow/engine/engine.go b/internal/workflow/engine/engine.go index 6c398dc76..86d8d8fc2 100644 --- a/internal/workflow/engine/engine.go +++ b/internal/workflow/engine/engine.go @@ -210,7 +210,7 @@ func (we *workflowEngine) executeNode(wfCtx *WorkflowContext, node *Node) error } }) } - if _, err := we.wfoutputRepo.Save(execCtx.ctx, output); err != nil { + if _, err := we.wfoutputRepo.Save(execCtx.Context(), output); err != nil { we.syslog.Error("failed to save node output", slog.Any("error", err)) } } diff --git a/internal/workflow/engine/executor_bizapply.go b/internal/workflow/engine/executor_bizapply.go index bd796d632..41e304038 100644 --- a/internal/workflow/engine/executor_bizapply.go +++ b/internal/workflow/engine/executor_bizapply.go @@ -118,7 +118,7 @@ func (ne *bizApplyNodeExecutor) Execute(execCtx *NodeExecutionContext) (*NodeExe WorkflowNodeId: execCtx.Node.Id, } certificate.PopulateFromPEM(obtainResp.FullChainCertificate, obtainResp.PrivateKey) - if certificate, err := ne.certificateRepo.Save(execCtx.ctx, certificate); err != nil { + if certificate, err := ne.certificateRepo.Save(execCtx.Context(), certificate); err != nil { ne.logger.Warn("could not save certificate") return execRes, err } else { @@ -128,7 +128,7 @@ func (ne *bizApplyNodeExecutor) Execute(execCtx *NodeExecutionContext) (*NodeExe // 保存 ARI 替换状态 if lastCertificate != nil && obtainResp.ARIReplaced { lastCertificate.IsRenewed = true - ne.certificateRepo.Save(execCtx.ctx, lastCertificate) + ne.certificateRepo.Save(execCtx.Context(), lastCertificate) } // 节点输出 @@ -140,13 +140,13 @@ func (ne *bizApplyNodeExecutor) Execute(execCtx *NodeExecutionContext) (*NodeExe } func (ne *bizApplyNodeExecutor) getLastOutputArtifacts(execCtx *NodeExecutionContext) (*domain.WorkflowOutput, *domain.Certificate, error) { - lastOutput, err := ne.wfoutputRepo.GetByWorkflowIdAndNodeId(execCtx.ctx, execCtx.WorkflowId, execCtx.Node.Id) + lastOutput, err := ne.wfoutputRepo.GetByWorkflowIdAndNodeId(execCtx.Context(), execCtx.WorkflowId, execCtx.Node.Id) if err != nil && !domain.IsRecordNotFoundError(err) { return nil, nil, fmt.Errorf("failed to get last output record of node #%s: %w", execCtx.Node.Id, err) } if lastOutput != nil { - lastCertificate, err := ne.certificateRepo.GetByWorkflowRunIdAndNodeId(execCtx.ctx, lastOutput.RunId, lastOutput.NodeId) + lastCertificate, err := ne.certificateRepo.GetByWorkflowRunIdAndNodeId(execCtx.Context(), lastOutput.RunId, lastOutput.NodeId) if err != nil && !domain.IsRecordNotFoundError(err) { return lastOutput, nil, fmt.Errorf("failed to get last certificate record of node #%s: %w", execCtx.Node.Id, err) } @@ -264,7 +264,7 @@ func (ne *bizApplyNodeExecutor) executeObtain(execCtx *NodeExecutionContext, nod // 读取质询提供商授权 providerAccessConfig := make(map[string]any) if nodeCfg.ProviderAccessId != "" { - if access, err := ne.accessRepo.GetById(execCtx.ctx, nodeCfg.ProviderAccessId); err != nil { + if access, err := ne.accessRepo.GetById(execCtx.Context(), nodeCfg.ProviderAccessId); err != nil { return nil, fmt.Errorf("failed to get access #%s record: %w", nodeCfg.ProviderAccessId, err) } else { providerAccessConfig = access.Config @@ -274,7 +274,7 @@ func (ne *bizApplyNodeExecutor) executeObtain(execCtx *NodeExecutionContext, nod // 读取证书颁发机构授权 caAccessConfig := make(map[string]any) if nodeCfg.CAProviderAccessId != "" { - if access, err := ne.accessRepo.GetById(execCtx.ctx, nodeCfg.CAProviderAccessId); err != nil { + if access, err := ne.accessRepo.GetById(execCtx.Context(), nodeCfg.CAProviderAccessId); err != nil { return nil, fmt.Errorf("failed to get access #%s record: %w", nodeCfg.CAProviderAccessId, err) } else { caAccessConfig = access.Config @@ -393,7 +393,7 @@ func (ne *bizApplyNodeExecutor) executeObtain(execCtx *NodeExecutionContext, nod } msender := mproc.NewSender[InData, OutData]("certapply", ne.logger) - moutput, err := msender.SendWithContext(execCtx.ctx, &InData{ + moutput, err := msender.SendWithContext(execCtx.Context(), &InData{ Account: legoUser, Request: obtainReq, }) @@ -423,7 +423,7 @@ func (ne *bizApplyNodeExecutor) executeObtain(execCtx *NodeExecutionContext, nod } // 执行申请证书请求 - obtainResp, err := legoClient.ObtainCertificate(execCtx.ctx, obtainReq) + obtainResp, err := legoClient.ObtainCertificate(execCtx.Context(), obtainReq) if err != nil { ne.logger.Warn("could not obtain certificate") return nil, err diff --git a/internal/workflow/engine/executor_bizdeploy.go b/internal/workflow/engine/executor_bizdeploy.go index 95d427c01..921750664 100644 --- a/internal/workflow/engine/executor_bizdeploy.go +++ b/internal/workflow/engine/executor_bizdeploy.go @@ -48,7 +48,7 @@ func (ne *bizDeployNodeExecutor) Execute(execCtx *NodeExecutionContext) (*NodeEx if inputStateValue, ok := inputState.Value.(string); ok { s := strings.Split(inputStateValue, "#") if len(s) == 2 { - certificate, err := ne.certificateRepo.GetById(execCtx.ctx, s[1]) + certificate, err := ne.certificateRepo.GetById(execCtx.Context(), s[1]) if err != nil { ne.logger.Warn("could not get input certificate") return execRes, err @@ -81,7 +81,7 @@ func (ne *bizDeployNodeExecutor) Execute(execCtx *NodeExecutionContext) (*NodeEx // 读取部署提供商授权 providerAccessConfig := make(map[string]any) if nodeCfg.ProviderAccessId != "" { - if access, err := ne.accessRepo.GetById(execCtx.ctx, nodeCfg.ProviderAccessId); err != nil { + if access, err := ne.accessRepo.GetById(execCtx.Context(), nodeCfg.ProviderAccessId); err != nil { return nil, fmt.Errorf("failed to get access #%s record: %w", nodeCfg.ProviderAccessId, err) } else { providerAccessConfig = access.Config @@ -97,7 +97,7 @@ func (ne *bizDeployNodeExecutor) Execute(execCtx *NodeExecutionContext) (*NodeEx Certificate: inputCertificate.Certificate, PrivateKey: inputCertificate.PrivateKey, } - if _, err := deployer.DeployCertificate(execCtx.ctx, deployReq); err != nil { + if _, err := deployer.DeployCertificate(execCtx.Context(), deployReq); err != nil { ne.logger.Warn("could not deploy certificate") return execRes, err } @@ -110,7 +110,7 @@ func (ne *bizDeployNodeExecutor) Execute(execCtx *NodeExecutionContext) (*NodeEx } func (ne *bizDeployNodeExecutor) getLastOutputArtifacts(execCtx *NodeExecutionContext) (*domain.WorkflowOutput, error) { - lastOutput, err := ne.wfoutputRepo.GetByWorkflowIdAndNodeId(execCtx.ctx, execCtx.WorkflowId, execCtx.Node.Id) + lastOutput, err := ne.wfoutputRepo.GetByWorkflowIdAndNodeId(execCtx.Context(), execCtx.WorkflowId, execCtx.Node.Id) if err != nil && !domain.IsRecordNotFoundError(err) { return nil, fmt.Errorf("failed to get last output record of node #%s: %w", execCtx.Node.Id, err) } diff --git a/internal/workflow/engine/executor_bizmonitor.go b/internal/workflow/engine/executor_bizmonitor.go index e0eb67e95..3006a777c 100644 --- a/internal/workflow/engine/executor_bizmonitor.go +++ b/internal/workflow/engine/executor_bizmonitor.go @@ -60,9 +60,10 @@ func (ne *bizMonitorNodeExecutor) Execute(execCtx *NodeExecutionContext) (*NodeE if attempt > 0 { ne.logger.Info(fmt.Sprintf("retry %d time(s) ...", attempt)) + ctx := execCtx.Context() select { - case <-execCtx.ctx.Done(): - return execRes, execCtx.ctx.Err() + case <-ctx.Done(): + return execRes, ctx.Err() case <-time.After(RETRY_INTERVAL): } } @@ -132,7 +133,7 @@ func (ne *bizMonitorNodeExecutor) tryRetrievePeerCertificates(execCtx *NodeExecu } url := fmt.Sprintf("https://%s/%s", addr, strings.TrimLeft(requestPath, "/")) - req, err := http.NewRequestWithContext(execCtx.ctx, http.MethodHead, url, nil) + req, err := http.NewRequestWithContext(execCtx.Context(), http.MethodHead, url, nil) if err != nil { err = fmt.Errorf("failed to create http request: %w", err) ne.logger.Warn(err.Error()) diff --git a/internal/workflow/engine/executor_biznotify.go b/internal/workflow/engine/executor_biznotify.go index db49cdcd9..2ce5384e8 100644 --- a/internal/workflow/engine/executor_biznotify.go +++ b/internal/workflow/engine/executor_biznotify.go @@ -33,7 +33,7 @@ func (ne *bizNotifyNodeExecutor) Execute(execCtx *NodeExecutionContext) (*NodeEx // 读取部署提供商授权 providerAccessConfig := make(map[string]any) if nodeCfg.ProviderAccessId != "" { - if access, err := ne.accessRepo.GetById(execCtx.ctx, nodeCfg.ProviderAccessId); err != nil { + if access, err := ne.accessRepo.GetById(execCtx.Context(), nodeCfg.ProviderAccessId); err != nil { return nil, fmt.Errorf("failed to get access #%s record: %w", nodeCfg.ProviderAccessId, err) } else { providerAccessConfig = access.Config @@ -74,7 +74,7 @@ func (ne *bizNotifyNodeExecutor) Execute(execCtx *NodeExecutionContext) (*NodeEx Subject: subject, Message: message, } - if _, err := notifier.SendNotification(execCtx.ctx, notifyReq); err != nil { + if _, err := notifier.SendNotification(execCtx.Context(), notifyReq); err != nil { ne.logger.Warn("could not send notification") return execRes, err } diff --git a/internal/workflow/engine/executor_bizupload.go b/internal/workflow/engine/executor_bizupload.go index c8ca789d8..4f5007ba2 100644 --- a/internal/workflow/engine/executor_bizupload.go +++ b/internal/workflow/engine/executor_bizupload.go @@ -181,7 +181,7 @@ func (ne *bizUploadNodeExecutor) Execute(execCtx *NodeExecutionContext) (*NodeEx WorkflowNodeId: execCtx.Node.Id, } certificate.PopulateFromPEM(certPEM, privkeyPEM) - if certificate, err := ne.certificateRepo.Save(execCtx.ctx, certificate); err != nil { + if certificate, err := ne.certificateRepo.Save(execCtx.Context(), certificate); err != nil { ne.logger.Warn("could not save certificate") return execRes, err } else { @@ -197,13 +197,13 @@ func (ne *bizUploadNodeExecutor) Execute(execCtx *NodeExecutionContext) (*NodeEx } func (ne *bizUploadNodeExecutor) getLastOutputArtifacts(execCtx *NodeExecutionContext) (*domain.WorkflowOutput, *domain.Certificate, error) { - lastOutput, err := ne.wfoutputRepo.GetByWorkflowIdAndNodeId(execCtx.ctx, execCtx.WorkflowId, execCtx.Node.Id) + lastOutput, err := ne.wfoutputRepo.GetByWorkflowIdAndNodeId(execCtx.Context(), execCtx.WorkflowId, execCtx.Node.Id) if err != nil && !domain.IsRecordNotFoundError(err) { return nil, nil, fmt.Errorf("failed to get last output record of node #%s: %w", execCtx.Node.Id, err) } if lastOutput != nil { - lastCertificate, err := ne.certificateRepo.GetByWorkflowRunIdAndNodeId(execCtx.ctx, lastOutput.RunId, lastOutput.NodeId) + lastCertificate, err := ne.certificateRepo.GetByWorkflowRunIdAndNodeId(execCtx.Context(), lastOutput.RunId, lastOutput.NodeId) if err != nil && !domain.IsRecordNotFoundError(err) { return lastOutput, nil, fmt.Errorf("failed to get last certificate record of node #%s: %w", execCtx.Node.Id, err) } diff --git a/internal/workflow/engine/executor_condition.go b/internal/workflow/engine/executor_condition.go index e1aabc6a0..4d9cdbcef 100644 --- a/internal/workflow/engine/executor_condition.go +++ b/internal/workflow/engine/executor_condition.go @@ -25,9 +25,10 @@ func (ne *conditionNodeExecutor) Execute(execCtx *NodeExecutionContext) (*NodeEx errs := make([]error, 0) blocks := lo.Filter(execCtx.Node.Blocks, func(n *Node, _ int) bool { return n.Type == NodeTypeBranchBlock }) for _, node := range blocks { + ctx := execCtx.Context() select { - case <-execCtx.ctx.Done(): - return execRes, execCtx.ctx.Err() + case <-ctx.Done(): + return execRes, ctx.Err() default: } diff --git a/internal/workflow/engine/executor_delay.go b/internal/workflow/engine/executor_delay.go index 93de77cba..3eed39398 100644 --- a/internal/workflow/engine/executor_delay.go +++ b/internal/workflow/engine/executor_delay.go @@ -18,7 +18,7 @@ func (ne *delayNodeExecutor) Execute(execCtx *NodeExecutionContext) (*NodeExecut nodeCfg := execCtx.Node.Data.Config.AsDelay() ne.logger.Info(fmt.Sprintf("delay for %d second(s) before continuing ...", nodeCfg.Wait)) - xwait.DelayWithContext(execCtx.ctx, time.Duration(nodeCfg.Wait)*time.Second) + xwait.DelayWithContext(execCtx.Context(), time.Duration(nodeCfg.Wait)*time.Second) return execRes, nil } diff --git a/internal/workflow/engine/executor_trycatch.go b/internal/workflow/engine/executor_trycatch.go index 1f3023679..3a24d384c 100644 --- a/internal/workflow/engine/executor_trycatch.go +++ b/internal/workflow/engine/executor_trycatch.go @@ -25,9 +25,10 @@ func (ne *tryCatchNodeExecutor) Execute(execCtx *NodeExecutionContext) (*NodeExe tryErrs := make([]error, 0) tryBlocks := lo.Filter(execCtx.Node.Blocks, func(n *Node, _ int) bool { return n.Type == NodeTypeTryBlock }) for _, node := range tryBlocks { + ctx := execCtx.Context() select { - case <-execCtx.ctx.Done(): - return execRes, execCtx.ctx.Err() + case <-ctx.Done(): + return execRes, ctx.Err() default: } @@ -45,8 +46,8 @@ func (ne *tryCatchNodeExecutor) Execute(execCtx *NodeExecutionContext) (*NodeExe catchBlocks := lo.Filter(execCtx.Node.Blocks, func(n *Node, _ int) bool { return n.Type == NodeTypeCatchBlock }) for _, node := range catchBlocks { select { - case <-execCtx.ctx.Done(): - return execRes, execCtx.ctx.Err() + case <-execCtx.Context().Done(): + return execRes, execCtx.Context().Err() default: }