ethl 0.1.42

use std::time::Duration;

use alloy::{
    network::Ethereum,
    providers::{DynProvider, Provider, ProviderBuilder, WsConnect},
};
use anyhow::Result;
use reqwest::Url;
use tracing::warn;

use crate::rpc::RpcError;
use crate::rpc::provider::{Provider as RpcProvider, ProviderCredentials, ProviderUrlError};

/// Determines which block number is used as the capture target in `advance_to_latest`.
///
/// Default is `Latest` (no lag) — existing L2 consumers are unchanged.
/// Mainnet callers should opt in to `Finalized` for reorg-proof capture.
#[derive(Debug, Clone, Copy, PartialEq, Default)]
pub enum CaptureTarget {
    /// Ethereum `finalized` block (≈ 2 epochs ≈ 64 blocks ≈ 13 min behind tip).
    /// Cryptographically final — cannot reorg without 1/3 of staked ETH slashed.
    /// Opt-in; mainnet-only. L2 `finalized` tracks L1 finality and lags too far.
    Finalized,
    /// Fixed number of blocks behind the current chain tip. Saturates at 0.
    Lag(u64),
    /// Current chain tip. No lag; suitable for L2s with instant-finality sequencers.
    #[default]
    Latest,
}

pub const MAX_RETRIES: u32 = 5;
pub const BASE_BACKOFF_SECS: u64 = 2;
pub const DEFAULT_HTTP_CONNECT_TIMEOUT_SECS: u64 = 5;
pub const DEFAULT_HTTP_REQUEST_TIMEOUT_SECS: u64 = 15;

#[derive(Debug, Clone)]
pub struct HttpRpcSettings {
    pub url: Url,
    /// Upper bound on in-flight requests dispatched to this provider at once
    /// (e.g. by `flush_window_range`'s concurrent by-hash log fetch). Read as
    /// `u32` here for config ergonomics; `fetch_concurrently` clamps it to at
    /// least 1 before using it to size its dispatch pool. Not every consumer of
    /// this field applies that clamp — check the specific call site.
    pub max_concurrency: u32,
    pub max_batch_size: u32,
    pub init_batch_size: u32,
    pub max_logs_per_request: u32,
}

impl HttpRpcSettings {
    pub fn new(url: String, max_concurrency: u32, max_batch_size: u32) -> Self {
        Self {
            url: url.parse().expect("Invalid URL"),
            max_concurrency,
            max_batch_size,
            max_logs_per_request: 2_500,
            init_batch_size: 500,
        }
    }

    pub fn host(&self) -> String {
        self.url.host_str().unwrap_or("unknown").to_string()
    }
}

#[derive(Debug, Clone)]
pub struct ProviderSettings {
    pub http_providers: Vec<HttpRpcSettings>,
    pub wss_endpoints: Vec<Url>,
    pub max_retries: u32,
    pub base_backoff_secs: u64,
    pub http_connect_timeout_secs: u64,
    pub http_request_timeout_secs: u64,
    /// Block to target when `advance_to_latest` is called. Default `Latest` (no lag).
    /// Set to `Finalized` for mainnet to avoid reorgs at the capture boundary.
    pub capture_target: CaptureTarget,
    /// Depth of the live-tip reorg detector's `(number, hash)` ring, in blocks.
    /// `0` disables detection (default) — every L2 and the snapshot path stay
    /// untouched. A mainnet live (`Latest`) consumer sets this to finality + margin
    /// (≈ 128); the walk-back can resolve a common ancestor up to this many blocks
    /// back, and a divergence older than the ring is reported as beyond finality.
    /// Only `stream_heads_with_logs` reads it; the capture path never does.
    pub reorg_detect_depth: u64,
    /// Seconds to wait between tip re-polls when a batch is past the provider tip.
    /// Default 60s (5 × 60 ≈ 5 min to stall alarm). Set to 0 in `from_mock` for fast tests.
    pub tip_wait_backoff_secs: u64,
    /// Number of consecutive flat-tip polling cycles before yielding `ProviderStalled`.
    pub tip_wait_max_stall_cycles: u32,
    /// Maximum wall-clock seconds a single tip-wait episode may run before yielding
    /// `ProviderStalled`, even if the tip is advancing. An "episode" is one entry into a
    /// `'tip_wait*` loop for a given `to_block`; the timer resets on each re-entry.
    ///
    /// Bounds the advancing-but-not-converging case only. A flat tip is caught by
    /// `tip_wait_max_stall_cycles` above, far sooner, so this is not the halt protection.
    /// Blunt by construction: a provider that would have converged just past the bound is
    /// terminated too. One global default for every chain; override per-request with
    /// `with_tip_wait_max_total_secs`. Set to `0` to disable (unbounded wait, old behavior).
    pub tip_wait_max_total_secs: u64,
    /// Maximum wall-clock seconds one coalesced flush's aggregate RPC work
    /// (`flush_window_range`'s concurrent, bounded per-block by-hash log
    /// fetches) may run before it's treated as failed and retried on the next
    /// provider.
    ///
    /// Only meaningful when `reorg_detect_depth > 0` (the by-hash flush path) — the
    /// snapshot/backfill path never calls `flush_window_range`. Exists because the
    /// live head-consuming loop does not poll new heads while a flush is in
    /// progress; a pathologically slow flush lets alloy's own WS subscription
    /// buffer heads internally (default capacity 16, silently drops the oldest on
    /// overflow) for the whole blocked duration. Bounding the flush turns an
    /// unbounded internal-overflow risk into a bounded, retried failure instead.
    ///
    /// Default `0` (disabled — old, unbounded behavior) so this is a no-op for
    /// every chain until a caller opts in with a real value. A reorg-exposed,
    /// sub-second-block chain should set this to a fraction of
    /// `16 × block_time_secs` (alloy's buffer capacity × block time) with real
    /// margin, since at steady cadence a flush is only 1-2 calls and nowhere near
    /// that budget — only an abnormally slow flush should ever approach it.
    pub flush_max_total_secs: u64,
    /// Milliseconds the live stream waits after the first observed head before
    /// flushing the pending window with one `eth_getLogs`. Purely a
    /// cost-vs-freshness knob: providers serve logs by the time `newHeads`
    /// arrives (measured ~0ms index lag), so raising this trades log latency for
    /// fewer requests. Default 30_000 (cost-oriented consumers); latency-bound
    /// consumers set it low (0 disables the delay — flush on every head).
    pub live_head_coalesce_window_ms: u64,
    // When set, connect_http() returns this provider instead of building from url.
    // Used by hermetic tests to inject a mock transport.
    injected_http: Option<DynProvider<Ethereum>>,
}

impl Default for ProviderSettings {
    fn default() -> Self {
        Self {
            http_providers: vec![],
            wss_endpoints: vec![],
            max_retries: MAX_RETRIES,
            base_backoff_secs: BASE_BACKOFF_SECS,
            http_connect_timeout_secs: DEFAULT_HTTP_CONNECT_TIMEOUT_SECS,
            http_request_timeout_secs: DEFAULT_HTTP_REQUEST_TIMEOUT_SECS,
            capture_target: CaptureTarget::Latest,
            reorg_detect_depth: 0,
            tip_wait_backoff_secs: 60,
            tip_wait_max_stall_cycles: 5,
            tip_wait_max_total_secs: 1800,
            flush_max_total_secs: 0,
            live_head_coalesce_window_ms: 30_000,
            injected_http: None,
        }
    }
}

#[derive(Debug, Clone, Default)]
pub struct ProviderOptions {
    pub ankr_api_key: Option<String>,
    pub infura_api_key: Option<String>,
    pub drpc_api_key: Option<String>,
    /// Full endpoints keyed by the chain each endpoint serves.
    pub quicknode_endpoints: Vec<(u64, String)>,
    pub alchemy_api_key: Option<String>,
    pub rpc_urls: Option<Vec<String>>,
    pub ws_urls: Option<Vec<String>>,
    /// Ordered HTTP provider preferences. Unlisted configured providers remain fallbacks.
    pub http_provider_preferences: Option<Vec<RpcProvider>>,
    /// Ordered WebSocket provider preferences. Unlisted configured providers remain fallbacks.
    pub websocket_provider_preferences: Option<Vec<RpcProvider>>,
}

impl ProviderOptions {
    pub fn ankr_api_key(mut self, key: impl AsRef<str>) -> Self {
        self.ankr_api_key = Some(key.as_ref().to_string());
        self
    }

    pub fn infura_api_key(mut self, key: impl AsRef<str>) -> Self {
        self.infura_api_key = Some(key.as_ref().to_string());
        self
    }

    pub fn drpc_api_key(mut self, key: impl AsRef<str>) -> Self {
        self.drpc_api_key = Some(key.as_ref().to_string());
        self
    }

    pub fn quicknode_endpoint(mut self, chain_id: u64, endpoint: impl AsRef<str>) -> Self {
        self.quicknode_endpoints
            .push((chain_id, endpoint.as_ref().to_string()));
        self
    }

    pub fn alchemy_api_key(mut self, key: impl AsRef<str>) -> Self {
        self.alchemy_api_key = Some(key.as_ref().to_string());
        self
    }

    pub fn http_provider_preferences(
        mut self,
        providers: impl IntoIterator<Item = RpcProvider>,
    ) -> Self {
        self.http_provider_preferences = Some(providers.into_iter().collect());
        self
    }

    pub fn websocket_provider_preferences(
        mut self,
        providers: impl IntoIterator<Item = RpcProvider>,
    ) -> Self {
        self.websocket_provider_preferences = Some(providers.into_iter().collect());
        self
    }

    pub fn add_http(mut self, url: String) -> Self {
        if self.rpc_urls.is_none() {
            self.rpc_urls = Some(vec![]);
        }
        self.rpc_urls.as_mut().unwrap().push(url);
        self
    }

    pub fn add_websocket(mut self, url: String) -> Self {
        if self.ws_urls.is_none() {
            self.ws_urls = Some(vec![]);
        }
        self.ws_urls.as_mut().unwrap().push(url);
        self
    }
}

/// Pushes `provider`'s HTTP + WS endpoints onto `settings` if it serves `chain_id`.
/// Returns the provider's name if it was skipped (chain not in its table) so the
/// caller can report which configured providers didn't cover the chain.
#[derive(Debug, Clone)]
struct ConfiguredProvider<T> {
    provider: Option<RpcProvider>,
    endpoint: T,
}

fn push_provider_if_supported(
    http_providers: &mut Vec<ConfiguredProvider<HttpRpcSettings>>,
    ws_providers: &mut Vec<ConfiguredProvider<Url>>,
    provider: RpcProvider,
    credentials: ProviderCredentials<'_>,
    chain_id: u64,
) -> Result<Option<&'static str>> {
    match provider.http_url(credentials, chain_id) {
        Ok(http_url) => {
            let ws_url = provider.ws_url(credentials, chain_id)?;
            http_providers.push(ConfiguredProvider {
                provider: Some(provider),
                endpoint: provider.rpc_settings(http_url.as_str()),
            });
            ws_providers.push(ConfiguredProvider {
                provider: Some(provider),
                endpoint: ws_url,
            });
            Ok(None)
        }
        // Exhaustive on purpose: a future ProviderUrlError variant must be handled
        // explicitly here (compiler-enforced), not silently swallowed as a skip.
        Err(
            ProviderUrlError::UnsupportedChain { provider: name, .. }
            | ProviderUrlError::EndpointNotConfigured { provider: name, .. },
        ) => Ok(Some(name)),
        Err(error) => Err(error.into()),
    }
}

fn select_providers<T: Clone>(
    providers: Vec<ConfiguredProvider<T>>,
    preferences: Option<&[RpcProvider]>,
    transport: &str,
) -> Result<Vec<ConfiguredProvider<T>>> {
    let Some(preferences) = preferences else {
        return Ok(providers);
    };

    if preferences.is_empty() {
        return Ok(providers);
    }

    for (index, provider) in preferences.iter().enumerate() {
        if preferences[..index].contains(provider) {
            anyhow::bail!(
                "{transport} provider preferences contain duplicate provider {}",
                provider.name()
            );
        }
    }

    let mut selected = Vec::with_capacity(providers.len());
    for provider in preferences {
        selected.extend(
            providers
                .iter()
                .filter(|configured| configured.provider == Some(*provider))
                .cloned(),
        );
    }

    selected.extend(providers.into_iter().filter(|configured| {
        configured
            .provider
            .is_none_or(|provider| !preferences.contains(&provider))
    }));

    Ok(selected)
}

fn missing_provider_error(chain_id: u64, kind: &str, skipped: &[&'static str]) -> anyhow::Error {
    if skipped.is_empty() {
        anyhow::anyhow!("At least one {kind} provider must be configured via API key or custom URL")
    } else {
        anyhow::anyhow!(
            "chain {chain_id} is not served by any configured provider (configured: {}); \
             add a provider that supports it",
            skipped.join(", ")
        )
    }
}

impl ProviderSettings {
    pub fn http(rpc_url: impl AsRef<str>) -> Result<Self> {
        let url = rpc_url.as_ref();
        let provider = RpcProvider::detect(url).map_err(|_| {
            anyhow::anyhow!("Unknown RPC provider for URL {url}. Provide explicit RPC settings.")
        })?;
        let http_setting = provider.rpc_settings(url);
        Ok(Self {
            http_providers: vec![http_setting],
            ..Self::default()
        })
    }

    pub fn with_max_retries(mut self, retries: u32) -> Self {
        self.max_retries = retries;
        self
    }

    pub fn with_base_backoff_secs(mut self, secs: u64) -> Self {
        self.base_backoff_secs = secs;
        self
    }

    pub fn with_http_connect_timeout_secs(mut self, secs: u64) -> Self {
        self.http_connect_timeout_secs = secs;
        self
    }

    pub fn with_http_request_timeout_secs(mut self, secs: u64) -> Self {
        self.http_request_timeout_secs = secs;
        self
    }

    pub fn with_tip_wait_backoff_secs(mut self, secs: u64) -> Self {
        self.tip_wait_backoff_secs = secs;
        self
    }

    pub fn with_tip_wait_max_stall_cycles(mut self, cycles: u32) -> Self {
        self.tip_wait_max_stall_cycles = cycles;
        self
    }

    pub fn with_tip_wait_max_total_secs(mut self, secs: u64) -> Self {
        self.tip_wait_max_total_secs = secs;
        self
    }

    pub fn build(options: ProviderOptions, chain_id: u64) -> Result<Self> {
        let mut settings = Self::default();
        let mut skipped_providers: Vec<&'static str> = Vec::new();
        let mut http_providers = Vec::new();
        let mut ws_providers = Vec::new();

        if let Some(key) = options.ankr_api_key
            && let Some(name) = push_provider_if_supported(
                &mut http_providers,
                &mut ws_providers,
                RpcProvider::Ankr,
                ProviderCredentials::ApiKey(&key),
                chain_id,
            )?
        {
            skipped_providers.push(name);
        }

        if let Some(key) = options.infura_api_key
            && let Some(name) = push_provider_if_supported(
                &mut http_providers,
                &mut ws_providers,
                RpcProvider::Infura,
                ProviderCredentials::ApiKey(&key),
                chain_id,
            )?
        {
            skipped_providers.push(name);
        }

        if let Some(key) = options.drpc_api_key
            && let Some(name) = push_provider_if_supported(
                &mut http_providers,
                &mut ws_providers,
                RpcProvider::Drpc,
                ProviderCredentials::ApiKey(&key),
                chain_id,
            )?
        {
            skipped_providers.push(name);
        }

        if !options.quicknode_endpoints.is_empty()
            && let Some(name) = push_provider_if_supported(
                &mut http_providers,
                &mut ws_providers,
                RpcProvider::QuickNode,
                ProviderCredentials::Endpoints(&options.quicknode_endpoints),
                chain_id,
            )?
        {
            skipped_providers.push(name);
        }

        if let Some(key) = options.alchemy_api_key
            && let Some(name) = push_provider_if_supported(
                &mut http_providers,
                &mut ws_providers,
                RpcProvider::Alchemy,
                ProviderCredentials::ApiKey(&key),
                chain_id,
            )?
        {
            skipped_providers.push(name);
        }

        if let Some(urls) = options.rpc_urls {
            for url in urls {
                let provider = RpcProvider::detect(&url).map_err(|_| {
                    anyhow::anyhow!(
                        "Unknown RPC provider for URL {url}.  Provide explicit RPC settings."
                    )
                })?;
                http_providers.push(ConfiguredProvider {
                    provider: Some(provider),
                    endpoint: provider.rpc_settings(&url),
                });
            }
        }

        if let Some(urls) = options.ws_urls {
            for url in urls {
                let parsed_url: Url = url
                    .parse()
                    .map_err(|e| anyhow::anyhow!("Invalid WS URL {}: {}", url, e))?;
                ws_providers.push(ConfiguredProvider {
                    provider: RpcProvider::detect(&url).ok(),
                    endpoint: parsed_url,
                });
            }
        }

        let http_providers = select_providers(
            http_providers,
            options.http_provider_preferences.as_deref(),
            "HTTP",
        )?;
        let ws_providers = select_providers(
            ws_providers,
            options.websocket_provider_preferences.as_deref(),
            "WebSocket",
        )?;

        if http_providers.is_empty() {
            return Err(missing_provider_error(chain_id, "HTTP", &skipped_providers));
        }

        if ws_providers.is_empty() {
            return Err(missing_provider_error(
                chain_id,
                "WebSocket",
                &skipped_providers,
            ));
        }

        settings.http_providers = http_providers
            .into_iter()
            .map(|provider| provider.endpoint)
            .collect();
        settings.wss_endpoints = ws_providers
            .into_iter()
            .map(|provider| provider.endpoint)
            .collect();

        Ok(settings)
    }

    pub fn new(http_providers: Vec<HttpRpcSettings>, wss_endpoints: Vec<Url>) -> Self {
        Self {
            http_providers,
            wss_endpoints,
            ..Self::default()
        }
    }

    pub fn http_settings(&self, index: usize) -> &HttpRpcSettings {
        let index = index % self.http_providers.len();
        &self.http_providers[index]
    }

    /// Build a `ProviderSettings` backed by a pre-built mock provider.
    /// All `connect_http` calls return clones of the same provider.
    /// Used by hermetic integration tests; not for production use.
    pub fn from_mock(provider: DynProvider<Ethereum>) -> Self {
        Self {
            injected_http: Some(provider),
            http_providers: vec![HttpRpcSettings {
                url: "http://mock".parse().expect("static url"),
                // Just the natural default for "one mock provider" — not verified
                // to be load-bearing for any test's dispatch or assertion order.
                // A test exercising real concurrent dispatch needs its own
                // `ProviderSettings` with a higher bound anyway (see
                // `flush_window_range_dispatches_concurrently_and_restores_ascending_order`).
                max_concurrency: 1,
                max_batch_size: 100_000,
                init_batch_size: 100_000,
                max_logs_per_request: 1_000_000,
            }],
            // Zero backoff so tip-wait integration tests don't sleep.
            tip_wait_backoff_secs: 0,
            // Zero window so hermetic live-stream tests don't sleep.
            live_head_coalesce_window_ms: 0,
            ..Self::default()
        }
    }

    pub fn connect_http(&self, index: usize) -> DynProvider<Ethereum> {
        if let Some(ref p) = self.injected_http {
            return p.clone();
        }
        assert!(
            !self.http_providers.is_empty(),
            "connect_http: no HTTP providers configured"
        );
        let index = index % self.http_providers.len();
        let provider = &self.http_providers[index];
        ProviderBuilder::new()
            .with_reqwest(provider.url.clone(), |client_builder| {
                client_builder
                    .connect_timeout(Duration::from_secs(self.http_connect_timeout_secs))
                    .timeout(Duration::from_secs(self.http_request_timeout_secs))
                    .build()
                    .expect("failed to build reqwest client")
            })
            .erased()
    }

    /// Build one persistent HTTP client per configured provider, via
    /// `connect_http` — never a hand-copied `ProviderBuilder` chain, so
    /// `injected_http`'s mock short-circuit (above) still applies per client
    /// and every `from_mock`-backed hermetic test stays unaffected.
    ///
    /// Intended to be built once per stream invocation (by the live-heads
    /// path) and reused across every flush and provider-fallback attempt in
    /// that invocation, rather than rebuilding a fresh client — with a fresh
    /// TCP+TLS handshake — on every call.
    pub(crate) fn build_reused_http_clients(&self) -> Vec<DynProvider<Ethereum>> {
        (0..self.http_providers.len())
            .map(|i| self.connect_http(i))
            .collect()
    }

    pub async fn connect_ws(&self, start_idx: usize) -> Result<DynProvider<Ethereum>, RpcError> {
        let mut attempt = 0;
        let mut provider_idx = start_idx;
        let mut last_endpoint_host: Option<String> = None;

        while attempt < MAX_RETRIES {
            let endpoint = &self.wss_endpoints[provider_idx % self.wss_endpoints.len()];
            let ws_conn = ProviderBuilder::new()
                .connect_ws(WsConnect::new(endpoint.clone()))
                .await;

            match ws_conn {
                Ok(ws) => return Ok(ws.erased()),
                Err(e) => {
                    let endpoint_host = endpoint.host_str().unwrap_or("unknown").to_string();
                    last_endpoint_host = Some(endpoint_host.clone());
                    attempt += 1;
                    let backoff = Duration::from_secs(self.base_backoff_secs.pow(attempt));
                    warn!(
                        "Failed to connect to WebSocket {}: {} (attempt {}/{}, backoff {:?})",
                        endpoint_host, e, attempt, self.max_retries, backoff
                    );

                    if attempt >= self.max_retries {
                        return Err(RpcError::ConnectionError(format!(
                            "Failed to connect to WebSocket {} after {}/{} attempts: {}",
                            endpoint_host, attempt, self.max_retries, e
                        )));
                    }

                    tokio::time::sleep(backoff).await;
                    provider_idx += 1;
                }
            }
        }

        Err(RpcError::ConnectionError(
            match last_endpoint_host.as_deref() {
                Some(host) => {
                    format!("WebSocket connection failed for {host}: max retries exceeded")
                }
                None => "WebSocket connection failed: max retries exceeded".to_string(),
            },
        ))
    }
}

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

    fn configured_provider_options() -> ProviderOptions {
        ProviderOptions::default()
            .ankr_api_key("ankr-key")
            .infura_api_key("infura-key")
            .drpc_api_key("drpc-key")
            .quicknode_endpoint(1, "https://eth-mainnet.quiknode.pro/quicknode-key")
            .alchemy_api_key("alchemy-key")
    }

    #[test]
    fn test_http_returns_error_for_unknown_provider() {
        let result = ProviderSettings::http("https://unknown-provider.example.com/rpc");
        assert!(result.is_err());
        let err = result.unwrap_err();
        assert!(
            err.to_string().contains("Unknown RPC provider"),
            "Error message should mention unknown provider, got: {}",
            err
        );
    }

    #[test]
    fn test_build_returns_error_for_unknown_rpc_url() {
        let options = ProviderOptions::default()
            .add_http("https://unknown-provider.example.com/rpc".to_string())
            .add_websocket("wss://eth-mainnet.g.alchemy.com/ws/v2/test".to_string());

        let result = ProviderSettings::build(options, 1);
        assert!(result.is_err());
        let err = result.unwrap_err();
        assert!(
            err.to_string().contains("Unknown RPC provider"),
            "Error message should mention unknown provider, got: {}",
            err
        );
    }

    #[test]
    fn test_build_returns_error_for_unsupported_chain_id() {
        let options = ProviderOptions::default().infura_api_key("test-key");

        let result = ProviderSettings::build(options, 10200);
        assert!(result.is_err());

        let err = result.unwrap_err().to_string();
        assert!(
            err.contains("chain 10200"),
            "Error should mention the chain ID, got: {}",
            err
        );
        assert!(
            err.contains("infura"),
            "Error should mention the skipped provider, got: {}",
            err
        );
    }

    #[test]
    fn build_skips_a_provider_that_does_not_serve_the_chain() {
        // Chain 4663 (Robinhood Chain) is Alchemy-only among the built-in key providers —
        // Ankr does not serve it. A configured-but-unsupported key must be skipped, not
        // fail the whole build, as long as another configured provider covers the chain.
        let options = ProviderOptions::default()
            .ankr_api_key("ankr-key")
            .alchemy_api_key("alchemy-key");

        let settings = ProviderSettings::build(options, 4663)
            .expect("alchemy alone should suffice for chain 4663");

        assert_eq!(settings.http_providers.len(), 1);
        assert_eq!(settings.wss_endpoints.len(), 1);
        assert!(settings.http_providers[0].host().contains("alchemy.com"));
    }

    #[test]
    fn provider_preferences_are_opt_in_and_preserve_default_order() {
        let settings = ProviderSettings::build(configured_provider_options(), 1).unwrap();

        let http_hosts = settings
            .http_providers
            .iter()
            .map(HttpRpcSettings::host)
            .collect::<Vec<_>>();
        let ws_hosts = settings
            .wss_endpoints
            .iter()
            .map(|url| url.host_str().unwrap_or("unknown").to_string())
            .collect::<Vec<_>>();

        assert_eq!(
            http_hosts,
            [
                "rpc.ankr.com",
                "mainnet.infura.io",
                "lb.drpc.live",
                "eth-mainnet.quiknode.pro",
                "eth-mainnet.g.alchemy.com",
            ]
        );
        assert_eq!(ws_hosts, http_hosts);
    }

    #[test]
    fn provider_preferences_reorder_transports_independently() {
        let options = configured_provider_options()
            .http_provider_preferences([RpcProvider::Alchemy, RpcProvider::Drpc])
            .websocket_provider_preferences([RpcProvider::Drpc, RpcProvider::Alchemy]);

        let settings = ProviderSettings::build(options, 1).unwrap();

        assert_eq!(settings.http_providers.len(), 5);
        assert!(settings.http_providers[0].host().contains("alchemy.com"));
        assert!(settings.http_providers[1].host().contains("drpc.live"));
        assert!(settings.http_providers[2].host().contains("ankr.com"));
        assert_eq!(settings.wss_endpoints.len(), 5);
        assert!(
            settings.wss_endpoints[0]
                .host_str()
                .unwrap_or("")
                .contains("drpc.live")
        );
        assert!(
            settings.wss_endpoints[1]
                .host_str()
                .unwrap_or("")
                .contains("alchemy.com")
        );
        assert!(
            settings.wss_endpoints[2]
                .host_str()
                .unwrap_or("")
                .contains("ankr.com")
        );
    }

    #[test]
    fn websocket_preferences_do_not_require_matching_http_providers() {
        let options = configured_provider_options()
            .http_provider_preferences([RpcProvider::Alchemy])
            .websocket_provider_preferences([RpcProvider::Drpc]);

        let settings = ProviderSettings::build(options, 1).unwrap();

        assert_eq!(settings.http_providers.len(), 5);
        assert!(settings.http_providers[0].host().contains("alchemy.com"));
        assert_eq!(settings.wss_endpoints.len(), 5);
        assert!(
            settings.wss_endpoints[0]
                .host_str()
                .unwrap_or("")
                .contains("drpc.live")
        );
    }

    #[test]
    fn provider_preferences_promote_available_choices_and_reject_duplicates() {
        let settings = ProviderSettings::build(
            ProviderOptions::default()
                .alchemy_api_key("alchemy-key")
                .http_provider_preferences([RpcProvider::Drpc, RpcProvider::Alchemy])
                .websocket_provider_preferences([RpcProvider::Drpc, RpcProvider::Alchemy]),
            1,
        )
        .unwrap();
        assert_eq!(settings.http_providers.len(), 1);
        assert!(settings.http_providers[0].host().contains("alchemy.com"));

        let empty = ProviderSettings::build(
            configured_provider_options().http_provider_preferences([]),
            1,
        )
        .unwrap();
        assert!(empty.http_providers[0].host().contains("ankr.com"));

        let duplicate = ProviderSettings::build(
            configured_provider_options()
                .websocket_provider_preferences([RpcProvider::Drpc, RpcProvider::Drpc]),
            1,
        )
        .unwrap_err();
        assert!(duplicate.to_string().contains("duplicate provider drpc"));

        let unavailable = ProviderSettings::build(
            ProviderOptions::default()
                .alchemy_api_key("alchemy-key")
                .http_provider_preferences([RpcProvider::Drpc]),
            1,
        )
        .unwrap();
        assert_eq!(unavailable.http_providers.len(), 1);
        assert!(unavailable.http_providers[0].host().contains("alchemy.com"));
    }

    #[test]
    fn test_default_http_timeouts() {
        let settings = ProviderSettings::default();
        assert_eq!(
            settings.http_connect_timeout_secs,
            DEFAULT_HTTP_CONNECT_TIMEOUT_SECS
        );
        assert_eq!(
            settings.http_request_timeout_secs,
            DEFAULT_HTTP_REQUEST_TIMEOUT_SECS
        );
    }

    #[test]
    fn quicknode_endpoint_is_selected_by_chain_and_preserves_its_path() {
        let options = ProviderOptions::default()
            .quicknode_endpoint(
                1,
                "https://eth-mainnet.quiknode.pro/ethereum-key/?region=us-east",
            )
            .quicknode_endpoint(
                8453,
                "https://base-mainnet.quiknode.pro/base-key/?region=us-east",
            );

        let settings = ProviderSettings::build(options, 8453).unwrap();

        assert_eq!(settings.http_providers.len(), 1);
        assert_eq!(settings.wss_endpoints.len(), 1);
        assert_eq!(
            settings.http_providers[0].url.as_str(),
            "https://base-mainnet.quiknode.pro/base-key/?region=us-east"
        );
        assert_eq!(
            settings.wss_endpoints[0].as_str(),
            "wss://base-mainnet.quiknode.pro/base-key/?region=us-east"
        );
    }

    #[test]
    fn quicknode_endpoint_scheme_is_normalized_for_each_transport() {
        let options = ProviderOptions::default().quicknode_endpoint(
            1,
            "wss://eth-mainnet.quiknode.pro/endpoint-key?region=us-east",
        );

        let settings = ProviderSettings::build(options, 1).unwrap();

        assert_eq!(
            settings.http_providers[0].url.as_str(),
            "https://eth-mainnet.quiknode.pro/endpoint-key?region=us-east"
        );
        assert_eq!(
            settings.wss_endpoints[0].as_str(),
            "wss://eth-mainnet.quiknode.pro/endpoint-key?region=us-east"
        );
    }

    #[test]
    fn quicknode_endpoint_must_be_a_valid_url() {
        let options = ProviderOptions::default().quicknode_endpoint(1, "not-a-url");

        let error = ProviderSettings::build(options, 1).unwrap_err();

        assert!(
            error
                .to_string()
                .contains("invalid endpoint for provider quicknode")
        );
    }

    #[test]
    fn test_default_live_head_coalesce_window() {
        // Cost-oriented default: tokindex rides this without opting in.
        assert_eq!(
            ProviderSettings::default().live_head_coalesce_window_ms,
            30_000
        );
    }

    #[test]
    fn test_from_mock_disables_live_head_coalesce_window() {
        use alloy::transports::mock::Asserter;

        // Hermetic stream tests must not sleep, so from_mock zeroes the window
        // the same way it zeroes tip_wait_backoff_secs.
        let provider = ProviderBuilder::new()
            .disable_recommended_fillers()
            .connect_mocked_client(Asserter::new())
            .erased();
        assert_eq!(
            ProviderSettings::from_mock(provider).live_head_coalesce_window_ms,
            0
        );
    }

    #[test]
    fn build_ethereum_mainnet_resolves_all_providers() {
        // chain_id 1 (Ethereum mainnet) must not hit the UnsupportedChain path for any key provider.
        // Positive counterpart to test_build_returns_error_for_unsupported_chain_id.
        let infura =
            ProviderSettings::build(ProviderOptions::default().infura_api_key("key"), 1).unwrap();
        assert_eq!(infura.http_providers.len(), 1);
        assert_eq!(infura.wss_endpoints.len(), 1);
        assert!(infura.http_providers[0].host().contains("infura.io"));
        assert!(
            infura.wss_endpoints[0]
                .host_str()
                .unwrap_or("")
                .contains("infura.io")
        );

        let ankr =
            ProviderSettings::build(ProviderOptions::default().ankr_api_key("key"), 1).unwrap();
        assert_eq!(ankr.http_providers.len(), 1);
        assert_eq!(ankr.wss_endpoints.len(), 1);
        assert!(ankr.http_providers[0].host().contains("ankr.com"));
        assert!(
            ankr.wss_endpoints[0]
                .host_str()
                .unwrap_or("")
                .contains("ankr.com")
        );

        let alchemy =
            ProviderSettings::build(ProviderOptions::default().alchemy_api_key("key"), 1).unwrap();
        assert_eq!(alchemy.http_providers.len(), 1);
        assert_eq!(alchemy.wss_endpoints.len(), 1);
        assert!(alchemy.http_providers[0].host().contains("alchemy.com"));
        assert!(
            alchemy.wss_endpoints[0]
                .host_str()
                .unwrap_or("")
                .contains("alchemy.com")
        );

        let drpc =
            ProviderSettings::build(ProviderOptions::default().drpc_api_key("key"), 1).unwrap();
        assert_eq!(drpc.http_providers.len(), 1);
        assert_eq!(drpc.wss_endpoints.len(), 1);
        assert!(drpc.http_providers[0].host().contains("drpc.live"));
        assert!(
            drpc.wss_endpoints[0]
                .host_str()
                .unwrap_or("")
                .contains("drpc.live")
        );

        let quicknode = ProviderSettings::build(
            ProviderOptions::default()
                .quicknode_endpoint(1, "https://eth-mainnet.quiknode.pro/endpoint-key"),
            1,
        )
        .unwrap();
        assert_eq!(quicknode.http_providers.len(), 1);
        assert_eq!(quicknode.wss_endpoints.len(), 1);
        assert!(quicknode.http_providers[0].host().contains("quiknode.pro"));
        assert!(
            quicknode.wss_endpoints[0]
                .host_str()
                .unwrap_or("")
                .contains("quiknode.pro")
        );
    }

    /// `connect_http` is unreachable at `N == 0` in production for the same
    /// reasons `resolve_http_provider` is (see `heads.rs`'s
    /// `resolve_http_provider_panics_with_labeled_message_on_empty_http_providers`),
    /// but the arithmetic underneath it still divides by zero if this guard is
    /// ever removed. The message must name the misconfiguration, not the
    /// arithmetic, and must differ from `resolve_http_provider`'s so a revert of
    /// just one guard is distinguishable from the other.
    #[test]
    #[should_panic(expected = "connect_http: no HTTP providers configured")]
    fn connect_http_panics_with_labeled_message_on_empty_http_providers() {
        let settings = ProviderSettings::new(
            vec![], // zero configured HTTP providers
            vec!["ws://127.0.0.1:9".parse().expect("static url")],
        );
        settings.connect_http(0);
    }
}