mur_chat/

notification.rs

//! Notification system — delivers workflow status updates via chat and other sinks.
//!
//! Provides a dispatcher that fans out notifications to multiple backends:
//! console, webhook, and chat platform.

use crate::platform::{ChatPlatform, WorkflowNotification};
use anyhow::{bail, Result};
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::net::IpAddr;
use std::sync::Arc;

/// Notification severity level.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum NotificationLevel {
    Info,
    Warning,
    Error,
    Success,
}

impl std::fmt::Display for NotificationLevel {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Info => write!(f, "INFO"),
            Self::Warning => write!(f, "WARN"),
            Self::Error => write!(f, "ERROR"),
            Self::Success => write!(f, "OK"),
        }
    }
}

/// A notification to be delivered.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Notification {
    pub level: NotificationLevel,
    pub title: String,
    pub body: String,
    #[serde(default)]
    pub metadata: HashMap<String, String>,
    pub timestamp: DateTime<Utc>,
}

impl Notification {
    pub fn new(
        level: NotificationLevel,
        title: impl Into<String>,
        body: impl Into<String>,
    ) -> Self {
        Self {
            level,
            title: title.into(),
            body: body.into(),
            metadata: HashMap::new(),
            timestamp: Utc::now(),
        }
    }

    pub fn with_meta(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
        self.metadata.insert(key.into(), value.into());
        self
    }
}

/// Console sink — prints notifications to stderr.
pub struct ConsoleSink;

impl ConsoleSink {
    pub async fn send(&self, notification: &Notification) -> Result<()> {
        let emoji = match notification.level {
            NotificationLevel::Info => "ℹ️",
            NotificationLevel::Warning => "⚠️",
            NotificationLevel::Error => "❌",
            NotificationLevel::Success => "✅",
        };
        eprintln!(
            "{} [{}] {}: {}",
            emoji, notification.level, notification.title, notification.body
        );
        Ok(())
    }
}

/// Webhook sink — POSTs notification as JSON to a URL.
pub struct WebhookSink {
    url: String,
    client: reqwest::Client,
}

impl WebhookSink {
    /// Create a new webhook sink, validating the URL at construction time.
    ///
    /// Rejects URLs that:
    /// - Are not valid URLs
    /// - Use a scheme other than http or https
    /// - Target localhost or private/loopback IP addresses (SSRF mitigation)
    pub fn new(url: impl Into<String>) -> Result<Self> {
        let url = url.into();
        Self::validate_url(&url)?;
        Ok(Self {
            url,
            client: reqwest::Client::builder()
                .connect_timeout(std::time::Duration::from_secs(10))
                .timeout(std::time::Duration::from_secs(30))
                .build()
                .unwrap_or_else(|_| reqwest::Client::new()),
        })
    }

    /// Validate a webhook URL: must be a valid URL with http(s) scheme and
    /// must not point to loopback or private IP addresses (SSRF protection).
    fn validate_url(url: &str) -> Result<()> {
        let parsed = reqwest::Url::parse(url)
            .map_err(|e| anyhow::anyhow!("Invalid webhook URL '{}': {}", url, e))?;

        match parsed.scheme() {
            "http" | "https" => {}
            other => bail!("Webhook URL scheme must be http or https, got '{}'", other),
        }

        if let Some(host) = parsed.host_str() {
            // Block well-known loopback hostnames
            let lower = host.to_lowercase();
            if lower == "localhost" || lower == "[::1]" {
                bail!("Webhook URL must not target localhost: {}", host);
            }

            // Block loopback and private IPs
            if let Ok(ip) = host.parse::<IpAddr>() {
                if ip.is_loopback() {
                    bail!("Webhook URL must not target loopback address: {}", ip);
                }
                if let IpAddr::V4(v4) = ip {
                    if v4.is_private() || v4.is_link_local() || v4.is_unspecified() {
                        bail!(
                            "Webhook URL must not target private/link-local IP: {}",
                            v4
                        );
                    }
                }
                if let IpAddr::V6(v6) = ip {
                    if v6.is_unspecified() {
                        bail!(
                            "Webhook URL must not target unspecified IPv6: {}",
                            v6
                        );
                    }
                }
            }
        } else {
            bail!("Webhook URL has no host");
        }

        Ok(())
    }

    pub async fn send(&self, notification: &Notification) -> Result<()> {
        self.client
            .post(&self.url)
            .json(notification)
            .send()
            .await?
            .error_for_status()?;
        Ok(())
    }
}

/// Chat notification sink — sends workflow notifications via ChatPlatform.
pub struct ChatNotificationSink<P: ChatPlatform> {
    platform: Arc<P>,
    default_channel: String,
}

impl<P: ChatPlatform + 'static> ChatNotificationSink<P> {
    pub fn new(platform: Arc<P>, default_channel: String) -> Self {
        Self {
            platform,
            default_channel,
        }
    }

    /// Notify about a workflow result.
    pub async fn notify_workflow(
        &self,
        notification: &WorkflowNotification,
        thread_id: Option<&str>,
    ) -> Result<String> {
        self.platform
            .send_notification(&self.default_channel, thread_id, notification)
            .await
    }

    /// Send a generic notification as a chat message.
    pub async fn send(&self, notification: &Notification) -> Result<()> {
        let emoji = match notification.level {
            NotificationLevel::Info => "ℹ️",
            NotificationLevel::Warning => "⚠️",
            NotificationLevel::Error => "❌",
            NotificationLevel::Success => "✅",
        };

        let msg = crate::platform::OutgoingMessage {
            channel_id: self.default_channel.clone(),
            text: format!(
                "{} *{}*\n{}",
                emoji, notification.title, notification.body
            ),
            thread_id: None,
            blocks: None,
        };

        self.platform.send_message(&msg).await?;
        Ok(())
    }
}

/// Dispatcher that sends notifications to console and optional webhook/chat sinks.
pub struct NotificationDispatcher {
    console: ConsoleSink,
    webhook: Option<WebhookSink>,
}

impl NotificationDispatcher {
    pub fn new() -> Self {
        Self {
            console: ConsoleSink,
            webhook: None,
        }
    }

    /// Add a webhook endpoint. Returns an error if the URL is invalid, uses a
    /// non-http(s) scheme, or targets a private/loopback IP address.
    pub fn with_webhook(mut self, url: impl Into<String>) -> Result<Self> {
        self.webhook = Some(WebhookSink::new(url)?);
        Ok(self)
    }

    /// Send a notification to all configured sinks.
    pub async fn notify(&self, notification: &Notification) {
        // Always log to console
        if let Err(e) = self.console.send(notification).await {
            tracing::warn!("Console notification failed: {}", e);
        }

        // Optionally send to webhook
        if let Some(ref webhook) = self.webhook {
            if let Err(e) = webhook.send(notification).await {
                tracing::warn!("Webhook notification failed: {}", e);
            }
        }
    }

    /// Convenience: send an info notification.
    pub async fn info(&self, title: impl Into<String>, body: impl Into<String>) {
        self.notify(&Notification::new(NotificationLevel::Info, title, body))
            .await;
    }

    /// Convenience: send an error notification.
    pub async fn error(&self, title: impl Into<String>, body: impl Into<String>) {
        self.notify(&Notification::new(NotificationLevel::Error, title, body))
            .await;
    }

    /// Convenience: send a success notification.
    pub async fn success(&self, title: impl Into<String>, body: impl Into<String>) {
        self.notify(&Notification::new(NotificationLevel::Success, title, body))
            .await;
    }
}

impl Default for NotificationDispatcher {
    fn default() -> Self {
        Self::new()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::platform::{
        ApprovalRequest, OutgoingMessage, ProgressUpdate,
    };
    use std::sync::Mutex;

    struct MockPlatform {
        notifications: Arc<Mutex<Vec<(bool, String)>>>,
        messages: Arc<Mutex<Vec<String>>>,
    }

    impl MockPlatform {
        fn new() -> Self {
            Self {
                notifications: Arc::new(Mutex::new(Vec::new())),
                messages: Arc::new(Mutex::new(Vec::new())),
            }
        }
    }

    impl ChatPlatform for MockPlatform {
        async fn send_message(&self, msg: &OutgoingMessage) -> anyhow::Result<String> {
            self.messages.lock().unwrap().push(msg.text.clone());
            Ok("ts".into())
        }
        async fn send_approval(
            &self,
            _channel_id: &str,
            _request: &ApprovalRequest,
        ) -> anyhow::Result<String> {
            Ok("ts".into())
        }
        async fn update_message(
            &self,
            _channel_id: &str,
            _message_id: &str,
            _text: &str,
        ) -> anyhow::Result<()> {
            Ok(())
        }
        async fn add_reaction(
            &self,
            _channel_id: &str,
            _message_id: &str,
            _emoji: &str,
        ) -> anyhow::Result<()> {
            Ok(())
        }
        async fn send_progress(
            &self,
            _channel_id: &str,
            _thread_id: &str,
            _progress: &ProgressUpdate,
        ) -> anyhow::Result<String> {
            Ok("ts".into())
        }
        async fn send_notification(
            &self,
            _channel_id: &str,
            _thread_id: Option<&str>,
            notification: &WorkflowNotification,
        ) -> anyhow::Result<String> {
            self.notifications
                .lock()
                .unwrap()
                .push((notification.success, notification.workflow_id.clone()));
            Ok("ts".into())
        }
        async fn start_thread(
            &self,
            _channel_id: &str,
            _execution_id: &str,
            _workflow_id: &str,
            _total_steps: usize,
            _shadow: bool,
        ) -> anyhow::Result<String> {
            Ok("thread-ts".into())
        }
    }

    #[tokio::test]
    async fn test_chat_notification_sink_success() {
        let platform = Arc::new(MockPlatform::new());
        let sink = ChatNotificationSink::new(platform.clone(), "#ops".into());

        let notif = WorkflowNotification {
            execution_id: "e1".into(),
            workflow_id: "deploy".into(),
            success: true,
            steps_completed: 3,
            total_steps: 3,
            duration_ms: 1500,
            error: None,
        };
        sink.notify_workflow(&notif, None).await.unwrap();

        let notifs = platform.notifications.lock().unwrap();
        assert_eq!(notifs.len(), 1);
        assert!(notifs[0].0);
        assert_eq!(notifs[0].1, "deploy");
    }

    #[tokio::test]
    async fn test_chat_notification_sink_failure() {
        let platform = Arc::new(MockPlatform::new());
        let sink = ChatNotificationSink::new(platform.clone(), "#ops".into());

        let notif = WorkflowNotification {
            execution_id: "e1".into(),
            workflow_id: "deploy".into(),
            success: false,
            steps_completed: 1,
            total_steps: 3,
            duration_ms: 500,
            error: Some("step failed".into()),
        };
        sink.notify_workflow(&notif, None).await.unwrap();

        let notifs = platform.notifications.lock().unwrap();
        assert_eq!(notifs.len(), 1);
        assert!(!notifs[0].0);
    }

    #[tokio::test]
    async fn test_chat_notification_sink_generic() {
        let platform = Arc::new(MockPlatform::new());
        let sink = ChatNotificationSink::new(platform.clone(), "#ops".into());

        let notification =
            Notification::new(NotificationLevel::Info, "Test", "hello world");
        sink.send(&notification).await.unwrap();

        let msgs = platform.messages.lock().unwrap();
        assert_eq!(msgs.len(), 1);
        assert!(msgs[0].contains("Test"));
    }

    #[tokio::test]
    async fn test_dispatcher() {
        let dispatcher = NotificationDispatcher::new();
        // Just verify it doesn't panic
        dispatcher.info("Test", "hello").await;
        dispatcher.error("Fail", "oops").await;
        dispatcher.success("Done", "all good").await;
    }

    #[test]
    fn test_notification_with_meta() {
        let n = Notification::new(NotificationLevel::Info, "Test", "body")
            .with_meta("workflow", "deploy")
            .with_meta("step", "3");
        assert_eq!(n.metadata.len(), 2);
        assert_eq!(n.metadata["workflow"], "deploy");
    }

    #[test]
    fn test_notification_level_display() {
        assert_eq!(NotificationLevel::Info.to_string(), "INFO");
        assert_eq!(NotificationLevel::Error.to_string(), "ERROR");
        assert_eq!(NotificationLevel::Success.to_string(), "OK");
        assert_eq!(NotificationLevel::Warning.to_string(), "WARN");
    }
}