electrum_client/

raw_client.rs

// SPDX-License-Identifier: MIT OR Apache-2.0

//! Raw client
//!
//! This module contains the definition of the raw client that wraps the transport method

use std::borrow::Borrow;
use std::collections::{BTreeMap, HashMap, VecDeque};
use std::io::{BufRead, BufReader, Read, Write};
use std::mem::drop;
use std::net::{TcpStream, ToSocketAddrs};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::mpsc::{channel, Receiver, Sender};
use std::sync::{Arc, Mutex, TryLockError};
use std::time::Duration;

#[allow(unused_imports)]
use log::{debug, error, info, trace, warn};

use bitcoin::consensus::encode::deserialize;
use bitcoin::hex::{DisplayHex, FromHex};
use bitcoin::{Script, Txid};

#[cfg(feature = "openssl")]
use openssl::ssl::{SslConnector, SslMethod, SslStream, SslVerifyMode};

#[cfg(any(feature = "rustls", feature = "rustls-ring"))]
#[allow(unused_imports)]
use rustls::{
    pki_types::ServerName,
    pki_types::{Der, TrustAnchor},
    ClientConfig, ClientConnection, RootCertStore, StreamOwned,
};

#[cfg(feature = "proxy")]
use crate::socks::{Socks5Stream, TargetAddr, ToTargetAddr};

use crate::stream::ClonableStream;

use crate::api::ElectrumApi;
use crate::batch::Batch;
use crate::config::AuthProvider;
use crate::types::*;

/// Client name sent to the server during protocol version negotiation.
pub const CLIENT_NAME: &str = "";

/// Minimum protocol version supported by this client.
pub const PROTOCOL_VERSION_MIN: &str = "1.4";

/// Maximum protocol version supported by this client.
pub const PROTOCOL_VERSION_MAX: &str = "1.6";

/// Checks if a protocol version string is at least the specified major.minor version.
fn is_protocol_version_at_least(version: &str, major: u32, minor: u32) -> bool {
    let mut parts = version.split('.');
    let v_major = parts.next().and_then(|s| s.parse::<u32>().ok());
    let v_minor = parts.next().and_then(|s| s.parse::<u32>().ok());
    match (v_major, v_minor) {
        (Some(v_major), Some(v_minor)) => v_major > major || (v_major == major && v_minor >= minor),
        _ => false,
    }
}

macro_rules! impl_batch_call {
    ( $self:expr, $data:expr, $call:ident ) => {{
        impl_batch_call!($self, $data, $call, )
    }};

    ( $self:expr, $data:expr, $call:ident, apply_deref ) => {{
        impl_batch_call!($self, $data, $call, *)
    }};

    ( $self:expr, $data:expr, $call:ident, $($apply_deref:tt)? ) => {{
        let mut batch = Batch::default();
        for i in $data {
            batch.$call($($apply_deref)* i.borrow());
        }

        let resp = $self.batch_call(&batch)?;
        let mut answer = Vec::new();

        for x in resp {
            answer.push(serde_json::from_value(x)?);
        }

        Ok(answer)
    }};
}

/// A trait for [`ToSocketAddrs`](https://doc.rust-lang.org/std/net/trait.ToSocketAddrs.html) that
/// can also be turned into a domain. Used when an SSL client needs to validate the server's
/// certificate.
pub trait ToSocketAddrsDomain: ToSocketAddrs {
    /// Returns the domain, if present
    fn domain(&self) -> Option<&str> {
        None
    }
}

impl ToSocketAddrsDomain for &str {
    fn domain(&self) -> Option<&str> {
        self.split(':').next()
    }
}

impl ToSocketAddrsDomain for (&str, u16) {
    fn domain(&self) -> Option<&str> {
        self.0.domain()
    }
}

#[cfg(feature = "proxy")]
impl ToSocketAddrsDomain for TargetAddr {
    fn domain(&self) -> Option<&str> {
        match self {
            TargetAddr::Ip(_) => None,
            TargetAddr::Domain(domain, _) => Some(domain.as_str()),
        }
    }
}

macro_rules! impl_to_socket_addrs_domain {
    ( $ty:ty ) => {
        impl ToSocketAddrsDomain for $ty {}
    };
}

impl_to_socket_addrs_domain!(std::net::SocketAddr);
impl_to_socket_addrs_domain!(std::net::SocketAddrV4);
impl_to_socket_addrs_domain!(std::net::SocketAddrV6);
impl_to_socket_addrs_domain!((std::net::IpAddr, u16));
impl_to_socket_addrs_domain!((std::net::Ipv4Addr, u16));
impl_to_socket_addrs_domain!((std::net::Ipv6Addr, u16));

/// Instance of an Electrum client
///
/// A [`RawClient`] maintains a constant connection with an Electrum server and exposes methods to
/// interact with it. It can also subscribe and receive notifications from the server about new
/// blocks or activity on a specific *scriptPubKey*.
///
/// The [`RawClient`] is modeled in such a way that allows the external caller to have full control over
/// its functionality: no threads or tasks are spawned internally to monitor the state of the
/// connection.
///
/// More transport methods can be used by manually creating an instance of this struct with an
/// arbitrary `S` type.
pub struct RawClient<S>
where
    S: Read + Write,
{
    stream: Mutex<ClonableStream<S>>,
    buf_reader: Mutex<BufReader<ClonableStream<S>>>,

    last_id: AtomicUsize,
    waiting_map: Mutex<HashMap<usize, Sender<ChannelMessage>>>,

    headers: Mutex<VecDeque<RawHeaderNotification>>,
    script_notifications: Mutex<HashMap<ScriptHash, VecDeque<ScriptStatus>>>,

    /// The protocol version negotiated with the server via `server.version`.
    protocol_version: Mutex<Option<String>>,

    /// Optional authorization provider for dynamic token injection (e.g., JWT).
    auth_provider: Option<AuthProvider>,

    calls: AtomicUsize,
}

// Custom Debug impl because AuthProvider doesn't implement Debug
impl<S> std::fmt::Debug for RawClient<S>
where
    S: Read + Write,
{
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("RawClient")
            .field("stream", &"<stream>")
            .field("buf_reader", &"<buf_reader>")
            .field("last_id", &self.last_id)
            .field("waiting_map", &self.waiting_map)
            .field("headers", &self.headers)
            .field("script_notifications", &self.script_notifications)
            .field(
                "auth_provider",
                &self.auth_provider.as_ref().map(|_| "<provider>"),
            )
            .finish()
    }
}

impl<S> From<S> for RawClient<S>
where
    S: Read + Write,
{
    fn from(stream: S) -> Self {
        let stream: ClonableStream<_> = stream.into();

        Self {
            buf_reader: Mutex::new(BufReader::new(stream.clone())),
            stream: Mutex::new(stream),

            last_id: AtomicUsize::new(0),
            waiting_map: Mutex::new(HashMap::new()),

            headers: Mutex::new(VecDeque::new()),
            script_notifications: Mutex::new(HashMap::new()),

            protocol_version: Mutex::new(None),

            auth_provider: None,

            calls: AtomicUsize::new(0),
        }
    }
}

/// Transport type used to establish a plaintext TCP connection with the server
pub type ElectrumPlaintextStream = TcpStream;
impl RawClient<ElectrumPlaintextStream> {
    /// Creates a new plaintext client and tries to connect to `socket_addr`.
    ///
    /// Automatically negotiates the protocol version with the server using
    /// `server.version` as required by the Electrum protocol.
    pub fn new<A: ToSocketAddrs>(
        socket_addrs: A,
        timeout: Option<Duration>,
        auth_provider: Option<AuthProvider>,
    ) -> Result<Self, Error> {
        let stream = match timeout {
            Some(timeout) => {
                let stream = connect_with_total_timeout(socket_addrs, timeout)?;
                stream.set_read_timeout(Some(timeout))?;
                stream.set_write_timeout(Some(timeout))?;
                stream
            }
            None => TcpStream::connect(socket_addrs)?,
        };

        let client = Self::from(stream)
            .with_auth(auth_provider)
            .negotiate_protocol_version()?;

        Ok(client)
    }
}

fn connect_with_total_timeout<A: ToSocketAddrs>(
    socket_addrs: A,
    mut timeout: Duration,
) -> Result<TcpStream, Error> {
    // Use the same algorithm as curl: 1/2 on the first host, 1/4 on the second one, etc.
    // https://curl.se/mail/lib-2014-11/0164.html

    let mut errors = Vec::new();

    let addrs = socket_addrs
        .to_socket_addrs()?
        .enumerate()
        .collect::<Vec<_>>();
    for (index, addr) in &addrs {
        if *index < addrs.len() - 1 {
            timeout = timeout.div_f32(2.0);
        }

        info!(
            "Trying to connect to {} (attempt {}/{}) with timeout {:?}",
            addr,
            index + 1,
            addrs.len(),
            timeout
        );
        match TcpStream::connect_timeout(addr, timeout) {
            Ok(socket) => return Ok(socket),
            Err(e) => {
                warn!("Connection error: {:?}", e);
                errors.push(e.into());
            }
        }
    }

    Err(Error::AllAttemptsErrored(errors))
}

#[cfg(feature = "openssl")]
/// Transport type used to establish an OpenSSL TLS encrypted/authenticated connection with the server
pub type ElectrumSslStream = SslStream<TcpStream>;
#[cfg(feature = "openssl")]
impl RawClient<ElectrumSslStream> {
    /// Creates a new SSL client and tries to connect to `socket_addr`. Optionally, if
    /// `validate_domain` is `true`, validate the server's certificate.
    pub fn new_ssl<A: ToSocketAddrsDomain + Clone>(
        socket_addrs: A,
        validate_domain: bool,
        timeout: Option<Duration>,
        auth_provider: Option<AuthProvider>,
    ) -> Result<Self, Error> {
        debug!(
            "new_ssl socket_addrs.domain():{:?} validate_domain:{} timeout:{:?}",
            socket_addrs.domain(),
            validate_domain,
            timeout
        );
        if validate_domain {
            socket_addrs.domain().ok_or(Error::MissingDomain)?;
        }
        match timeout {
            Some(timeout) => {
                let stream = connect_with_total_timeout(socket_addrs.clone(), timeout)?;
                stream.set_read_timeout(Some(timeout))?;
                stream.set_write_timeout(Some(timeout))?;
                Self::new_ssl_from_stream(socket_addrs, validate_domain, stream, auth_provider)
            }
            None => {
                let stream = TcpStream::connect(socket_addrs.clone())?;
                Self::new_ssl_from_stream(socket_addrs, validate_domain, stream, auth_provider)
            }
        }
    }

    /// Create a new SSL client using an existing TcpStream
    pub fn new_ssl_from_stream<A: ToSocketAddrsDomain>(
        socket_addrs: A,
        validate_domain: bool,
        stream: TcpStream,
        auth_provider: Option<AuthProvider>,
    ) -> Result<Self, Error> {
        let mut builder =
            SslConnector::builder(SslMethod::tls()).map_err(Error::InvalidSslMethod)?;

        // TODO: support for certificate pinning
        if validate_domain {
            socket_addrs.domain().ok_or(Error::MissingDomain)?;
        } else {
            builder.set_verify(SslVerifyMode::NONE);
        }
        let connector = builder.build();

        let domain = socket_addrs.domain().unwrap_or("NONE").to_string();

        let stream = connector
            .connect(&domain, stream)
            .map_err(Error::SslHandshakeError)?;

        let client = Self::from(stream)
            .with_auth(auth_provider)
            .negotiate_protocol_version()?;

        Ok(client)
    }
}

#[cfg(any(feature = "rustls", feature = "rustls-ring"))]
#[allow(unused)]
mod danger {
    use crate::raw_client::ServerName;
    use rustls::client::danger::{HandshakeSignatureValid, ServerCertVerified};
    use rustls::crypto::CryptoProvider;
    use rustls::pki_types::{CertificateDer, UnixTime};
    use rustls::DigitallySignedStruct;

    #[derive(Debug)]
    pub struct NoCertificateVerification(CryptoProvider);

    impl NoCertificateVerification {
        pub fn new(provider: CryptoProvider) -> Self {
            Self(provider)
        }
    }

    impl rustls::client::danger::ServerCertVerifier for NoCertificateVerification {
        fn verify_server_cert(
            &self,
            _end_entity: &CertificateDer<'_>,
            _intermediates: &[CertificateDer<'_>],
            _server_name: &ServerName<'_>,
            _ocsp: &[u8],
            _now: UnixTime,
        ) -> Result<ServerCertVerified, rustls::Error> {
            Ok(ServerCertVerified::assertion())
        }

        fn verify_tls12_signature(
            &self,
            _message: &[u8],
            _cert: &CertificateDer<'_>,
            _dss: &DigitallySignedStruct,
        ) -> Result<HandshakeSignatureValid, rustls::Error> {
            Ok(HandshakeSignatureValid::assertion())
        }

        fn verify_tls13_signature(
            &self,
            _message: &[u8],
            _cert: &CertificateDer<'_>,
            _dss: &DigitallySignedStruct,
        ) -> Result<HandshakeSignatureValid, rustls::Error> {
            Ok(HandshakeSignatureValid::assertion())
        }

        fn supported_verify_schemes(&self) -> Vec<rustls::SignatureScheme> {
            self.0.signature_verification_algorithms.supported_schemes()
        }
    }
}

#[cfg(all(
    any(feature = "rustls", feature = "rustls-ring"),
    not(feature = "openssl")
))]
/// Transport type used to establish a Rustls TLS encrypted/authenticated connection with the server
pub type ElectrumSslStream = StreamOwned<ClientConnection, TcpStream>;
#[cfg(all(
    any(feature = "rustls", feature = "rustls-ring"),
    not(feature = "openssl")
))]
impl RawClient<ElectrumSslStream> {
    /// Creates a new SSL client and tries to connect to `socket_addr`. Optionally, if
    /// `validate_domain` is `true`, validate the server's certificate.
    pub fn new_ssl<A: ToSocketAddrsDomain + Clone>(
        socket_addrs: A,
        validate_domain: bool,
        timeout: Option<Duration>,
        auth_provider: Option<AuthProvider>,
    ) -> Result<Self, Error> {
        debug!(
            "new_ssl socket_addrs.domain():{:?} validate_domain:{} timeout:{:?}",
            socket_addrs.domain(),
            validate_domain,
            timeout
        );

        if validate_domain {
            socket_addrs.domain().ok_or(Error::MissingDomain)?;
        }

        match timeout {
            Some(timeout) => {
                let stream = connect_with_total_timeout(socket_addrs.clone(), timeout)?;
                stream.set_read_timeout(Some(timeout))?;
                stream.set_write_timeout(Some(timeout))?;
                Self::new_ssl_from_stream(socket_addrs, validate_domain, stream, auth_provider)
            }
            None => {
                let stream = TcpStream::connect(socket_addrs.clone())?;
                Self::new_ssl_from_stream(socket_addrs, validate_domain, stream, auth_provider)
            }
        }
    }

    /// Create a new SSL client using an existing TcpStream
    pub fn new_ssl_from_stream<A: ToSocketAddrsDomain>(
        socket_addr: A,
        validate_domain: bool,
        tcp_stream: TcpStream,
        auth_provider: Option<AuthProvider>,
    ) -> Result<Self, Error> {
        use std::convert::TryFrom;

        if rustls::crypto::CryptoProvider::get_default().is_none() {
            // We install a crypto provider depending on the set feature.
            #[cfg(all(feature = "rustls", not(feature = "rustls-ring")))]
            rustls::crypto::CryptoProvider::install_default(
                rustls::crypto::aws_lc_rs::default_provider(),
            )
            .map_err(|_| {
                Error::CouldNotCreateConnection(rustls::Error::General(
                    "Failed to install CryptoProvider".to_string(),
                ))
            })?;

            #[cfg(feature = "rustls-ring")]
            rustls::crypto::CryptoProvider::install_default(
                rustls::crypto::ring::default_provider(),
            )
            .map_err(|_| {
                Error::CouldNotCreateConnection(rustls::Error::General(
                    "Failed to install CryptoProvider".to_string(),
                ))
            })?;
        }

        let builder = ClientConfig::builder();

        let config = if validate_domain {
            socket_addr.domain().ok_or(Error::MissingDomain)?;

            let store = webpki_roots::TLS_SERVER_ROOTS
                .iter()
                .map(|t| TrustAnchor {
                    subject: Der::from_slice(t.subject),
                    subject_public_key_info: Der::from_slice(t.spki),
                    name_constraints: t.name_constraints.map(Der::from_slice),
                })
                .collect::<RootCertStore>();

            // TODO: cert pinning
            builder.with_root_certificates(store).with_no_client_auth()
        } else {
            builder
                .dangerous()
                .with_custom_certificate_verifier(std::sync::Arc::new(
                    #[cfg(all(feature = "rustls", not(feature = "rustls-ring")))]
                    danger::NoCertificateVerification::new(rustls::crypto::aws_lc_rs::default_provider()),
                    #[cfg(feature = "rustls-ring")]
                    danger::NoCertificateVerification::new(rustls::crypto::ring::default_provider()),
                ))
                .with_no_client_auth()
        };

        let domain = socket_addr.domain().unwrap_or("NONE").to_string();
        let session = ClientConnection::new(
            std::sync::Arc::new(config),
            ServerName::try_from(domain.clone())
                .map_err(|_| Error::InvalidDNSNameError(domain.clone()))?,
        )
        .map_err(Error::CouldNotCreateConnection)?;
        let stream = StreamOwned::new(session, tcp_stream);

        let client = Self::from(stream)
            .with_auth(auth_provider)
            .negotiate_protocol_version()?;

        Ok(client)
    }
}

#[cfg(feature = "proxy")]
/// Transport type used to establish a connection to a server through a socks proxy
pub type ElectrumProxyStream = Socks5Stream;
#[cfg(feature = "proxy")]
impl RawClient<ElectrumProxyStream> {
    /// Creates a new socks client and tries to connect to `target_addr` using `proxy_addr` as a
    /// socks proxy server. The DNS resolution of `target_addr`, if required, is done
    /// through the proxy. This allows to specify, for instance, `.onion` addresses.
    pub fn new_proxy<T: ToTargetAddr>(
        target_addr: T,
        proxy: &crate::Socks5Config,
        timeout: Option<Duration>,
        auth_provider: Option<AuthProvider>,
    ) -> Result<Self, Error> {
        let mut stream = match proxy.credentials.as_ref() {
            Some(cred) => Socks5Stream::connect_with_password(
                &proxy.addr,
                target_addr,
                &cred.username,
                &cred.password,
                timeout,
            )?,
            None => Socks5Stream::connect(&proxy.addr, target_addr, timeout)?,
        };
        stream.get_mut().set_read_timeout(timeout)?;
        stream.get_mut().set_write_timeout(timeout)?;

        let client = Self::from(stream)
            .with_auth(auth_provider)
            .negotiate_protocol_version()?;

        Ok(client)
    }

    #[cfg(all(
        any(feature = "openssl", feature = "rustls", feature = "rustls-ring",),
        feature = "proxy",
    ))]
    /// Creates a new TLS client that connects to `target_addr` using `proxy_addr` as a socks proxy
    /// server. The DNS resolution of `target_addr`, if required, is done through the proxy. This
    /// allows to specify, for instance, `.onion` addresses.
    pub fn new_proxy_ssl<T: ToTargetAddr>(
        target_addr: T,
        validate_domain: bool,
        proxy: &crate::Socks5Config,
        timeout: Option<Duration>,
        auth_provider: Option<AuthProvider>,
    ) -> Result<RawClient<ElectrumSslStream>, Error> {
        let target = target_addr.to_target_addr()?;

        let mut stream = match proxy.credentials.as_ref() {
            Some(cred) => Socks5Stream::connect_with_password(
                &proxy.addr,
                target_addr,
                &cred.username,
                &cred.password,
                timeout,
            )?,
            None => Socks5Stream::connect(&proxy.addr, target.clone(), timeout)?,
        };

        stream.get_mut().set_read_timeout(timeout)?;
        stream.get_mut().set_write_timeout(timeout)?;

        RawClient::new_ssl_from_stream(target, validate_domain, stream.into_inner(), auth_provider)
    }
}

#[derive(Debug)]
enum ChannelMessage {
    Response(serde_json::Value),
    WakeUp,
    Error(Arc<std::io::Error>),
}

impl<S: Read + Write> RawClient<S> {
    // TODO: to enable this we have to find a way to allow concurrent read and writes to the
    // underlying transport struct. This can be done pretty easily for TcpStream because it can be
    // split into a "read" and a "write" object, but it's not as trivial for other types. Without
    // such thing, this causes a deadlock, because the reader thread takes a lock on the
    // `ClonableStream` before other threads can send a request to the server. They will block
    // waiting for the reader to release the mutex, but this will never happen because the server
    // didn't receive any request, so it has nothing to send back.
    //
    // pub fn reader_thread(&self) -> Result<(), Error> {
    //     self._reader_thread(None).map(|_| ())
    // }

    /// Sets the [`AuthProvider`] for this client, enabling authentication on all
    /// outgoing RPC requests.
    ///
    /// The `auth_provider` is a callback invoked before each request, allowing
    /// dynamic token strategies such as automatic JWT refresh without
    /// reconnecting the client. Passing `None` or not calling this method
    /// disables authentication.
    ///
    /// # Notes
    ///
    /// This method should be called **before** [`RawClient::negotiate_protocol_version`],
    /// as the initial `server.version` handshake also requires authentication
    /// on protected servers.
    fn with_auth(mut self, auth_provider: Option<AuthProvider>) -> Self {
        self.auth_provider = auth_provider;
        self
    }

    /// Negotiates the Electrum protocol version with the Electrum server.
    ///
    /// This sends `server.version` as the first message and stores the negotiated
    /// protocol version.
    ///
    /// As of Electrum Protocol v1.6 it's a mandatory step, see:
    /// <https://electrum-protocol.readthedocs.io/en/latest/protocol-changes.html#version-1-6>
    ///
    /// [`ClientType`]: crate::ClientType
    fn negotiate_protocol_version(self) -> Result<Self, Error> {
        let version_range = vec![
            PROTOCOL_VERSION_MIN.to_string(),
            PROTOCOL_VERSION_MAX.to_string(),
        ];
        let req = Request::new_id(
            self.last_id.fetch_add(1, Ordering::SeqCst),
            "server.version",
            vec![
                Param::String(CLIENT_NAME.to_string()),
                Param::StringVec(version_range),
            ],
        );
        let result = self.call(req)?;
        let response: ServerVersionRes = serde_json::from_value(result)?;

        *self.protocol_version.lock()? = Some(response.protocol_version);
        Ok(self)
    }

    fn _reader_thread(&self, until_message: Option<usize>) -> Result<serde_json::Value, Error> {
        let mut raw_resp = String::new();
        let resp = match self.buf_reader.try_lock() {
            Ok(mut reader) => {
                trace!(
                    "Starting reader thread with `until_message` = {:?}",
                    until_message
                );

                if let Some(until_message) = until_message {
                    // If we are trying to start a reader thread but the corresponding sender is
                    // missing from the map, exit immediately. We might have already received a
                    // response for that id, but we don't know it yet. Exiting here forces the
                    // calling code to fallback to the sender-receiver method, and it should find
                    // a message there waiting for it.
                    if self.waiting_map.lock()?.get(&until_message).is_none() {
                        return Err(Error::CouldntLockReader);
                    }
                }

                // Loop over every message
                loop {
                    raw_resp.clear();

                    match reader.read_line(&mut raw_resp) {
                        Ok(n_bytes_read) => {
                            if n_bytes_read == 0 {
                                trace!("Reached UnexpectedEof");
                                return Err(Error::IOError(std::io::Error::new(
                                    std::io::ErrorKind::UnexpectedEof,
                                    "unexpected EOF",
                                )));
                            }
                        }
                        Err(e) => {
                            let error = Arc::new(e);
                            for (_, s) in self.waiting_map.lock().unwrap().drain() {
                                s.send(ChannelMessage::Error(error.clone()))?;
                            }
                            return Err(Error::SharedIOError(error));
                        }
                    }
                    trace!("<== {}", raw_resp);

                    let resp: serde_json::Value = serde_json::from_str(&raw_resp)?;

                    // Normally there is and id, but it's missing for spontaneous notifications
                    // from the server
                    let resp_id = resp["id"]
                        .as_str()
                        .and_then(|s| s.parse().ok())
                        .or_else(|| resp["id"].as_u64().map(|i| i as usize));
                    match resp_id {
                        Some(resp_id) if until_message == Some(resp_id) => {
                            // We have a valid id and it's exactly the one we were waiting for!
                            trace!(
                                "Reader thread {} received a response for its request",
                                resp_id
                            );

                            // Remove ourselves from the "waiting map"
                            let mut map = self.waiting_map.lock()?;
                            map.remove(&resp_id);

                            // If the map is not empty, we select a random thread to become the
                            // new reader thread.
                            if let Some(err) = map.values().find_map(|sender| {
                                sender
                                    .send(ChannelMessage::WakeUp)
                                    .map_err(|err| {
                                        warn!("Unable to wake up a thread, trying some other");
                                        err
                                    })
                                    .err()
                            }) {
                                error!("All the threads has failed, giving up");
                                return Err(err)?;
                            }

                            break Ok(resp);
                        }
                        Some(resp_id) => {
                            // We have an id, but it's not our response. Notify the thread and
                            // move on
                            trace!("Reader thread received response for {}", resp_id);

                            if let Some(sender) = self.waiting_map.lock()?.remove(&resp_id) {
                                sender.send(ChannelMessage::Response(resp))?;
                            } else {
                                warn!("Missing listener for {}", resp_id);
                            }
                        }
                        None => {
                            // No id, that's probably a notification.
                            let mut resp = resp;

                            if let Some(method) = resp["method"].take().as_str() {
                                self.handle_notification(method, resp["params"].take())?;
                            } else {
                                warn!("Unexpected response: {:?}", resp);
                            }
                        }
                    }
                }
            }
            Err(TryLockError::WouldBlock) => {
                // If we "WouldBlock" here it means that there's already a reader thread
                // running somewhere.
                Err(Error::CouldntLockReader)
            }
            Err(TryLockError::Poisoned(e)) => Err(e)?,
        };

        let resp = resp?;
        if let Some(err) = resp.get("error") {
            Err(Error::Protocol(err.clone()))
        } else {
            Ok(resp)
        }
    }

    fn call(&self, req: Request) -> Result<serde_json::Value, Error> {
        // Add our listener to the map before we send the request, to make sure we don't get a
        // reply before the receiver is added
        let (sender, receiver) = channel();
        self.waiting_map.lock()?.insert(req.id, sender);

        // apply `authorization` token into `Request`, if any.
        let authorization = self
            .auth_provider
            .as_ref()
            .and_then(|auth_provider| auth_provider());

        let req = req.with_auth(authorization);

        let mut raw = serde_json::to_vec(&req)?;
        trace!("==> {}", String::from_utf8_lossy(&raw));

        raw.extend_from_slice(b"\n");
        let mut stream = self.stream.lock()?;
        stream.write_all(&raw)?;
        stream.flush()?;
        drop(stream); // release the lock

        self.increment_calls();

        let mut resp = match self.recv(&receiver, req.id) {
            Ok(resp) => resp,
            e @ Err(_) => {
                // In case of error our sender could still be left in the map, depending on where
                // the error happened. Just in case, try to remove it here
                self.waiting_map.lock()?.remove(&req.id);
                return e;
            }
        };
        Ok(resp["result"].take())
    }

    fn recv(
        &self,
        receiver: &Receiver<ChannelMessage>,
        req_id: usize,
    ) -> Result<serde_json::Value, Error> {
        loop {
            // Try to take the lock on the reader. If we manage to do so, we'll become the reader
            // thread until we get our reponse
            match self._reader_thread(Some(req_id)) {
                Ok(response) => break Ok(response),
                Err(Error::CouldntLockReader) => {
                    match receiver.recv()? {
                        // Received our response, returning it
                        ChannelMessage::Response(received) => break Ok(received),
                        ChannelMessage::WakeUp => {
                            // We have been woken up, this means that we should try becoming the
                            // reader thread ourselves
                            trace!("WakeUp for {}", req_id);

                            continue;
                        }
                        ChannelMessage::Error(e) => {
                            warn!("Received ChannelMessage::Error");

                            break Err(Error::SharedIOError(e));
                        }
                    }
                }
                e @ Err(_) => break e,
            }
        }
    }

    fn handle_notification(&self, method: &str, result: serde_json::Value) -> Result<(), Error> {
        match method {
            "blockchain.headers.subscribe" => self.headers.lock()?.append(
                &mut serde_json::from_value::<Vec<RawHeaderNotification>>(result)?
                    .into_iter()
                    .collect(),
            ),
            "blockchain.scripthash.subscribe" => {
                let unserialized: ScriptNotification = serde_json::from_value(result)?;
                let mut script_notifications = self.script_notifications.lock()?;

                let queue = script_notifications
                    .get_mut(&unserialized.scripthash)
                    .ok_or(Error::NotSubscribed(unserialized.scripthash))?;

                queue.push_back(unserialized.status);
            }
            _ => info!("received unknown notification for method `{}`", method),
        }

        Ok(())
    }

    pub(crate) fn internal_raw_call_with_vec(
        &self,
        method_name: &str,
        params: Vec<Param>,
    ) -> Result<serde_json::Value, Error> {
        let req = Request::new_id(
            self.last_id.fetch_add(1, Ordering::SeqCst),
            method_name,
            params,
        );
        let result = self.call(req)?;

        Ok(result)
    }

    #[inline]
    fn increment_calls(&self) {
        self.calls.fetch_add(1, Ordering::SeqCst);
    }
}

impl<T: Read + Write> ElectrumApi for RawClient<T> {
    fn raw_call(
        &self,
        method_name: &str,
        params: impl IntoIterator<Item = Param>,
    ) -> Result<serde_json::Value, Error> {
        self.internal_raw_call_with_vec(method_name, params.into_iter().collect())
    }

    fn batch_call(&self, batch: &Batch) -> Result<Vec<serde_json::Value>, Error> {
        let mut raw = Vec::new();

        let mut missing_responses = Vec::new();
        let mut answers = BTreeMap::new();

        // Add our listener to the map before we send the request

        for (idx, (method, params)) in batch.iter().enumerate() {
            let mut req = Request::new_id(
                self.last_id.fetch_add(1, Ordering::SeqCst),
                method,
                params.to_vec(),
            );

            // Although the library DOES NOT use JSON-RPC batch arrays,
            // It applies the `authorization` ONLY in the first `Request` of the `Batch`.
            //
            // JWT tokens can be 1KB+, therefore duplicating it across multiple requests adds significant overhead.
            // It assumes the server authenticates the `Batch` by the first `Request`. If a server implementation treats
            // each newline-delimited request independently, subsequently `Request`'s would be unauthenticated.
            //
            // It's a known trade-off, not a bug.
            if idx == 0 {
                // it should get the `authorization`, if there's an `auth_provider` available.
                let authorization = self
                    .auth_provider
                    .as_ref()
                    .and_then(|auth_provider| auth_provider());

                req = req.with_auth(authorization);
            }

            // Add distinct channel to each request so when we remove our request id (and sender) from the waiting_map
            // we can be sure that the response gets sent to the correct channel in self.recv
            let (sender, receiver) = channel();
            missing_responses.push((req.id, receiver));

            self.waiting_map.lock()?.insert(req.id, sender);

            raw.append(&mut serde_json::to_vec(&req)?);
            raw.extend_from_slice(b"\n");
        }

        if missing_responses.is_empty() {
            return Ok(vec![]);
        }

        trace!("==> {}", String::from_utf8_lossy(&raw));

        let mut stream = self.stream.lock()?;
        stream.write_all(&raw)?;
        stream.flush()?;
        drop(stream); // release the lock

        self.increment_calls();

        for (req_id, receiver) in missing_responses.iter() {
            match self.recv(receiver, *req_id) {
                Ok(mut resp) => answers.insert(req_id, resp["result"].take()),
                Err(e) => {
                    // In case of error our sender could still be left in the map, depending on where
                    // the error happened. Just in case, try to remove it here
                    warn!("got error for req_id {}: {:?}", req_id, e);
                    warn!("removing all waiting req of this batch");
                    let mut guard = self.waiting_map.lock()?;
                    for (req_id, _) in missing_responses.iter() {
                        guard.remove(req_id);
                    }
                    return Err(e);
                }
            };
        }

        Ok(answers.into_values().collect())
    }

    fn block_headers_subscribe_raw(&self) -> Result<RawHeaderNotification, Error> {
        let req = Request::new_id(
            self.last_id.fetch_add(1, Ordering::SeqCst),
            "blockchain.headers.subscribe",
            vec![],
        );
        let value = self.call(req)?;

        Ok(serde_json::from_value(value)?)
    }

    fn block_headers_pop_raw(&self) -> Result<Option<RawHeaderNotification>, Error> {
        Ok(self.headers.lock()?.pop_front())
    }

    fn block_header_raw(&self, height: usize) -> Result<Vec<u8>, Error> {
        let req = Request::new_id(
            self.last_id.fetch_add(1, Ordering::SeqCst),
            "blockchain.block.header",
            vec![Param::Usize(height)],
        );
        let result = self.call(req)?;

        Ok(Vec::<u8>::from_hex(
            result
                .as_str()
                .ok_or_else(|| Error::InvalidResponse(result.clone()))?,
        )?)
    }

    fn block_headers(&self, start_height: usize, count: usize) -> Result<GetHeadersRes, Error> {
        let req = Request::new_id(
            self.last_id.fetch_add(1, Ordering::SeqCst),
            "blockchain.block.headers",
            vec![Param::Usize(start_height), Param::Usize(count)],
        );
        let result = self.call(req)?;

        // Check protocol version to determine response format
        let is_v1_6_or_later = {
            let protocol_version = self.protocol_version.lock()?;
            protocol_version
                .as_ref()
                .map(|v| is_protocol_version_at_least(v, 1, 6))
                .unwrap_or(false)
        };

        if is_v1_6_or_later {
            // v1.6+: headers field contains array of hex strings
            let mut deserialized: GetHeadersRes = serde_json::from_value(result)?;
            for header_hex in &deserialized.header_hexes {
                let header_bytes = Vec::<u8>::from_hex(header_hex)?;
                deserialized.headers.push(deserialize(&header_bytes)?);
            }
            deserialized.header_hexes.clear();
            Ok(deserialized)
        } else {
            // v1.4: hex field contains concatenated headers
            let deserialized: GetHeadersResLegacy = serde_json::from_value(result)?;
            let mut headers = Vec::new();
            for i in 0..deserialized.count {
                let (start, end) = (i * 80, (i + 1) * 80);
                headers.push(deserialize(&deserialized.raw_headers[start..end])?);
            }
            Ok(GetHeadersRes {
                max: deserialized.max,
                count: deserialized.count,
                header_hexes: Vec::new(),
                headers,
            })
        }
    }

    fn estimate_fee(&self, number: usize, mode: Option<EstimationMode>) -> Result<f64, Error> {
        let mut params = vec![Param::Usize(number)];
        if let Some(mode) = mode {
            params.push(Param::String(mode.to_string()));
        }
        let req = Request::new_id(
            self.last_id.fetch_add(1, Ordering::SeqCst),
            "blockchain.estimatefee",
            params,
        );
        let result = self.call(req)?;

        result
            .as_f64()
            .ok_or_else(|| Error::InvalidResponse(result.clone()))
    }

    fn relay_fee(&self) -> Result<f64, Error> {
        let req = Request::new_id(
            self.last_id.fetch_add(1, Ordering::SeqCst),
            "blockchain.relayfee",
            vec![],
        );
        let result = self.call(req)?;

        result
            .as_f64()
            .ok_or_else(|| Error::InvalidResponse(result.clone()))
    }

    fn script_subscribe(&self, script: &Script) -> Result<Option<ScriptStatus>, Error> {
        let script_hash = script.to_electrum_scripthash();
        let mut script_notifications = self.script_notifications.lock()?;

        if script_notifications.contains_key(&script_hash) {
            return Err(Error::AlreadySubscribed(script_hash));
        }

        script_notifications.insert(script_hash, VecDeque::new());
        drop(script_notifications);

        let req = Request::new_id(
            self.last_id.fetch_add(1, Ordering::SeqCst),
            "blockchain.scripthash.subscribe",
            vec![Param::String(script_hash.to_hex())],
        );
        let value = self.call(req)?;

        Ok(serde_json::from_value(value)?)
    }

    fn batch_script_subscribe<'s, I>(&self, scripts: I) -> Result<Vec<Option<ScriptStatus>>, Error>
    where
        I: IntoIterator + Clone,
        I::Item: Borrow<&'s Script>,
    {
        {
            let mut script_notifications = self.script_notifications.lock()?;

            for script in scripts.clone() {
                let script_hash = script.borrow().to_electrum_scripthash();
                if script_notifications.contains_key(&script_hash) {
                    return Err(Error::AlreadySubscribed(script_hash));
                }
                script_notifications.insert(script_hash, VecDeque::new());
            }
        }
        impl_batch_call!(self, scripts, script_subscribe)
    }

    fn script_unsubscribe(&self, script: &Script) -> Result<bool, Error> {
        let script_hash = script.to_electrum_scripthash();
        let mut script_notifications = self.script_notifications.lock()?;

        if !script_notifications.contains_key(&script_hash) {
            return Err(Error::NotSubscribed(script_hash));
        }

        let req = Request::new_id(
            self.last_id.fetch_add(1, Ordering::SeqCst),
            "blockchain.scripthash.unsubscribe",
            vec![Param::String(script_hash.to_hex())],
        );
        let value = self.call(req)?;
        let answer = serde_json::from_value(value)?;

        script_notifications.remove(&script_hash);

        Ok(answer)
    }

    fn script_pop(&self, script: &Script) -> Result<Option<ScriptStatus>, Error> {
        let script_hash = script.to_electrum_scripthash();

        match self.script_notifications.lock()?.get_mut(&script_hash) {
            None => Err(Error::NotSubscribed(script_hash)),
            Some(queue) => Ok(queue.pop_front()),
        }
    }

    fn script_get_balance(&self, script: &Script) -> Result<GetBalanceRes, Error> {
        let params = vec![Param::String(script.to_electrum_scripthash().to_hex())];
        let req = Request::new_id(
            self.last_id.fetch_add(1, Ordering::SeqCst),
            "blockchain.scripthash.get_balance",
            params,
        );
        let result = self.call(req)?;

        Ok(serde_json::from_value(result)?)
    }
    fn batch_script_get_balance<'s, I>(&self, scripts: I) -> Result<Vec<GetBalanceRes>, Error>
    where
        I: IntoIterator + Clone,
        I::Item: Borrow<&'s Script>,
    {
        impl_batch_call!(self, scripts, script_get_balance)
    }

    fn script_get_history(&self, script: &Script) -> Result<Vec<GetHistoryRes>, Error> {
        let params = vec![Param::String(script.to_electrum_scripthash().to_hex())];
        let req = Request::new_id(
            self.last_id.fetch_add(1, Ordering::SeqCst),
            "blockchain.scripthash.get_history",
            params,
        );
        let result = self.call(req)?;

        Ok(serde_json::from_value(result)?)
    }
    fn batch_script_get_history<'s, I>(&self, scripts: I) -> Result<Vec<Vec<GetHistoryRes>>, Error>
    where
        I: IntoIterator + Clone,
        I::Item: Borrow<&'s Script>,
    {
        impl_batch_call!(self, scripts, script_get_history)
    }

    fn script_list_unspent(&self, script: &Script) -> Result<Vec<ListUnspentRes>, Error> {
        let params = vec![Param::String(script.to_electrum_scripthash().to_hex())];
        let req = Request::new_id(
            self.last_id.fetch_add(1, Ordering::SeqCst),
            "blockchain.scripthash.listunspent",
            params,
        );
        let result = self.call(req)?;
        let mut result: Vec<ListUnspentRes> = serde_json::from_value(result)?;

        // This should not be necessary, since the protocol documentation says that the txs should
        // be "in blockchain order" (https://electrumx.readthedocs.io/en/latest/protocol-methods.html#blockchain-scripthash-listunspent).
        // However, elects seems to be ignoring this at the moment, so we'll sort again here just
        // to make sure the result is consistent.
        result.sort_unstable_by_key(|k| (k.height, k.tx_pos));
        Ok(result)
    }

    fn batch_script_list_unspent<'s, I>(
        &self,
        scripts: I,
    ) -> Result<Vec<Vec<ListUnspentRes>>, Error>
    where
        I: IntoIterator + Clone,
        I::Item: Borrow<&'s Script>,
    {
        impl_batch_call!(self, scripts, script_list_unspent)
    }

    fn transaction_get_raw(&self, txid: &Txid) -> Result<Vec<u8>, Error> {
        let params = vec![Param::String(format!("{:x}", txid))];
        let req = Request::new_id(
            self.last_id.fetch_add(1, Ordering::SeqCst),
            "blockchain.transaction.get",
            params,
        );
        let result = self.call(req)?;

        Ok(Vec::<u8>::from_hex(
            result
                .as_str()
                .ok_or_else(|| Error::InvalidResponse(result.clone()))?,
        )?)
    }

    fn batch_transaction_get_raw<'t, I>(&self, txids: I) -> Result<Vec<Vec<u8>>, Error>
    where
        I: IntoIterator + Clone,
        I::Item: Borrow<&'t Txid>,
    {
        let txs_string: Result<Vec<String>, Error> = impl_batch_call!(self, txids, transaction_get);
        txs_string?
            .iter()
            .map(|s| Ok(Vec::<u8>::from_hex(s)?))
            .collect()
    }

    fn batch_block_header_raw<'s, I>(&self, heights: I) -> Result<Vec<Vec<u8>>, Error>
    where
        I: IntoIterator + Clone,
        I::Item: Borrow<u32>,
    {
        let headers_string: Result<Vec<String>, Error> =
            impl_batch_call!(self, heights, block_header, apply_deref);
        headers_string?
            .iter()
            .map(|s| Ok(Vec::<u8>::from_hex(s)?))
            .collect()
    }

    fn batch_estimate_fee<'s, I>(&self, numbers: I) -> Result<Vec<f64>, Error>
    where
        I: IntoIterator + Clone,
        I::Item: Borrow<usize>,
    {
        let mut batch = Batch::default();
        for i in numbers {
            batch.estimate_fee(*i.borrow(), None);
        }

        let resp = self.batch_call(&batch)?;
        let mut answer = Vec::new();

        for x in resp {
            answer.push(serde_json::from_value(x)?);
        }

        Ok(answer)
    }

    fn transaction_broadcast_raw(&self, raw_tx: &[u8]) -> Result<Txid, Error> {
        let params = vec![Param::String(raw_tx.to_lower_hex_string())];
        let req = Request::new_id(
            self.last_id.fetch_add(1, Ordering::SeqCst),
            "blockchain.transaction.broadcast",
            params,
        );
        let result = self.call(req)?;

        Ok(serde_json::from_value(result)?)
    }

    fn transaction_broadcast_package_raw<Tx: AsRef<[u8]>>(
        &self,
        raw_txs: &[Tx],
    ) -> Result<BroadcastPackageRes, Error> {
        let hex_txs: Vec<String> = raw_txs
            .iter()
            .map(|tx| tx.as_ref().to_lower_hex_string())
            .collect();
        let params = vec![Param::StringVec(hex_txs)];
        let req = Request::new_id(
            self.last_id.fetch_add(1, Ordering::SeqCst),
            "blockchain.transaction.broadcast_package",
            params,
        );
        let result = self.call(req)?;

        Ok(serde_json::from_value(result)?)
    }

    fn transaction_get_merkle(&self, txid: &Txid, height: usize) -> Result<GetMerkleRes, Error> {
        let params = vec![Param::String(format!("{:x}", txid)), Param::Usize(height)];
        let req = Request::new_id(
            self.last_id.fetch_add(1, Ordering::SeqCst),
            "blockchain.transaction.get_merkle",
            params,
        );
        let result = self.call(req)?;

        Ok(serde_json::from_value(result)?)
    }

    fn batch_transaction_get_merkle<I>(
        &self,
        txids_and_heights: I,
    ) -> Result<Vec<GetMerkleRes>, Error>
    where
        I: IntoIterator + Clone,
        I::Item: Borrow<(Txid, usize)>,
    {
        impl_batch_call!(self, txids_and_heights, transaction_get_merkle)
    }

    fn txid_from_pos(&self, height: usize, tx_pos: usize) -> Result<Txid, Error> {
        let params = vec![Param::Usize(height), Param::Usize(tx_pos)];
        let req = Request::new_id(
            self.last_id.fetch_add(1, Ordering::SeqCst),
            "blockchain.transaction.id_from_pos",
            params,
        );
        let result = self.call(req)?;

        Ok(serde_json::from_value(result)?)
    }

    fn txid_from_pos_with_merkle(
        &self,
        height: usize,
        tx_pos: usize,
    ) -> Result<TxidFromPosRes, Error> {
        let params = vec![
            Param::Usize(height),
            Param::Usize(tx_pos),
            Param::Bool(true),
        ];
        let req = Request::new_id(
            self.last_id.fetch_add(1, Ordering::SeqCst),
            "blockchain.transaction.id_from_pos",
            params,
        );
        let result = self.call(req)?;

        Ok(serde_json::from_value(result)?)
    }

    fn server_features(&self) -> Result<ServerFeaturesRes, Error> {
        let req = Request::new_id(
            self.last_id.fetch_add(1, Ordering::SeqCst),
            "server.features",
            vec![],
        );
        let result = self.call(req)?;

        Ok(serde_json::from_value(result)?)
    }

    fn mempool_get_info(&self) -> Result<MempoolInfoRes, Error> {
        let req = Request::new_id(
            self.last_id.fetch_add(1, Ordering::SeqCst),
            "mempool.get_info",
            vec![],
        );
        let result = self.call(req)?;

        Ok(serde_json::from_value(result)?)
    }

    fn ping(&self) -> Result<(), Error> {
        let req = Request::new_id(
            self.last_id.fetch_add(1, Ordering::SeqCst),
            "server.ping",
            vec![],
        );
        self.call(req)?;

        Ok(())
    }

    fn calls_made(&self) -> Result<usize, Error> {
        Ok(self.calls.load(Ordering::SeqCst))
    }
}

#[cfg(test)]
mod test {
    use std::str::FromStr;

    use crate::utils;

    use super::{ElectrumSslStream, RawClient};
    use crate::api::ElectrumApi;
    use crate::config::AuthProvider;

    // it's the default live testing electrum server, if you'd like to use a custom one set it up through
    // the environment variable `TEST_ELECTRUM_SERVER`.
    //
    // here's an useful list of live servers: https://1209k.com/bitcoin-eye/ele.php.
    const DEFAULT_TEST_ELECTRUM_SERVER: &str = "fortress.qtornado.com:443";

    fn get_test_auth_client(
        authorization_provider: Option<AuthProvider>,
    ) -> RawClient<ElectrumSslStream> {
        let server = std::env::var("TEST_ELECTRUM_SERVER")
            .unwrap_or(DEFAULT_TEST_ELECTRUM_SERVER.to_owned());

        RawClient::new_ssl(&*server, false, None, authorization_provider)
            .expect("should build the `RawClient` successfully!")
    }

    fn get_test_client() -> RawClient<ElectrumSslStream> {
        let server = std::env::var("TEST_ELECTRUM_SERVER")
            .unwrap_or(DEFAULT_TEST_ELECTRUM_SERVER.to_owned());

        RawClient::new_ssl(&*server, false, None, None)
            .expect("should build the `RawClient` successfully!")
    }

    #[test]
    fn test_server_features_simple() {
        let client = get_test_client();

        let resp = client.server_features().unwrap();
        assert_eq!(
            resp.genesis_hash,
            [
                0, 0, 0, 0, 0, 25, 214, 104, 156, 8, 90, 225, 101, 131, 30, 147, 79, 247, 99, 174,
                70, 162, 166, 193, 114, 179, 241, 182, 10, 140, 226, 111
            ],
        );
        assert_eq!(resp.hash_function, Some("sha256".into()));
        assert_eq!(resp.pruning, None);
    }

    #[test]
    fn test_mempool_get_info() {
        let client = get_test_client();

        let resp = client.mempool_get_info().unwrap();
        assert!(resp.mempoolminfee >= 0.0);
        assert!(resp.minrelaytxfee >= 0.0);
        assert!(resp.incrementalrelayfee >= 0.0);
    }

    #[test]
    fn test_transaction_broadcast_package() {
        let client = get_test_client();

        // Empty package should return an error or unsuccessful response
        let resp = client.transaction_broadcast_package_raw::<Vec<u8>>(&[]);
        // The server may reject an empty package with a protocol error
        assert!(resp.is_err() || !resp.unwrap().success);
    }

    #[test]
    #[ignore = "depends on a live server"]
    fn test_batch_response_ordering() {
        // The electrum.blockstream.info:50001 node always sends back ordered responses which will make this always pass.
        // However, many servers do not, so we use one of those servers for this test.
        let client = get_test_client();
        let heights: Vec<u32> = vec![1, 4, 8, 12, 222, 6666, 12];
        let result_times = [
            1231469665, 1231470988, 1231472743, 1231474888, 1231770653, 1236456633, 1231474888,
        ];
        // Check ordering 10 times. This usually fails within 5 if ordering is incorrect.
        for _ in 0..10 {
            let results = client.batch_block_header(&heights).unwrap();
            for (index, result) in results.iter().enumerate() {
                assert_eq!(result_times[index], result.time);
            }
        }
    }

    #[test]
    fn test_estimate_fee() {
        let client = get_test_client();

        let resp = client.estimate_fee(10, None).unwrap();
        assert!(resp > 0.0);
    }

    #[test]
    fn test_block_header() {
        let client = get_test_client();

        let resp = client.block_header(0).unwrap();
        assert_eq!(resp.version, bitcoin::block::Version::ONE);
        assert_eq!(resp.time, 1231006505);
        assert_eq!(resp.nonce, 0x7c2bac1d);
    }

    #[test]
    fn test_block_header_raw() {
        let client = get_test_client();

        let resp = client.block_header_raw(0).unwrap();
        assert_eq!(
            resp,
            vec![
                1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
                0, 0, 0, 0, 0, 0, 0, 0, 59, 163, 237, 253, 122, 123, 18, 178, 122, 199, 44, 62,
                103, 118, 143, 97, 127, 200, 27, 195, 136, 138, 81, 50, 58, 159, 184, 170, 75, 30,
                94, 74, 41, 171, 95, 73, 255, 255, 0, 29, 29, 172, 43, 124
            ]
        );
    }

    #[test]
    fn test_block_headers() {
        let client = get_test_client();

        let resp = client.block_headers(0, 4).unwrap();
        assert_eq!(resp.count, 4);
        assert_eq!(resp.max, 2016);
        assert_eq!(resp.headers.len(), 4);

        assert_eq!(resp.headers[0].time, 1231006505);
    }

    #[test]
    fn test_script_get_balance() {
        use std::str::FromStr;

        let client = get_test_client();

        // Realistically nobody will ever spend from this address, so we can expect the balance to
        // increase over time
        let addr = bitcoin::Address::from_str("1CounterpartyXXXXXXXXXXXXXXXUWLpVr")
            .unwrap()
            .assume_checked();
        let resp = client.script_get_balance(&addr.script_pubkey()).unwrap();
        assert!(resp.confirmed >= 213091301265);
    }

    #[test]
    fn test_script_get_history() {
        use std::str::FromStr;

        use bitcoin::Txid;

        let client = get_test_client();

        // Mt.Gox hack address
        let addr = bitcoin::Address::from_str("1FeexV6bAHb8ybZjqQMjJrcCrHGW9sb6uF")
            .unwrap()
            .assume_checked();
        let resp = client.script_get_history(&addr.script_pubkey()).unwrap();

        assert!(resp.len() >= 328);
        assert_eq!(
            resp[0].tx_hash,
            Txid::from_str("e67a0550848b7932d7796aeea16ab0e48a5cfe81c4e8cca2c5b03e0416850114")
                .unwrap()
        );
    }

    #[test]
    fn test_script_list_unspent() {
        use bitcoin::Txid;
        use std::str::FromStr;

        let client = get_test_client();

        // Peter todd's sha256 bounty address https://bitcointalk.org/index.php?topic=293382.0
        let addr = bitcoin::Address::from_str("35Snmmy3uhaer2gTboc81ayCip4m9DT4ko")
            .unwrap()
            .assume_checked();
        let resp = client.script_list_unspent(&addr.script_pubkey()).unwrap();

        assert!(resp.len() >= 9);
        let txid = "397f12ee15f8a3d2ab25c0f6bb7d3c64d2038ca056af10dd8251b98ae0f076b0";
        let txid = Txid::from_str(txid).unwrap();
        let txs: Vec<_> = resp.iter().filter(|e| e.tx_hash == txid).collect();
        assert_eq!(txs.len(), 1);
        assert_eq!(txs[0].value, 10000000);
        assert_eq!(txs[0].height, 257674);
        assert_eq!(txs[0].tx_pos, 1);
    }

    #[test]
    fn test_batch_script_list_unspent() {
        use std::str::FromStr;

        let client = get_test_client();

        // Peter todd's sha256 bounty address https://bitcointalk.org/index.php?topic=293382.0
        let script_1 = bitcoin::Address::from_str("35Snmmy3uhaer2gTboc81ayCip4m9DT4ko")
            .unwrap()
            .assume_checked()
            .script_pubkey();

        let resp = client
            .batch_script_list_unspent(vec![script_1.as_script()])
            .unwrap();
        assert_eq!(resp.len(), 1);
        assert!(resp[0].len() >= 9);
    }

    #[test]
    fn test_batch_estimate_fee() {
        let client = get_test_client();

        let resp = client.batch_estimate_fee(vec![10, 20]).unwrap();
        assert_eq!(resp.len(), 2);
        assert!(resp[0] > 0.0);
        assert!(resp[1] > 0.0);
    }

    #[test]
    fn test_transaction_get() {
        use bitcoin::{transaction, Txid};

        let client = get_test_client();

        let resp = client
            .transaction_get(
                &Txid::from_str("cc2ca076fd04c2aeed6d02151c447ced3d09be6fb4d4ef36cb5ed4e7a3260566")
                    .unwrap(),
            )
            .unwrap();
        assert_eq!(resp.version, transaction::Version::ONE);
        assert_eq!(resp.lock_time.to_consensus_u32(), 0);
    }

    #[test]
    fn test_transaction_get_raw() {
        use bitcoin::Txid;

        let client = get_test_client();

        let resp = client
            .transaction_get_raw(
                &Txid::from_str("cc2ca076fd04c2aeed6d02151c447ced3d09be6fb4d4ef36cb5ed4e7a3260566")
                    .unwrap(),
            )
            .unwrap();
        assert_eq!(
            resp,
            vec![
                1, 0, 0, 0, 0, 1, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
                0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 255, 255, 255, 255, 84, 3, 240, 156, 9, 27, 77,
                105, 110, 101, 100, 32, 98, 121, 32, 65, 110, 116, 80, 111, 111, 108, 49, 49, 57,
                174, 0, 111, 32, 7, 77, 101, 40, 250, 190, 109, 109, 42, 177, 148, 141, 80, 179,
                217, 145, 226, 160, 130, 29, 247, 67, 88, 237, 156, 37, 83, 175, 0, 199, 166, 31,
                151, 119, 28, 160, 172, 238, 16, 110, 4, 0, 0, 0, 0, 0, 0, 0, 203, 236, 0, 128, 36,
                97, 249, 5, 255, 255, 255, 255, 3, 84, 206, 172, 42, 0, 0, 0, 0, 25, 118, 169, 20,
                17, 219, 228, 140, 198, 182, 23, 249, 198, 173, 175, 77, 158, 213, 246, 37, 177,
                199, 203, 89, 136, 172, 0, 0, 0, 0, 0, 0, 0, 0, 38, 106, 36, 170, 33, 169, 237, 46,
                87, 139, 206, 44, 166, 198, 188, 147, 89, 55, 115, 69, 216, 233, 133, 221, 95, 144,
                199, 132, 33, 255, 166, 239, 165, 235, 96, 66, 142, 105, 140, 0, 0, 0, 0, 0, 0, 0,
                0, 38, 106, 36, 185, 225, 27, 109, 47, 98, 29, 126, 195, 244, 90, 94, 202, 137,
                211, 234, 106, 41, 76, 223, 58, 4, 46, 151, 48, 9, 88, 68, 112, 161, 41, 22, 17,
                30, 44, 170, 1, 32, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
                0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0
            ]
        )
    }

    #[test]
    fn test_transaction_get_merkle() {
        use bitcoin::Txid;

        let client = get_test_client();

        let txid =
            Txid::from_str("1f7ff3c407f33eabc8bec7d2cc230948f2249ec8e591bcf6f971ca9366c8788d")
                .unwrap();
        let resp = client.transaction_get_merkle(&txid, 630000).unwrap();
        assert_eq!(resp.block_height, 630000);
        assert_eq!(resp.pos, 68);
        assert_eq!(resp.merkle.len(), 12);
        assert_eq!(
            resp.merkle[0],
            [
                34, 65, 51, 64, 49, 139, 115, 189, 185, 246, 70, 225, 168, 193, 217, 195, 47, 66,
                179, 240, 153, 24, 114, 215, 144, 196, 212, 41, 39, 155, 246, 25
            ]
        );

        // Check we can verify the merkle proof validity, but fail if we supply wrong data.
        let block_header = client.block_header(resp.block_height).unwrap();
        assert!(utils::validate_merkle_proof(
            &txid,
            &block_header.merkle_root,
            &resp
        ));

        let mut fail_resp = resp.clone();
        fail_resp.pos = 13;
        assert!(!utils::validate_merkle_proof(
            &txid,
            &block_header.merkle_root,
            &fail_resp
        ));

        let fail_block_header = client.block_header(resp.block_height + 1).unwrap();
        assert!(!utils::validate_merkle_proof(
            &txid,
            &fail_block_header.merkle_root,
            &resp
        ));
    }

    #[test]
    fn test_batch_transaction_get_merkle() {
        use bitcoin::Txid;

        struct TestCase {
            txid: Txid,
            block_height: usize,
            exp_pos: usize,
            exp_bytes: [u8; 32],
        }

        let client = get_test_client();

        let test_cases: Vec<TestCase> = vec![
            TestCase {
                txid: Txid::from_str(
                    "1f7ff3c407f33eabc8bec7d2cc230948f2249ec8e591bcf6f971ca9366c8788d",
                )
                .unwrap(),
                block_height: 630000,
                exp_pos: 68,
                exp_bytes: [
                    34, 65, 51, 64, 49, 139, 115, 189, 185, 246, 70, 225, 168, 193, 217, 195, 47,
                    66, 179, 240, 153, 24, 114, 215, 144, 196, 212, 41, 39, 155, 246, 25,
                ],
            },
            TestCase {
                txid: Txid::from_str(
                    "70a8639bc9b743c0610d1231103a2f8e99f4a25670946b91f16c55a5373b37d1",
                )
                .unwrap(),
                block_height: 630001,
                exp_pos: 25,
                exp_bytes: [
                    169, 100, 34, 99, 168, 101, 25, 168, 184, 90, 77, 50, 151, 245, 130, 101, 193,
                    229, 136, 128, 63, 110, 241, 19, 242, 59, 184, 137, 245, 249, 188, 110,
                ],
            },
            TestCase {
                txid: Txid::from_str(
                    "a0db149ace545beabbd87a8d6b20ffd6aa3b5a50e58add49a3d435f898c272cf",
                )
                .unwrap(),
                block_height: 840000,
                exp_pos: 0,
                exp_bytes: [
                    43, 184, 95, 75, 0, 75, 230, 218, 84, 247, 102, 193, 124, 30, 133, 81, 135, 50,
                    113, 18, 194, 49, 239, 47, 243, 94, 186, 208, 234, 103, 198, 158,
                ],
            },
        ];

        let txids_and_heights: Vec<(Txid, usize)> = test_cases
            .iter()
            .map(|case| (case.txid, case.block_height))
            .collect();

        let resp = client
            .batch_transaction_get_merkle(&txids_and_heights)
            .unwrap();

        for (i, (res, test_case)) in resp.iter().zip(test_cases).enumerate() {
            assert_eq!(res.block_height, test_case.block_height);
            assert_eq!(res.pos, test_case.exp_pos);
            assert_eq!(res.merkle.len(), 12);
            assert_eq!(res.merkle[0], test_case.exp_bytes);

            // Check we can verify the merkle proof validity, but fail if we supply wrong data.
            let block_header = client.block_header(res.block_height).unwrap();
            assert!(utils::validate_merkle_proof(
                &txids_and_heights[i].0,
                &block_header.merkle_root,
                res
            ));

            let mut fail_res = res.clone();
            fail_res.pos = 13;
            assert!(!utils::validate_merkle_proof(
                &txids_and_heights[i].0,
                &block_header.merkle_root,
                &fail_res
            ));

            let fail_block_header = client.block_header(res.block_height + 1).unwrap();
            assert!(!utils::validate_merkle_proof(
                &txids_and_heights[i].0,
                &fail_block_header.merkle_root,
                res
            ));
        }
    }

    #[test]
    fn test_txid_from_pos() {
        use bitcoin::Txid;

        let client = get_test_client();

        let txid =
            Txid::from_str("1f7ff3c407f33eabc8bec7d2cc230948f2249ec8e591bcf6f971ca9366c8788d")
                .unwrap();
        let resp = client.txid_from_pos(630000, 68).unwrap();
        assert_eq!(resp, txid);
    }

    #[test]
    fn test_txid_from_pos_with_merkle() {
        use bitcoin::Txid;

        let client = get_test_client();

        let txid =
            Txid::from_str("1f7ff3c407f33eabc8bec7d2cc230948f2249ec8e591bcf6f971ca9366c8788d")
                .unwrap();
        let resp = client.txid_from_pos_with_merkle(630000, 68).unwrap();
        assert_eq!(resp.tx_hash, txid);
        assert_eq!(
            resp.merkle[0],
            [
                34, 65, 51, 64, 49, 139, 115, 189, 185, 246, 70, 225, 168, 193, 217, 195, 47, 66,
                179, 240, 153, 24, 114, 215, 144, 196, 212, 41, 39, 155, 246, 25
            ]
        );
    }

    #[test]
    fn test_ping() {
        let client = get_test_client();
        client.ping().unwrap();
    }

    #[test]
    fn test_block_headers_subscribe() {
        let client = get_test_client();
        let resp = client.block_headers_subscribe().unwrap();

        assert!(resp.height >= 639000);
    }

    #[test]
    fn test_script_subscribe() {
        use std::str::FromStr;

        let client = get_test_client();

        // Mt.Gox hack address
        let addr = bitcoin::Address::from_str("1FeexV6bAHb8ybZjqQMjJrcCrHGW9sb6uF")
            .unwrap()
            .assume_checked();

        // Just make sure that the call returns Ok(something)
        client.script_subscribe(&addr.script_pubkey()).unwrap();
    }

    #[test]
    fn test_request_after_error() {
        let client = get_test_client();

        assert!(client.transaction_broadcast_raw(&[0x00]).is_err());
        assert!(client.server_features().is_ok());
    }

    #[test]
    fn test_raw_call() {
        use crate::types::Param;

        let client = get_test_client();

        let params = vec![
            Param::String(
                "cc2ca076fd04c2aeed6d02151c447ced3d09be6fb4d4ef36cb5ed4e7a3260566".to_string(),
            ),
            Param::Bool(false),
        ];

        let resp = client
            .raw_call("blockchain.transaction.get", params)
            .unwrap();

        assert_eq!(
            resp,
            "01000000000101000000000000000000000000000000000000000000000000000\
            0000000000000ffffffff5403f09c091b4d696e656420627920416e74506f6f6c3\
            13139ae006f20074d6528fabe6d6d2ab1948d50b3d991e2a0821df74358ed9c255\
            3af00c7a61f97771ca0acee106e0400000000000000cbec00802461f905fffffff\
            f0354ceac2a000000001976a91411dbe48cc6b617f9c6adaf4d9ed5f625b1c7cb5\
            988ac0000000000000000266a24aa21a9ed2e578bce2ca6c6bc9359377345d8e98\
            5dd5f90c78421ffa6efa5eb60428e698c0000000000000000266a24b9e11b6d2f6\
            21d7ec3f45a5eca89d3ea6a294cdf3a042e973009584470a12916111e2caa01200\
            000000000000000000000000000000000000000000000000000000000000000000\
            00000"
        )
    }

    #[test]
    fn test_authorization_provider_with_client() {
        use std::sync::{Arc, RwLock};

        // Track how many times the provider is called
        let call_count = Arc::new(RwLock::new(0));
        let call_count_clone = call_count.clone();

        let auth_provider = Arc::new(move || {
            *call_count_clone.write().unwrap() += 1;
            Some("Bearer test-token-123".to_string())
        });

        let client = get_test_auth_client(Some(auth_provider));

        // Make a request - provider should be called
        let _ = client.server_features();

        // Provider should have been called at least once
        assert!(*call_count.read().unwrap() >= 1);
    }

    #[test]
    fn test_authorization_provider_dynamic_token_refresh() {
        use std::sync::{Arc, RwLock};

        // Simulate a token that can be refreshed
        let token = Arc::new(RwLock::new("initial-token".to_string()));
        let token_clone = token.clone();

        let auth_provider: AuthProvider =
            Arc::new(move || Some(token_clone.read().unwrap().clone()));

        let client = get_test_auth_client(Some(auth_provider.clone()));

        // Make first request with initial token
        let _ = client.server_features();

        // Simulate token refresh
        *token.write().unwrap() = "refreshed-token".to_string();

        // Make second request - should use the new token
        let _ = client.server_features();

        // Verify the provider now returns the refreshed token
        assert_eq!(auth_provider(), Some("refreshed-token".to_string()));
    }

    #[test]
    fn test_authorization_provider_returns_none() {
        use std::sync::Arc;

        let auth_provider: AuthProvider = Arc::new(|| None);

        let client = get_test_auth_client(Some(auth_provider));

        // Should still work when provider returns None
        let result = client.server_features();
        assert!(result.is_ok());
    }

    #[test]
    fn test_auth_provider_via_constructor() {
        use std::sync::Arc;

        let auth_provider: AuthProvider = Arc::new(|| Some("Bearer test".to_string()));

        let client = get_test_auth_client(Some(auth_provider));

        // Verify the provider was set
        let result = client.server_features();
        assert!(result.is_ok());
    }
}