iven/erp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore: apply cargo fmt across workspace and update docs

Browse Source 
- Run cargo fmt on all Rust crates for consistent formatting
- Update CLAUDE.md with WASM plugin commands and dev.ps1 instructions
- Update wiki: add WASM plugin architecture, rewrite dev environment docs
- Minor frontend cleanup (unused imports)
This commit is contained in:
iven
2026-04-15 00:49:20 +08:00
parent e16c1a85d7
commit 9568dd7875
113 changed files with 4355 additions and 937 deletions
Download Patch File Download Diff File Expand all files Collapse all files

249
crates/erp-workflow/src/engine/executor.rs
View File

@@ -2,15 +2,14 @@ use std::collections::HashMap;
use chrono::Utc;
use sea_orm::{
ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, Set, ConnectionTrait,
PaginatorTrait,
ActiveModelTrait, ColumnTrait, ConnectionTrait, EntityTrait, PaginatorTrait, QueryFilter, Set,
};
use uuid::Uuid;
use crate::dto::NodeType;
use crate::engine::expression::ExpressionEvaluator;
use crate::engine::model::FlowGraph;
use crate::entity::{token, process_instance, task};
use crate::entity::{process_instance, task, token};
use crate::error::{WorkflowError, WorkflowResult};
/// Token 驱动的流程执行引擎。
@@ -92,11 +91,16 @@ impl FlowExecutor {
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()))?;
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)
let current_node = graph
.nodes
.get(&node_id)
.ok_or_else(|| WorkflowError::InvalidDiagram(format!("节点不存在: {node_id}")))?;
match current_node.node_type {
@@ -177,11 +181,9 @@ impl FlowExecutor {
}
}
let target = matched_target
.or(default_target)
.ok_or_else(|| WorkflowError::ExpressionError(
"排他网关没有匹配的条件分支".to_string(),
))?;
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
}
@@ -219,136 +221,139 @@ impl FlowExecutor {
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>> {
) -> 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}")))?;
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
}
NodeType::ServiceTask => {
// ServiceTask 自动执行：当前阶段自动跳过（直接推进到后继节点）
// 创建一个立即消费的 token 记录（用于审计追踪）
let now = Utc::now();
let system_user = uuid::Uuid::nil();
let auto_token_id = Uuid::now_v7();
let token_model = token::ActiveModel {
id: Set(auto_token_id),
tenant_id: Set(tenant_id),
instance_id: Set(instance_id),
node_id: Set(node_id.to_string()),
status: Set("consumed".to_string()),
created_at: Set(now),
updated_at: Set(now),
created_by: Set(system_user),
updated_by: Set(system_user),
deleted_at: Set(None),
version: Set(1),
consumed_at: Set(Some(now)),
};
token_model
.insert(txn)
.await
.map_err(|e| WorkflowError::Validation(e.to_string()))?;
tracing::info!(node_id = node_id, node_name = %node.name, "ServiceTask 自动跳过（尚未实现 HTTP 调用）");
// 沿出边继续推进
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(
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,
&edge.target,
node_id,
graph,
variables,
txn,
)
.await?;
new_tokens.extend(tokens);
}
Ok(new_tokens)
}
_ => {
// UserTask / 网关（分支）等：创建活跃 token
let new_token_id = Uuid::now_v7();
let now = Utc::now();
let system_user = uuid::Uuid::nil();
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),
updated_at: Set(now),
created_by: Set(system_user),
updated_by: Set(system_user),
deleted_at: Set(None),
version: Set(1),
consumed_at: Set(None),
};
token_model
.insert(txn)
.await
.map_err(|e| WorkflowError::Validation(e.to_string()))?;
}
NodeType::ServiceTask => {
// ServiceTask 自动执行：当前阶段自动跳过（直接推进到后继节点）
// 创建一个立即消费的 token 记录（用于审计追踪）
let now = Utc::now();
let system_user = uuid::Uuid::nil();
let auto_token_id = Uuid::now_v7();
// UserTask: 同时创建 task 记录
if node.node_type == NodeType::UserTask {
let task_model = task::ActiveModel {
id: Set(Uuid::now_v7()),
let token_model = token::ActiveModel {
id: Set(auto_token_id),
tenant_id: Set(tenant_id),
instance_id: Set(instance_id),
token_id: Set(new_token_id),
node_id: Set(node_id.to_string()),
node_name: Set(Some(node.name.clone())),
assignee_id: Set(node.assignee_id),
candidate_groups: Set(node.candidate_groups.as_ref()
.map(|g| serde_json::to_value(g).unwrap_or_default())),
status: Set("pending".to_string()),
outcome: Set(None),
form_data: Set(None),
due_date: Set(None),
completed_at: Set(None),
status: Set("consumed".to_string()),
created_at: Set(now),
updated_at: Set(now),
created_by: Set(Uuid::nil()),
updated_by: Set(Uuid::nil()),
created_by: Set(system_user),
updated_by: Set(system_user),
deleted_at: Set(None),
version: Set(1),
consumed_at: Set(Some(now)),
};
task_model
token_model
.insert(txn)
.await
.map_err(|e| WorkflowError::Validation(e.to_string()))?;
}
Ok(vec![new_token_id])
tracing::info!(node_id = node_id, node_name = %node.name, "ServiceTask 自动跳过（尚未实现 HTTP 调用）");
// 沿出边继续推进
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)
}
_ => {
// UserTask / 网关（分支）等：创建活跃 token
let new_token_id = Uuid::now_v7();
let now = Utc::now();
let system_user = uuid::Uuid::nil();
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),
updated_at: Set(now),
created_by: Set(system_user),
updated_by: Set(system_user),
deleted_at: Set(None),
version: Set(1),
consumed_at: Set(None),
};
token_model
.insert(txn)
.await
.map_err(|e| WorkflowError::Validation(e.to_string()))?;
// UserTask: 同时创建 task 记录
if node.node_type == NodeType::UserTask {
let task_model = task::ActiveModel {
id: Set(Uuid::now_v7()),
tenant_id: Set(tenant_id),
instance_id: Set(instance_id),
token_id: Set(new_token_id),
node_id: Set(node_id.to_string()),
node_name: Set(Some(node.name.clone())),
assignee_id: Set(node.assignee_id),
candidate_groups: Set(node
.candidate_groups
.as_ref()
.map(|g| serde_json::to_value(g).unwrap_or_default())),
status: Set("pending".to_string()),
outcome: Set(None),
form_data: Set(None),
due_date: Set(None),
completed_at: Set(None),
created_at: Set(now),
updated_at: Set(now),
created_by: Set(Uuid::nil()),
updated_by: Set(Uuid::nil()),
deleted_at: Set(None),
version: Set(1),
};
task_model
.insert(txn)
.await
.map_err(|e| WorkflowError::Validation(e.to_string()))?;
}
Ok(vec![new_token_id])
}
}
}
})
}
@@ -445,7 +450,10 @@ impl FlowExecutor {
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()))?;
active
.update(txn)
.await
.map_err(|e| WorkflowError::Validation(e.to_string()))?;
// 写入完成事件到 outbox，由 relay 广播
let now = Utc::now();
@@ -461,7 +469,10 @@ impl FlowExecutor {
created_at: Set(now),
published_at: Set(None),
};
outbox_event.insert(txn).await.map_err(|e| WorkflowError::Validation(e.to_string()))?;
outbox_event
.insert(txn)
.await
.map_err(|e| WorkflowError::Validation(e.to_string()))?;
}
Ok(())

10
crates/erp-workflow/src/engine/expression.rs
View File

@@ -18,7 +18,10 @@ impl ExpressionEvaluator {
/// 求值单个条件表达式。
///
/// 表达式格式: `{left} {op} {right}` 或复合表达式 `{expr1} && {expr2}`
pub fn eval(expr: &str, variables: &HashMap<String, serde_json::Value>) -> WorkflowResult<bool> {
pub fn eval(
expr: &str,
variables: &HashMap<String, serde_json::Value>,
) -> WorkflowResult<bool> {
let expr = expr.trim();
// 处理逻辑 OR
@@ -72,7 +75,10 @@ impl ExpressionEvaluator {
}
/// 求值单个比较表达式。
fn eval_comparison(expr: &str, variables: &HashMap<String, serde_json::Value>) -> WorkflowResult<bool> {
fn eval_comparison(
expr: &str,
variables: &HashMap<String, serde_json::Value>,
) -> WorkflowResult<bool> {
let operators = [">=", "<=", "!=", "==", ">", "<"];
for op in &operators {

2
crates/erp-workflow/src/engine/mod.rs
View File

@@ -1,5 +1,5 @@
pub mod expression;
pub mod executor;
pub mod expression;
pub mod model;
pub mod parser;
pub mod timeout;

17
crates/erp-workflow/src/engine/model.rs
View File

@@ -75,13 +75,16 @@ impl FlowGraph {
}
for e in edges {
graph.edges.insert(e.id.clone(), FlowEdge {
id: e.id.clone(),
source: e.source.clone(),
target: e.target.clone(),
condition: e.condition.clone(),
label: e.label.clone(),
});
graph.edges.insert(
e.id.clone(),
FlowEdge {
id: e.id.clone(),
source: e.source.clone(),
target: e.target.clone(),
condition: e.condition.clone(),
label: e.label.clone(),
},
);
if let Some(out) = graph.outgoing.get_mut(&e.source) {
out.push(e.id.clone());

16
crates/erp-workflow/src/engine/parser.rs
View File

@@ -10,7 +10,10 @@ pub fn parse_and_validate(nodes: &[NodeDef], edges: &[EdgeDef]) -> WorkflowResul
}
// 检查恰好 1 个 StartEvent
let start_count = nodes.iter().filter(|n| n.node_type == NodeType::StartEvent).count();
let start_count = nodes
.iter()
.filter(|n| n.node_type == NodeType::StartEvent)
.count();
if start_count == 0 {
return Err(WorkflowError::InvalidDiagram(
"流程图必须包含一个开始事件".to_string(),
@@ -23,7 +26,10 @@ pub fn parse_and_validate(nodes: &[NodeDef], edges: &[EdgeDef]) -> WorkflowResul
}
// 检查至少 1 个 EndEvent
let end_count = nodes.iter().filter(|n| n.node_type == NodeType::EndEvent).count();
let end_count = nodes
.iter()
.filter(|n| n.node_type == NodeType::EndEvent)
.count();
if end_count == 0 {
return Err(WorkflowError::InvalidDiagram(
"流程图必须包含至少一个结束事件".to_string(),
@@ -31,8 +37,7 @@ pub fn parse_and_validate(nodes: &[NodeDef], edges: &[EdgeDef]) -> WorkflowResul
}
// 检查节点 ID 唯一性
let node_ids: std::collections::HashSet<&str> =
nodes.iter().map(|n| n.id.as_str()).collect();
let node_ids: std::collections::HashSet<&str> = nodes.iter().map(|n| n.id.as_str()).collect();
if node_ids.len() != nodes.len() {
return Err(WorkflowError::InvalidDiagram(
"节点 ID 不能重复".to_string(),
@@ -101,7 +106,8 @@ pub fn parse_and_validate(nodes: &[NodeDef], edges: &[EdgeDef]) -> WorkflowResul
}
// 排他网关的出边应该有条件（第一条可以无条件作为默认分支）
if node.node_type == NodeType::ExclusiveGateway && out.len() > 1 {
let with_condition: Vec<_> = out.iter().filter(|e| e.condition.is_some()).collect();
let with_condition: Vec<_> =
out.iter().filter(|e| e.condition.is_some()).collect();
if with_condition.is_empty() {
return Err(WorkflowError::InvalidDiagram(format!(
"排他网关 '{}' 有多条出边但没有条件表达式",

4
crates/erp-workflow/src/entity/mod.rs
View File

@@ -1,5 +1,5 @@
pub mod process_definition;
pub mod process_instance;
pub mod token;
pub mod task;
pub mod process_variable;
pub mod task;
pub mod token;

14
crates/erp-workflow/src/handler/definition_handler.rs
View File

@@ -94,8 +94,7 @@ where
{
require_permission(&ctx, "workflow.update")?;
let resp =
DefinitionService::update(id, ctx.tenant_id, ctx.user_id, &req, &state.db).await?;
let resp = DefinitionService::update(id, ctx.tenant_id, ctx.user_id, &req, &state.db).await?;
Ok(Json(ApiResponse::ok(resp)))
}
@@ -111,14 +110,9 @@ where
{
require_permission(&ctx, "workflow.publish")?;
let resp = DefinitionService::publish(
id,
ctx.tenant_id,
ctx.user_id,
&state.db,
&state.event_bus,
)
.await?;
let resp =
DefinitionService::publish(id, ctx.tenant_id, ctx.user_id, &state.db, &state.event_bus)
.await?;
Ok(Json(ApiResponse::ok(resp)))
}

6
crates/erp-workflow/src/handler/instance_handler.rs
View File

@@ -23,7 +23,8 @@ where
S: Clone + Send + Sync + 'static,
{
require_permission(&ctx, "workflow.start")?;
req.validate().map_err(|e| AppError::Validation(e.to_string()))?;
req.validate()
.map_err(|e| AppError::Validation(e.to_string()))?;
let resp = InstanceService::start(
ctx.tenant_id,
@@ -49,8 +50,7 @@ where
{
require_permission(&ctx, "workflow.list")?;
let (instances, total) =
InstanceService::list(ctx.tenant_id, &pagination, &state.db).await?;
let (instances, total) = InstanceService::list(ctx.tenant_id, &pagination, &state.db).await?;
let page = pagination.page.unwrap_or(1);
let page_size = pagination.limit();

9
crates/erp-workflow/src/handler/task_handler.rs
View File

@@ -80,7 +80,8 @@ where
S: Clone + Send + Sync + 'static,
{
require_permission(&ctx, "workflow.approve")?;
req.validate().map_err(|e| AppError::Validation(e.to_string()))?;
req.validate()
.map_err(|e| AppError::Validation(e.to_string()))?;
let resp = TaskService::complete(
id,
@@ -107,10 +108,10 @@ where
S: Clone + Send + Sync + 'static,
{
require_permission(&ctx, "workflow.delegate")?;
req.validate().map_err(|e| AppError::Validation(e.to_string()))?;
req.validate()
.map_err(|e| AppError::Validation(e.to_string()))?;
let resp =
TaskService::delegate(id, ctx.tenant_id, ctx.user_id, &req, &state.db).await?;
let resp = TaskService::delegate(id, ctx.tenant_id, ctx.user_id, &req, &state.db).await?;
Ok(Json(ApiResponse::ok(resp)))
}

10
crates/erp-workflow/src/module.rs
View File

@@ -7,9 +7,7 @@ use erp_core::error::AppResult;
use erp_core::events::EventBus;
use erp_core::module::ErpModule;
use crate::handler::{
definition_handler, instance_handler, task_handler,
};
use crate::handler::{definition_handler, instance_handler, task_handler};
/// Workflow module implementing the `ErpModule` trait.
///
@@ -37,8 +35,7 @@ impl WorkflowModule {
)
.route(
"/workflow/definitions/{id}",
get(definition_handler::get_definition)
.put(definition_handler::update_definition),
get(definition_handler::get_definition).put(definition_handler::update_definition),
)
.route(
"/workflow/definitions/{id}/publish",
@@ -47,8 +44,7 @@ impl WorkflowModule {
// Instance routes
.route(
"/workflow/instances",
post(instance_handler::start_instance)
.get(instance_handler::list_instances),
post(instance_handler::start_instance).get(instance_handler::list_instances),
)
.route(
"/workflow/instances/{id}",

74
crates/erp-workflow/src/service/definition_service.rs
View File

@@ -1,12 +1,8 @@
use chrono::Utc;
use sea_orm::{
ActiveModelTrait, ColumnTrait, EntityTrait, PaginatorTrait, QueryFilter, Set,
};
use sea_orm::{ActiveModelTrait, ColumnTrait, EntityTrait, PaginatorTrait, QueryFilter, Set};
use uuid::Uuid;
use crate::dto::{
CreateProcessDefinitionReq, ProcessDefinitionResp, UpdateProcessDefinitionReq,
};
use crate::dto::{CreateProcessDefinitionReq, ProcessDefinitionResp, UpdateProcessDefinitionReq};
use crate::engine::parser;
use crate::entity::process_definition;
use crate::error::{WorkflowError, WorkflowResult};
@@ -103,15 +99,25 @@ impl DefinitionService {
.await
.map_err(|e| WorkflowError::Validation(e.to_string()))?;
event_bus.publish(erp_core::events::DomainEvent::new(
"process_definition.created",
tenant_id,
serde_json::json!({ "definition_id": id, "key": req.key }),
), db).await;
event_bus
.publish(
erp_core::events::DomainEvent::new(
"process_definition.created",
tenant_id,
serde_json::json!({ "definition_id": id, "key": req.key }),
),
db,
)
.await;
audit_service::record(
AuditLog::new(tenant_id, Some(operator_id), "process_definition.create", "process_definition")
.with_resource_id(id),
AuditLog::new(
tenant_id,
Some(operator_id),
"process_definition.create",
"process_definition",
)
.with_resource_id(id),
db,
)
.await;
@@ -192,8 +198,13 @@ impl DefinitionService {
.map_err(|e| WorkflowError::Validation(e.to_string()))?;
audit_service::record(
AuditLog::new(tenant_id, Some(operator_id), "process_definition.update", "process_definition")
.with_resource_id(id),
AuditLog::new(
tenant_id,
Some(operator_id),
"process_definition.update",
"process_definition",
)
.with_resource_id(id),
db,
)
.await;
@@ -241,15 +252,25 @@ impl DefinitionService {
.await
.map_err(|e| WorkflowError::Validation(e.to_string()))?;
event_bus.publish(erp_core::events::DomainEvent::new(
"process_definition.published",
tenant_id,
serde_json::json!({ "definition_id": id }),
), db).await;
event_bus
.publish(
erp_core::events::DomainEvent::new(
"process_definition.published",
tenant_id,
serde_json::json!({ "definition_id": id }),
),
db,
)
.await;
audit_service::record(
AuditLog::new(tenant_id, Some(operator_id), "process_definition.publish", "process_definition")
.with_resource_id(id),
AuditLog::new(
tenant_id,
Some(operator_id),
"process_definition.publish",
"process_definition",
)
.with_resource_id(id),
db,
)
.await;
@@ -283,8 +304,13 @@ impl DefinitionService {
.map_err(|e| WorkflowError::Validation(e.to_string()))?;
audit_service::record(
AuditLog::new(tenant_id, Some(operator_id), "process_definition.delete", "process_definition")
.with_resource_id(id),
AuditLog::new(
tenant_id,
Some(operator_id),
"process_definition.delete",
"process_definition",
)
.with_resource_id(id),
db,
)
.await;

39
crates/erp-workflow/src/service/instance_service.rs
View File

@@ -2,8 +2,8 @@ use std::collections::HashMap;
use chrono::Utc;
use sea_orm::{
ActiveModelTrait, ColumnTrait, EntityTrait, PaginatorTrait, QueryFilter, Set,
TransactionTrait, ConnectionTrait,
ActiveModelTrait, ColumnTrait, ConnectionTrait, EntityTrait, PaginatorTrait, QueryFilter, Set,
TransactionTrait,
};
use uuid::Uuid;
@@ -94,7 +94,10 @@ impl InstanceService {
deleted_at: Set(None),
version: Set(1),
};
instance.insert(txn).await.map_err(|e| WorkflowError::Validation(e.to_string()))?;
instance
.insert(txn)
.await
.map_err(|e| WorkflowError::Validation(e.to_string()))?;
// 保存初始变量
if let Some(vars) = vars_to_save {
@@ -112,14 +115,8 @@ impl InstanceService {
}
// 启动执行引擎
FlowExecutor::start(
instance_id_clone,
tenant_id_clone,
&graph,
&variables,
txn,
)
.await?;
FlowExecutor::start(instance_id_clone, tenant_id_clone, &graph, &variables, txn)
.await?;
Ok(())
})
@@ -133,8 +130,13 @@ impl InstanceService {
), db).await;
audit_service::record(
AuditLog::new(tenant_id, Some(operator_id), "process_instance.start", "process_instance")
.with_resource_id(instance_id),
AuditLog::new(
tenant_id,
Some(operator_id),
"process_instance.start",
"process_instance",
)
.with_resource_id(instance_id),
db,
)
.await;
@@ -320,7 +322,9 @@ impl InstanceService {
id: Set(event_id),
tenant_id: Set(tenant_id),
event_type: Set(event_type),
payload: Set(Some(serde_json::json!({ "instance_id": id, "changed_by": operator_id }))),
payload: Set(Some(
serde_json::json!({ "instance_id": id, "changed_by": operator_id }),
)),
correlation_id: Set(Some(Uuid::now_v7())),
status: Set("pending".to_string()),
attempts: Set(0),
@@ -378,7 +382,12 @@ impl InstanceService {
) -> WorkflowResult<()> {
let id = Uuid::now_v7();
let (value_string, value_number, value_boolean, _value_date): (Option<String>, Option<f64>, Option<bool>, Option<chrono::DateTime<Utc>>) = match var_type {
let (value_string, value_number, value_boolean, _value_date): (
Option<String>,
Option<f64>,
Option<bool>,
Option<chrono::DateTime<Utc>>,
) = match var_type {
"string" => (value.as_str().map(|s| s.to_string()), None, None, None),
"number" => (None, value.as_f64(), None, None),
"boolean" => (None, None, value.as_bool(), None),

56
crates/erp-workflow/src/service/task_service.rs
View File

@@ -2,8 +2,8 @@ use std::collections::HashMap;
use chrono::Utc;
use sea_orm::{
ActiveModelTrait, ColumnTrait, ConnectionTrait, DatabaseBackend, EntityTrait,
PaginatorTrait, QueryFilter, Set, Statement, TransactionTrait,
ActiveModelTrait, ColumnTrait, ConnectionTrait, DatabaseBackend, EntityTrait, PaginatorTrait,
QueryFilter, Set, Statement, TransactionTrait,
};
use uuid::Uuid;
@@ -178,14 +178,10 @@ impl TaskService {
)));
}
let nodes: Vec<crate::dto::NodeDef> =
serde_json::from_value(definition.nodes.clone()).map_err(|e| {
WorkflowError::InvalidDiagram(format!("节点数据无效: {e}"))
})?;
let edges: Vec<crate::dto::EdgeDef> =
serde_json::from_value(definition.edges.clone()).map_err(|e| {
WorkflowError::InvalidDiagram(format!("连线数据无效: {e}"))
})?;
let nodes: Vec<crate::dto::NodeDef> = serde_json::from_value(definition.nodes.clone())
.map_err(|e| WorkflowError::InvalidDiagram(format!("节点数据无效: {e}")))?;
let edges: Vec<crate::dto::EdgeDef> = serde_json::from_value(definition.edges.clone())
.map_err(|e| WorkflowError::InvalidDiagram(format!("连线数据无效: {e}")))?;
let graph = parser::parse_and_validate(&nodes, &edges)?;
// 准备变量（从 req.form_data 中提取）
@@ -223,31 +219,29 @@ impl TaskService {
.map_err(|e| WorkflowError::Validation(e.to_string()))?;
// 推进 token
FlowExecutor::advance(
token_id,
instance_id,
tenant_id,
&graph,
&variables,
txn,
)
.await?;
FlowExecutor::advance(token_id, instance_id, tenant_id, &graph, &variables, txn)
.await?;
Ok(())
})
})
.await?;
event_bus.publish(erp_core::events::DomainEvent::new(
"task.completed",
tenant_id,
serde_json::json!({
"task_id": id,
"instance_id": instance_id,
"started_by": instance.started_by,
"outcome": req.outcome,
}),
), db).await;
event_bus
.publish(
erp_core::events::DomainEvent::new(
"task.completed",
tenant_id,
serde_json::json!({
"task_id": id,
"instance_id": instance_id,
"started_by": instance.started_by,
"outcome": req.outcome,
}),
),
db,
)
.await;
audit_service::record(
AuditLog::new(tenant_id, Some(operator_id), "task.complete", "task")
@@ -359,7 +353,9 @@ impl TaskService {
node_id: Set(node_id.to_string()),
node_name: Set(node_name.map(|s| s.to_string())),
assignee_id: Set(assignee_id),
candidate_groups: Set(candidate_groups.map(|g| serde_json::to_value(g).unwrap_or_default())),
candidate_groups: Set(
candidate_groups.map(|g| serde_json::to_value(g).unwrap_or_default())
),
status: Set("pending".to_string()),
outcome: Set(None),
form_data: Set(None),