DevFlow/crates/df-nodes/src/http_node.rs

//! HTTP 节点 — 发起 HTTP 请求(GET/POST/PUT/DELETE)
//!
//! 使用 reqwest(与 df-ai 同一版本 0.12)。config 提供 method/url/headers/body/timeout_secs。
//! 输出 status_code + body + (可选)响应头子集。

use std::time::Duration;

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

/// HTTP 节点
pub struct HttpNode;

/// 支持的 HTTP 方法
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum HttpMethod {
    Get,
    Post,
    Put,
    Delete,
}

impl HttpMethod {
    /// 从 config.method 字符串解析,大小写不敏感,非法值返回 Err。缺省返回 GET。
    pub fn parse(raw: Option<&str>) -> anyhow::Result<Self> {
        match raw.map(|s| s.to_ascii_uppercase()).as_deref() {
            None | Some("GET") => Ok(HttpMethod::Get),
            Some("POST") => Ok(HttpMethod::Post),
            Some("PUT") => Ok(HttpMethod::Put),
            Some("DELETE") => Ok(HttpMethod::Delete),
            Some(other) => anyhow::bail!(
                "HttpNode 非法 method: {}（合法值: GET|POST|PUT|DELETE）",
                other
            ),
        }
    }

    fn as_str(&self) -> &'static str {
        match self {
            HttpMethod::Get => "GET",
            HttpMethod::Post => "POST",
            HttpMethod::Put => "PUT",
            HttpMethod::Delete => "DELETE",
        }
    }
}

/// 从 NodeContext.config 解析出的 HttpNode 参数。
/// 抽离此结构便于单元测试 config 解析逻辑(无需发起真实网络请求)。
#[derive(Debug, Clone)]
pub struct HttpParams {
    pub method: HttpMethod,
    pub url: String,
    /// 请求头(JSON 对象 → Vec<(name, value)>)
    pub headers: Vec<(String, String)>,
    /// 请求体(可选)
    pub body: Option<String>,
    /// 超时秒数(缺省 30)
    pub timeout_secs: u64,
}

/// 默认超时
const DEFAULT_TIMEOUT_SECS: u64 = 30;

/// 从 config JSON 解析 HttpParams。url 必填;method 缺省 GET;timeout 缺省 30。
pub fn parse_params(config: &serde_json::Value) -> anyhow::Result<HttpParams> {
    let method = HttpMethod::parse(config.get("method").and_then(|v| v.as_str()))?;

    let url = config
        .get("url")
        .and_then(|v| v.as_str())
        .ok_or_else(|| anyhow::anyhow!("HttpNode 缺少必填参数: url"))?;
    if url.trim().is_empty() {
        anyhow::bail!("HttpNode url 不能为空");
    }

    // headers:JSON 对象 → 有序 (name, value) 列表。非对象/非字符串值忽略并 warn。
    let mut headers = Vec::new();
    if let Some(obj) = config.get("headers").and_then(|v| v.as_object()) {
        for (k, v) in obj {
            match v.as_str() {
                Some(s) => headers.push((k.clone(), s.to_string())),
                None => {
                    tracing::warn!(
                        header = %k,
                        "HttpNode headers 中 header 值非字符串,忽略"
                    );
                }
            }
        }
    }

    let body = config
        .get("body")
        .and_then(|v| v.as_str())
        .map(|s| s.to_string());

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

    Ok(HttpParams {
        method,
        url: url.to_string(),
        headers,
        body,
        timeout_secs,
    })
}

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

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

        tracing::info!(
            method = params.method.as_str(),
            url = %params.url,
            timeout_secs = params.timeout_secs,
            "HttpNode 发起请求"
        );

        let client = reqwest::Client::builder()
            .timeout(Duration::from_secs(params.timeout_secs))
            .build()?;

        let mut req = match params.method {
            HttpMethod::Get => client.get(&params.url),
            HttpMethod::Post => client.post(&params.url),
            HttpMethod::Put => client.put(&params.url),
            HttpMethod::Delete => client.delete(&params.url),
        };

        for (name, value) in &params.headers {
            req = req.header(name, value);
        }

        if let Some(body) = &params.body {
            req = req.body(body.clone());
        }

        let response = req.send().await?;
        let status_code = response.status().as_u16();

        let content_type = response
            .headers()
            .get(reqwest::header::CONTENT_TYPE)
            .and_then(|v| v.to_str().ok())
            .map(|s| s.to_string());

        let body_text = response.text().await?;

        tracing::info!(
            method = params.method.as_str(),
            url = %params.url,
            status_code,
            body_len = body_text.len(),
            "HttpNode 完成"
        );

        Ok(NodeOutput::from_value(serde_json::json!({
            "method": params.method.as_str(),
            "url": params.url,
            "status_code": status_code,
            "body": body_text,
            "content_type": content_type,
        })))
    }

    fn schema(&self) -> NodeSchema {
        NodeSchema {
            params: serde_json::json!({
                "type": "object",
                "properties": {
                    "method": {
                        "type": "string",
                        "enum": ["GET", "POST", "PUT", "DELETE"],
                        "default": "GET"
                    },
                    "url": { "type": "string", "description": "请求 URL" },
                    "headers": {
                        "type": "object",
                        "description": "请求头 JSON 对象",
                        "additionalProperties": { "type": "string" }
                    },
                    "body": { "type": "string", "description": "请求体(可选)" },
                    "timeout_secs": { "type": "integer", "default": 30 }
                },
                "required": ["url"]
            }),
            output: serde_json::json!({
                "type": "object",
                "properties": {
                    "method": { "type": "string" },
                    "url": { "type": "string" },
                    "status_code": { "type": "integer" },
                    "body": { "type": "string" },
                    "content_type": { "type": "string" }
                }
            }),
        }
    }

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

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

    // ── HttpMethod::parse ──

    #[test]
    fn method_defaults_to_get_when_absent() {
        assert_eq!(HttpMethod::parse(None).unwrap(), HttpMethod::Get);
    }

    #[test]
    fn method_case_insensitive() {
        assert_eq!(HttpMethod::parse(Some("get")).unwrap(), HttpMethod::Get);
        assert_eq!(HttpMethod::parse(Some("Post")).unwrap(), HttpMethod::Post);
        assert_eq!(HttpMethod::parse(Some("DELETE")).unwrap(), HttpMethod::Delete);
        assert_eq!(HttpMethod::parse(Some("pUt")).unwrap(), HttpMethod::Put);
    }

    #[test]
    fn method_invalid_errors() {
        let err = HttpMethod::parse(Some("PATCH")).unwrap_err().to_string();
        assert!(err.contains("非法 method"), "实际: {}", err);
        assert!(err.contains("PATCH"));
    }

    // ── parse_params(config) ──

    #[test]
    fn params_missing_url_errors() {
        let err = parse_params(&json!({ "method": "GET" }))
            .unwrap_err()
            .to_string();
        assert!(err.contains("url"), "实际: {}", err);
    }

    #[test]
    fn params_empty_url_errors() {
        let err = parse_params(&json!({ "url": "  " })).unwrap_err().to_string();
        assert!(err.contains("空"), "实际: {}", err);
    }

    #[test]
    fn params_defaults_method_and_timeout() {
        let p = parse_params(&json!({ "url": "https://example.com" })).unwrap();
        assert_eq!(p.method, HttpMethod::Get);
        assert_eq!(p.timeout_secs, DEFAULT_TIMEOUT_SECS);
        assert!(p.headers.is_empty());
        assert_eq!(p.body, None);
    }

    #[test]
    fn params_custom_timeout() {
        let p = parse_params(&json!({ "url": "https://x", "timeout_secs": 5 })).unwrap();
        assert_eq!(p.timeout_secs, 5);
    }

    #[test]
    fn params_parses_headers_object() {
        let p = parse_params(&json!({
            "url": "https://x",
            "headers": {
                "Authorization": "Bearer abc",
                "X-Trace-Id": "123"
            }
        }))
        .unwrap();
        // headers 顺序由 serde_json BTreeMap 保证;只校验集合
        let map: std::collections::HashMap<&str, &str> =
            p.headers.iter().map(|(k, v)| (k.as_str(), v.as_str())).collect();
        assert_eq!(map.get("Authorization"), Some(&"Bearer abc"));
        assert_eq!(map.get("X-Trace-Id"), Some(&"123"));
    }

    #[test]
    fn params_non_string_header_value_skipped() {
        // 非字符串值不应炸,应被跳过
        let p = parse_params(&json!({
            "url": "https://x",
            "headers": { "X-Num": 123, "X-Ok": "yes" }
        }))
        .unwrap();
        let names: Vec<&str> = p.headers.iter().map(|(k, _)| k.as_str()).collect();
        assert!(names.contains(&"X-Ok"));
        assert!(!names.contains(&"X-Num"), "非字符串 header 应被跳过");
    }

    #[test]
    fn params_parses_body() {
        let p = parse_params(&json!({
            "url": "https://x",
            "method": "POST",
            "body": "{\"k\":1}"
        }))
        .unwrap();
        assert_eq!(p.method, HttpMethod::Post);
        assert_eq!(p.body.as_deref(), Some("{\"k\":1}"));
    }

    #[test]
    fn params_invalid_method_propagates() {
        let err = parse_params(&json!({ "url": "https://x", "method": "TRACE" }))
            .unwrap_err()
            .to_string();
        assert!(err.contains("非法 method"), "实际: {}", err);
    }
}