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

feat(flow): 添加动态API路由支持通过流程code执行

Browse Source 
refactor(engine): 优化节点执行耗时记录
fix(db): 修正结果模式获取逻辑忽略connection.mode
style(i18n): 统一节点描述和输出模式选项的国际化
test(flow): 新增测试流程定义文件
refactor(react): 简化开发环境日志降噪处理
This commit is contained in:
ayou
2025-09-20 00:12:40 +08:00
parent 62789fce42
commit d8116ff8dc
13 changed files with 162 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@ -4,6 +4,7 @@ use futures::future::join_all;
use rhai::Engine;
use tracing::info;
use std::time::Instant;
// === 表达式评估支持：thread_local 引擎与 AST 缓存，避免全局 Sync/Send 限制 ===
use std::cell::RefCell;
@ -204,6 +205,9 @@ async fn drive_from(
steps += 1;
// 读取节点
let node = match node_map.get(&current) { Some(n) => n, None => break };
// 进入节点：打点
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));
@ -287,7 +291,16 @@ async fn drive_from(
}
}
if matches!(node.kind, NodeKind::End) { break; }
// 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));
}
info!(target: "udmin.flow", "leave node: {} {} ms", node_id_str, duration);
break;
}
// 选择下一批 link：仅在 Condition 节点上评估条件；其他节点忽略条件，直接沿第一条边前进
let mut nexts: Vec<String> = Vec::new();
@ -347,9 +360,25 @@ async fn drive_from(
}
}
if nexts.is_empty() { break; }
// 无后继：记录耗时后结束
if nexts.is_empty() {
let duration = node_start.elapsed().as_millis();
{
let mut lg = logs.lock().await;
lg.push(format!("leave node: {} {} ms", node_id_str, duration));
}
info!(target: "udmin.flow", "leave node: {} {} ms", node_id_str, duration);
break;
}
// 单分支：记录耗时后前进
if nexts.len() == 1 {
let duration = node_start.elapsed().as_millis();
{
let mut lg = logs.lock().await;
lg.push(format!("leave node: {} {} ms", node_id_str, duration));
}
info!(target: "udmin.flow", "leave node: {} {} ms", node_id_str, duration);
current = nexts.remove(0);
continue;
}
@ -369,6 +398,13 @@ async fn drive_from(
current = nexts.into_iter().next().unwrap();
// 在一个安全点等待已分支的完成（这里选择在下一轮进入前等待）
let _ = join_all(futs).await;
// 多分支：记录当前节点耗时（包含等待其他分支完成的时间）
let duration = node_start.elapsed().as_millis();
{
let mut lg = logs.lock().await;
lg.push(format!("leave node: {} {} ms", node_id_str, duration));
}
info!(target: "udmin.flow", "leave node: {} {} ms", node_id_str, duration);
}
Ok(())

4
backend/src/flow/executors/db.rs
View File

@ -25,8 +25,8 @@ impl Executor for DbTask {
// 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);
// 提前读取结果模式：仅使用 db.output.mode/db.outputMode/db.mode，忽略 connection.mode
let result_mode = mode_from_db;
info!(target = "udmin.flow", "db task: exec sql: {}", sql);
// 4) 获取连接：必须显式声明 db.connection，禁止回退到项目全局数据库，避免安全风险

60
backend/src/routes/dynamic_api.rs Normal file
View File

@ -0,0 +1,60 @@
use axum::{Router, routing::post, extract::{State, Path}, Json};
use crate::{db::Db, response::ApiResponse, services::flow_service, error::AppError};
use serde_json::Value;
use tracing::{info, error};
pub fn router() -> Router<Db> {
Router::new()
.route("/dynamic/{flow_code}", post(execute_flow))
}
async fn execute_flow(
State(db): State<Db>,
Path(flow_code): Path<String>,
Json(payload): Json<Value>
) -> Result<Json<ApiResponse<Value>>, AppError> {
info!(target = "udmin", "dynamic_api.execute_flow: start flow_code={}", flow_code);
// 1. 通过code查询流程
let flow_doc = match flow_service::get_by_code(&db, &flow_code).await {
Ok(doc) => doc,
Err(e) => {
error!(target = "udmin", error = ?e, "dynamic_api.execute_flow: flow not found flow_code={}", flow_code);
return Err(flow_service::ae(e));
}
};
// 2. 执行流程
let flow_id = flow_doc.id.clone();
info!(target = "udmin", "dynamic_api.execute_flow: found flow id={} for code={}", flow_id, flow_code);
match flow_service::run(&db, &flow_id, flow_service::RunReq { input: payload }, Some((0, "接口".to_string()))).await {
Ok(result) => {
info!(target = "udmin", "dynamic_api.execute_flow: execution successful flow_code={}", flow_code);
// 仅返回上下文中的 http_resp / http_response，如果不存在则返回空对象 {}
let ctx = result.ctx;
let data = match ctx {
Value::Object(mut map) => {
if let Some(v) = map.remove("http_resp") {
v
} else if let Some(v) = map.remove("http_response") {
v
} else {
Value::Object(serde_json::Map::new())
}
}
_ => Value::Object(serde_json::Map::new()),
};
Ok(Json(ApiResponse::ok(data)))
},
Err(e) => {
error!(target = "udmin", error = ?e, "dynamic_api.execute_flow: execution failed flow_code={}", flow_code);
let mut full = e.to_string();
for cause in e.chain().skip(1) {
full.push_str(" | ");
full.push_str(&cause.to_string());
}
Err(AppError::InternalMsg(full))
}
}
}

7
backend/src/routes/mod.rs
View File

@ -1,13 +1,13 @@
pub mod auth;
pub mod users;
pub mod roles;
pub mod menus;
pub mod departments;
pub mod logs;
// 新增岗位
pub mod positions;
pub mod menus;
pub mod logs;
pub mod flows;
pub mod flow_run_logs;
pub mod dynamic_api;
use axum::Router;
use crate::db::Db;
@ -23,4 +23,5 @@ pub fn api_router() -> Router<Db> {
.merge(flows::router())
.merge(positions::router())
.merge(flow_run_logs::router())
.merge(dynamic_api::router())
}

11
backend/src/services/flow_service.rs
View File

@ -144,6 +144,17 @@ pub async fn get(db: &Db, id: &str) -> anyhow::Result<FlowDoc> {
Ok(FlowDoc { id: row.id, yaml, design_json })
}
pub async fn get_by_code(db: &Db, code: &str) -> anyhow::Result<FlowDoc> {
let row = db_flow::Entity::find()
.filter(db_flow::Column::Code.eq(code))
.one(db)
.await?;
let row = row.ok_or_else(|| anyhow::anyhow!("flow not found with code: {}", code))?;
let yaml = row.yaml.unwrap_or_default();
let design_json = row.design_json.and_then(|s| serde_json::from_str::<serde_json::Value>(&s).ok());
Ok(FlowDoc { id: row.id, yaml, design_json })
}
pub async fn update(db: &Db, id: &str, req: FlowUpdateReq) -> anyhow::Result<FlowDoc> {
if let Some(yaml) = &req.yaml {
let _parsed: FlowDSL = serde_yaml::from_str(yaml).context("invalid flow yaml")?;

5
backend/test_flow_create.json Normal file
View File

@ -0,0 +1,5 @@
{
"name": "测试流程",
"code": "test-flow",
"yaml": "# demo flow\nnodes:\n - { id: start, kind: start, name: 开始 }\n - { id: assign, kind: custom, name: 赋值, script: ctx.x = ctx.input.name || 'default'; }\n - { id: end, kind: end, name: 结束 }\nedges:\n - { from: start, to: assign }\n - { from: assign, to: end }\n"
}

8
frontend/src/flows/form-components/form-content/index.tsx
View File

@ -5,7 +5,7 @@
import React from 'react';
import { FlowNodeRegistry } from '@flowgram.ai/free-layout-editor';
import { FlowNodeRegistry, I18n } from '@flowgram.ai/free-layout-editor';
import { useIsSidebar, useNodeRenderContext } from '../../hooks';
import { FormTitleDescription, FormWrapper } from './styles';
@ -21,7 +21,11 @@ export function FormContent(props: { children?: React.ReactNode }) {
return (
<FormWrapper>
<>
{isSidebar && <FormTitleDescription>{registry.info?.description}</FormTitleDescription>}
{isSidebar && (
<FormTitleDescription>
{I18n.t((registry.info?.description as any) || '')}
</FormTitleDescription>
)}
{(expanded || isSidebar) && props.children}
</>
</FormWrapper>

16
frontend/src/flows/form-components/form-header/title-input.tsx
View File

@ -39,10 +39,18 @@ export function TitleInput(props: {
onBlur={() => updateTitleEdit(false)}
/>
) : (
// 对默认的 Start/End 标题进行按需本地化显示
<Text ellipsis={{ showTooltip: true }}>{
value === 'Start' || value === 'End' ? I18n.t(value as any) : (value as any)
}</Text>
// 对默认的 Start/End 标题进行按需本地化显示（大小写与首尾空白规整）
<Text ellipsis={{ showTooltip: true }}>
{(() => {
const raw = (value ?? '') as string;
const norm = raw.trim().toLowerCase();
if (norm === 'start' || norm === 'end') {
const key = norm === 'start' ? 'Start' : 'End';
return I18n.t(key as any);
}
return value as any;
})()}
</Text>
)}
<Feedback errors={fieldState?.errors} />
</div>

13
frontend/src/flows/hooks/use-editor-props.tsx
View File

@ -415,8 +415,19 @@ export function useEditorProps(
'SQL': 'SQL',
'Params': '参数',
'Output Key': '输出键',
// ==== DB Node: Output Mode and options ====
'Output Mode': '输出模式',
'Rows': '行数组',
'First Row': '首行对象',
'Affected Rows': '影响行数',
},
'en-US': {
// ==== DB Node: Output Mode and options ====
'Output Mode': 'Output Mode',
'Rows': 'Rows',
'First Row': 'First Row',
'Affected Rows': 'Affected Rows',
},
'en-US': {},
},
},
plugins: () => [

6
frontend/src/flows/nodes/db/form-meta.tsx
View File

@ -186,9 +186,9 @@ export const FormRender = ({ form }: FormRenderProps<any>) => {
onChange={(v) => field.onChange(v as string)}
style={{ width: 200 }}
optionList={[
{ label: 'rows', value: 'rows' },
{ label: 'first', value: 'first' },
{ label: 'affected', value: 'affected' },
{ label: I18n.t('Rows'), value: 'rows' },
{ label: I18n.t('First Row'), value: 'first' },
{ label: I18n.t('Affected Rows'), value: 'affected' },
]}
/>
)}

2
frontend/src/flows/nodes/end/index.ts
View File

@ -24,7 +24,7 @@ export const EndNodeRegistry: FlowNodeRegistry = {
info: {
icon: iconEnd,
description:
I18n.t('The final node of the workflow, used to return the result information after the workflow is run.'),
I18n.t('流程结束节点，用于返回流程运行后的结果信息。'),
},
/**
* Render node via formMeta

2
frontend/src/flows/nodes/start/index.ts
View File

@ -25,7 +25,7 @@ export const StartNodeRegistry: FlowNodeRegistry = {
info: {
icon: iconStart,
description:
I18n.t('The starting node of the workflow, used to set the information needed to initiate the workflow.'),
I18n.t('流程开始节点，用于设置启动流程所需的信息。'),
},
/**
* Render node via formMeta

45
frontend/src/utils/react18-polyfill.ts
View File

@ -1,7 +1,7 @@
/**
* React 18 兼容性补丁 + 开发期告警修复
* - 解决第三方库使用旧版 ReactDOM.render API 的问题
* - 在开发环境下拦截并剔除部分第三方库会误透传到原生 DOM 的非标准属性
* - 在开发环境下提供最小化的日志降噪（不再篡改 React.createElement，以避免 ESM 导入赋值报错）
*/
import * as ReactDOM from 'react-dom/client'
import * as React from 'react'
@ -33,41 +33,6 @@ export function setupReact18Polyfill() {
}
}
// 开发环境：剔除被第三方库误透传到原生 DOM 的非标准属性，防止 React 告警
export function setupDevSanitizeDOMProps() {
if (!import.meta.env.DEV) return
const ANY_REACT = React as any
// 避免重复打补丁
if (ANY_REACT.__patched_createElement__) return
const origCreateElement = React.createElement as any
const STRIP_KEYS = new Set(['localeCode', 'defaultCurrency', 'showCurrencySymbol'])
const patchedCreateElement = (type: any, props: any, ...children: any[]) => {
if (typeof type === 'string' && props && typeof props === 'object') {
let mutated = false
const nextProps: any = {}
for (const key in props) {
if (Object.prototype.hasOwnProperty.call(props, key)) {
if (STRIP_KEYS.has(key)) {
mutated = true
continue
}
nextProps[key] = props[key]
}
}
return origCreateElement(type, mutated ? nextProps : props, ...children)
}
return origCreateElement(type, props, ...children)
}
;(React as any).createElement = patchedCreateElement
ANY_REACT.__patched_createElement__ = true
if (typeof console !== 'undefined' && console.debug) {
console.debug('[DEV] React.createElement patched to sanitize DOM props')
}
}
// 开发环境：抑制特定第三方库产生的已知无害告警（仅字符串匹配，不影响其他日志）
export function setupDevConsoleSuppression() {
if (!import.meta.env.DEV) return
@ -88,10 +53,8 @@ export function setupDevConsoleSuppression() {
})
.join(' ')
const hitsReactUnrecognized = joined.includes('React does not recognize the')
const hitsKnownKeys = joined.includes('localeCode') || joined.includes('defaultCurrency') || joined.includes('showCurrencySymbol')
if (hitsReactUnrecognized && hitsKnownKeys) return true
// 常见的第三方库误传非标准 DOM 属性引发的 React 告警
if (joined.includes('React does not recognize the')) return true
if (joined.includes('[DOM] Input elements should have autocomplete attributes')) return true
return false
} catch {
@ -121,6 +84,6 @@ export function setupDevConsoleSuppression() {
export function setupReactDevFixes() {
if (!import.meta.env.DEV) return
setupReact18Polyfill()
setupDevSanitizeDOMProps()
// 注意：不再对 React.createElement 打补丁，避免 ESM 导入不可变导致的构建错误
setupDevConsoleSuppression()
}