opensource/certimate
3
0
Fork 0
mirror of https://github.com/certimate-go/certimate.git synced 2026-09-24 23:10:13 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor: clean code

Browse Source
This commit is contained in:
Fu Diwei
2026-01-22 21:21:16 +08:00
parent db347bb1b8
commit d587a91795
10 changed files with 35 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/workflow/engine/context.go
+4
View File
@@ -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,
internal/workflow/engine/engine.go
+1 -1
View File
@@ -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))
}
}
internal/workflow/engine/executor_bizapply.go
+8 -8
View File
@@ -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
internal/workflow/engine/executor_bizdeploy.go
+4 -4
View File
@@ -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)
}
internal/workflow/engine/executor_bizmonitor.go
+4 -3
View File
@@ -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())
internal/workflow/engine/executor_biznotify.go
+2 -2
View File
@@ -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
}
internal/workflow/engine/executor_bizupload.go
+3 -3
View File
@@ -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)
}
internal/workflow/engine/executor_condition.go
+3 -2
View File
@@ -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:
}
internal/workflow/engine/executor_delay.go
+1 -1
View File
@@ -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
}
internal/workflow/engine/executor_trycatch.go
+5 -4
View File
@@ -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:
}