use std::future::Future;
use std::pin::Pin;
use std::time::Duration;

use crate::body::RequestBodyLocal;
use crate::error::Error;
use crate::pool::PooledConnection;
use crate::proxy::ProxyConfig;
use crate::runtime::{ConnectorLocal, RuntimeLocal, SocketConfig};

use super::HttpEngineLocal;

impl<R: RuntimeLocal, C: ConnectorLocal + Clone> HttpEngineLocal<R, C> {
    pub(super) async fn connect_via_proxy_local(
        &self,
        proxy: &ProxyConfig,
        target_authority: &http::uri::Authority,
        is_https: bool,
        connect_timeout: Option<Duration>,
    ) -> Result<PooledConnection<RequestBodyLocal>, Error> {
        let proxy_authority = proxy.authority()?;
        let default_port = proxy.default_port();
        let proxy_addr = self
            .core
            .resolve_authority(proxy_authority, default_port)
            .await?;
        let tcp_stream = if let Some(local_addr) = self.core.local_address {
            self.connector
                .connect_bound(proxy_addr, local_addr)
                .await
                .map_err(Error::Io)?
        } else {
            self.connector
                .connect(proxy_addr)
                .await
                .map_err(Error::Io)?
        };
        #[cfg(target_os = "linux")]
        if let Some(ref iface) = self.core.interface {
            tcp_stream.bind_device(iface).map_err(Error::Io)?;
        }
        if let Some(time) = self.core.tcp_keepalive {
            tcp_stream
                .set_keepalive(
                    time,
                    self.core.tcp_keepalive_interval,
                    self.core.tcp_keepalive_retries,
                )
                .map_err(Error::Io)?;
        }
        if self.core.tcp_fast_open {
            let _ = tcp_stream.set_fast_open();
        }

        if proxy.scheme == crate::proxy::ProxyScheme::Socks5
            || proxy.scheme == crate::proxy::ProxyScheme::Socks5h
        {
            let host = target_authority.host();
            let port = target_authority
                .port_u16()
                .unwrap_or(if is_https { 443 } else { 80 });
            let dns = if proxy.scheme == crate::proxy::ProxyScheme::Socks5h {
                crate::socks5::Socks5Dns::Remote
            } else {
                crate::socks5::Socks5Dns::Local
            };
            let mut std_stream = self.connector.into_std_tcp(tcp_stream).map_err(Error::Io)?;
            if let Some(timeout) = connect_timeout {
                std_stream
                    .set_read_timeout(Some(timeout))
                    .map_err(Error::Io)?;
                std_stream
                    .set_write_timeout(Some(timeout))
                    .map_err(Error::Io)?;
            }
            crate::socks5::socks5_handshake(&mut std_stream, host, port, proxy.auth.as_ref(), dns)
                .map_err(Error::Io)?;
            if connect_timeout.is_some() {
                std_stream.set_read_timeout(None).map_err(Error::Io)?;
                std_stream.set_write_timeout(None).map_err(Error::Io)?;
            }
            let tcp_stream = self.connector.from_std_tcp(std_stream).map_err(Error::Io)?;
            if is_https {
                self.connect_tls_local(tcp_stream, host).await
            } else {
                self.connect_h1_local(tcp_stream).await
            }
        } else if proxy.scheme == crate::proxy::ProxyScheme::Socks4 {
            let host = target_authority.host();
            let port = target_authority
                .port_u16()
                .unwrap_or(if is_https { 443 } else { 80 });
            let mut std_stream = self.connector.into_std_tcp(tcp_stream).map_err(Error::Io)?;
            if let Some(timeout) = connect_timeout {
                std_stream
                    .set_read_timeout(Some(timeout))
                    .map_err(Error::Io)?;
                std_stream
                    .set_write_timeout(Some(timeout))
                    .map_err(Error::Io)?;
            }
            crate::socks4::socks4a_handshake(&mut std_stream, host, port, proxy.auth.as_ref())
                .map_err(Error::Io)?;
            if connect_timeout.is_some() {
                std_stream.set_read_timeout(None).map_err(Error::Io)?;
                std_stream.set_write_timeout(None).map_err(Error::Io)?;
            }
            let tcp_stream = self.connector.from_std_tcp(std_stream).map_err(Error::Io)?;
            if is_https {
                self.connect_tls_local(tcp_stream, host).await
            } else {
                self.connect_h1_local(tcp_stream).await
            }
        } else if proxy.scheme == crate::proxy::ProxyScheme::Https {
            #[cfg(all(feature = "rustls", feature = "compio"))]
            {
                use crate::tls::TlsConnectLocal;
                let tls_connector = self
                    .core
                    .tls
                    .as_ref()
                    .ok_or_else(|| Error::Tls("no TLS connector configured".into()))?;
                let tls_stream =
                    <crate::tls::RustlsConnector as TlsConnectLocal<C::Stream>>::connect_local(
                        tls_connector,
                        proxy_authority.host(),
                        tcp_stream,
                    )
                    .await
                    .map_err(|e| Error::Tls(Box::new(e)))?;
                if is_https {
                    self.connect_tunnel_local(tls_stream, proxy, target_authority, connect_timeout)
                        .await
                } else {
                    // HTTPS proxy for HTTP target: CONNECT through TLS pipe.
                    let port = target_authority.port_u16().unwrap_or(80);
                    let target = format!("{}:{port}", target_authority.host());
                    let tunnel_stream =
                        do_connect_handshake_local(tls_stream, proxy, &target).await?;
                    if self.core.http2_prior_knowledge {
                        self.connect_h2_prior_knowledge_local(tunnel_stream).await
                    } else {
                        self.connect_h1_local(tunnel_stream).await
                    }
                }
            }
            #[cfg(not(all(feature = "rustls", feature = "compio")))]
            {
                Err(Error::Tls(
                    "HTTPS proxy requires rustls + compio features".into(),
                ))
            }
        } else if is_https {
            self.connect_tunnel_local(tcp_stream, proxy, target_authority, connect_timeout)
                .await
        } else {
            // HTTP proxy for HTTP target: CONNECT to create a raw pipe.
            let port = target_authority.port_u16().unwrap_or(80);
            let target = format!("{}:{port}", target_authority.host());
            let tunnel_stream = do_connect_handshake_local(tcp_stream, proxy, &target).await?;
            if self.core.http2_prior_knowledge {
                self.connect_h2_prior_knowledge_local(tunnel_stream).await
            } else {
                self.connect_h1_local(tunnel_stream).await
            }
        }
    }

    async fn connect_tunnel_local<S>(
        &self,
        stream: S,
        proxy: &ProxyConfig,
        target_authority: &http::uri::Authority,
        _connect_timeout: Option<Duration>,
    ) -> Result<PooledConnection<RequestBodyLocal>, Error>
    where
        S: hyper::rt::Read + hyper::rt::Write + Unpin + 'static,
    {
        let port = target_authority.port_u16().unwrap_or(443);
        let target = format!("{}:{port}", target_authority.host());
        let tunnel_stream = do_connect_handshake_local(stream, proxy, &target).await?;
        #[cfg(all(feature = "rustls", feature = "compio"))]
        {
            let stream = tunnel_stream;
            let host = target_authority.host();
            use crate::tls::TlsConnectLocal;
            use std::time::Instant;

            let tls_connector = self
                .core
                .tls
                .as_ref()
                .ok_or_else(|| Error::Tls("no TLS connector configured".into()))?;

            let tls_start = Instant::now();
            let tls_stream = <crate::tls::RustlsConnector as TlsConnectLocal<S>>::connect_local(
                tls_connector,
                host,
                stream,
            )
            .await
            .map_err(|e| {
                #[cfg(feature = "tracing")]
                tracing::trace!(host = host, error = %e, "tls.handshake.error");
                Error::Tls(Box::new(e))
            })?;

            let tls_duration = tls_start.elapsed();
            let alpn =
                crate::tls::RustlsConnector::negotiated_protocol(tls_stream.tls_connection());
            let tls_info = tls_stream.tls_info();

            match alpn {
                Some(crate::tls::AlpnProtocol::H2) => {
                    let mut builder = hyper::client::conn::http2::Builder::new(
                        crate::runtime::executor::completion_executor::<R>(),
                    );
                    if let Some(ref h2) = self.core.http2 {
                        h2.apply(&mut builder);
                    }
                    let (sender, conn) = builder.handshake(tls_stream).await?;
                    R::spawn_local(async move {
                        let _ = conn.await;
                    });
                    let mut pooled = PooledConnection::new_h2(sender);
                    pooled.tls_info = Some(tls_info);
                    pooled.tls_handshake_duration = Some(tls_duration);
                    Ok(pooled)
                }
                _ => {
                    let (sender, conn) = hyper::client::conn::http1::handshake(tls_stream).await?;

                    let handle = crate::upgrade::UpgradeHandleLocal::new();
                    let handle_clone = handle.clone();

                    R::spawn_local(async move {
                        match conn.without_shutdown().await {
                            Ok(parts) => {
                                let upgraded =
                                    crate::upgrade::UpgradedLocal::new(parts.io, parts.read_buf);
                                handle_clone.fulfill(upgraded);
                            }
                            Err(_) => {
                                handle_clone.fail();
                            }
                        }
                    });

                    let mut pooled = PooledConnection::new_h1(sender);
                    pooled.tls_info = Some(tls_info);
                    pooled.tls_handshake_duration = Some(tls_duration);
                    pooled.upgrade_handle_local = Some(handle);
                    Ok(pooled)
                }
            }
        }
        #[cfg(not(all(feature = "rustls", feature = "compio")))]
        {
            drop(tunnel_stream);
            Err(Error::Tls(
                "HTTPS CONNECT tunnel requires rustls + compio features".into(),
            ))
        }
    }

    async fn connect_two_hop_local(
        &self,
        first: &ProxyConfig,
        second: &ProxyConfig,
        target_authority: &http::uri::Authority,
        is_https: bool,
        connect_timeout: Option<Duration>,
    ) -> Result<PooledConnection<RequestBodyLocal>, Error> {
        let second_authority = second.authority()?;
        let second_default_port = second.default_port();
        let second_host = second_authority.host();
        let second_port = second_authority.port_u16().unwrap_or(second_default_port);

        let first_authority = first.authority()?;
        let first_addr = self
            .core
            .resolve_authority(first_authority, first.default_port())
            .await?;

        let tcp_stream = if let Some(local_addr) = self.core.local_address {
            self.connector
                .connect_bound(first_addr, local_addr)
                .await
                .map_err(Error::Io)?
        } else {
            self.connector
                .connect(first_addr)
                .await
                .map_err(Error::Io)?
        };

        #[cfg(target_os = "linux")]
        if let Some(ref iface) = self.core.interface {
            tcp_stream.bind_device(iface).map_err(Error::Io)?;
        }
        if let Some(time) = self.core.tcp_keepalive {
            tcp_stream
                .set_keepalive(
                    time,
                    self.core.tcp_keepalive_interval,
                    self.core.tcp_keepalive_retries,
                )
                .map_err(Error::Io)?;
        }
        if self.core.tcp_fast_open {
            let _ = tcp_stream.set_fast_open();
        }

        if first.scheme == crate::proxy::ProxyScheme::Socks5
            || first.scheme == crate::proxy::ProxyScheme::Socks5h
        {
            let dns = if first.scheme == crate::proxy::ProxyScheme::Socks5h {
                crate::socks5::Socks5Dns::Remote
            } else {
                crate::socks5::Socks5Dns::Local
            };
            let mut std_stream = self.connector.into_std_tcp(tcp_stream).map_err(Error::Io)?;
            if let Some(timeout) = connect_timeout {
                std_stream
                    .set_read_timeout(Some(timeout))
                    .map_err(Error::Io)?;
                std_stream
                    .set_write_timeout(Some(timeout))
                    .map_err(Error::Io)?;
            }
            crate::socks5::socks5_handshake(
                &mut std_stream,
                second_host,
                second_port,
                first.auth.as_ref(),
                dns,
            )
            .map_err(Error::Io)?;
            if connect_timeout.is_some() {
                std_stream.set_read_timeout(None).map_err(Error::Io)?;
                std_stream.set_write_timeout(None).map_err(Error::Io)?;
            }
            let stream = self.connector.from_std_tcp(std_stream).map_err(Error::Io)?;
            self.connect_second_hop_local(
                stream,
                second,
                target_authority,
                is_https,
                connect_timeout,
            )
            .await
        } else if first.scheme == crate::proxy::ProxyScheme::Socks4 {
            let mut std_stream = self.connector.into_std_tcp(tcp_stream).map_err(Error::Io)?;
            if let Some(timeout) = connect_timeout {
                std_stream
                    .set_read_timeout(Some(timeout))
                    .map_err(Error::Io)?;
                std_stream
                    .set_write_timeout(Some(timeout))
                    .map_err(Error::Io)?;
            }
            crate::socks4::socks4a_handshake(
                &mut std_stream,
                second_host,
                second_port,
                first.auth.as_ref(),
            )
            .map_err(Error::Io)?;
            if connect_timeout.is_some() {
                std_stream.set_read_timeout(None).map_err(Error::Io)?;
                std_stream.set_write_timeout(None).map_err(Error::Io)?;
            }
            let stream = self.connector.from_std_tcp(std_stream).map_err(Error::Io)?;
            self.connect_second_hop_local(
                stream,
                second,
                target_authority,
                is_https,
                connect_timeout,
            )
            .await
        } else if first.scheme == crate::proxy::ProxyScheme::Https {
            #[cfg(all(feature = "rustls", feature = "compio"))]
            {
                use crate::tls::TlsConnectLocal;
                let tls_connector = self
                    .core
                    .tls
                    .as_ref()
                    .ok_or_else(|| Error::Tls("no TLS connector configured".into()))?;
                let tls_stream =
                    <crate::tls::RustlsConnector as TlsConnectLocal<C::Stream>>::connect_local(
                        tls_connector,
                        first_authority.host(),
                        tcp_stream,
                    )
                    .await
                    .map_err(|e| Error::Tls(Box::new(e)))?;
                let second_target = format!(
                    "{}:{}",
                    second_authority.host(),
                    second_authority.port_u16().unwrap_or(second.default_port())
                );
                let stream = do_connect_handshake_local(tls_stream, first, &second_target).await?;
                if is_https {
                    self.connect_tunnel_local(stream, second, target_authority, connect_timeout)
                        .await
                } else {
                    self.connect_plaintext_local(stream).await
                }
            }
            #[cfg(not(all(feature = "rustls", feature = "compio")))]
            {
                Err(Error::Tls(
                    "HTTPS proxy requires rustls + compio features".into(),
                ))
            }
        } else {
            // HTTP proxy: CONNECT through first to reach second
            let second_target = format!(
                "{}:{}",
                second_authority.host(),
                second_authority.port_u16().unwrap_or(second.default_port())
            );
            let stream = do_connect_handshake_local(tcp_stream, first, &second_target).await?;
            self.connect_second_hop_local(
                stream,
                second,
                target_authority,
                is_https,
                connect_timeout,
            )
            .await
        }
    }

    /// Second-hop dispatch for the local (!Send) path.
    /// The stream is already connected to `second`'s proxy.
    async fn connect_second_hop_local(
        &self,
        stream: C::Stream,
        second: &ProxyConfig,
        target_authority: &http::uri::Authority,
        is_https: bool,
        connect_timeout: Option<Duration>,
    ) -> Result<PooledConnection<RequestBodyLocal>, Error> {
        let target_host = target_authority.host();
        let target_port = target_authority
            .port_u16()
            .unwrap_or(if is_https { 443 } else { 80 });

        if second.scheme == crate::proxy::ProxyScheme::Socks5
            || second.scheme == crate::proxy::ProxyScheme::Socks5h
        {
            let dns = if second.scheme == crate::proxy::ProxyScheme::Socks5h {
                crate::socks5::Socks5Dns::Remote
            } else {
                crate::socks5::Socks5Dns::Local
            };
            let mut std_stream = self.connector.into_std_tcp(stream).map_err(Error::Io)?;
            if let Some(timeout) = connect_timeout {
                std_stream
                    .set_read_timeout(Some(timeout))
                    .map_err(Error::Io)?;
                std_stream
                    .set_write_timeout(Some(timeout))
                    .map_err(Error::Io)?;
            }
            crate::socks5::socks5_handshake(
                &mut std_stream,
                target_host,
                target_port,
                second.auth.as_ref(),
                dns,
            )
            .map_err(Error::Io)?;
            if connect_timeout.is_some() {
                std_stream.set_read_timeout(None).map_err(Error::Io)?;
                std_stream.set_write_timeout(None).map_err(Error::Io)?;
            }
            let stream = self.connector.from_std_tcp(std_stream).map_err(Error::Io)?;
            if is_https {
                self.connect_tls_local(stream, target_host).await
            } else {
                self.connect_plaintext_local(stream).await
            }
        } else if second.scheme == crate::proxy::ProxyScheme::Socks4 {
            let mut std_stream = self.connector.into_std_tcp(stream).map_err(Error::Io)?;
            if let Some(timeout) = connect_timeout {
                std_stream
                    .set_read_timeout(Some(timeout))
                    .map_err(Error::Io)?;
                std_stream
                    .set_write_timeout(Some(timeout))
                    .map_err(Error::Io)?;
            }
            crate::socks4::socks4a_handshake(
                &mut std_stream,
                target_host,
                target_port,
                second.auth.as_ref(),
            )
            .map_err(Error::Io)?;
            if connect_timeout.is_some() {
                std_stream.set_read_timeout(None).map_err(Error::Io)?;
                std_stream.set_write_timeout(None).map_err(Error::Io)?;
            }
            let stream = self.connector.from_std_tcp(std_stream).map_err(Error::Io)?;
            if is_https {
                self.connect_tls_local(stream, target_host).await
            } else {
                self.connect_plaintext_local(stream).await
            }
        } else if second.scheme == crate::proxy::ProxyScheme::Https {
            #[cfg(all(feature = "rustls", feature = "compio"))]
            {
                use crate::tls::TlsConnectLocal;
                let tls_connector = self
                    .core
                    .tls
                    .as_ref()
                    .ok_or_else(|| Error::Tls("no TLS connector configured".into()))?;
                let second_authority = second.authority()?;
                let tls_stream =
                    <crate::tls::RustlsConnector as TlsConnectLocal<C::Stream>>::connect_local(
                        tls_connector,
                        second_authority.host(),
                        stream,
                    )
                    .await
                    .map_err(|e| Error::Tls(Box::new(e)))?;
                if is_https {
                    self.connect_tunnel_local(tls_stream, second, target_authority, connect_timeout)
                        .await
                } else {
                    self.connect_plaintext_local(tls_stream).await
                }
            }
            #[cfg(not(all(feature = "rustls", feature = "compio")))]
            {
                Err(Error::Tls(
                    "HTTPS proxy requires rustls + compio features".into(),
                ))
            }
        } else {
            // HTTP: CONNECT through second to reach target
            if is_https {
                self.connect_tunnel_local(stream, second, target_authority, connect_timeout)
                    .await
            } else {
                self.connect_plaintext_local(stream).await
            }
        }
    }

    pub(super) async fn connect_via_proxy_chain_local(
        &self,
        chain: &crate::proxy::ProxyChain,
        target_authority: &http::uri::Authority,
        is_https: bool,
        connect_timeout: Option<Duration>,
    ) -> Result<PooledConnection<RequestBodyLocal>, Error> {
        match chain.len() {
            0 => Err(Error::Other("empty proxy chain".into())),
            1 => {
                self.connect_via_proxy_local(
                    &chain.proxies[0],
                    target_authority,
                    is_https,
                    connect_timeout,
                )
                .await
            }
            2 => {
                self.connect_two_hop_local(
                    &chain.proxies[0],
                    &chain.proxies[1],
                    target_authority,
                    is_https,
                    connect_timeout,
                )
                .await
            }
            n => Err(Error::Other(
                format!("proxy chains longer than 2 hops are not yet supported (got {n})").into(),
            )),
        }
    }

    pub(super) fn connect_plaintext_local<S>(
        &self,
        stream: S,
    ) -> Pin<Box<dyn Future<Output = Result<PooledConnection<RequestBodyLocal>, Error>> + '_>>
    where
        S: hyper::rt::Read + hyper::rt::Write + Unpin + 'static,
    {
        self.connect_plaintext_local_with_hint(stream, false)
    }

    pub(super) fn connect_plaintext_local_with_hint<S>(
        &self,
        stream: S,
        force_h2c: bool,
    ) -> Pin<Box<dyn Future<Output = Result<PooledConnection<RequestBodyLocal>, Error>> + '_>>
    where
        S: hyper::rt::Read + hyper::rt::Write + Unpin + 'static,
    {
        if self.core.http2_prior_knowledge || force_h2c {
            Box::pin(self.connect_h2_prior_knowledge_local(stream))
        } else {
            Box::pin(self.connect_h1_local(stream))
        }
    }

    pub(super) async fn connect_h1_local<S>(
        &self,
        stream: S,
    ) -> Result<PooledConnection<RequestBodyLocal>, Error>
    where
        S: hyper::rt::Read + hyper::rt::Write + Unpin + 'static,
    {
        let (sender, conn) = hyper::client::conn::http1::handshake(stream).await?;

        let handle = crate::upgrade::UpgradeHandleLocal::new();
        let handle_clone = handle.clone();

        R::spawn_local(async move {
            match conn.without_shutdown().await {
                Ok(parts) => {
                    let upgraded = crate::upgrade::UpgradedLocal::new(parts.io, parts.read_buf);
                    handle_clone.fulfill(upgraded);
                }
                Err(_) => {
                    handle_clone.fail();
                }
            }
        });

        let mut pooled = PooledConnection::new_h1(sender);
        pooled.upgrade_handle_local = Some(handle);
        Ok(pooled)
    }

    pub(super) async fn connect_h2_prior_knowledge_local<S>(
        &self,
        stream: S,
    ) -> Result<PooledConnection<RequestBodyLocal>, Error>
    where
        S: hyper::rt::Read + hyper::rt::Write + Unpin + 'static,
    {
        let mut builder = hyper::client::conn::http2::Builder::new(
            crate::runtime::executor::completion_executor::<R>(),
        );
        if let Some(ref h2) = self.core.http2 {
            h2.apply(&mut builder);
        }
        let (sender, conn) = builder.handshake(stream).await?;
        R::spawn_local(async move {
            let _ = conn.await;
        });
        Ok(PooledConnection::new_h2(sender))
    }

    #[cfg(all(feature = "rustls", feature = "compio"))]
    pub(super) async fn connect_tls_local(
        &self,
        tcp_stream: C::Stream,
        host: &str,
    ) -> Result<PooledConnection<RequestBodyLocal>, Error> {
        use crate::tls::TlsConnectLocal;
        use std::time::Instant;

        let tls_start = Instant::now();

        let tls_connector = self
            .core
            .tls
            .as_ref()
            .ok_or_else(|| Error::Tls("no TLS connector configured".into()))?;

        let tls_stream =
            <crate::tls::RustlsConnector as TlsConnectLocal<C::Stream>>::connect_local(
                tls_connector,
                host,
                tcp_stream,
            )
            .await
            .map_err(|e| Error::Tls(Box::new(e)))?;

        let tls_duration = tls_start.elapsed();

        let alpn = crate::tls::RustlsConnector::negotiated_protocol(tls_stream.tls_connection());
        let tls_info = tls_stream.tls_info();

        match alpn {
            Some(crate::tls::AlpnProtocol::H2) => {
                let mut builder = hyper::client::conn::http2::Builder::new(
                    crate::runtime::executor::completion_executor::<R>(),
                );
                if let Some(ref h2) = self.core.http2 {
                    h2.apply(&mut builder);
                }
                let (sender, conn) = builder.handshake(tls_stream).await?;
                R::spawn_local(async move {
                    let _ = conn.await;
                });
                let mut pooled = PooledConnection::new_h2(sender);
                pooled.tls_info = Some(tls_info);
                pooled.tls_handshake_duration = Some(tls_duration);
                Ok(pooled)
            }
            _ => {
                let (sender, conn) = hyper::client::conn::http1::handshake(tls_stream).await?;

                let handle = crate::upgrade::UpgradeHandleLocal::new();
                let handle_clone = handle.clone();

                R::spawn_local(async move {
                    match conn.without_shutdown().await {
                        Ok(parts) => {
                            let upgraded =
                                crate::upgrade::UpgradedLocal::new(parts.io, parts.read_buf);
                            handle_clone.fulfill(upgraded);
                        }
                        Err(_) => {
                            handle_clone.fail();
                        }
                    }
                });

                let mut pooled = PooledConnection::new_h1(sender);
                pooled.tls_info = Some(tls_info);
                pooled.tls_handshake_duration = Some(tls_duration);
                pooled.upgrade_handle_local = Some(handle);
                Ok(pooled)
            }
        }
    }

    #[cfg(all(feature = "rustls", not(feature = "compio")))]
    pub(super) async fn connect_tls_local(
        &self,
        _tcp_stream: C::Stream,
        _host: &str,
    ) -> Result<PooledConnection<RequestBodyLocal>, Error> {
        Err(Error::Tls(
            "TLS with !Send streams requires the compio feature".into(),
        ))
    }

    #[cfg(not(feature = "rustls"))]
    pub(super) async fn connect_tls_local(
        &self,
        _tcp_stream: C::Stream,
        _host: &str,
    ) -> Result<PooledConnection<RequestBodyLocal>, Error> {
        Err(Error::Tls(
            "HTTPS requires the `rustls` TLS backend feature".into(),
        ))
    }
}

/// Perform an HTTP CONNECT handshake through `stream` to `target`.
///
/// Sends `CONNECT target HTTP/1.1`, reads the response, validates HTTP 200,
/// and returns the stream unchanged on success. Type-preserving: the returned
/// stream is the same `S` that was passed in, making it reusable for proxy
/// chaining.
///
/// This is the `!Send` (local/completion runtime) variant.
async fn do_connect_handshake_local<S>(
    mut stream: S,
    proxy: &ProxyConfig,
    target: &str,
) -> Result<S, Error>
where
    S: hyper::rt::Read + hyper::rt::Write + Unpin + 'static,
{
    let mut connect_msg = format!("CONNECT {target} HTTP/1.1\r\nHost: {target}\r\n");
    if let Some(auth_value) = proxy.connect_header(target) {
        connect_msg.push_str(&format!("Proxy-Authorization: {auth_value}\r\n"));
    }
    connect_msg.push_str("\r\n");

    let buf = connect_msg.into_bytes();
    let mut written = 0;
    while written < buf.len() {
        let n = std::future::poll_fn(|cx| Pin::new(&mut stream).poll_write(cx, &buf[written..]))
            .await
            .map_err(Error::Io)?;
        if n == 0 {
            return Err(Error::Io(std::io::Error::new(
                std::io::ErrorKind::WriteZero,
                "proxy closed connection during CONNECT handshake",
            )));
        }
        written += n;
    }

    // Flush after write: completion-based runtimes (compio) may buffer
    // writes internally; poll_flush ensures bytes reach the proxy before
    // we start reading the CONNECT response.
    std::future::poll_fn(|cx| Pin::new(&mut stream).poll_flush(cx))
        .await
        .map_err(Error::Io)?;

    let mut resp_buf = Vec::with_capacity(256);
    loop {
        let mut one = [0u8; 1];
        let mut read_buf = hyper::rt::ReadBuf::new(&mut one);
        std::future::poll_fn(|cx| Pin::new(&mut stream).poll_read(cx, read_buf.unfilled()))
            .await
            .map_err(Error::Io)?;

        if read_buf.filled().is_empty() {
            return Err(Error::Other("proxy closed connection".into()));
        }
        resp_buf.push(one[0]);

        if resp_buf.len() >= 4 && resp_buf[resp_buf.len() - 4..] == *b"\r\n\r\n" {
            break;
        }

        if resp_buf.len() > 8192 {
            return Err(Error::Other("CONNECT response too large".into()));
        }
    }

    let resp_str = String::from_utf8_lossy(&resp_buf);
    let status_line = resp_str
        .lines()
        .next()
        .ok_or_else(|| Error::Other("empty CONNECT response".into()))?;

    let status_code = super::connect::parse_connect_status(status_line)?;
    if status_code != 200 {
        return Err(Error::Other(
            format!("CONNECT tunnel failed: {status_line}").into(),
        ));
    }

    Ok(stream)
}

#[cfg(all(test, feature = "compio"))]
mod compio_tests {
    use super::super::HttpEngineLocal;
    use crate::runtime::compio_rt::{CompioIo, CompioRuntime, TcpConnector};

    /// Helper: build an HttpEngineLocal with default settings (http2_prior_knowledge = false).
    fn make_local_engine() -> HttpEngineLocal<CompioRuntime, TcpConnector> {
        HttpEngineLocal::<CompioRuntime, TcpConnector>::new()
    }

    /// Helper: build an HttpEngineLocal with http2_prior_knowledge = true.
    fn make_h2_local_engine() -> HttpEngineLocal<CompioRuntime, TcpConnector> {
        HttpEngineLocal::<CompioRuntime, TcpConnector>::builder()
            .http2_prior_knowledge()
            .build_local()
            .unwrap()
    }

    #[test]
    fn connect_h1_local_succeeds() {
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();
        listener.set_nonblocking(true).unwrap();
        let async_listener = async_io::Async::new(listener).unwrap();

        compio_runtime::Runtime::new().unwrap().block_on(async {
            let client_tcp = async_io::Async::<std::net::TcpStream>::connect(addr)
                .await
                .unwrap();
            let (server_tcp, _) = async_listener.accept().await.unwrap();

            // Keep server alive — drain reads in a background task.
            compio_runtime::spawn(async move {
                use futures_io::AsyncRead;
                let mut server = server_tcp;
                let mut buf = [0u8; 4096];
                while std::future::poll_fn(|cx| {
                    std::pin::Pin::new(&mut server).poll_read(cx, &mut buf)
                })
                .await
                .unwrap_or(0)
                    > 0
                {}
            })
            .detach();

            let io = CompioIo::new(client_tcp);
            let engine = make_local_engine();
            let result = engine.connect_h1_local(io).await;
            assert!(result.is_ok());
            let pooled = result.unwrap();
            assert!(matches!(pooled.conn, crate::pool::HttpConnection::H1(_)));
        });
    }

    #[test]
    fn connect_h2_prior_knowledge_local_succeeds() {
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();
        listener.set_nonblocking(true).unwrap();
        let async_listener = async_io::Async::new(listener).unwrap();

        compio_runtime::Runtime::new().unwrap().block_on(async {
            let client_tcp = async_io::Async::<std::net::TcpStream>::connect(addr)
                .await
                .unwrap();
            let (server_tcp, _) = async_listener.accept().await.unwrap();

            // Spawn an h2 server on the server side
            compio_runtime::spawn(async move {
                let io = CompioIo::new(server_tcp);
                let builder = hyper::server::conn::http2::Builder::new(
                    crate::runtime::executor::completion_executor::<CompioRuntime>(),
                );
                let _ = builder
                    .serve_connection(
                        io,
                        hyper::service::service_fn(|_req| async {
                            Ok::<_, std::convert::Infallible>(hyper::Response::new(
                                http_body_util::Empty::<bytes::Bytes>::new(),
                            ))
                        }),
                    )
                    .await;
            })
            .detach();

            let io = CompioIo::new(client_tcp);
            let engine = make_local_engine();
            let result = engine.connect_h2_prior_knowledge_local(io).await;
            assert!(result.is_ok());
            let pooled = result.unwrap();
            assert!(matches!(pooled.conn, crate::pool::HttpConnection::H2(_)));
        });
    }

    #[test]
    fn connect_plaintext_local_defaults_to_h1() {
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();
        listener.set_nonblocking(true).unwrap();
        let async_listener = async_io::Async::new(listener).unwrap();

        compio_runtime::Runtime::new().unwrap().block_on(async {
            let client_tcp = async_io::Async::<std::net::TcpStream>::connect(addr)
                .await
                .unwrap();
            let (server_tcp, _) = async_listener.accept().await.unwrap();

            compio_runtime::spawn(async move {
                use futures_io::AsyncRead;
                let mut server = server_tcp;
                let mut buf = [0u8; 4096];
                while std::future::poll_fn(|cx| {
                    std::pin::Pin::new(&mut server).poll_read(cx, &mut buf)
                })
                .await
                .unwrap_or(0)
                    > 0
                {}
            })
            .detach();

            let io = CompioIo::new(client_tcp);
            let engine = make_local_engine();
            let result = engine.connect_plaintext_local(io).await;
            assert!(result.is_ok());
            let pooled = result.unwrap();
            assert!(matches!(pooled.conn, crate::pool::HttpConnection::H1(_)));
        });
    }

    #[test]
    fn connect_plaintext_local_with_hint_false_uses_h1() {
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();
        listener.set_nonblocking(true).unwrap();
        let async_listener = async_io::Async::new(listener).unwrap();

        compio_runtime::Runtime::new().unwrap().block_on(async {
            let client_tcp = async_io::Async::<std::net::TcpStream>::connect(addr)
                .await
                .unwrap();
            let (server_tcp, _) = async_listener.accept().await.unwrap();

            compio_runtime::spawn(async move {
                use futures_io::AsyncRead;
                let mut server = server_tcp;
                let mut buf = [0u8; 4096];
                while std::future::poll_fn(|cx| {
                    std::pin::Pin::new(&mut server).poll_read(cx, &mut buf)
                })
                .await
                .unwrap_or(0)
                    > 0
                {}
            })
            .detach();

            let io = CompioIo::new(client_tcp);
            let engine = make_local_engine();
            let result = engine.connect_plaintext_local_with_hint(io, false).await;
            assert!(result.is_ok());
            let pooled = result.unwrap();
            assert!(matches!(pooled.conn, crate::pool::HttpConnection::H1(_)));
        });
    }

    #[test]
    fn connect_plaintext_local_with_hint_true_uses_h2() {
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();
        listener.set_nonblocking(true).unwrap();
        let async_listener = async_io::Async::new(listener).unwrap();

        compio_runtime::Runtime::new().unwrap().block_on(async {
            let client_tcp = async_io::Async::<std::net::TcpStream>::connect(addr)
                .await
                .unwrap();
            let (server_tcp, _) = async_listener.accept().await.unwrap();

            // Spawn an h2 server
            compio_runtime::spawn(async move {
                let io = CompioIo::new(server_tcp);
                let builder = hyper::server::conn::http2::Builder::new(
                    crate::runtime::executor::completion_executor::<CompioRuntime>(),
                );
                let _ = builder
                    .serve_connection(
                        io,
                        hyper::service::service_fn(|_req| async {
                            Ok::<_, std::convert::Infallible>(hyper::Response::new(
                                http_body_util::Empty::<bytes::Bytes>::new(),
                            ))
                        }),
                    )
                    .await;
            })
            .detach();

            let io = CompioIo::new(client_tcp);
            let engine = make_local_engine();
            let result = engine.connect_plaintext_local_with_hint(io, true).await;
            assert!(result.is_ok());
            let pooled = result.unwrap();
            assert!(matches!(pooled.conn, crate::pool::HttpConnection::H2(_)));
        });
    }

    #[test]
    fn connect_plaintext_local_with_http2_prior_knowledge_uses_h2() {
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();
        listener.set_nonblocking(true).unwrap();
        let async_listener = async_io::Async::new(listener).unwrap();

        compio_runtime::Runtime::new().unwrap().block_on(async {
            let client_tcp = async_io::Async::<std::net::TcpStream>::connect(addr)
                .await
                .unwrap();
            let (server_tcp, _) = async_listener.accept().await.unwrap();

            // Spawn an h2 server
            compio_runtime::spawn(async move {
                let io = CompioIo::new(server_tcp);
                let builder = hyper::server::conn::http2::Builder::new(
                    crate::runtime::executor::completion_executor::<CompioRuntime>(),
                );
                let _ = builder
                    .serve_connection(
                        io,
                        hyper::service::service_fn(|_req| async {
                            Ok::<_, std::convert::Infallible>(hyper::Response::new(
                                http_body_util::Empty::<bytes::Bytes>::new(),
                            ))
                        }),
                    )
                    .await;
            })
            .detach();

            let io = CompioIo::new(client_tcp);
            let engine = make_h2_local_engine();
            // http2_prior_knowledge is true, so connect_plaintext_local uses h2
            let result = engine.connect_plaintext_local(io).await;
            assert!(result.is_ok());
            let pooled = result.unwrap();
            assert!(matches!(pooled.conn, crate::pool::HttpConnection::H2(_)));
        });
    }

    #[test]
    fn connect_h1_local_server_closes_immediately() {
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();
        listener.set_nonblocking(true).unwrap();
        let async_listener = async_io::Async::new(listener).unwrap();

        compio_runtime::Runtime::new().unwrap().block_on(async {
            let client_tcp = async_io::Async::<std::net::TcpStream>::connect(addr)
                .await
                .unwrap();
            let (server_tcp, _) = async_listener.accept().await.unwrap();
            // Drop server immediately
            drop(server_tcp);

            let io = CompioIo::new(client_tcp);
            let engine = make_local_engine();
            let result = engine.connect_h1_local(io).await;
            // h1 handshake does not require server response — it just creates the sender
            assert!(result.is_ok());
        });
    }

    #[test]
    fn connect_h2_local_server_closes_immediately_fails() {
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();
        listener.set_nonblocking(true).unwrap();
        let async_listener = async_io::Async::new(listener).unwrap();

        compio_runtime::Runtime::new().unwrap().block_on(async {
            let client_tcp = async_io::Async::<std::net::TcpStream>::connect(addr)
                .await
                .unwrap();
            let (server_tcp, _) = async_listener.accept().await.unwrap();
            // Drop server immediately — h2 handshake requires preface exchange.
            // On compio the close may not propagate before hyper returns the sender,
            // so we just verify the handshake completes (either way) and that a
            // subsequent request on the dead connection would fail.
            drop(server_tcp);

            let io = CompioIo::new(client_tcp);
            let engine = make_local_engine();
            let result = engine.connect_h2_prior_knowledge_local(io).await;
            match result {
                Ok(pooled) => {
                    // Handshake "succeeded" but the connection is dead.
                    // Verify it's at least an H2 connection.
                    assert!(matches!(pooled.conn, crate::pool::HttpConnection::H2(_)));
                }
                Err(_) => {
                    // On some platforms/timings, the close is detected during handshake
                }
            }
        });
    }

    #[cfg(feature = "rustls")]
    #[test]
    fn connect_tls_local_on_plain_stream_fails() {
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();

        compio_runtime::Runtime::new().unwrap().block_on(async {
            // Keep server alive in background (std listener accepts synchronously)
            let accept_handle = std::thread::spawn(move || {
                let (mut conn, _) = listener.accept().unwrap();
                use std::io::Read;
                let mut buf = [0u8; 4096];
                let _ = conn.read(&mut buf);
            });

            crate::tls::install_default_crypto_provider();
            let engine = HttpEngineLocal::<CompioRuntime, TcpConnector>::builder()
                .tls(crate::tls::RustlsConnector::with_webpki_roots())
                .build_local()
                .unwrap();
            let connector = TcpConnector;
            let stream = crate::runtime::ConnectorLocal::connect(&connector, addr)
                .await
                .unwrap();
            let result = engine.connect_tls_local(stream, "example.com").await;
            // TLS handshake should fail on a non-TLS stream
            assert!(result.is_err());

            drop(accept_handle);
        });
    }
}

#[cfg(all(test, feature = "tokio"))]
mod tokio_tests {
    use std::cell::Cell;
    use std::pin::Pin;
    use std::task::{Context, Poll};

    use super::*;

    /// Mock stream that records whether poll_flush was called before poll_read.
    struct FlushTrackingStream {
        response: Vec<u8>,
        read_pos: usize,
        flushed: Cell<bool>,
    }

    impl FlushTrackingStream {
        fn new(response: &[u8]) -> Self {
            Self {
                response: response.to_vec(),
                read_pos: 0,
                flushed: Cell::new(false),
            }
        }
        fn was_flushed(&self) -> bool {
            self.flushed.get()
        }
    }

    impl hyper::rt::Write for FlushTrackingStream {
        fn poll_write(
            self: Pin<&mut Self>,
            _cx: &mut Context<'_>,
            buf: &[u8],
        ) -> Poll<std::io::Result<usize>> {
            Poll::Ready(Ok(buf.len()))
        }
        fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
            self.flushed.set(true);
            Poll::Ready(Ok(()))
        }
        fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
            Poll::Ready(Ok(()))
        }
    }

    impl hyper::rt::Read for FlushTrackingStream {
        fn poll_read(
            self: Pin<&mut Self>,
            _cx: &mut Context<'_>,
            mut buf: hyper::rt::ReadBufCursor<'_>,
        ) -> Poll<std::io::Result<()>> {
            let this = self.get_mut();
            // Assert flush was called before any read
            assert!(
                this.flushed.get(),
                "poll_flush must be called before poll_read"
            );
            if this.read_pos < this.response.len() {
                let remaining = &this.response[this.read_pos..];
                let to_copy = remaining.len().min(buf.remaining());
                let dest = unsafe { buf.as_mut() };
                // Manually copy from initialized bytes into MaybeUninit buffer
                for (i, &byte) in remaining[..to_copy].iter().enumerate() {
                    dest[i].write(byte);
                }
                unsafe { buf.advance(to_copy) };
                this.read_pos += to_copy;
            }
            Poll::Ready(Ok(()))
        }
    }

    #[test]
    fn connect_handshake_flushes_before_read() {
        let rt = tokio::runtime::Runtime::new().unwrap();
        rt.block_on(async {
            let proxy = crate::ProxyConfig::http("http://proxy:8080").unwrap();
            let response = b"HTTP/1.1 200 Connection Established\r\n\r\n";
            let stream = FlushTrackingStream::new(response);

            let result = do_connect_handshake_local(stream, &proxy, "example.com:80").await;
            assert!(
                result.is_ok(),
                "handshake should succeed: {:?}",
                result.err()
            );
            assert!(
                result.unwrap().was_flushed(),
                "stream should have been flushed"
            );
        });
    }

    #[test]
    fn connect_handshake_flush_failure_propagates() {
        let rt = tokio::runtime::Runtime::new().unwrap();
        rt.block_on(async {
            let proxy = crate::ProxyConfig::http("http://proxy:8080").unwrap();

            // Stream that fails on flush
            #[derive(Debug)]
            struct FlushFailingStream;
            impl hyper::rt::Write for FlushFailingStream {
                fn poll_write(
                    self: Pin<&mut Self>,
                    _cx: &mut Context<'_>,
                    buf: &[u8],
                ) -> Poll<std::io::Result<usize>> {
                    Poll::Ready(Ok(buf.len()))
                }
                fn poll_flush(
                    self: Pin<&mut Self>,
                    _cx: &mut Context<'_>,
                ) -> Poll<std::io::Result<()>> {
                    Poll::Ready(Err(std::io::Error::new(
                        std::io::ErrorKind::BrokenPipe,
                        "flush failed",
                    )))
                }
                fn poll_shutdown(
                    self: Pin<&mut Self>,
                    _cx: &mut Context<'_>,
                ) -> Poll<std::io::Result<()>> {
                    Poll::Ready(Ok(()))
                }
            }
            impl hyper::rt::Read for FlushFailingStream {
                fn poll_read(
                    self: Pin<&mut Self>,
                    _cx: &mut Context<'_>,
                    _buf: hyper::rt::ReadBufCursor<'_>,
                ) -> Poll<std::io::Result<()>> {
                    Poll::Ready(Ok(()))
                }
            }

            let result =
                do_connect_handshake_local(FlushFailingStream, &proxy, "example.com:80").await;
            assert!(result.is_err());
            assert!(
                format!("{}", result.unwrap_err()).contains("flush"),
                "error should mention flush failure"
            );
        });
    }

    #[test]
    fn connect_local_uses_same_parse_connect_status() {
        // connect_local.rs uses super::connect::parse_connect_status
        // Verify it's accessible and correct
        let result = super::super::connect::parse_connect_status("HTTP/1.1 200 OK");
        assert_eq!(result.unwrap(), 200);
    }
}
