refactor(workflow_ui): 抽出 queue 子模块

Co-Authored-By: Claude <noreply@anthropic.com>

src/routes/workflow_ui/mod.rs

@@ -4,7 +4,7 @@ use crate::{
     services::workspace::{self, WorkspaceAssetInput},
     workflow::{
         models::{RunContext, WorkflowGraph},
-        registry, storage,
+        registry,
     },
     AppState,
 };
@@ -16,7 +16,7 @@ use axum::{
         DefaultBodyLimit, FromRequest, Path, Query, State,
     },
     http::{header, HeaderMap, HeaderValue, Method, StatusCode, Uri},
-    response::{Html, IntoResponse, Response},
+    response::{IntoResponse, Response},
     routing::{any, get, post},
     Json, Router,
 };
@@ -24,7 +24,6 @@ use include_dir::{include_dir, Dir};
 use once_cell::sync::Lazy;
 use serde::{Deserialize, Serialize};
 use serde_json::{json, Map, Value};
-use sqlx::Row;
 use std::{
     collections::{HashMap, HashSet},
     path::{Path as StdPath, PathBuf},
@@ -63,22 +62,22 @@ struct CachedJson {
 }
 
 #[derive(Debug, Deserialize, Serialize)]
-struct PromptPayload {
+pub(super) struct PromptPayload {
-    prompt: Value,
+    pub(super) prompt: Value,
     #[serde(default)]
-    extra_data: Value,
+    pub(super) extra_data: Value,
     #[serde(default)]
-    client_id: Option<String>,
+    pub(super) client_id: Option<String>,
     #[serde(default)]
-    front: Option<bool>,
+    pub(super) front: Option<bool>,
     #[serde(default)]
-    number: Option<i64>,
+    pub(super) number: Option<i64>,
 }
 
 #[derive(Debug, Deserialize)]
-struct HistoryQuery {
+pub(super) struct HistoryQuery {
     #[serde(default = "default_history_limit")]
-    max_items: i64,
+    pub(super) max_items: i64,
 }
 
 #[derive(Debug, Deserialize)]
@@ -106,11 +105,11 @@ struct UploadQuery {
 }
 
 #[derive(Debug, Deserialize, Serialize)]
-struct QueueMutation {
+pub(super) struct QueueMutation {
     #[serde(default)]
-    delete: Vec<String>,
+    pub(super) delete: Vec<String>,
     #[serde(default)]
-    clear: bool,
+    pub(super) clear: bool,
 }
 
 #[derive(Debug, Deserialize)]
@@ -124,7 +123,7 @@ use auth::{auth_caller, caller_from_headers, caller_is_admin, caller_user_id};
 mod node_defs;
 use node_defs::{
     core_node_definitions, core_workflow_template_index, default_boosapi_workflow_graph,
-    node_type_def_to_comfy, normalize_prompt_payload, text_to_video_workflow_graph,
+    node_type_def_to_comfy, text_to_video_workflow_graph,
 };
 mod models;
 use models::{
@@ -145,6 +144,11 @@ use comfy_proxy::{
 };
 mod validate;
 use validate::validate_workflow;
+mod queue;
+use queue::{
+    history, history_by_id, history_mutate, interrupt, internal_logs, internal_logs_raw, queue,
+    queue_mutate, queue_prompt, queue_status, repair_native_prompt_payload, view,
+};
 
 pub fn routes() -> Router<AppState> {
     Router::new()
@@ -252,7 +256,7 @@ pub(super) fn user_home(user_id: Option<i64>) -> PathBuf {
     )
 }
 
-fn guess_mime_from_path(path: &PathBuf) -> &'static str {
+pub(super) fn guess_mime_from_path(path: &PathBuf) -> &'static str {
     match path
         .extension()
         .and_then(|ext| ext.to_str())
@@ -572,7 +576,7 @@ fn normalize_model_folder_payload(value: &mut Value) {
     }
 }
 
-fn prompt_contains_boosapi_nodes(prompt: &Value) -> bool {
+pub(super) fn prompt_contains_boosapi_nodes(prompt: &Value) -> bool {
     prompt
         .as_object()
         .map(|nodes| {
@@ -889,137 +893,7 @@ pub(super) fn workflow_validation_hints(errors: usize, warnings: usize) -> Vec<S
     hints
 }
 
-async fn queue_status(State(state): State<AppState>, headers: HeaderMap) -> AppResult<Json<Value>> {
+pub(super) fn workflow_run_to_history(run: crate::workflow::models::WorkflowRunSummary) -> Value {
-    let _ = auth_caller_optional(&state, &headers);
-    if state.config.comfyui.uses_native_backend() {
-        return native_workflow_json(&state, Method::GET, "/prompt", None).await;
-    }
-    Ok(Json(json!({
-        "status_str": "success",
-        "completed": true,
-        "messages": []
-    })))
-}
-
-async fn queue(State(state): State<AppState>, headers: HeaderMap) -> AppResult<Json<Value>> {
-    if state.config.comfyui.uses_native_backend() {
-        let _ = auth_caller_optional(&state, &headers);
-        return native_workflow_json(&state, Method::GET, "/queue", None).await;
-    }
-    let caller =
-        auth_caller_optional(&state, &headers).unwrap_or(WorkflowCaller::User { user_id: 0 });
-    let rows = storage::list_runs(
-        &state.db,
-        caller_user_id(&caller),
-        caller_is_admin(&caller),
-        25,
-    )
-    .await?;
-    let running = rows
-        .into_iter()
-        .filter(|run| run.status == "running")
-        .map(workflow_run_to_queue_task)
-        .collect::<Vec<_>>();
-    Ok(Json(json!({
-        "Running": running,
-        "Pending": []
-    })))
-}
-
-async fn queue_mutate(
-    State(state): State<AppState>,
-    headers: HeaderMap,
-    Json(_payload): Json<QueueMutation>,
-) -> AppResult<Json<Value>> {
-    let _ = auth_caller(&state, &headers).await?;
-    if state.config.comfyui.uses_native_backend() {
-        let value = serde_json::to_value(&_payload).unwrap_or_else(|_| json!({}));
-        return native_workflow_json(&state, Method::POST, "/queue", Some(value)).await;
-    }
-    Ok(Json(json!({
-        "success": true,
-        "Running": [],
-        "Pending": []
-    })))
-}
-
-async fn history(
-    State(state): State<AppState>,
-    headers: HeaderMap,
-    Query(query): Query<HistoryQuery>,
-) -> AppResult<Json<Value>> {
-    if state.config.comfyui.uses_native_backend() {
-        let _ = auth_caller_optional(&state, &headers);
-        let path = format!("/history?max_items={}", query.max_items);
-        return native_workflow_json(&state, Method::GET, &path, None).await;
-    }
-    let caller =
-        auth_caller_optional(&state, &headers).unwrap_or(WorkflowCaller::User { user_id: 0 });
-    let rows = storage::list_runs(
-        &state.db,
-        caller_user_id(&caller),
-        caller_is_admin(&caller),
-        query.max_items,
-    )
-    .await?;
-    let history = rows
-        .into_iter()
-        .map(workflow_run_to_history_task)
-        .collect::<Vec<_>>();
-    Ok(Json(json!({ "History": history })))
-}
-
-async fn history_mutate(
-    State(state): State<AppState>,
-    headers: HeaderMap,
-    Json(payload): Json<QueueMutation>,
-) -> AppResult<Json<Value>> {
-    let caller = auth_caller(&state, &headers).await?;
-    if state.config.comfyui.uses_native_backend() {
-        let value = serde_json::to_value(&payload).unwrap_or_else(|_| json!({}));
-        return native_workflow_json(&state, Method::POST, "/history", Some(value)).await;
-    }
-    let affected = if payload.clear {
-        storage::clear_runs(&state, caller_user_id(&caller), caller_is_admin(&caller)).await?
-            as usize
-    } else {
-        storage::delete_runs(
-            &state,
-            &payload.delete,
-            caller_user_id(&caller),
-            caller_is_admin(&caller),
-        )
-        .await?
-    };
-    Ok(Json(json!({
-        "success": true,
-        "deleted": affected
-    })))
-}
-
-async fn history_by_id(
-    State(state): State<AppState>,
-    headers: HeaderMap,
-    Path(id): Path<String>,
-) -> AppResult<Json<Value>> {
-    if state.config.comfyui.uses_native_backend() {
-        let _ = auth_caller_optional(&state, &headers);
-        let path = format!("/history/{}", simple_urlencode(&id));
-        return native_workflow_json(&state, Method::GET, &path, None).await;
-    }
-    let caller =
-        auth_caller_optional(&state, &headers).unwrap_or(WorkflowCaller::User { user_id: 0 });
-    let run = storage::get_run(
-        &state.db,
-        &id,
-        caller_user_id(&caller),
-        caller_is_admin(&caller),
-    )
-    .await?;
-    Ok(Json(workflow_run_to_history(run)))
-}
-
-fn workflow_run_to_history(run: crate::workflow::models::WorkflowRunSummary) -> Value {
     let outputs = run_outputs_for_comfy(&run);
     let status_str = if run.status == "failed" {
         "error"
@@ -1038,7 +912,7 @@ fn workflow_run_to_history(run: crate::workflow::models::WorkflowRunSummary) ->
     })
 }
 
-fn workflow_run_to_history_task(run: crate::workflow::models::WorkflowRunSummary) -> Value {
+pub(super) fn workflow_run_to_history_task(run: crate::workflow::models::WorkflowRunSummary) -> Value {
     let history = workflow_run_to_history(run);
     json!({
         "taskType": "History",
@@ -1048,7 +922,7 @@ fn workflow_run_to_history_task(run: crate::workflow::models::WorkflowRunSummary
     })
 }
 
-fn workflow_run_to_queue_task(run: crate::workflow::models::WorkflowRunSummary) -> Value {
+pub(super) fn workflow_run_to_queue_task(run: crate::workflow::models::WorkflowRunSummary) -> Value {
     json!({
         "taskType": "Running",
         "prompt": workflow_run_to_prompt_tuple(&run),
@@ -1061,7 +935,7 @@ fn workflow_run_to_queue_task(run: crate::workflow::models::WorkflowRunSummary)
     })
 }
 
-fn prompt_error_response(message: impl Into<String>, details: impl Into<String>) -> Response {
+pub(super) fn prompt_error_response(message: impl Into<String>, details: impl Into<String>) -> Response {
     let message = message.into();
     let details = details.into();
     (
@@ -1085,58 +959,6 @@ fn prompt_error_response(message: impl Into<String>, details: impl Into<String>)
         .into_response()
 }
 
-async fn queue_prompt(
-    State(state): State<AppState>,
-    headers: HeaderMap,
-    Json(mut payload): Json<PromptPayload>,
-) -> AppResult<Response> {
-    let caller = auth_caller(&state, &headers).await?;
-    if state.config.comfyui.uses_native_backend() {
-        if state.config.comfyui.uses_cloud_backend() {
-            repair_native_prompt_payload(&state, &mut payload).await;
-            return native_workflow_response(
-                &state,
-                Method::POST,
-                "/prompt",
-                Some(serde_json::to_value(payload).unwrap_or_else(|_| json!({}))),
-            )
-            .await;
-        }
-        if prompt_contains_boosapi_nodes(&payload.prompt) {
-            return queue_boosapi_prompt(state, caller, payload).await;
-        }
-        repair_native_prompt_payload(&state, &mut payload).await;
-        return native_workflow_response(
-            &state,
-            Method::POST,
-            "/prompt",
-            Some(serde_json::to_value(payload).unwrap_or_else(|_| json!({}))),
-        )
-        .await;
-    }
-    queue_boosapi_prompt(state, caller, payload).await
-}
-
-async fn repair_native_prompt_payload(state: &AppState, payload: &mut PromptPayload) {
-    inject_native_prompt_cloud_api_key(state, payload);
-    let object_info = native_workflow_json_cached(
-        state,
-        Method::GET,
-        "/object_info",
-        None,
-        native_workflow_metadata_ttl(state),
-    )
-    .await
-    .ok()
-    .map(|value| value.0);
-    if let Some(object_info) = object_info.as_ref() {
-        repair_dynamic_combo_nodes(&mut payload.prompt, object_info);
-    }
-    repair_resize_image_mask_nodes(&mut payload.prompt, &payload.extra_data);
-    repair_text_generate_ltx2_prompt_nodes(&mut payload.prompt);
-    repair_wan2_api_nodes(&mut payload.prompt);
-}
-
 fn inject_native_prompt_cloud_api_key(state: &AppState, payload: &mut PromptPayload) {
     if !state.config.comfyui.uses_cloud_backend() {
         return;
@@ -1154,7 +976,7 @@ fn inject_native_prompt_cloud_api_key(state: &AppState, payload: &mut PromptPayl
     );
 }
 
-fn repair_resize_image_mask_nodes(prompt: &mut Value, extra_data: &Value) {
+pub(super) fn repair_resize_image_mask_nodes(prompt: &mut Value, extra_data: &Value) {
     let Some(prompt_nodes) = prompt.as_object_mut() else {
         return;
     };
@@ -1253,7 +1075,7 @@ fn repair_resize_image_mask_node_inputs_without_workflow(
     );
 }
 
-fn repair_text_generate_ltx2_prompt_nodes(prompt: &mut Value) {
+pub(super) fn repair_text_generate_ltx2_prompt_nodes(prompt: &mut Value) {
     let Some(prompt_nodes) = prompt.as_object_mut() else {
         return;
     };
@@ -1305,7 +1127,7 @@ struct DynamicComboField {
     max: Option<f64>,
 }
 
-fn repair_dynamic_combo_nodes(prompt: &mut Value, object_info: &Value) {
+pub(super) fn repair_dynamic_combo_nodes(prompt: &mut Value, object_info: &Value) {
     let Some(prompt_nodes) = prompt.as_object_mut() else {
         return;
     };
@@ -1591,7 +1413,7 @@ fn clamp_f64_to_field(value: f64, field: &DynamicComboField) -> f64 {
     value.clamp(min, max)
 }
 
-fn repair_wan2_api_nodes(prompt: &mut Value) {
+pub(super) fn repair_wan2_api_nodes(prompt: &mut Value) {
     let Some(prompt_nodes) = prompt.as_object_mut() else {
         return;
     };
@@ -1777,221 +1599,6 @@ fn id_to_prompt_key(id: &Value) -> String {
     id.as_str().unwrap_or_default().to_string()
 }
 
-async fn queue_boosapi_prompt(
-    state: AppState,
-    caller: WorkflowCaller,
-    payload: PromptPayload,
-) -> AppResult<Response> {
-    let prompt = payload.prompt.clone();
-    let graph = match normalize_prompt_payload(&prompt) {
-        Ok(graph) => graph,
-        Err(err) => {
-            return Ok(prompt_error_response("工作流无法执行", err.to_string()));
-        }
-    };
-    if graph.nodes.is_empty() {
-        return Ok(prompt_error_response(
-            "画布里还没有可执行节点",
-            "请先从左侧节点库添加 BoosAPI 节点并连接后再执行。",
-        ));
-    }
-    let name = payload
-        .extra_data
-        .pointer("/extra_pnginfo/workflow/name")
-        .and_then(Value::as_str)
-        .unwrap_or("ComfyUI workflow");
-    let comfy_workflow = payload
-        .extra_data
-        .pointer("/extra_pnginfo/workflow")
-        .cloned()
-        .unwrap_or_else(|| workflow_graph_to_comfy_workflow(&graph));
-    let workflow = storage::insert_workflow(
-        &state,
-        caller_user_id(&caller),
-        name,
-        Some("Imported from ComfyUI frontend"),
-        &serde_json::to_value(&graph).unwrap_or_else(|_| json!({})),
-        "private",
-        false,
-    )
-    .await?;
-
-    let ctx = workflow_context(&state, &caller).await?;
-    let progress_session_id = payload
-        .client_id
-        .clone()
-        .filter(|value| !value.trim().is_empty())
-        .unwrap_or_else(|| "boosapi-workflow".to_string());
-    let (run_tx, run_rx) = mpsc::sync_channel::<String>(1);
-    let progress =
-        workflow_progress_callback_with_run_sender(Some(progress_session_id.clone()), Some(run_tx));
-    let state_for_task = state.clone();
-    let graph_value = serde_json::to_value(&graph).unwrap_or_else(|_| json!({}));
-    let task_graph_value = graph_value.clone();
-    let user_id = caller_user_id(&caller);
-    let allow_admin = caller_is_admin(&caller);
-    let session_for_task = progress_session_id.clone();
-    tokio::spawn(async move {
-        let result = crate::workflow::executor::run_workflow_with_progress(
-            &state_for_task,
-            workflow.id,
-            task_graph_value,
-            json!({}),
-            ctx,
-            progress,
-        )
-        .await;
-        match result {
-            Ok(run_id) => {
-                if let Ok(run) =
-                    storage::get_run(&state_for_task.db, &run_id, user_id, allow_admin).await
-                {
-                    let outputs = run_outputs_for_comfy(&run);
-                    publish_run_events(Some(session_for_task.as_str()), &run, &outputs);
-                }
-            }
-            Err(err) => {
-                tracing::warn!("workflow background execution failed: {err}");
-            }
-        }
-    });
-    let run_id = match tokio::task::spawn_blocking(move || run_rx.recv()).await {
-        Ok(Ok(run_id)) => run_id,
-        Ok(Err(_)) | Err(_) => {
-            return Ok(prompt_error_response(
-                "工作流执行失败",
-                "后台任务没有成功进入队列，请稍后重试。",
-            ));
-        }
-    };
-
-    Ok(Json(json!({
-        "prompt_id": run_id,
-        "number": payload.number.unwrap_or(0),
-        "node_errors": [],
-        "client_id": payload.client_id,
-        "front": payload.front.unwrap_or(false),
-        "outputs": {},
-        "workflow": comfy_workflow
-    }))
-    .into_response())
-}
-
-async fn interrupt(State(state): State<AppState>, headers: HeaderMap) -> AppResult<Json<Value>> {
-    let _ = auth_caller_optional(&state, &headers);
-    if state.config.comfyui.uses_native_backend() {
-        return native_workflow_json(&state, Method::POST, "/interrupt", Some(json!({}))).await;
-    }
-    Ok(Json(json!({ "success": true })))
-}
-
-async fn view(
-    State(state): State<AppState>,
-    headers: HeaderMap,
-    Query(params): Query<HashMap<String, String>>,
-) -> AppResult<Response> {
-    let caller = auth_caller(&state, &headers).await?;
-    if state.config.comfyui.uses_native_backend() {
-        return native_workflow_view_response(&state, &caller, &params).await;
-    }
-    let file = params.get("filename").cloned().unwrap_or_default();
-    if file.is_empty() {
-        return Err(AppError::BadRequest("filename is required".into()));
-    }
-    let file_type = params
-        .get("type")
-        .cloned()
-        .unwrap_or_else(|| "input".to_string());
-    let subfolder = params.get("subfolder").cloned().unwrap_or_default();
-    let user_root = user_home(caller_user_id(&caller))
-        .join("userdata")
-        .join(&file_type);
-    let local_path = if subfolder.trim().is_empty() {
-        user_root.join(&file)
-    } else {
-        user_root.join(safe_view_path(&subfolder)).join(&file)
-    };
-    if tokio::fs::metadata(&local_path).await.is_ok() {
-        let bytes = tokio::fs::read(&local_path)
-            .await
-            .map_err(|err| AppError::UpstreamError(err.to_string()))?;
-        return Ok((
-            StatusCode::OK,
-            [(
-                axum::http::header::CONTENT_TYPE,
-                guess_mime_from_path(&local_path),
-            )],
-            bytes,
-        )
-            .into_response());
-    }
-
-    let row = sqlx::query(
-        "SELECT file_path, remote_url, mime_type
-         FROM workspace_assets
-         WHERE file_name = ? OR public_id = ?
-         ORDER BY id DESC
-         LIMIT 1",
-    )
-    .bind(&file)
-    .bind(&file)
-    .fetch_optional(&*state.db)
-    .await?;
-    let Some(row) = row else {
-        return Err(AppError::NotFound("Workspace asset not found".into()));
-    };
-    let file_path: Option<String> = row.try_get("file_path").ok();
-    if let Some(path) = file_path {
-        let path_buf = PathBuf::from(path);
-        match tokio::fs::read(&path_buf).await {
-            Ok(bytes) => {
-                return Ok((
-                    StatusCode::OK,
-                    [(
-                        axum::http::header::CONTENT_TYPE,
-                        guess_mime_from_path(&path_buf),
-                    )],
-                    bytes,
-                )
-                    .into_response())
-            }
-            Err(err) => return Err(AppError::UpstreamError(err.to_string())),
-        }
-    }
-    let remote_url: Option<String> = row.try_get("remote_url").ok();
-    if let Some(url) = remote_url
-        .as_deref()
-        .filter(|value| !value.trim().is_empty())
-    {
-        return Ok((
-            StatusCode::TEMPORARY_REDIRECT,
-            [(axum::http::header::LOCATION, url.to_string())],
-        )
-            .into_response());
-    }
-    Err(AppError::NotFound(
-        remote_url.unwrap_or_else(|| "asset unavailable".into()),
-    ))
-}
-
-async fn internal_logs(
-    State(state): State<AppState>,
-    headers: HeaderMap,
-) -> AppResult<Html<String>> {
-    let _ = auth_caller(&state, &headers).await?;
-    Ok(Html("".to_string()))
-}
-
-async fn internal_logs_raw(
-    State(state): State<AppState>,
-    headers: HeaderMap,
-) -> AppResult<Json<Value>> {
-    let _ = auth_caller(&state, &headers).await?;
-    Ok(Json(
-        json!({ "size": { "cols": 0, "row": 0 }, "entries": [] }),
-    ))
-}
-
 async fn internal_logs_subscribe(
     State(state): State<AppState>,
     headers: HeaderMap,
@@ -2221,7 +1828,7 @@ pub(super) fn native_workflow_request(
     request
 }
 
-async fn native_workflow_json(
+pub(super) async fn native_workflow_json(
     state: &AppState,
     method: Method,
     path: &str,
@@ -2309,7 +1916,7 @@ fn workflow_cache_kind_from_key(key: &str) -> &'static str {
     }
 }
 
-async fn native_workflow_response(
+pub(super) async fn native_workflow_response(
     state: &AppState,
     method: Method,
     path: &str,
@@ -2338,7 +1945,7 @@ async fn native_workflow_response(
         .into_response())
 }
 
-async fn native_workflow_view_response(
+pub(super) async fn native_workflow_view_response(
     state: &AppState,
     caller: &WorkflowCaller,
     params: &HashMap<String, String>,
@@ -2586,7 +2193,7 @@ pub(super) fn build_query_string(params: &HashMap<String, String>) -> String {
     format!("?{}", parts.join("&"))
 }
 
-fn simple_urlencode(value: &str) -> String {
+pub(super) fn simple_urlencode(value: &str) -> String {
     value
         .replace('%', "%25")
         .replace(' ', "%20")
@@ -2614,7 +2221,7 @@ fn workflow_timestamp_ms(value: &str) -> i64 {
         .unwrap_or_else(|_| timestamp_ms())
 }
 
-fn workflow_graph_to_comfy_workflow(graph: &WorkflowGraph) -> Value {
+pub(super) fn workflow_graph_to_comfy_workflow(graph: &WorkflowGraph) -> Value {
     let nodes = graph
         .nodes
         .iter()
@@ -2701,7 +2308,7 @@ fn run_status_messages(run: &crate::workflow::models::WorkflowRunSummary) -> Vec
     messages
 }
 
-fn run_outputs_for_comfy(run: &crate::workflow::models::WorkflowRunSummary) -> Value {
+pub(super) fn run_outputs_for_comfy(run: &crate::workflow::models::WorkflowRunSummary) -> Value {
     let mut outputs = Map::new();
     for node in &run.node_runs {
         if let Some(node_outputs) = &node.outputs {
@@ -2865,7 +2472,7 @@ fn workflow_node_label(node_type: &str, node_id: &str) -> String {
     }
 }
 
-fn workflow_progress_callback_with_run_sender(
+pub(super) fn workflow_progress_callback_with_run_sender(
     session_id: Option<String>,
     run_sender: Option<mpsc::SyncSender<String>>,
 ) -> Option<WorkflowProgressCallback> {
@@ -3022,7 +2629,7 @@ fn workflow_progress_callback_with_run_sender(
     }))
 }
 
-fn publish_run_events(
+pub(super) fn publish_run_events(
     session_id: Option<&str>,
     run: &crate::workflow::models::WorkflowRunSummary,
     outputs: &Value,
@@ -3551,7 +3158,7 @@ async fn native_workflow_ws_bridge(
     }
 }
 
-async fn workflow_context(state: &AppState, caller: &WorkflowCaller) -> AppResult<RunContext> {
+pub(super) async fn workflow_context(state: &AppState, caller: &WorkflowCaller) -> AppResult<RunContext> {
     let bearer = match caller {
         WorkflowCaller::Admin => state.builtin_keys.comfyui_agent.1.clone(),
         WorkflowCaller::User { user_id } => {

src/routes/workflow_ui/queue.rs

@@ -0,0 +1,435 @@
+use crate::{
+    error::{AppError, AppResult},
+    workflow::storage,
+    AppState,
+};
+use axum::{
+    extract::{Path, Query, State},
+    http::{HeaderMap, Method, StatusCode},
+    response::{Html, IntoResponse, Response},
+    Json,
+};
+use serde_json::{json, Value};
+use sqlx::Row;
+use std::{collections::HashMap, path::PathBuf, sync::mpsc};
+
+use super::auth::{auth_caller, caller_is_admin, caller_user_id};
+use super::node_defs::normalize_prompt_payload;
+use super::{
+    auth_caller_optional, guess_mime_from_path, inject_native_prompt_cloud_api_key,
+    native_workflow_json, native_workflow_json_cached, native_workflow_metadata_ttl,
+    native_workflow_response, native_workflow_view_response, prompt_contains_boosapi_nodes,
+    prompt_error_response, publish_run_events, repair_dynamic_combo_nodes,
+    repair_resize_image_mask_nodes, repair_text_generate_ltx2_prompt_nodes, repair_wan2_api_nodes,
+    run_outputs_for_comfy, safe_view_path, simple_urlencode, user_home, workflow_context,
+    workflow_graph_to_comfy_workflow, workflow_progress_callback_with_run_sender,
+    workflow_run_to_history, workflow_run_to_history_task, workflow_run_to_queue_task,
+    HistoryQuery, PromptPayload, QueueMutation, WorkflowCaller,
+};
+
+pub(super) async fn queue_status(
+    State(state): State<AppState>,
+    headers: HeaderMap,
+) -> AppResult<Json<Value>> {
+    let _ = auth_caller_optional(&state, &headers);
+    if state.config.comfyui.uses_native_backend() {
+        return native_workflow_json(&state, Method::GET, "/prompt", None).await;
+    }
+    Ok(Json(json!({
+        "status_str": "success",
+        "completed": true,
+        "messages": []
+    })))
+}
+
+pub(super) async fn queue(
+    State(state): State<AppState>,
+    headers: HeaderMap,
+) -> AppResult<Json<Value>> {
+    if state.config.comfyui.uses_native_backend() {
+        let _ = auth_caller_optional(&state, &headers);
+        return native_workflow_json(&state, Method::GET, "/queue", None).await;
+    }
+    let caller =
+        auth_caller_optional(&state, &headers).unwrap_or(WorkflowCaller::User { user_id: 0 });
+    let rows = storage::list_runs(
+        &state.db,
+        caller_user_id(&caller),
+        caller_is_admin(&caller),
+        25,
+    )
+    .await?;
+    let running = rows
+        .into_iter()
+        .filter(|run| run.status == "running")
+        .map(workflow_run_to_queue_task)
+        .collect::<Vec<_>>();
+    Ok(Json(json!({
+        "Running": running,
+        "Pending": []
+    })))
+}
+
+pub(super) async fn queue_mutate(
+    State(state): State<AppState>,
+    headers: HeaderMap,
+    Json(_payload): Json<QueueMutation>,
+) -> AppResult<Json<Value>> {
+    let _ = auth_caller(&state, &headers).await?;
+    if state.config.comfyui.uses_native_backend() {
+        let value = serde_json::to_value(&_payload).unwrap_or_else(|_| json!({}));
+        return native_workflow_json(&state, Method::POST, "/queue", Some(value)).await;
+    }
+    Ok(Json(json!({
+        "success": true,
+        "Running": [],
+        "Pending": []
+    })))
+}
+
+pub(super) async fn history(
+    State(state): State<AppState>,
+    headers: HeaderMap,
+    Query(query): Query<HistoryQuery>,
+) -> AppResult<Json<Value>> {
+    if state.config.comfyui.uses_native_backend() {
+        let _ = auth_caller_optional(&state, &headers);
+        let path = format!("/history?max_items={}", query.max_items);
+        return native_workflow_json(&state, Method::GET, &path, None).await;
+    }
+    let caller =
+        auth_caller_optional(&state, &headers).unwrap_or(WorkflowCaller::User { user_id: 0 });
+    let rows = storage::list_runs(
+        &state.db,
+        caller_user_id(&caller),
+        caller_is_admin(&caller),
+        query.max_items,
+    )
+    .await?;
+    let history = rows
+        .into_iter()
+        .map(workflow_run_to_history_task)
+        .collect::<Vec<_>>();
+    Ok(Json(json!({ "History": history })))
+}
+
+pub(super) async fn history_mutate(
+    State(state): State<AppState>,
+    headers: HeaderMap,
+    Json(payload): Json<QueueMutation>,
+) -> AppResult<Json<Value>> {
+    let caller = auth_caller(&state, &headers).await?;
+    if state.config.comfyui.uses_native_backend() {
+        let value = serde_json::to_value(&payload).unwrap_or_else(|_| json!({}));
+        return native_workflow_json(&state, Method::POST, "/history", Some(value)).await;
+    }
+    let affected = if payload.clear {
+        storage::clear_runs(&state, caller_user_id(&caller), caller_is_admin(&caller)).await?
+            as usize
+    } else {
+        storage::delete_runs(
+            &state,
+            &payload.delete,
+            caller_user_id(&caller),
+            caller_is_admin(&caller),
+        )
+        .await?
+    };
+    Ok(Json(json!({
+        "success": true,
+        "deleted": affected
+    })))
+}
+
+pub(super) async fn history_by_id(
+    State(state): State<AppState>,
+    headers: HeaderMap,
+    Path(id): Path<String>,
+) -> AppResult<Json<Value>> {
+    if state.config.comfyui.uses_native_backend() {
+        let _ = auth_caller_optional(&state, &headers);
+        let path = format!("/history/{}", simple_urlencode(&id));
+        return native_workflow_json(&state, Method::GET, &path, None).await;
+    }
+    let caller =
+        auth_caller_optional(&state, &headers).unwrap_or(WorkflowCaller::User { user_id: 0 });
+    let run = storage::get_run(
+        &state.db,
+        &id,
+        caller_user_id(&caller),
+        caller_is_admin(&caller),
+    )
+    .await?;
+    Ok(Json(workflow_run_to_history(run)))
+}
+
+pub(super) async fn queue_prompt(
+    State(state): State<AppState>,
+    headers: HeaderMap,
+    Json(mut payload): Json<PromptPayload>,
+) -> AppResult<Response> {
+    let caller = auth_caller(&state, &headers).await?;
+    if state.config.comfyui.uses_native_backend() {
+        if state.config.comfyui.uses_cloud_backend() {
+            repair_native_prompt_payload(&state, &mut payload).await;
+            return native_workflow_response(
+                &state,
+                Method::POST,
+                "/prompt",
+                Some(serde_json::to_value(payload).unwrap_or_else(|_| json!({}))),
+            )
+            .await;
+        }
+        if prompt_contains_boosapi_nodes(&payload.prompt) {
+            return queue_boosapi_prompt(state, caller, payload).await;
+        }
+        repair_native_prompt_payload(&state, &mut payload).await;
+        return native_workflow_response(
+            &state,
+            Method::POST,
+            "/prompt",
+            Some(serde_json::to_value(payload).unwrap_or_else(|_| json!({}))),
+        )
+        .await;
+    }
+    queue_boosapi_prompt(state, caller, payload).await
+}
+
+pub(super) async fn repair_native_prompt_payload(state: &AppState, payload: &mut PromptPayload) {
+    inject_native_prompt_cloud_api_key(state, payload);
+    let object_info = native_workflow_json_cached(
+        state,
+        Method::GET,
+        "/object_info",
+        None,
+        native_workflow_metadata_ttl(state),
+    )
+    .await
+    .ok()
+    .map(|value| value.0);
+    if let Some(object_info) = object_info.as_ref() {
+        repair_dynamic_combo_nodes(&mut payload.prompt, object_info);
+    }
+    repair_resize_image_mask_nodes(&mut payload.prompt, &payload.extra_data);
+    repair_text_generate_ltx2_prompt_nodes(&mut payload.prompt);
+    repair_wan2_api_nodes(&mut payload.prompt);
+}
+
+pub(super) async fn queue_boosapi_prompt(
+    state: AppState,
+    caller: WorkflowCaller,
+    payload: PromptPayload,
+) -> AppResult<Response> {
+    let prompt = payload.prompt.clone();
+    let graph = match normalize_prompt_payload(&prompt) {
+        Ok(graph) => graph,
+        Err(err) => {
+            return Ok(prompt_error_response("工作流无法执行", err.to_string()));
+        }
+    };
+    if graph.nodes.is_empty() {
+        return Ok(prompt_error_response(
+            "画布里还没有可执行节点",
+            "请先从左侧节点库添加 BoosAPI 节点并连接后再执行。",
+        ));
+    }
+    let name = payload
+        .extra_data
+        .pointer("/extra_pnginfo/workflow/name")
+        .and_then(Value::as_str)
+        .unwrap_or("ComfyUI workflow");
+    let comfy_workflow = payload
+        .extra_data
+        .pointer("/extra_pnginfo/workflow")
+        .cloned()
+        .unwrap_or_else(|| workflow_graph_to_comfy_workflow(&graph));
+    let workflow = storage::insert_workflow(
+        &state,
+        caller_user_id(&caller),
+        name,
+        Some("Imported from ComfyUI frontend"),
+        &serde_json::to_value(&graph).unwrap_or_else(|_| json!({})),
+        "private",
+        false,
+    )
+    .await?;
+
+    let ctx = workflow_context(&state, &caller).await?;
+    let progress_session_id = payload
+        .client_id
+        .clone()
+        .filter(|value| !value.trim().is_empty())
+        .unwrap_or_else(|| "boosapi-workflow".to_string());
+    let (run_tx, run_rx) = mpsc::sync_channel::<String>(1);
+    let progress =
+        workflow_progress_callback_with_run_sender(Some(progress_session_id.clone()), Some(run_tx));
+    let state_for_task = state.clone();
+    let graph_value = serde_json::to_value(&graph).unwrap_or_else(|_| json!({}));
+    let task_graph_value = graph_value.clone();
+    let user_id = caller_user_id(&caller);
+    let allow_admin = caller_is_admin(&caller);
+    let session_for_task = progress_session_id.clone();
+    tokio::spawn(async move {
+        let result = crate::workflow::executor::run_workflow_with_progress(
+            &state_for_task,
+            workflow.id,
+            task_graph_value,
+            json!({}),
+            ctx,
+            progress,
+        )
+        .await;
+        match result {
+            Ok(run_id) => {
+                if let Ok(run) =
+                    storage::get_run(&state_for_task.db, &run_id, user_id, allow_admin).await
+                {
+                    let outputs = run_outputs_for_comfy(&run);
+                    publish_run_events(Some(session_for_task.as_str()), &run, &outputs);
+                }
+            }
+            Err(err) => {
+                tracing::warn!("workflow background execution failed: {err}");
+            }
+        }
+    });
+    let run_id = match tokio::task::spawn_blocking(move || run_rx.recv()).await {
+        Ok(Ok(run_id)) => run_id,
+        Ok(Err(_)) | Err(_) => {
+            return Ok(prompt_error_response(
+                "工作流执行失败",
+                "后台任务没有成功进入队列，请稍后重试。",
+            ));
+        }
+    };
+
+    Ok(Json(json!({
+        "prompt_id": run_id,
+        "number": payload.number.unwrap_or(0),
+        "node_errors": [],
+        "client_id": payload.client_id,
+        "front": payload.front.unwrap_or(false),
+        "outputs": {},
+        "workflow": comfy_workflow
+    }))
+    .into_response())
+}
+
+pub(super) async fn interrupt(
+    State(state): State<AppState>,
+    headers: HeaderMap,
+) -> AppResult<Json<Value>> {
+    let _ = auth_caller_optional(&state, &headers);
+    if state.config.comfyui.uses_native_backend() {
+        return native_workflow_json(&state, Method::POST, "/interrupt", Some(json!({}))).await;
+    }
+    Ok(Json(json!({ "success": true })))
+}
+
+pub(super) async fn view(
+    State(state): State<AppState>,
+    headers: HeaderMap,
+    Query(params): Query<HashMap<String, String>>,
+) -> AppResult<Response> {
+    let caller = auth_caller(&state, &headers).await?;
+    if state.config.comfyui.uses_native_backend() {
+        return native_workflow_view_response(&state, &caller, &params).await;
+    }
+    let file = params.get("filename").cloned().unwrap_or_default();
+    if file.is_empty() {
+        return Err(AppError::BadRequest("filename is required".into()));
+    }
+    let file_type = params
+        .get("type")
+        .cloned()
+        .unwrap_or_else(|| "input".to_string());
+    let subfolder = params.get("subfolder").cloned().unwrap_or_default();
+    let user_root = user_home(caller_user_id(&caller))
+        .join("userdata")
+        .join(&file_type);
+    let local_path = if subfolder.trim().is_empty() {
+        user_root.join(&file)
+    } else {
+        user_root.join(safe_view_path(&subfolder)).join(&file)
+    };
+    if tokio::fs::metadata(&local_path).await.is_ok() {
+        let bytes = tokio::fs::read(&local_path)
+            .await
+            .map_err(|err| AppError::UpstreamError(err.to_string()))?;
+        return Ok((
+            StatusCode::OK,
+            [(
+                axum::http::header::CONTENT_TYPE,
+                guess_mime_from_path(&local_path),
+            )],
+            bytes,
+        )
+            .into_response());
+    }
+
+    let row = sqlx::query(
+        "SELECT file_path, remote_url, mime_type
+         FROM workspace_assets
+         WHERE file_name = ? OR public_id = ?
+         ORDER BY id DESC
+         LIMIT 1",
+    )
+    .bind(&file)
+    .bind(&file)
+    .fetch_optional(&*state.db)
+    .await?;
+    let Some(row) = row else {
+        return Err(AppError::NotFound("Workspace asset not found".into()));
+    };
+    let file_path: Option<String> = row.try_get("file_path").ok();
+    if let Some(path) = file_path {
+        let path_buf = PathBuf::from(path);
+        match tokio::fs::read(&path_buf).await {
+            Ok(bytes) => {
+                return Ok((
+                    StatusCode::OK,
+                    [(
+                        axum::http::header::CONTENT_TYPE,
+                        guess_mime_from_path(&path_buf),
+                    )],
+                    bytes,
+                )
+                    .into_response())
+            }
+            Err(err) => return Err(AppError::UpstreamError(err.to_string())),
+        }
+    }
+    let remote_url: Option<String> = row.try_get("remote_url").ok();
+    if let Some(url) = remote_url
+        .as_deref()
+        .filter(|value| !value.trim().is_empty())
+    {
+        return Ok((
+            StatusCode::TEMPORARY_REDIRECT,
+            [(axum::http::header::LOCATION, url.to_string())],
+        )
+            .into_response());
+    }
+    Err(AppError::NotFound(
+        remote_url.unwrap_or_else(|| "asset unavailable".into()),
+    ))
+}
+
+pub(super) async fn internal_logs(
+    State(state): State<AppState>,
+    headers: HeaderMap,
+) -> AppResult<Html<String>> {
+    let _ = auth_caller(&state, &headers).await?;
+    Ok(Html("".to_string()))
+}
+
+pub(super) async fn internal_logs_raw(
+    State(state): State<AppState>,
+    headers: HeaderMap,
+) -> AppResult<Json<Value>> {
+    let _ = auth_caller(&state, &headers).await?;
+    Ok(Json(
+        json!({ "size": { "cols": 0, "row": 0 }, "entries": [] }),
+    ))
+}
+