//! Integration tests for the direct op (`channels/tunnel/direct`,
//! ADR-007) end to end over the duplex harness: the params → spec →
//! establisher → registration → `TunnelSession::open_direct` chain.
//!
//! Covered: e2e round-trips through `pump_bidi` (tcp + udp — the
//! framed adapter wraps at the dial closure, datagram boundary
//! preserved), the schema-class failure for malformed targets (no
//! phantom session), the `NOT_FOUND` posture on a producer without
//! the direct op registered, the structural scope separation
//! (`tunnel:open` does NOT imply `tunnel:direct` — FORBIDDEN, and
//! vice versa), and the CF-006 per-call opener identity on the
//! direct establisher (witness).

mod harness;

use std::sync::Arc;

use alkcall::protocol::wire::CallError;
use alktunnels::params::{Substrate, SubstrateAddr};
use alktunnels::{TunnelEstablishError, TunnelSession};

use harness::{
    all_substrate_dial, consumer_identity, echo_dial, failing_dial, framed_udp_echo_dial, wire,
    wire_direct, wire_direct_forward, wire_direct_with_identity, Topology,
};

fn target_hostport(host: &str, port: u16) -> SubstrateAddr {
    SubstrateAddr::HostPort {
        host: host.to_string(),
        port,
    }
}

/// Call the producer's direct op on channel 0 (the raw call surface),
/// returning the wire `CallError` on failure.
async fn call_direct(
    topo: &Topology,
    input: serde_json::Value,
) -> Result<serde_json::Value, CallError> {
    let payload = serde_json::json!({
        "operationId": "channels/tunnel/direct",
        "input": input,
    });
    let response = topo.consumer_call.call_with_payload(payload).await;
    response.result
}

/// Adopt the producer-allocated channel and pump it against in-process
/// halves (the consumer-side data plane, driven manually).
async fn adopt_and_pump(
    topo: &Topology,
    channel_id: u32,
) -> (
    tokio::io::ReadHalf<tokio::io::DuplexStream>,
    tokio::io::WriteHalf<tokio::io::DuplexStream>,
    tokio::task::JoinHandle<(u64, u64)>,
) {
    let (send, recv) = topo
        .consumer_manager
        .adopt_channel(channel_id, "alk/tunnel", None)
        .await
        .expect("adopt producer-allocated channel");
    let bidi = alkcall::core::types::BiStream::from_joined(recv, send);
    let (channel_end, local_end) = tokio::io::duplex(64 * 1024);
    let (c_read, c_write) = tokio::io::split(channel_end);
    let pump = tokio::spawn(alkcall::channels::pump::pump_bidi(bidi, c_read, c_write));
    let (l_read, l_write) = tokio::io::split(local_end);
    (l_read, l_write, pump)
}

/// Only channel 0 may remain — a failed open leaves no data channel
/// on either side (the phantom-channel property).
fn no_data_channels(topo: &Topology) -> bool {
    let consumer = topo.consumer_manager.channel_ids();
    let producer = topo.producer.manager().channel_ids();
    consumer.iter().all(|&id| id == 0) && producer.iter().all(|&id| id == 0)
}

fn establishment_reason_of(err: &CallError) -> Option<&str> {
    err.details
        .as_ref()
        .and_then(|d| d.get("reason"))
        .and_then(|r| r.as_str())
}

#[tokio::test]
async fn direct_tcp_round_trips_through_pump_bidi() {
    let topo = wire_direct(echo_dial()).await;

    let out = call_direct(
        &topo,
        serde_json::json!({
            "substrate": "tcp",
            "target": {"host": "db.internal", "port": 5432}
        }),
    )
    .await
    .expect("direct open must succeed");
    let channel_id = out["channel_id"].as_u64().expect("channel_id") as u32;
    assert!(topo.producer.manager().has_channel(channel_id));

    let (mut read, mut write, pump) = adopt_and_pump(&topo, channel_id).await;
    use tokio::io::{AsyncReadExt, AsyncWriteExt};
    write.write_all(b"ping-direct").await.expect("write");
    write.flush().await.expect("flush");
    let mut buf = vec![0u8; 11];
    tokio::time::timeout(std::time::Duration::from_secs(5), read.read_exact(&mut buf))
        .await
        .expect("round trip timed out")
        .expect("read");
    assert_eq!(&buf, b"ping-direct");

    // EOF propagates and the wrapper reaps the producer-side channel.
    drop(write);
    drop(read);
    let (c2p, p2c) = tokio::time::timeout(std::time::Duration::from_secs(5), pump)
        .await
        .expect("pump completion timed out")
        .expect("pump task");
    assert!(c2p > 0 && p2c > 0, "both pumps moved bytes: {c2p}, {p2c}");
    tokio::time::sleep(std::time::Duration::from_millis(100)).await;
    assert!(
        !topo.producer.manager().has_channel(channel_id),
        "wrapper reaped the producer-side channel after pump completion"
    );
}

#[tokio::test]
async fn direct_udp_round_trips_framed() {
    // The UDP dial closure wraps the framed adapter (ADR-003) at the
    // boundary — the establisher stays substrate-agnostic; the
    // session's datagram plane sees boundary-preserved frames.
    let topo = wire_direct(all_substrate_dial()).await;

    let out = call_direct(
        &topo,
        serde_json::json!({
            "substrate": "udp",
            "target": {"host": "10.0.0.7", "port": 53}
        }),
    )
    .await
    .expect("udp direct open must succeed");
    let channel_id = out["channel_id"].as_u64().expect("channel_id") as u32;

    let mut session = TunnelSession::adopt(
        &topo.consumer_manager,
        channel_id,
        Substrate::Udp,
        "alk/tunnel",
    )
    .await
    .expect("adopt into a udp session");
    session
        .send_datagram(b"datagram-direct")
        .await
        .expect("send");
    let dg = tokio::time::timeout(std::time::Duration::from_secs(5), session.recv_datagram())
        .await
        .expect("round trip timed out")
        .expect("recv")
        .expect("datagram");
    assert_eq!(dg.as_ref(), b"datagram-direct");
    session.close().await;
}

#[tokio::test]
async fn malformed_target_is_schema_class_and_leaves_no_channel() {
    // The registry's input-schema gate runs before the establisher: a
    // unix target without `path` (and a tcp target without `port`)
    // fails as `INVALID_INPUT` — never a phantom session, never a
    // establisher dial.
    let topo = wire_direct(echo_dial()).await;

    let err = call_direct(
        &topo,
        serde_json::json!({
            "substrate": "unix",
            "target": {"host": "nope", "port": 1}
        }),
    )
    .await
    .expect_err("unix target without path must fail the schema gate");
    assert_eq!(err.code, "INVALID_INPUT");
    assert!(establishment_reason_of(&err).is_none());
    assert!(no_data_channels(&topo));

    let err = call_direct(
        &topo,
        serde_json::json!({
            "substrate": "tcp",
            "target": {"host": "nope"}
        }),
    )
    .await
    .expect_err("tcp target without port must fail the schema gate");
    assert_eq!(err.code, "INVALID_INPUT");
    assert!(no_data_channels(&topo));

    // Missing target entirely.
    let err = call_direct(&topo, serde_json::json!({"substrate": "tcp"}))
        .await
        .expect_err("missing target must fail the schema gate");
    assert_eq!(err.code, "INVALID_INPUT");
    assert!(no_data_channels(&topo));
}

#[tokio::test]
async fn dial_failure_passes_through_typed() {
    // The dial closure's own typed errors pass through unchanged
    // (ADR-007 §4's table minus the registry row).
    let topo = wire_direct(failing_dial("no route to target")).await;

    let err = call_direct(
        &topo,
        serde_json::json!({
            "substrate": "tcp",
            "target": {"host": "blackhole.internal", "port": 9}
        }),
    )
    .await
    .expect_err("failed dial must fail the open");
    assert_eq!(err.code, "channel:open_failed");
    assert_eq!(establishment_reason_of(&err), Some("dial_failed"));
    assert!(no_data_channels(&topo));
}

#[tokio::test]
async fn producer_without_direct_op_rejects_loudly() {
    // An old producer (only the base op registered) rejects the
    // unknown op loudly — NOT_FOUND, the SSH "unknown channel type"
    // posture.
    let registry = alktunnels::producer::ResourceRegistry::new();
    registry
        .register("echo", Substrate::Tcp, "in-process")
        .await;
    let topo = wire(registry, echo_dial()).await;
    assert!(topo
        .producer_registry
        .registration("channels/tunnel/direct")
        .is_none());

    let err = call_direct(
        &topo,
        serde_json::json!({
            "substrate": "tcp",
            "target": {"host": "db.internal", "port": 5432}
        }),
    )
    .await
    .expect_err("unknown op must be rejected loudly");
    assert_eq!(err.code, "NOT_FOUND");
    assert!(no_data_channels(&topo));
}

#[tokio::test]
async fn tunnel_open_scope_does_not_imply_tunnel_direct() {
    // ADR-006 Amendment 1: the scopes are structurally separate
    // grants — an identity holding `tunnel:open` (but not
    // `tunnel:direct`) is FORBIDDEN on the direct op.
    let topo = wire_direct_with_identity(consumer_identity()).await;

    let err = call_direct(
        &topo,
        serde_json::json!({
            "substrate": "tcp",
            "target": {"host": "db.internal", "port": 5432}
        }),
    )
    .await
    .expect_err("tunnel:open must NOT authorize the direct op");
    assert_eq!(err.code, "FORBIDDEN");
    assert!(no_data_channels(&topo));
}

#[tokio::test]
async fn direct_scope_does_not_imply_tunnel_open() {
    // The separation is symmetric: `tunnel:direct` does not authorize
    // the base open op either (two capabilities; a deployment
    // granting both grants both explicitly). A tunnel:direct-only
    // identity is FORBIDDEN on `channels/tunnel/sub` even though the
    // producer registered it.
    let registry = alktunnels::producer::ResourceRegistry::new();
    registry
        .register("echo", Substrate::Tcp, "in-process")
        .await;
    let topo = harness::wire_with(
        registry,
        harness::RegistrationMode::Dial(echo_dial()),
        Some(harness::direct_identity()),
        None,
        Arc::new(alkcall::core::auth::NoopIdentityProvider),
    )
    .await;
    let payload = serde_json::json!({
        "operationId": "channels/tunnel/sub",
        "input": {"resource": "echo", "substrate": "tcp"},
    });
    let response = topo.consumer_call.call_with_payload(payload).await;
    let err = response
        .result
        .expect_err("tunnel:direct must NOT authorize the base op");
    assert_eq!(err.code, "FORBIDDEN");
    assert!(no_data_channels(&topo));
}

#[tokio::test]
async fn direct_establisher_witnesses_per_call_identity() {
    // CF-006: the direct establisher's per-call auth carried the END
    // CALLER's identity (the direct op is the arbitrary-egress
    // capability — identity visibility matters more, not less).
    let topo = wire_direct(echo_dial()).await;

    let out = call_direct(
        &topo,
        serde_json::json!({
            "substrate": "tcp",
            "target": {"host": "db.internal", "port": 5432}
        }),
    )
    .await
    .expect("direct open authorized (both-scopes identity)");
    assert!(out["channel_id"].is_u64());

    let witness = topo.identity_witness.lock().await.clone();
    assert_eq!(
        witness.as_deref(),
        Some("consumer-both"),
        "direct establisher saw the per-call opener identity (CF-006)"
    );
}

#[tokio::test]
async fn establisher_parse_backstop_maps_handler_error() {
    // The establisher's serde parse is the backstop behind the schema
    // gate: feed an input the schema would have rejected but a direct
    // establisher call never schema-checks (the raw-call shape —
    // assembly variants, or a registry whose schema gate is looser)
    // and observe the typed `handler_error` mapping (ADR-007 §4).
    // An unknown substrate value fails the enum parse.
    let establisher = alktunnels::producer::direct_establisher(echo_dial());
    let err = establisher(
        serde_json::json!({
            "substrate": "sctp",
            "target": {"host": "h", "port": 1}
        }),
        alkcall::core::auth::AuthContext::anonymous(b"alk/tunnel"),
    )
    .await;
    match err {
        Err(alkcall::channels::operations::EstablishmentError::HandlerError { message }) => {
            assert!(message.contains("invalid direct params"), "{message}");
        }
        Err(other) => panic!("expected HandlerError backstop, got {other:?}"),
        Ok(_) => panic!("expected HandlerError backstop, got Ok"),
    }
}

#[tokio::test]
async fn direct_establisher_renders_target_to_backing_string() {
    // The design pin: the establisher renders the target to the SAME
    // backing-string form the registry backings use and calls the
    // injected DialFn with it — the dial closure observes the exact
    // rendered form ("host:port" / path), per ADR-007 §4.
    let observed: Arc<tokio::sync::Mutex<Vec<(Substrate, String)>>> =
        Arc::new(tokio::sync::Mutex::new(Vec::new()));
    let obs = Arc::clone(&observed);
    let dial: alktunnels::producer::DialFn =
        Arc::new(move |substrate: Substrate, backing: &str| {
            let obs = Arc::clone(&obs);
            let backing = backing.to_string();
            Box::pin(async move {
                obs.lock().await.push((substrate, backing));
                Err(TunnelEstablishError::DialFailed("stop here".to_string()))
            })
        });
    let establisher = alktunnels::producer::direct_establisher(dial);
    let auth = alkcall::core::auth::AuthContext::anonymous(b"alk/tunnel");

    let _ = establisher(
        serde_json::json!({"substrate": "tcp", "target": {"host": "db.internal", "port": 5432}}),
        auth.clone(),
    )
    .await;
    let _ = establisher(
        serde_json::json!({"substrate": "unix", "target": {"path": "/run/pg/.s.PGSQL.5432"}}),
        auth,
    )
    .await;

    let seen = observed.lock().await.clone();
    assert_eq!(
        seen,
        vec![
            (Substrate::Tcp, "db.internal:5432".to_string()),
            (Substrate::Unix, "/run/pg/.s.PGSQL.5432".to_string()),
        ],
        "targets rendered to the registry backing-string form"
    );
}

// --- consumer session level (TunnelSession::open_direct) ---------------

#[tokio::test]
async fn open_direct_session_round_trip_and_teardown() {
    let topo = wire_direct_forward(echo_dial()).await;

    let mut session = TunnelSession::open_direct(
        &topo.consumer,
        target_hostport("db.internal", 5432),
        Substrate::Tcp,
    )
    .await
    .expect("open_direct");
    assert!(topo.producer_manager.has_channel(session.channel_id));

    use tokio::io::{AsyncReadExt, AsyncWriteExt};
    let (read, write) = session.stream_halves().expect("stream session has halves");
    write.write_all(b"ping-direct").await.expect("write");
    write.flush().await.expect("flush");
    let mut buf = vec![0u8; 11];
    tokio::time::timeout(std::time::Duration::from_secs(5), read.read_exact(&mut buf))
        .await
        .expect("round trip")
        .expect("read");
    assert_eq!(&buf, b"ping-direct");

    // Teardown semantics identical to `open` (ADR-007 §5).
    let channel_id = session.channel_id;
    let reaped = session.close().await;
    assert!(reaped, "close reaped the adopted entry");
    tokio::time::sleep(std::time::Duration::from_millis(100)).await;
    assert!(
        !topo.producer_manager.has_channel(channel_id),
        "producer side reaped after pump completion"
    );
}

#[tokio::test]
async fn open_direct_session_datagram_boundary_preserved() {
    let topo = wire_direct_forward(framed_udp_echo_dial()).await;

    let mut session = TunnelSession::open_direct(
        &topo.consumer,
        target_hostport("10.0.0.7", 53),
        Substrate::Udp,
    )
    .await
    .expect("udp open_direct");
    session.send_datagram(b"one").await.expect("send");
    session.send_datagram(b"").await.expect("send empty");
    let first = tokio::time::timeout(std::time::Duration::from_secs(5), session.recv_datagram())
        .await
        .expect("round trip")
        .expect("recv")
        .expect("datagram");
    assert_eq!(first.as_ref(), b"one");
    let second = tokio::time::timeout(std::time::Duration::from_secs(5), session.recv_datagram())
        .await
        .expect("round trip")
        .expect("recv")
        .expect("datagram");
    assert!(
        second.is_empty(),
        "the empty datagram crosses framed (F-2 layering)"
    );
    session.close().await;
}

#[tokio::test]
async fn open_direct_malformed_target_no_phantom_session() {
    // The session-level failure surface: a malformed target is the
    // schema/invalid_input-class failure (registry-side); a failed
    // open never yields a session.
    let topo = wire_direct_forward(echo_dial()).await;

    let err = match TunnelSession::open_direct(
        &topo.consumer,
        SubstrateAddr::Path {
            path: "/nope".to_string(),
        },
        Substrate::Tcp,
    )
    .await
    {
        Err(e) => e,
        Ok(_) => panic!("path-shaped target on tcp must fail"),
    };
    let call = err.open_ref().call_error().expect("wire error");
    assert_eq!(call.code, "INVALID_INPUT");

    // Both managers hold only channel 0 (the phantom-channel property).
    assert!(
        topo.consumer
            .manager()
            .channel_ids()
            .iter()
            .all(|&id| id == 0)
            && topo
                .producer_manager
                .channel_ids()
                .iter()
                .all(|&id| id == 0),
        "no phantom session"
    );
}

#[tokio::test]
async fn open_direct_scope_separated_session_level() {
    // Session-level scope separation (ADR-006 Amendment 1): a
    // tunnel:open-only consumer client is FORBIDDEN on open_direct —
    // the scopes are separate grants, and the session constructor
    // states which capability it invokes.
    let topo = harness::wire_direct_forward_with_identity(echo_dial(), consumer_identity()).await;

    let err = match TunnelSession::open_direct(
        &topo.consumer,
        target_hostport("db.internal", 5432),
        Substrate::Tcp,
    )
    .await
    {
        Err(e) => e,
        Ok(_) => panic!("tunnel:open-only identity must NOT open_direct"),
    };
    let call = err.open_ref().call_error().expect("wire error");
    assert_eq!(call.code, "FORBIDDEN");
    assert!(
        topo.consumer
            .manager()
            .channel_ids()
            .iter()
            .all(|&id| id == 0)
            && topo
                .producer_manager
                .channel_ids()
                .iter()
                .all(|&id| id == 0),
        "no phantom session"
    );
}

#[tokio::test]
async fn open_direct_witness_carries_session_identity() {
    let topo = wire_direct_forward(echo_dial()).await;
    let _session = TunnelSession::open_direct(
        &topo.consumer,
        target_hostport("db.internal", 5432),
        Substrate::Tcp,
    )
    .await
    .expect("open_direct");
    let witness = topo.identity_witness.lock().await.clone();
    assert_eq!(
        witness.as_deref(),
        Some("consumer-both"),
        "the session path's establisher saw the per-call opener identity"
    );
}