DevFlow/src-tauri/src/commands/workflow.rs

//! 工作流相关命令 — 触发执行、查询执行记录
//!
//! 事件转发：run_workflow 在执行前订阅事件总线，
//! 将 WorkflowEvent 包装为 `{ execution_id, event }` 通过 `workflow-event` 事件发给前端。

use serde::Serialize;
use tauri::{AppHandle, Emitter, State};
use tokio::sync::broadcast::error::RecvError;

use df_core::events::{WorkflowEvent, HumanApprovalResponse};
use df_core::types::new_id;
use df_storage::crud::WorkflowRepo;
use df_storage::models::WorkflowRecord;
use df_workflow::dag_def::DagDef;
use df_workflow::executor::DagExecutor;

use crate::state::AppState;

use super::now_millis;

/// 转发到前端的事件载荷
#[derive(Debug, Clone, Serialize)]
struct WorkflowEventPayload {
    /// 工作流执行记录 ID
    execution_id: String,
    /// 原始工作流事件（serde tag = "type"）
    event: WorkflowEvent,
}

/// 触发工作流执行（核心命令）
///
/// 流程：build_dag 校验 → 写入执行记录(status=running) → 后台异步执行 →
/// 事件经 EventBus 转发到前端 → 完成后更新执行记录状态。
/// 立即返回执行记录 ID，前端凭此关联后续事件。
#[tauri::command]
pub async fn run_workflow(
    app: AppHandle,
    state: State<'_, AppState>,
    name: String,
    dag: DagDef,
    config: serde_json::Value,
) -> Result<String, String> {
    // 1. 先构建运行时 DAG，校验失败直接返回，不落库
    let runtime_dag = state.registry.build_dag(&dag).map_err(|e| e.to_string())?;

    // 2. 写入执行记录（status=running）
    let execution_id = new_id();
    let dag_json = serde_json::to_string(&dag).map_err(|e| e.to_string())?;
    let record = WorkflowRecord {
        id: execution_id.clone(),
        name,
        dag_json,
        status: "running".to_string(),
        triggered_by: Some("manual".to_string()),
        project_id: None,
        task_id: None,
        created_at: now_millis(),
        completed_at: None,
    };
    state
        .workflows
        .insert(record)
        .await
        .map_err(|e| e.to_string())?;

    // 3. 订阅事件总线，把事件转发给前端（收到完成/失败事件后退出）
    let mut rx = state.event_bus.subscribe();
    let forward_app = app.clone();
    let forward_exec_id = execution_id.clone();
    tauri::async_runtime::spawn(async move {
        loop {
            match rx.recv().await {
                Ok(event) => {
                    let finished = matches!(
                        event,
                        WorkflowEvent::WorkflowCompleted { .. }
                            | WorkflowEvent::WorkflowFailed { .. }
                    );
                    let payload = WorkflowEventPayload {
                        execution_id: forward_exec_id.clone(),
                        event,
                    };
                    if let Err(e) = forward_app.emit("workflow-event", &payload) {
                        tracing::warn!("工作流事件转发失败: {}", e);
                    }
                    if finished {
                        break;
                    }
                }
                // 消费过慢丢失部分事件时继续接收
                Err(RecvError::Lagged(n)) => {
                    tracing::warn!("工作流事件消费滞后，丢失 {} 条", n);
                }
                Err(RecvError::Closed) => break,
            }
        }
    });

    // 4. 后台异步执行 DAG，完成后更新执行记录状态
    let event_bus = state.event_bus.clone();
    let db = state.db.clone();
    let exec_id = execution_id.clone();
    tauri::async_runtime::spawn(async move {
        let mut executor = DagExecutor::new(event_bus.clone(), exec_id.clone());
        let result = executor.run(&runtime_dag, config).await;

        let (status, error) = match &result {
            Ok(_) => ("completed", None),
            Err(e) => ("failed", Some(format!("{:#}", e))),
        };

        // 更新执行记录（Repo 内部仅持有 Arc 连接，重建开销可忽略）
        let workflows = WorkflowRepo::new(&db);
        if let Err(e) = workflows.update_field(&exec_id, "status", status).await {
            tracing::error!("更新工作流状态失败: {}", e);
        }
        if let Err(e) = workflows
            .update_field(&exec_id, "completed_at", &now_millis())
            .await
        {
            tracing::error!("更新工作流完成时间失败: {}", e);
        }

        // 执行失败时补发 WorkflowFailed（执行器内部只发 NodeFailed）
        if let Some(error) = error {
            event_bus
                .send(WorkflowEvent::WorkflowFailed {
                    error,
                    failed_node: String::new(),
                })
                .await;
        }
    });

    Ok(execution_id)
}

/// 列出全部工作流执行记录
#[tauri::command]
pub async fn list_workflow_executions(
    state: State<'_, AppState>,
) -> Result<Vec<WorkflowRecord>, String> {
    state.workflows.list_all().await.map_err(|e| e.to_string())
}

/// 按 ID 查询工作流执行记录
#[tauri::command]
pub async fn get_workflow_execution(
    state: State<'_, AppState>,
    id: String,
) -> Result<Option<WorkflowRecord>, String> {
    state
        .workflows
        .get_by_id(&id)
        .await
        .map_err(|e| e.to_string())
}

/// 发送人工审批响应
#[tauri::command]
pub async fn approve_human_approval(
    app: AppHandle,
    state: State<'_, AppState>,
    execution_id: String,
    node_id: String,
    decision: String,
    comment: Option<String>,
) -> Result<(), String> {
    // 发送审批响应到事件总线
    let response = HumanApprovalResponse {
        execution_id: execution_id.clone(),
        node_id: node_id.clone(),
        decision,
        comment,
    };

    let event = WorkflowEvent::HumanApprovalResponse {
        execution_id: response.execution_id,
        node_id: response.node_id,
        decision: response.decision,
        comment: response.comment,
    };

    // 直接发送到全局事件总线
    state.event_bus.send(event).await;
    Ok(())
}