scatter-proxy 0.8.0

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

use tracing::info;

use crate::classifier::BodyClassifier;
use crate::config::ScatterProxyConfig;
use crate::error::ScatterProxyError;
use crate::metrics::PoolMetrics;
use crate::task::TaskHandle;
use crate::ScatterProxy;

/// Routes proxied requests to per-host [`ScatterProxy`] instances.
///
/// Each registered host gets its own independent proxy pool, health tracker,
/// and scheduler.  Isolation means:
///
/// - A struggling host (all proxies in cooldown, low success rate) cannot
///   starve tasks for other hosts.
/// - Proxy eviction is scoped per-host: a proxy that repeatedly fails for
///   host A is marked dead only in A's pool, while remaining available to B.
/// - [`metrics_for`] gives a clean, host-scoped view of throughput and health.
///
/// Each host is configured independently via its own [`ScatterProxyConfig`],
/// so you can tune sources, concurrency, timeouts, and logging per target.
///
/// # Example
///
/// ```no_run
/// use scatter_proxy::{ScatterProxyRouter, ScatterProxyConfig, DefaultClassifier};
///
/// # async fn example() -> Result<(), scatter_proxy::ScatterProxyError> {
/// let router = ScatterProxyRouter::new(
///     [
///         ("szse.cn",     ScatterProxyConfig { max_inflight: 20, ..Default::default() }),
///         ("sse.com.cn",  ScatterProxyConfig { max_inflight: 10, ..Default::default() }),
///         ("cninfo.com.cn", ScatterProxyConfig::default()),
///     ],
///     DefaultClassifier,
/// ).await?;
///
/// let client = reqwest::Client::new();
/// let req = client.get("http://szse.cn/api/data").build().unwrap();
/// let handle = router.submit(req).await?;
/// // handle.await or handle.with_timeout(…) as usual
///
/// // Per-host observability
/// if let Some(m) = router.metrics_for("szse.cn") {
///     println!("szse success rate: {:.0}%", m.success_rate_1m * 100.0);
/// }
///
/// router.shutdown().await;
/// # Ok(())
/// # }
/// ```
///
/// [`metrics_for`]: ScatterProxyRouter::metrics_for
pub struct ScatterProxyRouter {
    routes: HashMap<String, ScatterProxy>,
}

impl ScatterProxyRouter {
    /// Build a router with one independent [`ScatterProxy`] pool per host.
    ///
    /// `routes` is an iterable of `(hostname, config)` pairs.  Each host gets
    /// its own pool configured independently.  If `config.name` is not set it
    /// defaults to the hostname so that metrics logs are prefixed correctly.
    ///
    /// # Errors
    ///
    /// Returns [`ScatterProxyError::Init`] if any host pool fails to initialise
    /// (e.g. the proxy source URL is unreachable and the list is empty).
    pub async fn new(
        routes: impl IntoIterator<Item = (impl Into<String>, ScatterProxyConfig)>,
        classifier: impl BodyClassifier,
    ) -> Result<Self, ScatterProxyError> {
        let classifier: Arc<dyn BodyClassifier> = Arc::new(classifier);
        let mut map = HashMap::new();

        for (host, mut config) in routes {
            let host = host.into();
            if config.name.is_none() {
                config.name = Some(host.clone());
            }
            let sp = ScatterProxy::new_arc(config, Arc::clone(&classifier)).await?;
            map.insert(host, sp);
        }

        info!(hosts = map.len(), "ScatterProxyRouter initialised");
        Ok(Self { routes: map })
    }

    /// Submit a request to the pool registered for its host.
    ///
    /// Blocks until the pool has capacity, then returns a [`TaskHandle`].
    ///
    /// # Errors
    ///
    /// Returns [`ScatterProxyError::UnknownHost`] if the request's host is not
    /// registered in this router.
    pub async fn submit(&self, request: reqwest::Request) -> Result<TaskHandle, ScatterProxyError> {
        let sp = self.route(&request)?;
        Ok(sp.submit(request).await)
    }

    /// Non-blocking submit variant.
    ///
    /// # Errors
    ///
    /// Returns [`ScatterProxyError::UnknownHost`] if the host is not registered,
    /// or [`ScatterProxyError::PoolFull`] if the host's pool is at capacity.
    pub fn try_submit(&self, request: reqwest::Request) -> Result<TaskHandle, ScatterProxyError> {
        let sp = self.route(&request)?;
        sp.try_submit(request)
    }

    /// Returns a [`PoolMetrics`] snapshot for the given host.
    ///
    /// Returns `None` if the host is not registered in this router.
    pub fn metrics_for(&self, host: &str) -> Option<PoolMetrics> {
        self.routes.get(host).map(|sp| sp.metrics())
    }

    /// Returns [`PoolMetrics`] snapshots for every registered host.
    pub fn all_metrics(&self) -> HashMap<String, PoolMetrics> {
        self.routes
            .iter()
            .map(|(host, sp)| (host.clone(), sp.metrics()))
            .collect()
    }

    /// Returns the list of registered hostnames.
    pub fn hosts(&self) -> Vec<&str> {
        self.routes.keys().map(String::as_str).collect()
    }

    /// Gracefully shut down all host pools (persists state for each if configured).
    pub async fn shutdown(self) {
        for (_, sp) in self.routes {
            sp.shutdown().await;
        }
    }

    fn route(&self, request: &reqwest::Request) -> Result<&ScatterProxy, ScatterProxyError> {
        let host = request.url().host_str().unwrap_or("").to_string();
        self.routes
            .get(&host)
            .ok_or(ScatterProxyError::UnknownHost(host))
    }
}