//! Tests for the `local` feature's BIND-FIRST half (shape 3, ADR-008):
//! the real-socket `BindFn` sources (`bind_tcp_first` /
//! `bind_unix_first` — the `bound` reply field's data, port 0
//! resolved), the accept-pair loops (peer + halves per accept), and
//! the assembly convenience closures (`local_bind_first` +
//! `local_bindfirst_accept_loop`), through the real bind-first
//! establisher path over REAL sockets.
//!
//! Covered (the task's test plan): the bound address is the REAL
//! OS-chosen one (a fake `bound` would strand the test's dials), two
//! clients dial → two accepts → two forwarded channels pumping
//! independently (peer fidelity: each forwarded open carries the REAL
//! accepted peer), unix bind-first parity (OQ-TN-14 — the bind path is
//! the same shape as the dial path), and the udp bind-first refusal
//! (a UDP socket has no accept).

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

mod harness;

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

use alkcall::channels::operations::{ChannelCore, OpenHandler};
use alkcall::core::auth::AuthContext;
use alkcall::protocol::wire::CallError;
use alktunnels::params::{parse_listen_reply, Substrate, SubstrateAddr, TunnelForwardedParams};
use alktunnels::producer::ResourceRegistry;
use serde_json::Value;
use tracing_subscriber::EnvFilter;

fn init_tracing() {
    let _ = tracing_subscriber::fmt()
        .with_env_filter(EnvFilter::from_default_env())
        .try_init();
}

/// The forwarded-op recording stub's peer log: the `peer` param of
/// every forwarded open the producer's dispatcher made (the
/// reply#2-fidelity probe surface — ADR-008 §3).
type PeerLog = Arc<tokio::sync::Mutex<Vec<SubstrateAddr>>>;

/// The forwarded-op opener-side serving stub (the queue-fed shape):
/// each forwarded open pops the NEXT app-side pair (the "local BIND
/// client" halves the test registered) and pumps the forwarded
/// channel against it, recording the open's `peer` param (the
/// reply#2-fidelity probe surface — ADR-008 §3). One listener session
/// = one token = MANY forwarded opens (per accept), so the
/// correlation is a queue, not a one-shot map.
fn recording_forwarded_handler(app_sides: AppSideQueue, peers: PeerLog) -> OpenHandler {
    Arc::new(
        move |input: Value, _plan, conn: alkcall::core::types::Connection, _auth| {
            let app_sides = Arc::clone(&app_sides);
            let peers = Arc::clone(&peers);
            tokio::spawn(async move {
                let forwarded: TunnelForwardedParams = match serde_json::from_value(input) {
                    Ok(p) => p,
                    Err(e) => {
                        tracing::error!(error = %e, "recording stub: params parse failed");
                        return;
                    }
                };
                peers.lock().await.push(forwarded.peer.clone());
                let target = app_sides.lock().await.pop_front();
                let Some(target) = target else {
                    tracing::error!(
                        listen = forwarded.listen,
                        "recording stub: no app-side pair registered"
                    );
                    return;
                };
                let Ok(bidi) = conn.accept_bi().await else {
                    tracing::error!("recording stub: accept_bi failed");
                    return;
                };
                alkcall::channels::pump::pump_bidi(bidi, target.read, target.write).await;
            })
        },
    )
}

/// The app-side pairs the stub pops per forwarded open (the "local
/// BIND client" halves — the stub's pump target; the far end is the
/// test's echo task).
type AppSideQueue =
    Arc<tokio::sync::Mutex<std::collections::VecDeque<alktunnels::producer::TargetHandle>>>;

/// Install the recording stub on the opener (consumer) side (post-hoc
/// on the captured serving registry — W2).
async fn install_recording_stub(topo: &harness::Topology) -> (AppSideQueue, PeerLog) {
    let app_sides: AppSideQueue = Arc::new(tokio::sync::Mutex::new(Default::default()));
    let peers: PeerLog = Arc::new(tokio::sync::Mutex::new(Vec::new()));
    let core = ChannelCore::new(
        topo.consumer_manager.clone(),
        alkcall::channels::policy::default_policy(),
    );
    core.register_openable(
        alktunnels::params::tunnel_forwarded_spec(),
        recording_forwarded_handler(Arc::clone(&app_sides), Arc::clone(&peers)),
        &topo.consumer_registry,
        AuthContext::anonymous(b"alk/tunnel"),
    )
    .expect("register recording forwarded stub on the opener");
    (app_sides, peers)
}

/// Call the producer's open op on channel 0 with the FULL reply (the
/// bind-first reply projection probe — `bound` + `listen`).
async fn open_listen_with_reply(
    topo: &harness::Topology,
    resource: &str,
    substrate: Substrate,
) -> Result<(u32, Value), CallError> {
    let substrate_str = match substrate {
        Substrate::Tcp => "tcp",
        Substrate::Udp => "udp",
        Substrate::Unix => "unix",
    };
    let payload = serde_json::json!({
        "operationId": "channels/tunnel/sub",
        "input": {
            "resource": resource,
            "substrate": substrate_str,
        },
    });
    let response = topo.consumer_call.call_with_payload(payload).await;
    let out = response.result?;
    let channel_id = out["channel_id"].as_u64().expect("channel_id") as u32;
    Ok((channel_id, out))
}

/// One app-side session: a raw duplex pair whose FAR end echoes (the
/// app-side session shape) — the stub pumps the forwarded channel
/// against the near end (pushed into the stub's queue); the test
/// drives the REAL accepted socket at the tunnel's other side, whose
/// bytes echo through the app-side session.
fn echo_app_side() -> alktunnels::producer::TargetHandle {
    let (a, b) = tokio::io::duplex(64 * 1024);
    let (a_read, a_write) = tokio::io::split(a);
    let (b_read, b_write) = tokio::io::split(b);
    tokio::spawn(async move {
        use tokio::io::{AsyncReadExt, AsyncWriteExt};
        let (mut r, mut w) = (b_read, b_write);
        let mut buf = vec![0u8; 8192];
        loop {
            match r.read(&mut buf).await {
                Ok(0) | Err(_) => return,
                Ok(n) => {
                    if w.write_all(&buf[..n]).await.is_err() {
                        return;
                    }
                }
            }
        }
    });
    alktunnels::producer::TargetHandle {
        read: Box::new(a_read),
        write: Box::new(a_write),
    }
}

/// THE bind-first flow over REAL TCP sockets (the `local` shape-3
/// gate): `bind_tcp_first` binds port 0 → the reply's `bound` is the
/// REAL OS-chosen address (the test dials it — a fake bound would
/// strand the dial) → two clients dial → two accepts → two forwarded
/// channels → both pump independently, and each forwarded open
/// carries the REAL accepted peer (peer fidelity, reply#2 class).
#[tokio::test]
async fn bindfirst_tcp_real_sockets_bound_fidelity_and_two_pumps() {
    init_tracing();
    let registry = ResourceRegistry::new();
    registry
        .register("bind", Substrate::Tcp, "127.0.0.1:0")
        .await;
    let topo = harness::wire_bindfirst(
        registry,
        alktunnels::local::local_bind_first(),
        alktunnels::local::local_bindfirst_accept_loop(),
    )
    .await;

    let (channel_id, reply) = open_listen_with_reply(&topo, "bind", Substrate::Tcp)
        .await
        .expect("listen open");
    let projected = parse_listen_reply(&reply).expect("projected reply");
    let bind_addr = match &projected.bound {
        Some(SubstrateAddr::HostPort { host, port }) => {
            assert_eq!(host, "127.0.0.1", "the bound host is the bound loopback");
            assert_ne!(*port, 0, "port 0 was resolved to the OS-chosen port");
            format!("{host}:{port}")
        }
        other => panic!("bind-first open carries the real bound address: {other:?}"),
    };
    let token = projected.listen.expect("the `listen` token");
    assert_ne!(token, 0, "token 0 is never minted (the wire-illegal value)");

    let (app_sides, peers) = install_recording_stub(&topo).await;

    // The opener adopts the LISTEN channel (the idle side). The
    // halves are HELD — dropping them EOFs the idle stream and tears
    // the listener down (the ADR-008 sentinel is live even here).
    let (_idle_send, _idle_recv) = topo
        .consumer_manager
        .adopt_channel(channel_id, alktunnels::TUNNEL_ALPN, None)
        .await
        .expect("adopt listen channel");

    // Two app-side sessions (the stub pops one per forwarded open).
    app_sides.lock().await.push_back(echo_app_side());
    app_sides.lock().await.push_back(echo_app_side());

    // Two clients dial the reported bound address (the reply#1
    // fidelity pin: `bound` is the REAL address, not a placeholder).
    use tokio::io::{AsyncReadExt, AsyncWriteExt};
    let mut client1 = tokio::net::TcpStream::connect(&bind_addr)
        .await
        .expect("client 1 dials the bound address");
    let mut client2 = tokio::net::TcpStream::connect(&bind_addr)
        .await
        .expect("client 2 dials the bound address");

    // Both pump independently: interleaved writes, per-socket echoes.
    client1.write_all(b"client-one").await.expect("write 1");
    client2.write_all(b"client-two").await.expect("write 2");
    let mut buf1 = vec![0u8; 10];
    let mut buf2 = vec![0u8; 10];
    let (r1, r2) = tokio::join!(
        tokio::time::timeout(Duration::from_secs(5), client1.read_exact(&mut buf1)),
        tokio::time::timeout(Duration::from_secs(5), client2.read_exact(&mut buf2)),
    );
    r1.expect("round trip 1").expect("read 1");
    r2.expect("round trip 2").expect("read 2");
    assert_eq!(&buf1, b"client-one", "pump 1 serves ITS socket");
    assert_eq!(&buf2, b"client-two", "pump 2 serves ITS socket");

    // Peer fidelity: each forwarded open carried the REAL accepted
    // peer (loopback + a real ephemeral port; distinct per socket).
    let logged = peers.lock().await.clone();
    assert_eq!(logged.len(), 2, "one forwarded open per accept");
    let mut ports = std::collections::HashSet::new();
    for peer in &logged {
        match peer {
            SubstrateAddr::HostPort { host, port } => {
                assert_eq!(host, "127.0.0.1", "the REAL peer host");
                assert_ne!(*port, 0, "the REAL peer port");
                ports.insert(*port);
            }
            other => panic!("tcp peer is HostPort-shaped: {other:?}"),
        }
    }
    assert_eq!(ports.len(), 2, "two accepts carried distinct real peers");
}

/// Unix bind-first parity (OQ-TN-14 — the bind path ships with the
/// dial path): `bind_unix_first` binds a temp-path listener → the
/// reply's `bound` is the REAL bound socket path → a unix client
/// (bound to its own path, so the peer is representable) connects →
/// one accept → one forwarded channel carrying the REAL peer path →
/// byte round-trip.
#[tokio::test]
async fn bindfirst_unix_real_sockets_bound_and_pumped() {
    init_tracing();
    let dir = std::env::temp_dir().join(format!("alktunnels-bindfirst-{}", std::process::id()));
    let _ = std::fs::create_dir_all(&dir);
    let listen_path = dir.join("listen.sock");
    let client_path = dir.join("client.sock");
    let _ = std::fs::remove_file(&listen_path);
    let _ = std::fs::remove_file(&client_path);
    let listen_str = listen_path.to_string_lossy().into_owned();

    let registry = ResourceRegistry::new();
    registry
        .register("ubind", Substrate::Unix, &listen_str)
        .await;
    let topo = harness::wire_bindfirst(
        registry,
        alktunnels::local::local_bind_first(),
        alktunnels::local::local_bindfirst_accept_loop(),
    )
    .await;

    let (channel_id, reply) = open_listen_with_reply(&topo, "ubind", Substrate::Unix)
        .await
        .expect("listen open");
    let projected = parse_listen_reply(&reply).expect("projected reply");
    assert_eq!(
        projected.bound,
        Some(SubstrateAddr::Path {
            path: listen_str.clone()
        }),
        "the `bound` field is the REAL bound socket path"
    );
    let token = projected.listen.expect("the `listen` token");
    assert_ne!(token, 0, "token 0 is never minted (the wire-illegal value)");

    let (app_sides, peers) = install_recording_stub(&topo).await;

    // The opener adopts the LISTEN channel (the idle side) — held for
    // the test's lifetime (dropping the halves EOFs the idle stream
    // and tears the listener down, the ADR-008 sentinel).
    let (_idle_send, _idle_recv) = topo
        .consumer_manager
        .adopt_channel(channel_id, alktunnels::TUNNEL_ALPN, None)
        .await
        .expect("adopt listen channel");

    // One app-side session (the stub pops it for the forwarded open).
    app_sides.lock().await.push_back(echo_app_side());

    // The unix client binds its own socket at a named path FIRST (an
    // unnamed peer cannot be represented in the forwarded `peer`
    // param — ADR-008 §3; the named-peer shape is the parity case).
    // socket2: std has no `UnixStream::bind`; a socket2 socket binds
    // the sockaddr at the path, then connects (the peer address is
    // the client's own bound path).
    let client_sock = socket2::Socket::new(socket2::Domain::UNIX, socket2::Type::STREAM, None)
        .expect("socket2 client");
    client_sock
        .bind(&socket2::SockAddr::unix(&client_path).expect("client sockaddr"))
        .expect("bind client path");
    client_sock
        .connect(&socket2::SockAddr::unix(&listen_path).expect("listen sockaddr"))
        .expect("client dials");
    let client_std: std::os::unix::net::UnixStream = client_sock.into();
    client_std.set_nonblocking(true).expect("nonblocking");
    let mut client = tokio::net::UnixStream::from_std(client_std).expect("tokio client");

    use tokio::io::{AsyncReadExt, AsyncWriteExt};
    client.write_all(b"unix-bind").await.expect("write");
    let mut buf = vec![0u8; 9];
    tokio::time::timeout(Duration::from_secs(5), client.read_exact(&mut buf))
        .await
        .expect("round trip")
        .expect("read");
    assert_eq!(&buf, b"unix-bind");

    // Peer fidelity: the forwarded open carried the client's REAL
    // bound path.
    let logged = peers.lock().await.clone();
    assert_eq!(
        logged,
        vec![SubstrateAddr::Path {
            path: client_path.to_string_lossy().into_owned()
        }],
        "the forwarded open carried the REAL accepted peer path"
    );

    let _ = std::fs::remove_file(&listen_path);
    let _ = std::fs::remove_file(&client_path);
}

/// The udp bind-first refusal: a UDP socket has no accept, so the
/// bind closure fails loud (`handler_error`-class) — the shape is not
/// defined there, never a silent fake listener.
#[tokio::test]
async fn udp_bind_first_fails_loud() {
    let registry = ResourceRegistry::new();
    registry
        .register("ubind", Substrate::Udp, "127.0.0.1:0")
        .await;
    let topo = harness::wire_bindfirst(
        registry,
        alktunnels::local::local_bind_first(),
        alktunnels::local::local_bindfirst_accept_loop(),
    )
    .await;

    let payload = serde_json::json!({
        "operationId": "channels/tunnel/sub",
        "input": {"resource": "ubind", "substrate": "udp"},
    });
    let response = topo.consumer_call.call_with_payload(payload).await;
    let err = response.result.expect_err("udp bind-first must fail");
    assert_eq!(err.code, "channel:open_failed");
    let reason = err
        .details
        .as_ref()
        .and_then(|d| d.get("reason"))
        .and_then(|r| r.as_str())
        .expect("reason in details");
    assert_eq!(reason, "handler_error");
}

/// The accept-pair loop's mismatch probe: a listener NOT produced by
/// this module's bind closures fails loud inside the closure (the
/// downcast is the assembly side of the type-erased injection — a
/// foreign listener type is `handler_error`, never a phantom accept).
#[tokio::test]
async fn accept_pair_loop_downcast_miss_fails_loud() {
    let loop_ = alktunnels::local::local_bindfirst_accept_loop();
    let foreign: Arc<dyn std::any::Any + Send + Sync> = Arc::new(4242u32);
    let err = loop_(foreign)
        .await
        .expect_err("a foreign listener fails loud");
    assert!(
        matches!(
            err,
            alktunnels::producer::TunnelEstablishError::HandlerError(_)
        ),
        "the downcast miss is handler_error-class: {err:?}"
    );
}

/// The unnamed-peer/unnamed-bind fail-loud arms (G-5 — ADR-008 §3's
/// two pinned sentences: a producer that cannot determine the peer
/// does not serve a bind-first listener over that substrate):
///
/// - **Unnamed bind** — an empty-path `bind_unix` is the kernel's
///   autobind (the kernel assigns an abstract name, addlen ==
///   offsetof(sun_path)); the bound addr has no filesystem path, so
///   `local_addr` (and thus `bind_unix_first`) fails loud with the
///   "no filesystem path" handler error instead of serving an
///   unrepresentable `bound`.
/// - **Unnamed peer** — an addressless client (socket2 autobind
///   here; a plain unbound client behaves identically through tokio
///   accept — mio maps any abstract-namespace peer to unnamed) is
///   accepted by a NAMED listener; `accept_pair` fails loud because
///   the forwarded op's `peer` param is required on the wire.
#[cfg(target_os = "linux")]
#[tokio::test]
async fn unix_unnamed_bind_and_unnamed_peer_fail_loud() {
    // Arm 1: the autobind listener — `bind_unix("")`. tokio passes the
    // empty path through to `bind(2)` with addlen == offsetof(sun_path):
    // the kernel autobinds an abstract name, no filesystem path exists.
    let halves = alktunnels::local::bind_unix("")
        .expect("autobind bind succeeds (the kernel assigns the abstract name)");
    let err = halves
        .local_addr()
        .expect_err("an autobound listener has no filesystem path");
    assert!(
        matches!(
            err,
            alktunnels::producer::TunnelEstablishError::HandlerError(ref m)
                if m.contains("no filesystem path"),
        ),
        "unnamed bind maps to the handler error: {err:?}"
    );

    // Arm 2: the unnamed peer — a socket2 autobind client connects to a
    // NAMED listener; `accept_pair` must refuse (the forwarded `peer`
    // param requires an address).
    let dir = std::env::temp_dir().join(format!("alktunnels-g5-{}", std::process::id()));
    let _ = std::fs::create_dir_all(&dir);
    let listen_path = dir.join("named.sock");
    let _ = std::fs::remove_file(&listen_path);

    let named =
        alktunnels::local::bind_unix(&listen_path.to_string_lossy()).expect("bind named listener");
    assert!(named.local_addr().is_ok(), "the named listener has a path");

    let client = socket2::Socket::new(socket2::Domain::UNIX, socket2::Type::STREAM, None)
        .expect("socket2 client");
    client
        .bind(&socket2::SockAddr::unix("").expect("empty sockaddr (the autobind shape)"))
        .expect("autobind client bind");
    client
        .connect(&socket2::SockAddr::unix(&listen_path).expect("listen sockaddr"))
        .expect("autobind client dials");

    let err = named
        .accept_pair()
        .await
        .expect_err("an unnamed peer cannot be represented in the forwarded op");
    assert!(
        matches!(
            err,
            alktunnels::producer::TunnelEstablishError::HandlerError(ref m)
                if m.contains("unnamed peer"),
        ),
        "unnamed peer maps to the handler error: {err:?}"
    );

    let _ = std::fs::remove_file(&listen_path);
}