//! human 节点端到端集成测试 — DagDef → NodeRegistry.build_dag → DagExecutor.run
//!
//! df-workflow 不依赖 df-nodes(反向依赖),此处用等价阻塞节点模拟 HumanNode 的
//! 「发审批请求 → 挂起等待 → 外部 approve → 返回结果」链路,覆盖
//! DagDef 序列化 → 注册表 → 执行器全链路 + 事件序列断言(R6 send 缺 await /
//! R7 契约失配的存活土壤)。

use std::time::Duration;

use async_trait::async_trait;
use serde_json::json;

use df_types::events::{SelectType, WorkflowEvent};
use df_types::types::NodeStatus;
use df_workflow::dag_def::DagDef;
use df_workflow::eventbus::{EventBus, EventSubscriber};
use df_workflow::executor::DagExecutor;
use df_workflow::node::{Node, NodeContext, NodeOutput, NodeResult, NodeSchema};
use df_workflow::registry::NodeRegistry;

/// 前驱节点:sleep 后返回空输出(仿 executor 单测 SleepNode)。
struct SleepNode {
    sleep_ms: u64,
}

#[async_trait]
impl Node for SleepNode {
    async fn execute(&self, _ctx: NodeContext) -> NodeResult {
        tokio::time::sleep(Duration::from_millis(self.sleep_ms)).await;
        Ok(NodeOutput::empty())
    }

    fn schema(&self) -> NodeSchema {
        NodeSchema {
            params: json!(null),
            output: json!(null),
        }
    }

    fn node_type(&self) -> &str {
        "sleep"
    }
}

/// 模拟 HumanNode 的阻塞审批节点:先 subscribe 再发 HumanApprovalRequest(broadcast 不回放),
/// select! 等待匹配 execution_id + node_id 的 HumanApprovalResponse,收到则返回 decision。
struct ApprovalNode;

#[async_trait]
impl Node for ApprovalNode {
    async fn execute(&self, ctx: NodeContext) -> NodeResult {
        let mut rx = ctx.event_bus.subscribe();
        let _ = ctx
            .event_bus
            .send(WorkflowEvent::HumanApprovalRequest {
                execution_id: ctx.execution_id.clone(),
                node_id: ctx.node_id.clone(),
                title: "确认发布".to_string(),
                description: String::new(),
                options: vec!["同意".into(), "拒绝".into()],
                select_type: SelectType::Single,
            })
            .await;
        let timeout_secs = ctx
            .config
            .get("timeout_secs")
            .and_then(|v| v.as_u64())
            .unwrap_or(2);
        let deadline = tokio::time::Instant::now() + Duration::from_secs(timeout_secs);
        loop {
            tokio::select! {
                recv = rx.recv() => match recv {
                    Ok(WorkflowEvent::HumanApprovalResponse {
                        execution_id, node_id, decision, decisions, ..
                    }) if execution_id == ctx.execution_id && node_id == ctx.node_id => {
                        let primary = if decision.is_empty() {
                            decisions.first().cloned().unwrap_or_default()
                        } else {
                            decision
                        };
                        return Ok(NodeOutput::from_value(json!({ "decision": primary })));
                    }
                    Ok(_) => continue,
                    Err(_) => return Err(anyhow::anyhow!("事件总线关闭,审批无法完成")),
                },
                _ = tokio::time::sleep_until(deadline) => {
                    return Err(anyhow::anyhow!("人工审批等待超时({timeout_secs}s)"));
                }
            }
        }
    }

    fn schema(&self) -> NodeSchema {
        NodeSchema {
            params: json!(null),
            output: json!(null),
        }
    }

    fn is_blocking(&self) -> bool {
        true
    }

    fn node_type(&self) -> &str {
        "human"
    }
}

/// 从事件流中等待一条 HumanApprovalRequest(跳过 NodeStarted/NodeCompleted 等其他事件)。
async fn wait_approval_request(mut rx: EventSubscriber) -> WorkflowEvent {
    tokio::time::timeout(Duration::from_millis(1000), async move {
        loop {
            if let Ok(ev @ WorkflowEvent::HumanApprovalRequest { .. }) = rx.recv().await {
                return ev;
            }
        }
    })
    .await
    .expect("应收到 HumanApprovalRequest,实际超时")
}

/// 端到端主链路:前驱 sleep(a) → human(b)。
/// 执行到 human 后挂起等审批 → 外部 approve(模拟 approve_human_approval IPC) → 返回结果、工作流完成。
#[tokio::test]
async fn human_workflow_end_to_end_approval() {
    let bus = EventBus::new();
    let rx = bus.subscribe(); // 先 subscribe 再执行(broadcast 不回放)
    let exec_id = "exec-e2e-approve";

    // DagDef(可序列化定义)→ NodeRegistry.build_dag → 运行时 Dag
    let mut def = DagDef::new();
    def.add_node("a", "sleep", json!({}));
    def.add_node("b", "human", json!({}));
    def.add_edge("a", "b");
    let mut registry = NodeRegistry::new();
    registry.register("sleep", |_| Box::new(SleepNode { sleep_ms: 10 }));
    registry.register("human", |_| Box::new(ApprovalNode));
    let dag = registry.build_dag(&def).expect("build_dag 应成功");

    let mut executor = DagExecutor::new(bus.clone(), exec_id.into());
    let sm = executor.state_machine(); // 共享状态机(spawn 后仍可读)
    let run_handle = tokio::spawn(async move { executor.run(&dag, json!({})).await });

    // a 层完成、b 层 human subscribe + send Request 后应收到审批请求
    let request = wait_approval_request(rx).await;
    match request {
        WorkflowEvent::HumanApprovalRequest {
            node_id, title, options, ..
        } => {
            assert_eq!(node_id, "b");
            assert_eq!(title, "确认发布");
            assert_eq!(options, vec!["同意".to_string(), "拒绝".to_string()]);
        }
        other => panic!("期望 HumanApprovalRequest,收到 {:?}", other),
    }

    // 外部 approve(模拟 approve_human_approval IPC 发送 HumanApprovalResponse 到总线)
    bus.send(WorkflowEvent::HumanApprovalResponse {
        execution_id: exec_id.into(),
        node_id: "b".to_string(),
        decision: "同意".into(),
        decisions: vec![],
        comment: Some("可以发布".into()),
    })
    .await;

    let outputs = run_handle
        .await
        .expect("run 不应 panic")
        .expect("run 应成功");
    assert!(outputs.contains_key("a"), "outputs 应含前驱节点 a");
    assert_eq!(outputs["b"].data["decision"], json!("同意"));
    assert_eq!(sm.get(&"a".to_string()), NodeStatus::Completed);
    assert_eq!(sm.get(&"b".to_string()), NodeStatus::Completed);
}

/// 取消路径:外部 set_cancelled(模拟 cancel_workflow_node IPC 写共享状态机)
/// → human select! 轮询 is_cancelled → Err → 状态保持 Cancelled。
#[tokio::test]
async fn human_workflow_cancel_via_shared_state_machine() {
    let bus = EventBus::new();
    let rx = bus.subscribe();
    let exec_id = "exec-e2e-cancel";

    let mut def = DagDef::new();
    def.add_node("h", "human", json!({}));
    let mut registry = NodeRegistry::new();
    registry.register("human", |_| Box::new(ApprovalNode));
    let dag = registry.build_dag(&def).expect("build_dag 应成功");

    let mut executor = DagExecutor::new(bus.clone(), exec_id.into());
    let sm = executor.state_machine();
    let run_handle = tokio::spawn(async move { executor.run(&dag, json!({})).await });

    // 等 human 发出审批请求(确认已 subscribe + send + 进入 select! 轮询)
    let _ = wait_approval_request(rx).await;

    // 外部 cancel:写共享状态机置 Cancelled,humar select! 的 cancel_tick 会读到
    sm.set_cancelled("h".to_string());

    let result = run_handle.await.unwrap();
    assert!(result.is_err(), "取消应致 run 返回 Err");
    let err = result.unwrap_err().to_string();
    assert!(
        err.contains("取消") || err.contains("执行失败"),
        "应返回取消相关错误,实际: {err}"
    );
    // 状态保持 Cancelled(未被 set_failed 覆盖为 Failed)—— 取消终态不被失败语义覆盖
    assert_eq!(sm.get(&"h".to_string()), NodeStatus::Cancelled);
}

/// 挂起等待路径:收到请求后不 approve → human 挂起直到超时 → run 返 Err。
/// 证明节点真在等待外部响应(而非空转直接返回)。
#[tokio::test]
async fn human_workflow_waits_then_times_out_without_approve() {
    let bus = EventBus::new();
    let rx = bus.subscribe();
    let exec_id = "exec-e2e-timeout";

    let mut def = DagDef::new();
    def.add_node("h", "human", json!({ "timeout_secs": 1 }));
    let mut registry = NodeRegistry::new();
    registry.register("human", |_| Box::new(ApprovalNode));
    let dag = registry.build_dag(&def).expect("build_dag 应成功");

    let mut executor = DagExecutor::new(bus.clone(), exec_id.into());
    let run_handle = tokio::spawn(async move { executor.run(&dag, json!({})).await });

    let _ = wait_approval_request(rx).await;
    let result = run_handle.await.unwrap();
    assert!(result.is_err(), "无审批响应应超时失败");
    // executor 包了「节点 h 执行失败」context,用 {:#} 展开整条错误链断言根因
    let chain = format!("{:#}", result.unwrap_err());
    assert!(chain.contains("超时"), "应返回超时错误,实际: {chain}");
}