//! Per-host request pacing.
//!
//! `RateLimiter` is a single-token bucket: each `wait().await` returns
//! immediately when at least `interval` has elapsed since the last call,
//! otherwise sleeps just enough to satisfy it. Uses
//! `tokio::time::Instant` so tests can run under `start_paused` virtual
//! time without sleeping for real.
//!
//! `HostRateLimiters` is the multi-host wrapper actually used by the
//! crawler — concurrent workers issuing requests to different origins
//! (catalog vs. CDN) don't contend on a shared budget; each host gets
//! its own bucket. `wait_for(url)` extracts the host, lazily creates a
//! limiter for it, and serializes only against other callers hitting
//! the same host.

use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Mutex;
use tokio::time::Instant;

#[derive(Debug)]
pub struct RateLimiter {
    interval: Duration,
    last: Option<Instant>,
}

impl RateLimiter {
    pub fn new(interval: Duration) -> Self {
        Self {
            interval,
            last: None,
        }
    }

    pub async fn wait(&mut self) {
        if let Some(last) = self.last {
            let elapsed = last.elapsed();
            if elapsed < self.interval {
                tokio::time::sleep(self.interval - elapsed).await;
            }
        }
        self.last = Some(Instant::now());
    }
}

/// Once the per-host map grows past this many entries, the next `wait_for`
/// sweeps out hosts idle longer than [`IDLE_EVICT_AFTER`]. A long crawl that
/// touches thousands of distinct CDN shards would otherwise retain one bucket
/// per host for the daemon's whole lifetime. Far above any single source's
/// real host count, so the sweep is rare.
const MAX_TRACKED_HOSTS: usize = 1024;

/// A host bucket untouched for at least this long is evicted on the next
/// over-cap sweep. Generous: a host still being crawled is touched every
/// `interval`, so only genuinely-finished hosts age out.
const IDLE_EVICT_AFTER: Duration = Duration::from_secs(3600);

/// A host's bucket plus when it was last used, so idle entries can be evicted.
#[derive(Debug)]
struct HostEntry {
    limiter: Arc<Mutex<RateLimiter>>,
    last_used: Instant,
}

/// Per-host rate limiter map. The outer `Mutex<HashMap>` is held only
/// during the entry-or-insert + Arc clone; the per-host `Mutex<RateLimiter>`
/// is held during the actual `wait().await`. So N workers calling
/// `wait_for(url)` on N different hosts contend nowhere except the brief
/// HashMap lookup; workers hitting the same host serialize on that
/// host's bucket.
#[derive(Debug)]
pub struct HostRateLimiters {
    default_interval: Duration,
    overrides: HashMap<String, Duration>,
    map: Mutex<HashMap<String, HostEntry>>,
}

impl HostRateLimiters {
    pub fn new(default_interval: Duration) -> Self {
        Self {
            default_interval,
            overrides: HashMap::new(),
            map: Mutex::new(HashMap::new()),
        }
    }

    /// Set a per-host interval that overrides `default_interval`. Calls
    /// after a host's limiter has been instantiated do *not* re-create
    /// it — set all overrides before the first `wait_for` to that host.
    pub fn with_override(mut self, host: impl Into<String>, interval: Duration) -> Self {
        self.overrides.insert(host.into(), interval);
        self
    }

    /// Block until the per-host budget allows the next request to
    /// `url`'s host. Returns an error only when the URL has no host
    /// (malformed input).
    pub async fn wait_for(&self, url: &str) -> anyhow::Result<()> {
        let host = host_of(url)
            .ok_or_else(|| anyhow::anyhow!("no host in url: {url}"))?;
        let limiter = {
            let mut map = self.map.lock().await;
            let now = Instant::now();
            // Bound the map: when it grows past the soft cap, drop hosts that
            // have been idle past the TTL. Only fires over-cap, so the common
            // path is a plain lookup. A host still being crawled is touched
            // every `interval` and so never ages out.
            if map.len() >= MAX_TRACKED_HOSTS {
                map.retain(|_, e| now.duration_since(e.last_used) < IDLE_EVICT_AFTER);
            }
            let entry = map.entry(host.clone()).or_insert_with(|| {
                let interval = self
                    .overrides
                    .get(&host)
                    .copied()
                    .unwrap_or(self.default_interval);
                HostEntry {
                    limiter: Arc::new(Mutex::new(RateLimiter::new(interval))),
                    last_used: now,
                }
            });
            entry.last_used = now;
            entry.limiter.clone()
        };
        limiter.lock().await.wait().await;
        Ok(())
    }

    /// Number of host buckets currently tracked. Test/observability hook.
    #[cfg(test)]
    pub(crate) async fn tracked_hosts(&self) -> usize {
        self.map.lock().await.len()
    }
}

// `host_of` was duplicated across session/rate_limit/pipeline; the
// canonical version now lives in `crawler::url_utils`.
use crate::crawler::url_utils::host_of;

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

    #[tokio::test(start_paused = true)]
    async fn first_call_does_not_sleep() {
        let mut rl = RateLimiter::new(Duration::from_millis(100));
        let t0 = Instant::now();
        rl.wait().await;
        assert_eq!(Instant::now() - t0, Duration::ZERO);
    }

    #[tokio::test(start_paused = true)]
    async fn second_call_sleeps_to_fill_interval() {
        let mut rl = RateLimiter::new(Duration::from_millis(100));
        let t0 = Instant::now();
        rl.wait().await;
        rl.wait().await;
        // Second call had to wait the full 100ms after the (instant)
        // first call.
        assert_eq!(Instant::now() - t0, Duration::from_millis(100));
    }

    #[tokio::test(start_paused = true)]
    async fn no_sleep_if_interval_already_elapsed() {
        let mut rl = RateLimiter::new(Duration::from_millis(100));
        rl.wait().await;
        tokio::time::sleep(Duration::from_millis(250)).await;
        let t0 = Instant::now();
        rl.wait().await;
        // Already 250ms past — no further wait needed.
        assert_eq!(Instant::now() - t0, Duration::ZERO);
    }

    #[test]
    fn host_of_parses_scheme_path_and_port() {
        assert_eq!(host_of("https://Example.com/path").as_deref(), Some("example.com"));
        assert_eq!(host_of("http://cdn.foo.bar/img.jpg").as_deref(), Some("cdn.foo.bar"));
        assert_eq!(host_of("http://localhost:8080/x").as_deref(), Some("localhost"));
        assert!(host_of("not a url").is_none());
    }

    #[tokio::test(start_paused = true)]
    async fn host_rate_limiters_pace_per_host() {
        // Two hosts at 100ms each. Two consecutive calls to the SAME
        // host wait 100ms total. Two consecutive calls to DIFFERENT
        // hosts both fire immediately.
        let rl = HostRateLimiters::new(Duration::from_millis(100));

        let t0 = Instant::now();
        rl.wait_for("https://a.example/x").await.unwrap();
        rl.wait_for("https://b.example/y").await.unwrap();
        assert_eq!(Instant::now() - t0, Duration::ZERO, "different hosts don't contend");

        let t1 = Instant::now();
        rl.wait_for("https://a.example/x").await.unwrap();
        assert_eq!(
            Instant::now() - t1,
            Duration::from_millis(100),
            "second call to same host waits a full interval"
        );
    }

    #[tokio::test(start_paused = true)]
    async fn host_rate_limiters_evict_idle_hosts_over_cap() {
        // Fill the map to the soft cap with distinct hosts (first call to a
        // fresh host never sleeps, so this is fast even under real time).
        let rl = HostRateLimiters::new(Duration::from_millis(1));
        for i in 0..MAX_TRACKED_HOSTS {
            rl.wait_for(&format!("https://host{i}.example/x")).await.unwrap();
        }
        assert_eq!(rl.tracked_hosts().await, MAX_TRACKED_HOSTS);

        // Let every tracked host age past the idle TTL, then touch one new
        // host: the over-cap sweep should evict all the idle ones, leaving
        // only the freshly-inserted entry.
        tokio::time::sleep(IDLE_EVICT_AFTER + Duration::from_secs(1)).await;
        rl.wait_for("https://newcomer.example/y").await.unwrap();
        assert_eq!(
            rl.tracked_hosts().await,
            1,
            "idle hosts should be swept once the map is over the cap"
        );
    }

    #[tokio::test(start_paused = true)]
    async fn host_rate_limiters_keep_recently_used_hosts() {
        // A host touched within the TTL must survive a sweep so active crawls
        // aren't reset. Fill to the cap, then re-touch one host right before
        // adding a newcomer that triggers the sweep.
        let rl = HostRateLimiters::new(Duration::from_millis(1));
        for i in 0..MAX_TRACKED_HOSTS {
            rl.wait_for(&format!("https://host{i}.example/x")).await.unwrap();
        }
        tokio::time::sleep(IDLE_EVICT_AFTER + Duration::from_secs(1)).await;
        // Re-touch host0 so it's recent again.
        rl.wait_for("https://host0.example/x").await.unwrap();
        // Newcomer triggers the sweep (map is at cap+1 conceptually).
        rl.wait_for("https://newcomer.example/y").await.unwrap();
        // host0 (recent) + newcomer survive; the rest aged out.
        assert_eq!(rl.tracked_hosts().await, 2);
    }

    #[tokio::test(start_paused = true)]
    async fn host_rate_limiters_honor_overrides() {
        let rl = HostRateLimiters::new(Duration::from_millis(1000))
            .with_override("fast.example", Duration::from_millis(100));

        rl.wait_for("https://fast.example/a").await.unwrap();
        let t0 = Instant::now();
        rl.wait_for("https://fast.example/b").await.unwrap();
        assert_eq!(Instant::now() - t0, Duration::from_millis(100));
    }
}
