//! Tests for the `local` feature: real-socket halves through the real
//! establisher path (`local_dial` injected as the `DialFn`), TCP/UDP/
//! unix round-trips against real local echo servers, the OQ-TN-13
//! truncation fail-loud probe, and the 1400-byte MTU datagram.
//!
//! The forward topology (`wire_forward`) carries the full path: the
//! consumer opens (`TunnelSession::open`), the producer's establisher
//! dials via `local_dial`, the pump handler pumps — real sockets at
//! both ends.

#![cfg(feature = "local")]

mod harness;

use std::sync::Arc;

use alktunnels::params::{Substrate, TunnelParams};
use alktunnels::producer::ResourceRegistry;
use alktunnels::TunnelSession;

use harness::wire_forward;

fn params(resource: &str, substrate: Substrate) -> TunnelParams {
    TunnelParams {
        resource: resource.to_string(),
        substrate,
    }
}

/// A TCP echo server on an ephemeral port.
async fn tcp_echo_server() -> std::net::SocketAddr {
    let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0))
        .await
        .expect("bind echo");
    let addr = listener.local_addr().expect("addr");
    tokio::spawn(async move {
        loop {
            let Ok((sock, _)) = listener.accept().await else {
                return;
            };
            tokio::spawn(async move {
                let (mut r, mut w) = tokio::io::split(sock);
                let _ = tokio::io::copy(&mut r, &mut w).await;
            });
        }
    });
    addr
}

/// A UDP echo server on an ephemeral port (one task, echo loop).
async fn udp_echo_server() -> std::net::SocketAddr {
    let sock = tokio::net::UdpSocket::bind(("127.0.0.1", 0))
        .await
        .expect("bind udp echo");
    let addr = sock.local_addr().expect("addr");
    tokio::spawn(async move {
        let mut buf = vec![0u8; 65535];
        loop {
            let Ok((n, peer)) = sock.recv_from(&mut buf).await else {
                return;
            };
            if sock.send_to(&buf[..n], peer).await.is_err() {
                return;
            }
        }
    });
    addr
}

#[tokio::test]
async fn tcp_dial_round_trips_through_the_real_establisher_path() {
    let addr = tcp_echo_server().await;
    let registry = ResourceRegistry::new();
    registry
        .register("echo", Substrate::Tcp, &addr.to_string())
        .await;
    let topo = wire_forward(registry, alktunnels::local::local_dial()).await;

    let mut session = TunnelSession::open(&topo.consumer, params("echo", Substrate::Tcp))
        .await
        .expect("tcp open");
    use tokio::io::{AsyncReadExt, AsyncWriteExt};
    let (read, write) = session.stream_halves().expect("halves");
    write.write_all(b"tcp-local").await.expect("write");
    write.flush().await.expect("flush");
    let mut buf = vec![0u8; 9];
    tokio::time::timeout(std::time::Duration::from_secs(5), read.read_exact(&mut buf))
        .await
        .expect("round trip")
        .expect("read");
    assert_eq!(&buf, b"tcp-local");
    session.close().await;
}

#[tokio::test]
async fn udp_datagram_round_trip_through_the_framed_adapter() {
    // The FRAMED adapter end-to-end (ADR-003): session sends the codec
    // frame; the pump copies raw bytes; the dialed UdpHalf strips the
    // frame on recv and delivers ONE datagram to the target; the echo
    // comes back through the same path.
    let addr = udp_echo_server().await;
    let registry = ResourceRegistry::new();
    registry
        .register("dns", Substrate::Udp, &addr.to_string())
        .await;
    let topo = wire_forward(registry, alktunnels::local::local_dial()).await;

    let mut session = TunnelSession::open(&topo.consumer, params("dns", Substrate::Udp))
        .await
        .expect("udp open");
    session.send_datagram(b"query").await.expect("send");
    let dg = tokio::time::timeout(std::time::Duration::from_secs(5), session.recv_datagram())
        .await
        .expect("recv timed out")
        .expect("io ok")
        .expect("datagram");
    assert_eq!(&dg[..], b"query");
    session.close().await;
}

#[tokio::test]
async fn udp_large_datagram_through_bounded_buffers() {
    // The POC's sizing test: a 1400-byte datagram (max ethernet MTU
    // payload) round-trips through the codec + the bounded-buffer path.
    let addr = udp_echo_server().await;
    let registry = ResourceRegistry::new();
    registry
        .register("big", Substrate::Udp, &addr.to_string())
        .await;
    let topo = wire_forward(registry, alktunnels::local::local_dial()).await;

    let mut session = TunnelSession::open(&topo.consumer, params("big", Substrate::Udp))
        .await
        .expect("udp open");
    let payload: Vec<u8> = (0..1400u32).map(|i| (i % 7) as u8).collect();
    session.send_datagram(&payload).await.expect("send");
    let got = tokio::time::timeout(std::time::Duration::from_secs(5), session.recv_datagram())
        .await
        .expect("recv timed out")
        .expect("io ok")
        .expect("datagram");
    assert_eq!(got.len(), payload.len());
    assert_eq!(&got[..], &payload[..]);
    session.close().await;
}

#[tokio::test]
async fn unix_dial_round_trips() {
    // OQ-TN-14: unix ships with local v1 (same halves shape as TCP).
    let dir = std::env::temp_dir().join(format!("alktunnels-test-{}", std::process::id()));
    let _ = std::fs::create_dir_all(&dir);
    let path = dir.join("echo.sock");
    let _ = std::fs::remove_file(&path);

    let listener = tokio::net::UnixListener::bind(&path).expect("bind unix echo");
    tokio::spawn(async move {
        loop {
            let Ok((sock, _)) = listener.accept().await else {
                return;
            };
            tokio::spawn(async move {
                let (mut r, mut w) = tokio::io::split(sock);
                let _ = tokio::io::copy(&mut r, &mut w).await;
            });
        }
    });

    let registry = ResourceRegistry::new();
    registry
        .register("uds", Substrate::Unix, &path.to_string_lossy())
        .await;
    let topo = wire_forward(registry, alktunnels::local::local_dial()).await;

    let mut session = TunnelSession::open(&topo.consumer, params("uds", Substrate::Unix))
        .await
        .expect("unix open");
    use tokio::io::{AsyncReadExt, AsyncWriteExt};
    let (read, write) = session.stream_halves().expect("halves");
    write.write_all(b"unix-local").await.expect("write");
    write.flush().await.expect("flush");
    let mut buf = vec![0u8; 10];
    tokio::time::timeout(std::time::Duration::from_secs(5), read.read_exact(&mut buf))
        .await
        .expect("round trip")
        .expect("read");
    assert_eq!(&buf, b"unix-local");
    session.close().await;
    let _ = std::fs::remove_file(&path);
}

#[tokio::test]
async fn truncation_fails_loud_oq_tn_13() {
    // A receive buffer smaller than the datagram surfaces as an error
    // (never a silent truncation). Probed at the adapter level: feed a
    // >buffer datagram directly into a UdpHalf pair and read with a
    // small buffer.
    let sock_a = tokio::net::UdpSocket::bind(("127.0.0.1", 0))
        .await
        .expect("bind a");
    let sock_b = tokio::net::UdpSocket::bind(("127.0.0.1", 0))
        .await
        .expect("bind b");
    sock_a
        .connect(sock_b.local_addr().expect("b addr"))
        .await
        .expect("connect a");
    sock_b
        .connect(sock_a.local_addr().expect("a addr"))
        .await
        .expect("connect b");

    // The producer-side adapter (what the establisher dials): sock_a
    // wrapped as halves. Send a large datagram from the "wire" side.
    let (read, _write) = alktunnels::local::UdpHalf::split(sock_a);
    let mut read = read;
    let payload = vec![0u8; 60000]; // well-formed for UDP but big
    sock_b.send(&payload).await.expect("send big");

    // The pump reads with the codec's chunk size (16 KiB) — smaller
    // than 60000: the adapter must FAIL LOUD, not truncate.
    use tokio::io::AsyncReadExt;
    let mut chunk = vec![0u8; 16 * 1024];
    let result =
        tokio::time::timeout(std::time::Duration::from_secs(5), read.read(&mut chunk)).await;
    match result {
        Ok(Err(e)) => {
            assert_eq!(e.kind(), std::io::ErrorKind::InvalidData);
        }
        other => panic!("expected InvalidData error, got {other:?}"),
    }
}

#[tokio::test]
async fn dial_refusal_maps_to_dial_failed() {
    // Error mapping: a refused dial (nothing listening on the port) is
    // `dial_failed` on the wire (`channel:open_failed` reason).
    let registry = ResourceRegistry::new();
    registry
        .register("dead", Substrate::Tcp, "127.0.0.1:1")
        .await;
    let topo = wire_forward(registry, alktunnels::local::local_dial()).await;

    let err = match TunnelSession::open(&topo.consumer, params("dead", Substrate::Tcp)).await {
        Ok(_) => panic!("refused dial must fail"),
        Err(e) => e,
    };
    assert_eq!(err.open_ref().establishment_reason(), Some("dial_failed"));
}

#[tokio::test]
async fn listen_producer_over_a_real_tcp_listener() {
    // The listen shape + local halves compose: a real TcpListenerHalves
    // accept loop feeds an AcceptQueue; a listen-registered producer
    // pops accepted sockets; the consumer pumps the tunnel against the
    // accepted local socket.
    let queue = alktunnels::producer::AcceptQueue::new();
    let listener = alktunnels::local::bind_tcp("127.0.0.1:0")
        .await
        .expect("bind");
    let addr = listener.local_addr().expect("addr");

    let accept_loop = tokio::spawn({
        let queue = queue.clone();
        async move { listener.accept_loop(queue).await }
    });

    // A local client connects (the accepted connection's source). The
    // driving end comes from `try_clone` — the pump owns the accepted
    // handle; the driver pushes/reads the echo through the same real
    // socket.
    // std-level try_clone BEFORE converting to tokio streams: one fd
    // for the driver, one wrapped for the pump (two independent
    // handles to the same socket). O_NONBLOCK is shared across the
    // dup, so BOTH ends go nonblocking and the driver is driven
    // async too.
    let local_std = std::net::TcpStream::connect(addr).expect("std connect");
    let driver_std = local_std.try_clone().expect("try_clone driver");
    local_std.set_nonblocking(true).expect("nonblocking");
    let mut driver = tokio::net::TcpStream::from_std(driver_std).expect("tokio driver");

    let registry = ResourceRegistry::new();
    registry
        .register("listener", Substrate::Tcp, "assembly-owned")
        .await;
    let accept_fn: alktunnels::producer::AcceptFn = Arc::new(move || {
        let queue = queue.clone();
        Box::pin(async move {
            queue.pop().await.ok_or_else(|| {
                alktunnels::producer::TunnelEstablishError::DialFailed(
                    "listener closed".to_string(),
                )
            })
        })
    });
    let topo = harness::wire_listen(registry, accept_fn).await;

    let channel_id = alktunnels::open_reverse_channel(
        &topo.consumer_call,
        &params("listener", Substrate::Tcp),
        None,
    )
    .await
    .expect("listen open");

    let session = TunnelSession::adopt(
        &topo.consumer_manager,
        channel_id,
        Substrate::Tcp,
        alktunnels::TUNNEL_ALPN,
    )
    .await
    .expect("adopt");
    // The producer's pump (wrapper-managed) copies channel <-> the
    // accepted handle; the accepted handle IS the client socket — so
    // the session's borrowed halves drive it directly: the driver end
    // writes into the client socket, the pump carries it over the
    // channel, and the session halves read it (and back).
    let mut session = session;
    use tokio::io::{AsyncReadExt, AsyncWriteExt};
    driver.write_all(b"listen-local").await.expect("write");
    driver.flush().await.expect("flush");
    let (read, _write) = session.stream_halves().expect("halves");
    let mut buf = vec![0u8; 12];
    tokio::time::timeout(std::time::Duration::from_secs(5), read.read_exact(&mut buf))
        .await
        .expect("round trip")
        .expect("read");
    assert_eq!(&buf, b"listen-local");

    session.close().await;
    accept_loop.abort();
}