admin/udmin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(flow): 新增流式执行模式与SSE支持

Browse Source 
新增流式执行模式，通过SSE实时推送节点执行事件与日志
重构HTTP执行器与中间件，提取通用HTTP客户端组件
优化前端测试面板，支持流式模式切换与实时日志展示
更新依赖版本并修复密码哈希的随机数生成器问题
修复前端节点类型映射问题，确保Code节点表单可用
This commit is contained in:
ayou
2025-09-21 01:48:24 +08:00
parent 296f0ae9f6
commit dd7857940f
24 changed files with 1695 additions and 885 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
backend/src/flow/context.rs
View File

@ -14,14 +14,29 @@ pub enum ExecutionMode {
impl Default for ExecutionMode { fn default() -> Self { ExecutionMode::Sync } }
// 新增：流式事件（用于 SSE）
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type")] // 带判别字段，便于前端识别事件类型
pub enum StreamEvent {
#[serde(rename = "node")]
Node { node_id: String, logs: Vec<String>, ctx: serde_json::Value },
#[serde(rename = "done")]
Done { ok: bool, ctx: serde_json::Value, logs: Vec<String> },
#[serde(rename = "error")]
Error { message: String },
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DriveOptions {
#[serde(default)]
pub max_steps: usize,
#[serde(default)]
pub execution_mode: ExecutionMode,
// 新增：事件通道（仅运行时使用，不做序列化/反序列化）
#[serde(default, skip_serializing, skip_deserializing)]
pub event_tx: Option<tokio::sync::mpsc::Sender<StreamEvent>>,
}
impl Default for DriveOptions {
fn default() -> Self { Self { max_steps: 10_000, execution_mode: ExecutionMode::Sync } }
fn default() -> Self { Self { max_steps: 10_000, execution_mode: ExecutionMode::Sync, event_tx: None } }
}

48
backend/src/flow/dsl.rs
View File

@ -1,3 +1,12 @@
//! 模块：流程 DSL 与自由布局 Design JSON 的解析、校验与构建。
//! 主要内容：
//! - FlowDSL/NodeDSL/EdgeDSL：较为“表述性”的简化 DSL 结构（用于外部接口/入库）。
//! - DesignSyntax/NodeSyntax/EdgeSyntax：与前端自由布局 JSON 对齐的结构（含 source_port_id 等）。
//! - validate_design：基础校验（节点 ID 唯一、至少包含一个 start 与一个 end、边的引用合法）。
//! - build_chain_from_design：将自由布局 JSON 转换为内部 ChainDef（含条件节点 AND 组装等启发式与兼容逻辑）。
//! - chain_from_design_json：统一入口，支持字符串/对象两种输入，做兼容字段回填后再校验并构建。
//! 说明：尽量保持向后兼容；在条件节点的出边组装上采用启发式（例如：单出边 + 多条件 => 组装为 AND 条件组）。
use serde::{Deserialize, Serialize};
use serde_json::Value;
use anyhow::bail;
@ -5,36 +14,53 @@ use anyhow::bail;
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FlowDSL {
#[serde(default)]
/// 流程名称（可选）
pub name: String,
#[serde(default, alias = "executionMode")]
/// 执行模式（兼容前端 executionMode），如：sync/async（目前仅占位）
pub execution_mode: Option<String>,
/// 节点列表（按声明顺序）
pub nodes: Vec<NodeDSL>,
#[serde(default)]
/// 边列表（from -> to，可选 condition）
pub edges: Vec<EdgeDSL>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct NodeDSL {
/// 节点唯一 ID（字符串）
pub id: String,
#[serde(default)]
/// 节点类型：start / end / task / condition（开始/结束/任务/条件）
pub kind: String, // 节点类型：start / end / task / condition（开始/结束/任务/条件）
#[serde(default)]
/// 节点显示名称（可选）
pub name: String,
#[serde(default)]
/// 任务标识（绑定执行器），如 http/db/variable/script_*（可选）
pub task: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct EdgeDSL {
/// 起点节点 ID（别名：source/from）
#[serde(alias = "source", alias = "from", rename = "from")]
pub from: String,
/// 终点节点 ID（别名：target/to）
#[serde(alias = "target", alias = "to", rename = "to")]
pub to: String,
#[serde(default)]
/// 条件表达式（字符串）：
/// - 若为 JSON 字符串（以 { 或 [ 开头），则按 JSON 条件集合进行求值；
/// - 否则按 Rhai 表达式求值；
/// - 空字符串/None 表示无条件。
pub condition: Option<String>,
}
impl From<FlowDSL> for super::domain::ChainDef {
/// 将简化 DSL 转换为内部 ChainDef：
/// - kind 映射：start/end/condition/其他->task；支持 decision 别名 -> condition。
/// - 直接搬运 edges 的 from/to/condition。
fn from(v: FlowDSL) -> Self {
super::domain::ChainDef {
name: v.name,
@ -71,34 +97,47 @@ impl From<FlowDSL> for super::domain::ChainDef {
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DesignSyntax {
#[serde(default)]
/// 设计名称（可选）
pub name: String,
#[serde(default)]
/// 节点集合（自由布局）
pub nodes: Vec<NodeSyntax>,
#[serde(default)]
/// 边集合（自由布局）
pub edges: Vec<EdgeSyntax>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct NodeSyntax {
/// 节点 ID
pub id: String,
#[serde(rename = "type", default)]
/// 前端类型：start | end | condition | http | db | task | script_*（用于推断具体执行器）
pub kind: String, // 取值: start | end | condition | http | db | task（开始/结束/条件/HTTP/数据库/通用任务）
#[serde(default)]
/// 节点附加数据：title/conditions/scripts 等
pub data: serde_json::Value,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct EdgeSyntax {
/// 起点（兼容 sourceNodeID/source/from）
#[serde(alias = "sourceNodeID", alias = "source", alias = "from")]
pub from: String,
/// 终点（兼容 targetNodeID/target/to）
#[serde(alias = "targetNodeID", alias = "target", alias = "to")]
pub to: String,
#[serde(default)]
/// 源端口 ID：用于条件节点端口到条件 key 的兼容映射；
/// 特殊值 and/all/group/true 表示将节点内所有 conditions 的 value 组装为 AND 组。
pub source_port_id: Option<String>,
}
/// 从 design_json（前端流程 JSON）构建 ChainDef
/// 设计级别校验：
/// - 节点 ID 唯一且非空；
/// - 至少一个 start 与一个 end；
/// - 边的 from/to 必须指向已知节点。
fn validate_design(design: &DesignSyntax) -> anyhow::Result<()> {
use std::collections::HashSet;
let mut ids = HashSet::new();
@ -129,6 +168,13 @@ fn validate_design(design: &DesignSyntax) -> anyhow::Result<()> {
Ok(())
}
/// 将自由布局 DesignSyntax 转换为内部 ChainDef：
/// - 节点：推断 kind/name/task（含 scripts 与 inline script/expr 的兼容）；
/// - 边：
/// * 条件节点：支持 source_port_id 到 data.conditions 的旧版映射；
/// * 当 source_port_id 为空或为 and/all/group/true，取 conditions 的 value 组成 AND 组；
/// * 启发式：若条件节点仅一条出边且包含多个 conditions，即便 source_port_id 指向具体 key，也回退为 AND 组；
/// * 非条件节点：不处理条件。
fn build_chain_from_design(design: &DesignSyntax) -> anyhow::Result<super::domain::ChainDef> {
use super::domain::{ChainDef, NodeDef, NodeId, NodeKind, LinkDef};

252
backend/src/flow/engine.rs
View File

@ -11,7 +11,17 @@ use std::cell::RefCell;
use rhai::AST;
use regex::Regex;
// 将常用的正则匹配暴露给表达式使用
// 模块：流程执行引擎（engine.rs）
// 作用：驱动 ChainDef 流程图，支持：
// - 同步/异步（Fire-and-Forget）任务执行
// - 条件路由（Rhai 表达式与 JSON 条件）与无条件回退
// - 并发分支 fan-out 与 join_all 等待
// - SSE 实时事件推送（逐行增量 + 节点级切片）
// 设计要点：
// - 表达式执行使用 thread_local 的 Rhai Engine 与 AST 缓存，避免全局 Send/Sync 限制
// - 共享上下文使用 RwLock 包裹 serde_json::Value；日志聚合使用 Mutex<Vec<String>>
// - 不做冲突校验：允许并发修改；最后写回/写入按代码路径覆盖
//
fn regex_match(s: &str, pat: &str) -> bool {
Regex::new(pat).map(|re| re.is_match(s)).unwrap_or(false)
}
@ -135,7 +145,7 @@ impl FlowEngine {
pub fn builder() -> FlowEngineBuilder { FlowEngineBuilder::default() }
pub async fn drive(&self, chain: &ChainDef, ctx: serde_json::Value, opts: DriveOptions) -> anyhow::Result<(serde_json::Value, Vec<String>)> {
// 1) 选取起点
// 1) 选取起点：优先 Start；否则入度为 0；再否则第一个节点
// 查找 start：优先 Start 节点；否则选择入度为 0 的第一个节点；再否则回退第一个节点
let start = if let Some(n) = chain
.nodes
@ -208,10 +218,30 @@ async fn drive_from(
// 进入节点：打点
let node_id_str = node.id.0.clone();
let node_start = Instant::now();
{
let mut lg = logs.lock().await;
lg.push(format!("enter node: {}", node.id.0));
// 进入节点前记录当前日志长度，便于节点结束时做切片
let pre_len = { logs.lock().await.len() };
// 在每次追加日志时同步发送一条增量 SSE 事件（仅 1 行日志），以提升实时性
// push_and_emit：
// - 先将单行日志 push 到共享日志
// - 若存在 SSE 通道，截取上下文快照并发送单行增量事件
async fn push_and_emit(
logs: &std::sync::Arc<tokio::sync::Mutex<Vec<String>>>,
opts: &super::context::DriveOptions,
node_id: &str,
ctx: &std::sync::Arc<tokio::sync::RwLock<serde_json::Value>>,
msg: String,
) {
{
let mut lg = logs.lock().await;
lg.push(msg.clone());
}
if let Some(tx) = opts.event_tx.as_ref() {
let ctx_snapshot = { ctx.read().await.clone() };
crate::middlewares::sse::emit_node(&tx, node_id.to_string(), vec![msg], ctx_snapshot).await;
}
}
// enter 节点也实时推送
push_and_emit(&logs, &opts, &node_id_str, &ctx, format!("enter node: {}", node.id.0)).await;
info!(target: "udmin.flow", "enter node: {}", node.id.0);
// 执行任务
@ -221,13 +251,21 @@ async fn drive_from(
ExecutionMode::Sync => {
// 使用快照执行，结束后整体写回（允许最后写入覆盖并发修改；程序端不做冲突校验）
let mut local_ctx = { ctx.read().await.clone() };
task.execute(&node.id, node, &mut local_ctx).await?;
{ let mut w = ctx.write().await; *w = local_ctx; }
{
let mut lg = logs.lock().await;
lg.push(format!("exec task: {} (sync)", task_name));
match task.execute(&node.id, node, &mut local_ctx).await {
Ok(_) => {
{ let mut w = ctx.write().await; *w = local_ctx; }
push_and_emit(&logs, &opts, &node_id_str, &ctx, format!("exec task: {} (sync)", task_name)).await;
info!(target: "udmin.flow", "exec task: {} (sync)", task_name);
}
Err(e) => {
let err_msg = format!("task error: {}: {}", task_name, e);
push_and_emit(&logs, &opts, &node_id_str, &ctx, err_msg.clone()).await;
// 捕获快照并返回 DriveError
let ctx_snapshot = { ctx.read().await.clone() };
let logs_snapshot = { logs.lock().await.clone() };
return Err(anyhow::Error::new(DriveError { node_id: node_id_str.clone(), ctx: ctx_snapshot, logs: logs_snapshot, message: err_msg }));
}
}
info!(target: "udmin.flow", "exec task: {} (sync)", task_name);
}
ExecutionMode::AsyncFireAndForget => {
// fire-and-forget：基于快照执行，不写回共享 ctx（变量任务除外：做有界差异写回）
@ -238,6 +276,7 @@ async fn drive_from(
let node_def = node.clone();
let logs_clone = logs.clone();
let ctx_clone = ctx.clone();
let event_tx_opt = opts.event_tx.clone();
tokio::spawn(async move {
let mut c = task_ctx.clone();
let _ = task_arc.execute(&node_id, &node_def, &mut c).await;
@ -268,25 +307,31 @@ async fn drive_from(
let mut lg = logs_clone.lock().await;
lg.push(format!("exec task done (async): {} (writeback variable)", name_for_log));
}
// 实时推送异步完成日志
if let Some(tx) = event_tx_opt.as_ref() {
let ctx_snapshot = { ctx_clone.read().await.clone() };
crate::middlewares::sse::emit_node(&tx, node_id.0.clone(), vec![format!("exec task done (async): {} (writeback variable)", name_for_log)], ctx_snapshot).await;
}
info!(target: "udmin.flow", "exec task done (async): {} (writeback variable)", name_for_log);
} else {
{
let mut lg = logs_clone.lock().await;
lg.push(format!("exec task done (async): {}", name_for_log));
}
// 实时推送异步完成日志
if let Some(tx) = event_tx_opt.as_ref() {
let ctx_snapshot = { ctx_clone.read().await.clone() };
crate::middlewares::sse::emit_node(&tx, node_id.0.clone(), vec![format!("exec task done (async): {}", name_for_log)], ctx_snapshot).await;
}
info!(target: "udmin.flow", "exec task done (async): {}", name_for_log);
}
});
{
let mut lg = logs.lock().await;
lg.push(format!("spawn task: {} (async)", task_name));
}
push_and_emit(&logs, &opts, &node_id_str, &ctx, format!("spawn task: {} (async)", task_name)).await;
info!(target: "udmin.flow", "spawn task: {} (async)", task_name);
}
}
} else {
let mut lg = logs.lock().await;
lg.push(format!("task not found: {} (skip)", task_name));
push_and_emit(&logs, &opts, &node_id_str, &ctx, format!("task not found: {} (skip)", task_name)).await;
info!(target: "udmin.flow", "task not found: {} (skip)", task_name);
}
}
@ -294,11 +339,13 @@ async fn drive_from(
// End 节点：记录耗时后结束
if matches!(node.kind, NodeKind::End) {
let duration = node_start.elapsed().as_millis();
{
let mut lg = logs.lock().await;
lg.push(format!("leave node: {} {} ms", node_id_str, duration));
}
push_and_emit(&logs, &opts, &node_id_str, &ctx, format!("leave node: {} {} ms", node_id_str, duration)).await;
info!(target: "udmin.flow", "leave node: {} {} ms", node_id_str, duration);
if let Some(tx) = opts.event_tx.as_ref() {
let node_logs = { let lg = logs.lock().await; lg[pre_len..].to_vec() };
let ctx_snapshot = { ctx.read().await.clone() };
crate::middlewares::sse::emit_node(&tx, node_id_str.clone(), node_logs, ctx_snapshot).await;
}
break;
}
@ -367,7 +414,13 @@ async fn drive_from(
let mut lg = logs.lock().await;
lg.push(format!("leave node: {} {} ms", node_id_str, duration));
}
push_and_emit(&logs, &opts, &node_id_str, &ctx, format!("leave node: {} {} ms", node_id_str, duration)).await;
info!(target: "udmin.flow", "leave node: {} {} ms", node_id_str, duration);
if let Some(tx) = opts.event_tx.as_ref() {
let node_logs = { let lg = logs.lock().await; lg[pre_len..].to_vec() };
let ctx_snapshot = { ctx.read().await.clone() };
crate::middlewares::sse::emit_node(&tx, node_id_str.clone(), node_logs, ctx_snapshot).await;
}
break;
}
@ -379,6 +432,11 @@ async fn drive_from(
lg.push(format!("leave node: {} {} ms", node_id_str, duration));
}
info!(target: "udmin.flow", "leave node: {} {} ms", node_id_str, duration);
if let Some(tx) = opts.event_tx.as_ref() {
let node_logs = { let lg = logs.lock().await; lg[pre_len..].to_vec() };
let ctx_snapshot = { ctx.read().await.clone() };
crate::middlewares::sse::emit_node(&tx, node_id_str.clone(), node_logs, ctx_snapshot).await;
}
current = nexts.remove(0);
continue;
}
@ -405,6 +463,11 @@ async fn drive_from(
lg.push(format!("leave node: {} {} ms", node_id_str, duration));
}
info!(target: "udmin.flow", "leave node: {} {} ms", node_id_str, duration);
if let Some(tx) = opts.event_tx.as_ref() {
let node_logs = { let lg = logs.lock().await; lg[pre_len..].to_vec() };
let ctx_snapshot = { ctx.read().await.clone() };
crate::middlewares::sse::emit_node(&tx, node_id_str.clone(), node_logs, ctx_snapshot).await;
}
}
Ok(())
@ -427,138 +490,19 @@ impl Default for FlowEngine {
fn default() -> Self { Self { tasks: crate::flow::task::default_registry() } }
}
/* moved to executors::condition
fn eval_condition_json(ctx: &serde_json::Value, cond: &serde_json::Value) -> anyhow::Result<bool> {
-fn eval_condition_json(ctx: &serde_json::Value, cond: &serde_json::Value) -> anyhow::Result<bool> {
- // 支持前端 Condition 组件导出的: { left:{type, content}, operator, right? }
- use serde_json::Value as V;
-
- let left = cond.get("left").ok_or_else(|| anyhow::anyhow!("missing left"))?;
- let op_raw = cond.get("operator").and_then(|v| v.as_str()).unwrap_or("");
- let right = cond.get("right");
-
- let lval = resolve_value(ctx, left)?;
- let rval = match right { Some(v) => Some(resolve_value(ctx, v)?), None => None };
-
- // 归一化操作符：忽略大小写，替换下划线为空格
- let op = op_raw.trim().to_lowercase().replace('_', " ");
-
- // 工具函数
- fn to_f64(v: &V) -> Option<f64> {
- match v {
- V::Number(n) => n.as_f64(),
- V::String(s) => s.parse::<f64>().ok(),
- _ => None,
- }
- }
- fn is_empty_val(v: &V) -> bool {
- match v {
- V::Null => true,
- V::String(s) => s.trim().is_empty(),
- V::Array(a) => a.is_empty(),
- V::Object(m) => m.is_empty(),
- _ => false,
- }
- }
- fn json_equal(a: &V, b: &V) -> bool {
- match (a, b) {
- (V::Number(_), V::Number(_)) | (V::Number(_), V::String(_)) | (V::String(_), V::Number(_)) => {
- match (to_f64(a), to_f64(b)) { (Some(x), Some(y)) => x == y, _ => a == b }
- }
- _ => a == b,
- }
- }
- fn contains(left: &V, right: &V) -> bool {
- match (left, right) {
- (V::String(s), V::String(t)) => s.contains(t),
- (V::Array(arr), r) => arr.iter().any(|x| json_equal(x, r)),
- (V::Object(map), V::String(key)) => map.contains_key(key),
- _ => false,
- }
- }
- fn in_op(left: &V, right: &V) -> bool {
- match right {
- V::Array(arr) => arr.iter().any(|x| json_equal(left, x)),
- V::Object(map) => match left { V::String(k) => map.contains_key(k), _ => false },
- V::String(hay) => match left { V::String(needle) => hay.contains(needle), _ => false },
- _ => false,
- }
- }
- fn bool_like(v: &V) -> bool {
- match v {
- V::Bool(b) => *b,
- V::Null => false,
- V::Number(n) => n.as_f64().map(|x| x != 0.0).unwrap_or(false),
- V::String(s) => {
- let s_l = s.trim().to_lowercase();
- if s_l == "true" { true } else if s_l == "false" { false } else { !s_l.is_empty() }
- }
- V::Array(a) => !a.is_empty(),
- V::Object(m) => !m.is_empty(),
- }
- }
-
- let res = match (op.as_str(), &lval, &rval) {
- // 等于 / 不等于（适配所有 JSON 类型；数字按 f64 比较，其他走深度相等）
- ("equal" | "equals" | "==" | "eq", l, Some(r)) => json_equal(l, r),
- ("not equal" | "!=" | "not equals" | "neq", l, Some(r)) => !json_equal(l, r),
-
- // 数字比较
- ("greater than" | ">" | "gt", l, Some(r)) => match (to_f64(l), to_f64(r)) { (Some(a), Some(b)) => a > b, _ => false },
- ("greater than or equal" | ">=" | "gte" | "ge", l, Some(r)) => match (to_f64(l), to_f64(r)) { (Some(a), Some(b)) => a >= b, _ => false },
- ("less than" | "<" | "lt", l, Some(r)) => match (to_f64(l), to_f64(r)) { (Some(a), Some(b)) => a < b, _ => false },
- ("less than or equal" | "<=" | "lte" | "le", l, Some(r)) => match (to_f64(l), to_f64(r)) { (Some(a), Some(b)) => a <= b, _ => false },
-
- // 包含 / 不包含（字符串、数组、对象(键)）
- ("contains", l, Some(r)) => contains(l, r),
- ("not contains", l, Some(r)) => !contains(l, r),
-
- // 成员关系：left in right / not in
- ("in", l, Some(r)) => in_op(l, r),
- ("not in" | "nin", l, Some(r)) => !in_op(l, r),
-
- // 为空 / 非空（字符串、数组、对象、null）
- ("is empty" | "empty" | "isempty", l, _) => is_empty_val(l),
- ("is not empty" | "not empty" | "notempty", l, _) => !is_empty_val(l),
-
- // 布尔判断（对各类型进行布尔化）
- ("is true" | "is true?" | "istrue", l, _) => bool_like(l),
- ("is false" | "isfalse", l, _) => !bool_like(l),
-
- _ => false,
- };
- Ok(res)
-}
-
-fn resolve_value(ctx: &serde_json::Value, v: &serde_json::Value) -> anyhow::Result<serde_json::Value> {
- use serde_json::Value as V;
- let t = v.get("type").and_then(|v| v.as_str()).unwrap_or("");
- match t {
- "constant" => Ok(v.get("content").cloned().unwrap_or(V::Null)),
- "ref" => {
- // content: [nodeId, field]
- if let Some(arr) = v.get("content").and_then(|v| v.as_array()) {
- if arr.len() >= 2 {
- if let (Some(node), Some(field)) = (arr[0].as_str(), arr[1].as_str()) {
- let val = ctx
- .get("nodes")
- .and_then(|n| n.get(node))
- .and_then(|m| m.get(field))
- .cloned()
- .or_else(|| ctx.get(field).cloned())
- .unwrap_or(V::Null);
- return Ok(val);
- }
- }
- }
- Ok(V::Null)
- }
- "expression" => {
- let expr = v.get("content").and_then(|x| x.as_str()).unwrap_or("");
- if expr.trim().is_empty() { return Ok(V::Null); }
- Ok(super::engine::eval_rhai_expr_json(expr, ctx).unwrap_or(V::Null))
- }
- _ => Ok(V::Null),
#[derive(Debug, Clone)]
pub struct DriveError {
pub node_id: String,
pub ctx: serde_json::Value,
pub logs: Vec<String>,
pub message: String,
}
impl std::fmt::Display for DriveError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "{}", self.message)
}
}
*/
impl std::error::Error for DriveError {}

85
backend/src/flow/executors/http.rs
View File

@ -2,8 +2,7 @@ use async_trait::async_trait;
use serde_json::{Value, json, Map};
use tracing::info;
use std::collections::HashMap;
use std::time::Duration;
use reqwest::Certificate;
use crate::middlewares::http_client::{execute_http, HttpClientOptions, HttpRequest};
use crate::flow::task::Executor;
use crate::flow::domain::{NodeDef, NodeId};
@ -34,64 +33,29 @@ impl Executor for HttpTask {
return Ok(());
};
// 3) 解析配置
// 3) 解析配置 -> 转换为中间件请求参数
let (method, url, headers, query, body, opts) = parse_http_config(cfg)?;
info!(target = "udmin.flow", "http task: {} {}", method, url);
// 4) 发送请求（支持 HTTPS 相关选项）
let client = {
let mut builder = reqwest::Client::builder();
if let Some(ms) = opts.timeout_ms { builder = builder.timeout(Duration::from_millis(ms)); }
if opts.insecure { builder = builder.danger_accept_invalid_certs(true); }
if opts.http1_only { builder = builder.http1_only(); }
if let Some(pem) = opts.ca_pem {
if let Ok(cert) = Certificate::from_pem(pem.as_bytes()) {
builder = builder.add_root_certificate(cert);
}
}
builder.build()?
let req = HttpRequest {
method,
url,
headers,
query,
body,
};
let client_opts = HttpClientOptions {
timeout_ms: opts.timeout_ms,
insecure: opts.insecure,
ca_pem: opts.ca_pem,
http1_only: opts.http1_only,
};
let mut req = client.request(method.parse()?, url);
if let Some(hs) = headers {
use reqwest::header::{HeaderMap, HeaderName, HeaderValue};
let mut map = HeaderMap::new();
for (k, v) in hs {
if let (Ok(name), Ok(value)) = (HeaderName::from_bytes(k.as_bytes()), HeaderValue::from_str(&v)) {
map.insert(name, value);
}
}
req = req.headers(map);
}
if let Some(qs) = query {
// 将查询参数转成 (String, String) 列表，便于 reqwest 序列化
let mut pairs: Vec<(String, String)> = Vec::new();
for (k, v) in qs {
let s = match v {
Value::String(s) => s,
Value::Number(n) => n.to_string(),
Value::Bool(b) => b.to_string(),
other => other.to_string(),
};
pairs.push((k, s));
}
req = req.query(&pairs);
}
if let Some(b) = body { req = req.json(&b); }
let resp = req.send().await?;
let status = resp.status().as_u16();
let headers_out: Map<String, Value> = resp
.headers()
.iter()
.map(|(k, v)| (k.to_string(), Value::String(v.to_str().unwrap_or("").to_string())))
.collect();
// 尝试以 JSON 解析，否则退回文本
let text = resp.text().await?;
let parsed_body: Value = serde_json::from_str(&text).unwrap_or_else(|_| Value::String(text));
// 4) 调用中间件发送请求
let out = execute_http(req, client_opts).await?;
let status = out.status;
let headers_out = out.headers;
let parsed_body = out.body;
// 5) 将结果写回 ctx
let result = json!({
@ -138,8 +102,15 @@ fn parse_http_config(cfg: Value) -> anyhow::Result<(
let query = m.remove("query").and_then(|v| v.as_object().cloned());
let body = m.remove("body");
// 可选 HTTPS/超时/HTTP 版本配置
let timeout_ms = m.remove("timeout_ms").and_then(|v| v.as_u64());
// 统一解析超时配置（内联）
let timeout_ms = if let Some(ms) = m.remove("timeout_ms").and_then(|v| v.as_u64()) {
Some(ms)
} else if let Some(Value::Object(mut to)) = m.remove("timeout") {
to.remove("timeout").and_then(|v| v.as_u64())
} else {
None
};
let insecure = m.remove("insecure").and_then(|v| v.as_bool()).unwrap_or(false);
let http1_only = m.remove("http1_only").and_then(|v| v.as_bool()).unwrap_or(false);
let ca_pem = m.remove("ca_pem").and_then(|v| v.as_str().map(|s| s.to_string()));

1
backend/src/flow/executors/mod.rs
View File

@ -1,6 +1,5 @@
pub mod http;
pub mod db;
// removed: pub mod expr;
pub mod variable;
pub mod script_rhai;
pub mod script_js;

219
backend/src/flow/log_handler.rs Normal file
View File

@ -0,0 +1,219 @@
use async_trait::async_trait;
use chrono::{DateTime, FixedOffset};
use serde_json::Value;
use tokio::sync::mpsc::Sender;
use crate::flow::context::StreamEvent;
use crate::services::flow_run_log_service::{self, CreateRunLogInput};
use crate::db::Db;
/// 流程执行日志处理器抽象接口
#[async_trait]
pub trait FlowLogHandler: Send + Sync {
/// 记录流程开始执行
async fn log_start(&self, flow_id: &str, flow_code: Option<&str>, input: &Value, operator: Option<(i64, String)>) -> anyhow::Result<()>;
/// 记录流程执行失败（仅包含错误信息）
async fn log_error(&self, flow_id: &str, flow_code: Option<&str>, input: &Value, error_msg: &str, operator: Option<(i64, String)>, started_at: DateTime<FixedOffset>, duration_ms: i64) -> anyhow::Result<()>;
/// 记录流程执行失败（包含部分输出与累计日志）
async fn log_error_detail(&self, flow_id: &str, flow_code: Option<&str>, input: &Value, output: &Value, logs: &[String], error_msg: &str, operator: Option<(i64, String)>, started_at: DateTime<FixedOffset>, duration_ms: i64) -> anyhow::Result<()> {
// 默认实现：退化为仅错误信息
self.log_error(flow_id, flow_code, input, error_msg, operator, started_at, duration_ms).await
}
/// 记录流程执行成功
async fn log_success(&self, flow_id: &str, flow_code: Option<&str>, input: &Value, output: &Value, logs: &[String], operator: Option<(i64, String)>, started_at: DateTime<FixedOffset>, duration_ms: i64) -> anyhow::Result<()>;
/// 推送节点执行事件（仅SSE实现需要）
async fn emit_node_event(&self, _node_id: &str, _event_type: &str, _data: &Value) -> anyhow::Result<()> {
// 默认空实现，数据库日志处理器不需要
Ok(())
}
/// 推送完成事件（仅SSE实现需要）
async fn emit_done(&self, _success: bool, _output: &Value, _logs: &[String]) -> anyhow::Result<()> {
// 默认空实现，数据库日志处理器不需要
Ok(())
}
}
/// 数据库日志处理器
pub struct DatabaseLogHandler {
db: Db,
}
impl DatabaseLogHandler {
pub fn new(db: Db) -> Self {
Self { db }
}
}
#[async_trait]
impl FlowLogHandler for DatabaseLogHandler {
async fn log_start(&self, _flow_id: &str, _flow_code: Option<&str>, _input: &Value, _operator: Option<(i64, String)>) -> anyhow::Result<()> {
// 数据库日志处理器不需要记录开始事件，只在结束时记录
Ok(())
}
async fn log_error(&self, flow_id: &str, flow_code: Option<&str>, input: &Value, error_msg: &str, operator: Option<(i64, String)>, started_at: DateTime<FixedOffset>, duration_ms: i64) -> anyhow::Result<()> {
let (user_id, username) = operator.map(|(u, n)| (Some(u), Some(n))).unwrap_or((None, None));
flow_run_log_service::create(&self.db, CreateRunLogInput {
flow_id: flow_id.to_string(),
flow_code: flow_code.map(|s| s.to_string()),
input: Some(serde_json::to_string(input).unwrap_or_default()),
output: None,
ok: false,
logs: Some(error_msg.to_string()),
user_id,
username,
started_at,
duration_ms,
}).await.map_err(|e| anyhow::anyhow!("Failed to create error log: {}", e))?;
Ok(())
}
async fn log_error_detail(&self, flow_id: &str, flow_code: Option<&str>, input: &Value, output: &Value, logs: &[String], error_msg: &str, operator: Option<(i64, String)>, started_at: DateTime<FixedOffset>, duration_ms: i64) -> anyhow::Result<()> {
let (user_id, username) = operator.map(|(u, n)| (Some(u), Some(n))).unwrap_or((None, None));
// 将 error_msg 附加到日志尾部（若最后一条不同），确保日志中有清晰的错误描述且不重复
let mut all_logs = logs.to_vec();
if all_logs.last().map(|s| s != error_msg).unwrap_or(true) {
all_logs.push(error_msg.to_string());
}
flow_run_log_service::create(&self.db, CreateRunLogInput {
flow_id: flow_id.to_string(),
flow_code: flow_code.map(|s| s.to_string()),
input: Some(serde_json::to_string(input).unwrap_or_default()),
output: Some(serde_json::to_string(output).unwrap_or_default()),
ok: false,
logs: Some(serde_json::to_string(&all_logs).unwrap_or_default()),
user_id,
username,
started_at,
duration_ms,
}).await.map_err(|e| anyhow::anyhow!("Failed to create error log with details: {}", e))?;
Ok(())
}
async fn log_success(&self, flow_id: &str, flow_code: Option<&str>, input: &Value, output: &Value, logs: &[String], operator: Option<(i64, String)>, started_at: DateTime<FixedOffset>, duration_ms: i64) -> anyhow::Result<()> {
let (user_id, username) = operator.map(|(u, n)| (Some(u), Some(n))).unwrap_or((None, None));
flow_run_log_service::create(&self.db, CreateRunLogInput {
flow_id: flow_id.to_string(),
flow_code: flow_code.map(|s| s.to_string()),
input: Some(serde_json::to_string(input).unwrap_or_default()),
output: Some(serde_json::to_string(output).unwrap_or_default()),
ok: true,
logs: Some(serde_json::to_string(logs).unwrap_or_default()),
user_id,
username,
started_at,
duration_ms,
}).await.map_err(|e| anyhow::anyhow!("Failed to create success log: {}", e))?;
Ok(())
}
}
/// SSE日志处理器
pub struct SseLogHandler {
db: Db,
event_tx: Sender<StreamEvent>,
}
impl SseLogHandler {
pub fn new(db: Db, event_tx: Sender<StreamEvent>) -> Self {
Self { db, event_tx }
}
}
#[async_trait]
impl FlowLogHandler for SseLogHandler {
async fn log_start(&self, _flow_id: &str, _flow_code: Option<&str>, _input: &Value, _operator: Option<(i64, String)>) -> anyhow::Result<()> {
// SSE处理器也不需要记录开始事件
Ok(())
}
async fn log_error(&self, flow_id: &str, flow_code: Option<&str>, input: &Value, error_msg: &str, operator: Option<(i64, String)>, started_at: DateTime<FixedOffset>, duration_ms: i64) -> anyhow::Result<()> {
// 先推送SSE错误事件（不在此处发送 done，交由调用方统一携带 ctx/logs 发送）
crate::middlewares::sse::emit_error(&self.event_tx, error_msg.to_string()).await;
// 然后记录到数据库（仅错误信息）
let (user_id, username) = operator.map(|(u, n)| (Some(u), Some(n))).unwrap_or((None, None));
flow_run_log_service::create(&self.db, CreateRunLogInput {
flow_id: flow_id.to_string(),
flow_code: flow_code.map(|s| s.to_string()),
input: Some(serde_json::to_string(input).unwrap_or_default()),
output: None,
ok: false,
logs: Some(error_msg.to_string()),
user_id,
username,
started_at,
duration_ms,
}).await.map_err(|e| anyhow::anyhow!("Failed to create error log: {}", e))?;
Ok(())
}
async fn log_error_detail(&self, flow_id: &str, flow_code: Option<&str>, input: &Value, output: &Value, logs: &[String], error_msg: &str, operator: Option<(i64, String)>, started_at: DateTime<FixedOffset>, duration_ms: i64) -> anyhow::Result<()> {
// 先推送SSE错误事件（不在此处发送 done，交由调用方统一携带 ctx/logs 发送）
crate::middlewares::sse::emit_error(&self.event_tx, error_msg.to_string()).await;
// 然后记录到数据库（包含部分输出与累计日志），避免重复附加相同错误信息
let (user_id, username) = operator.map(|(u, n)| (Some(u), Some(n))).unwrap_or((None, None));
let mut all_logs = logs.to_vec();
if all_logs.last().map(|s| s != error_msg).unwrap_or(true) {
all_logs.push(error_msg.to_string());
}
flow_run_log_service::create(&self.db, CreateRunLogInput {
flow_id: flow_id.to_string(),
flow_code: flow_code.map(|s| s.to_string()),
input: Some(serde_json::to_string(input).unwrap_or_default()),
output: Some(serde_json::to_string(output).unwrap_or_default()),
ok: false,
logs: Some(serde_json::to_string(&all_logs).unwrap_or_default()),
user_id,
username,
started_at,
duration_ms,
}).await.map_err(|e| anyhow::anyhow!("Failed to create error log with details: {}", e))?;
Ok(())
}
async fn log_success(&self, flow_id: &str, flow_code: Option<&str>, input: &Value, output: &Value, logs: &[String], operator: Option<(i64, String)>, started_at: DateTime<FixedOffset>, duration_ms: i64) -> anyhow::Result<()> {
// 先推送SSE完成事件
crate::middlewares::sse::emit_done(&self.event_tx, true, output.clone(), logs.to_vec()).await;
// 然后记录到数据库
let (user_id, username) = operator.map(|(u, n)| (Some(u), Some(n))).unwrap_or((None, None));
flow_run_log_service::create(&self.db, CreateRunLogInput {
flow_id: flow_id.to_string(),
flow_code: flow_code.map(|s| s.to_string()),
input: Some(serde_json::to_string(input).unwrap_or_default()),
output: Some(serde_json::to_string(output).unwrap_or_default()),
ok: true,
logs: Some(serde_json::to_string(logs).unwrap_or_default()),
user_id,
username,
started_at,
duration_ms,
}).await.map_err(|e| anyhow::anyhow!("Failed to create success log: {}", e))?;
Ok(())
}
async fn emit_node_event(&self, node_id: &str, event_type: &str, data: &Value) -> anyhow::Result<()> {
// 推送节点事件到SSE
let event = StreamEvent::Node {
node_id: node_id.to_string(),
logs: vec![event_type.to_string()],
ctx: data.clone(),
};
if let Err(_e) = self.event_tx.send(event).await {
// 通道可能已关闭，忽略错误
}
Ok(())
}
async fn emit_done(&self, success: bool, output: &Value, logs: &[String]) -> anyhow::Result<()> {
crate::middlewares::sse::emit_done(&self.event_tx, success, output.clone(), logs.to_vec()).await;
Ok(())
}
}

31
backend/src/flow/mappers/http.rs
View File

@ -1,6 +1,6 @@
use serde_json::Value;
// Extract http config: method, url, headers, query, body from a node
// 从节点中提取 HTTP 配置：method、url、headers、query、body
pub fn extract_http_cfg(n: &Value) -> Option<Value> {
let data = n.get("data");
let api = data.and_then(|d| d.get("api"));
@ -28,7 +28,7 @@ pub fn extract_http_cfg(n: &Value) -> Option<Value> {
http_obj.insert("method".into(), Value::String(method));
http_obj.insert("url".into(), Value::String(url));
// Optional: headers
// 可选：headers
if let Some(hs) = api.and_then(|a| a.get("headers")).and_then(|v| v.as_object()) {
let mut heads = serde_json::Map::new();
for (k, v) in hs.iter() {
@ -41,7 +41,7 @@ pub fn extract_http_cfg(n: &Value) -> Option<Value> {
}
}
// Optional: query
// 可选：query
if let Some(qs) = api.and_then(|a| a.get("query")).and_then(|v| v.as_object()) {
let mut query = serde_json::Map::new();
for (k, v) in qs.iter() {
@ -52,7 +52,7 @@ pub fn extract_http_cfg(n: &Value) -> Option<Value> {
}
}
// Optional: body
// 可选：body
if let Some(body_obj) = data.and_then(|d| d.get("body")).and_then(|v| v.as_object()) {
if let Some(Value::Object(json_body)) = body_obj.get("json") {
http_obj.insert("body".into(), Value::Object(json_body.clone()));
@ -61,5 +61,28 @@ pub fn extract_http_cfg(n: &Value) -> Option<Value> {
}
}
// 可选：超时（统一处理：数字或对象）
if let Some(to_val) = data.and_then(|d| d.get("timeout")) {
match to_val {
Value::Number(n) => {
http_obj.insert("timeout_ms".into(), Value::Number(n.clone()));
}
Value::Object(obj) => {
// 只读访问对象中的字段并规范化
let mut t = serde_json::Map::new();
if let Some(ms) = obj.get("timeout").and_then(|v| v.as_u64()) {
t.insert("timeout".into(), Value::Number(serde_json::Number::from(ms)));
}
if let Some(rt) = obj.get("retryTimes").and_then(|v| v.as_u64()) {
t.insert("retryTimes".into(), Value::Number(serde_json::Number::from(rt)));
}
if !t.is_empty() {
http_obj.insert("timeout".into(), Value::Object(t));
}
}
_ => {}
}
}
Some(Value::Object(http_obj))
}

4
backend/src/flow/mod.rs
View File

@ -3,6 +3,6 @@ pub mod context;
pub mod task;
pub mod engine;
pub mod dsl;
// removed: pub mod storage;
pub mod executors;
pub mod mappers;
pub mod mappers;
pub mod log_handler;