//! The consumer-half integration tests: `Socks5ChannelSession`
//! (CONNECT) end to end through the harness's consumer-side topology
//! (the producer's real command-dispatch path over channels; the
//! alkcall 0.8.0 establishment surface consumed from the consumer's
//! own `ChannelClient`).
//!
//! Covered: connect → echo round-trip (payload bytes both
//! directions); the dial-witness target forms (IP-string parses to an
//! IP target, domain passes through unresolved (ADR-012)); refused
//! dial → the RFC reply code surfaced as a typed session error; the
//! scope-less FORBIDDEN open → typed open error, no session; drop
//! teardown (no leaked channel entries, the connection stays
//! healthy); half-close semantics (the write-half shutdown EOFs the
//! producer's pump cleanly); the compile-asserted no-`Clone` posture.

mod harness;

use std::sync::Arc;

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

use alksocks::consumer::{Socks5ChannelSession, TakenInner};
use alksocks::egress::SocksDialError;
use alksocks::error::SocksSessionError;
use alksocks::rfc1928::consts;
use alksocks::rfc1928::util::target_addr::{TargetAddr, ToTargetAddr};
use alksocks::rfc1928::SocksError as SocksErrorKind;

use harness::{
    echo_dial, refusing_dial, wire_consumer_side, wire_consumer_side_with, ConsumerSideTopology,
};

/// A `SocksOpenError`-carrying open failure (the FORBIDDEN probe's
/// expectation shape).
fn expect_forbidden(err: &SocksSessionError) -> bool {
    match err {
        SocksSessionError::Open(open) => match open.open_ref() {
            alkcall::channels::client::ChannelOpenError::CallFailed { error } => {
                error.code == "FORBIDDEN"
            }
            _ => false,
        },
        _ => false,
    }
}

/// A dial that always refuses (the refusal-riding-in-stream probe's
/// dial; the RFC reply code must surface through the session type,
/// not as an open failure).
fn refusing_session_dial() -> alksocks::egress::DialFn {
    refusing_dial(SocksDialError::Refused("target down".into()))
}

/// Both managers hold only channel 0 (the phantom-channel property,
/// asserted from both sides). The producer's entry is reaped by its
/// open-op wrapper when the pump handler exits (an await-able
/// teardown that races the consumer-side reap; this probe polls past
/// the race instead of asserting instantly.
async fn no_leaked_channels(topo: &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");
}

/// The happy path: `connect` → the session is a raw byte pipe → echo
/// round-trip both directions; the dial seam saw the parsed target;
/// the establisher witnessed the opener identity (CF-006).
#[tokio::test]
async fn connect_echo_round_trip_and_dial_witness() {
    let topo = wire_consumer_side(echo_dial()).await;

    let mut session = Socks5ChannelSession::connect(
        topo.consumer.as_ref(),
        ("echo.test", 443u16).to_target_addr().unwrap(),
    )
    .await
    .expect("the session connects");

    // The `split()`-returned halves are the last use of the session;
    // the tail test's alias import lives here.
    session
        .split()
        .expect("post-reply pipe")
        .1
        .write_all(b"ping")
        .await
        .expect("write the probe");
    let (mut read, _) = session.split().expect("post-reply pipe");
    let mut buf = [0u8; 4];
    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"ping");

    // The establisher witnessed the per-call opener identity.
    assert_eq!(
        topo.identity_witness.lock().await.as_deref(),
        Some("consumer")
    );

    session.close().await;
    no_leaked_channels(&topo).await;
}

/// Target construction (ADR-012): an IP-string form parses to an IP
/// target (the dial seam receives `TargetAddr::Ip`); the domain form
/// passes through unresolved (`TargetAddr::Domain`). Both ride the
/// same happy path.
#[tokio::test]
async fn target_construction_ip_strings_parse_domains_pass_through() {
    let targets = Arc::new(tokio::sync::Mutex::new(Vec::new()));
    let topo = wire_consumer_side(harness::witnessing_echo_dial(Arc::clone(&targets))).await;

    // The domain form: passes through unresolved (the dial's job).
    let session = Socks5ChannelSession::connect(
        topo.consumer.as_ref(),
        ("echo.test", 443u16).to_target_addr().unwrap(),
    )
    .await
    .expect("domain connect");
    session.close().await;

    // The IP-string form: parsed to an IP target by `ToTargetAddr`.
    let session = Socks5ChannelSession::connect(
        topo.consumer.as_ref(),
        ("10.0.0.1", 8443u16).to_target_addr().unwrap(),
    )
    .await
    .expect("ip-string connect");
    session.close().await;

    let dialed = targets.lock().await.clone();
    assert_eq!(
        dialed,
        vec![
            TargetAddr::Domain("echo.test".to_owned(), 443),
            TargetAddr::Ip(std::net::SocketAddr::from(([10, 0, 0, 1], 8443)))
        ],
        "the dial seam received the parsed target forms (ADR-012)"
    );
    no_leaked_channels(&topo).await;
}

/// A refused dial: the RFC reply code surfaces through the session
/// type (`SocksSessionError::Rfc`, the in-stream refusal, never an
/// open failure), and no phantom channel remains on either side.
#[tokio::test]
async fn refused_dial_surfaces_rfc_reply_code_no_phantom_channel() {
    let topo = wire_consumer_side(refusing_session_dial()).await;

    let err = Socks5ChannelSession::connect(
        topo.consumer.as_ref(),
        ("down.test", 80u16).to_target_addr().unwrap(),
    )
    .await
    .expect_err("the dial refused; the reply code rides the session error");

    match &err {
        SocksSessionError::Rfc(SocksErrorKind::ReplyError(code)) => {
            assert_eq!(code.as_u8(), consts::SOCKS5_REPLY_CONNECTION_REFUSED);
        }
        other => panic!("expected an Rfc reply-code refusal, got {other:?}"),
    }

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

    // The connection is healthy: a fresh connect through a working
    // dial succeeds immediately.
    let topo2 = wire_consumer_side(echo_dial()).await;
    let session = Socks5ChannelSession::connect(
        topo2.consumer.as_ref(),
        ("echo.test", 443u16).to_target_addr().unwrap(),
    )
    .await
    .expect("fresh connect after the refusal");
    session.close().await;
}

/// A scope-less identity: the registry's FORBIDDEN open failure is
/// the typed `SocksOpenError` through `SocksSessionError::Open`;
/// no session, no phantom channel (the open gate is the boundary).
#[tokio::test]
async fn forbidden_open_is_a_typed_open_error_with_no_session() {
    let bare = alkcall::core::auth::Identity {
        id: "scopeless".to_string(),
        scopes: vec![],
        resources: Default::default(),
    };
    let topo = wire_consumer_side_with(echo_dial(), None, None, Some(bare)).await;

    let err = Socks5ChannelSession::connect(
        topo.consumer.as_ref(),
        ("gated.test", 443u16).to_target_addr().unwrap(),
    )
    .await
    .expect_err("a scope-less identity fails closed");

    assert!(
        expect_forbidden(&err),
        "FORBIDDEN is the open error: {err:?}"
    );
    no_leaked_channels(&topo).await;
}

/// Drop teardown: dropping a live session mid-pipe reaps the adopted
/// channel entry (no leak) and the connection stays healthy: a
/// fresh connect succeeds immediately after.
#[tokio::test]
async fn drop_teardown_reaps_and_connection_stays_healthy() {
    let topo = wire_consumer_side(echo_dial()).await;

    let mut session = Socks5ChannelSession::connect(
        topo.consumer.as_ref(),
        ("dropped.test", 443u16).to_target_addr().unwrap(),
    )
    .await
    .expect("the session connects");
    session
        .split()
        .expect("pipe")
        .1
        .write_all(b"mid")
        .await
        .expect("write before the drop");
    let (mut read, _) = session.split().expect("pipe");
    let mut buf = [0u8; 3];
    read.read_exact(&mut buf).await.expect("echo before drop");
    assert_eq!(&buf, b"mid");

    // Drop mid-stream (the teardown-ownership posture).
    drop(session);
    no_leaked_channels(&topo).await;

    // The connection is healthy: a fresh session round-trips.
    let mut fresh = Socks5ChannelSession::connect(
        topo.consumer.as_ref(),
        ("fresh.test", 443u16).to_target_addr().unwrap(),
    )
    .await
    .expect("fresh connect after the drop");
    fresh
        .split()
        .expect("pipe")
        .1
        .write_all(b"again")
        .await
        .expect("write");
    let (mut read, _) = fresh.split().expect("pipe");
    let mut buf = [0u8; 5];
    tokio::time::timeout(
        std::time::Duration::from_secs(10),
        read.read_exact(&mut buf),
    )
    .await
    .expect("echo timed out")
    .expect("read the fresh echo");
    assert_eq!(&buf, b"again");
    fresh.close().await;
    no_leaked_channels(&topo).await;
}

/// Half-close semantics: shutting the write half EOFs the producer's
/// pump cleanly (the EOF sentinel), while the read half stays alive
/// until the target's data flows. The pump's completion is the
/// producer-side evidence (the handler tracks the data plane; R-02).
#[tokio::test]
async fn half_close_write_eof_ends_the_producer_pump() {
    let topo = wire_consumer_side(echo_dial()).await;

    let session = Socks5ChannelSession::connect(
        topo.consumer.as_ref(),
        ("halfclose.test", 443u16).to_target_addr().unwrap(),
    )
    .await
    .expect("the session connects");
    let channel_id = session.channel_id();
    assert!(channel_id >= 1, "a producer-allocated id");
    assert!(topo.consumer.manager().has_channel(channel_id));

    // Split into owned halves via `into_inner` (the advanced-use
    // accessor), then half-close the write half.
    let TakenInner {
        session: _owner,
        inner,
    } = session.into_inner().expect("the session is live");
    let (mut read, mut write) = tokio::io::split(inner);
    write.write_all(b"pre").await.expect("write");
    let mut buf = [0u8; 3];
    read.read_exact(&mut buf).await.expect("echo");
    assert_eq!(&buf, b"pre");
    write.shutdown().await.expect("half-close the pipe");

    // The producer's pump observes the EOF sentinel and ends (the
    // two-pump shape's shutdown-on-completion; the echo target's
    // pump exits with it). The session-side observable: the read
    // half drains to EOF.
    drop(write);
    let mut eof = [0u8; 1];
    tokio::time::timeout(std::time::Duration::from_secs(10), async {
        loop {
            match read.read(&mut eof).await {
                Ok(0) | Err(_) => break,
                Ok(_) => continue,
            }
        }
    })
    .await
    .expect("the read resolves after the half-close");
    drop(read);
    drop(_owner);
    no_leaked_channels(&topo).await;
}

/// `into_inner` yields the forked client type paired with the
/// teardown-owning session; the pair's `close` reaps the channel.
#[tokio::test]
async fn into_inner_pairs_the_pipe_with_the_teardown_owner() {
    let topo = wire_consumer_side(echo_dial()).await;

    let session = Socks5ChannelSession::connect(
        topo.consumer.as_ref(),
        ("inner.test", 443u16).to_target_addr().unwrap(),
    )
    .await
    .expect("the session connects");
    let channel_id = session.channel_id();
    let TakenInner { session, inner } = session.into_inner().expect("the session is live");
    assert!(topo.consumer.manager().has_channel(channel_id));

    let (mut read, write) = tokio::io::split(inner);
    let mut buf = [0u8; 1];
    // The pipe is idle; the close below reaps the entry.
    let _ = tokio::time::timeout(std::time::Duration::from_millis(50), read.read(&mut buf)).await;
    drop(write);
    drop(read);

    let reaped = session.close().await;
    assert!(reaped, "the adopted channel entry existed");
    no_leaked_channels(&topo).await;
}

/// The no-`Clone` posture: the type is `!Clone` because its fields
/// are (the `compile_fail` doctest pins the type-level posture).
#[test]
fn session_type_is_not_clone() {
    // A `Clone`-bound assertion on a non-`Clone` type cannot be a
    // runtime test; the doctest is the compile-assert. This probe
    // pins the Send shape the type promises.
    fn assert_send<T: Send>() {}
    assert_send::<Socks5ChannelSession>();
}

/// The read half stays alive across a half-close (the echo flows
/// after the write half's EOF was already seen by the producer): a
/// payload written before the shutdown round-trips, and the read
/// half observes the target's own EOF after it (the two-pump
/// half-close semantics, both directions).
#[tokio::test]
async fn both_directions_survive_across_halves() {
    let topo = wire_consumer_side(echo_dial()).await;

    let mut session = Socks5ChannelSession::connect(
        topo.consumer.as_ref(),
        ("halves.test", 443u16).to_target_addr().unwrap(),
    )
    .await
    .expect("the session connects");

    // Write half: two writes, flush each (the chunk stream carries
    // them in order).
    let (_, mut write) = session.split().expect("pipe");
    write.write_all(b"one").await.expect("write one");
    write.flush().await.expect("flush");
    write.write_all(b"two").await.expect("write two");
    write.flush().await.expect("flush");
    drop(write);

    // Read half: the echo arrives in order.
    let (mut read, _) = session.split().expect("pipe");
    let mut buf = [0u8; 6];
    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"onetwo");

    session.close().await;
    no_leaked_channels(&topo).await;
}

/// The `split()` accessor's `Option` shape: `None` after the pipe
/// was taken (the spent-session posture; the borrow checker makes
/// the direct post-`into_inner` call the only reachable shape).
#[tokio::test]
async fn close_after_into_inner_reaps() {
    let topo = wire_consumer_side(echo_dial()).await;
    let session = Socks5ChannelSession::connect(
        topo.consumer.as_ref(),
        ("spent.test", 443u16).to_target_addr().unwrap(),
    )
    .await
    .expect("the session connects");
    let channel_id = session.channel_id();
    let TakenInner { session, .. } = session.into_inner().expect("the session is live");
    // The inner pipe was taken; `close` still reaps the entry.
    assert!(session.close().await, "the entry existed before the close");
    no_leaked_channels(&topo).await;
    assert!(channel_id >= 1);
}