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

feat(flows): 新增流程编辑器基础功能与相关组件

Browse Source 
feat(backend): 添加流程模型与服务支持
feat(frontend): 实现流程编辑器UI与交互
feat(assets): 添加流程节点图标资源
feat(plugins): 实现上下文菜单和运行时插件
feat(components): 新增基础节点和侧边栏组件
feat(routes): 添加流程相关路由配置
feat(models): 创建流程和运行日志数据模型
feat(services): 实现流程服务层逻辑
feat(migration): 添加流程相关数据库迁移
feat(config): 更新前端配置支持流程编辑器
feat(utils): 增强axios错误处理和工具函数
This commit is contained in:
ayou
2025-09-15 00:27:13 +08:00
parent 9da3978f91
commit b0963e5e37
291 changed files with 17947 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
backend/src/flow/context.rs Normal file
View File

@ -0,0 +1,27 @@
use serde::{Deserialize, Serialize};
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct FlowContext {
#[serde(default)]
pub data: serde_json::Value,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub enum ExecutionMode {
#[serde(rename = "sync")] Sync,
#[serde(rename = "async")] AsyncFireAndForget,
}
impl Default for ExecutionMode { fn default() -> Self { ExecutionMode::Sync } }
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DriveOptions {
#[serde(default)]
pub max_steps: usize,
#[serde(default)]
pub execution_mode: ExecutionMode,
}
impl Default for DriveOptions {
fn default() -> Self { Self { max_steps: 10_000, execution_mode: ExecutionMode::Sync } }
}

44
backend/src/flow/domain.rs Normal file
View File

@ -0,0 +1,44 @@
use serde::{Deserialize, Serialize};
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Hash, Default)]
pub struct NodeId(pub String);
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum NodeKind {
Start,
End,
Task,
Decision,
}
impl Default for NodeKind {
fn default() -> Self { Self::Task }
}
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct NodeDef {
pub id: NodeId,
#[serde(default)]
pub kind: NodeKind,
#[serde(default)]
pub name: String,
#[serde(default)]
pub task: Option<String>, // 绑定的任务组件标识
}
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct LinkDef {
pub from: NodeId,
pub to: NodeId,
#[serde(default)]
pub condition: Option<String>, // 条件脚本，返回 bool
}
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct ChainDef {
#[serde(default)]
pub name: String,
pub nodes: Vec<NodeDef>,
#[serde(default)]
pub links: Vec<LinkDef>,
}

263
backend/src/flow/dsl.rs Normal file
View File

@ -0,0 +1,263 @@
use serde::{Deserialize, Serialize};
use serde_json::Value;
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FlowDSL {
#[serde(default)]
pub name: String,
#[serde(default, alias = "executionMode")]
pub execution_mode: Option<String>,
pub nodes: Vec<NodeDSL>,
#[serde(default)]
pub edges: Vec<EdgeDSL>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct NodeDSL {
pub id: String,
#[serde(default)]
pub kind: String, // start / end / task / decision
#[serde(default)]
pub name: String,
#[serde(default)]
pub task: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct EdgeDSL {
#[serde(alias = "source", alias = "from", rename = "from")]
pub from: String,
#[serde(alias = "target", alias = "to", rename = "to")]
pub to: String,
#[serde(default)]
pub condition: Option<String>,
}
impl From<FlowDSL> for super::domain::ChainDef {
fn from(v: FlowDSL) -> Self {
super::domain::ChainDef {
name: v.name,
nodes: v
.nodes
.into_iter()
.map(|n| super::domain::NodeDef {
id: super::domain::NodeId(n.id),
kind: match n.kind.to_lowercase().as_str() {
"start" => super::domain::NodeKind::Start,
"end" => super::domain::NodeKind::End,
"decision" => super::domain::NodeKind::Decision,
_ => super::domain::NodeKind::Task,
},
name: n.name,
task: n.task,
})
.collect(),
links: v
.edges
.into_iter()
.map(|e| super::domain::LinkDef {
from: super::domain::NodeId(e.from),
to: super::domain::NodeId(e.to),
condition: e.condition,
})
.collect(),
}
}
}
// ===== New: Parse design_json (free layout JSON) to ChainDef and build execution context =====
/// Build ChainDef from design_json (front-end flow JSON)
pub fn chain_from_design_json(design: &Value) -> anyhow::Result<super::domain::ChainDef> {
use super::domain::{ChainDef, NodeDef, NodeId, NodeKind, LinkDef};
// Accept both JSON object and stringified JSON
let parsed: Option<Value> = match design {
Value::String(s) => serde_json::from_str::<Value>(s).ok(),
_ => None,
};
let design = parsed.as_ref().unwrap_or(design);
let name = design
.get("name")
.and_then(|v| v.as_str())
.unwrap_or("")
.to_string();
let nodes_arr = design.get("nodes").and_then(|v| v.as_array()).cloned().unwrap_or_default();
let mut nodes: Vec<NodeDef> = Vec::new();
for n in &nodes_arr {
let id = n.get("id").and_then(|v| v.as_str()).unwrap_or_default();
let t = n.get("type").and_then(|v| v.as_str()).unwrap_or("task");
let name_field = n
.get("data")
.and_then(|d| d.get("title"))
.and_then(|v| v.as_str())
.unwrap_or("")
.to_string();
let kind = match t {
"start" => NodeKind::Start,
"end" => NodeKind::End,
"condition" => NodeKind::Decision,
_ => NodeKind::Task,
};
// Map type to task executor id (only for executable nodes). Others will be None.
let task = match t {
"http" => Some("http".to_string()),
"db" => Some("db".to_string()),
_ => None,
};
nodes.push(NodeDef { id: NodeId(id.to_string()), kind, name: name_field, task });
}
let mut links: Vec<LinkDef> = Vec::new();
if let Some(arr) = design.get("edges").and_then(|v| v.as_array()) {
for e in arr {
let from = e
.get("sourceNodeID")
.or_else(|| e.get("source"))
.or_else(|| e.get("from"))
.and_then(|v| v.as_str())
.unwrap_or_default()
.to_string();
let to = e
.get("targetNodeID")
.or_else(|| e.get("target"))
.or_else(|| e.get("to"))
.and_then(|v| v.as_str())
.unwrap_or_default()
.to_string();
// Try build structured condition for edges from a condition node via sourcePortID mapping
let mut cond: Option<String> = None;
if let Some(spid) = e.get("sourcePortID").and_then(|v| v.as_str()) {
// find source node
if let Some(src_node) = nodes_arr.iter().find(|n| n.get("id").and_then(|v| v.as_str()) == Some(from.as_str())) {
if src_node.get("type").and_then(|v| v.as_str()) == Some("condition") {
if let Some(conds) = src_node.get("data").and_then(|d| d.get("conditions")).and_then(|v| v.as_array()) {
if let Some(item) = conds.iter().find(|c| c.get("key").and_then(|v| v.as_str()) == Some(spid)) {
if let Some(val) = item.get("value") {
// store JSON string for engine to interpret at runtime
if let Ok(s) = serde_json::to_string(val) { cond = Some(s); }
}
}
}
}
}
}
links.push(LinkDef { from: NodeId(from), to: NodeId(to), condition: cond });
}
}
Ok(ChainDef { name, nodes, links })
}
/// Trim whitespace and strip wrapping quotes/backticks if present
fn sanitize_wrapped(s: &str) -> String {
let mut t = s.trim();
if t.len() >= 2 {
let bytes = t.as_bytes();
let first = bytes[0] as char;
let last = bytes[t.len() - 1] as char;
if (first == '`' && last == '`') || (first == '"' && last == '"') || (first == '\'' && last == '\'') {
t = &t[1..t.len() - 1];
t = t.trim();
// Handle stray trailing backslash left by an attempted escape of the closing quote/backtick
if t.ends_with('\\') {
t = &t[..t.len() - 1];
}
}
}
t.to_string()
}
/// Build ctx supplement from design_json: fill node-scope configs for executors, e.g., nodes.<id>.http
pub fn ctx_from_design_json(design: &Value) -> Value {
use serde_json::json;
// Accept both JSON object and stringified JSON
let parsed: Option<Value> = match design {
Value::String(s) => serde_json::from_str::<Value>(s).ok(),
_ => None,
};
let design = parsed.as_ref().unwrap_or(design);
let mut nodes_map = serde_json::Map::new();
if let Some(arr) = design.get("nodes").and_then(|v| v.as_array()) {
for n in arr {
let id = match n.get("id").and_then(|v| v.as_str()) {
Some(s) => s,
None => continue,
};
let node_type = n.get("type").and_then(|v| v.as_str()).unwrap_or("");
let mut node_cfg = serde_json::Map::new();
match node_type {
"http" => {
// Extract http config: method, url, headers, query, body
let data = n.get("data");
let api = data.and_then(|d| d.get("api"));
let method = api.and_then(|a| a.get("method")).and_then(|v| v.as_str()).unwrap_or("GET").to_string();
let url_val = api.and_then(|a| a.get("url"));
let raw_url = match url_val {
Some(Value::String(s)) => s.clone(),
Some(Value::Object(obj)) => obj.get("content").and_then(|v| v.as_str()).unwrap_or("").to_string(),
_ => String::new(),
};
let url = sanitize_wrapped(&raw_url);
if !url.is_empty() {
let mut http_obj = serde_json::Map::new();
http_obj.insert("method".into(), Value::String(method));
http_obj.insert("url".into(), Value::String(url));
// Optionally: headers/query/body
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() {
if let Some(s) = v.as_str() { heads.insert(k.clone(), Value::String(s.to_string())); }
}
if !heads.is_empty() { http_obj.insert("headers".into(), Value::Object(heads)); }
}
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() { query.insert(k.clone(), v.clone()); }
if !query.is_empty() { http_obj.insert("query".into(), Value::Object(query)); }
}
if let Some(body_obj) = data.and_then(|d| d.get("body")).and_then(|v| v.as_object()) {
// try body.content or body.json
if let Some(Value::Object(json_body)) = body_obj.get("json") { http_obj.insert("body".into(), Value::Object(json_body.clone())); }
else if let Some(Value::String(s)) = body_obj.get("content") { http_obj.insert("body".into(), Value::String(s.clone())); }
}
node_cfg.insert("http".into(), Value::Object(http_obj));
}
}
"db" => {
// Extract db config: sql, params, outputKey
let data = n.get("data");
if let Some(db_cfg) = data.and_then(|d| d.get("db")).and_then(|v| v.as_object()) {
let mut db_obj = serde_json::Map::new();
// sql can be string or object with content
let raw_sql = db_cfg.get("sql");
let sql = match raw_sql {
Some(Value::String(s)) => sanitize_wrapped(s),
Some(Value::Object(o)) => o.get("content").and_then(|v| v.as_str()).map(sanitize_wrapped).unwrap_or_default(),
_ => String::new(),
};
if !sql.is_empty() { db_obj.insert("sql".into(), Value::String(sql)); }
if let Some(p) = db_cfg.get("params") { db_obj.insert("params".into(), p.clone()); }
if let Some(Value::String(k)) = db_cfg.get("outputKey") { db_obj.insert("outputKey".into(), Value::String(k.clone())); }
if let Some(conn) = db_cfg.get("connection") { db_obj.insert("connection".into(), conn.clone()); }
if !db_obj.is_empty() { node_cfg.insert("db".into(), Value::Object(db_obj)); }
}
}
_ => {}
}
if !node_cfg.is_empty() { nodes_map.insert(id.to_string(), Value::Object(node_cfg)); }
}
}
json!({ "nodes": Value::Object(nodes_map) })
}

173
backend/src/flow/engine.rs Normal file
View File

@ -0,0 +1,173 @@
use std::collections::HashMap;
use rhai::Engine;
use tracing::info;
use super::{context::{DriveOptions, ExecutionMode}, domain::{ChainDef, NodeKind}, task::TaskRegistry};
pub struct FlowEngine {
pub tasks: TaskRegistry,
}
impl FlowEngine {
pub fn new(tasks: TaskRegistry) -> Self { Self { tasks } }
pub async fn drive(&self, chain: &ChainDef, mut ctx: serde_json::Value, opts: DriveOptions) -> anyhow::Result<(serde_json::Value, Vec<String>)> {
let mut logs = Vec::new();
// 查找 start：优先 Start 节点；否则选择入度为 0 的第一个节点；再否则回退第一个节点
let start = if let Some(n) = chain
.nodes
.iter()
.find(|n| matches!(n.kind, NodeKind::Start))
{
n.id.0.clone()
} else {
// 计算入度
let mut indeg: HashMap<&str, usize> = HashMap::new();
for n in &chain.nodes { indeg.entry(n.id.0.as_str()).or_insert(0); }
for l in &chain.links { *indeg.entry(l.to.0.as_str()).or_insert(0) += 1; }
if let Some(n) = chain.nodes.iter().find(|n| indeg.get(n.id.0.as_str()).copied().unwrap_or(0) == 0) {
n.id.0.clone()
} else {
chain
.nodes
.first()
.ok_or_else(|| anyhow::anyhow!("empty chain"))?
.id
.0
.clone()
}
};
// 邻接表（按 links 的原始顺序保序）
let mut adj: HashMap<&str, Vec<&super::domain::LinkDef>> = HashMap::new();
for l in &chain.links { adj.entry(&l.from.0).or_default().push(l); }
let node_map: HashMap<&str, &super::domain::NodeDef> = chain.nodes.iter().map(|n| (n.id.0.as_str(), n)).collect();
let mut current = start;
let mut steps = 0usize;
while steps < opts.max_steps {
steps += 1;
let node = node_map.get(current.as_str()).ok_or_else(|| anyhow::anyhow!("node not found"))?;
logs.push(format!("enter node: {}", node.id.0));
info!(target: "udmin.flow", "enter node: {}", node.id.0);
// 任务执行
if let Some(task_name) = &node.task {
if let Some(task) = self.tasks.get(task_name) {
match opts.execution_mode {
ExecutionMode::Sync => {
if let serde_json::Value::Object(obj) = &mut ctx { obj.insert("__current_node_id".to_string(), serde_json::Value::String(node.id.0.clone())); }
task.execute(&mut ctx).await?;
logs.push(format!("exec task: {} (sync)", task_name));
info!(target: "udmin.flow", "exec task: {} (sync)", task_name);
}
ExecutionMode::AsyncFireAndForget => {
// fire-and-forget: 复制一份上下文供该任务使用，主流程不等待
let mut task_ctx = ctx.clone();
if let serde_json::Value::Object(obj) = &mut task_ctx { obj.insert("__current_node_id".to_string(), serde_json::Value::String(node.id.0.clone())); }
let task_arc = task.clone();
let name_for_log = task_name.clone();
tokio::spawn(async move {
let _ = task_arc.execute(&mut task_ctx).await;
info!(target: "udmin.flow", "exec task done (async): {}", name_for_log);
});
logs.push(format!("spawn task: {} (async)", task_name));
info!(target: "udmin.flow", "spawn task: {} (async)", task_name);
}
}
} else {
logs.push(format!("task not found: {} (skip)", task_name));
info!(target: "udmin.flow", "task not found: {} (skip)", task_name);
}
}
if matches!(node.kind, NodeKind::End) { break; }
// 选择下一条 link：优先有条件的且为真；否则保序选择第一条无条件边
let mut next: Option<String> = None;
if let Some(links) = adj.get(node.id.0.as_str()) {
// 先检测条件边
for link in links.iter() {
if let Some(cond_str) = &link.condition {
// 两种情况：
// 1) 前端序列化的 JSON，形如 { left: {type, content}, operator, right? }
// 2) 直接是 rhai 表达式字符串
let ok = if cond_str.trim_start().starts_with('{') {
match serde_json::from_str::<serde_json::Value>(cond_str) {
Ok(v) => eval_condition_json(&ctx, &v).unwrap_or(false),
Err(_) => false,
}
} else {
let mut scope = rhai::Scope::new();
scope.push("ctx", rhai::serde::to_dynamic(ctx.clone()).map_err(|e| anyhow::anyhow!(e.to_string()))?);
let engine = Engine::new();
engine.eval_with_scope::<bool>(&mut scope, cond_str).unwrap_or(false)
};
if ok { next = Some(link.to.0.clone()); break; }
}
}
// 若没有命中条件边，则取第一条无条件边
if next.is_none() {
for link in links.iter() {
if link.condition.is_none() { next = Some(link.to.0.clone()); break; }
}
}
}
match next { Some(n) => current = n, None => break }
}
Ok((ctx, logs))
}
}
fn eval_condition_json(ctx: &serde_json::Value, cond: &serde_json::Value) -> anyhow::Result<bool> {
// 目前支持前端 Condition 组件导出的: { left:{type, content}, operator, right? }
let left = cond.get("left").ok_or_else(|| anyhow::anyhow!("missing left"))?;
let op = 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 };
use serde_json::Value as V;
let res = match (op, &lval, &rval) {
("contains", V::String(s), Some(V::String(t))) => s.contains(t),
("equals", V::String(s), Some(V::String(t))) => s == t,
("equals", V::Number(a), Some(V::Number(b))) => a == b,
("is_true", V::Bool(b), _) => *b,
("is_false", V::Bool(b), _) => !*b,
("gt", V::Number(a), Some(V::Number(b))) => a.as_f64().unwrap_or(0.0) > b.as_f64().unwrap_or(0.0),
("lt", V::Number(a), Some(V::Number(b))) => a.as_f64().unwrap_or(0.0) < b.as_f64().unwrap_or(0.0),
_ => 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)
}
_ => Ok(V::Null),
}
}

261
backend/src/flow/executors/db.rs Normal file
View File

@ -0,0 +1,261 @@
use async_trait::async_trait;
use serde_json::{json, Value};
use tracing::info;
use crate::flow::task::TaskComponent;
#[derive(Default)]
pub struct DbTask;
#[async_trait]
impl TaskComponent for DbTask {
async fn execute(&self, ctx: &mut Value) -> anyhow::Result<()> {
// 1) 获取当前节点ID
let node_id_opt = ctx
.get("__current_node_id")
.and_then(|v| v.as_str())
.map(|s| s.to_string());
// 2) 读取 db 配置：仅节点级 db，不再回退到全局 ctx.db，避免误用项目数据库
let cfg = match (&node_id_opt, ctx.get("nodes")) {
(Some(node_id), Some(nodes)) => nodes.get(&node_id).and_then(|n| n.get("db")).cloned(),
_ => None,
};
let Some(cfg) = cfg else {
info!(target = "udmin.flow", "db task: no config found, skip");
return Ok(());
};
// 3) 解析配置（包含可选连接信息）
let (sql, params, output_key, conn, mode_from_db) = parse_db_config(cfg)?;
// 提前读取结果模式，优先 connection.mode，其次 db.output.mode/db.outputMode/db.mode
let result_mode = get_result_mode_from_conn(&conn).or(mode_from_db);
info!(target = "udmin.flow", "db task: exec sql: {}", sql);
// 4) 获取连接：必须显式声明 db.connection，禁止回退到项目全局数据库，避免安全风险
let db: std::borrow::Cow<'_, crate::db::Db>;
let tmp_conn; // 用于在本作用域内持有临时连接
use sea_orm::{Statement, ConnectionTrait};
let conn_cfg = conn.ok_or_else(|| anyhow::anyhow!("db task: connection config is required (db.connection)"))?;
// 构造 URL 并建立临时连接
let url = extract_connection_url(conn_cfg)?;
use sea_orm::{ConnectOptions, Database};
use std::time::Duration;
let mut opt = ConnectOptions::new(url);
opt.max_connections(20)
.min_connections(1)
.connect_timeout(Duration::from_secs(8))
.idle_timeout(Duration::from_secs(120))
.sqlx_logging(true);
tmp_conn = Database::connect(opt).await?;
db = std::borrow::Cow::Owned(tmp_conn);
// 判定是否为 SELECT：简单判断前缀，允许前导空白与括号
let is_select = {
let s = sql.trim_start();
let s = s.trim_start_matches('(');
s.to_uppercase().starts_with("SELECT")
};
// 构建参数列表（支持位置和命名两种形式）
let params_vec: Vec<sea_orm::Value> = match params {
None => vec![],
Some(Value::Array(arr)) => arr.into_iter().map(json_to_db_value).collect::<anyhow::Result<_>>()?,
Some(Value::Object(obj)) => {
// 对命名参数对象，保持插入顺序不可控，这里仅将值收集为位置绑定，建议 SQL 使用 `?` 占位
obj.into_iter().map(|(_, v)| json_to_db_value(v)).collect::<anyhow::Result<_>>()?
}
Some(v) => {
// 其它类型：当作单个位置参数
vec![json_to_db_value(v)?]
}
};
let stmt = Statement::from_sql_and_values(db.get_database_backend(), &sql, params_vec);
let result = if is_select {
let rows = db.query_all(stmt).await?;
// 将 QueryResult 转换为 JSON 数组
let mut out = Vec::with_capacity(rows.len());
for row in rows {
let mut obj = serde_json::Map::new();
// 读取列名列表
let cols = row.column_names();
for (idx, col_name) in cols.iter().enumerate() {
let key = col_name.to_string();
// 尝试以通用 JSON 值提取（优先字符串、数值、布尔、二进制、null）
let val = try_get_as_json(&row, idx, &key);
obj.insert(key, val);
}
out.push(Value::Object(obj));
}
// 默认 rows 模式：直接返回数组
match result_mode.as_deref() {
// 返回首行字段对象（无则 Null）
Some("fields") | Some("first") => {
if let Some(Value::Object(m)) = out.get(0) { Value::Object(m.clone()) } else { Value::Null }
}
// 默认与显式 rows 都返回数组
_ => Value::Array(out),
}
} else {
let exec = db.execute(stmt).await?;
// 非 SELECT 默认返回受影响行数
match result_mode.as_deref() {
// 如显式要求 rows，则返回空数组
Some("rows") => json!([]),
_ => json!(exec.rows_affected()),
}
};
// 5) 写回 ctx（并对敏感信息脱敏）
let write_key = output_key.unwrap_or_else(|| "db_response".to_string());
if let (Some(node_id), Some(obj)) = (node_id_opt, ctx.as_object_mut()) {
if let Some(nodes) = obj.get_mut("nodes").and_then(|v| v.as_object_mut()) {
if let Some(target) = nodes.get_mut(&node_id).and_then(|v| v.as_object_mut()) {
// 写入结果
target.insert(write_key, result);
// 对密码字段脱敏（保留其它配置不变）
if let Some(dbv) = target.get_mut("db") {
if let Some(dbo) = dbv.as_object_mut() {
if let Some(connv) = dbo.get_mut("connection") {
match connv {
Value::Object(m) => {
if let Some(pw) = m.get_mut("password") {
*pw = Value::String("***".to_string());
}
if let Some(Value::String(url)) = m.get_mut("url") {
*url = "***".to_string();
}
}
Value::String(s) => {
*s = "***".to_string();
}
_ => {}
}
}
}
}
return Ok(());
}
}
}
if let Value::Object(map) = ctx { map.insert(write_key, result); }
Ok(())
}
}
fn parse_db_config(cfg: Value) -> anyhow::Result<(String, Option<Value>, Option<String>, Option<Value>, Option<String>)> {
match cfg {
Value::String(sql) => Ok((sql, None, None, None, None)),
Value::Object(mut m) => {
let sql = m
.remove("sql")
.and_then(|v| v.as_str().map(|s| s.to_string()))
.ok_or_else(|| anyhow::anyhow!("db config missing sql"))?;
let params = m.remove("params");
let output_key = m.remove("outputKey").and_then(|v| v.as_str().map(|s| s.to_string()));
// 在移除 connection 前，从 db 层读取可能的输出模式
let mode_from_db = {
// db.output.mode
let from_output = m.get("output").and_then(|v| v.as_object()).and_then(|o| o.get("mode")).and_then(|v| v.as_str()).map(|s| s.to_string());
// db.outputMode 或 db.mode
let from_flat = m.get("outputMode").and_then(|v| v.as_str()).map(|s| s.to_string())
.or_else(|| m.get("mode").and_then(|v| v.as_str()).map(|s| s.to_string()));
from_output.or(from_flat)
};
let conn = m.remove("connection");
// 安全策略：必须显式声明连接，禁止默认落到全局数据库
if conn.is_none() {
return Err(anyhow::anyhow!("db config missing connection (db.connection is required)"));
}
Ok((sql, params, output_key, conn, mode_from_db))
}
_ => Err(anyhow::anyhow!("invalid db config")),
}
}
fn extract_connection_url(cfg: Value) -> anyhow::Result<String> {
match cfg {
Value::String(url) => Ok(url),
Value::Object(mut m) => {
if let Some(url) = m.remove("url").and_then(|v| v.as_str().map(|s| s.to_string())) {
return Ok(url);
}
let driver = m
.remove("driver")
.and_then(|v| v.as_str().map(|s| s.to_string()))
.unwrap_or_else(|| "mysql".to_string());
// sqlite 特殊处理：仅需要 database（文件路径或 :memory:）
if driver == "sqlite" {
let database = m.remove("database").and_then(|v| v.as_str().map(|s| s.to_string())).ok_or_else(|| anyhow::anyhow!("connection.database is required for sqlite unless url provided"))?;
return Ok(format!("sqlite://{}", database));
}
let host = m.remove("host").and_then(|v| v.as_str().map(|s| s.to_string())).unwrap_or_else(|| "localhost".to_string());
let port = m.remove("port").map(|v| match v { Value::Number(n) => n.to_string(), Value::String(s) => s, _ => String::new() });
let database = m.remove("database").and_then(|v| v.as_str().map(|s| s.to_string())).ok_or_else(|| anyhow::anyhow!("connection.database is required unless url provided"))?;
let username = m.remove("username").and_then(|v| v.as_str().map(|s| s.to_string())).ok_or_else(|| anyhow::anyhow!("connection.username is required unless url provided"))?;
let password = m.remove("password").and_then(|v| v.as_str().map(|s| s.to_string())).unwrap_or_default();
let port_part = port.filter(|s| !s.is_empty()).map(|s| format!(":{}", s)).unwrap_or_default();
let url = format!(
"{}://{}:{}@{}{}{}",
driver,
percent_encoding::utf8_percent_encode(&username, percent_encoding::NON_ALPHANUMERIC),
percent_encoding::utf8_percent_encode(&password, percent_encoding::NON_ALPHANUMERIC),
host,
port_part,
format!("/{}", database)
);
Ok(url)
}
_ => Err(anyhow::anyhow!("invalid connection config")),
}
}
fn get_result_mode_from_conn(conn: &Option<Value>) -> Option<String> {
match conn {
Some(Value::Object(m)) => m.get("mode").and_then(|v| v.as_str()).map(|s| s.to_string()),
_ => None,
}
}
fn json_to_db_value(v: Value) -> anyhow::Result<sea_orm::Value> {
use sea_orm::Value as DbValue;
let dv = match v {
Value::Null => DbValue::String(None),
Value::Bool(b) => DbValue::Bool(Some(b)),
Value::Number(n) => {
if let Some(i) = n.as_i64() { DbValue::BigInt(Some(i)) }
else if let Some(u) = n.as_u64() { DbValue::BigUnsigned(Some(u)) }
else if let Some(f) = n.as_f64() { DbValue::Double(Some(f)) }
else { DbValue::String(None) }
}
Value::String(s) => DbValue::String(Some(Box::new(s))),
Value::Array(arr) => {
// 无通用跨库数组类型：存为 JSON 字符串
let s = serde_json::to_string(&Value::Array(arr))?;
DbValue::String(Some(Box::new(s)))
}
Value::Object(obj) => {
let s = serde_json::to_string(&Value::Object(obj))?;
DbValue::String(Some(Box::new(s)))
}
};
Ok(dv)
}
fn try_get_as_json(row: &sea_orm::QueryResult, idx: usize, col_name: &str) -> Value {
use sea_orm::TryGetable;
// 尝试多种基础类型
if let Ok(v) = row.try_get::<Option<String>>("", col_name) { return v.map(Value::String).unwrap_or(Value::Null); }
if let Ok(v) = row.try_get::<Option<i64>>("", col_name) { return v.map(|x| json!(x)).unwrap_or(Value::Null); }
if let Ok(v) = row.try_get::<Option<u64>>("", col_name) { return v.map(|x| json!(x)).unwrap_or(Value::Null); }
if let Ok(v) = row.try_get::<Option<f64>>("", col_name) { return v.map(|x| json!(x)).unwrap_or(Value::Null); }
if let Ok(v) = row.try_get::<Option<bool>>("", col_name) { return v.map(|x| json!(x)).unwrap_or(Value::Null); }
// 回退：按索引读取成字符串
if let Ok(v) = row.try_get_by_index::<Option<String>>(idx) { return v.map(Value::String).unwrap_or(Value::Null); }
Value::Null
}

161
backend/src/flow/executors/http.rs Normal file
View File

@ -0,0 +1,161 @@
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::flow::task::TaskComponent;
#[derive(Default)]
pub struct HttpTask;
#[derive(Default, Clone)]
struct HttpOpts {
timeout_ms: Option<u64>,
insecure: bool,
ca_pem: Option<String>,
http1_only: bool,
}
#[async_trait]
impl TaskComponent for HttpTask {
async fn execute(&self, ctx: &mut Value) -> anyhow::Result<()> {
// 1) 读取当前节点ID（由引擎在执行前写入 ctx.__current_node_id）
let node_id_opt = ctx
.get("__current_node_id")
.and_then(|v| v.as_str())
.map(|s| s.to_string());
// 2) 从 ctx 中提取 http 配置
// 优先 nodes.<node_id>.http，其次全局 http
let cfg = match (&node_id_opt, ctx.get("nodes")) {
(Some(node_id), Some(nodes)) => nodes.get(&node_id).and_then(|n| n.get("http")).cloned(),
_ => None,
}.or_else(|| ctx.get("http").cloned());
let Some(cfg) = cfg else {
// 未提供配置，直接跳过（也可选择返回错误）
info!(target = "udmin.flow", "http task: no config found, skip");
return Ok(());
};
// 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 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));
// 5) 将结果写回 ctx
let result = json!({
"status": status,
"headers": headers_out,
"body": parsed_body,
});
// 优先写 nodes.<node_id>.http_response，否则写入全局 http_response
if let (Some(node_id), Some(obj)) = (node_id_opt, ctx.as_object_mut()) {
if let Some(nodes) = obj.get_mut("nodes").and_then(|v| v.as_object_mut()) {
if let Some(target) = nodes.get_mut(&node_id).and_then(|v| v.as_object_mut()) {
target.insert("http_response".to_string(), result);
return Ok(());
}
}
}
// 退回：写入全局
if let Value::Object(map) = ctx {
map.insert("http_response".to_string(), result);
}
Ok(())
}
}
fn parse_http_config(cfg: Value) -> anyhow::Result<(
String,
String,
Option<HashMap<String, String>>,
Option<Map<String, Value>>,
Option<Value>,
HttpOpts,
)> {
// 支持两种配置：
// 1) 字符串：视为 URL，方法 GET
// 2) 对象：{ method, url, headers, query, body }
match cfg {
Value::String(url) => Ok(("GET".into(), url, None, None, None, HttpOpts::default())),
Value::Object(mut m) => {
let method = m.remove("method").and_then(|v| v.as_str().map(|s| s.to_uppercase())).unwrap_or_else(|| "GET".into());
let url = m.remove("url").and_then(|v| v.as_str().map(|s| s.to_string()))
.ok_or_else(|| anyhow::anyhow!("http config missing url"))?;
let headers = m.remove("headers").and_then(|v| v.as_object().cloned()).map(|obj| {
obj.into_iter().filter_map(|(k, v)| v.as_str().map(|s| (k, s.to_string()))).collect::<HashMap<String, String>>()
});
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 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()));
let opts = HttpOpts { timeout_ms, insecure, ca_pem, http1_only };
Ok((method, url, headers, query, body, opts))
}
_ => Err(anyhow::anyhow!("invalid http config")),
}
}

2
backend/src/flow/executors/mod.rs Normal file
View File

@ -0,0 +1,2 @@
pub mod http;
pub mod db;

7
backend/src/flow/mod.rs Normal file
View File

@ -0,0 +1,7 @@
pub mod domain;
pub mod context;
pub mod task;
pub mod engine;
pub mod dsl;
pub mod storage;
pub mod executors;

15
backend/src/flow/storage.rs Normal file
View File

@ -0,0 +1,15 @@
use once_cell::sync::Lazy;
use std::collections::HashMap;
use std::sync::Mutex;
static STORE: Lazy<Mutex<HashMap<String, String>>> = Lazy::new(|| Mutex::new(HashMap::new()));
pub fn list() -> Vec<(String, String)> {
STORE.lock().unwrap().iter().map(|(k, v)| (k.clone(), v.clone())).collect()
}
pub fn get(id: &str) -> Option<String> { STORE.lock().unwrap().get(id).cloned() }
pub fn put(id: String, yaml: String) { STORE.lock().unwrap().insert(id, yaml); }
pub fn del(id: &str) -> Option<String> { STORE.lock().unwrap().remove(id) }

41
backend/src/flow/task.rs Normal file
View File

@ -0,0 +1,41 @@
use async_trait::async_trait;
use serde_json::Value;
#[async_trait]
pub trait TaskComponent: Send + Sync {
async fn execute(&self, ctx: &mut Value) -> anyhow::Result<()>;
}
pub type TaskRegistry = std::collections::HashMap<String, std::sync::Arc<dyn TaskComponent>>;
use std::sync::{Arc, RwLock, OnceLock};
pub fn default_registry() -> TaskRegistry {
let mut reg: TaskRegistry = TaskRegistry::new();
reg.insert("http".into(), Arc::new(crate::flow::executors::http::HttpTask::default()));
reg.insert("db".into(), Arc::new(crate::flow::executors::db::DbTask::default()));
reg
}
// ===== Global registry (for DI/registry center) =====
static GLOBAL_TASK_REGISTRY: OnceLock<RwLock<TaskRegistry>> = OnceLock::new();
/// Get a snapshot of current registry (clone of HashMap). If not initialized, it will be filled with default_registry().
pub fn get_registry() -> TaskRegistry {
let lock = GLOBAL_TASK_REGISTRY.get_or_init(|| RwLock::new(default_registry()));
lock.read().expect("lock poisoned").clone()
}
/// Register/override a single task into global registry.
pub fn register_global_task(name: impl Into<String>, task: Arc<dyn TaskComponent>) {
let lock = GLOBAL_TASK_REGISTRY.get_or_init(|| RwLock::new(default_registry()));
let mut w = lock.write().expect("lock poisoned");
w.insert(name.into(), task);
}
/// Initialize or mutate the global registry with a custom initializer.
pub fn init_global_registry_with(init: impl FnOnce(&mut TaskRegistry)) {
let lock = GLOBAL_TASK_REGISTRY.get_or_init(|| RwLock::new(default_registry()));
let mut w = lock.write().expect("lock poisoned");
init(&mut w);
}