sz_rust_workflow/engine/
instance.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2024-2026 SZ-Rust Team
3//
4use std::sync::Arc;
5
6use chrono::Utc;
7use uuid::Uuid;
8
9use crate::error::{WorkflowError, WorkflowErrorCode, WorkflowResult};
10use crate::instance::{FlowInstance, InstanceStatus, PageRequest, PageResult, Task};
11use crate::integration::SensitiveFieldRegistry;
12use crate::observability::{AuditAction, AuditLogger, WorkflowEvent, WorkflowEventBus};
13use crate::repository::{DefinitionRepository, InstanceRepository};
14use crate::scheduling::candidate_resolver::CandidateResolver;
15
16use super::history::HistoryRecorder;
17use super::task_manager::TaskManager;
18
19/// 实例摘要。
20#[derive(Debug, Clone, serde::Serialize)]
21pub struct InstanceSummary {
22    pub instance_id: String,
23    pub flow_key: String,
24    pub status: InstanceStatus,
25    pub current_nodes: Vec<String>,
26}
27
28/// 实例详情。
29#[derive(Debug, Clone, serde::Serialize)]
30pub struct InstanceDetail {
31    pub instance: FlowInstance,
32    pub current_tasks: Vec<Task>,
33}
34
35/// 实例管理器。
36pub struct InstanceManager {
37    instance_repo: Arc<dyn InstanceRepository>,
38    definition_repo: Arc<dyn DefinitionRepository>,
39    task_manager: Arc<TaskManager>,
40    candidate_resolver: Arc<dyn CandidateResolver>,
41    history_recorder: Arc<HistoryRecorder>,
42    audit: Arc<AuditLogger>,
43    event_bus: Arc<dyn WorkflowEventBus>,
44    #[allow(dead_code)] // 保留：实例历史脱敏预留
45    sensitive_registry: Arc<SensitiveFieldRegistry>,
46}
47
48impl InstanceManager {
49    #[allow(clippy::too_many_arguments)]
50    pub fn new(
51        instance_repo: Arc<dyn InstanceRepository>,
52        definition_repo: Arc<dyn DefinitionRepository>,
53        task_manager: Arc<TaskManager>,
54        candidate_resolver: Arc<dyn CandidateResolver>,
55        history_recorder: Arc<HistoryRecorder>,
56        audit: Arc<AuditLogger>,
57        event_bus: Arc<dyn WorkflowEventBus>,
58        #[allow(dead_code)] // 保留：实例历史脱敏预留
59        sensitive_registry: Arc<SensitiveFieldRegistry>,
60    ) -> Self {
61        Self {
62            instance_repo,
63            definition_repo,
64            task_manager,
65            candidate_resolver,
66            history_recorder,
67            audit,
68            event_bus,
69            sensitive_registry,
70        }
71    }
72
73    /// 启动实例。
74    pub async fn start(
75        &self,
76        flow_key: &str,
77        context: serde_json::Value,
78        initiator: &str,
79    ) -> WorkflowResult<InstanceSummary> {
80        let def = self
81            .definition_repo
82            .get_active(flow_key)
83            .await?
84            .ok_or_else(|| {
85                WorkflowError::with_field(
86                    WorkflowErrorCode::DefinitionNotFound,
87                    "流程定义不存在或无生效版本",
88                    "flow_key",
89                    flow_key,
90                )
91            })?;
92
93        let instance_id = Uuid::new_v4().to_string();
94        let instance = FlowInstance::new(
95            instance_id.clone(),
96            flow_key,
97            def.version.clone(),
98            initiator,
99            context,
100            &def.start_node,
101        );
102        self.instance_repo.create(&instance).await?;
103
104        self.history_recorder
105            .record_node_event(
106                &instance_id,
107                None,
108                &def.start_node,
109                crate::instance::HistoryEntryType::NodeEnter,
110            )
111            .await
112            .ok();
113
114        if let Some(start_node) = def.find_node(&def.start_node) {
115            if let crate::definition::NodeConfig::Start { next } = &start_node.config {
116                if let Some(next_node) = def.find_node(next) {
117                    if let crate::definition::NodeConfig::Approval {
118                        candidate_strategy, ..
119                    } = &next_node.config
120                    {
121                        let candidates = self
122                            .candidate_resolver
123                            .resolve(candidate_strategy, &instance.context)
124                            .await?;
125                        self.task_manager
126                            .create_tasks(&instance_id, next, candidates)
127                            .await
128                            .ok();
129                    }
130                }
131            }
132        }
133
134        self.event_bus
135            .publish(WorkflowEvent::InstanceStarted {
136                instance_id: instance_id.clone(),
137                flow_key: flow_key.into(),
138                initiator: initiator.into(),
139                timestamp: Utc::now(),
140            })
141            .await
142            .ok();
143
144        self.audit.log_action(
145            AuditAction::Start,
146            initiator,
147            &instance_id,
148            InstanceStatus::Created,
149            InstanceStatus::Running,
150            serde_json::json!({"flow_key": flow_key}),
151        );
152
153        Ok(InstanceSummary {
154            instance_id,
155            flow_key: flow_key.into(),
156            status: InstanceStatus::Running,
157            current_nodes: instance.current_nodes.clone(),
158        })
159    }
160
161    /// 挂起实例。
162    pub async fn suspend(&self, instance_id: &str, actor: &str) -> WorkflowResult<()> {
163        let instance = self.load_instance(instance_id).await?;
164        if instance.status != InstanceStatus::Running {
165            return Err(WorkflowError::with_field(
166                WorkflowErrorCode::IllegalStatusTransition,
167                format!("非 running 状态不可挂起：{}", instance.status),
168                "status",
169                &instance.status.to_string(),
170            ));
171        }
172        let mut updated = instance.clone();
173        updated.status = InstanceStatus::Suspended;
174        updated.bump_version();
175        self.save_with_lock(&updated, instance.version_lock).await?;
176
177        self.event_bus
178            .publish(WorkflowEvent::InstanceSuspended {
179                instance_id: instance_id.into(),
180                actor: actor.into(),
181                timestamp: Utc::now(),
182            })
183            .await
184            .ok();
185        self.audit.log_action(
186            AuditAction::Suspend,
187            actor,
188            instance_id,
189            InstanceStatus::Running,
190            InstanceStatus::Suspended,
191            serde_json::json!({}),
192        );
193        Ok(())
194    }
195
196    /// 恢复实例。
197    pub async fn resume(&self, instance_id: &str, actor: &str) -> WorkflowResult<()> {
198        let instance = self.load_instance(instance_id).await?;
199        if instance.status != InstanceStatus::Suspended {
200            return Err(WorkflowError::with_field(
201                WorkflowErrorCode::IllegalStatusTransition,
202                format!("非 suspended 状态不可恢复：{}", instance.status),
203                "status",
204                &instance.status.to_string(),
205            ));
206        }
207        let mut updated = instance.clone();
208        updated.status = InstanceStatus::Running;
209        updated.bump_version();
210        self.save_with_lock(&updated, instance.version_lock).await?;
211
212        self.event_bus
213            .publish(WorkflowEvent::InstanceResumed {
214                instance_id: instance_id.into(),
215                actor: actor.into(),
216                timestamp: Utc::now(),
217            })
218            .await
219            .ok();
220        self.audit.log_action(
221            AuditAction::Resume,
222            actor,
223            instance_id,
224            InstanceStatus::Suspended,
225            InstanceStatus::Running,
226            serde_json::json!({}),
227        );
228        Ok(())
229    }
230
231    /// 终止实例。
232    pub async fn terminate(&self, instance_id: &str, actor: &str) -> WorkflowResult<()> {
233        let instance = self.load_instance(instance_id).await?;
234        let mut updated = instance.clone();
235        updated.status = InstanceStatus::Terminated;
236        updated.bump_version();
237        self.save_with_lock(&updated, instance.version_lock).await?;
238        self.task_manager
239            .invalidate_by_instance(instance_id)
240            .await?;
241
242        self.event_bus
243            .publish(WorkflowEvent::InstanceTerminated {
244                instance_id: instance_id.into(),
245                actor: actor.into(),
246                timestamp: Utc::now(),
247            })
248            .await
249            .ok();
250        self.audit.log_action(
251            AuditAction::Terminate,
252            actor,
253            instance_id,
254            instance.status,
255            InstanceStatus::Terminated,
256            serde_json::json!({}),
257        );
258        Ok(())
259    }
260
261    /// 查询实例详情。
262    pub async fn query(&self, instance_id: &str) -> WorkflowResult<InstanceDetail> {
263        let instance = self.load_instance(instance_id).await?;
264        let tasks = self.task_manager.list_by_instance(instance_id).await?;
265        Ok(InstanceDetail {
266            instance,
267            current_tasks: tasks,
268        })
269    }
270
271    /// 按状态分页查询。
272    pub async fn list_by_status(
273        &self,
274        status: InstanceStatus,
275        page: PageRequest,
276    ) -> WorkflowResult<PageResult<FlowInstance>> {
277        self.instance_repo.list_by_status(status, page).await
278    }
279
280    /// 查询历史轨迹（敏感字段脱敏）。
281    pub async fn query_history(
282        &self,
283        instance_id: &str,
284    ) -> WorkflowResult<Vec<crate::instance::HistoryEntry>> {
285        let entries = self
286            .history_recorder
287            .history_repo
288            .list_by_instance(instance_id)
289            .await?;
290        Ok(entries)
291    }
292
293    async fn load_instance(&self, instance_id: &str) -> WorkflowResult<FlowInstance> {
294        self.instance_repo.get(instance_id).await?.ok_or_else(|| {
295            WorkflowError::with_field(
296                WorkflowErrorCode::InstanceNotFound,
297                "实例不存在",
298                "instance_id",
299                instance_id,
300            )
301        })
302    }
303
304    async fn save_with_lock(&self, instance: &FlowInstance, expected: u64) -> WorkflowResult<()> {
305        let success = self
306            .instance_repo
307            .update_with_version(instance, expected)
308            .await?;
309        if !success {
310            return Err(WorkflowError::new(
311                WorkflowErrorCode::OptimisticLockConflict,
312                "乐观锁冲突",
313            ));
314        }
315        Ok(())
316    }
317}
sz_rust_workflow/engine/instance.rs

sz_rust_workflow/engine/
instance.rs
