//! 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::sync::Arc;
use std::time::Duration;

use alksocks::composed::{bindfirst_accept, register_forwarded_serving};
use alksocks::consumer::BindSession;
use alksocks::rfc1928::util::target_addr::ToTargetAddr;
use alktunnels::params::{Substrate, SubstrateAddr};
use alktunnels::producer::{
    BindFirstAcceptLoop, BindFn, BoundListener, ResourceRegistry, TargetHandle,
    TunnelEstablishError,
};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::sync::Notify;

/// The resource the bind-first producer serves (the tests' only
/// subscription — the registry namespace gate).
const RESOURCE: &str = "bind";

/// The bound address the test bind reports (the "OS-chosen" stand-in —
/// the test's queue listener owns it; fidelity is the assert's job).
const BOUND_PORT: u16 = 42113;

/// The peer address the accepted pairs report (the "app server" — the
/// forwarded op's `peer` param carries it to reply#2).
const PEER_PORT: u16 = 55231;

/// The test bind (the assembly's bind seam): reports the test's
/// bound address + an opaque listener witness.
fn test_bind() -> BindFn {
    Arc::new(|_substrate, _backing| {
        Box::pin(async {
            Ok(BoundListener {
                listener: Arc::new(()) as Arc<dyn std::any::Any + Send + Sync>,
                bound: SubstrateAddr::HostPort {
                    host: "127.0.0.1".to_string(),
                    port: BOUND_PORT,
                },
            })
        })
    })
}

/// The accept-queue pair the test accept loop pops from (the alktunnels
/// listener test's shape: a VecDeque + Notify — one accepted pair per
/// dispatcher call).
type AcceptQueue = (
    Arc<tokio::sync::Mutex<std::collections::VecDeque<(SubstrateAddr, TargetHandle)>>>,
    Arc<tokio::sync::Notify>,
);

fn fresh_queue() -> AcceptQueue {
    (
        Arc::new(tokio::sync::Mutex::new(std::collections::VecDeque::new())),
        Arc::new(Notify::new()),
    )
}

/// An accepted "connection" for the test accept loop: a duplex pair
/// with an echo task on the far end (the app server's socket stand-in
/// — the data plane's echo confirms the full round trip).
fn accepted_halves(peer_addr: (&str, u16)) -> (SubstrateAddr, TargetHandle) {
    let (consumer_side, target_side) = tokio::io::duplex(64 * 1024);
    let (mut t_read, mut t_write) = tokio::io::split(target_side);
    tokio::spawn(async move {
        let mut buf = vec![0u8; 8192];
        loop {
            match t_read.read(&mut buf).await {
                Ok(0) | Err(_) => return,
                Ok(n) => {
                    if t_write.write_all(&buf[..n]).await.is_err() {
                        return;
                    }
                }
            }
        }
    });
    let (c_read, c_write) = tokio::io::split(consumer_side);
    (
        SubstrateAddr::HostPort {
            host: peer_addr.0.to_string(),
            port: peer_addr.1,
        },
        TargetHandle {
            read: Box::new(c_read),
            write: Box::new(c_write),
        },
    )
}

/// The test accept loop (the dispatcher invokes the closure per
/// accept): pop one accepted pair, or park until notified.
fn queue_loop(queue: AcceptQueue) -> BindFirstAcceptLoop {
    let (queue, notify) = queue;
    Arc::new(move |_listener| {
        let queue = Arc::clone(&queue);
        let notify = Arc::clone(&notify);
        Box::pin(async move {
            let popped = {
                let notified = std::pin::pin!(notify.notified());
                let mut q = queue.lock().await;
                if let Some(pair) = q.pop_front() {
                    Some(pair)
                } else {
                    drop(q);
                    notified.await;
                    let mut q = queue.lock().await;
                    q.pop_front()
                }
            };
            popped.ok_or(TunnelEstablishError::DialFailed(
                "listener closed".to_string(),
            ))
        })
    })
}

/// Push an accepted pair into the test's accept queue.
async fn push_accept(queue: &AcceptQueue, peer: (&str, u16)) {
    let (peer_addr, handle) = accepted_halves(peer);
    queue.0.lock().await.push_back((peer_addr, handle));
    queue.1.notify_waiters();
}

/// 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: AcceptQueue,
}

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

async fn wire_suite_with_identity(identity: alkcall::core::auth::Identity) -> Suite {
    let resource_registry = ResourceRegistry::new();
    resource_registry
        .register(RESOURCE, Substrate::Tcp, "assembly-owned")
        .await;
    let queue = fresh_queue();
    let topo = harness::wire_bindfirst_composed(
        resource_registry,
        test_bind(),
        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(Arc::clone(&topo.opener), Arc::clone(&registry), 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 no-leak probe: the opener side ends with channel 0 (+ the
/// live channels the test names); the producer side ends with channel
/// 0, the SOCKS5 channels the test names, AND any odd id — the
/// producer-side ADOPTED forwarded channels (opener-allocated ids,
/// adopted inside the producer's `open_forwarded` caller; per-session
/// ownership per alktunnels ADR-005 — their entries reap at transport
/// EOF, not at the listener's teardown; upstream's own bind-first
/// suites make the same concession in `no_stray_channels`).
async fn assert_only_channels(
    topo: &harness::BindFirstTopology,
    opener_extra: &[u32],
    producer_extra: &[u32],
) {
    let opener_ids = topo.opener.manager().channel_ids();
    let producer_ids = topo.producer_manager.channel_ids();
    let opener_ok = opener_ids
        .iter()
        .all(|id| *id == 0 || opener_extra.contains(id));
    let producer_ok = producer_ids
        .iter()
        .all(|id| *id == 0 || *id % 2 == 1 || producer_extra.contains(id));
    assert!(
        opener_ok && producer_ok,
        "channel sets: opener={opener_ids:?} producer={producer_ids:?}"
    );
}

/// 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,
        alksocks::rfc1928::util::target_addr::TargetAddr::Ip(SocketAddr::from((
            [127, 0, 0, 1],
            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.
    push_accept(&suite.queue, ("10.0.0.1", 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,
        alksocks::rfc1928::util::target_addr::TargetAddr::Ip(SocketAddr::from((
            [10, 0, 0, 1],
            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_only_channels`
    // — its adopter's session dropped with the plane; the entry reaps
    // at transport EOF, not at the listener's).
    assert_only_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_only_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).
    push_accept(&suite.queue, ("10.0.0.1", PEER_PORT)).await;
    push_accept(&suite.queue, ("10.0.0.2", 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_only_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(harness::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.
    push_accept(&suite.queue, ("10.0.0.1", 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).
    push_accept(&suite.queue, ("10.0.0.1", 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.
    push_accept(&suite.queue, ("10.0.0.1", 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_only_channels(&suite.topo, &[0], &[0]).await;
}