//! The composed accept integration tests (ADR-007 variant 2 — the
//! BIND accept whose listener lives on an alktunnels 0.2.0 bind-first
//! listen resource, end to end over channels): the graduation's
//! vanilla fidelity (reply#1 = the real `bound`; reply#2 = the
//! forwarded op's real `peer` — no stand-ins), the raw pass-through
//! data plane, the client-gone teardown (drop-witness), two concurrent
//! BIND sessions' distinct tokens, and the per-accept isolation.
//!
//! The topology: the opener (the SOCKS5 consumer side) dials the
//! transport; the tunnel producer serves BOTH the alktunnels bind-first
//! open op and the SOCKS5 open op on one registry. The forwarded op
//! serves on the opener side (post-hoc `register_forwarded_serving` —
//! the correlation registry the composed accept factories ride).

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

mod harness;

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

use alksocks::composed::{bindfirst_accept, register_forwarded_serving};
use alksocks::consumer::BindSession;
use alksocks::rfc1928::util::target_addr::{TargetAddr, ToTargetAddr};
use alktunnels::params::Substrate;
use alktunnels::producer::ResourceRegistry;
use tokio::io::{AsyncReadExt, AsyncWriteExt};

use harness::{
    assert_composed_channels, composed_fresh_queue, composed_opener_identity, composed_push_accept,
    composed_queue_loop, composed_scopeless_opener_identity, composed_test_bind,
    wire_bindfirst_composed, COMPOSED_BOUND_PORT, COMPOSED_PEER_PORT, COMPOSED_RESOURCE,
};

/// The full wiring: the bind-first topology with the composed accept
/// factory + the forwarded serving registered post-hoc on the opener's
/// registry. Returns everything the tests drive.
struct Suite {
    topo: harness::BindFirstTopology,
    queue: harness::ComposedAcceptQueue,
}

async fn wire_suite() -> Suite {
    wire_suite_with_identity(composed_opener_identity()).await
}

async fn wire_suite_with_identity(identity: alkcall::core::auth::Identity) -> Suite {
    let resource_registry = ResourceRegistry::new();
    resource_registry
        .register(COMPOSED_RESOURCE, Substrate::Tcp, "assembly-owned")
        .await;
    let queue = composed_fresh_queue();
    let topo = wire_bindfirst_composed(
        resource_registry,
        composed_test_bind(),
        composed_queue_loop(queue.clone()),
        None, // the SOCKS5 openable registers post-hoc, below — the composed factory needs the opener's client
        identity,
    )
    .await;
    // The correlation registry rides the opener's serving registry
    // (the forwarded op's serving half — the provider's opener half).
    let registry = register_forwarded_serving(
        &alkcall::channels::operations::ChannelCore::new(
            topo.opener.manager().clone(),
            alkcall::channels::policy::default_policy(),
        ),
        &topo.opener_registry,
        alkcall::core::auth::AuthContext::anonymous(b"alk/tunnel"),
    );
    // The SOCKS5 openable registers post-hoc (W2's shape) on the
    // producer's registry — with the REAL composed factory, which
    // closes over the opener's client + the correlation registry (both
    // only exist now).
    let factory = bindfirst_accept(topo.opener.clone(), registry, COMPOSED_RESOURCE);
    let core = alkcall::channels::operations::ChannelCore::new(
        topo.producer_manager.clone(),
        alkcall::channels::policy::default_policy(),
    );
    alksocks::producer::Socks5Producer::builder()
        .accept(factory)
        .build()
        .register(
            &core,
            &topo.producer_registry,
            alkcall::core::auth::AuthContext::anonymous(b"alk/socks5"),
            None,
        )
        .expect("register socks5 openable post-hoc (the composed accept factory)");
    Suite { topo, queue }
}

/// Open a BIND session through the composed accept (the SOCKS5 BIND
/// client over channels — the consumer session type). The session
/// drives the whole pipeline: open → negotiate → BIND → reply#1 (the
/// factory's real `bound`) → quiet wait (the factory's accept future)
/// → reply#2 (the real peer).
async fn bind_session(suite: &Suite) -> BindSession {
    BindSession::bind(
        suite.topo.opener.as_ref(),
        ("bind.test", 2121).to_target_addr().unwrap(),
    )
    .await
    .expect("the session binds through the composed accept")
}

/// THE gate: the full composed BIND story — reply#1 carries the real
/// `bound` (the graduation's vanilla fidelity), an app server "dials"
/// (the queue accept), reply#2 carries the real peer (the forwarded
/// op's `peer` param), and the pass-through data plane round-trips
/// through the forwarded channel to the echo task and back.
#[tokio::test]
async fn composed_bind_full_story_bound_peer_and_round_trip() {
    let _ = tracing_subscriber::fmt()
        .with_max_level(tracing::level_filters::LevelFilter::TRACE)
        .with_test_writer()
        .try_init();
    let suite = wire_suite().await;

    // The BIND session rides the real composed factory (the SOCKS5
    // registration's accept seam — replaced post-hoc per-suite; the
    // registration reads through the shared registry per dispatch, so
    // the session's open dispatches with the factory in place).
    let mut session = bind_session(&suite).await;
    let channel_id = session.channel_id();
    assert!(
        suite.topo.producer_manager.has_channel(channel_id),
        "the SOCKS5 channel is live on the producer side"
    );

    // Reply#1: the REAL bound (the open-op reply's `bound` field — no
    // expected-addr stand-in; the test bind's reported address).
    let bind_addr = session.reply1().expect("reply#1").bind_addr;
    assert_eq!(
        bind_addr,
        TargetAddr::Ip(SocketAddr::from(([127, 0, 0, 1], COMPOSED_BOUND_PORT))),
        "reply#1 advertises the real OS-chosen bound"
    );

    // The app server dials the far listener (the queue accept) — the
    // producer's accept loop opens the forwarded channel toward the
    // opener; the correlation registry delivers it into the accept
    // future; reply#2 carries the real peer.
    composed_push_accept(&suite.queue, ("10.0.0.1", COMPOSED_PEER_PORT)).await;
    let mut plane = tokio::time::timeout(Duration::from_secs(10), session.await_inbound())
        .await
        .expect("reply#2 within timeout")
        .expect("await_inbound ok")
        .expect("the plane exists");
    assert_eq!(
        plane.peer,
        TargetAddr::Ip(SocketAddr::from(([10, 0, 0, 1], COMPOSED_PEER_PORT))),
        "reply#2 carries the forwarded op's real peer"
    );

    // The pass-through data plane: client bytes cross the SOCKS5
    // channel → the producer's pump → the forwarded channel → the
    // accepted halves' echo → and back.
    plane.write_all(b"bind-ping").await.expect("write");
    plane.flush().await.expect("flush");
    let mut buf = [0u8; 9];
    tokio::time::timeout(Duration::from_secs(10), plane.read_exact(&mut buf))
        .await
        .expect("round trip")
        .expect("read");
    assert_eq!(&buf, b"bind-ping");

    // Teardown: the session reaps its SOCKS5 channel; the data plane's
    // halves drop (the owner wrapper reaps the forwarded channel).
    drop(plane);
    assert!(session.close().await, "the SOCKS5 channel reaped");
    // The teardown cascade: the data plane's halves drop → the
    // forwarded handler's pump EOFs → the wrapper reaps the forwarded
    // channel → the listener dropped with the handle → the listen
    // channel EOFs → the far accept loop exits. Settle, then assert.
    for _ in 0..50 {
        tokio::time::sleep(Duration::from_millis(100)).await;
        let opener_ids = suite.topo.opener.manager().channel_ids();
        if opener_ids.iter().all(|&id| id == 0) {
            break;
        }
    }
    // The opener side holds only channel 0 (the producer-allocated
    // forwarded channel's entry rides the opener's manager per its own
    // adopter contract — alktunnels ADR-005). The producer side may
    // retain the adopted forwarded channel (see `assert_composed_channels`
    // — its adopter's session dropped with the plane; the entry reaps
    // at transport EOF, not at the listener's).
    assert_composed_channels(&suite.topo, &[0], &[channel_id]).await;
}

/// The client-gone teardown (the drop-witness): dropping the BIND
/// session mid-quiet-wait releases the far listener — the accept
/// future owns it end-to-end (the quiet wait's `select!` drop IS the
/// teardown).
#[tokio::test]
async fn client_gone_teardown_releases_the_listener() {
    let suite = wire_suite().await;
    let mut session = bind_session(&suite).await;
    let socks5_channel = session.channel_id();
    let _ = session.reply1().expect("reply#1");

    // The client vanishes (drop = the quiet wait's select! observes
    // the channel EOF → the accept future drops → the TunnelListener
    // inside it drops → the listen channel EOFs → the far accept
    // dispatcher tears down).
    drop(session);

    // The SOCKS5 channel reaps; the listen channel (the far listener's
    // carrier) reaps too — no leaked listener, no leaked channel.
    let mut socks5_gone = false;
    let mut listen_gone = false;
    for _ in 0..50 {
        tokio::time::sleep(Duration::from_millis(200)).await;
        socks5_gone = !suite.topo.producer_manager.has_channel(socks5_channel);
        // The listen channel's id is unknown here (the accept future
        // held it); the aggregate leak probe is the witness: after the
        // teardown cascade every channel is gone on both sides.
        listen_gone = suite
            .topo
            .producer_manager
            .channel_ids()
            .iter()
            .all(|&id| id == 0);
        if socks5_gone && listen_gone {
            break;
        }
    }
    assert!(
        socks5_gone,
        "the SOCKS5 channel reaped after the client dropped"
    );
    assert!(
        listen_gone,
        "the listen channel reaped (no leaked listener): producer={:?}",
        suite.topo.producer_manager.channel_ids()
    );
    // The accept queue was never consumed (the listener died before
    // any app server arrived).
    assert!(
        suite.queue.0.lock().await.is_empty(),
        "the dropped listener consumed nothing"
    );
    assert_composed_channels(&suite.topo, &[0], &[0]).await;
}

/// Two concurrent BIND sessions get distinct tokens (the mint is
/// per-open); each accept future correlates only ITS forwarded
/// session.
#[tokio::test]
async fn concurrent_bind_sessions_get_distinct_tokens() {
    let suite = wire_suite().await;
    let mut s1 = bind_session(&suite).await;
    let mut s2 = bind_session(&suite).await;
    assert_ne!(s1.channel_id(), s2.channel_id(), "distinct SOCKS5 channels");
    let _ = s1.reply1().expect("reply#1 s1");
    let _ = s2.reply1().expect("reply#2 s2");

    // Two accepts: the tokens are distinct (two listen opens minted
    // distinct tokens — observably: both sessions resolve their own
    // forwarded sessions, the correlation delivered each to the right
    // waiter).
    composed_push_accept(&suite.queue, ("10.0.0.1", COMPOSED_PEER_PORT)).await;
    composed_push_accept(&suite.queue, ("10.0.0.2", COMPOSED_PEER_PORT + 1)).await;

    let mut p1 = tokio::time::timeout(Duration::from_secs(10), s1.await_inbound())
        .await
        .expect("s1 reply#2")
        .expect("s1 plane ok")
        .expect("s1 plane");
    let mut p2 = tokio::time::timeout(Duration::from_secs(10), s2.await_inbound())
        .await
        .expect("s2 reply#2")
        .expect("s2 plane ok")
        .expect("s2 plane");

    // Each plane pumps only its own accepted connection (the
    // correlation routed by token): distinct peer ports arrive, each
    // on the session whose listener accepted it.
    let peers = [p1.peer.clone(), p2.peer.clone()];
    assert_ne!(peers[0], peers[1], "the two accepts routed distinctly");

    // Both data planes round-trip independently.
    p1.write_all(b"one").await.expect("p1 write");
    p2.write_all(b"two").await.expect("p2 write");
    let mut b1 = [0u8; 3];
    let mut b2 = [0u8; 3];
    tokio::time::timeout(Duration::from_secs(10), p1.read_exact(&mut b1))
        .await
        .expect("p1 echo")
        .expect("p1 read");
    tokio::time::timeout(Duration::from_secs(10), p2.read_exact(&mut b2))
        .await
        .expect("p2 echo")
        .expect("p2 read");
    assert_eq!(&b1, b"one");
    assert_eq!(&b2, b"two");

    drop(p1);
    drop(p2);
    assert!(s1.close().await);
    assert!(s2.close().await);
    tokio::time::sleep(Duration::from_millis(300)).await;
    assert_composed_channels(&suite.topo, &[0], &[0]).await;
}

/// A forwarded-open failure is per-accept isolated: an opener identity
/// without `tunnel:forwarded` makes the forwarded open fail FORBIDDEN
/// — the BIND session's accept future never resolves a forwarded
/// session (reply#2's mapped refusal), the listen channel stays
/// usable, and the failure does not tear the listener down.
#[tokio::test]
async fn forwarded_open_failure_is_per_accept_isolated() {
    let suite = wire_suite_with_identity(composed_scopeless_opener_identity()).await;
    let mut session = bind_session(&suite).await;
    let _ = session.reply1().expect("reply#1");

    // The accept fires (the queue accept consumed — the producer's
    // accept loop is alive), but the forwarded open fails FORBIDDEN.
    composed_push_accept(&suite.queue, ("10.0.0.1", COMPOSED_PEER_PORT)).await;

    // The forwarded-open failure is per-accept isolated AT THE TUNNEL
    // LAYER: the producer's accept loop drops that accepted connection
    // and keeps serving. The accept future stays parked (the SOCKS5
    // producer never learns of the tunnel-layer failure — the
    // quiet-wait shape is unaffected), so the queue drains both
    // accepts and the listener survives (the alktunnels producer
    // suite's `forwarded_open_failure_isolates_the_accept` pin, end to
    // end).
    composed_push_accept(&suite.queue, ("10.0.0.1", COMPOSED_PEER_PORT)).await;
    tokio::time::sleep(Duration::from_millis(300)).await;
    assert!(
        suite.queue.0.lock().await.is_empty(),
        "the accept loop consumed both accepts (alive throughout)"
    );
    drop(session);
    tokio::time::sleep(Duration::from_millis(300)).await;
}

/// The dropped listener (client gone mid-quiet-wait) leaves a
/// forwarded open nowhere to go: the correlation miss drops the
/// forwarded channel (teardown-at-birth — pinned by the dead-oneshot
/// probe's aggregate-leak witness).
#[tokio::test]
async fn late_forwarded_open_after_client_gone_drops_the_channel() {
    let suite = wire_suite().await;
    let mut session = bind_session(&suite).await;
    let _ = session.reply1().expect("reply#1");
    drop(session);

    // The client is gone; the listener is being torn down. An accept
    // arriving in the race window may reach the dispatcher before the
    // EOF sentinel does; the forwarded open then finds a dead oneshot
    // (the correlation miss) and drops the channel.
    composed_push_accept(&suite.queue, ("10.0.0.1", COMPOSED_PEER_PORT)).await;
    tokio::time::sleep(Duration::from_millis(500)).await;
    // Either the listener was already gone (the accept queued
    // unconsumed) or the forwarded channel dropped at birth — either
    // way: no leaked channels on the opener side.
    assert_composed_channels(&suite.topo, &[0], &[0]).await;
}