//! The consumer BIND integration tests — `BindSession` end to end
//! through the harness's consumer-side topology (the producer's real
//! command-dispatch path over channels; the BIND phase model, ADR-007,
//! driven by the wrapper-aware client, POC #5's `read_reply2` shape).
//!
//! Covered: bind → reply#1 (BND.ADDR as the harness accept advertises)
//! → reply#2 (the peer address) → pass-through echo both directions;
//! the command-gate refusal surfaced as a typed error (a
//! `socks5:open`-only identity cannot bind); factory-failure and
//! accept-stage-failure mappings surfaced; quiet-wait garbage from the
//! client side ends the session; the three-command regression guard
//! from the consumer side.

mod harness;

use tokio::io::{AsyncReadExt, AsyncWriteExt};

use alksocks::consumer::{BindDataPlane, BindReply1, BindSession, Socks5ChannelSession};
use alksocks::egress::{BindAcceptError, BindAcceptFactoryError};
use alksocks::error::SocksSessionError;
use alksocks::rfc1928::consts;
use alksocks::rfc1928::util::target_addr::{TargetAddr, ToTargetAddr};
use alksocks::SocksError as SocksErrorKind;

use harness::{
    echo_egress, refusing_accept, refusing_accept_factory, wire_consumer_side_with, ECHO_BIND_PORT,
};

/// Both scopes: the open gate (`socks5:open`) AND the command gate
/// (`socks5:bind`) — a BIND session needs both grants explicitly
/// (ADR-006: never implied).
fn bind_scoped() -> alkcall::core::auth::Identity {
    harness::both_scopes_identity()
}

/// The no-leaked-channels probe (the consumer-connect suite's shape,
/// re-pinned for the BIND sessions): both managers hold only channel
/// 0 — the producer's entry reaps by its open-op wrapper; the probe
/// polls past the reap race.
async fn no_leaked_channels(topo: &harness::ConsumerSideTopology) {
    let deadline = std::time::Duration::from_secs(5);
    let consumer_ids = topo.consumer.manager().channel_ids();
    assert!(
        consumer_ids.iter().all(|&id| id == 0),
        "consumer leaked channel entries: {consumer_ids:?}"
    );
    tokio::time::timeout(deadline, async {
        loop {
            let producer_ids = topo.producer_manager.channel_ids();
            if producer_ids.iter().all(|&id| id == 0) {
                return;
            }
            tokio::time::sleep(std::time::Duration::from_millis(10)).await;
        }
    })
    .await
    .expect("the producer's handler-exit teardown reaps the entry");
}

/// A session error that carries an RFC reply code (the typed
/// reply-code-carrying shape the BIND paths promise).
fn reply_code_of(err: &SocksSessionError) -> Option<u8> {
    match err {
        SocksSessionError::Rfc(SocksErrorKind::ReplyError(code)) => Some(code.as_u8()),
        _ => None,
    }
}

/// The happy path: `bind` → reply#1 carries the harness accept's
/// advertised BND.ADDR (`ECHO_BIND_PORT` — the reply#1 carrier); the
/// app server's out-of-band dial shape is the `BindReply1` handle;
/// `await_inbound` reads reply#2 (the peer address, 10.0.0.7:44444 —
/// the harness's fixed stand-in); the data plane two-pumps raw bytes
/// both directions; `close` reaps the adopted channel.
#[tokio::test]
async fn bind_round_trip_reply1_reply2_then_pass_through() {
    let topo = wire_consumer_side_with(
        harness::echo_dial(),
        None,
        Some(harness::echo_accept()),
        Some(bind_scoped()),
    )
    .await;

    let mut session = BindSession::bind(
        topo.consumer.as_ref(),
        ("bind.test", ECHO_BIND_PORT).to_target_addr().unwrap(),
    )
    .await
    .expect("the session binds");

    // Reply#1: the advertised BND.ADDR (the harness accept's static
    // in-process address).
    let channel_id = session.channel_id();
    assert!(channel_id >= 1, "a producer-allocated id");
    assert!(topo.consumer.manager().has_channel(channel_id));
    let BindReply1 { bind_addr, .. } = session.reply1().expect("reply#1's handle");
    assert_eq!(
        bind_addr,
        TargetAddr::Ip(std::net::SocketAddr::from(([127, 0, 0, 1], ECHO_BIND_PORT))),
        "reply#1 carries the advertised BND.ADDR"
    );

    // Reply#2: the accepted peer (the harness's fixed stand-in), then
    // the raw pass-through data plane.
    let mut plane = session
        .await_inbound()
        .await
        .expect("reply#2 read")
        .expect("the plane exists before reply#2");
    assert_eq!(
        plane.peer,
        TargetAddr::Ip(std::net::SocketAddr::from(([10, 0, 0, 7], 44444))),
        "reply#2 carries the accepted peer"
    );

    // Phase 2: raw pass-through, both directions.
    plane.write_all(b"ping").await.expect("write");
    plane.flush().await.expect("flush");
    let mut buf = [0u8; 4];
    tokio::time::timeout(
        std::time::Duration::from_secs(10),
        plane.read_exact(&mut buf),
    )
    .await
    .expect("echo timed out")
    .expect("read the echo");
    assert_eq!(&buf, b"ping");

    // Graceful teardown: half-close (the EOF sentinel) + reap.
    drop(plane);
    assert!(session.close().await, "the adopted channel entry existed");
    no_leaked_channels(&topo).await;
}

/// The `BindDataPlane` accessors: `into_inner` yields the owned
/// forked client type (read/write rides it directly); the session
/// still owns the teardown — `close` after the plane is drained reaps
/// the entry.
#[tokio::test]
async fn bind_data_plane_accessors_into_inner_and_socket_mut() {
    let topo = wire_consumer_side_with(
        harness::echo_dial(),
        None,
        Some(harness::echo_accept()),
        Some(bind_scoped()),
    )
    .await;

    let mut session = BindSession::bind(
        topo.consumer.as_ref(),
        ("plane.test", ECHO_BIND_PORT).to_target_addr().unwrap(),
    )
    .await
    .expect("the session binds");
    session
        .reply1()
        .expect("reply#1's handle")
        .raw_mut()
        .get_socket_mut()
        .write_all(b"pre")
        .await
        .expect("pre-reply#2 write");
    let mut plane = session
        .await_inbound()
        .await
        .expect("reply#2 read")
        .expect("the plane");
    // `get_socket_mut` is the borrowed socket accessor (the POC
    // shape) — the same stream, reached through the plane.
    plane
        .get_socket_mut()
        .write_all(b"raw")
        .await
        .expect("write");
    let mut inner = plane.into_inner();
    let mut buf = [0u8; 6];
    tokio::time::timeout(
        std::time::Duration::from_secs(10),
        inner.read_exact(&mut buf),
    )
    .await
    .expect("echo timed out")
    .expect("read the echo");
    assert_eq!(&buf, b"preraw", "both write paths rode the same pipe");
    drop(inner);

    assert!(session.close().await, "the entry existed");
    no_leaked_channels(&topo).await;
}

/// The command gate (ADR-006): a `socks5:open`-only identity cannot
/// bind — the refusal surfaces as a typed session error carrying the
/// RFC reply code (`ConnectionNotAllowed` 0x02), and no phantom
/// channel remains on either side.
#[tokio::test]
async fn bind_command_gate_refusal_is_a_typed_reply_code_error() {
    let topo = wire_consumer_side_with(
        harness::echo_dial(),
        None,
        Some(harness::echo_accept()),
        Some(harness::consumer_identity()),
    )
    .await;

    let err = BindSession::bind(
        topo.consumer.as_ref(),
        ("gate.test", ECHO_BIND_PORT).to_target_addr().unwrap(),
    )
    .await
    .expect_err("the gate refuses a socks5:open-only identity");

    assert_eq!(
        reply_code_of(&err),
        Some(consts::SOCKS5_REPLY_CONNECTION_NOT_ALLOWED),
        "the command gate's 0x02 rides the session error: {err:?}"
    );

    // Teardown-at-birth: the adopted channel was reaped before the
    // error returned (no phantom channel, either side).
    no_leaked_channels(&topo).await;
}

/// Factory failure (pre-reply#1) maps to reply#1's code, surfaced
/// through the session type — no session lives past the refusal, no
/// phantom listener.
#[tokio::test]
async fn bind_factory_failure_surfaces_reply1_code() {
    let topo = wire_consumer_side_with(
        harness::echo_dial(),
        None,
        Some(refusing_accept_factory(BindAcceptFactoryError::Refused(
            "port held".into(),
        ))),
        Some(bind_scoped()),
    )
    .await;

    let err = BindSession::bind(
        topo.consumer.as_ref(),
        ("factory.fail", ECHO_BIND_PORT).to_target_addr().unwrap(),
    )
    .await
    .expect_err("the factory refused");

    assert_eq!(
        reply_code_of(&err),
        Some(consts::SOCKS5_REPLY_CONNECTION_REFUSED),
        "the factory refusal maps to reply#1's code: {err:?}"
    );
    no_leaked_channels(&topo).await;
}

/// Accept-stage failure (post-reply#1) maps to reply#2's code —
/// surfaced at `await_inbound`, no data plane; the session's channel
/// is still owned (close/drop reaps it).
#[tokio::test]
async fn bind_accept_stage_failure_surfaces_reply2_code() {
    let topo = wire_consumer_side_with(
        harness::echo_dial(),
        None,
        Some(refusing_accept(BindAcceptError::NotAllowed(
            "policy".into(),
        ))),
        Some(bind_scoped()),
    )
    .await;

    let mut session = BindSession::bind(
        topo.consumer.as_ref(),
        ("accept.fail", ECHO_BIND_PORT).to_target_addr().unwrap(),
    )
    .await
    .expect("reply#1 succeeded — the factory resolved");

    let err = match session.await_inbound().await {
        Err(err) => err,
        Ok(Some(_)) => panic!("the accept-stage refusal must not yield a data plane"),
        Ok(None) => panic!("the plane existed before reply#2"),
    };
    assert_eq!(
        reply_code_of(&err),
        Some(consts::SOCKS5_REPLY_CONNECTION_NOT_ALLOWED),
        "reply#2's mapped code surfaces at await_inbound: {err:?}"
    );

    // The channel entry stays adopted (reply#1 succeeded; the session
    // owns it until the app closes it) — close reaps it.
    assert!(session.close().await, "the entry existed");
    no_leaked_channels(&topo).await;
}

/// Quiet-wait garbage from the client side: a control byte after
/// reply#1 is protocol garbage — the producer ends the session with
/// no data plane, so the consumer's `await_inbound` read EOFs (an I/O
/// error, not a reply#2).
#[tokio::test]
async fn bind_quiet_wait_garbage_from_the_client_ends_the_session() {
    let parked: alksocks::egress::BindAcceptFactory =
        std::sync::Arc::new(move |_target: TargetAddr| {
            Box::pin(async move {
                Ok(alksocks::egress::BindAccept::new(
                    std::net::SocketAddr::from(([127, 0, 0, 1], ECHO_BIND_PORT)),
                    std::future::pending::<
                        Result<alksocks::egress::BindTargetHandle, BindAcceptError>,
                    >(),
                ))
            })
        });
    let topo = wire_consumer_side_with(
        harness::echo_dial(),
        None,
        Some(parked),
        Some(bind_scoped()),
    )
    .await;

    let mut session = BindSession::bind(
        topo.consumer.as_ref(),
        ("quiet.test", ECHO_BIND_PORT).to_target_addr().unwrap(),
    )
    .await
    .expect("reply#1 succeeded");

    // Protocol garbage in the quiet wait (the client's side).
    session
        .reply1()
        .expect("reply#1's handle")
        .raw_mut()
        .get_socket_mut()
        .write_all(&[0x05, 0x02])
        .await
        .expect("garbage written");

    // The session ends: the reply#2 read EOFs (no reply#2, no data
    // plane).
    let err = match session.await_inbound().await {
        Err(err) => err,
        Ok(Some(_)) => panic!("the garbage must not yield a data plane"),
        Ok(None) => panic!("the plane existed before reply#2"),
    };
    assert!(
        matches!(err, SocksSessionError::Io(_)),
        "the quiet-wait garbage surfaces as the EOF'd reply#2 read: {err:?}"
    );

    // The channel is still owned (reply#1 succeeded); close reaps it.
    assert!(session.close().await, "the entry existed");
    no_leaked_channels(&topo).await;
}

/// The three-command regression guard from the consumer side (POC #5
/// test 7's consumer-side mirror): one registration, all three
/// commands served through their consumer sessions — CONNECT
/// (`Socks5ChannelSession`), BIND (`BindSession`), ASSOCIATE rides
/// its task; here CONNECT + BIND coexist on one registration with the
/// both-scopes identity, fresh channel per session.
#[tokio::test]
async fn bind_and_connect_serve_on_the_same_registration() {
    let topo = wire_consumer_side_with(
        harness::echo_dial(),
        Some(echo_egress()),
        Some(harness::echo_accept()),
        Some(bind_scoped()),
    )
    .await;

    // CONNECT serves (the CONNECT session over the same registration).
    let mut connect_session = Socks5ChannelSession::connect(
        topo.consumer.as_ref(),
        ("echo.test", 443u16).to_target_addr().unwrap(),
    )
    .await
    .expect("the connect session");
    connect_session
        .split()
        .expect("pipe")
        .1
        .write_all(b"tcp")
        .await
        .expect("write");
    let (mut read, _) = connect_session.split().expect("pipe");
    let mut buf = [0u8; 3];
    tokio::time::timeout(
        std::time::Duration::from_secs(10),
        read.read_exact(&mut buf),
    )
    .await
    .expect("echo timed out")
    .expect("read the echo");
    assert_eq!(&buf, b"tcp");
    connect_session.close().await;

    // BIND serves the same registration (fresh channel).
    let mut bind_session = BindSession::bind(
        topo.consumer.as_ref(),
        ("bind.test", ECHO_BIND_PORT).to_target_addr().unwrap(),
    )
    .await
    .expect("the bind session");
    let mut plane = bind_session
        .await_inbound()
        .await
        .expect("reply#2")
        .expect("the plane");
    plane.write_all(b"bind").await.expect("write");
    plane.flush().await.expect("flush");
    let mut buf = [0u8; 4];
    tokio::time::timeout(
        std::time::Duration::from_secs(10),
        plane.read_exact(&mut buf),
    )
    .await
    .expect("echo timed out")
    .expect("read the echo");
    assert_eq!(&buf, b"bind");
    drop(plane);
    bind_session.close().await;
    no_leaked_channels(&topo).await;
}

/// Drop teardown: dropping a live BIND session mid-pipe reaps the
/// adopted channel entry (no leak) and the connection stays healthy —
/// a fresh bind succeeds immediately after (the CONNECT session's
/// drop posture, re-pinned for BIND).
#[tokio::test]
async fn bind_drop_teardown_reaps_and_connection_stays_healthy() {
    let topo = wire_consumer_side_with(
        harness::echo_dial(),
        None,
        Some(harness::echo_accept()),
        Some(bind_scoped()),
    )
    .await;

    let mut session = BindSession::bind(
        topo.consumer.as_ref(),
        ("dropped.test", ECHO_BIND_PORT).to_target_addr().unwrap(),
    )
    .await
    .expect("the session binds");
    let BindReply1 { bind_addr, .. } = session.reply1().expect("reply#1's handle");
    assert_eq!(
        bind_addr,
        TargetAddr::Ip(std::net::SocketAddr::from(([127, 0, 0, 1], ECHO_BIND_PORT)))
    );

    // Drop mid-pipe (before reply#2 — the quiet-wait phase).
    drop(session);
    no_leaked_channels(&topo).await;

    // The connection is healthy: a fresh bind round-trips.
    let mut fresh = BindSession::bind(
        topo.consumer.as_ref(),
        ("fresh.test", ECHO_BIND_PORT).to_target_addr().unwrap(),
    )
    .await
    .expect("fresh bind after the drop");
    let mut plane = fresh
        .await_inbound()
        .await
        .expect("reply#2")
        .expect("the plane");
    plane.write_all(b"again").await.expect("write");
    plane.flush().await.expect("flush");
    let mut buf = [0u8; 5];
    tokio::time::timeout(
        std::time::Duration::from_secs(10),
        plane.read_exact(&mut buf),
    )
    .await
    .expect("echo timed out")
    .expect("read the fresh echo");
    assert_eq!(&buf, b"again");
    drop(plane);
    fresh.close().await;
    no_leaked_channels(&topo).await;
}

/// The no-`Clone` posture for the BIND session (the CONNECT session's
/// compile shape, mirrored): the doctest on the type is the
/// compile-assert; this probe pins `Send`.
#[test]
fn bind_session_type_is_send() {
    fn assert_send<T: Send>() {}
    assert_send::<BindSession>();
    assert_send::<BindDataPlane>();
    assert_send::<BindReply1<'static>>();
}

/// The C-1 re-entry guard (review 002): a second `await_inbound` on
/// the same session cannot re-read the data plane — it returns the
/// spent shape (`Ok(None)`), and the data-plane bytes survive
/// untouched (the first plane's echo round-trip is intact after the
/// second call; the pinning test for the silent-corruption path).
#[tokio::test]
async fn double_await_inbound_returns_the_spent_shape_and_bytes_survive() {
    let topo = wire_consumer_side_with(
        harness::echo_dial(),
        None,
        Some(harness::echo_accept()),
        Some(bind_scoped()),
    )
    .await;

    let mut session = BindSession::bind(
        topo.consumer.as_ref(),
        ("reentry.test", ECHO_BIND_PORT).to_target_addr().unwrap(),
    )
    .await
    .expect("the session binds");

    // The first read: the plane exists (and consumes the phase —
    // the plane owns the stream; the session stays the teardown
    // owner).
    let mut plane = session
        .await_inbound()
        .await
        .expect("reply#2 read")
        .expect("the plane exists before reply#2");
    plane.write_all(b"ping").await.expect("write");
    plane.flush().await.expect("flush");
    let mut buf = [0u8; 4];
    tokio::time::timeout(
        std::time::Duration::from_secs(10),
        plane.read_exact(&mut buf),
    )
    .await
    .expect("echo timed out")
    .expect("read the echo");
    assert_eq!(&buf, b"ping", "the first plane's round-trip");
    drop(plane);

    // The second call: the spent shape — `Ok(None)`, never a
    // re-read of the live data plane.
    let second = session
        .await_inbound()
        .await
        .expect("the spent shape is an Ok, not an error");
    assert!(second.is_none(), "the phase was consumed");

    // The data-plane bytes survive: the first plane's echo
    // round-trip was complete and no re-read corrupted the pipe —
    // the channel is closed with the reap (the session owner's
    // `close`), and a fresh bind on the same registration
    // round-trips with its own bytes intact.
    assert!(session.close().await, "the entry existed");
    no_leaked_channels(&topo).await;

    let mut fresh = BindSession::bind(
        topo.consumer.as_ref(),
        ("after.test", ECHO_BIND_PORT).to_target_addr().unwrap(),
    )
    .await
    .expect("fresh bind after the spent call");
    let mut plane = fresh
        .await_inbound()
        .await
        .expect("reply#2")
        .expect("the plane");
    plane.write_all(b"again").await.expect("write");
    plane.flush().await.expect("flush");
    let mut buf = [0u8; 5];
    tokio::time::timeout(
        std::time::Duration::from_secs(10),
        plane.read_exact(&mut buf),
    )
    .await
    .expect("echo timed out")
    .expect("read the fresh echo");
    assert_eq!(&buf, b"again", "the fresh session's bytes are intact");
    drop(plane);
    fresh.close().await;
    no_leaked_channels(&topo).await;
}

/// The `reply1()`-twice re-entry path (C-1's other door, closed by
/// the D-1 one-reader decision): `reply1()` is a view — fresh handles
/// per call, no read phase — so repeated `reply1()` calls carry the
/// same advertised address without consuming anything, and the one
/// normative reader (`await_inbound`) still yields exactly one plane.
#[tokio::test]
async fn reply1_twice_is_a_view_and_the_reader_yields_one_plane() {
    let topo = wire_consumer_side_with(
        harness::echo_dial(),
        None,
        Some(harness::echo_accept()),
        Some(bind_scoped()),
    )
    .await;

    let mut session = BindSession::bind(
        topo.consumer.as_ref(),
        ("view.test", ECHO_BIND_PORT).to_target_addr().unwrap(),
    )
    .await
    .expect("the session binds");

    let expected = TargetAddr::Ip(std::net::SocketAddr::from(([127, 0, 0, 1], ECHO_BIND_PORT)));

    // Repeated views: every handle carries the same advertised
    // address; nothing is consumed.
    let BindReply1 { bind_addr, .. } = session.reply1().expect("reply#1's handle");
    assert_eq!(bind_addr, expected);
    let BindReply1 { bind_addr, .. } = session.reply1().expect("reply#1's handle again");
    assert_eq!(bind_addr, expected, "a view, not a reader");

    // The one normative reader: exactly one plane.
    let mut plane = session
        .await_inbound()
        .await
        .expect("reply#2 read")
        .expect("the plane");
    plane.write_all(b"once").await.expect("write");
    plane.flush().await.expect("flush");
    let mut buf = [0u8; 4];
    tokio::time::timeout(
        std::time::Duration::from_secs(10),
        plane.read_exact(&mut buf),
    )
    .await
    .expect("echo timed out")
    .expect("read the echo");
    assert_eq!(&buf, b"once");
    drop(plane);

    // The re-entry is closed on this door too.
    assert!(
        session
            .await_inbound()
            .await
            .expect("spent shape")
            .is_none(),
        "no second plane after the reader ran"
    );
    session.close().await;
    no_leaked_channels(&topo).await;
}