//! Docker 节点 — 在 Docker 容器内执行命令
//!
//! 通过 `docker run --rm` 一次性容器执行命令。复用 df_execute::shell::execute 调用本地
//! docker CLI,与 ScriptNode/GitNode 路径一致(不内嵌 docker SDK,CLI 更通用透明)。
//!
//! 执行流程:
//! 1. 先 `docker --version` 探测 Docker 可用性(未装/未运行直接报错,避免容器启动失败
//!    时退出码语义混淆)。
//! 2. 构建 `docker run --rm {volumes} {env} -w {working_dir} {image} {command}`。
//! 3. 经 df_execute::shell::execute 执行,回传 stdout/stderr/exit_code。

use async_trait::async_trait;
use df_workflow::node::{Node, NodeContext, NodeOutput, NodeResult, NodeSchema};

/// Docker 节点
pub struct DockerNode;

/// 卷挂载配置项(volumes 数组元素)
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct VolumeMount {
    pub host: String,
    pub container: String,
}

/// 从 NodeContext.config 解析出的 DockerNode 参数。
/// 抽离此结构便于单元测试 config 解析逻辑(无需起 shell / 真实容器)。
#[derive(Debug, Clone)]
pub struct DockerParams {
    /// Docker 镜像名(必填,如 "rust:latest")
    pub image: String,
    /// 容器内执行命令(必填)
    pub command: String,
    /// 容器内工作目录(默认 "/workspace")
    pub working_dir: String,
    /// 超时秒数(默认 300)
    pub timeout_secs: u64,
    /// 卷挂载列表(可选)
    pub volumes: Vec<VolumeMount>,
    /// 环境变量(可选)
    pub env: std::collections::HashMap<String, String>,
}

/// 从 config JSON 解析 DockerParams。image / command 必填,其余按默认/可选。
pub fn parse_params(config: &serde_json::Value) -> anyhow::Result<DockerParams> {
    let image = config
        .get("image")
        .and_then(|v| v.as_str())
        .ok_or_else(|| anyhow::anyhow!("DockerNode 缺少必填参数: image"))?
        .to_string();

    let command = config
        .get("command")
        .and_then(|v| v.as_str())
        .ok_or_else(|| anyhow::anyhow!("DockerNode 缺少必填参数: command"))?
        .to_string();

    let working_dir = config
        .get("working_dir")
        .and_then(|v| v.as_str())
        .map(|s| s.to_string())
        .unwrap_or_else(|| "/workspace".to_string());

    let timeout_secs = config
        .get("timeout_secs")
        .and_then(|v| v.as_u64())
        .unwrap_or(300);

    // volumes: JSON 数组,每项 {"host": "...", "container": "..."}
    // 容错:类型不符/缺字段项跳过(不整体失败,单条坏配置不阻塞整个工作流)。
    let volumes = config
        .get("volumes")
        .and_then(|v| v.as_array())
        .map(|arr| {
            arr.iter()
                .filter_map(|item| {
                    let host = item.get("host")?.as_str()?.to_string();
                    let container = item.get("container")?.as_str()?.to_string();
                    Some(VolumeMount { host, container })
                })
                .collect()
        })
        .unwrap_or_default();

    // env: JSON 对象 { KEY: VALUE },值统一转字符串。
    let env = config
        .get("env")
        .and_then(|v| v.as_object())
        .map(|obj| {
            obj.iter()
                .filter_map(|(k, v)| v.as_str().map(|s| (k.clone(), s.to_string())))
                .collect()
        })
        .unwrap_or_default();

    Ok(DockerParams {
        image,
        command,
        working_dir,
        timeout_secs,
        volumes,
        env,
    })
}

/// 探测 Docker 是否可用:`docker --version` 退出码 0 视为可用。
async fn check_docker_available() -> anyhow::Result<()> {
    let request = df_execute::shell::ShellRequest {
        command: "docker --version".to_string(),
        working_dir: None,
        env: std::collections::HashMap::new(),
        timeout_secs: Some(15),
        shell_type: Default::default(),
    };
    let result = df_execute::shell::execute(request).await;
    match result {
        Ok(r) if r.exit_code.unwrap_or(-1) == 0 => Ok(()),
        _ => anyhow::bail!("Docker 未安装或未运行"),
    }
}

/// 构建 docker run 命令字符串。
/// 卷/环境变量值用 shell_quote 包裹,防止空格/特殊字符注入。
fn build_command(params: &DockerParams) -> String {
    let mut parts: Vec<String> = vec!["docker run --rm".to_string()];

    for v in &params.volumes {
        parts.push(format!(
            "-v {}:{}",
            shell_quote(&v.host),
            shell_quote(&v.container)
        ));
    }

    for (k, val) in &params.env {
        parts.push(format!("-e {}={}", k, shell_quote(val)));
    }

    parts.push(format!("-w {}", shell_quote(&params.working_dir)));
    parts.push(shell_quote(&params.image));
    // command 原样追加(用户自行决定是否含参数 / shell 元字符),不做引号包裹,
    // 与脚本节点一致由 shell 解释器解析。
    parts.push(params.command.clone());

    parts.join(" ")
}

/// 简单 shell 引号包裹:含空格/特殊字符时用双引号包裹并转义内嵌双引号。
fn shell_quote(s: &str) -> String {
    if s
        .chars()
        .any(|c| c.is_whitespace() || c == '"' || c == '$' || c == '`')
    {
        format!("\"{}\"", s.replace('"', "\\\""))
    } else {
        s.to_string()
    }
}

#[async_trait]
impl Node for DockerNode {
    async fn execute(&self, ctx: NodeContext) -> NodeResult {
        tracing::info!("DockerNode 执行: node_id={}", ctx.node_id);

        // 1. Docker 可用性探测(未装/未运行直接 fail-fast)。
        if let Err(e) = check_docker_available().await {
            anyhow::bail!(e.to_string());
        }

        let params = parse_params(&ctx.config)?;
        let command = build_command(&params);

        tracing::info!(
            image = %params.image,
            working_dir = %params.working_dir,
            timeout_secs = params.timeout_secs,
            "DockerNode 构建命令: {}",
            command
        );

        let request = df_execute::shell::ShellRequest {
            command,
            // 宿主机工作目录对 docker run 无意义,置 None。
            working_dir: None,
            env: std::collections::HashMap::new(),
            timeout_secs: Some(params.timeout_secs),
            shell_type: Default::default(),
        };

        let result = df_execute::shell::execute(request).await?;
        let exit_code = result.exit_code.unwrap_or(-1);

        tracing::info!(
            exit_code,
            duration_ms = result.duration_ms,
            "DockerNode 完成"
        );

        Ok(NodeOutput::from_value(serde_json::json!({
            "image": params.image,
            "stdout": result.stdout,
            "stderr": result.stderr,
            "exit_code": exit_code,
            "duration_ms": result.duration_ms,
        })))
    }

    fn schema(&self) -> NodeSchema {
        NodeSchema {
            params: serde_json::json!({
                "type": "object",
                "properties": {
                    "image": { "type": "string", "description": "Docker 镜像名(如 rust:latest)" },
                    "command": { "type": "string", "description": "容器内执行命令" },
                    "working_dir": { "type": "string", "description": "容器工作目录(默认 /workspace)" },
                    "timeout_secs": { "type": "integer", "description": "超时秒数(默认 300)" },
                    "volumes": {
                        "type": "array",
                        "description": "卷挂载 [{host, container}]",
                        "items": {
                            "type": "object",
                            "properties": {
                                "host": { "type": "string" },
                                "container": { "type": "string" }
                            }
                        }
                    },
                    "env": { "type": "object", "description": "环境变量键值对" }
                },
                "required": ["image", "command"]
            }),
            output: serde_json::json!({
                "type": "object",
                "properties": {
                    "image": { "type": "string" },
                    "stdout": { "type": "string" },
                    "stderr": { "type": "string" },
                    "exit_code": { "type": "integer" },
                    "duration_ms": { "type": "integer" }
                }
            }),
        }
    }

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

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;

    // ── parse_params: 必填缺失 ──

    #[test]
    fn params_missing_image_errors() {
        let err = parse_params(&json!({ "command": "ls" }))
            .unwrap_err()
            .to_string();
        assert!(err.contains("image"), "实际: {}", err);
    }

    #[test]
    fn params_missing_command_errors() {
        let err = parse_params(&json!({ "image": "rust:latest" }))
            .unwrap_err()
            .to_string();
        assert!(err.contains("command"), "实际: {}", err);
    }

    // ── parse_params: 默认值 ──

    #[test]
    fn params_minimal_uses_defaults() {
        let p = parse_params(&json!({ "image": "alpine", "command": "echo hi" })).unwrap();
        assert_eq!(p.image, "alpine");
        assert_eq!(p.command, "echo hi");
        assert_eq!(p.working_dir, "/workspace");
        assert_eq!(p.timeout_secs, 300);
        assert!(p.volumes.is_empty());
        assert!(p.env.is_empty());
    }

    // ── parse_params: 完整字段 ──

    #[test]
    fn params_full_fields() {
        let p = parse_params(&json!({
            "image": "rust:latest",
            "command": "cargo test",
            "working_dir": "/app",
            "timeout_secs": 120
        }))
        .unwrap();
        assert_eq!(p.image, "rust:latest");
        assert_eq!(p.command, "cargo test");
        assert_eq!(p.working_dir, "/app");
        assert_eq!(p.timeout_secs, 120);
    }

    // ── parse_params: volumes 解析 ──

    #[test]
    fn params_volumes_parsed() {
        let p = parse_params(&json!({
            "image": "node",
            "command": "npm ci",
            "volumes": [
                { "host": "/host/a", "container": "/c/a" },
                { "host": "/host/b", "container": "/c/b" }
            ]
        }))
        .unwrap();
        assert_eq!(p.volumes.len(), 2);
        assert_eq!(
            p.volumes[0],
            VolumeMount {
                host: "/host/a".into(),
                container: "/c/a".into()
            }
        );
        assert_eq!(p.volumes[1].container, "/c/b");
    }

    #[test]
    fn params_volumes_skips_malformed_items() {
        // 缺 container / 非 object 项应被跳过,不整体失败。
        let p = parse_params(&json!({
            "image": "node",
            "command": "ls",
            "volumes": [
                { "host": "/ok", "container": "/ok" },
                { "host": "/no-container" },
                "not-an-object",
                { "container": "/no-host" }
            ]
        }))
        .unwrap();
        assert_eq!(p.volumes.len(), 1);
        assert_eq!(p.volumes[0].host, "/ok");
    }

    // ── parse_params: env 解析 ──

    #[test]
    fn params_env_parsed() {
        let p = parse_params(&json!({
            "image": "python",
            "command": "pytest",
            "env": {
                "FOO": "bar",
                "DEBUG": "1"
            }
        }))
        .unwrap();
        assert_eq!(p.env.len(), 2);
        assert_eq!(p.env.get("FOO").map(|s| s.as_str()), Some("bar"));
        assert_eq!(p.env.get("DEBUG").map(|s| s.as_str()), Some("1"));
    }

    #[test]
    fn params_env_empty_when_non_object() {
        // env 非 object(误传字符串)时回退空 map,不报错。
        let p = parse_params(&json!({
            "image": "python",
            "command": "ls",
            "env": "should-be-object"
        }))
        .unwrap();
        assert!(p.env.is_empty());
    }

    // ── build_command ──

    #[test]
    fn command_minimal_shape() {
        let p = parse_params(&json!({
            "image": "alpine",
            "command": "echo hello"
        }))
        .unwrap();
        let cmd = build_command(&p);
        assert!(cmd.starts_with("docker run --rm"), "实际: {}", cmd);
        assert!(cmd.contains("-w /workspace"), "实际: {}", cmd);
        assert!(cmd.contains(" alpine "), "实际: {}", cmd);
        assert!(cmd.ends_with("echo hello"), "实际: {}", cmd);
    }

    #[test]
    fn command_includes_volumes_and_env() {
        let p = parse_params(&json!({
            "image": "rust:latest",
            "command": "cargo build",
            "working_dir": "/app",
            "volumes": [
                { "host": "/host/src", "container": "/app" }
            ],
            "env": { "CARGO_HOME": "/cargo" }
        }))
        .unwrap();
        let cmd = build_command(&p);
        assert!(cmd.contains("-v /host/src:/app"), "实际: {}", cmd);
        assert!(cmd.contains("-e CARGO_HOME=/cargo"), "实际: {}", cmd);
        assert!(cmd.contains("-w /app"), "实际: {}", cmd);
    }

    // ── shell_quote ──

    #[test]
    fn shell_quote_plain_passthrough() {
        assert_eq!(shell_quote("abc"), "abc");
    }

    #[test]
    fn shell_quote_wraps_spaces() {
        assert_eq!(shell_quote("/a b/c"), "\"/a b/c\"");
    }
}