arcly-http 0.2.0

//! Cluster-wide rate limiting behind a pluggable async backend.
//!
//! The local [`RateLimit`](crate::resilience::RateLimit) is per-instance: on a
//! 10-replica deployment a "30 req/min" limit silently becomes 300 req/min and
//! resets on every pod restart. `DistributedRateLimit` delegates the counting
//! to a shared backend (Redis sliding window in the enterprise example) so the
//! limit holds across the whole fleet.
//!
//! ## Design rules
//!
//! - **Async by nature** — a network counter cannot hide behind the sync
//!   [`Guard`](crate::auth::guards::Guard) trait, so this type exposes
//!   `async fn check(&ctx)`; handlers `.await` it explicitly. No `block_on`,
//!   no executor stalls.
//! - **One atomic round-trip** — `RateLimitBackend::hit` must count and decide
//!   in a single atomic operation (Lua script / GETDEL-style). Split
//!   check-then-increment is a TOCTOU bug, the same class this codebase
//!   already eliminated from refresh-token rotation.
//! - **Explicit failure policy** — "backend down" is its own decision
//!   ([`RateDecision::Unavailable`]); whether that allows or denies is the
//!   caller's declared [`FailurePolicy`], never an accident of error mapping.
//! - **Optional by construction** — no backend in the DI container means the
//!   check is a no-op, so dev environments need zero configuration.

use std::net::IpAddr;

use futures::future::BoxFuture;

use crate::web::{Error, RequestContext};

/// Outcome of one counted hit against the shared limiter.
pub enum RateDecision {
    Allow {
        remaining: u32,
    },
    Deny {
        retry_after_secs: u32,
    },
    /// The backend did not answer — resolved by [`FailurePolicy`].
    Unavailable,
}

/// Shared counting backend (Redis, Memcached, …). Object-safe via `BoxFuture`,
/// following the same pattern as `SessionStore` and `OAuth2Provider`.
///
/// Implementations MUST count-and-decide atomically in a single round-trip.
pub trait RateLimitBackend: Send + Sync + 'static {
    fn hit<'a>(&'a self, key: &'a str, max: u32, window_secs: u32) -> BoxFuture<'a, RateDecision>;
}

/// What to do when the backend is unreachable.
#[derive(Clone, Copy, PartialEq, Eq)]
pub enum FailurePolicy {
    /// Let traffic through (availability over strictness) — sane default.
    FailOpen,
    /// Reject with `503` — for abuse-sensitive endpoints like `/auth/login`.
    FailClosed,
}

/// Cluster-wide limiter. Build as a `static` and call from handlers:
///
/// ```ignore
/// static LOGIN_RATE: DistributedRateLimit =
///     DistributedRateLimit::new("login", 10, 60).fail_closed();
///
/// // in the handler:
/// LOGIN_RATE.check(&ctx).await?;
/// ```
pub struct DistributedRateLimit {
    pub name: &'static str,
    pub max: u32,
    pub window_secs: u32,
    pub policy: FailurePolicy,
}

impl DistributedRateLimit {
    pub const fn new(name: &'static str, max: u32, window_secs: u32) -> Self {
        Self {
            name,
            max,
            window_secs,
            policy: FailurePolicy::FailOpen,
        }
    }

    /// Reject with `503` when the backend is down (instead of letting traffic through).
    pub const fn fail_closed(mut self) -> Self {
        self.policy = FailurePolicy::FailClosed;
        self
    }

    /// Rate-limit key for this request's principal.
    ///
    /// Authenticated identity (`sub` claim) wins over network address: it
    /// keeps users behind one NAT from sharing a bucket, and an attacker with
    /// a stolen account can't escape their per-user limit by rotating IPs.
    /// Anonymous traffic falls back to the first `X-Forwarded-For` hop.
    fn principal(ctx: &RequestContext) -> String {
        if let Some(sub) = ctx
            .claims()
            .and_then(|c| c.get("sub"))
            .and_then(|v| v.as_str())
        {
            return format!("sub:{sub}");
        }
        ctx.header("x-forwarded-for")
            .and_then(|h| h.split(',').next())
            .and_then(|s| s.trim().parse::<IpAddr>().ok())
            .map(|ip| format!("ip:{ip}"))
            .unwrap_or_else(|| "anon".to_owned())
    }

    pub async fn check(&self, ctx: &RequestContext) -> Result<(), Error> {
        // No backend registered → limiter disabled (dev mode). The local
        // `RateLimit` guard can still provide per-instance protection.
        let Some(backend) = ctx.try_inject::<Box<dyn RateLimitBackend>>() else {
            return Ok(());
        };

        let key = format!("rl:{}:{}", self.name, Self::principal(ctx));
        match backend.hit(&key, self.max, self.window_secs).await {
            RateDecision::Allow { .. } => Ok(()),
            RateDecision::Deny { .. } => Err(Error::TooManyRequests),
            RateDecision::Unavailable => match self.policy {
                FailurePolicy::FailOpen => Ok(()),
                FailurePolicy::FailClosed => {
                    Err(Error::ServiceUnavailable("rate limit backend unavailable"))
                }
            },
        }
    }
}