af-notify 0.4.0

//! `af-notify` — notification dispatcher + transport seam. Port of the
//! reusable core of `agent_core/notifications/`.
//!
//! The governance rule that carries over: **every outbound message goes through
//! the dispatcher**, never a raw `api.telegram.org/...` POST. A product builds a
//! typed [`Notification`] (transport-agnostic blocks) and dispatches it; each
//! [`Sender`] renders the blocks its own way (a chat transport flattens to
//! text, a rich transport keeps structure).
//!
//! Transports (Telegram / email / Discord) are **pluggable adapters** that
//! implement [`Sender`] — the core never depends on a specific provider.
//!
//! ```
//! use af_notify::{Block, Dispatcher, LogSender, Notification};
//! use std::sync::Arc;
//!
//! # tokio_test(async {
//! let mut d = Dispatcher::new();
//! d.register(Arc::new(LogSender));
//! let notif = Notification::new()
//!     .title("Run finished")
//!     .block(Block::text("Your workflow completed."))
//!     .block(Block::fields(vec![("duration".into(), "1.2s".into())]));
//! d.dispatch("log", "ops", &notif).await.unwrap();
//! # });
//! # fn tokio_test<F: std::future::Future>(_: F) {}
//! ```

#![deny(missing_docs)]
#![deny(rustdoc::broken_intra_doc_links)]

use std::collections::HashMap;
use std::sync::Arc;

use async_trait::async_trait;
use serde::{Deserialize, Serialize};

/// One piece of a notification. Transport-agnostic; senders decide how to
/// render each block.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub enum Block {
    /// A heading / section title.
    Heading(String),
    /// A paragraph of body text.
    Text(String),
    /// Key/value pairs (rendered as a definition list or `key: value` lines).
    Fields(Vec<(String, String)>),
    /// A visual separator.
    Divider,
    /// A call to action with a label + URL.
    Action {
        /// Button or link text.
        label: String,
        /// Destination URL.
        url: String,
    },
}

impl Block {
    /// A heading block.
    pub fn heading(s: impl Into<String>) -> Self {
        Block::Heading(s.into())
    }
    /// A paragraph block.
    pub fn text(s: impl Into<String>) -> Self {
        Block::Text(s.into())
    }
    /// A key/value table block.
    pub fn fields(kv: Vec<(String, String)>) -> Self {
        Block::Fields(kv)
    }
    /// A call-to-action link block.
    pub fn action(label: impl Into<String>, url: impl Into<String>) -> Self {
        Block::Action {
            label: label.into(),
            url: url.into(),
        }
    }
}

/// A transport-agnostic notification.
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
pub struct Notification {
    /// Display title.
    pub title: Option<String>,
    /// Ordered content blocks rendered by each transport.
    pub blocks: Vec<Block>,
}

impl Notification {
    /// An empty notification without a title.
    pub fn new() -> Self {
        Self::default()
    }

    /// Set the title.
    pub fn title(mut self, t: impl Into<String>) -> Self {
        self.title = Some(t.into());
        self
    }

    /// Append a content block.
    pub fn block(mut self, b: Block) -> Self {
        self.blocks.push(b);
        self
    }

    /// Reference plain-text rendering for chat/SMS transports.
    pub fn to_plain_text(&self) -> String {
        let mut out = String::new();
        if let Some(t) = &self.title {
            out.push_str(t);
            out.push_str("\n\n");
        }
        for block in &self.blocks {
            match block {
                Block::Heading(h) => {
                    out.push_str(h);
                    out.push('\n');
                }
                Block::Text(t) => {
                    out.push_str(t);
                    out.push('\n');
                }
                Block::Fields(kv) => {
                    for (k, v) in kv {
                        out.push_str(&format!("{k}: {v}\n"));
                    }
                }
                Block::Divider => out.push_str("---\n"),
                Block::Action { label, url } => out.push_str(&format!("{label}: {url}\n")),
            }
        }
        out.trim_end().to_string()
    }
}

/// Failure while dispatching a notification.
#[derive(Debug, thiserror::Error)]
pub enum NotifyError {
    /// No sender registered for channel ''.
    #[error("no sender registered for channel '{0}'")]
    UnknownChannel(String),
    /// The channel sender failed.
    #[error("transport '{transport}' failed: {reason}")]
    Transport {
        /// Channel name of the failing sender.
        transport: String,
        /// Transport-reported failure.
        reason: String,
    },
}

/// A transport. The `name` is the channel key callers dispatch to
/// (`"telegram"`, `"email"`, `"log"`, …).
#[async_trait]
pub trait Sender: Send + Sync {
    /// Channel name this sender serves (for example `telegram`).
    fn name(&self) -> &str;
    /// Deliver `notif` to `recipient` (chat id / email address / webhook …).
    async fn send(&self, recipient: &str, notif: &Notification) -> Result<(), NotifyError>;
}

/// A durably queued notification claimed by a worker.
#[derive(Debug, Clone, PartialEq)]
pub struct OutboxItem {
    /// Stable identifier of this record.
    pub id: String,
    /// Tenant that owns this record.
    pub tenant_id: String,
    /// Subject (user or service principal) acting on or owning this record.
    pub subject_id: String,
    /// Channel the notification is routed to.
    pub channel: String,
    /// Channel-specific recipient address.
    pub recipient: String,
    /// The notification to deliver.
    pub notification: Notification,
    /// Delivery attempts made so far.
    pub attempts: u32,
    /// Fencing token bumped on every lease acquisition; stale holders cannot write.
    pub lease_version: i64,
}

/// A notification to enqueue.
#[derive(Debug, Clone, PartialEq)]
pub struct NewOutboxItem {
    /// Tenant that owns the notification.
    pub tenant_id: String,
    /// Subject the notification is about.
    pub subject_id: String,
    /// Caller-supplied key that makes repeated submissions return the first result.
    pub idempotency_key: String,
    /// Channel the notification is routed to.
    pub channel: String,
    /// Channel-specific recipient address.
    pub recipient: String,
    /// The notification to deliver.
    pub notification: Notification,
}

/// Lease-fenced durable queue between producers and transport senders.
#[async_trait]
pub trait DurableOutbox: Send + Sync {
    /// Persist a notification and return its id.
    async fn enqueue(&self, item: NewOutboxItem) -> Result<String, String>;
    /// Claim up to `batch` due items for `worker_id` with a lease of `lease_secs`.
    async fn claim(
        &self,
        worker_id: &str,
        lease_secs: i64,
        batch: usize,
    ) -> Result<Vec<OutboxItem>, String>;
    /// Mark a claimed item delivered; fails when the lease was lost.
    async fn mark_sent(&self, id: &str, lease_version: i64) -> Result<(), String>;
    /// Release a claimed item for a later attempt with the given error.
    async fn retry(
        &self,
        id: &str,
        lease_version: i64,
        error: &str,
        delay_secs: i64,
    ) -> Result<(), String>;
}

/// Routes notifications to registered senders by channel name.
#[derive(Default, Clone)]
pub struct Dispatcher {
    senders: HashMap<String, Arc<dyn Sender>>,
}

impl Dispatcher {
    /// A dispatcher with no senders registered.
    pub fn new() -> Self {
        Self::default()
    }

    /// Register a sender for its channel; a later registration for the same channel replaces the earlier one.
    pub fn register(&mut self, sender: Arc<dyn Sender>) -> &mut Self {
        self.senders.insert(sender.name().to_string(), sender);
        self
    }

    /// Channels with a registered sender.
    pub fn channels(&self) -> impl Iterator<Item = &str> {
        self.senders.keys().map(|s| s.as_str())
    }

    /// Dispatch to the sender registered under `channel`.
    pub async fn dispatch(
        &self,
        channel: &str,
        recipient: &str,
        notif: &Notification,
    ) -> Result<(), NotifyError> {
        let sender = self
            .senders
            .get(channel)
            .ok_or_else(|| NotifyError::UnknownChannel(channel.to_string()))?;
        sender.send(recipient, notif).await
    }

    /// Claim due outbox items and deliver each through its channel sender, recording success or scheduling a retry.
    pub async fn drain(
        &self,
        outbox: &dyn DurableOutbox,
        worker_id: &str,
        lease_secs: i64,
        batch: usize,
        retry_delay_secs: i64,
    ) -> Result<usize, String> {
        let items = outbox
            .claim(worker_id, lease_secs, batch.clamp(1, 100))
            .await?;
        for item in &items {
            match self
                .dispatch(&item.channel, &item.recipient, &item.notification)
                .await
            {
                Ok(()) => outbox.mark_sent(&item.id, item.lease_version).await?,
                Err(error) => {
                    outbox
                        .retry(
                            &item.id,
                            item.lease_version,
                            &error.to_string(),
                            retry_delay_secs,
                        )
                        .await?
                }
            }
        }
        Ok(items.len())
    }
}

/// Default sender: emits the rendered notification to the log. Useful in dev
/// and as the reference transport.
pub struct LogSender;

#[async_trait]
impl Sender for LogSender {
    fn name(&self) -> &str {
        "log"
    }
    async fn send(&self, recipient: &str, notif: &Notification) -> Result<(), NotifyError> {
        tracing::info!(target: "notify", recipient, body = %notif.to_plain_text(), "notification");
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::Mutex;

    #[derive(Default)]
    struct MemoryOutbox {
        items: Mutex<Vec<OutboxItem>>,
        sent: Mutex<Vec<String>>,
    }

    #[async_trait]
    impl DurableOutbox for MemoryOutbox {
        async fn enqueue(&self, item: NewOutboxItem) -> Result<String, String> {
            let id = item.idempotency_key.clone();
            self.items.lock().unwrap().push(OutboxItem {
                id: id.clone(),
                tenant_id: item.tenant_id,
                subject_id: item.subject_id,
                channel: item.channel,
                recipient: item.recipient,
                notification: item.notification,
                attempts: 0,
                lease_version: 1,
            });
            Ok(id)
        }
        async fn claim(&self, _: &str, _: i64, _: usize) -> Result<Vec<OutboxItem>, String> {
            Ok(self.items.lock().unwrap().clone())
        }
        async fn mark_sent(&self, id: &str, _: i64) -> Result<(), String> {
            self.sent.lock().unwrap().push(id.into());
            Ok(())
        }
        async fn retry(&self, _: &str, _: i64, _: &str, _: i64) -> Result<(), String> {
            Ok(())
        }
    }

    #[test]
    fn renders_plain_text() {
        let n = Notification::new()
            .title("Run finished")
            .block(Block::heading("Summary"))
            .block(Block::text("All good."))
            .block(Block::fields(vec![("duration".into(), "1.2s".into())]))
            .block(Block::action("View", "https://x/y"));
        let txt = n.to_plain_text();
        assert!(txt.starts_with("Run finished"));
        assert!(txt.contains("duration: 1.2s"));
        assert!(txt.contains("View: https://x/y"));
    }

    struct CapturingSender(Mutex<Vec<String>>);
    #[async_trait]
    impl Sender for CapturingSender {
        fn name(&self) -> &str {
            "capture"
        }
        async fn send(&self, recipient: &str, notif: &Notification) -> Result<(), NotifyError> {
            self.0
                .lock()
                .unwrap()
                .push(format!("{recipient}|{}", notif.to_plain_text()));
            Ok(())
        }
    }

    #[tokio::test]
    async fn dispatch_routes_to_named_sender() {
        let sender = Arc::new(CapturingSender(Mutex::new(Vec::new())));
        let mut d = Dispatcher::new();
        d.register(sender.clone());

        let n = Notification::new().block(Block::text("hi"));
        d.dispatch("capture", "user1", &n).await.unwrap();

        assert_eq!(sender.0.lock().unwrap().len(), 1);
        assert!(sender.0.lock().unwrap()[0].starts_with("user1|hi"));
    }

    #[tokio::test]
    async fn unknown_channel_errors() {
        let d = Dispatcher::new();
        let n = Notification::new();
        assert!(matches!(
            d.dispatch("nope", "x", &n).await,
            Err(NotifyError::UnknownChannel(_))
        ));
    }

    #[tokio::test]
    async fn durable_dispatch_marks_claimed_messages_sent() {
        let sender = Arc::new(CapturingSender(Mutex::new(Vec::new())));
        let mut dispatcher = Dispatcher::new();
        dispatcher.register(sender);
        let outbox = MemoryOutbox::default();
        outbox
            .enqueue(NewOutboxItem {
                tenant_id: "tenant".into(),
                subject_id: "subject".into(),
                idempotency_key: "one".into(),
                channel: "capture".into(),
                recipient: "recipient".into(),
                notification: Notification::new().block(Block::text("hello")),
            })
            .await
            .unwrap();
        assert_eq!(
            dispatcher
                .drain(&outbox, "worker", 30, 10, 5)
                .await
                .unwrap(),
            1
        );
        assert_eq!(&*outbox.sent.lock().unwrap(), &["one"]);
    }
}