From d3b9c8ff5539f9f4810e665e73c2afde68e04e25 Mon Sep 17 00:00:00 2001 From: Fu Diwei Date: Thu, 21 Aug 2025 12:21:52 +0800 Subject: [PATCH] refactor: clean code --- internal/workflow/engine/context.go | 139 +----------------- internal/workflow/engine/engine.go | 4 +- internal/workflow/engine/executor.go | 24 +-- internal/workflow/engine/executor_bizapply.go | 8 +- .../workflow/engine/executor_bizdeploy.go | 4 +- .../workflow/engine/executor_bizmonitor.go | 8 +- .../workflow/engine/executor_biznotify.go | 2 +- .../workflow/engine/executor_bizupload.go | 8 +- .../workflow/engine/executor_condition.go | 2 +- internal/workflow/engine/models.go | 8 - internal/workflow/engine/state.go | 139 ++++++++++++++++++ 11 files changed, 175 insertions(+), 171 deletions(-) create mode 100644 internal/workflow/engine/state.go diff --git a/internal/workflow/engine/context.go b/internal/workflow/engine/context.go index 36b3ac875..06797d621 100644 --- a/internal/workflow/engine/context.go +++ b/internal/workflow/engine/context.go @@ -2,8 +2,6 @@ import ( "context" - "slices" - "sync" ) type WorkflowContext struct { @@ -12,8 +10,8 @@ type WorkflowContext struct { RunGraph *Graph engine WorkflowEngine - variables WorkflowIOsManager - inputs WorkflowIOsManager + variables StateManager + inputs StateManager ctx context.Context } @@ -30,13 +28,13 @@ func (c *WorkflowContext) SetEngine(engine WorkflowEngine) *WorkflowContext { return c } -func (c *WorkflowContext) SetVariablesManager(variables WorkflowIOsManager) *WorkflowContext { - c.variables = variables +func (c *WorkflowContext) SetVariablesManager(manager StateManager) *WorkflowContext { + c.variables = manager return c } -func (c *WorkflowContext) SetInputsManager(inputs WorkflowIOsManager) *WorkflowContext { - c.inputs = inputs +func (c *WorkflowContext) SetInputsManager(manager StateManager) *WorkflowContext { + c.inputs = manager return c } @@ -58,128 +56,3 @@ func (c *WorkflowContext) Clone() *WorkflowContext { ctx: c.ctx, } } - -type WorkflowIOsManager interface { - All() []NodeIOEntry - Erase() - - Add(entry NodeIOEntry) - Get(name string) (*NodeIOEntry, bool) - GetScoped(scope string, name string) (*NodeIOEntry, bool) - Take(name string) (*NodeIOEntry, bool) - TakeScoped(scope string, name string) (*NodeIOEntry, bool) - Remove(name string) bool - RemoveScoped(scope string, name string) bool -} - -type workflowIOsManager struct { - entriesMtx sync.RWMutex - entries []NodeIOEntry -} - -var _ WorkflowIOsManager = (*workflowIOsManager)(nil) - -func (m *workflowIOsManager) All() []NodeIOEntry { - m.entriesMtx.RLock() - defer m.entriesMtx.RUnlock() - - if m.entries == nil { - return make([]NodeIOEntry, 0) - } - - return slices.Clone(m.entries) -} - -func (m *workflowIOsManager) Erase() { - m.entriesMtx.Lock() - defer m.entriesMtx.Unlock() - - m.entries = make([]NodeIOEntry, 0) -} - -func (m *workflowIOsManager) Add(entry NodeIOEntry) { - m.entriesMtx.Lock() - defer m.entriesMtx.Unlock() - - if m.entries == nil { - m.entries = make([]NodeIOEntry, 0) - } - - for i, item := range m.entries { - if item.Scope == entry.Scope && item.Key == entry.Key { - m.entries[i] = entry - return - } - } - m.entries = append(m.entries, entry) -} - -func (m *workflowIOsManager) Get(name string) (*NodeIOEntry, bool) { - return m.GetScoped("", name) -} - -func (m *workflowIOsManager) GetScoped(scope string, name string) (*NodeIOEntry, bool) { - m.entriesMtx.RLock() - defer m.entriesMtx.RUnlock() - - if m.entries == nil { - return nil, false - } - - for _, item := range m.entries { - if item.Scope == scope && item.Key == name { - return &item, true - } - } - return nil, false -} - -func (m *workflowIOsManager) Take(name string) (*NodeIOEntry, bool) { - return m.TakeScoped("", name) -} - -func (m *workflowIOsManager) TakeScoped(scope, name string) (*NodeIOEntry, bool) { - m.entriesMtx.Lock() - defer m.entriesMtx.Unlock() - - if m.entries == nil { - return nil, false - } - - for i, item := range m.entries { - if item.Scope == scope && item.Key == name { - m.entries = slices.Delete(m.entries, i, i+1) - return &item, true - } - } - return nil, false -} - -func (m *workflowIOsManager) Remove(name string) bool { - return m.RemoveScoped("", name) -} - -func (m *workflowIOsManager) RemoveScoped(scope string, name string) bool { - _, ok := m.TakeScoped(scope, name) - return ok -} - -func newWorkflowIOsManager() WorkflowIOsManager { - return &workflowIOsManager{ - entries: make([]NodeIOEntry, 0), - } -} - -const ( - wfIOTypeCertificate = "certificate" -) - -const ( - wfIOKeyCertificate = "certificate" -) - -const ( - wfVariableKeyCertificateValidity = "certificate.validity" // ValueType: "bool" - wfVariableKeyCertificateDaysLeft = "certificate.daysLeft" // ValueType: "int32" - wfVariableKeyNodeSkipped = "node.skipped" // ValueType: "bool" -) diff --git a/internal/workflow/engine/engine.go b/internal/workflow/engine/engine.go index ab3bbad16..fbd039bf2 100644 --- a/internal/workflow/engine/engine.go +++ b/internal/workflow/engine/engine.go @@ -52,8 +52,8 @@ func (we *workflowEngine) Invoke(ctx context.Context, workflowId string, runId s wfCtx := (&WorkflowContext{}). SetExecutingWorkflow(workflowId, runId, runGraph). SetEngine(we). - SetVariablesManager(newWorkflowIOsManager()). - SetInputsManager(newWorkflowIOsManager()). + SetVariablesManager(newStateManager()). + SetInputsManager(newStateManager()). SetContext(ctx) if err := we.executeBlocks(wfCtx, runGraph.Nodes); err != nil { if !errors.Is(err, errInterrupted) { diff --git a/internal/workflow/engine/executor.go b/internal/workflow/engine/executor.go index b664688fd..763f6e828 100644 --- a/internal/workflow/engine/executor.go +++ b/internal/workflow/engine/executor.go @@ -41,13 +41,13 @@ func (c *NodeExecutionContext) SetEngine(engine WorkflowEngine) *NodeExecutionCo return c } -func (c *NodeExecutionContext) SetVariablesManager(variablesMgr WorkflowIOsManager) *NodeExecutionContext { - c.variables = variablesMgr +func (c *NodeExecutionContext) SetVariablesManager(manager StateManager) *NodeExecutionContext { + c.variables = manager return c } -func (c *NodeExecutionContext) SetInputsManager(inputsMgr WorkflowIOsManager) *NodeExecutionContext { - c.inputs = inputsMgr +func (c *NodeExecutionContext) SetInputsManager(manager StateManager) *NodeExecutionContext { + c.inputs = manager return c } @@ -60,13 +60,13 @@ type NodeExecutionResult struct { Interrupted bool // 是否中断执行（通常由 End 节点主动触发） variablesMtx sync.RWMutex - Variables []NodeIOEntry + Variables []StateEntry outputsMtx sync.RWMutex - Outputs []NodeIOEntry + Outputs []StateEntry } func (r *NodeExecutionResult) AddVariable(scope string, key string, value any, valueType string) { - r.AddVariableEntry(NodeIOEntry{ + r.AddVariableEntry(StateEntry{ Scope: scope, Type: "", Key: key, @@ -75,12 +75,12 @@ func (r *NodeExecutionResult) AddVariable(scope string, key string, value any, v }) } -func (r *NodeExecutionResult) AddVariableEntry(entry NodeIOEntry) { +func (r *NodeExecutionResult) AddVariableEntry(entry StateEntry) { r.variablesMtx.Lock() defer r.variablesMtx.Unlock() if r.Variables == nil { - r.Variables = make([]NodeIOEntry, 0) + r.Variables = make([]StateEntry, 0) } for i, item := range r.Variables { @@ -93,7 +93,7 @@ func (r *NodeExecutionResult) AddVariableEntry(entry NodeIOEntry) { } func (r *NodeExecutionResult) AddOutput(scope string, type_ string, key string, value any, valueType string) { - r.AddOutputEntry(NodeIOEntry{ + r.AddOutputEntry(StateEntry{ Scope: scope, Type: type_, Key: key, @@ -102,12 +102,12 @@ func (r *NodeExecutionResult) AddOutput(scope string, type_ string, key string, }) } -func (r *NodeExecutionResult) AddOutputEntry(entry NodeIOEntry) { +func (r *NodeExecutionResult) AddOutputEntry(entry StateEntry) { r.outputsMtx.Lock() defer r.outputsMtx.Unlock() if r.Outputs == nil { - r.Outputs = make([]NodeIOEntry, 0) + r.Outputs = make([]StateEntry, 0) } for i, t := range r.Outputs { diff --git a/internal/workflow/engine/executor_bizapply.go b/internal/workflow/engine/executor_bizapply.go index 4e8010e85..b70ded109 100644 --- a/internal/workflow/engine/executor_bizapply.go +++ b/internal/workflow/engine/executor_bizapply.go @@ -35,7 +35,7 @@ func (ne *bizApplyNodeExecutor) Execute(execCtx *NodeExecutionContext) (*NodeExe if skippable, reason := ne.checkCanSkip(execCtx, lastOutput, lastCertificate); skippable { ne.logger.Info(fmt.Sprintf("skip this application, because %s", reason)) - execRes.AddVariable(execCtx.Node.Id, wfVariableKeyNodeSkipped, true, "boolean") + execRes.AddVariable(execCtx.Node.Id, stateVariableKeyNodeSkipped, true, "boolean") return execRes, nil } else if reason != "" { ne.logger.Info(fmt.Sprintf("re-apply, because %s", reason)) @@ -109,9 +109,9 @@ func (ne *bizApplyNodeExecutor) Execute(execCtx *NodeExecutionContext) (*NodeExe } // 记录中间结果 - execRes.AddVariable(execCtx.Node.Id, wfVariableKeyNodeSkipped, false, "boolean") - execRes.AddVariable(execCtx.Node.Id, wfVariableKeyCertificateValidity, true, "boolean") - execRes.AddVariable(execCtx.Node.Id, wfVariableKeyCertificateDaysLeft, int32(time.Until(certificate.ValidityNotAfter).Hours()/24), "number") + execRes.AddVariable(execCtx.Node.Id, stateVariableKeyNodeSkipped, false, "boolean") + execRes.AddVariable(execCtx.Node.Id, stateVariableKeyCertificateValidity, true, "boolean") + execRes.AddVariable(execCtx.Node.Id, stateVariableKeyCertificateDaysLeft, int32(time.Until(certificate.ValidityNotAfter).Hours()/24), "number") ne.logger.Info("application completed") return execRes, nil diff --git a/internal/workflow/engine/executor_bizdeploy.go b/internal/workflow/engine/executor_bizdeploy.go index 048e9a34d..d3798738e 100644 --- a/internal/workflow/engine/executor_bizdeploy.go +++ b/internal/workflow/engine/executor_bizdeploy.go @@ -42,7 +42,7 @@ func (ne *bizDeployNodeExecutor) Execute(execCtx *NodeExecutionContext) (*NodeEx if skippable, reason := ne.checkCanSkip(execCtx, lastOutput); skippable { ne.logger.Info(fmt.Sprintf("skip this deployment, because %s", reason)) - execRes.AddVariable(execCtx.Node.Id, wfVariableKeyNodeSkipped, true, "boolean") + execRes.AddVariable(execCtx.Node.Id, stateVariableKeyNodeSkipped, true, "boolean") return execRes, nil } else if reason != "" { ne.logger.Info(fmt.Sprintf("re-deploy, because %s", reason)) @@ -82,7 +82,7 @@ func (ne *bizDeployNodeExecutor) Execute(execCtx *NodeExecutionContext) (*NodeEx } // 记录中间结果 - execRes.AddVariable(execCtx.Node.Id, wfVariableKeyNodeSkipped, false, "boolean") + execRes.AddVariable(execCtx.Node.Id, stateVariableKeyNodeSkipped, false, "boolean") ne.logger.Info("deployment completed") return execRes, nil diff --git a/internal/workflow/engine/executor_bizmonitor.go b/internal/workflow/engine/executor_bizmonitor.go index 2e96aa08c..14bd2cc76 100644 --- a/internal/workflow/engine/executor_bizmonitor.go +++ b/internal/workflow/engine/executor_bizmonitor.go @@ -68,8 +68,8 @@ func (ne *bizMonitorNodeExecutor) Execute(execCtx *NodeExecutionContext) (*NodeE if len(certs) == 0 { ne.logger.Warn("no ssl certificates retrieved in http response") - execRes.AddVariable(execCtx.Node.Id, wfVariableKeyCertificateValidity, false, "boolean") - execRes.AddVariable(execCtx.Node.Id, wfVariableKeyCertificateDaysLeft, 0, "number") + execRes.AddVariable(execCtx.Node.Id, stateVariableKeyCertificateValidity, false, "boolean") + execRes.AddVariable(execCtx.Node.Id, stateVariableKeyCertificateDaysLeft, 0, "number") } else { cert := certs[0] // 只取证书链中的第一个证书，即服务器证书 ne.logger.Info(fmt.Sprintf("ssl certificate retrieved (serial='%s', subject='%s', issuer='%s', not_before='%s', not_after='%s', sans='%s')", @@ -87,8 +87,8 @@ func (ne *bizMonitorNodeExecutor) Execute(execCtx *NodeExecutionContext) (*NodeE validated := isCertPeriodValid && isCertHostMatched daysLeft := int(math.Floor(cert.NotAfter.Sub(now).Hours() / 24)) - execRes.AddVariable(execCtx.Node.Id, wfVariableKeyCertificateValidity, validated, "boolean") - execRes.AddVariable(execCtx.Node.Id, wfVariableKeyCertificateDaysLeft, daysLeft, "number") + execRes.AddVariable(execCtx.Node.Id, stateVariableKeyCertificateValidity, validated, "boolean") + execRes.AddVariable(execCtx.Node.Id, stateVariableKeyCertificateDaysLeft, daysLeft, "number") if validated { ne.logger.Info(fmt.Sprintf("the certificate is valid, and will expire in %d day(s)", daysLeft)) diff --git a/internal/workflow/engine/executor_biznotify.go b/internal/workflow/engine/executor_biznotify.go index af629f30b..8cd872c3e 100644 --- a/internal/workflow/engine/executor_biznotify.go +++ b/internal/workflow/engine/executor_biznotify.go @@ -56,7 +56,7 @@ func (ne *bizNotifyNodeExecutor) checkCanSkip(execCtx *NodeExecutionContext) (_s var total, skipped int32 for _, variable := range execCtx.variables.All() { - if variable.Scope != "" && variable.Key == wfVariableKeyNodeSkipped { + if variable.Scope != "" && variable.Key == stateVariableKeyNodeSkipped { total++ if variable.Value == true { skipped++ diff --git a/internal/workflow/engine/executor_bizupload.go b/internal/workflow/engine/executor_bizupload.go index e3afdc77f..80dce6fe4 100644 --- a/internal/workflow/engine/executor_bizupload.go +++ b/internal/workflow/engine/executor_bizupload.go @@ -33,7 +33,7 @@ func (ne *bizUploadNodeExecutor) Execute(execCtx *NodeExecutionContext) (*NodeEx if skippable, reason := ne.checkCanSkip(execCtx, lastOutput, lastCertificate); skippable { ne.logger.Info(fmt.Sprintf("skip this uploading, because %s", reason)) - execRes.AddVariable(execCtx.Node.Id, wfVariableKeyNodeSkipped, true, "boolean") + execRes.AddVariable(execCtx.Node.Id, stateVariableKeyNodeSkipped, true, "boolean") return execRes, nil } else if reason != "" { ne.logger.Info(fmt.Sprintf("re-upload, because %s", reason)) @@ -70,9 +70,9 @@ func (ne *bizUploadNodeExecutor) Execute(execCtx *NodeExecutionContext) (*NodeEx } // 记录中间结果 - execRes.AddVariable(execCtx.Node.Id, wfVariableKeyNodeSkipped, false, "boolean") - execRes.AddVariable(execCtx.Node.Id, wfVariableKeyCertificateValidity, true, "boolean") - execRes.AddVariable(execCtx.Node.Id, wfVariableKeyCertificateDaysLeft, int32(time.Until(certificate.ValidityNotAfter).Hours()/24), "number") + execRes.AddVariable(execCtx.Node.Id, stateVariableKeyNodeSkipped, false, "boolean") + execRes.AddVariable(execCtx.Node.Id, stateVariableKeyCertificateValidity, true, "boolean") + execRes.AddVariable(execCtx.Node.Id, stateVariableKeyCertificateDaysLeft, int32(time.Until(certificate.ValidityNotAfter).Hours()/24), "number") ne.logger.Info("uploading completed") return execRes, nil diff --git a/internal/workflow/engine/executor_condition.go b/internal/workflow/engine/executor_condition.go index f03de4c92..01aca9528 100644 --- a/internal/workflow/engine/executor_condition.go +++ b/internal/workflow/engine/executor_condition.go @@ -65,7 +65,7 @@ func (ne *branchBlockNodeExecutor) Execute(execCtx *NodeExecutionContext) (*Node if nodeCfg.Expression == nil { ne.logger.Info("enter this branch without any conditions") } else { - variables := lo.Reduce(execCtx.variables.All(), func(acc map[string]map[string]any, entry NodeIOEntry, _ int) map[string]map[string]any { + variables := lo.Reduce(execCtx.variables.All(), func(acc map[string]map[string]any, entry StateEntry, _ int) map[string]map[string]any { if _, ok := acc[entry.Scope]; !ok { acc[entry.Scope] = make(map[string]any) } diff --git a/internal/workflow/engine/models.go b/internal/workflow/engine/models.go index 63765f26c..079b676c5 100644 --- a/internal/workflow/engine/models.go +++ b/internal/workflow/engine/models.go @@ -23,12 +23,4 @@ const ( NodeTypeBizNotify = domain.WorkflowNodeTypeBizNotify ) -type NodeIOEntry struct { - Scope string // 零值时表示全局的，否则表示指定节点的 - Type string // 仅表示输入输出有值，表示变量无值 - Key string - Value any - ValueType string `options:"string | number | boolean"` -} - type Graph = domain.WorkflowGraph diff --git a/internal/workflow/engine/state.go b/internal/workflow/engine/state.go new file mode 100644 index 000000000..3274a4f49 --- /dev/null +++ b/internal/workflow/engine/state.go @@ -0,0 +1,139 @@ +package engine + +import ( + "slices" + "sync" +) + +type StateEntry struct { + Scope string // 零值时表示全局的，否则表示指定节点的 + Type string // 仅表示输入输出有值，表示变量无值 + Key string + Value any + ValueType string `options:"string | number | boolean"` +} + +type StateManager interface { + All() []StateEntry + Erase() + + Add(entry StateEntry) + Get(name string) (*StateEntry, bool) + GetScoped(scope string, name string) (*StateEntry, bool) + Take(name string) (*StateEntry, bool) + TakeScoped(scope string, name string) (*StateEntry, bool) + Remove(name string) bool + RemoveScoped(scope string, name string) bool +} + +type stateManager struct { + entriesMtx sync.RWMutex + entries []StateEntry +} + +var _ StateManager = (*stateManager)(nil) + +func (m *stateManager) All() []StateEntry { + m.entriesMtx.RLock() + defer m.entriesMtx.RUnlock() + + if m.entries == nil { + return make([]StateEntry, 0) + } + + return slices.Clone(m.entries) +} + +func (m *stateManager) Erase() { + m.entriesMtx.Lock() + defer m.entriesMtx.Unlock() + + m.entries = make([]StateEntry, 0) +} + +func (m *stateManager) Add(entry StateEntry) { + m.entriesMtx.Lock() + defer m.entriesMtx.Unlock() + + if m.entries == nil { + m.entries = make([]StateEntry, 0) + } + + for i, item := range m.entries { + if item.Scope == entry.Scope && item.Key == entry.Key { + m.entries[i] = entry + return + } + } + m.entries = append(m.entries, entry) +} + +func (m *stateManager) Get(name string) (*StateEntry, bool) { + return m.GetScoped("", name) +} + +func (m *stateManager) GetScoped(scope string, name string) (*StateEntry, bool) { + m.entriesMtx.RLock() + defer m.entriesMtx.RUnlock() + + if m.entries == nil { + return nil, false + } + + for _, item := range m.entries { + if item.Scope == scope && item.Key == name { + return &item, true + } + } + return nil, false +} + +func (m *stateManager) Take(name string) (*StateEntry, bool) { + return m.TakeScoped("", name) +} + +func (m *stateManager) TakeScoped(scope, name string) (*StateEntry, bool) { + m.entriesMtx.Lock() + defer m.entriesMtx.Unlock() + + if m.entries == nil { + return nil, false + } + + for i, item := range m.entries { + if item.Scope == scope && item.Key == name { + m.entries = slices.Delete(m.entries, i, i+1) + return &item, true + } + } + return nil, false +} + +func (m *stateManager) Remove(name string) bool { + return m.RemoveScoped("", name) +} + +func (m *stateManager) RemoveScoped(scope string, name string) bool { + _, ok := m.TakeScoped(scope, name) + return ok +} + +func newStateManager() StateManager { + return &stateManager{ + entries: make([]StateEntry, 0), + } +} + +const ( + stateIOTypeCertificate = "certificate" +) + +const ( + stateIOKeyCertificate = "certificate" +) + +const ( + stateVariableKeyCertificateValidity = "certificate.validity" // ValueType: "bool" + stateVariableKeyCertificateDaysLeft = "certificate.daysLeft" // ValueType: "int32" + stateVariableKeyNodeSkipped = "node.skipped" // ValueType: "bool" +)