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

feat: systematic functional audit — fix 18 issues across Phase A/B

Browse Source 
Phase A (P1 production blockers):
- A1: Apply IP rate limiting to public routes (login/refresh)
- A2: Publish domain events for workflow instance state transitions
  (completed/suspended/resumed/terminated) via outbox pattern
- A3: Replace hardcoded nil UUID default tenant with dynamic DB lookup
- A4: Add GET /api/v1/audit-logs query endpoint with pagination
- A5: Enhance CORS wildcard warning for production environments

Phase B (P2 functional gaps):
- B1: Remove dead erp-common crate (zero references in codebase)
- B2: Refactor 5 settings pages to use typed API modules instead of
  direct client calls; create api/themes.ts; delete dead errors.ts
- B3: Add resume/suspend buttons to InstanceMonitor page
- B4: Remove unused EventHandler trait from erp-core
- B5: Handle task.completed events in message module (send notifications)
- B6: Wire TimeoutChecker as 60s background task
- B7: Auto-skip ServiceTask nodes instead of crashing the process
- B8: Remove empty register_routes() from ErpModule trait and modules
This commit is contained in:
iven
2026-04-12 15:22:28 +08:00
parent 685df5e458
commit 14f431efff
34 changed files with 785 additions and 304 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -246,10 +246,49 @@ impl FlowExecutor {
.await
}
NodeType::ServiceTask => {
// ServiceTask 尚未实现：无法自动执行服务调用，直接报错
return Err(WorkflowError::Validation(
format!("ServiceTask ({}) 尚未实现，流程无法继续", node.name),
));
// 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(
instance_id,
tenant_id,
&edge.target,
graph,
variables,
txn,
)
.await?;
new_tokens.extend(tokens);
}
Ok(new_tokens)
}
_ => {
// UserTask / 网关（分支）等：创建活跃 token
@@ -407,6 +446,22 @@ impl FlowExecutor {
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()))?;
// 写入完成事件到 outbox，由 relay 广播
let now = Utc::now();
let outbox_event = erp_core::entity::domain_event::ActiveModel {
id: Set(Uuid::now_v7()),
tenant_id: Set(tenant_id),
event_type: Set("process_instance.completed".to_string()),
payload: Set(Some(serde_json::json!({ "instance_id": instance_id }))),
correlation_id: Set(Some(Uuid::now_v7())),
status: Set("pending".to_string()),
attempts: Set(0),
last_error: Set(None),
created_at: Set(now),
published_at: Set(None),
};
outbox_event.insert(txn).await.map_err(|e| WorkflowError::Validation(e.to_string()))?;
}
Ok(())

29
crates/erp-workflow/src/engine/timeout.rs
View File

@@ -1,7 +1,7 @@
// 超时检查框架 — 占位实现
// 超时检查框架
//
// 当前版本仅提供接口定义，实际超时检查逻辑将在后续迭代中实现。
// Task 表的 due_date 字段已支持设置超时时间。
// TimeoutChecker 定期扫描 tasks 表中已超时但仍处于 pending 状态的任务，
// 以便触发自动完成或升级逻辑（后续迭代实现）。
use chrono::Utc;
use sea_orm::{ColumnTrait, EntityTrait, QueryFilter};
@@ -10,11 +10,11 @@ use uuid::Uuid;
use crate::entity::task;
use crate::error::WorkflowResult;
/// 超时检查服务（占位）。
/// 超时检查服务。
pub struct TimeoutChecker;
impl TimeoutChecker {
/// 查询已超时但未完成的任务列表。
/// 查询指定租户下已超时但未完成的任务列表。
///
/// 返回 due_date < now 且 status = 'pending' 的任务 ID。
pub async fn find_overdue_tasks(
@@ -33,4 +33,23 @@ impl TimeoutChecker {
Ok(overdue.iter().map(|t| t.id).collect())
}
/// 查询所有租户中已超时但未完成的任务列表。
///
/// 返回 due_date < now 且 status = 'pending' 的任务 ID。
/// 用于后台定时任务的全量扫描。
pub async fn find_all_overdue_tasks(
db: &sea_orm::DatabaseConnection,
) -> WorkflowResult<Vec<Uuid>> {
let now = Utc::now();
let overdue = task::Entity::find()
.filter(task::Column::Status.eq("pending"))
.filter(task::Column::DueDate.lt(now))
.filter(task::Column::DeletedAt.is_null())
.all(db)
.await
.map_err(|e| crate::error::WorkflowError::Validation(e.to_string()))?;
Ok(overdue.iter().map(|t| t.id).collect())
}
}

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

@@ -1,5 +1,6 @@
use axum::Router;
use axum::routing::{get, post};
use std::time::Duration;
use uuid::Uuid;
use erp_core::error::AppResult;
@@ -83,6 +84,38 @@ impl WorkflowModule {
post(task_handler::delegate_task),
)
}
/// 启动超时检查后台任务。
///
/// 每 60 秒扫描一次 tasks 表，查找 due_date 已过期但仍处于 pending 状态的任务。
/// 发现超时任务时记录 warning 日志，后续迭代将实现自动完成/升级逻辑。
pub fn start_timeout_checker(db: sea_orm::DatabaseConnection) {
tokio::spawn(async move {
let mut interval = tokio::time::interval(Duration::from_secs(60));
// 首次跳过，等一个完整间隔再执行
interval.tick().await;
loop {
interval.tick().await;
match crate::engine::timeout::TimeoutChecker::find_all_overdue_tasks(&db).await {
Ok(overdue) => {
if !overdue.is_empty() {
tracing::warn!(
count = overdue.len(),
task_ids = ?overdue,
"发现超时未完成的任务 — TODO: 实现自动完成/升级逻辑"
);
}
}
Err(e) => {
tracing::warn!(error = %e, "超时检查任务执行失败");
}
}
}
});
}
}
impl Default for WorkflowModule {
@@ -105,11 +138,6 @@ impl ErpModule for WorkflowModule {
vec!["auth"]
}
fn register_routes(&self, router: Router) -> Router {
// Actual route registration is done via protected_routes(), called by erp-server.
router
}
fn register_event_handlers(&self, _bus: &EventBus) {}
async fn on_tenant_created(&self, _tenant_id: Uuid) -> AppResult<()> {

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

@@ -312,6 +312,27 @@ impl InstanceService {
.await
.map_err(|e| WorkflowError::Validation(e.to_string()))?;
// 发布状态变更领域事件（通过 outbox 模式，由 relay 广播）
let event_type = format!("process_instance.{}", to_status);
let event_id = Uuid::now_v7();
let now = Utc::now();
let outbox_event = erp_core::entity::domain_event::ActiveModel {
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 }))),
correlation_id: Set(Some(Uuid::now_v7())),
status: Set("pending".to_string()),
attempts: Set(0),
last_error: Set(None),
created_at: Set(now),
published_at: Set(None),
};
match outbox_event.insert(db).await {
Ok(_) => {}
Err(e) => tracing::warn!(error = %e, "领域事件持久化失败"),
}
let action = format!("process_instance.{}", to_status);
audit_service::record(
AuditLog::new(tenant_id, Some(operator_id), action, "process_instance")