ethl 0.1.34

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, 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,
    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.
    ///
    /// Default 1800 s (30 min): rides the known ~23-min Arbitrum catch-up case while
    /// bounding a crawling-but-never-converging provider. Tradeoff: this is a blunt bound —
    /// a provider that would have caught up just past 30 min is also terminated. It is a
    /// single global default applied to all chains; callers may 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 sequential 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 quicknode_api_key: Option<String>,
    pub alchemy_api_key: Option<String>,
    pub rpc_urls: Option<Vec<String>>,
    pub ws_urls: Option<Vec<String>>,
}

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 quicknode_api_key(mut self, key: impl AsRef<str>) -> Self {
        self.quicknode_api_key = Some(key.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 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.
fn push_provider_if_supported(
    settings: &mut ProviderSettings,
    provider: RpcProvider,
    key: &str,
    chain_id: u64,
) -> Result<Option<&'static str>> {
    match provider.http_url(key, chain_id) {
        Ok(http_url) => {
            let ws_url = provider.ws_url(key, chain_id)?;
            settings
                .http_providers
                .push(provider.rpc_settings(http_url.as_str()));
            settings.wss_endpoints.push(ws_url);
            Ok(None)
        }
        // Exhaustive on purpose: ProviderUrlError has exactly one variant today. A
        // future variant must be handled explicitly here (compiler-enforced), not
        // silently swallowed as a skip.
        Err(ProviderUrlError::UnsupportedChain { provider: name, .. }) => Ok(Some(name)),
    }
}

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();

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

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

        if let Some(key) = options.quicknode_api_key
            && let Some(name) =
                push_provider_if_supported(&mut settings, RpcProvider::QuickNode, &key, chain_id)?
        {
            skipped_providers.push(name);
        }

        if let Some(key) = options.alchemy_api_key
            && let Some(name) =
                push_provider_if_supported(&mut settings, RpcProvider::Alchemy, &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."
                    )
                })?;
                settings.http_providers.push(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))?;
                settings.wss_endpoints.push(parsed_url);
            }
        }

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

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

        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"),
                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();
        }
        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()
    }

    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::*;

    #[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 four provider tables —
        // 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 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 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 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 quicknode =
            ProviderSettings::build(ProviderOptions::default().quicknode_api_key("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")
        );
    }
}