feat(workflow): add workflow engine module (Phase 4)
Implement complete workflow engine with BPMN subset support: Backend (erp-workflow crate): - Token-driven execution engine with exclusive/parallel gateway support - BPMN parser with flow graph validation - Expression evaluator for conditional branching - Process definition CRUD with draft/publish lifecycle - Process instance management (start, suspend, terminate) - Task service (pending, complete, delegate) - PostgreSQL advisory locks for concurrent safety - 5 database tables: process_definitions, process_instances, tokens, tasks, process_variables - 13 API endpoints with RBAC protection - Timeout checker framework (placeholder) Frontend: - Workflow page with 4 tabs (definitions, pending, completed, monitor) - React Flow visual process designer (@xyflow/react) - Process viewer with active node highlighting - 3 API client modules for workflow endpoints - Sidebar menu integration
This commit is contained in:
iven
371
crates/erp-workflow/src/engine/executor.rs
Normal file
371
crates/erp-workflow/src/engine/executor.rs
Normal file
@@ -0,0 +1,371 @@
|
||||
use std::collections::HashMap;
|
||||
|
||||
use chrono::Utc;
|
||||
use sea_orm::{
|
||||
ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, Set, ConnectionTrait,
|
||||
PaginatorTrait,
|
||||
};
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::dto::NodeType;
|
||||
use crate::engine::expression::ExpressionEvaluator;
|
||||
use crate::engine::model::FlowGraph;
|
||||
use crate::entity::{token, process_instance};
|
||||
use crate::error::{WorkflowError, WorkflowResult};
|
||||
|
||||
/// Token 驱动的流程执行引擎。
|
||||
///
|
||||
/// 核心职责：
|
||||
/// - 在流程启动时，于 StartEvent 创建第一个 token
|
||||
/// - 在任务完成时推进 token 到下一个节点
|
||||
/// - 处理网关分支/汇合逻辑
|
||||
/// - 在 EndEvent 完成实例
|
||||
pub struct FlowExecutor;
|
||||
|
||||
impl FlowExecutor {
|
||||
/// 启动流程：在 StartEvent 的后继节点创建 token。
|
||||
///
|
||||
/// 返回创建的 token ID 列表。
|
||||
pub async fn start(
|
||||
instance_id: Uuid,
|
||||
tenant_id: Uuid,
|
||||
graph: &FlowGraph,
|
||||
variables: &HashMap<String, serde_json::Value>,
|
||||
txn: &impl ConnectionTrait,
|
||||
) -> WorkflowResult<Vec<Uuid>> {
|
||||
let start_id = graph
|
||||
.start_node_id
|
||||
.as_ref()
|
||||
.ok_or_else(|| WorkflowError::InvalidDiagram("流程图没有开始事件".to_string()))?;
|
||||
|
||||
// 获取 StartEvent 的出边，推进到后继节点
|
||||
let outgoing = graph.get_outgoing_edges(start_id);
|
||||
if outgoing.is_empty() {
|
||||
return Err(WorkflowError::InvalidDiagram(
|
||||
"开始事件没有出边".to_string(),
|
||||
));
|
||||
}
|
||||
|
||||
// StartEvent 只有一条出边
|
||||
let first_edge = &outgoing[0];
|
||||
let target_node_id = &first_edge.target;
|
||||
|
||||
Self::create_token_at_node(
|
||||
instance_id,
|
||||
tenant_id,
|
||||
target_node_id,
|
||||
graph,
|
||||
variables,
|
||||
txn,
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
/// 推进 token：消费当前 token，在下一节点创建新 token。
|
||||
///
|
||||
/// 返回新创建的 token ID 列表。
|
||||
pub async fn advance(
|
||||
token_id: Uuid,
|
||||
instance_id: Uuid,
|
||||
tenant_id: Uuid,
|
||||
graph: &FlowGraph,
|
||||
variables: &HashMap<String, serde_json::Value>,
|
||||
txn: &impl ConnectionTrait,
|
||||
) -> WorkflowResult<Vec<Uuid>> {
|
||||
// 读取当前 token
|
||||
let current_token = token::Entity::find_by_id(token_id)
|
||||
.one(txn)
|
||||
.await
|
||||
.map_err(|e| WorkflowError::Validation(e.to_string()))?
|
||||
.ok_or_else(|| WorkflowError::NotFound(format!("Token 不存在: {token_id}")))?;
|
||||
|
||||
if current_token.status != "active" {
|
||||
return Err(WorkflowError::InvalidState(format!(
|
||||
"Token 状态不是 active: {}",
|
||||
current_token.status
|
||||
)));
|
||||
}
|
||||
|
||||
let node_id = current_token.node_id.clone();
|
||||
|
||||
// 消费当前 token
|
||||
let mut active: token::ActiveModel = current_token.into();
|
||||
active.status = Set("consumed".to_string());
|
||||
active.consumed_at = Set(Some(Utc::now()));
|
||||
active.update(txn).await.map_err(|e| WorkflowError::Validation(e.to_string()))?;
|
||||
|
||||
// 获取当前节点的出边
|
||||
let outgoing = graph.get_outgoing_edges(&node_id);
|
||||
let current_node = graph.nodes.get(&node_id)
|
||||
.ok_or_else(|| WorkflowError::InvalidDiagram(format!("节点不存在: {node_id}")))?;
|
||||
|
||||
match current_node.node_type {
|
||||
NodeType::ExclusiveGateway => {
|
||||
// 排他网关：求值条件，选择一条分支
|
||||
Self::advance_exclusive_gateway(
|
||||
instance_id,
|
||||
tenant_id,
|
||||
&outgoing,
|
||||
graph,
|
||||
variables,
|
||||
txn,
|
||||
)
|
||||
.await
|
||||
}
|
||||
NodeType::ParallelGateway => {
|
||||
// 并行网关：为每条出边创建 token
|
||||
Self::advance_parallel_gateway(
|
||||
instance_id,
|
||||
tenant_id,
|
||||
&outgoing,
|
||||
graph,
|
||||
variables,
|
||||
txn,
|
||||
)
|
||||
.await
|
||||
}
|
||||
_ => {
|
||||
// 普通节点：沿出边前进
|
||||
if outgoing.is_empty() {
|
||||
// 没有出边（理论上只有 EndEvent 会到这里）
|
||||
Ok(vec![])
|
||||
} else {
|
||||
let mut new_tokens = Vec::new();
|
||||
for edge in &outgoing {
|
||||
let tokens = Self::create_token_at_node(
|
||||
instance_id,
|
||||
tenant_id,
|
||||
&edge.target,
|
||||
graph,
|
||||
variables,
|
||||
txn,
|
||||
)
|
||||
.await?;
|
||||
new_tokens.extend(tokens);
|
||||
}
|
||||
Ok(new_tokens)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// 排他网关分支：求值条件，选择第一个满足条件的分支。
|
||||
async fn advance_exclusive_gateway(
|
||||
instance_id: Uuid,
|
||||
tenant_id: Uuid,
|
||||
outgoing: &[&crate::engine::model::FlowEdge],
|
||||
graph: &FlowGraph,
|
||||
variables: &HashMap<String, serde_json::Value>,
|
||||
txn: &impl ConnectionTrait,
|
||||
) -> WorkflowResult<Vec<Uuid>> {
|
||||
let mut default_target: Option<&str> = None;
|
||||
let mut matched_target: Option<&str> = None;
|
||||
|
||||
for edge in outgoing {
|
||||
if let Some(condition) = &edge.condition {
|
||||
match ExpressionEvaluator::eval(condition, variables) {
|
||||
Ok(true) => {
|
||||
matched_target = Some(&edge.target);
|
||||
break;
|
||||
}
|
||||
Ok(false) => continue,
|
||||
Err(_) => continue, // 条件求值失败，跳过
|
||||
}
|
||||
} else {
|
||||
// 无条件的边作为默认分支
|
||||
default_target = Some(&edge.target);
|
||||
}
|
||||
}
|
||||
|
||||
let target = matched_target
|
||||
.or(default_target)
|
||||
.ok_or_else(|| WorkflowError::ExpressionError(
|
||||
"排他网关没有匹配的条件分支".to_string(),
|
||||
))?;
|
||||
|
||||
Self::create_token_at_node(instance_id, tenant_id, target, graph, variables, txn).await
|
||||
}
|
||||
|
||||
/// 并行网关分支：为每条出边创建 token。
|
||||
async fn advance_parallel_gateway(
|
||||
instance_id: Uuid,
|
||||
tenant_id: Uuid,
|
||||
outgoing: &[&crate::engine::model::FlowEdge],
|
||||
graph: &FlowGraph,
|
||||
variables: &HashMap<String, serde_json::Value>,
|
||||
txn: &impl ConnectionTrait,
|
||||
) -> WorkflowResult<Vec<Uuid>> {
|
||||
let mut new_tokens = Vec::new();
|
||||
for edge in outgoing {
|
||||
let tokens = Self::create_token_at_node(
|
||||
instance_id,
|
||||
tenant_id,
|
||||
&edge.target,
|
||||
graph,
|
||||
variables,
|
||||
txn,
|
||||
)
|
||||
.await?;
|
||||
new_tokens.extend(tokens);
|
||||
}
|
||||
Ok(new_tokens)
|
||||
}
|
||||
|
||||
/// 在指定节点创建 token，并根据节点类型执行相应逻辑。
|
||||
fn create_token_at_node<'a>(
|
||||
instance_id: Uuid,
|
||||
tenant_id: Uuid,
|
||||
node_id: &'a str,
|
||||
graph: &'a FlowGraph,
|
||||
variables: &'a HashMap<String, serde_json::Value>,
|
||||
txn: &'a impl ConnectionTrait,
|
||||
) -> std::pin::Pin<Box<dyn std::future::Future<Output = WorkflowResult<Vec<Uuid>>> + Send + 'a>> {
|
||||
Box::pin(async move {
|
||||
let node = graph.nodes.get(node_id)
|
||||
.ok_or_else(|| WorkflowError::InvalidDiagram(format!("节点不存在: {node_id}")))?;
|
||||
|
||||
match node.node_type {
|
||||
NodeType::EndEvent => {
|
||||
// 到达 EndEvent，不创建新 token
|
||||
// 检查实例是否所有 token 都完成
|
||||
Self::check_instance_completion(instance_id, tenant_id, txn).await?;
|
||||
Ok(vec![])
|
||||
}
|
||||
NodeType::ParallelGateway
|
||||
if Self::is_join_gateway(node_id, graph) =>
|
||||
{
|
||||
// 并行网关汇合：等待所有入边 token 到达
|
||||
Self::handle_join_gateway(
|
||||
instance_id,
|
||||
tenant_id,
|
||||
node_id,
|
||||
graph,
|
||||
variables,
|
||||
txn,
|
||||
)
|
||||
.await
|
||||
}
|
||||
_ => {
|
||||
// UserTask / ServiceTask / 网关（分支）等：创建活跃 token
|
||||
let new_token_id = Uuid::now_v7();
|
||||
let now = Utc::now();
|
||||
|
||||
let token_model = token::ActiveModel {
|
||||
id: Set(new_token_id),
|
||||
tenant_id: Set(tenant_id),
|
||||
instance_id: Set(instance_id),
|
||||
node_id: Set(node_id.to_string()),
|
||||
status: Set("active".to_string()),
|
||||
created_at: Set(now),
|
||||
consumed_at: Set(None),
|
||||
};
|
||||
token_model
|
||||
.insert(txn)
|
||||
.await
|
||||
.map_err(|e| WorkflowError::Validation(e.to_string()))?;
|
||||
|
||||
Ok(vec![new_token_id])
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
/// 判断并行网关是否是汇合模式（入边数 > 出边数，或者入边数 > 1）。
|
||||
fn is_join_gateway(node_id: &str, graph: &FlowGraph) -> bool {
|
||||
let incoming = graph.get_incoming_edges(node_id);
|
||||
incoming.len() > 1
|
||||
}
|
||||
|
||||
/// 处理并行网关汇合逻辑。
|
||||
///
|
||||
/// 当所有入边的源节点都有已消费的 token 时，创建新 token 推进到后继。
|
||||
async fn handle_join_gateway(
|
||||
instance_id: Uuid,
|
||||
tenant_id: Uuid,
|
||||
node_id: &str,
|
||||
graph: &FlowGraph,
|
||||
variables: &HashMap<String, serde_json::Value>,
|
||||
txn: &impl ConnectionTrait,
|
||||
) -> WorkflowResult<Vec<Uuid>> {
|
||||
let incoming = graph.get_incoming_edges(node_id);
|
||||
|
||||
// 检查所有入边的源节点是否都有已消费/已完成的 token
|
||||
for edge in &incoming {
|
||||
let has_consumed = token::Entity::find()
|
||||
.filter(token::Column::InstanceId.eq(instance_id))
|
||||
.filter(token::Column::NodeId.eq(&edge.source))
|
||||
.filter(token::Column::Status.is_in(["consumed", "active"]))
|
||||
.one(txn)
|
||||
.await
|
||||
.map_err(|e| WorkflowError::Validation(e.to_string()))?;
|
||||
|
||||
if has_consumed.is_none() {
|
||||
// 还有分支没有到达，等待
|
||||
return Ok(vec![]);
|
||||
}
|
||||
|
||||
// 检查是否还有活跃的 token（来自其他分支）
|
||||
let has_active = token::Entity::find()
|
||||
.filter(token::Column::InstanceId.eq(instance_id))
|
||||
.filter(token::Column::NodeId.eq(&edge.source))
|
||||
.filter(token::Column::Status.eq("active"))
|
||||
.one(txn)
|
||||
.await
|
||||
.map_err(|e| WorkflowError::Validation(e.to_string()))?;
|
||||
|
||||
if has_active.is_some() {
|
||||
// 还有分支在执行中，等待
|
||||
return Ok(vec![]);
|
||||
}
|
||||
}
|
||||
|
||||
// 所有分支都完成了，沿出边继续
|
||||
let outgoing = graph.get_outgoing_edges(node_id);
|
||||
let mut new_tokens = Vec::new();
|
||||
for edge in &outgoing {
|
||||
let tokens = Self::create_token_at_node(
|
||||
instance_id,
|
||||
tenant_id,
|
||||
&edge.target,
|
||||
graph,
|
||||
variables,
|
||||
txn,
|
||||
)
|
||||
.await?;
|
||||
new_tokens.extend(tokens);
|
||||
}
|
||||
Ok(new_tokens)
|
||||
}
|
||||
|
||||
/// 检查实例是否所有 token 都已完成，如果是则完成实例。
|
||||
async fn check_instance_completion(
|
||||
instance_id: Uuid,
|
||||
tenant_id: Uuid,
|
||||
txn: &impl ConnectionTrait,
|
||||
) -> WorkflowResult<()> {
|
||||
let active_count = token::Entity::find()
|
||||
.filter(token::Column::InstanceId.eq(instance_id))
|
||||
.filter(token::Column::Status.eq("active"))
|
||||
.count(txn)
|
||||
.await
|
||||
.map_err(|e| WorkflowError::Validation(e.to_string()))?;
|
||||
|
||||
if active_count == 0 {
|
||||
// 所有 token 都完成，标记实例完成
|
||||
let instance = process_instance::Entity::find_by_id(instance_id)
|
||||
.one(txn)
|
||||
.await
|
||||
.map_err(|e| WorkflowError::Validation(e.to_string()))?
|
||||
.filter(|i| i.tenant_id == tenant_id && i.deleted_at.is_none())
|
||||
.ok_or_else(|| WorkflowError::NotFound(format!("流程实例不存在: {instance_id}")))?;
|
||||
|
||||
let mut active: process_instance::ActiveModel = instance.into();
|
||||
active.status = Set("completed".to_string());
|
||||
active.completed_at = Set(Some(Utc::now()));
|
||||
active.updated_at = Set(Utc::now());
|
||||
active.update(txn).await.map_err(|e| WorkflowError::Validation(e.to_string()))?;
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user