use std::{fmt::Debug, time::Duration};

use alloy::{
    eips::{BlockId, BlockNumberOrTag},
    network::{Ethereum, Network},
    primitives::BlockHash,
    providers::{Provider, RootProvider},
    pubsub::Subscription,
    rpc::types::{Filter, Log},
    transports::{RpcError, TransportErrorKind},
};
use backon::{ExponentialBuilder, Retryable};
use tokio::time::timeout;
use tracing::{error, info};

use crate::robust_provider::Error;

/// Provider wrapper with built-in retry and timeout mechanisms.
///
/// This wrapper around Alloy providers automatically handles retries,
/// timeouts, and error logging for RPC calls.
/// The first provider in the vector is treated as the primary provider.
#[derive(Clone)]
pub struct RobustProvider<N: Network = Ethereum> {
    pub(crate) providers: Vec<RootProvider<N>>,
    pub(crate) max_timeout: Duration,
    pub(crate) max_retries: usize,
    pub(crate) min_delay: Duration,
}

impl<N: Network> RobustProvider<N> {
    /// Get a reference to the primary provider (the first provider in the list)
    ///
    /// # Panics
    ///
    /// If there are no providers set (this should never happen)
    #[must_use]
    pub fn primary(&self) -> &RootProvider<N> {
        // Safe to unwrap because we always have at least one provider
        self.providers.first().expect("providers vector should never be empty")
    }

    /// Fetch a block by [`BlockNumberOrTag`] with retry and timeout.
    ///
    /// # Errors
    ///
    /// See [retry errors](#retry-errors).
    pub async fn get_block_by_number(
        &self,
        number: BlockNumberOrTag,
    ) -> Result<N::BlockResponse, Error> {
        info!("eth_getBlockByNumber called");
        let result = self
            .retry_with_total_timeout(
                move |provider| async move { provider.get_block_by_number(number).await },
                false,
            )
            .await;
        if let Err(e) = &result {
            error!(error = %e, "eth_getByBlockNumber failed");
        }

        result?.ok_or_else(|| Error::BlockNotFound(number.into()))
    }

    /// Fetch a block number by [`BlockId`]  with retry and timeout.
    ///
    /// # Errors
    ///
    /// See [retry errors](#retry-errors).
    pub async fn get_block(&self, id: BlockId) -> Result<N::BlockResponse, Error> {
        info!("eth_getBlock called");
        let result = self
            .retry_with_total_timeout(|provider| async move { provider.get_block(id).await }, false)
            .await;
        if let Err(e) = &result {
            error!(error = %e, "eth_getByBlockNumber failed");
        }
        result?.ok_or_else(|| Error::BlockNotFound(id))
    }

    /// Fetch the latest block number with retry and timeout.
    ///
    /// # Errors
    ///
    /// See [retry errors](#retry-errors).
    pub async fn get_block_number(&self) -> Result<u64, Error> {
        info!("eth_getBlockNumber called");
        let result = self
            .retry_with_total_timeout(
                move |provider| async move { provider.get_block_number().await },
                false,
            )
            .await;
        if let Err(e) = &result {
            error!(error = %e, "eth_getBlockNumber failed");
        }
        result
    }

    /// Fetch the latest confirmed block number with retry and timeout.
    ///
    /// This method fetches the latest block number and subtracts the specified
    /// number of confirmations to get a "confirmed" block number.
    ///
    /// # Arguments
    ///
    /// * `confirmations` - The number of block confirmations to wait for. The returned block number
    ///   will be `latest_block - confirmations`.
    ///
    /// # Errors
    ///
    /// See [retry errors](#retry-errors).
    pub async fn get_latest_confirmed(&self, confirmations: u64) -> Result<u64, Error> {
        info!("get_latest_confirmed called with confirmations={}", confirmations);
        let latest_block = self.get_block_number().await?;
        let confirmed_block = latest_block.saturating_sub(confirmations);
        Ok(confirmed_block)
    }

    /// Fetch a block by [`BlockHash`] with retry and timeout.
    ///
    /// # Errors
    ///
    /// See [retry errors](#retry-errors).
    pub async fn get_block_by_hash(&self, hash: BlockHash) -> Result<N::BlockResponse, Error> {
        info!("eth_getBlockByHash called");
        let result = self
            .retry_with_total_timeout(
                move |provider| async move { provider.get_block_by_hash(hash).await },
                false,
            )
            .await;
        if let Err(e) = &result {
            error!(error = %e, "eth_getBlockByHash failed");
        }

        result?.ok_or_else(|| Error::BlockNotFound(hash.into()))
    }

    /// Fetch logs for the given [`Filter`] with retry and timeout.
    ///
    /// # Errors
    ///
    /// See [retry errors](#retry-errors).
    pub async fn get_logs(&self, filter: &Filter) -> Result<Vec<Log>, Error> {
        info!("eth_getLogs called");
        let result = self
            .retry_with_total_timeout(
                move |provider| async move { provider.get_logs(filter).await },
                false,
            )
            .await;
        if let Err(e) = &result {
            error!(error = %e, "eth_getLogs failed");
        }
        result
    }

    /// Subscribe to new block headers with retry and timeout.
    ///
    /// # Errors
    ///
    /// see [retry errors](#retry-errors).
    pub async fn subscribe_blocks(&self) -> Result<Subscription<N::HeaderResponse>, Error> {
        info!("eth_subscribe called");
        let result = self
            .retry_with_total_timeout(
                move |provider| async move { provider.subscribe_blocks().await },
                true,
            )
            .await;
        if let Err(e) = &result {
            error!(error = %e, "eth_subscribe failed");
        }
        result
    }

    /// Execute `operation` with exponential backoff and a total timeout.
    ///
    /// Wraps the retry logic with `tokio::time::timeout(self.max_timeout, ...)` so
    /// the entire operation (including time spent inside the RPC call) cannot exceed
    /// `max_timeout`.
    ///
    /// If the timeout is exceeded and fallback providers are available, it will
    /// attempt to use each fallback provider in sequence.
    ///
    /// If `require_pubsub` is true, providers that don't support pubsub will be skipped.
    ///
    /// # Errors
    /// <a name="retry-errors"></a>
    ///
    /// - Returns [`RpcError<TransportErrorKind>`] with message "total operation timeout exceeded
    ///   and all fallback providers failed" if the overall timeout elapses and no fallback
    ///   providers succeed.
    /// - Returns [`RpcError::Transport(TransportErrorKind::PubsubUnavailable)`] if `require_pubsub`
    ///   is true and all providers don't support pubsub.
    /// - Propagates any [`RpcError<TransportErrorKind>`] from the underlying retries.
    async fn retry_with_total_timeout<T: Debug, F, Fut>(
        &self,
        operation: F,
        require_pubsub: bool,
    ) -> Result<T, Error>
    where
        F: Fn(RootProvider<N>) -> Fut,
        Fut: Future<Output = Result<T, RpcError<TransportErrorKind>>>,
    {
        let mut providers = self.providers.iter();
        let primary = providers.next().expect("should have primary provider");

        let result = self.try_provider_with_timeout(primary, &operation).await;

        if result.is_ok() {
            return result;
        }

        let mut last_error = result.unwrap_err();

        let num_providers = self.providers.len();
        if num_providers > 1 {
            info!("Primary provider failed, trying fallback provider(s)");
        }

        // This loop starts at index 1 automatically
        for (idx, provider) in providers.enumerate() {
            let fallback_num = idx + 1;
            if require_pubsub && !Self::supports_pubsub(provider) {
                info!("Fallback provider {} doesn't support pubsub, skipping", fallback_num);
                continue;
            }
            info!("Attempting fallback provider {}/{}", fallback_num, num_providers - 1);

            match self.try_provider_with_timeout(provider, &operation).await {
                Ok(value) => {
                    info!(provider_num = fallback_num, "Fallback provider succeeded");
                    return Ok(value);
                }
                Err(e) => {
                    error!(provider_num = fallback_num, err = %e, "Fallback provider failed");
                    last_error = e;
                }
            }
        }

        // Return the last error encountered
        error!("All providers failed or timed out - returning the last providers attempt's error");
        Err(last_error)
    }

    /// Try executing an operation with a specific provider with retry and timeout.
    async fn try_provider_with_timeout<T, F, Fut>(
        &self,
        provider: &RootProvider<N>,
        operation: F,
    ) -> Result<T, Error>
    where
        F: Fn(RootProvider<N>) -> Fut,
        Fut: Future<Output = Result<T, RpcError<TransportErrorKind>>>,
    {
        let retry_strategy = ExponentialBuilder::default()
            .with_max_times(self.max_retries)
            .with_min_delay(self.min_delay);

        timeout(
            self.max_timeout,
            (|| operation(provider.clone()))
                .retry(retry_strategy)
                .notify(|err: &RpcError<TransportErrorKind>, dur: Duration| {
                    info!(error = %err, "RPC error retrying after {:?}", dur);
                })
                .sleep(tokio::time::sleep),
        )
        .await
        .map_err(Error::from)?
        .map_err(Error::from)
    }

    /// Check if a provider supports pubsub
    fn supports_pubsub(provider: &RootProvider<N>) -> bool {
        provider.client().pubsub_frontend().is_some()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::robust_provider::RobustProviderBuilder;
    use alloy::{
        consensus::BlockHeader,
        providers::{ProviderBuilder, WsConnect, ext::AnvilApi},
    };
    use alloy_node_bindings::Anvil;
    use std::sync::atomic::{AtomicUsize, Ordering};
    use tokio::time::sleep;

    fn test_provider(timeout: u64, max_retries: usize, min_delay: u64) -> RobustProvider {
        RobustProvider {
            providers: vec![RootProvider::new_http("http://localhost:8545".parse().unwrap())],
            max_timeout: Duration::from_millis(timeout),
            max_retries,
            min_delay: Duration::from_millis(min_delay),
        }
    }

    #[tokio::test]
    async fn test_retry_with_timeout_succeeds_on_first_attempt() {
        let provider = test_provider(100, 3, 10);

        let call_count = AtomicUsize::new(0);

        let result = provider
            .retry_with_total_timeout(
                |_| async {
                    call_count.fetch_add(1, Ordering::SeqCst);
                    let count = call_count.load(Ordering::SeqCst);
                    Ok(count)
                },
                false,
            )
            .await;

        assert!(matches!(result, Ok(1)));
    }

    #[tokio::test]
    async fn test_retry_with_timeout_retries_on_error() {
        let provider = test_provider(100, 3, 10);

        let call_count = AtomicUsize::new(0);

        let result = provider
            .retry_with_total_timeout(
                |_| async {
                    call_count.fetch_add(1, Ordering::SeqCst);
                    let count = call_count.load(Ordering::SeqCst);
                    match count {
                        3 => Ok(count),
                        _ => Err(TransportErrorKind::BackendGone.into()),
                    }
                },
                false,
            )
            .await;

        assert!(matches!(result, Ok(3)));
    }

    #[tokio::test]
    async fn test_retry_with_timeout_fails_after_max_retries() {
        let provider = test_provider(100, 2, 10);

        let call_count = AtomicUsize::new(0);

        let result: Result<(), Error> = provider
            .retry_with_total_timeout(
                |_| async {
                    call_count.fetch_add(1, Ordering::SeqCst);
                    Err(TransportErrorKind::BackendGone.into())
                },
                false,
            )
            .await;

        assert!(matches!(result, Err(Error::RpcError(_))));
        assert_eq!(call_count.load(Ordering::SeqCst), 3);
    }

    #[tokio::test]
    async fn test_retry_with_timeout_respects_max_timeout() {
        let max_timeout = 50;
        let provider = test_provider(max_timeout, 10, 1);

        let result = provider
            .retry_with_total_timeout(
                move |_provider| async move {
                    sleep(Duration::from_millis(max_timeout + 10)).await;
                    Ok(42)
                },
                false,
            )
            .await;

        assert!(matches!(result, Err(Error::Timeout)));
    }

    #[tokio::test]
    #[ignore = "Either anvil or the failover for subscription is flaky so best to ignore for now"]
    async fn test_subscribe_fails_causes_backup_to_be_used() -> anyhow::Result<()> {
        let anvil_1 = Anvil::new().try_spawn()?;

        let ws_provider_1 =
            ProviderBuilder::new().connect(anvil_1.ws_endpoint_url().as_str()).await?;

        let anvil_2 = Anvil::new().try_spawn()?;

        let ws_provider_2 =
            ProviderBuilder::new().connect(anvil_2.ws_endpoint_url().as_str()).await?;

        let robust = RobustProviderBuilder::fragile(ws_provider_1.clone())
            .fallback(ws_provider_2.clone())
            .max_timeout(Duration::from_secs(1))
            .build()
            .await?;

        drop(anvil_1);

        let mut subscription = robust.subscribe_blocks().await?;

        ws_provider_2.anvil_mine(Some(2), None).await?;

        assert_eq!(1, subscription.recv().await?.number());
        assert_eq!(2, subscription.recv().await?.number());
        assert!(subscription.is_empty());

        Ok(())
    }

    #[tokio::test]
    async fn test_subscribe_fails_when_all_providers_lack_pubsub() -> anyhow::Result<()> {
        let anvil = Anvil::new().try_spawn()?;

        let http_provider = ProviderBuilder::new().connect_http(anvil.endpoint_url());

        let robust = RobustProviderBuilder::new(http_provider.clone())
            .fallback(http_provider)
            .max_timeout(Duration::from_secs(5))
            .min_delay(Duration::from_millis(100))
            .build()
            .await?;

        let result = robust.subscribe_blocks().await.unwrap_err();

        match result {
            Error::RpcError(e) => {
                assert!(matches!(
                    e.as_ref(),
                    RpcError::Transport(TransportErrorKind::PubsubUnavailable)
                ));
            }
            other => panic!("Expected PubsubUnavailable error type, got: {other:?}"),
        }

        Ok(())
    }

    #[tokio::test]
    async fn test_subscribe_succeeds_if_primary_provider_lacks_pubsub_but_fallback_supports_it()
    -> anyhow::Result<()> {
        let anvil = Anvil::new().try_spawn()?;

        let http_provider = ProviderBuilder::new().connect_http(anvil.endpoint_url());
        let ws_provider = ProviderBuilder::new()
            .connect_ws(WsConnect::new(anvil.ws_endpoint_url().as_str()))
            .await?;

        let robust = RobustProviderBuilder::fragile(http_provider)
            .fallback(ws_provider)
            .max_timeout(Duration::from_secs(5))
            .build()
            .await?;

        let result = robust.subscribe_blocks().await;
        assert!(result.is_ok());

        Ok(())
    }

    #[tokio::test]
    async fn test_ws_fails_http_fallback_returns_primary_error() -> anyhow::Result<()> {
        let anvil_1 = Anvil::new().try_spawn()?;

        let ws_provider =
            ProviderBuilder::new().connect(anvil_1.ws_endpoint_url().as_str()).await?;

        let anvil_2 = Anvil::new().try_spawn()?;
        let http_provider = ProviderBuilder::new().connect_http(anvil_2.endpoint_url());

        let robust = RobustProviderBuilder::fragile(ws_provider.clone())
            .fallback(http_provider)
            .max_timeout(Duration::from_millis(500))
            .build()
            .await?;

        // force ws_provider to fail and return BackendGone
        drop(anvil_1);

        let err = robust.subscribe_blocks().await.unwrap_err();

        // The error should be either a Timeout or BackendGone from the primary WS provider,
        // NOT a PubsubUnavailable error (which would indicate HTTP fallback was attempted)
        match err {
            Error::Timeout => {}
            Error::RpcError(e) => {
                assert!(matches!(e.as_ref(), RpcError::Transport(TransportErrorKind::BackendGone)));
            }
            Error::BlockNotFound(id) => panic!("Unexpected error type: BlockNotFound({id})"),
        }

        Ok(())
    }
}
