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

//! 通知节点 — desktop(本地日志,后续接 tauri-plugin-notification) / webhook
//!
//! - type=desktop:tracing::info! 输出(桌面通知集成延后,见 PROGRESS.md 后续 Sprint)
//! - type=webhook:POST JSON {title, message} 到 webhook_url
//!
//! 节点语义:尽力而为,通知失败不阻断工作流(webhook 发送失败时输出 success=false 但仍 Ok 返回)。

use std::time::Duration;

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

/// 通知节点
pub struct NotifyNode;

/// 通知类型
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum NotifyType {
    Desktop,
    Webhook,
}

impl NotifyType {
    /// 从 config.type 字符串解析,非法值返回 Err。
    pub fn parse(raw: &str) -> anyhow::Result<Self> {
        match raw {
            "desktop" => Ok(NotifyType::Desktop),
            "webhook" => Ok(NotifyType::Webhook),
            other => anyhow::bail!(
                "NotifyNode 非法 type: {}（合法值: desktop|webhook）",
                other
            ),
        }
    }

    fn as_str(&self) -> &'static str {
        match self {
            NotifyType::Desktop => "desktop",
            NotifyType::Webhook => "webhook",
        }
    }
}

/// 从 NodeContext.config 解析出的 NotifyNode 参数。
/// 抽离此结构便于单元测试 config 解析逻辑(无需发起网络请求)。
#[derive(Debug, Clone)]
pub struct NotifyParams {
    pub notify_type: NotifyType,
    pub title: String,
    pub message: String,
    /// webhook 类型必填;desktop 类型忽略
    pub webhook_url: Option<String>,
}

/// 从 config JSON 解析 NotifyParams。type 必填;webhook 类型要求 webhook_url。
pub fn parse_params(config: &serde_json::Value) -> anyhow::Result<NotifyParams> {
    let raw_type = config
        .get("type")
        .and_then(|v| v.as_str())
        .ok_or_else(|| anyhow::anyhow!("NotifyNode 缺少必填参数: type"))?;
    let notify_type = NotifyType::parse(raw_type)?;

    let title = config
        .get("title")
        .and_then(|v| v.as_str())
        .unwrap_or("通知")
        .to_string();

    let message = config
        .get("message")
        .and_then(|v| v.as_str())
        .unwrap_or("")
        .to_string();

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

    // webhook 类型必须有 webhook_url
    if notify_type == NotifyType::Webhook {
        if webhook_url.as_deref().map(|s| s.trim().is_empty()).unwrap_or(true) {
            anyhow::bail!("NotifyNode type=webhook 缺少 webhook_url");
        }
    }

    Ok(NotifyParams {
        notify_type,
        title,
        message,
        webhook_url,
    })
}

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

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

        match params.notify_type {
            NotifyType::Desktop => {
                // 桌面通知集成延后;当前仅日志,后续接 tauri-plugin-notification。
                tracing::info!(
                    title = %params.title,
                    message = %params.message,
                    "NotifyNode desktop 通知(日志占位,集成待后续 Sprint)"
                );
                Ok(NodeOutput::from_value(serde_json::json!({
                    "type": params.notify_type.as_str(),
                    "title": params.title,
                    "message": params.message,
                    "delivered": true,
                })))
            }
            NotifyType::Webhook => {
                // webhook_url 在 parse_params 已校验非空
                let url = params.webhook_url.as_deref().unwrap();
                tracing::info!(
                    title = %params.title,
                    url = %url,
                    "NotifyNode webhook 推送"
                );

                let payload = serde_json::json!({
                    "title": params.title,
                    "message": params.message,
                });

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

                // 尽力而为:发送失败不阻断工作流,降级为 success=false + error 字段返回。
                let (status_code, success, error) = match client
                    .post(url)
                    .json(&payload)
                    .send()
                    .await
                {
                    Ok(resp) => {
                        let code = resp.status().as_u16();
                        (Some(code), code >= 200 && code < 300, None)
                    }
                    Err(e) => (None, false, Some(e.to_string())),
                };

                if !success {
                    tracing::warn!(
                        url = %url,
                        status_code = ?status_code,
                        error = ?error,
                        "NotifyNode webhook 发送失败(不阻断工作流)"
                    );
                }

                Ok(NodeOutput::from_value(serde_json::json!({
                    "type": params.notify_type.as_str(),
                    "title": params.title,
                    "message": params.message,
                    "webhook_url": url,
                    "status_code": status_code,
                    "success": success,
                    "error": error,
                })))
            }
        }
    }

    fn schema(&self) -> NodeSchema {
        NodeSchema {
            params: serde_json::json!({
                "type": "object",
                "properties": {
                    "type": {
                        "type": "string",
                        "enum": ["desktop", "webhook"]
                    },
                    "title": { "type": "string", "description": "通知标题" },
                    "message": { "type": "string", "description": "通知正文" },
                    "webhook_url": { "type": "string", "description": "webhook URL(webhook 类型必填)" }
                },
                "required": ["type"]
            }),
            output: serde_json::json!({
                "type": "object",
                "properties": {
                    "type": { "type": "string" },
                    "title": { "type": "string" },
                    "message": { "type": "string" },
                    "delivered": { "type": "boolean" },
                    "webhook_url": { "type": "string" },
                    "status_code": { "type": "integer" },
                    "success": { "type": "boolean" },
                    "error": { "type": "string" }
                }
            }),
        }
    }

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

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

    // ── NotifyType::parse ──

    #[test]
    fn type_parses_desktop_and_webhook() {
        assert_eq!(NotifyType::parse("desktop").unwrap(), NotifyType::Desktop);
        assert_eq!(NotifyType::parse("webhook").unwrap(), NotifyType::Webhook);
    }

    #[test]
    fn type_invalid_errors() {
        let err = NotifyType::parse("email").unwrap_err().to_string();
        assert!(err.contains("非法 type"), "实际: {}", err);
        assert!(err.contains("email"));
    }

    // ── parse_params(config) ──

    #[test]
    fn params_missing_type_errors() {
        let err = parse_params(&json!({ "title": "t" })).unwrap_err().to_string();
        assert!(err.contains("type"), "实际: {}", err);
    }

    #[test]
    fn params_invalid_type_errors() {
        let err = parse_params(&json!({ "type": "sms" })).unwrap_err().to_string();
        assert!(err.contains("非法 type"), "实际: {}", err);
    }

    #[test]
    fn params_desktop_defaults_title_and_message() {
        let p = parse_params(&json!({ "type": "desktop" })).unwrap();
        assert_eq!(p.notify_type, NotifyType::Desktop);
        assert_eq!(p.title, "通知"); // 默认标题
        assert_eq!(p.message, ""); // 默认空正文
        assert_eq!(p.webhook_url, None); // desktop 不需要
    }

    #[test]
    fn params_desktop_with_custom_fields() {
        let p = parse_params(&json!({
            "type": "desktop",
            "title": "构建完成",
            "message": "已发布 v1.0"
        }))
        .unwrap();
        assert_eq!(p.title, "构建完成");
        assert_eq!(p.message, "已发布 v1.0");
    }

    #[test]
    fn params_webhook_requires_url() {
        let err = parse_params(&json!({ "type": "webhook" }))
            .unwrap_err()
            .to_string();
        assert!(err.contains("webhook_url"), "实际: {}", err);
    }

    #[test]
    fn params_webhook_empty_url_errors() {
        let err = parse_params(&json!({ "type": "webhook", "webhook_url": "  " }))
            .unwrap_err()
            .to_string();
        assert!(err.contains("webhook_url"), "实际: {}", err);
    }

    #[test]
    fn params_webhook_valid() {
        let p = parse_params(&json!({
            "type": "webhook",
            "title": "告警",
            "message": "CPU > 90%",
            "webhook_url": "https://hooks.example.com/x"
        }))
        .unwrap();
        assert_eq!(p.notify_type, NotifyType::Webhook);
        assert_eq!(p.title, "告警");
        assert_eq!(p.message, "CPU > 90%");
        assert_eq!(p.webhook_url.as_deref(), Some("https://hooks.example.com/x"));
    }
}