//! The UDP front door's integration tests (feature `local`): the POC
//! #2 test-8 shape end to end: a real UDP echo target, the producer
//! with `local::udp_egress()`, the door's relay socket, a vanilla
//! RFC-header'd datagram in → the echo back RFC-header'd (the RFC
//! codec boundary pinned: the in-band wire stays `{addr, data}`
//! frames). FRAG ≠ 0 refused (dropped, association healthy); the
//! empty-datagram invariant through the door (F-2); the idle bound
//! closes an idle association; a fresh association succeeds after
//! teardown.
#![cfg(feature = "local")]

mod harness;

fn init_trace() {
    let _ = tracing_subscriber::fmt()
        .with_max_level(tracing::level_filters::LevelFilter::DEBUG)
        .with_test_writer()
        .try_init();
}

use std::net::SocketAddr;
use std::sync::Arc;
use std::time::Duration;

use tokio::net::UdpSocket;
use tokio::time::timeout;

use alksocks::UdpFrontDoor;

use harness::echo_dial;

const BOUND: Duration = Duration::from_secs(10);

/// A UDP echo server on a real kernel socket; returns its address
/// (the vanilla client's datagram target, the POC #2 shape).
async fn udp_echo_server() -> SocketAddr {
    let sock = 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
}

/// Build an RFC-header'd datagram (the vanilla client's sendto form;
/// the codec appears at the door boundary, so the test hand-rolls the
/// same header: RSV(2) FRAG(1) ATYP(1) addr port).
fn rfc_datagram(target: SocketAddr, payload: &[u8]) -> Vec<u8> {
    let SocketAddr::V4(v4) = target else {
        panic!("v4 targets only");
    };
    let mut dg = vec![0u8, 0, 0, 0x01];
    dg.extend_from_slice(&v4.ip().octets());
    dg.extend_from_slice(&v4.port().to_be_bytes());
    dg.extend_from_slice(payload);
    dg
}

/// Parse an RFC-header'd datagram back → (addr, payload).
fn parse_rfc_datagram(wire: &[u8]) -> (SocketAddr, &[u8]) {
    assert_eq!(&wire[..2], &[0, 0], "RSV zero");
    assert_eq!(wire[3], 0x01, "v4 ATYP");
    let ip = std::net::Ipv4Addr::new(wire[4], wire[5], wire[6], wire[7]);
    let port = u16::from_be_bytes([wire[8], wire[9]]);
    (SocketAddr::from((ip, port)), &wire[10..])
}

/// The happy path (the POC #2 test-8 shape, end to end): the door's
/// relay socket ← a vanilla RFC-header'd datagram → decoded at the
/// door's codec boundary → `{addr, data}` in-band → the producer's
/// stage → the real UDP echo target → back in-band → RFC-header'd →
/// the client's source address. The empty-datagram invariant rides
/// the same boundary (F-2: zero-length payload round-trips, never
/// EOF), and the in-band codec boundary is pinned (the door's relay
/// carries only RFC-header'd datagrams; the in-band wire carries only
/// `{addr, data}` frames: the round-trip through the real echo
/// target proves the two vocabularies never mixed).
#[tokio::test]
async fn udp_door_vanilla_client_round_trip_and_empty_datagram() {
    init_trace();
    let echo_addr = udp_echo_server().await;
    let topo = harness::wire_consumer_side_with(
        echo_dial(),
        Some(alksocks::udp_egress()),
        None,
        Some(harness::consumer_identity()),
    )
    .await;

    let relay = UdpSocket::bind("127.0.0.1:0").await.expect("bind relay");
    let relay_addr = relay.local_addr().expect("relay addr");
    let client = UdpSocket::bind("127.0.0.1:0").await.expect("client socket");
    let door = UdpFrontDoor::builder()
        .bind(relay_addr)
        .build()
        .expect("the door builds");
    let _serve = tokio::spawn({
        let client = Arc::clone(&topo.consumer);
        async move { door.serve_on(relay, client).await.expect("the door serves") }
    });

    // Pre: a normal datagram round-trips.
    client
        .send_to(&rfc_datagram(echo_addr, b"pre"), relay_addr)
        .await
        .expect("vanilla sendto the relay");
    let mut rx = vec![0u8; 65535];
    let (n, from) = timeout(BOUND, client.recv_from(&mut rx))
        .await
        .expect("reply within bound")
        .expect("recv");
    assert_eq!(from, relay_addr, "the reply comes from the relay socket");
    let (addr, payload) = parse_rfc_datagram(&rx[..n]);
    assert_eq!(addr, echo_addr, "the echo reports the target");
    assert_eq!(payload, b"pre");

    // The empty datagram (F-2): a zero-length payload round-trips,
    // never EOF: the association stays healthy after it.
    client
        .send_to(&rfc_datagram(echo_addr, b""), relay_addr)
        .await
        .expect("send the empty datagram");
    let (n, _) = timeout(BOUND, client.recv_from(&mut rx))
        .await
        .expect("empty-datagram reply within bound")
        .expect("recv");
    let (addr, payload) = parse_rfc_datagram(&rx[..n]);
    assert_eq!(addr, echo_addr);
    assert!(
        payload.is_empty(),
        "the empty datagram round-trips (never EOF)"
    );

    // Post: the association survived; another datagram round-trips.
    client
        .send_to(&rfc_datagram(echo_addr, b"post"), relay_addr)
        .await
        .expect("send post");
    let (n, _) = timeout(BOUND, client.recv_from(&mut rx))
        .await
        .expect("post reply within bound")
        .expect("recv");
    let (_, payload) = parse_rfc_datagram(&rx[..n]);
    assert_eq!(payload, b"post");

    // The in-band vocabulary pin: the association's adopted channel
    // exists (the door opened it in-band via AssociateSession) and the
    // producer's stage saw the real echo target (the datagrams were
    // carried as `{addr, data}` frames (the echo server, which speaks
    // only raw kernel datagrams, answered), never RFC-header'd bytes.
    assert!(
        topo.consumer
            .manager()
            .channel_ids()
            .iter()
            .any(|&id| id != 0),
        "the door's association rode an in-band channel"
    );
}

/// The reader-state pin through the door's bridge (review 003 R-1,
/// the POC #2 finding carried to the doors): two replies arriving
/// back-to-back (the doubled echo: both frames ride the producer's
/// frame writer; the door's channel read may batch them into one
/// chunk) both reach the vanilla client. A bridge that re-created
/// its decoder per read lost the second datagram.
#[tokio::test]
async fn batched_in_band_replies_both_reach_the_vanilla_client() {
    let echo_addr = udp_echo_server().await;
    let topo = harness::wire_consumer_side_with(
        echo_dial(),
        Some(harness::doubled_echo_egress_factory()),
        None,
        Some(harness::consumer_identity()),
    )
    .await;

    let relay = UdpSocket::bind("127.0.0.1:0").await.expect("bind relay");
    let relay_addr = relay.local_addr().expect("relay addr");
    let client = UdpSocket::bind("127.0.0.1:0").await.expect("client socket");
    let door = UdpFrontDoor::builder()
        .bind(relay_addr)
        .build()
        .expect("the door builds");
    let _serve = tokio::spawn({
        let client = Arc::clone(&topo.consumer);
        async move { door.serve_on(relay, client).await.expect("the door serves") }
    });

    // One datagram in → two echo datagrams out (both from the relay
    // socket, both carrying the target as their source address; the
    // bridge's pending queue drained in order, none lost).
    client
        .send_to(&rfc_datagram(echo_addr, b"once"), relay_addr)
        .await
        .expect("send");
    let mut rx = vec![0u8; 65535];
    for _ in 0..2 {
        let (n, from) = timeout(BOUND, client.recv_from(&mut rx))
            .await
            .expect("reply within bound")
            .expect("recv");
        assert_eq!(from, relay_addr, "the reply comes from the relay socket");
        let (addr, payload) = parse_rfc_datagram(&rx[..n]);
        assert_eq!(addr, echo_addr);
        assert_eq!(payload, b"once");
    }

    // No third reply: exactly the two echoes arrived (the burst was
    // queued, not duplicated).
    let mut probe = vec![0u8; 65535];
    assert!(
        timeout(Duration::from_millis(300), client.recv_from(&mut probe))
            .await
            .is_err(),
        "exactly two replies arrived"
    );
}

/// FRAG ≠ 0 is refused at the door (the v1 posture): the fragment is
/// dropped (nothing is egressed), and the association stays healthy
/// (a following normal datagram round-trips).
#[tokio::test]
async fn frag_nonzero_refused_and_association_stays_healthy() {
    let echo_addr = udp_echo_server().await;
    let topo = harness::wire_consumer_side_with(
        echo_dial(),
        Some(alksocks::udp_egress()),
        None,
        Some(harness::consumer_identity()),
    )
    .await;

    let relay = UdpSocket::bind("127.0.0.1:0").await.expect("bind relay");
    let relay_addr = relay.local_addr().expect("relay addr");
    let client = UdpSocket::bind("127.0.0.1:0").await.expect("client socket");
    let door = UdpFrontDoor::builder()
        .bind(relay_addr)
        .build()
        .expect("the door builds");
    let _serve = tokio::spawn({
        let client = Arc::clone(&topo.consumer);
        async move { door.serve_on(relay, client).await.expect("the door serves") }
    });

    // A FRAG=1 datagram: dropped at the door's codec boundary (no
    // echo comes back, because the datagram never went in-band).
    let SocketAddr::V4(v4) = echo_addr else {
        panic!("v4 target")
    };
    let mut frag_dg = vec![0u8, 0, 1, 0x01];
    frag_dg.extend_from_slice(&v4.ip().octets());
    frag_dg.extend_from_slice(&v4.port().to_be_bytes());
    frag_dg.extend_from_slice(b"fragmented");
    client
        .send_to(&frag_dg, relay_addr)
        .await
        .expect("send frag");

    // A normal datagram right after: the FRAG drop left the door
    // serving: the association is healthy.
    client
        .send_to(&rfc_datagram(echo_addr, b"healthy"), relay_addr)
        .await
        .expect("send after frag");
    let mut rx = vec![0u8; 65535];
    let (n, _) = timeout(BOUND, client.recv_from(&mut rx))
        .await
        .expect("reply within bound")
        .expect("recv");
    let (_, payload) = parse_rfc_datagram(&rx[..n]);
    assert_eq!(
        payload, b"healthy",
        "the association survived the FRAG drop"
    );

    // And the fragment's payload never echoed: only one reply arrived
    // (the recv above consumed it; the next recv times out).
    let mut probe = vec![0u8; 65535];
    assert!(
        timeout(Duration::from_millis(300), client.recv_from(&mut probe))
            .await
            .is_err(),
        "the FRAG ≠ 0 datagram was dropped (never egressed)"
    );
}

/// The client idle bound (ADR-014's door-side defense, constructor
/// config): a client that sends no datagrams within the window gets
/// its association closed door-side: the door's task ends, the
/// association's channel is reaped (association lifetime = channel
/// lifetime, ADR-003), and a fresh association succeeds after the
/// teardown (the connection is healthy).
#[tokio::test]
async fn idle_bound_closes_an_idle_association_and_a_fresh_one_succeeds() {
    let echo_addr = udp_echo_server().await;
    let topo = harness::wire_consumer_side_with(
        echo_dial(),
        Some(alksocks::udp_egress()),
        None,
        Some(harness::consumer_identity()),
    )
    .await;

    // First association: never sends, so the idle bound fires within
    // (window + slack) and the door's task ends.
    let relay = UdpSocket::bind("127.0.0.1:0").await.expect("bind relay");
    let door = UdpFrontDoor::builder()
        .bind(relay.local_addr().unwrap())
        .idle(Duration::from_millis(300))
        .build()
        .expect("the door builds");
    let serve = tokio::spawn({
        let client = Arc::clone(&topo.consumer);
        async move { door.serve_on(relay, client).await.expect("the door serves") }
    });

    let started = std::time::Instant::now();
    timeout(Duration::from_secs(3), serve)
        .await
        .expect("the door's task ended")
        .expect("serve returns cleanly on the idle exit");
    let elapsed = started.elapsed();
    assert!(
        elapsed >= Duration::from_millis(250),
        "the door waited the window before closing: {elapsed:?}"
    );

    // The association's channel was torn down with the door.
    tokio::time::sleep(Duration::from_millis(100)).await;
    assert!(
        topo.consumer
            .manager()
            .channel_ids()
            .iter()
            .all(|&id| id == 0),
        "the idle teardown reaped the association's channel"
    );

    // A fresh association succeeds after the teardown (the connection
    // is healthy, the POC's strongest observable).
    let relay2 = UdpSocket::bind("127.0.0.1:0").await.expect("bind relay 2");
    let relay2_addr = relay2.local_addr().unwrap();
    let door2 = UdpFrontDoor::builder()
        .bind(relay2_addr)
        .build()
        .expect("the door builds");
    let _serve2 = tokio::spawn({
        let client = Arc::clone(&topo.consumer);
        async move {
            door2
                .serve_on(relay2, client)
                .await
                .expect("the fresh door serves")
        }
    });
    let client = UdpSocket::bind("127.0.0.1:0").await.expect("client socket");
    client
        .send_to(&rfc_datagram(echo_addr, b"fresh"), relay2_addr)
        .await
        .expect("send through the fresh door");
    let mut rx = vec![0u8; 65535];
    let (n, _) = timeout(BOUND, client.recv_from(&mut rx))
        .await
        .expect("fresh reply within bound")
        .expect("recv");
    let (addr, payload) = parse_rfc_datagram(&rx[..n]);
    assert_eq!(addr, echo_addr);
    assert_eq!(payload, b"fresh");
}

/// A fresh association succeeds after teardown, driven through the
/// same relay-port shape as the happy path (the idle test's follow-up,
/// pinned on its own so the "connection healthy" observable stands
/// alone).
#[tokio::test]
async fn fresh_association_succeeds_after_teardown() {
    let echo_addr = udp_echo_server().await;
    let topo = harness::wire_consumer_side_with(
        echo_dial(),
        Some(alksocks::udp_egress()),
        None,
        Some(harness::consumer_identity()),
    )
    .await;

    // First association: one datagram, then the client drops the
    // socket (the door's association ends with the idle bound).
    let relay = UdpSocket::bind("127.0.0.1:0").await.expect("bind relay");
    let relay_addr = relay.local_addr().expect("relay addr");
    let door = UdpFrontDoor::builder()
        .bind(relay_addr)
        .idle(Duration::from_millis(200))
        .build()
        .expect("the door builds");
    let serve = tokio::spawn({
        let client = Arc::clone(&topo.consumer);
        async move { door.serve_on(relay, client).await.expect("the door serves") }
    });
    let vanilla = UdpSocket::bind("127.0.0.1:0")
        .await
        .expect("vanilla socket");
    vanilla
        .send_to(&rfc_datagram(echo_addr, b"first"), relay_addr)
        .await
        .expect("send first");
    let mut rx = vec![0u8; 65535];
    let (n, _) = timeout(BOUND, vanilla.recv_from(&mut rx))
        .await
        .expect("first reply within bound")
        .expect("recv");
    assert_eq!(parse_rfc_datagram(&rx[..n]).1, b"first");
    timeout(Duration::from_secs(3), serve)
        .await
        .expect("the first association ended (idle bound)")
        .expect("clean exit");

    // A fresh association on the same topology succeeds end to end.
    let relay2 = UdpSocket::bind("127.0.0.1:0").await.expect("bind relay 2");
    let relay2_addr = relay2.local_addr().expect("relay2 addr");
    let door2 = UdpFrontDoor::builder()
        .bind(relay2_addr)
        .build()
        .expect("the door builds");
    let serve2 = tokio::spawn({
        let client = Arc::clone(&topo.consumer);
        async move {
            door2
                .serve_on(relay2, client)
                .await
                .expect("the fresh door serves")
        }
    });
    let client = UdpSocket::bind("127.0.0.1:0").await.expect("client socket");
    client
        .send_to(&rfc_datagram(echo_addr, b"fresh"), relay2_addr)
        .await
        .expect("send through the fresh door");
    let (n, _) = timeout(BOUND, client.recv_from(&mut rx))
        .await
        .expect("fresh reply within bound")
        .expect("recv");
    let (addr, payload) = parse_rfc_datagram(&rx[..n]);
    assert_eq!(addr, echo_addr);
    assert_eq!(payload, b"fresh");
    serve2.abort();
}

/// The `serve()` wrapper (review 003 N-8: every other test drove
/// `serve_on`; the bind-then-serve wrapper round-trips too; the port is
/// configured explicitly so the vanilla client can address the
/// relay).
#[tokio::test]
async fn udp_door_serve_wrapper_round_trips() {
    init_trace();
    let echo_addr = udp_echo_server().await;
    let topo = harness::wire_consumer_side_with(
        echo_dial(),
        Some(alksocks::udp_egress()),
        None,
        Some(harness::consumer_identity()),
    )
    .await;

    // A free loopback port, probed up front (the door binds exactly
    // it, the wrapper's own bind decision).
    let probe = UdpSocket::bind("127.0.0.1:0").await.expect("probe bind");
    let relay_addr = probe.local_addr().expect("probe addr");
    drop(probe);

    let client = UdpSocket::bind("127.0.0.1:0").await.expect("client socket");
    let door = UdpFrontDoor::builder()
        .bind(relay_addr)
        .build()
        .expect("the door builds");
    let _serve = tokio::spawn({
        let client = Arc::clone(&topo.consumer);
        async move { door.serve(client).await.expect("the door serves") }
    });

    client
        .send_to(&rfc_datagram(echo_addr, b"wrapped"), relay_addr)
        .await
        .expect("vanilla sendto the relay");
    let mut rx = vec![0u8; 65535];
    let (n, from) = timeout(BOUND, client.recv_from(&mut rx))
        .await
        .expect("reply within bound")
        .expect("recv");
    assert_eq!(from, relay_addr, "the reply comes from the wrapper's bind");
    let (_, payload) = parse_rfc_datagram(&rx[..n]);
    assert_eq!(payload, b"wrapped");
}

/// The NAT port-rebind shape (OQ-SK2-09's option-2 fix): the client's
/// source IP pins on the first datagram, and a later datagram from the
/// same IP on a *new source port* moves the reply path (the rebind
/// keeps receiving replies. The RFC drop rule's complement: replies go
/// to the latest-seen port from the pinned IP.
#[tokio::test]
async fn nat_port_rebind_keeps_receiving_replies() {
    let echo_addr = udp_echo_server().await;
    let topo = harness::wire_consumer_side_with(
        echo_dial(),
        Some(alksocks::udp_egress()),
        None,
        Some(harness::consumer_identity()),
    )
    .await;

    let relay = UdpSocket::bind("127.0.0.1:0").await.expect("bind relay");
    let relay_addr = relay.local_addr().expect("relay addr");
    let door = UdpFrontDoor::builder()
        .bind(relay_addr)
        .build()
        .expect("the door builds");
    let _serve = tokio::spawn({
        let client = Arc::clone(&topo.consumer);
        async move { door.serve_on(relay, client).await.expect("the door serves") }
    });

    // Association born on socket A (its source port pins the reply
    // path's IP; the port rides the latest-seen shape).
    let client_a = UdpSocket::bind("127.0.0.1:0").await.expect("client A");
    client_a
        .send_to(&rfc_datagram(echo_addr, b"before"), relay_addr)
        .await
        .expect("send from A");
    let mut rx = vec![0u8; 65535];
    let (n, _) = timeout(BOUND, client_a.recv_from(&mut rx))
        .await
        .expect("reply within bound")
        .expect("recv");
    assert_eq!(parse_rfc_datagram(&rx[..n]).1, b"before");

    // The "NAT rebind": a fresh socket, same IP, new source port. The
    // datagram is accepted (same IP) and the reply goes to the *new*
    // port; the rebind keeps receiving.
    let client_b = UdpSocket::bind("127.0.0.1:0").await.expect("client B");
    client_b
        .send_to(&rfc_datagram(echo_addr, b"after"), relay_addr)
        .await
        .expect("send from B");
    let (n, _) = timeout(BOUND, client_b.recv_from(&mut rx))
        .await
        .expect("rebound reply within bound")
        .expect("recv");
    assert_eq!(parse_rfc_datagram(&rx[..n]).1, b"after");

    // A (the old port) no longer receives: the reply path followed
    // the rebind.
    let mut probe = vec![0u8; 65535];
    assert!(
        timeout(Duration::from_millis(300), client_a.recv_from(&mut probe))
            .await
            .is_err(),
        "the reply followed the rebind (the old port stopped receiving)"
    );
}

/// The RFC 1928 peer posture's drop rule (OQ-SK2-09's option-2 fix):
/// a datagram from a *different source IP* is dropped (never
/// egressed), and the association stays healthy for the pinned peer.
/// (Loopback-only IPs are used: the pinned peer keeps receiving, the
/// foreign-IP sender's datagram disappears.)
#[tokio::test]
async fn foreign_source_ip_datagrams_dropped_and_association_stays_healthy() {
    init_trace();
    let echo_addr = udp_echo_server().await;
    let topo = harness::wire_consumer_side_with(
        echo_dial(),
        Some(alksocks::udp_egress()),
        None,
        Some(harness::consumer_identity()),
    )
    .await;

    let relay = UdpSocket::bind("127.0.0.1:0").await.expect("bind relay");
    let relay_addr = relay.local_addr().expect("relay addr");
    let door = UdpFrontDoor::builder()
        .bind(relay_addr)
        .build()
        .expect("the door builds");
    let _serve = tokio::spawn({
        let client = Arc::clone(&topo.consumer);
        async move { door.serve_on(relay, client).await.expect("the door serves") }
    });

    // The real client pins the peer on loopback (127.0.0.1).
    let client = UdpSocket::bind("127.0.0.1:0").await.expect("client");
    client
        .send_to(&rfc_datagram(echo_addr, b"pinned"), relay_addr)
        .await
        .expect("send pinned");
    let mut rx = vec![0u8; 65535];
    let (n, _) = timeout(BOUND, client.recv_from(&mut rx))
        .await
        .expect("reply within bound")
        .expect("recv");
    assert_eq!(parse_rfc_datagram(&rx[..n]).1, b"pinned");

    // A "foreign-IP" sender: 127.0.0.2 (loopback, a different IP,
    // bindable without privileges). Its datagram is dropped at the
    // door's peer gate: no echo comes back to it, and nothing is
    // egressed to the echo target on its behalf.
    let foreign = UdpSocket::bind("127.0.0.2:0").await.expect("foreign bind");
    foreign
        .send_to(&rfc_datagram(echo_addr, b"hijack"), relay_addr)
        .await
        .expect("send foreign");
    let mut probe = vec![0u8; 65535];
    assert!(
        timeout(Duration::from_millis(400), foreign.recv_from(&mut probe))
            .await
            .is_err(),
        "the foreign-IP datagram was dropped (no reply to the spoofer)"
    );

    // The pinned peer is unaffected: its next datagram round-trips
    // (the drop left the association healthy; the gate never evicts).
    client
        .send_to(&rfc_datagram(echo_addr, b"still"), relay_addr)
        .await
        .expect("send still");
    let (n, _) = timeout(BOUND, client.recv_from(&mut rx))
        .await
        .expect("still-reply within bound")
        .expect("recv");
    assert_eq!(
        parse_rfc_datagram(&rx[..n]).1,
        b"still",
        "the association survived the foreign-IP drop"
    );
}