ethl 0.1.44

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,
    /// How deep a recovery window the live-tip detector retains, in blocks: the
    /// span of its `(number, hash)` ring, and the budget the backward reorg walk
    /// may spend resolving a common ancestor. `0` disables detection (default) —
    /// every L2 and the snapshot path stay untouched.
    ///
    /// Local capacity, never a chain property. It bounds how far back we can
    /// still name an ancestor from what we retained; running past it establishes
    /// no fork depth and no finality, which is why exhausting it reports
    /// `Gap` rather than a claim about the chain. Size it to the depth
    /// of state the consumer can actually roll back to.
    /// Only `stream_heads_with_logs` reads it; the capture path never does.
    pub recovery_window_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 `recovery_window_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,
            recovery_window_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, exhaustive HTTP provider allowlist. `None` (default) selects and
    /// orders every configured provider in statement order (ankr, infura, drpc,
    /// quicknode, alchemy). `Some(list)` is exhaustive: a key-derived provider
    /// not on the list is never used on this transport, however healthy,
    /// however configured — position determines priority, and index 0 serves
    /// `connect_http(0)`, which retries onto nothing. `Some(vec![])` and a
    /// duplicate entry are both build-time errors. Operator-typed `rpc_urls`
    /// are never filtered by this list.
    pub http_provider_allowlist: Option<Vec<RpcProvider>>,
    /// Ordered, exhaustive WebSocket provider allowlist. Same semantics as
    /// `http_provider_allowlist`, independent per transport; operator-typed
    /// `ws_urls` are never filtered by this list.
    pub websocket_provider_allowlist: 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_allowlist(
        mut self,
        providers: impl IntoIterator<Item = RpcProvider>,
    ) -> Self {
        self.http_provider_allowlist = Some(providers.into_iter().collect());
        self
    }

    pub fn websocket_provider_allowlist(
        mut self,
        providers: impl IntoIterator<Item = RpcProvider>,
    ) -> Self {
        self.websocket_provider_allowlist = 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
    }
}

/// Distinguishes a provider derived from a configured API key or endpoint set
/// from one an operator named directly via `rpc_urls` / `ws_urls`. The
/// allowlist governs key-derived entries only (decision 7); an operator-typed
/// URL survives whatever the allowlist says about its detected provider.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ProviderOrigin {
    KeyDerived,
    OperatorTyped,
}

/// 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>,
    origin: ProviderOrigin,
    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),
                origin: ProviderOrigin::KeyDerived,
                endpoint: provider.rpc_settings(http_url.as_str()),
            });
            ws_providers.push(ConfiguredProvider {
                provider: Some(provider),
                origin: ProviderOrigin::KeyDerived,
                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()),
    }
}

/// Selects and orders `providers` for one transport against its declared
/// allowlist (decisions 1-3, 5-7). `None` returns `providers` untouched —
/// statement order, PR #58's default (decision 4). `Some(list)` is validated
/// (non-empty, duplicate-free) and then exhaustive: every key-derived entry
/// not on `list` is dropped and its name returned in the excluded set;
/// operator-typed entries are never dropped, grouping with their provider's
/// list position when listed and trailing in supplied order otherwise.
fn select_providers<T: Clone>(
    providers: Vec<ConfiguredProvider<T>>,
    allowlist: Option<&[RpcProvider]>,
    chain_id: u64,
    transport: &str,
) -> Result<(Vec<ConfiguredProvider<T>>, Vec<&'static str>)> {
    let Some(allowlist) = allowlist else {
        return Ok((providers, Vec::new()));
    };

    if allowlist.is_empty() {
        anyhow::bail!(
            "chain {chain_id}: {transport} provider allowlist is empty; \
             no provider may serve this chain"
        );
    }

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

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

    let mut excluded = Vec::new();
    for configured in &providers {
        let listed = configured
            .provider
            .is_some_and(|provider| allowlist.contains(&provider));
        if listed {
            continue;
        }
        match configured.origin {
            // Key-derived entries always carry a known provider.
            ProviderOrigin::KeyDerived => {
                if let Some(provider) = configured.provider {
                    excluded.push(provider.name());
                }
            }
            ProviderOrigin::OperatorTyped => selected.push(configured.clone()),
        }
    }

    Ok((selected, excluded))
}

/// Registers one key-derived provider: pushes its endpoints if an API key is
/// present and the chain is served, else records it in `no_key` (unconfigured)
/// or `no_slug` (configured but not serving this chain) for
/// `missing_provider_error`. Shared by Ankr, Infura, dRPC, and Alchemy, whose
/// `build()` wiring differs only in provider and key.
fn register_key_provider(
    http_providers: &mut Vec<ConfiguredProvider<HttpRpcSettings>>,
    ws_providers: &mut Vec<ConfiguredProvider<Url>>,
    provider: RpcProvider,
    api_key: Option<&str>,
    chain_id: u64,
    no_key: &mut Vec<&'static str>,
    no_slug: &mut Vec<&'static str>,
) -> Result<()> {
    match api_key {
        Some(key) => {
            if let Some(name) = push_provider_if_supported(
                http_providers,
                ws_providers,
                provider,
                ProviderCredentials::ApiKey(key),
                chain_id,
            )? {
                no_slug.push(name);
            }
        }
        None => no_key.push(provider.name()),
    }
    Ok(())
}

/// The three distinct ways a chain can end up with no usable provider on one
/// transport (decision 8): unconfigured, configured but not serving this
/// chain, and configured-and-serving but excluded by this transport's own
/// allowlist. An operator reading this at 4am must be able to tell which
/// lever fixes it without reading source.
fn missing_provider_error(
    chain_id: u64,
    kind: &str,
    no_key: &[&'static str],
    no_slug: &[&'static str],
    excluded_by_policy: &[&'static str],
) -> anyhow::Error {
    if no_key.is_empty() && no_slug.is_empty() && excluded_by_policy.is_empty() {
        return anyhow::anyhow!(
            "At least one {kind} provider must be configured via API key or custom URL"
        );
    }

    let mut reasons: Vec<String> = Vec::new();
    if !no_key.is_empty() {
        reasons.push(format!("not configured: {}", no_key.join(", ")));
    }
    if !no_slug.is_empty() {
        reasons.push(format!(
            "configured but does not serve chain {chain_id}: {}",
            no_slug.join(", ")
        ));
    }
    if !excluded_by_policy.is_empty() {
        reasons.push(format!(
            "configured and serves chain {chain_id} but excluded by the {kind} provider \
             allowlist: {}",
            excluded_by_policy.join(", ")
        ));
    }

    anyhow::anyhow!(
        "chain {chain_id} has no usable {kind} provider ({}); add a provider that supports it \
         or adjust the allowlist",
        reasons.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();
        // Shared across both transports: whether a provider carries a key
        // and whether it serves this chain don't depend on transport, only
        // the allowlist below does (decision 3).
        let mut no_key: Vec<&'static str> = Vec::new();
        let mut no_slug: Vec<&'static str> = Vec::new();
        let mut http_providers = Vec::new();
        let mut ws_providers = Vec::new();

        register_key_provider(
            &mut http_providers,
            &mut ws_providers,
            RpcProvider::Ankr,
            options.ankr_api_key.as_deref(),
            chain_id,
            &mut no_key,
            &mut no_slug,
        )?;

        register_key_provider(
            &mut http_providers,
            &mut ws_providers,
            RpcProvider::Infura,
            options.infura_api_key.as_deref(),
            chain_id,
            &mut no_key,
            &mut no_slug,
        )?;

        register_key_provider(
            &mut http_providers,
            &mut ws_providers,
            RpcProvider::Drpc,
            options.drpc_api_key.as_deref(),
            chain_id,
            &mut no_key,
            &mut no_slug,
        )?;

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

        register_key_provider(
            &mut http_providers,
            &mut ws_providers,
            RpcProvider::Alchemy,
            options.alchemy_api_key.as_deref(),
            chain_id,
            &mut no_key,
            &mut no_slug,
        )?;

        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),
                    origin: ProviderOrigin::OperatorTyped,
                    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(),
                    origin: ProviderOrigin::OperatorTyped,
                    endpoint: parsed_url,
                });
            }
        }

        let (http_providers, http_excluded) = select_providers(
            http_providers,
            options.http_provider_allowlist.as_deref(),
            chain_id,
            "HTTP",
        )?;
        let (ws_providers, ws_excluded) = select_providers(
            ws_providers,
            options.websocket_provider_allowlist.as_deref(),
            chain_id,
            "WebSocket",
        )?;

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

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

        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 allowlist_selects_and_orders_transports_independently() {
        // Exclusion, not reordering: [Alchemy, Drpc] on HTTP yields exactly
        // those two, not five with three retained as fallbacks. Each
        // transport's list governs its own transport only.
        let options = configured_provider_options()
            .http_provider_allowlist([RpcProvider::Alchemy, RpcProvider::Drpc])
            .websocket_provider_allowlist([RpcProvider::Drpc, RpcProvider::Alchemy]);

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

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

    #[test]
    fn websocket_allowlist_does_not_require_matching_http_providers() {
        // Full key set, [Alchemy] HTTP / [Drpc] WS: one provider per
        // transport, each the declared one -- per-transport divergence.
        let options = configured_provider_options()
            .http_provider_allowlist([RpcProvider::Alchemy])
            .websocket_provider_allowlist([RpcProvider::Drpc]);

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

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

    #[test]
    fn allowlist_promotes_an_available_choice_ahead_of_an_unconfigured_one() {
        // Drpc heads the list but carries no key; it lands in the no-key
        // group, not an error, since Alchemy -- also listed -- is configured
        // and serves the chain.
        let settings = ProviderSettings::build(
            ProviderOptions::default()
                .alchemy_api_key("alchemy-key")
                .http_provider_allowlist([RpcProvider::Drpc, RpcProvider::Alchemy])
                .websocket_provider_allowlist([RpcProvider::Drpc, RpcProvider::Alchemy]),
            1,
        )
        .unwrap();
        assert_eq!(settings.http_providers.len(), 1);
        assert!(settings.http_providers[0].host().contains("alchemy.com"));
    }

    #[test]
    fn allowlist_rejects_an_empty_list() {
        // Some(vec![]) is "no provider may serve this chain" -- a
        // configuration mistake worth failing loudly, not a synonym for
        // None (decision 5). This FLIPS PR #58's landed behavior, where an
        // empty preference list was inert.
        let error =
            ProviderSettings::build(configured_provider_options().http_provider_allowlist([]), 1)
                .unwrap_err();
        assert!(error.to_string().contains("empty"), "got: {error}");
    }

    #[test]
    fn allowlist_rejects_a_duplicate_entry() {
        let error = ProviderSettings::build(
            configured_provider_options()
                .websocket_provider_allowlist([RpcProvider::Drpc, RpcProvider::Drpc]),
            1,
        )
        .unwrap_err();
        assert!(error.to_string().contains("duplicate provider drpc"));
    }

    #[test]
    fn allowlist_excludes_rather_than_falls_back_to_an_unlisted_configured_provider() {
        // Drpc heads the list but carries no key, and Alchemy -- configured
        // and serving chain 1 -- is excluded for not being on the list. This
        // FLIPS PR #58's landed behavior, where an unlisted configured
        // provider was retained as a fallback: the allowlist is exhaustive
        // (decision 2), so this must error, naming both causes.
        let error = ProviderSettings::build(
            ProviderOptions::default()
                .alchemy_api_key("alchemy-key")
                .http_provider_allowlist([RpcProvider::Drpc]),
            1,
        )
        .unwrap_err()
        .to_string();
        assert!(
            error.contains("excluded by the HTTP provider allowlist"),
            "got: {error}"
        );
        assert!(error.contains("alchemy"), "got: {error}");
        assert!(error.contains("not configured"), "got: {error}");
        assert!(error.contains("drpc"), "got: {error}");
    }

    #[test]
    fn allowlist_filters_and_orders_the_built_providers() {
        // The goal test: chain 1 with Ankr, Infura, and Alchemy keys; both
        // transport lists set to [Alchemy, Ankr]. The built providers are
        // exactly those two, in that order -- Infura's absence proves
        // exclusion, not just reordering. Ignoring the list fails this on
        // order; treating it as a preference fails it on count.
        let options = ProviderOptions::default()
            .ankr_api_key("ankr-key")
            .infura_api_key("infura-key")
            .alchemy_api_key("alchemy-key")
            .http_provider_allowlist([RpcProvider::Alchemy, RpcProvider::Ankr])
            .websocket_provider_allowlist([RpcProvider::Alchemy, RpcProvider::Ankr]);

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

        let http_hosts: Vec<String> = settings.http_providers.iter().map(|s| s.host()).collect();
        assert_eq!(http_hosts.len(), 2);
        assert!(http_hosts[0].contains("alchemy.com"));
        assert!(http_hosts[1].contains("ankr.com"));

        let ws_hosts: Vec<String> = settings
            .wss_endpoints
            .iter()
            .map(|url| url.host_str().unwrap_or("unknown").to_string())
            .collect();
        assert_eq!(ws_hosts, http_hosts);
    }

    #[test]
    fn build_allowlist_excludes_the_only_provider_with_a_slug() {
        // Chain 4663 (Robinhood) is Alchemy-only among the key providers. An
        // allowlist naming only Ankr must fail outright, not fall back to
        // Alchemy -- the allowlist is exhaustive, never a last resort
        // (decision 2).
        let options = ProviderOptions::default()
            .alchemy_api_key("alchemy-key")
            .http_provider_allowlist([RpcProvider::Ankr]);

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

    #[test]
    fn build_error_names_providers_with_no_key_or_endpoint_configured() {
        // No keys and no QuickNode endpoints at all: the message must name
        // every provider in the "not configured" group, not the old generic
        // "at least one provider" catch-all -- closing the gap where
        // key-absent providers vanished silently in the `if let` chain.
        let error = ProviderSettings::build(ProviderOptions::default(), 1)
            .unwrap_err()
            .to_string();
        assert!(error.contains("not configured"), "got: {error}");
        for provider in ["ankr", "infura", "drpc", "quicknode", "alchemy"] {
            assert!(error.contains(provider), "expected {provider} in: {error}");
        }
    }

    #[test]
    fn build_error_names_a_key_provider_that_does_not_serve_the_chain() {
        // Infura has a key but no slug for chain 10200; the message must
        // name it in the "does not serve" group, distinct from "not
        // configured".
        let options = ProviderOptions::default().infura_api_key("key");
        let error = ProviderSettings::build(options, 10200)
            .unwrap_err()
            .to_string();
        assert!(error.contains("chain 10200"), "got: {error}");
        assert!(error.contains("does not serve"), "got: {error}");
        assert!(error.contains("infura"), "got: {error}");
    }

    #[test]
    fn build_error_names_a_quicknode_endpoint_missing_for_the_chain() {
        // QuickNode's own "does not serve" axis: an endpoint is configured
        // for chain 1 but not chain 8453, so it is EndpointNotConfigured for
        // 8453 -- the same group as a key provider without a chain slug.
        let options =
            ProviderOptions::default().quicknode_endpoint(1, "https://eth.quiknode.pro/key");
        let error = ProviderSettings::build(options, 8453)
            .unwrap_err()
            .to_string();
        assert!(error.contains("chain 8453"), "got: {error}");
        assert!(error.contains("does not serve"), "got: {error}");
        assert!(error.contains("quicknode"), "got: {error}");
    }

    #[test]
    fn build_error_names_a_provider_excluded_by_the_allowlist() {
        // Alchemy is configured and serves chain 4663, but the allowlist
        // excludes it; the message must name it in the "excluded" group,
        // distinct from both "not configured" and "does not serve".
        let options = ProviderOptions::default()
            .alchemy_api_key("alchemy-key")
            .http_provider_allowlist([RpcProvider::Ankr]);
        let error = ProviderSettings::build(options, 4663)
            .unwrap_err()
            .to_string();
        assert!(error.contains("chain 4663"), "got: {error}");
        assert!(
            error.contains("excluded by the HTTP provider allowlist"),
            "got: {error}"
        );
        assert!(error.contains("alchemy"), "got: {error}");
    }

    #[test]
    fn build_keeps_explicit_urls_despite_an_allowlist_that_excludes_their_provider() {
        // rpc_urls / ws_urls name a specific operator-typed endpoint; the
        // allowlist governs key-derived providers only (decision 7).
        let options = ProviderOptions::default()
            .alchemy_api_key("alchemy-key")
            .http_provider_allowlist([RpcProvider::Alchemy])
            .add_http("https://rpc.ankr.com/eth/explicit-key".to_string())
            .add_websocket("wss://rpc.ankr.com/eth/explicit-key".to_string());

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

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

    #[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);
    }
}