eric / rustwarden

//! In-memory cache plus durable write-through [`GuardStore`].
//!
//! The hot path (`is_banned`, `acl_for_domain`, `active_bans_for_domain`)
//! reads from the in-process `DashMap`-backed [`MemoryStore`] cache
//! (O(1), no I/O). Writes propagate to the durable backend in a
//! detached Tokio task — fire-and-forget — so the federation hot path
//! is never blocked by Postgres latency.
//!
//! On boot, call [`HybridStore::rehydrate`] to repopulate the cache
//! from the durable backend (active bans + every ACL row).
//!
//! Events are **not** cached: the analytical
//! `count_events_since` query is delegated straight to the durable
//! backend because event volume can be high and they are read only
//! by the detector chain (already off the request path).

use std::sync::Arc;

use async_trait::async_trait;
use chrono::{DateTime, Utc};
use rustwarden_core::error::AppError;
use uuid::Uuid;

use super::{AclEntry, BanEntry, GuardStore, MemoryStore};
use crate::events::{GuardEvent, GuardEventKind};

/// Composite store: `MemoryStore` cache on top, durable backend below.
pub struct HybridStore {
    cache: MemoryStore,
    durable: Arc<dyn GuardStore>,
}

impl HybridStore {
    /// Build a new hybrid store. Call [`Self::rehydrate`] right after
    /// to populate the cache with the current state of `durable`.
    #[must_use]
    pub fn new(durable: Arc<dyn GuardStore>) -> Self {
        Self {
            cache: MemoryStore::new(),
            durable,
        }
    }

    /// Borrow the durable backend (useful for tests + analytics).
    #[must_use]
    pub fn durable(&self) -> &Arc<dyn GuardStore> {
        &self.durable
    }

    /// Populate the cache from the durable backend. Should be called
    /// once at app boot, before serving requests.
    ///
    /// # Errors
    ///
    /// Propagates errors from the durable backend.
    pub async fn rehydrate(&self) -> Result<(usize, usize), AppError> {
        let now = Utc::now();
        let bans = self.durable.all_active_bans(now).await?;
        let bans_count = bans.len();
        for b in bans {
            self.cache.insert_ban(&b).await?;
        }
        let acls = self.durable.list_acl().await?;
        let acl_count = acls.len();
        for a in acls {
            self.cache.upsert_acl(&a).await?;
        }
        tracing::info!(
            bans = bans_count,
            acl = acl_count,
            "fediverse guard hybrid store rehydrated"
        );
        Ok((bans_count, acl_count))
    }
}

/// Spawn a fire-and-forget write to the durable backend. The
/// federation hot path must never await persistence.
fn spawn_durable<F>(future: F)
where
    F: std::future::Future<Output = Result<(), AppError>> + Send + 'static,
{
    tokio::spawn(async move {
        if let Err(e) = future.await {
            tracing::warn!(error = %e, "fediverse guard durable write failed");
        }
    });
}

#[async_trait]
impl GuardStore for HybridStore {
    async fn insert_event(&self, event: &GuardEvent) -> Result<(), AppError> {
        // Events are not cached. Delegate to durable directly so the
        // count_events_since query reflects this insert; spawn the
        // actual I/O so callers don't block on Postgres latency.
        let durable = self.durable.clone();
        let event = event.clone();
        spawn_durable(async move { durable.insert_event(&event).await });
        Ok(())
    }

    async fn count_events_since(
        &self,
        domain: &str,
        kind: GuardEventKind,
        since: DateTime<Utc>,
    ) -> Result<u64, AppError> {
        // Analytical query — go to durable.
        self.durable.count_events_since(domain, kind, since).await
    }

    async fn insert_ban(&self, ban: &BanEntry) -> Result<(), AppError> {
        self.cache.insert_ban(ban).await?;
        let durable = self.durable.clone();
        let ban = ban.clone();
        spawn_durable(async move { durable.insert_ban(&ban).await });
        Ok(())
    }

    async fn active_bans_for_domain(
        &self,
        domain: &str,
        now: DateTime<Utc>,
    ) -> Result<Vec<BanEntry>, AppError> {
        self.cache.active_bans_for_domain(domain, now).await
    }

    async fn all_active_bans(&self, now: DateTime<Utc>) -> Result<Vec<BanEntry>, AppError> {
        self.cache.all_active_bans(now).await
    }

    async fn lift_ban(
        &self,
        ban_id: Uuid,
        by: Option<Uuid>,
        at: DateTime<Utc>,
    ) -> Result<(), AppError> {
        self.cache.lift_ban(ban_id, by, at).await?;
        let durable = self.durable.clone();
        spawn_durable(async move { durable.lift_ban(ban_id, by, at).await });
        Ok(())
    }

    async fn upsert_acl(&self, entry: &AclEntry) -> Result<(), AppError> {
        self.cache.upsert_acl(entry).await?;
        let durable = self.durable.clone();
        let entry = entry.clone();
        spawn_durable(async move { durable.upsert_acl(&entry).await });
        Ok(())
    }

    async fn acl_for_domain(&self, domain: &str) -> Result<Option<AclEntry>, AppError> {
        self.cache.acl_for_domain(domain).await
    }

    async fn delete_acl(&self, id: Uuid) -> Result<(), AppError> {
        self.cache.delete_acl(id).await?;
        let durable = self.durable.clone();
        spawn_durable(async move { durable.delete_acl(id).await });
        Ok(())
    }

    async fn list_acl(&self) -> Result<Vec<AclEntry>, AppError> {
        self.cache.list_acl().await
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::events::{GuardEvent, GuardEventKind, Severity};
    use crate::store::{AclKind, MemoryStore};

    fn durable_with_active_ban() -> Arc<dyn GuardStore> {
        Arc::new(MemoryStore::new())
    }

    #[tokio::test]
    async fn rehydrate_loads_active_bans_and_acl_from_durable() {
        let durable = Arc::new(MemoryStore::new());
        let now = Utc::now();

        // Seed durable with one active ban + one expired (should not
        // be cached) + one ACL entry.
        let active = BanEntry {
            id: Uuid::new_v4(),
            domain: "evil.tld".into(),
            actor_uri: None,
            reason: "spam".into(),
            severity: "suspend".into(),
            triggered_by: "auto:test".into(),
            banned_at: now,
            expires_at: Some(now + chrono::Duration::hours(1)),
            lifted_at: None,
            lifted_by: None,
        };
        let expired = BanEntry {
            id: Uuid::new_v4(),
            domain: "stale.tld".into(),
            actor_uri: None,
            reason: "old".into(),
            severity: "suspend".into(),
            triggered_by: "auto:test".into(),
            banned_at: now - chrono::Duration::hours(3),
            expires_at: Some(now - chrono::Duration::hours(1)),
            lifted_at: None,
            lifted_by: None,
        };
        durable.insert_ban(&active).await.unwrap();
        durable.insert_ban(&expired).await.unwrap();
        durable
            .upsert_acl(&AclEntry {
                id: Uuid::new_v4(),
                kind: AclKind::Allow,
                domain: "peer.tld".into(),
                reason: None,
                created_by: Uuid::new_v4(),
                created_at: now,
                updated_at: now,
            })
            .await
            .unwrap();

        let hybrid = HybridStore::new(durable.clone() as Arc<dyn GuardStore>);
        let (bans_n, acl_n) = hybrid.rehydrate().await.unwrap();
        assert_eq!(bans_n, 1, "only the active ban must be cached");
        assert_eq!(acl_n, 1);

        // Cache should answer the hot path without hitting durable.
        let active_cached = hybrid
            .active_bans_for_domain("evil.tld", now)
            .await
            .unwrap();
        assert_eq!(active_cached.len(), 1);
        let acl_cached = hybrid.acl_for_domain("peer.tld").await.unwrap();
        assert!(acl_cached.is_some());
    }

    #[tokio::test]
    async fn writes_reach_durable_after_spawn_drains() {
        let durable: Arc<dyn GuardStore> = durable_with_active_ban();
        let hybrid = HybridStore::new(durable.clone());

        let now = Utc::now();
        let ban = BanEntry {
            id: Uuid::new_v4(),
            domain: "x.tld".into(),
            actor_uri: None,
            reason: "test".into(),
            severity: "suspend".into(),
            triggered_by: "auto:t".into(),
            banned_at: now,
            expires_at: Some(now + chrono::Duration::hours(1)),
            lifted_at: None,
            lifted_by: None,
        };
        hybrid.insert_ban(&ban).await.unwrap();

        // Cache is immediate.
        let cached = hybrid.active_bans_for_domain("x.tld", now).await.unwrap();
        assert_eq!(cached.len(), 1);

        // The detached task needs a tick to land in durable.
        tokio::task::yield_now().await;
        // Allow the spawned task to run.
        for _ in 0..5 {
            let durable_view = durable.active_bans_for_domain("x.tld", now).await.unwrap();
            if !durable_view.is_empty() {
                return;
            }
            tokio::time::sleep(std::time::Duration::from_millis(5)).await;
        }
        panic!("durable write never landed");
    }

    #[tokio::test]
    async fn count_events_since_delegates_to_durable() {
        let durable: Arc<dyn GuardStore> = Arc::new(MemoryStore::new());
        // Seed event directly in durable (HybridStore would have spawned).
        durable
            .insert_event(&GuardEvent::new(
                GuardEventKind::SignatureFailure,
                "x.tld",
                Severity::Medium,
            ))
            .await
            .unwrap();

        let hybrid = HybridStore::new(durable);
        let cutoff = Utc::now() - chrono::Duration::minutes(1);
        let n = hybrid
            .count_events_since("x.tld", GuardEventKind::SignatureFailure, cutoff)
            .await
            .unwrap();
        assert_eq!(n, 1);
    }
}