use std::future::Future;
use std::io;
use std::net::SocketAddr;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::Duration;

use hyper::rt::{self, Read, Write};
use pin_project_lite::pin_project;

use super::{ConnectorLocal, RuntimeCompletion, RuntimePoll};

/// Tokio async runtime implementation.
pub struct TokioRuntime;

// ── New trait impls (v0.2) ──────────────────────────────────────────────────

impl RuntimeCompletion for TokioRuntime {
    type Sleep = TokioSleep;

    fn sleep(duration: Duration) -> Self::Sleep {
        TokioSleep {
            inner: tokio::time::sleep(duration),
        }
    }

    fn block_on<F: Future>(future: F) -> Result<F::Output, crate::error::Error> {
        if tokio::runtime::Handle::try_current().is_ok() {
            return Err(crate::error::Error::Other(
                "blocking client cannot be used from within an async runtime".into(),
            ));
        }
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .map_err(crate::error::Error::Io)?;
        Ok(rt.block_on(future))
    }
}

impl RuntimePoll for TokioRuntime {
    fn spawn_send<F: Future<Output = ()> + Send + 'static>(future: F) {
        tokio::spawn(future);
    }
}

// ── SocketConfig ──────────────────────────────────────────────────────────

impl super::SocketConfig for TokioIo<tokio::net::TcpStream> {
    fn set_keepalive(
        &self,
        time: Duration,
        interval: Option<Duration>,
        retries: Option<u32>,
    ) -> io::Result<()> {
        use socket2::SockRef;
        let sock_ref = SockRef::from(self.inner());
        let mut keepalive = socket2::TcpKeepalive::new().with_time(time);
        if let Some(interval) = interval {
            keepalive = keepalive.with_interval(interval);
        }
        #[cfg(any(
            target_os = "linux",
            target_os = "macos",
            target_os = "ios",
            target_os = "freebsd",
            target_os = "netbsd",
        ))]
        if let Some(retries) = retries {
            keepalive = keepalive.with_retries(retries);
        }
        #[cfg(not(any(
            target_os = "linux",
            target_os = "macos",
            target_os = "ios",
            target_os = "freebsd",
            target_os = "netbsd",
        )))]
        let _ = retries;
        sock_ref.set_tcp_keepalive(&keepalive)
    }

    #[cfg(target_os = "linux")]
    fn set_fast_open(&self) -> io::Result<()> {
        use socket2::SockRef;
        use std::os::unix::io::AsRawFd;

        unsafe extern "C" {
            fn setsockopt(
                sockfd: std::ffi::c_int,
                level: std::ffi::c_int,
                optname: std::ffi::c_int,
                optval: *const std::ffi::c_void,
                optlen: u32,
            ) -> std::ffi::c_int;
        }

        let sock_ref = SockRef::from(self.inner());
        let fd = sock_ref.as_raw_fd();
        const IPPROTO_TCP: std::ffi::c_int = 6;
        const TCP_FASTOPEN_CONNECT: std::ffi::c_int = 30;
        let optval: std::ffi::c_int = 1;
        unsafe {
            let ret = setsockopt(
                fd,
                IPPROTO_TCP,
                TCP_FASTOPEN_CONNECT,
                &optval as *const std::ffi::c_int as *const std::ffi::c_void,
                std::mem::size_of::<std::ffi::c_int>() as u32,
            );
            if ret != 0 {
                return Err(io::Error::last_os_error());
            }
        }
        Ok(())
    }

    #[cfg(target_os = "linux")]
    fn bind_device(&self, interface: &str) -> io::Result<()> {
        use socket2::SockRef;
        let sock_ref = SockRef::from(self.inner());
        sock_ref.bind_device(Some(interface.as_bytes()))
    }
}

// ── TcpConnector ──────────────────────────────────────────────────────────

/// TCP connector for the Tokio runtime.
#[derive(Clone, Copy, Default)]
pub struct TcpConnector;

impl ConnectorLocal for TcpConnector {
    type Stream = TokioIo<tokio::net::TcpStream>;

    async fn connect(&self, addr: SocketAddr) -> io::Result<Self::Stream> {
        let stream = tokio::net::TcpStream::connect(addr).await?;
        stream.set_nodelay(true)?;
        Ok(TokioIo::new(stream))
    }

    async fn connect_bound(
        &self,
        addr: SocketAddr,
        local: std::net::IpAddr,
    ) -> io::Result<Self::Stream> {
        let socket = if addr.is_ipv4() {
            tokio::net::TcpSocket::new_v4()?
        } else {
            tokio::net::TcpSocket::new_v6()?
        };
        socket.bind(std::net::SocketAddr::new(local, 0))?;
        let stream = socket.connect(addr).await?;
        stream.set_nodelay(true)?;
        Ok(TokioIo::new(stream))
    }

    fn from_std_tcp(&self, stream: std::net::TcpStream) -> io::Result<Self::Stream> {
        stream.set_nonblocking(true)?;
        stream.set_nodelay(true)?;
        let tokio_stream = tokio::net::TcpStream::from_std(stream)?;
        Ok(TokioIo::new(tokio_stream))
    }

    fn into_std_tcp(&self, stream: Self::Stream) -> io::Result<std::net::TcpStream> {
        let std_stream = stream.into_inner().into_std()?;
        std_stream.set_nonblocking(false)?;
        Ok(std_stream)
    }
}

#[allow(clippy::manual_async_fn)]
impl super::ConnectorSend for TcpConnector {
    type Stream = TokioIo<tokio::net::TcpStream>;

    fn connect(&self, addr: SocketAddr) -> impl Future<Output = io::Result<Self::Stream>> + Send {
        async move {
            let stream = tokio::net::TcpStream::connect(addr).await?;
            stream.set_nodelay(true)?;
            Ok(TokioIo::new(stream))
        }
    }

    fn connect_bound(
        &self,
        addr: SocketAddr,
        local: std::net::IpAddr,
    ) -> impl Future<Output = io::Result<Self::Stream>> + Send {
        async move {
            let socket = if addr.is_ipv4() {
                tokio::net::TcpSocket::new_v4()?
            } else {
                tokio::net::TcpSocket::new_v6()?
            };
            socket.bind(std::net::SocketAddr::new(local, 0))?;
            let stream = socket.connect(addr).await?;
            stream.set_nodelay(true)?;
            Ok(TokioIo::new(stream))
        }
    }

    fn from_std_tcp(&self, stream: std::net::TcpStream) -> io::Result<Self::Stream> {
        stream.set_nonblocking(true)?;
        stream.set_nodelay(true)?;
        let tokio_stream = tokio::net::TcpStream::from_std(stream)?;
        Ok(TokioIo::new(tokio_stream))
    }

    fn into_std_tcp(&self, stream: Self::Stream) -> io::Result<std::net::TcpStream> {
        let std_stream = stream.into_inner().into_std()?;
        std_stream.set_nonblocking(false)?;
        Ok(std_stream)
    }
}

// ── DefaultResolver ───────────────────────────────────────────────────────

/// Default DNS resolver using Tokio's `lookup_host`.
pub struct DefaultResolver;

impl super::Resolve for DefaultResolver {
    fn resolve(
        &self,
        host: &str,
        port: u16,
    ) -> Pin<Box<dyn Future<Output = io::Result<SocketAddr>> + Send>> {
        let addr = format!("{host}:{port}");
        Box::pin(async move {
            tokio::net::lookup_host(addr).await?.next().ok_or_else(|| {
                io::Error::new(io::ErrorKind::AddrNotAvailable, "no addresses found")
            })
        })
    }

    fn resolve_all(
        &self,
        host: &str,
        port: u16,
    ) -> Pin<Box<dyn Future<Output = io::Result<Vec<SocketAddr>>> + Send>> {
        let addr = format!("{host}:{port}");
        Box::pin(async move {
            let addrs: Vec<SocketAddr> = tokio::net::lookup_host(addr).await?.collect();
            if addrs.is_empty() {
                return Err(io::Error::new(
                    io::ErrorKind::AddrNotAvailable,
                    "no addresses found",
                ));
            }
            Ok(addrs)
        })
    }
}

// -- TokioSleep: bridges hyper::rt::Sleep to tokio::time::Sleep --

pin_project! {
    /// Tokio-backed sleep future.
    pub struct TokioSleep {
        #[pin]
        inner: tokio::time::Sleep,
    }
}

impl Future for TokioSleep {
    type Output = ();

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        self.project().inner.poll(cx)
    }
}

// -- TokioIo: bridges tokio::io::{AsyncRead, AsyncWrite} to hyper::rt::{Read, Write} --

pin_project! {
    /// Adapter bridging tokio's `AsyncRead`/`AsyncWrite` to hyper's `Read`/`Write`.
    pub struct TokioIo<T> {
        #[pin]
        inner: T,
    }
}

impl<T> TokioIo<T> {
    /// Wrap a tokio I/O type.
    pub fn new(inner: T) -> Self {
        Self { inner }
    }

    /// Get a reference to the inner I/O type.
    pub fn inner(&self) -> &T {
        &self.inner
    }

    /// Consume the adapter and return the inner I/O type.
    pub fn into_inner(self) -> T {
        self.inner
    }
}

impl<T> Read for TokioIo<T>
where
    T: tokio::io::AsyncRead,
{
    fn poll_read(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        mut buf: rt::ReadBufCursor<'_>,
    ) -> Poll<io::Result<()>> {
        let n = unsafe {
            let mut tbuf = tokio::io::ReadBuf::uninit(buf.as_mut());
            match tokio::io::AsyncRead::poll_read(self.project().inner, cx, &mut tbuf) {
                Poll::Ready(Ok(())) => tbuf.filled().len(),
                other => return other,
            }
        };
        unsafe {
            buf.advance(n);
        }
        Poll::Ready(Ok(()))
    }
}

impl<T> Write for TokioIo<T>
where
    T: tokio::io::AsyncWrite,
{
    fn poll_write(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &[u8],
    ) -> Poll<io::Result<usize>> {
        tokio::io::AsyncWrite::poll_write(self.project().inner, cx, buf)
    }

    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
        tokio::io::AsyncWrite::poll_flush(self.project().inner, cx)
    }

    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
        tokio::io::AsyncWrite::poll_shutdown(self.project().inner, cx)
    }

    fn poll_write_vectored(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        bufs: &[io::IoSlice<'_>],
    ) -> Poll<io::Result<usize>> {
        tokio::io::AsyncWrite::poll_write_vectored(self.project().inner, cx, bufs)
    }

    fn is_write_vectored(&self) -> bool {
        tokio::io::AsyncWrite::is_write_vectored(&self.inner)
    }
}

#[cfg(test)]
#[allow(deprecated)]
mod tests {
    use super::*;

    // ── New trait tests (v0.2) ──────────────────────────────────────────────

    #[tokio::test]
    async fn runtime_completion_sleep() {
        use crate::runtime::RuntimeCompletion;
        let start = std::time::Instant::now();
        <TokioRuntime as RuntimeCompletion>::sleep(Duration::from_millis(10)).await;
        assert!(start.elapsed() >= Duration::from_millis(10));
    }

    #[tokio::test]
    async fn runtime_poll_spawn_send() {
        use crate::runtime::RuntimePoll;
        use std::sync::Arc;
        use std::sync::atomic::{AtomicBool, Ordering};
        let flag = Arc::new(AtomicBool::new(false));
        let flag2 = flag.clone();
        TokioRuntime::spawn_send(async move {
            flag2.store(true, Ordering::SeqCst);
        });
        tokio::task::yield_now().await;
        assert!(flag.load(Ordering::SeqCst));
    }

    #[tokio::test]
    async fn connector_connect_works() {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let connector = super::TcpConnector;
        let stream = connector.connect(addr).await.unwrap();
        assert!(Write::is_write_vectored(&stream));
    }

    // ── ConnectorSend & SocketConfig & Resolver trait tests ────────────────

    #[tokio::test]
    async fn connector_send_connect_works() {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let connector = TcpConnector;
        let stream = <TcpConnector as crate::runtime::ConnectorSend>::connect(&connector, addr)
            .await
            .unwrap();
        drop(stream);
    }

    #[tokio::test]
    async fn connector_send_connect_bound_works() {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let connector = TcpConnector;
        let stream = <TcpConnector as crate::runtime::ConnectorSend>::connect_bound(
            &connector,
            addr,
            "127.0.0.1".parse().unwrap(),
        )
        .await
        .unwrap();
        drop(stream);
    }

    #[tokio::test]
    async fn connector_send_from_std_tcp_works() {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let std_stream = std::net::TcpStream::connect(addr).unwrap();
        std_stream.set_nonblocking(true).unwrap();
        let connector = TcpConnector;
        let result =
            <TcpConnector as crate::runtime::ConnectorSend>::from_std_tcp(&connector, std_stream);
        assert!(result.is_ok());
    }

    #[test]
    fn connector_send_connect_future_is_send() {
        fn assert_send<T: Send>(_: &T) {}
        let connector = TcpConnector;
        let addr: std::net::SocketAddr = "127.0.0.1:1".parse().unwrap();
        let fut = <TcpConnector as crate::runtime::ConnectorSend>::connect(&connector, addr);
        assert_send(&fut);
    }

    #[tokio::test]
    async fn socket_config_set_keepalive_via_trait() {
        use crate::runtime::SocketConfig;
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let connector = TcpConnector;
        let stream = <TcpConnector as crate::runtime::ConnectorSend>::connect(&connector, addr)
            .await
            .unwrap();
        let result = stream.set_keepalive(
            Duration::from_secs(60),
            Some(Duration::from_secs(10)),
            Some(3),
        );
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn socket_config_set_fast_open_via_trait() {
        use crate::runtime::SocketConfig;
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let connector = TcpConnector;
        let stream = <TcpConnector as crate::runtime::ConnectorSend>::connect(&connector, addr)
            .await
            .unwrap();
        // On Linux this calls setsockopt(TCP_FASTOPEN_CONNECT) which may return
        // ENOPROTOOPT if the kernel doesn't support TFO. On other platforms the
        // default trait impl returns Ok. Either way, it must not panic.
        let _ = stream.set_fast_open();
    }

    #[tokio::test]
    async fn default_resolver_resolves_localhost() {
        use crate::runtime::Resolve;
        let resolver = DefaultResolver;
        let addr = resolver.resolve("localhost", 80).await.unwrap();
        assert!(addr.ip().is_loopback());
    }

    #[tokio::test]
    async fn default_resolver_resolve_all_localhost() {
        use crate::runtime::Resolve;
        let resolver = DefaultResolver;
        let addrs = resolver.resolve_all("localhost", 80).await.unwrap();
        assert!(!addrs.is_empty());
        assert!(addrs.iter().all(|a| a.ip().is_loopback()));
    }

    #[tokio::test]
    async fn default_resolver_invalid_host_errors() {
        use crate::runtime::Resolve;
        let resolver = DefaultResolver;
        let result = resolver
            .resolve("this.host.does.not.exist.invalid", 80)
            .await;
        assert!(result.is_err());
    }

    // ── block_on tests ─────────────────────────────────────────────────

    #[test]
    fn block_on_returns_value() {
        use crate::runtime::RuntimeCompletion;
        let result = TokioRuntime::block_on(async { 123 + 456 }).unwrap();
        assert_eq!(result, 579);
    }

    #[test]
    fn block_on_runs_async_io() {
        use crate::runtime::RuntimeCompletion;
        let result = TokioRuntime::block_on(async {
            let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
            listener.local_addr().unwrap().port()
        })
        .unwrap();
        assert!(result > 0);
    }

    #[tokio::test]
    async fn block_on_inside_runtime_returns_error() {
        use crate::runtime::RuntimeCompletion;
        let result = TokioRuntime::block_on(async { 42 });
        assert!(result.is_err());
        let err_msg = result.unwrap_err().to_string();
        assert!(err_msg.contains("blocking client cannot be used from within an async runtime"));
    }

    // ── TokioIo read/write edge cases ──────────────────────────────────

    #[tokio::test]
    async fn tokio_io_read_eof_returns_zero_advance() {
        use std::future::poll_fn;
        let (client, server) = tokio::io::duplex(1024);
        drop(server); // immediately close the write end

        let mut io = TokioIo::new(client);
        let mut buf = [0u8; 64];
        let mut read_buf = hyper::rt::ReadBuf::new(&mut buf);

        poll_fn(|cx| Pin::new(&mut io).poll_read(cx, read_buf.unfilled()))
            .await
            .unwrap();
        assert_eq!(
            read_buf.filled().len(),
            0,
            "EOF should produce 0 filled bytes"
        );
    }

    #[tokio::test]
    async fn tokio_io_write_and_flush() {
        use std::future::poll_fn;
        use tokio::io::AsyncReadExt;

        let (client, mut server) = tokio::io::duplex(1024);
        let mut io = TokioIo::new(client);

        let data = b"tokio io test";
        let n = poll_fn(|cx| Pin::new(&mut io).poll_write(cx, data))
            .await
            .unwrap();
        assert_eq!(n, data.len());

        poll_fn(|cx| Pin::new(&mut io).poll_flush(cx))
            .await
            .unwrap();

        let mut buf = vec![0u8; data.len()];
        server.read_exact(&mut buf).await.unwrap();
        assert_eq!(&buf, data);
    }

    #[tokio::test]
    async fn tokio_io_shutdown_closes_write() {
        use std::future::poll_fn;
        use tokio::io::AsyncReadExt;

        let (client, mut server) = tokio::io::duplex(1024);
        let mut io = TokioIo::new(client);

        poll_fn(|cx| Pin::new(&mut io).poll_shutdown(cx))
            .await
            .unwrap();

        // Server should see EOF
        let mut buf = [0u8; 1];
        let n = server.read(&mut buf).await.unwrap();
        assert_eq!(n, 0);
    }

    // ── ConnectorLocal connect_bound IPv6 path ──────────────────────────────

    #[tokio::test]
    async fn connector_connect_bound_works_local() {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let connector = TcpConnector;
        let stream = <TcpConnector as crate::runtime::ConnectorLocal>::connect_bound(
            &connector,
            addr,
            "127.0.0.1".parse().unwrap(),
        )
        .await
        .unwrap();
        assert!(stream.inner().peer_addr().is_ok());
    }

    // ── TokioSleep ─────────────────────────────────────────────────────

    #[tokio::test]
    async fn tokio_sleep_completes() {
        use crate::runtime::RuntimeCompletion;
        let start = std::time::Instant::now();
        <TokioRuntime as RuntimeCompletion>::sleep(Duration::from_millis(20)).await;
        assert!(start.elapsed() >= Duration::from_millis(20));
    }

    // ── DefaultResolver resolve_all ────────────────────────────────────

    #[tokio::test]
    async fn default_resolver_resolve_all_returns_multiple() {
        use crate::runtime::Resolve;
        let resolver = DefaultResolver;
        let addrs = resolver.resolve_all("localhost", 80).await.unwrap();
        assert!(!addrs.is_empty());
        for addr in &addrs {
            assert_eq!(addr.port(), 80);
        }
    }

    #[tokio::test]
    async fn default_resolver_resolve_all_invalid_host_errors() {
        use crate::runtime::Resolve;
        let resolver = DefaultResolver;
        let result = resolver
            .resolve_all("this.host.does.not.exist.invalid", 80)
            .await;
        assert!(result.is_err());
    }

    // ── ConnectorLocal from_std_tcp (ConnectorLocal trait, not ConnectorSend) ────

    #[tokio::test]
    async fn connector_trait_from_std_tcp_works() {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let std_stream = std::net::TcpStream::connect(addr).unwrap();
        let connector = TcpConnector;
        let result =
            <TcpConnector as crate::runtime::ConnectorLocal>::from_std_tcp(&connector, std_stream);
        assert!(result.is_ok());
        let stream = result.unwrap();
        assert!(stream.inner().peer_addr().is_ok());
    }

    // ── ConnectorLocal connect_bound IPv6 path ─────────────────────────────

    #[tokio::test]
    async fn connector_connect_bound_ipv6() {
        // Bind a listener on IPv6 loopback
        let listener = match tokio::net::TcpListener::bind("[::1]:0").await {
            Ok(l) => l,
            Err(_) => return, // Skip if IPv6 not available
        };
        let addr = listener.local_addr().unwrap();
        let connector = TcpConnector;
        let stream = <TcpConnector as crate::runtime::ConnectorLocal>::connect_bound(
            &connector,
            addr,
            "::1".parse().unwrap(),
        )
        .await
        .unwrap();
        assert!(stream.inner().peer_addr().is_ok());
        assert!(stream.inner().peer_addr().unwrap().is_ipv6());
    }

    // ── ConnectorSend connect_bound IPv6 path ─────────────────────────

    #[tokio::test]
    async fn connector_send_connect_bound_ipv6() {
        // Bind a listener on IPv6 loopback
        let listener = match tokio::net::TcpListener::bind("[::1]:0").await {
            Ok(l) => l,
            Err(_) => return, // Skip if IPv6 not available
        };
        let addr = listener.local_addr().unwrap();
        let connector = TcpConnector;
        let stream = <TcpConnector as crate::runtime::ConnectorSend>::connect_bound(
            &connector,
            addr,
            "::1".parse().unwrap(),
        )
        .await
        .unwrap();
        assert!(stream.inner().peer_addr().is_ok());
        assert!(stream.inner().peer_addr().unwrap().is_ipv6());
    }
}
