476 lines
17 KiB
Rust
476 lines
17 KiB
Rust
//! Bundled Node.js sidecar runner.
|
||
//!
|
||
//! 与 QuickJS 路径完全对齐：
|
||
//! - 脚本源码由后端 `GET /api/v1/nodejs-scripts?name=<file>.js` 下发，
|
||
//! 约定响应形如 `{ ok: true, data: { source: "..." } }`；
|
||
//! - 在 `src-tauri/resources/nodejs/node.exe`（开发期）/ 同目录
|
||
//! `resources/nodejs/node.exe`（打包后）的 Node 解释器中执行；
|
||
//! - 进度 / 日志由用户脚本通过 `globalThis.__native.emitProgress(s)`
|
||
//! / `__native.log(level, msg, fields)` 写到 stderr，Rust 这边解析后
|
||
//! 转成 `PipelineEvent` 推给前端；
|
||
//! - 用户脚本通过 `require('./xxx.js')` 引用同目录模块；Rust 在启动前
|
||
//! 按 require 图从服务端递归拉取并写入临时 bundle，Node 原生 CommonJS
|
||
//! `require` 加载（比 ESM `import` 更简单，与 `.cjs` runner 一致）；
|
||
//! - 用户脚本的最终结果通过 `globalThis.__nodejsMain(params)` 返回，runner
|
||
//! 把 JSON.stringify 后的结果以 `__RESULT__<json>` 写到 stdout。
|
||
//!
|
||
//! 这套协议有意保持极小：跟 QuickJS 的 `globalThis.__quickjsMain` / `__native`
|
||
//! 看起来完全对称，方便用户在两种运行时之间迁移脚本。
|
||
|
||
use std::collections::HashMap;
|
||
use std::path::{Path, PathBuf};
|
||
use std::process::Stdio;
|
||
|
||
use once_cell::sync::Lazy;
|
||
use regex::Regex;
|
||
use serde_json::Value;
|
||
use thiserror::Error;
|
||
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
|
||
use tokio::process::Command;
|
||
use tokio::sync::mpsc::UnboundedSender;
|
||
use uuid::Uuid;
|
||
|
||
use crate::js_runtime::PipelineEvent;
|
||
|
||
/// 与后端 `nodejs_scripts._SAFE_NAME` 一致：仅允许扁平 `.js` 文件名。
|
||
static RE_SAFE_SCRIPT: Lazy<Regex> =
|
||
Lazy::new(|| Regex::new(r"^[a-zA-Z0-9][a-zA-Z0-9._-]*\.js$").unwrap());
|
||
|
||
static RE_REQUIRE: Lazy<Regex> =
|
||
Lazy::new(|| Regex::new(r#"require\s*\(\s*['"]([^'"]+\.js)['"]\s*\)"#).unwrap());
|
||
|
||
const INLINE_ENTRY: &str = "__inline_entry__.js";
|
||
|
||
#[derive(Debug, Error)]
|
||
pub enum NodeError {
|
||
#[error("node.exe 未找到（请检查 resources/nodejs/ 是否存在）")]
|
||
NotFound,
|
||
#[error("io: {0}")]
|
||
Io(#[from] std::io::Error),
|
||
#[error("拉取脚本失败: {0}")]
|
||
FetchScript(String),
|
||
#[error("node 进程非零退出 status={status:?}, stderr={stderr}")]
|
||
NonZero { status: Option<i32>, stderr: String },
|
||
#[error("脚本未输出 __RESULT__ 行")]
|
||
NoResult,
|
||
#[error("json: {0}")]
|
||
Json(#[from] serde_json::Error),
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// 定位 node.exe
|
||
// ---------------------------------------------------------------------------
|
||
fn exe_name() -> &'static str {
|
||
if cfg!(windows) { "node.exe" } else { "node" }
|
||
}
|
||
|
||
pub fn locate_node() -> Option<PathBuf> {
|
||
if let Ok(p) = std::env::var("AICLIENT_NODE_PATH") {
|
||
let pb = PathBuf::from(p);
|
||
if pb.exists() {
|
||
return Some(pb);
|
||
}
|
||
}
|
||
if let Ok(exe) = std::env::current_exe() {
|
||
if let Some(dir) = exe.parent() {
|
||
let c = dir.join("resources").join("nodejs").join(exe_name());
|
||
if c.exists() {
|
||
return Some(c);
|
||
}
|
||
}
|
||
}
|
||
let dev = PathBuf::from("src-tauri/resources/nodejs").join(exe_name());
|
||
if dev.exists() {
|
||
return Some(dev);
|
||
}
|
||
let dev2 = PathBuf::from("resources/nodejs").join(exe_name());
|
||
if dev2.exists() {
|
||
return Some(dev2);
|
||
}
|
||
Some(PathBuf::from(exe_name()))
|
||
}
|
||
|
||
/// 用于 `NODE_PATH`：让用户脚本里的 `require('puppeteer')` 等解析到
|
||
/// `resources/nodejs/node_modules/`（需在该目录执行 `npm install`）。
|
||
pub fn nodejs_dir() -> Option<PathBuf> {
|
||
locate_node().and_then(|p| p.parent().map(|x| x.to_path_buf()))
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// Runner 模板（文件模式：入口脚本与 require 依赖均落盘，Node 原生 require）
|
||
// ---------------------------------------------------------------------------
|
||
fn build_file_runner(entry_script: &str) -> String {
|
||
format!(
|
||
r#"
|
||
// ---- aiclient auto-generated nodejs runner (do not edit) ----
|
||
"use strict";
|
||
(function () {{
|
||
function emit(obj) {{
|
||
try {{
|
||
process.stderr.write("__EVT__" + JSON.stringify(obj) + "\n");
|
||
}} catch (_) {{}}
|
||
}}
|
||
globalThis.__native = {{
|
||
emitProgress(status) {{
|
||
emit({{ type: "progress", status: String(status) }});
|
||
}},
|
||
log(level, msg, fields) {{
|
||
emit({{
|
||
type: "log",
|
||
level: String(level || "info"),
|
||
msg: String(msg),
|
||
fields: fields == null ? null : fields,
|
||
}});
|
||
}},
|
||
}};
|
||
}})();
|
||
|
||
(async () => {{
|
||
let raw = "";
|
||
process.stdin.setEncoding("utf8");
|
||
for await (const chunk of process.stdin) raw += chunk;
|
||
const params = raw ? JSON.parse(raw) : null;
|
||
|
||
require("./{entry_script}");
|
||
|
||
if (typeof globalThis.__nodejsMain !== "function") {{
|
||
process.stderr.write(
|
||
"__EVT__" + JSON.stringify({{
|
||
type: "log",
|
||
level: "error",
|
||
msg: "脚本未注册 globalThis.__nodejsMain",
|
||
}}) + "\n"
|
||
);
|
||
process.exit(1);
|
||
}}
|
||
try {{
|
||
const result = await globalThis.__nodejsMain(params);
|
||
process.stdout.write(
|
||
"__RESULT__" + JSON.stringify(result === undefined ? null : result) + "\n"
|
||
);
|
||
process.exit(0);
|
||
}} catch (e) {{
|
||
const msg = (e && (e.stack || e.message)) || String(e);
|
||
process.stderr.write(
|
||
"__EVT__" + JSON.stringify({{ type: "log", level: "error", msg }}) + "\n"
|
||
);
|
||
process.exit(2);
|
||
}}
|
||
}})();
|
||
"#
|
||
)
|
||
}
|
||
|
||
fn normalize_script_name(name: &str) -> Option<String> {
|
||
let base = Path::new(name.trim())
|
||
.file_name()
|
||
.and_then(|s| s.to_str())?;
|
||
if RE_SAFE_SCRIPT.is_match(base) {
|
||
Some(base.to_string())
|
||
} else {
|
||
None
|
||
}
|
||
}
|
||
|
||
fn extract_require_deps(source: &str) -> Vec<String> {
|
||
RE_REQUIRE
|
||
.captures_iter(source)
|
||
.filter_map(|c| c.get(1).map(|m| m.as_str().to_string()))
|
||
.collect()
|
||
}
|
||
|
||
/// 从入口脚本名出发，按 `require('*.js')` 递归拉取服务端模块。
|
||
async fn resolve_script_bundle(entry: &str) -> Result<HashMap<String, String>, NodeError> {
|
||
let entry = normalize_script_name(entry)
|
||
.ok_or_else(|| NodeError::FetchScript(format!("无效的入口脚本名: {entry}")))?;
|
||
|
||
let mut modules: HashMap<String, String> = HashMap::new();
|
||
let mut stack = vec![entry];
|
||
|
||
while let Some(name) = stack.pop() {
|
||
if modules.contains_key(&name) {
|
||
continue;
|
||
}
|
||
let source = fetch_node_script_source(&name).await?;
|
||
for dep in extract_require_deps(&source) {
|
||
if let Some(dep_name) = normalize_script_name(&dep) {
|
||
if !modules.contains_key(&dep_name) {
|
||
stack.push(dep_name);
|
||
}
|
||
}
|
||
}
|
||
modules.insert(name, source);
|
||
}
|
||
|
||
Ok(modules)
|
||
}
|
||
|
||
/// 调试源码：仅拉取其中 require 到的服务端模块，入口为内存写入的 `__inline_entry__.js`。
|
||
async fn resolve_script_bundle_for_inline(
|
||
inline_source: &str,
|
||
) -> Result<HashMap<String, String>, NodeError> {
|
||
let mut modules: HashMap<String, String> = HashMap::new();
|
||
let mut stack: Vec<String> = extract_require_deps(inline_source)
|
||
.into_iter()
|
||
.filter_map(|d| normalize_script_name(&d))
|
||
.collect();
|
||
|
||
while let Some(name) = stack.pop() {
|
||
if modules.contains_key(&name) {
|
||
continue;
|
||
}
|
||
let source = fetch_node_script_source(&name).await?;
|
||
for dep in extract_require_deps(&source) {
|
||
if let Some(dep_name) = normalize_script_name(&dep) {
|
||
if !modules.contains_key(&dep_name) {
|
||
stack.push(dep_name);
|
||
}
|
||
}
|
||
}
|
||
modules.insert(name, source);
|
||
}
|
||
|
||
modules.insert(INLINE_ENTRY.to_string(), inline_source.to_string());
|
||
Ok(modules)
|
||
}
|
||
|
||
async fn write_script_bundle(
|
||
modules: &HashMap<String, String>,
|
||
entry: &str,
|
||
) -> Result<(PathBuf, PathBuf), NodeError> {
|
||
let tmp_dir = std::env::temp_dir().join("aiclient-node");
|
||
tokio::fs::create_dir_all(&tmp_dir).await?;
|
||
let bundle_dir = tmp_dir.join(format!("bundle-{}", Uuid::new_v4()));
|
||
tokio::fs::create_dir_all(&bundle_dir).await?;
|
||
|
||
for (name, source) in modules {
|
||
tokio::fs::write(bundle_dir.join(name), source).await?;
|
||
}
|
||
|
||
let runner_path = bundle_dir.join("_runner.cjs");
|
||
let runner_src = build_file_runner(entry);
|
||
tokio::fs::write(&runner_path, runner_src).await?;
|
||
|
||
Ok((bundle_dir, runner_path))
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// 从后端拉脚本源码
|
||
// ---------------------------------------------------------------------------
|
||
fn api_base() -> String {
|
||
std::env::var("AICLIENT_API_BASE").unwrap_or_else(|_| "http://127.0.0.1:8001".to_string())
|
||
}
|
||
|
||
async fn fetch_node_script_source(script_name: &str) -> Result<String, NodeError> {
|
||
let q = urlencoding::encode(script_name);
|
||
let path = format!("/nodejs-scripts?name={q}");
|
||
let (_status, v) = crate::api_client::request_json("GET", &path, None, HashMap::new())
|
||
.await
|
||
.map_err(|e| NodeError::FetchScript(e))?;
|
||
if !v.get("ok").and_then(|x| x.as_bool()).unwrap_or(false) {
|
||
let msg = v
|
||
.get("message")
|
||
.and_then(|x| x.as_str())
|
||
.unwrap_or("未知错误");
|
||
return Err(NodeError::FetchScript(msg.to_string()));
|
||
}
|
||
let source = v
|
||
.get("data")
|
||
.and_then(|d| d.get("source"))
|
||
.and_then(|s| s.as_str())
|
||
.ok_or_else(|| NodeError::FetchScript("响应缺少 data.source".into()))?;
|
||
Ok(source.to_string())
|
||
}
|
||
|
||
/// 后端 `scripts/nodejs` 目录下的 `.js` 文件名列表。
|
||
pub async fn list_scripts() -> Result<Vec<String>, NodeError> {
|
||
let (_status, v) =
|
||
crate::api_client::request_json("GET", "/nodejs-scripts/list", None, HashMap::new())
|
||
.await
|
||
.map_err(|e| NodeError::FetchScript(e))?;
|
||
if !v.get("ok").and_then(|x| x.as_bool()).unwrap_or(false) {
|
||
let msg = v
|
||
.get("message")
|
||
.and_then(|x| x.as_str())
|
||
.unwrap_or("未知错误");
|
||
return Err(NodeError::FetchScript(msg.to_string()));
|
||
}
|
||
let names = v
|
||
.get("data")
|
||
.and_then(|d| d.get("names"))
|
||
.and_then(|n| n.as_array())
|
||
.map(|arr| {
|
||
arr.iter()
|
||
.filter_map(|x| x.as_str().map(|s| s.to_string()))
|
||
.collect::<Vec<_>>()
|
||
})
|
||
.unwrap_or_default();
|
||
Ok(names)
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// 公开入口
|
||
// ---------------------------------------------------------------------------
|
||
|
||
/// 按脚本名运行：去后端拉入口及 require 依赖 → 落盘 → 启动 node → 收集结果。
|
||
pub async fn run_node_script(
|
||
script_name: String,
|
||
params: Value,
|
||
events: UnboundedSender<PipelineEvent>,
|
||
config_env: HashMap<String, String>,
|
||
) -> Result<Value, NodeError> {
|
||
let entry = normalize_script_name(&script_name)
|
||
.ok_or_else(|| NodeError::FetchScript(format!("无效的脚本名称: {script_name}")))?;
|
||
let modules = resolve_script_bundle(&entry).await?;
|
||
run_script_bundle(&entry, modules, params, events, config_env).await
|
||
}
|
||
|
||
/// 调试入口：直接传源码；其中 `require('*.js')` 仍从服务端拉取依赖。
|
||
pub async fn run_node_script_source(
|
||
source: String,
|
||
params: Value,
|
||
events: UnboundedSender<PipelineEvent>,
|
||
config_env: HashMap<String, String>,
|
||
) -> Result<Value, NodeError> {
|
||
let modules = resolve_script_bundle_for_inline(&source).await?;
|
||
run_script_bundle(INLINE_ENTRY, modules, params, events, config_env).await
|
||
}
|
||
|
||
async fn run_script_bundle(
|
||
entry: &str,
|
||
modules: HashMap<String, String>,
|
||
params: Value,
|
||
events: UnboundedSender<PipelineEvent>,
|
||
config_env: HashMap<String, String>,
|
||
) -> Result<Value, NodeError> {
|
||
let node = locate_node().ok_or(NodeError::NotFound)?;
|
||
let node_modules_cwd = nodejs_dir().unwrap_or_else(std::env::temp_dir);
|
||
|
||
let (bundle_dir, script_path) = write_script_bundle(&modules, entry).await?;
|
||
|
||
let mut cmd = Command::new(&node);
|
||
cmd.arg(&script_path)
|
||
.current_dir(&bundle_dir)
|
||
.env(
|
||
"NODE_PATH",
|
||
node_modules_cwd
|
||
.join("node_modules")
|
||
.to_string_lossy()
|
||
.to_string(),
|
||
)
|
||
.stdin(Stdio::piped())
|
||
.stdout(Stdio::piped())
|
||
.stderr(Stdio::piped());
|
||
for (key, value) in config_env {
|
||
cmd.env(key, value);
|
||
}
|
||
if let Some(ffmpeg_path) = crate::ffmpeg::locate_ffmpeg() {
|
||
if ffmpeg_path.exists() {
|
||
cmd.env("AICLIENT_FFMPEG_PATH", ffmpeg_path);
|
||
}
|
||
}
|
||
let mut child = cmd.spawn()?;
|
||
|
||
// 把 params JSON 灌进 stdin
|
||
if let Some(mut stdin) = child.stdin.take() {
|
||
let json = serde_json::to_vec(¶ms)?;
|
||
stdin.write_all(&json).await?;
|
||
stdin.shutdown().await?;
|
||
}
|
||
|
||
// stderr → 解析 __EVT__<json> → PipelineEvent
|
||
let stderr = child.stderr.take().expect("stderr piped");
|
||
let events_tx = events.clone();
|
||
let stderr_task = tokio::spawn(async move {
|
||
let mut reader = BufReader::new(stderr).lines();
|
||
let mut tail = String::new();
|
||
while let Ok(Some(line)) = reader.next_line().await {
|
||
if let Some(rest) = line.strip_prefix("__EVT__") {
|
||
if let Ok(v) = serde_json::from_str::<Value>(rest) {
|
||
let typ = v.get("type").and_then(|x| x.as_str()).unwrap_or("");
|
||
match typ {
|
||
"progress" => {
|
||
let s = v
|
||
.get("status")
|
||
.and_then(|x| x.as_str())
|
||
.unwrap_or("")
|
||
.to_string();
|
||
let _ = events_tx.send(PipelineEvent::Progress(s));
|
||
}
|
||
"log" => {
|
||
let level = v
|
||
.get("level")
|
||
.and_then(|x| x.as_str())
|
||
.unwrap_or("info")
|
||
.to_string();
|
||
let msg = v
|
||
.get("msg")
|
||
.and_then(|x| x.as_str())
|
||
.unwrap_or("")
|
||
.to_string();
|
||
let fields = v.get("fields").cloned().filter(|x| !x.is_null());
|
||
let _ = events_tx.send(PipelineEvent::Log { level, msg, fields });
|
||
}
|
||
_ => {
|
||
// 兜底：未知 __EVT__ 类型也当 log 抛出
|
||
let _ = events_tx.send(PipelineEvent::Log {
|
||
level: "info".into(),
|
||
msg: rest.to_string(),
|
||
fields: None,
|
||
});
|
||
}
|
||
}
|
||
}
|
||
} else {
|
||
// 纯文本 stderr（比如 node 自己打印的 warning） → 当 info log
|
||
let _ = events_tx.send(PipelineEvent::Log {
|
||
level: "warn".into(),
|
||
msg: line.clone(),
|
||
fields: None,
|
||
});
|
||
tail.push_str(&line);
|
||
tail.push('\n');
|
||
}
|
||
}
|
||
tail
|
||
});
|
||
|
||
// stdout → 找 __RESULT__ 那一行
|
||
let stdout = child.stdout.take().expect("stdout piped");
|
||
let events_tx2 = events.clone();
|
||
let stdout_task = tokio::spawn(async move {
|
||
let mut reader = BufReader::new(stdout).lines();
|
||
let mut result: Option<String> = None;
|
||
while let Ok(Some(line)) = reader.next_line().await {
|
||
if let Some(rest) = line.strip_prefix("__RESULT__") {
|
||
result = Some(rest.to_string());
|
||
} else if !line.is_empty() {
|
||
// 用户 console.log 的内容也透传给前端
|
||
let _ = events_tx2.send(PipelineEvent::Log {
|
||
level: "info".into(),
|
||
msg: line,
|
||
fields: None,
|
||
});
|
||
}
|
||
}
|
||
result
|
||
});
|
||
|
||
let status = child.wait().await?;
|
||
let stderr_tail = stderr_task.await.unwrap_or_default();
|
||
let result_line = stdout_task.await.unwrap_or(None);
|
||
|
||
let _ = tokio::fs::remove_dir_all(&bundle_dir).await;
|
||
|
||
if !status.success() {
|
||
return Err(NodeError::NonZero {
|
||
status: status.code(),
|
||
stderr: stderr_tail,
|
||
});
|
||
}
|
||
|
||
let result_str = result_line.ok_or(NodeError::NoResult)?;
|
||
let v: Value = serde_json::from_str(&result_str)?;
|
||
Ok(v)
|
||
}
|