af-notify 0.5.0

//! Transport-neutral notification commands and the durable delivery worker.

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

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

use af_context::{NotificationAttemptId, NotificationId, RequestContext, SubjectId, TenantId};
use async_trait::async_trait;
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use tokio::time::Instant;
use tokio_util::sync::CancellationToken;

/// One transport-neutral piece of notification content.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub enum Block {
    /// A heading.
    Heading(String),
    /// A paragraph.
    Text(String),
    /// Key/value fields.
    Fields(Vec<(String, String)>),
    /// A separator.
    Divider,
    /// A labelled destination.
    Action {
        /// Link label.
        label: String,
        /// Link destination.
        url: String,
    },
}

impl Block {
    /// Builds a heading block.
    pub fn heading(value: impl Into<String>) -> Self {
        Self::Heading(value.into())
    }
    /// Builds a text block.
    pub fn text(value: impl Into<String>) -> Self {
        Self::Text(value.into())
    }
    /// Builds a field block.
    pub fn fields(value: Vec<(String, String)>) -> Self {
        Self::Fields(value)
    }
    /// Builds an action block.
    pub fn action(label: impl Into<String>, url: impl Into<String>) -> Self {
        Self::Action {
            label: label.into(),
            url: url.into(),
        }
    }
}

/// Transport-neutral notification payload.
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
pub struct Notification {
    /// Optional display title.
    pub title: Option<String>,
    /// Ordered content blocks.
    pub blocks: Vec<Block>,
}

impl Notification {
    /// Builds an empty notification.
    pub fn new() -> Self {
        Self::default()
    }
    /// Builds a localized title and text body from shared catalog keys.
    pub fn from_template(
        catalog: &af_i18n::I18n,
        locale: &str,
        title_key: &str,
        body_key: &str,
        args: &[(&str, &str)],
    ) -> Self {
        Self::new()
            .title(catalog.t(locale, title_key, args))
            .block(Block::text(catalog.t(locale, body_key, args)))
    }
    /// Sets the title.
    pub fn title(mut self, value: impl Into<String>) -> Self {
        self.title = Some(value.into());
        self
    }
    /// Appends a content block.
    pub fn block(mut self, value: Block) -> Self {
        self.blocks.push(value);
        self
    }
    /// Renders a conservative plain-text representation.
    pub fn to_plain_text(&self) -> String {
        let mut out = String::new();
        if let Some(title) = &self.title {
            out.push_str(title);
            out.push_str("\n\n");
        }
        for block in &self.blocks {
            match block {
                Block::Heading(value) | Block::Text(value) => {
                    out.push_str(value);
                    out.push('\n');
                }
                Block::Fields(fields) => {
                    for (key, value) in fields {
                        out.push_str(&format!("{key}: {value}\n"));
                    }
                }
                Block::Divider => out.push_str("---\n"),
                Block::Action { label, url } => out.push_str(&format!("{label}: {url}\n")),
            }
        }
        out.trim_end().to_owned()
    }
}

/// Durable notification state.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum NotificationStatus {
    /// Waiting for a worker.
    Pending,
    /// Leased to a worker.
    Sending,
    /// Waiting for a retry deadline.
    RetryScheduled,
    /// Delivered successfully.
    Sent,
    /// Exhausted or permanently rejected.
    DeadLetter,
}

impl NotificationStatus {
    /// Stable database/wire value.
    pub fn as_str(self) -> &'static str {
        match self {
            Self::Pending => "pending",
            Self::Sending => "sending",
            Self::RetryScheduled => "retry_scheduled",
            Self::Sent => "sent",
            Self::DeadLetter => "dead_letter",
        }
    }
}

/// Transport failure retry classification.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum DeliveryClass {
    /// A later attempt may succeed.
    Retryable,
    /// Repeating the same request cannot succeed.
    Permanent,
    /// The provider did not expose enough information.
    Unknown,
}

/// Notification boundary failure.
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
pub enum NotifyError {
    /// No sender is registered for a channel.
    #[error("no sender registered for channel '{0}'")]
    UnknownChannel(String),
    /// Retryable transport failure.
    #[error("transport temporarily unavailable")]
    Retryable,
    /// Permanent transport rejection.
    #[error("transport rejected the notification")]
    Permanent,
    /// Transport result cannot be classified safely.
    #[error("transport outcome is unknown")]
    Unknown,
    /// Delivery was cancelled before completion.
    #[error("notification delivery cancelled")]
    Cancelled,
    /// Delivery exceeded its deadline.
    #[error("notification delivery deadline exceeded")]
    DeadlineExceeded,
}

impl NotifyError {
    /// Classification persisted in attempt history.
    pub fn class(&self) -> DeliveryClass {
        match self {
            Self::Retryable | Self::DeadlineExceeded => DeliveryClass::Retryable,
            Self::Permanent | Self::UnknownChannel(_) => DeliveryClass::Permanent,
            Self::Unknown | Self::Cancelled => DeliveryClass::Unknown,
        }
    }
}

/// Durable repository failure.
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
pub enum NotifyStoreError {
    /// Input failed validation.
    #[error("invalid notification request: {0}")]
    Invalid(String),
    /// Notification does not exist in the tenant.
    #[error("notification not found")]
    NotFound,
    /// One idempotency key was reused for different content.
    #[error("idempotency key was reused with different notification content")]
    IdempotencyConflict,
    /// A stale worker attempted to settle a newer lease.
    #[error("notification lease was lost")]
    LeaseLost,
    /// The caller lacks dead-letter retry authority.
    #[error("dead-letter retry is not authorized")]
    Unauthorized,
    /// Storage is unavailable.
    #[error("notification store unavailable: {0}")]
    Unavailable(String),
}

/// Cancellation and deadline supplied to one transport request.
#[derive(Clone)]
pub struct DeliveryContext {
    /// Cancellation shared with the durable worker.
    pub cancellation: CancellationToken,
    /// Absolute request deadline.
    pub deadline: Instant,
}

/// New durable notification command.
#[derive(Debug, Clone, PartialEq)]
pub struct NewOutboxItem {
    /// Owning tenant.
    pub tenant_id: TenantId,
    /// Acting subject.
    pub subject_id: SubjectId,
    /// Caller idempotency key.
    pub idempotency_key: String,
    /// Registered transport channel.
    pub channel: String,
    /// Channel-specific recipient.
    pub recipient: String,
    /// Content to deliver.
    pub notification: Notification,
    /// Maximum delivery attempts.
    pub max_attempts: u32,
}

/// Durable notification projection.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct OutboxItem {
    /// Stable identity.
    pub id: NotificationId,
    /// Owning tenant.
    pub tenant_id: TenantId,
    /// Acting subject.
    pub subject_id: SubjectId,
    /// Registered transport channel.
    pub channel: String,
    /// Channel-specific recipient.
    pub recipient: String,
    /// Content to deliver.
    pub notification: Notification,
    /// Current durable state.
    pub status: NotificationStatus,
    /// Attempts claimed so far.
    pub attempts: u32,
    /// Maximum attempts before dead-lettering.
    pub max_attempts: u32,
    /// Current lease fence.
    pub lease_version: i64,
    /// Last redacted failure class, if any.
    pub last_error: Option<String>,
    /// Creation time.
    pub created_at: DateTime<Utc>,
    /// Last projection update.
    pub updated_at: DateTime<Utc>,
}

/// One immutable delivery-attempt event.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct AttemptEvent {
    /// Event identity.
    pub id: NotificationAttemptId,
    /// Parent notification.
    pub notification_id: NotificationId,
    /// Attempt number.
    pub attempt: u32,
    /// Lease fence that authored the event.
    pub lease_version: i64,
    /// Stable event kind.
    pub kind: String,
    /// Optional failure classification.
    pub class: Option<DeliveryClass>,
    /// Event time.
    pub created_at: DateTime<Utc>,
}

/// Store-owned durable queue and query contract.
#[async_trait]
pub trait DurableOutbox: Send + Sync {
    /// Subscribes to in-process projection changes; consumers always read the
    /// durable row after receiving an id.
    fn subscribe(&self) -> tokio::sync::broadcast::Receiver<NotificationId>;
    /// Enqueues idempotently, rejecting payload mismatches.
    async fn enqueue(&self, item: NewOutboxItem) -> Result<OutboxItem, NotifyStoreError>;
    /// Reads one tenant-owned notification.
    async fn get(
        &self,
        context: &RequestContext,
        id: &NotificationId,
    ) -> Result<OutboxItem, NotifyStoreError>;
    /// Lists tenant-owned notifications after an optional cursor.
    async fn list(
        &self,
        context: &RequestContext,
        status: Option<NotificationStatus>,
        limit: usize,
        after: Option<&NotificationId>,
    ) -> Result<Vec<OutboxItem>, NotifyStoreError>;
    /// Lists immutable attempt history.
    async fn attempts(
        &self,
        context: &RequestContext,
        id: &NotificationId,
    ) -> Result<Vec<AttemptEvent>, NotifyStoreError>;
    /// Claims due work across tenants from a BYPASSRLS worker connection.
    async fn claim(
        &self,
        worker_id: &str,
        lease_secs: i64,
        batch: usize,
    ) -> Result<Vec<OutboxItem>, NotifyStoreError>;
    /// Records successful delivery under the current lease fence.
    async fn mark_sent(
        &self,
        id: &NotificationId,
        lease_version: i64,
    ) -> Result<(), NotifyStoreError>;
    /// Records a classified failure, scheduling retry or dead-lettering atomically.
    async fn record_failure(
        &self,
        id: &NotificationId,
        lease_version: i64,
        class: DeliveryClass,
        delay: Duration,
    ) -> Result<NotificationStatus, NotifyStoreError>;
    /// Requeues a tenant-owned dead letter after API authorization.
    async fn retry_dead_letter(
        &self,
        context: &RequestContext,
        id: &NotificationId,
    ) -> Result<OutboxItem, NotifyStoreError>;
}

/// A single-attempt transport. Durable retry belongs to [`Dispatcher`].
#[async_trait]
pub trait Sender: Send + Sync {
    /// Registered channel name.
    fn name(&self) -> &str;
    /// Performs exactly one bounded delivery attempt.
    async fn send(
        &self,
        context: &DeliveryContext,
        recipient: &str,
        notification: &Notification,
    ) -> Result<(), NotifyError>;
}

/// Bounded retry policy with deterministic jitter.
#[derive(Debug, Clone, Copy)]
pub struct RetryPolicy {
    /// Initial delay.
    pub base: Duration,
    /// Maximum delay.
    pub cap: Duration,
}

/// One durable worker's bounded execution settings.
#[derive(Debug, Clone)]
pub struct WorkerConfig {
    /// Stable worker identity used in leases.
    pub worker_id: String,
    /// Lease duration in seconds.
    pub lease_secs: i64,
    /// Maximum records claimed in one pass.
    pub batch: usize,
    /// Deadline for each single transport attempt.
    pub request_timeout: Duration,
    /// Durable retry delay policy.
    pub retry: RetryPolicy,
}

impl RetryPolicy {
    /// Computes capped exponential backoff with stable id-derived jitter.
    pub fn delay(self, id: &NotificationId, attempt: u32) -> Duration {
        let exponent = attempt.saturating_sub(1).min(20);
        let raw = self
            .base
            .as_millis()
            .saturating_mul(1u128 << exponent)
            .min(self.cap.as_millis());
        let hash = id.as_bytes().iter().fold(0u64, |value, byte| {
            value.wrapping_mul(31).wrapping_add(u64::from(*byte))
        });
        let jitter = 90 + hash % 21;
        Duration::from_millis(
            u64::try_from(raw.saturating_mul(u128::from(jitter)) / 100).unwrap_or(u64::MAX),
        )
    }
}

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

impl Dispatcher {
    /// Builds an empty dispatcher.
    pub fn new() -> Self {
        Self::default()
    }
    /// Registers or replaces one channel sender.
    pub fn register(&mut self, sender: Arc<dyn Sender>) -> &mut Self {
        self.senders.insert(sender.name().to_owned(), sender);
        self
    }
    /// Returns registered channel names.
    pub fn channels(&self) -> impl Iterator<Item = &str> {
        self.senders.keys().map(String::as_str)
    }
    /// Performs one transport attempt.
    pub async fn dispatch(
        &self,
        context: &DeliveryContext,
        channel: &str,
        recipient: &str,
        notification: &Notification,
    ) -> Result<(), NotifyError> {
        let sender = self
            .senders
            .get(channel)
            .ok_or_else(|| NotifyError::UnknownChannel(channel.to_owned()))?;
        sender.send(context, recipient, notification).await
    }
    /// Claims and settles one bounded worker batch.
    pub async fn drain(
        &self,
        outbox: &dyn DurableOutbox,
        config: &WorkerConfig,
        cancellation: &CancellationToken,
    ) -> Result<usize, NotifyStoreError> {
        let items = outbox
            .claim(
                &config.worker_id,
                config.lease_secs.max(
                    i64::try_from(
                        config
                            .request_timeout
                            .as_secs()
                            .saturating_mul(config.batch.clamp(1, 100) as u64)
                            .saturating_add(1),
                    )
                    .unwrap_or(i64::MAX),
                ),
                config.batch.clamp(1, 100),
            )
            .await?;
        for item in &items {
            if cancellation.is_cancelled() {
                break;
            }
            let context = DeliveryContext {
                cancellation: cancellation.child_token(),
                deadline: Instant::now() + config.request_timeout,
            };
            match self
                .dispatch(&context, &item.channel, &item.recipient, &item.notification)
                .await
            {
                Ok(()) => outbox.mark_sent(&item.id, item.lease_version).await?,
                Err(error) => {
                    outbox
                        .record_failure(
                            &item.id,
                            item.lease_version,
                            error.class(),
                            config.retry.delay(&item.id, item.attempts),
                        )
                        .await?;
                }
            }
        }
        Ok(items.len())
    }
}

/// Development sender that logs rendered content.
pub struct LogSender;

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

/// In-memory fakes for host and failure-injection tests.
pub mod testing;

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

    #[test]
    fn rendering_and_backoff_are_stable() {
        let notification = Notification::new()
            .title("Completed")
            .block(Block::fields(vec![("duration".into(), "1s".into())]));
        assert_eq!(notification.to_plain_text(), "Completed\n\nduration: 1s");
        let id = NotificationId::parse("n-1").unwrap();
        let policy = RetryPolicy {
            base: Duration::from_secs(1),
            cap: Duration::from_secs(60),
        };
        assert_eq!(policy.delay(&id, 2), policy.delay(&id, 2));
        assert!(policy.delay(&id, 3) > policy.delay(&id, 2));
        let mut catalog = af_i18n::I18n::new("en").unwrap();
        catalog
            .load_locale(
                "en",
                serde_json::json!({"title":"Done", "body":"Hello {name}"}),
            )
            .unwrap();
        assert_eq!(
            Notification::from_template(&catalog, "en", "title", "body", &[("name", "Ada")])
                .to_plain_text(),
            "Done\n\nHello Ada"
        );
    }
}