eric / rustwarden

//! Outbound ActivityPub delivery queue + worker (RFC 0011).
//!
//! Persistent journal of activities that must be delivered to remote
//! inboxes (auth `Accept{AuthRequest}`, invitations, lifecycle signals).
//! The transport layer is abstracted behind `OutboundDelivery` so
//! that RFC 0010 can plug in HTTP-Sig signing without disturbing the
//! queue/worker plumbing.
//!

use async_trait::async_trait;
use chrono::Utc;
use sea_orm::{
    ActiveModelTrait, ColumnTrait, DatabaseConnection, EntityTrait, ExprTrait, QueryFilter,
    QueryOrder, Set,
};
use std::time::Duration;
use uuid::Uuid;

use crate::models::fediverse_outbound_queue;
use rustwarden_core::error::AppError;

/// Lifecycle of an outbound queue row.
///
/// In the Phase-2 slice livered by RFC 0011 only `Pending → Delivered`
/// and `Pending → Failed` transitions occur. RFC 0010 introduces the
/// retry/back-off ladder that uses `Failed → Pending` and the terminal
/// `Failed → Dead`.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum OutboundStatus {
    /// Eligible for delivery.
    Pending,
    /// Successfully delivered.
    Delivered,
    /// Last attempt failed; not (yet) eligible for retry.
    Failed,
    /// Reached `max_attempts`, no further attempts will be made.
    Dead,
}

impl OutboundStatus {
    /// RFC 0010 retry ladder. The returned offset is added to `now`
    /// to compute `next_retry_at` after a failed delivery. Sequence:
    /// 5 min → 30 min → 2 h → 6 h → 24 h → 48 h, capped at 48 h.
    /// `attempts` is the **post-increment** value (number of failed
    /// attempts so far).
    pub fn next_retry_offset(attempts: i32) -> std::time::Duration {
        const LADDER_SECS: &[u64] = &[5 * 60, 30 * 60, 2 * 3600, 6 * 3600, 24 * 3600, 48 * 3600];
        let bounded = std::cmp::Ord::max(attempts, 1) as usize - 1;
        let idx = bounded.min(LADDER_SECS.len() - 1);
        std::time::Duration::from_secs(LADDER_SECS[idx])
    }

    /// Wire form stored in the `status` column.
    pub fn as_str(self) -> &'static str {
        match self {
            Self::Pending => "pending",
            Self::Delivered => "delivered",
            Self::Failed => "failed",
            Self::Dead => "dead",
        }
    }

    /// Best-effort parse — unknown values fall back to `Pending` so a
    /// data corruption never silently drops a row.
    pub fn parse(raw: &str) -> Self {
        match raw.trim().to_ascii_lowercase().as_str() {
            "delivered" => Self::Delivered,
            "failed" => Self::Failed,
            "dead" => Self::Dead,
            _ => Self::Pending,
        }
    }
}

/// Caller-supplied metadata used to enqueue an activity for delivery.
#[derive(Debug, Clone)]
pub struct EnqueueRequest {
    /// Canonical AP `id` of the activity (e.g. `https://a.tld/activities/uuid`).
    pub activity_id: String,
    /// Pre-rendered JSON-LD body that will be POSTed to `target_inbox`.
    pub payload_json: String,
    /// Inbox URL.
    pub target_inbox: String,
    /// Domain owning the inbox (used by trust checks).
    pub target_domain: String,
    /// Local actor on whose behalf the activity is signed. Nullable
    /// while only `NoopDelivery` is wired; required by RFC 0010.
    pub actor_user_id: Option<Uuid>,
}

/// Transport layer abstraction.
///
/// The Phase-2 slice ships only [`NoopDelivery`]. RFC 0010 will swap in
/// `HttpSignedDelivery` that uses RFC 0005 keys to sign requests.
#[async_trait]
pub trait OutboundDelivery: Send + Sync {
    /// Attempt to deliver `item` to its `target_inbox`. Returning `Ok`
    /// promotes the row to `Delivered`; returning `Err` flips it to
    /// `Failed` and increments `attempts`.
    async fn deliver(&self, item: &fediverse_outbound_queue::Model) -> Result<(), AppError>;
}

/// No-op transport for the slice livered by RFC 0011: pretends every
/// delivery succeeded so the queue/worker plumbing can be exercised
/// end-to-end without an HTTP stack. Production deployments must swap
/// for `HttpSignedDelivery` once RFC 0010 lands.
pub struct NoopDelivery;

#[async_trait]
impl OutboundDelivery for NoopDelivery {
    async fn deliver(&self, item: &fediverse_outbound_queue::Model) -> Result<(), AppError> {
        tracing::debug!(
            target = %item.target_inbox,
            activity = %item.activity_id,
            "NoopDelivery: pretending delivery succeeded",
        );
        Ok(())
    }
}

/// Stateless service exposing enqueue + drain operations on the
/// outbound queue.
pub struct FediverseOutboundService;

impl FediverseOutboundService {
    /// Persist a new `pending` row.
    pub async fn enqueue(
        db: &DatabaseConnection,
        request: EnqueueRequest,
    ) -> Result<fediverse_outbound_queue::Model, AppError> {
        let now = Utc::now();
        let model = fediverse_outbound_queue::ActiveModel {
            id: Set(Uuid::new_v4()),
            activity_id: Set(request.activity_id),
            payload_json: Set(request.payload_json),
            target_inbox: Set(request.target_inbox),
            target_domain: Set(request.target_domain),
            status: Set(OutboundStatus::Pending.as_str().to_string()),
            attempts: Set(0),
            max_attempts: Set(8),
            last_error: Set(None),
            next_retry_at: Set(None),
            created_at: Set(now),
            delivered_at: Set(None),
            actor_user_id: Set(request.actor_user_id),
        };
        Ok(model.insert(db).await?)
    }

    /// Pick one eligible row, hand it to `transport`, and persist the
    /// outcome. Returns the updated row, or `None` if the queue is
    /// empty.
    pub async fn drain_one<T: OutboundDelivery>(
        db: &DatabaseConnection,
        transport: &T,
    ) -> Result<Option<fediverse_outbound_queue::Model>, AppError> {
        let now = Utc::now();
        let candidate = fediverse_outbound_queue::Entity::find()
            .filter(fediverse_outbound_queue::Column::Status.eq(OutboundStatus::Pending.as_str()))
            .filter(
                fediverse_outbound_queue::Column::NextRetryAt
                    .is_null()
                    .or(fediverse_outbound_queue::Column::NextRetryAt.lte(now)),
            )
            .order_by_asc(fediverse_outbound_queue::Column::CreatedAt)
            .one(db)
            .await?;

        let Some(model) = candidate else {
            return Ok(None);
        };

        let delivery = transport.deliver(&model).await;
        let next_attempts = model.attempts + 1;
        let max_attempts = model.max_attempts;
        let model_for_update = model.clone();
        let mut active: fediverse_outbound_queue::ActiveModel = model_for_update.into();
        active.attempts = Set(next_attempts);
        match &delivery {
            Ok(()) => {
                active.status = Set(OutboundStatus::Delivered.as_str().to_string());
                active.delivered_at = Set(Some(Utc::now()));
                active.last_error = Set(None);
                active.next_retry_at = Set(None);
            }
            Err(err) => {
                active.last_error = Set(Some(err.to_string()));
                if next_attempts >= max_attempts {
                    // RFC 0010: terminal — no further retry.
                    active.status = Set(OutboundStatus::Dead.as_str().to_string());
                    active.next_retry_at = Set(None);
                } else {
                    // RFC 0010: stay `pending` and gate on `next_retry_at`
                    // so the partial-index scan ignores the row until the
                    // backoff has elapsed.
                    let offset = OutboundStatus::next_retry_offset(next_attempts);
                    let next = Utc::now()
                        + chrono::Duration::from_std(offset).map_err(|e| {
                            AppError::Internal(format!(
                                "retry offset out of range for chrono::Duration: {e}",
                            ))
                        })?;
                    active.status = Set(OutboundStatus::Pending.as_str().to_string());
                    active.next_retry_at = Set(Some(next));
                }
            }
        }
        let updated = active.update(db).await?;

        // RFC 0019: structured tracing events — drive Prometheus / Loki /
        // OTEL exporters without coupling rustwarden-core to a metrics
        // crate. Gated by `app_settings.fediverse_observability_enabled`
        // (default `false`) — opt-in to avoid log noise on instances
        // that pipe nothing to an observability backend.
        let observability = is_fediverse_observability_enabled(db)
            .await
            .unwrap_or(false);
        if observability {
            match (&delivery, updated.status.as_str()) {
                (Ok(()), _) => {
                    tracing::info!(
                        target: "rustwarden::fediverse::outbound",
                        activity_id = %updated.activity_id,
                        target_inbox = %updated.target_inbox,
                        target_domain = %updated.target_domain,
                        attempts = updated.attempts,
                        event = "delivered",
                    );
                }
                (Err(err), status) if status == OutboundStatus::Dead.as_str() => {
                    tracing::error!(
                        target: "rustwarden::fediverse::outbound",
                        activity_id = %updated.activity_id,
                        target_domain = %updated.target_domain,
                        last_error = %err,
                        attempts = updated.attempts,
                        max_attempts = updated.max_attempts,
                        event = "dead",
                    );
                }
                (Err(err), _) => {
                    tracing::warn!(
                        target: "rustwarden::fediverse::outbound",
                        activity_id = %updated.activity_id,
                        target_domain = %updated.target_domain,
                        last_error = %err,
                        attempts = updated.attempts,
                        next_retry_at = ?updated.next_retry_at,
                        event = "deferred",
                    );
                }
            }
        }

        Ok(Some(updated))
    }
}

/// RFC 0019 + S4 observability toggle — `app_settings.fediverse_observability_enabled`,
/// default **false**. When `false`, fediverse routes/services skip the
/// structured `tracing::info!` / `warn!` / `error!` emit-points used for
/// metric aggregation. Cheap (1 SELECT by key) ; cached by the upstream
/// `AppSettingsService` if the consumer wires its standard cache.
pub async fn is_fediverse_observability_enabled(db: &DatabaseConnection) -> Result<bool, AppError> {
    rustwarden_core::services::app_settings_service::AppSettingsService::get_bool(
        db,
        "fediverse_observability_enabled",
        false,
    )
    .await
}

/// Background loop that periodically drains the queue.
///
/// Spawned by the federation router builder (RFC 0011 hooks
/// the spawn into the federation router builder in a follow-up patch — RFC 0006
/// will be the first consumer and triggers the spawn explicitly).
pub struct OutboundWorker<T: OutboundDelivery> {
    db: DatabaseConnection,
    transport: T,
    poll_interval: Duration,
}

impl<T: OutboundDelivery> OutboundWorker<T> {
    /// Build a worker that polls every `poll_interval`.
    pub fn new(db: DatabaseConnection, transport: T, poll_interval: Duration) -> Self {
        Self {
            db,
            transport,
            poll_interval,
        }
    }

    /// Run the polling loop forever. Errors during a single delivery
    /// are logged but do not stop the loop. Cancellation is achieved
    /// by cancelling the surrounding task.
    pub async fn run(self) {
        loop {
            match FediverseOutboundService::drain_one(&self.db, &self.transport).await {
                Ok(Some(_)) => continue, // drain greedily while items exist
                Ok(None) => tokio::time::sleep(self.poll_interval).await,
                Err(err) => {
                    tracing::error!(
                        target: "rustwarden::fediverse::outbound",
                        error = %err,
                        event = "drain_failed",
                    );
                    tokio::time::sleep(self.poll_interval).await;
                }
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use sea_orm::{DatabaseBackend, MockDatabase, MockExecResult};

    fn pending_row(id: Uuid) -> fediverse_outbound_queue::Model {
        let now = Utc::now();
        fediverse_outbound_queue::Model {
            id,
            activity_id: format!("https://a.tld/activities/{id}"),
            payload_json: "{}".to_string(),
            target_inbox: "https://b.tld/inbox".to_string(),
            target_domain: "b.tld".to_string(),
            status: "pending".to_string(),
            attempts: 0,
            max_attempts: 8,
            last_error: None,
            next_retry_at: None,
            created_at: now,
            delivered_at: None,
            actor_user_id: None,
        }
    }

    struct ErringDelivery;

    #[async_trait]
    impl OutboundDelivery for ErringDelivery {
        async fn deliver(&self, _item: &fediverse_outbound_queue::Model) -> Result<(), AppError> {
            Err(AppError::Internal("simulated transport failure".into()))
        }
    }

    #[test]
    fn outbound_status_round_trips_through_string() {
        for s in [
            OutboundStatus::Pending,
            OutboundStatus::Delivered,
            OutboundStatus::Failed,
            OutboundStatus::Dead,
        ] {
            assert_eq!(OutboundStatus::parse(s.as_str()), s);
        }
        // Unknown wire value falls back safely.
        assert_eq!(OutboundStatus::parse("garbage"), OutboundStatus::Pending);
    }

    #[test]
    fn enqueue_request_clone_works() {
        let r = EnqueueRequest {
            activity_id: "act".into(),
            payload_json: "{}".into(),
            target_inbox: "ib".into(),
            target_domain: "d".into(),
            actor_user_id: None,
        };
        let c = r.clone();
        assert_eq!(c.activity_id, "act");
    }

    #[tokio::test]
    async fn enqueue_persists_pending_row() {
        let id = Uuid::new_v4();
        let row = pending_row(id);
        let db: DatabaseConnection = MockDatabase::new(DatabaseBackend::Postgres)
            .append_query_results([vec![row.clone()]])
            .into_connection();

        let inserted = FediverseOutboundService::enqueue(
            &db,
            EnqueueRequest {
                activity_id: row.activity_id.clone(),
                payload_json: row.payload_json.clone(),
                target_inbox: row.target_inbox.clone(),
                target_domain: row.target_domain.clone(),
                actor_user_id: None,
            },
        )
        .await
        .expect("enqueue ok");
        assert_eq!(inserted.status, "pending");
        assert_eq!(inserted.attempts, 0);
    }

    #[tokio::test]
    async fn enqueue_uses_provided_activity_id() {
        let id = Uuid::new_v4();
        let mut row = pending_row(id);
        row.activity_id = "https://example.test/specific-id".into();
        let db: DatabaseConnection = MockDatabase::new(DatabaseBackend::Postgres)
            .append_query_results([vec![row.clone()]])
            .into_connection();

        let inserted = FediverseOutboundService::enqueue(
            &db,
            EnqueueRequest {
                activity_id: row.activity_id.clone(),
                payload_json: "{}".into(),
                target_inbox: row.target_inbox.clone(),
                target_domain: row.target_domain.clone(),
                actor_user_id: None,
            },
        )
        .await
        .unwrap();
        assert_eq!(inserted.activity_id, "https://example.test/specific-id");
    }

    #[tokio::test]
    async fn drain_one_returns_none_when_empty() {
        let db: DatabaseConnection = MockDatabase::new(DatabaseBackend::Postgres)
            .append_query_results([Vec::<fediverse_outbound_queue::Model>::new()])
            .into_connection();
        let result = FediverseOutboundService::drain_one(&db, &NoopDelivery)
            .await
            .unwrap();
        assert!(result.is_none());
    }

    #[tokio::test]
    async fn drain_one_marks_delivered_on_success() {
        let id = Uuid::new_v4();
        let row = pending_row(id);
        let mut after = row.clone();
        after.status = "delivered".into();
        after.attempts = 1;
        after.delivered_at = Some(Utc::now());
        let db: DatabaseConnection = MockDatabase::new(DatabaseBackend::Postgres)
            // SELECT for candidate
            .append_query_results([vec![row.clone()]])
            // Row returned by .update()
            .append_query_results([vec![after]])
            .append_exec_results([MockExecResult {
                last_insert_id: 0,
                rows_affected: 1,
            }])
            .into_connection();

        let updated = FediverseOutboundService::drain_one(&db, &NoopDelivery)
            .await
            .unwrap()
            .expect("row processed");
        assert_eq!(updated.status, "delivered");
        assert_eq!(updated.attempts, 1);
        assert!(updated.last_error.is_none());
    }

    #[tokio::test]
    async fn drain_one_marks_failed_pending_with_next_retry_at_when_below_max_attempts() {
        // RFC 0010: a single failure on a row with attempts < max
        // keeps the row `pending` but gated on `next_retry_at`.
        let id = Uuid::new_v4();
        let row = pending_row(id);
        let mut after = row.clone();
        after.status = "pending".into();
        after.attempts = 1;
        after.last_error = Some("simulated transport failure".into());
        after.next_retry_at = Some(Utc::now() + chrono::Duration::minutes(5));
        let db: DatabaseConnection = MockDatabase::new(DatabaseBackend::Postgres)
            .append_query_results([vec![row.clone()]])
            .append_query_results([vec![after]])
            .append_exec_results([MockExecResult {
                last_insert_id: 0,
                rows_affected: 1,
            }])
            .into_connection();

        let updated = FediverseOutboundService::drain_one(&db, &ErringDelivery)
            .await
            .unwrap()
            .expect("row processed");
        assert_eq!(updated.status, "pending");
        assert!(updated.last_error.is_some());
        assert!(updated.next_retry_at.is_some());
    }

    #[tokio::test]
    async fn drain_one_marks_dead_when_attempts_reach_max() {
        let id = Uuid::new_v4();
        let mut row = pending_row(id);
        row.attempts = 7;
        row.max_attempts = 8;
        let mut after = row.clone();
        after.status = "dead".into();
        after.attempts = 8;
        after.last_error = Some("simulated transport failure".into());
        let db: DatabaseConnection = MockDatabase::new(DatabaseBackend::Postgres)
            .append_query_results([vec![row]])
            .append_query_results([vec![after]])
            .append_exec_results([MockExecResult {
                last_insert_id: 0,
                rows_affected: 1,
            }])
            .into_connection();

        let updated = FediverseOutboundService::drain_one(&db, &ErringDelivery)
            .await
            .unwrap()
            .unwrap();
        assert_eq!(updated.status, "dead");
        assert!(updated.next_retry_at.is_none());
    }

    #[test]
    fn next_retry_offset_grows_then_caps() {
        let secs: Vec<u64> = (1..=8)
            .map(|i| OutboundStatus::next_retry_offset(i).as_secs())
            .collect();
        // Strictly increasing for the first 6 entries.
        assert!(secs[0] < secs[1]);
        assert!(secs[1] < secs[2]);
        assert!(secs[2] < secs[3]);
        assert!(secs[3] < secs[4]);
        assert!(secs[4] < secs[5]);
        // Cap at 48h afterwards.
        assert_eq!(secs[5], 48 * 3600);
        assert_eq!(secs[6], 48 * 3600);
        assert_eq!(secs[7], 48 * 3600);
    }

    #[tokio::test]
    async fn drain_one_increments_attempts() {
        let id = Uuid::new_v4();
        let mut row = pending_row(id);
        row.attempts = 3;
        let mut after = row.clone();
        after.status = "delivered".into();
        after.attempts = 4;
        after.delivered_at = Some(Utc::now());
        let db: DatabaseConnection = MockDatabase::new(DatabaseBackend::Postgres)
            .append_query_results([vec![row.clone()]])
            .append_query_results([vec![after]])
            .append_exec_results([MockExecResult {
                last_insert_id: 0,
                rows_affected: 1,
            }])
            .into_connection();

        let updated = FediverseOutboundService::drain_one(&db, &NoopDelivery)
            .await
            .unwrap()
            .unwrap();
        assert_eq!(updated.attempts, 4);
    }

    #[tokio::test]
    async fn noop_delivery_returns_ok() {
        let row = pending_row(Uuid::new_v4());
        assert!(NoopDelivery.deliver(&row).await.is_ok());
    }

    #[tokio::test]
    async fn worker_can_be_built_and_dropped() {
        // Smoke test : building the struct and dropping it before the
        // loop runs must not panic.
        let db: DatabaseConnection = MockDatabase::new(DatabaseBackend::Postgres).into_connection();
        let _worker = OutboundWorker::new(db, NoopDelivery, Duration::from_millis(10));
        // Drop happens at end of scope.
    }
}