//! The dial surface's contract rows (`w4-dial`), driven through the
//! in-process fake `sshd` ([`sshd_double`](common::sshd_double), the
//! client-direction vehicle) plus the crate-to-crate smoke against
//! the alkssh server direction. The row/stamp discipline
//! (`tests/common/mod.rs`'s doc) applies verbatim.

mod common;

use std::sync::Arc;
use std::time::Duration;

use alkssh::{
    DialConfig, DialError, HostKeyPolicy, KeySource, KeySourceError, Limits, RejectAll,
    ServeConfig, SshAdapter, VaultKeySource, dial, fingerprint, russh_client_config,
};
use common::{
    MemHostKeyPolicy, MemHostKeySource, MemIdentityProvider, TestVault, event_capture, fixture,
    identity, sshd_double,
};
use russh::{ChannelOpenFailure, Error};

use sshd_double::SshdDouble;

/// A vault-backed key source over the alpha fixture, the standard
/// custody posture (the fixture PEM sealed, decrypt-at-dial behind
/// the production source).
fn alpha_key_source() -> Arc<dyn KeySource> {
    let vault = TestVault::unlocked();
    let blob = vault.seal(fixture("alpha"));
    Arc::new(VaultKeySource::new(vault.handle(), Some(blob)))
}

/// The dial-target address of a new double; one fake `sshd` per row
/// (the vehicle's shape decision).
async fn spawn_double() -> (SshdDouble, sshd_double::SshdAccepts) {
    sshd_double::spawn(vec![fixture("bravo").private_key()], true).await
}

fn port_of(double: &SshdDouble) -> u16 {
    double
        .address
        .rsplit_once(':')
        .and_then(|(_, port)| port.parse().ok())
        .expect("the double's address carries its port")
}

/// The ADR-005 failure postures, one vault-backed source each.
/// `127.0.0.1:9` is deliberately unbound in every no-dial row's dial
/// (see the row's doc).
fn postured_key_source(posture: &'static str) -> (Arc<dyn KeySource>, KeySourceError) {
    let vault = TestVault::unlocked();
    let alpha = fixture("alpha");
    match posture {
        "locked" => {
            let blob = vault.seal(alpha);
            vault.lock();
            (
                Arc::new(VaultKeySource::new(vault.handle(), Some(blob))),
                KeySourceError::Locked,
            )
        }
        "absent" => (
            Arc::new(VaultKeySource::new(vault.handle(), None)),
            KeySourceError::Absent,
        ),
        "decryption" => {
            let mut blob = vault.seal(alpha);
            blob.data = "not the ciphertext".into();
            (
                Arc::new(VaultKeySource::new(vault.handle(), Some(blob))),
                KeySourceError::Decryption(String::new()),
            )
        }
        "unknown-version" => {
            let mut blob = vault.seal(alpha);
            blob.key_version = 0;
            (
                Arc::new(VaultKeySource::new(vault.handle(), Some(blob))),
                KeySourceError::UnknownKeyVersion(String::new()),
            )
        }
        "decode" => {
            let blob = vault.encrypt("this is not a key");
            (
                Arc::new(VaultKeySource::new(vault.handle(), Some(blob))),
                KeySourceError::Decode(String::new()),
            )
        }
        other => panic!("no posture named {other}"),
    }
}

/// - *Contract stamp:* `client.md` §Lifecycle primitives (the
///   explicit-vs-drop boundary, the OQ-12 primitives-only posture):
///   **drop is never the clean path**. Dropping the (feature-free,
///   last) wire-stack handle ends the runner and the transport goes
///   down abruptly; the double's serving outcome is the error wire
///   class, never the clean session end only `close()` produces (see
///   `dial_success_lands_an_authenticated_connection`'s `Ok`).
#[tokio::test]
async fn dropping_the_connection_is_an_abrupt_teardown_close_is_the_only_clean_path() {
    let (mut double, accepts) = spawn_double().await;
    accepts.register(fixture("alpha").fingerprint);

    let policy = Arc::new(MemHostKeyPolicy::new());
    policy.pin("127.0.0.1", fixture("bravo").fingerprint);
    let config = DialConfig::new(
        double.address.clone(),
        "runner",
        alpha_key_source(),
        Arc::clone(&policy) as Arc<dyn HostKeyPolicy>,
    );
    let connection = dial(config).await.expect("the dial succeeds");

    drop(connection);

    let outcome = double.next_session_outcome().await;
    assert!(
        matches!(outcome, Some(Err(_))),
        "the un-closed drop tears the peer's session down abruptly; \
         never the clean `Ok`, got {outcome:?}"
    );
    double.shutdown();
}

/// - *Contract stamp:* the dial surface's success row: decrypt-at-dial
///   → the policy accept → the single publickey attempt: an
///   authenticated connection, the dial data carried on its surface,
///   and the clean close the peer observes.
#[tokio::test]
async fn dial_success_lands_an_authenticated_connection() {
    let (mut double, accepts) = spawn_double().await;
    accepts.register(fixture("alpha").fingerprint);

    let policy = Arc::new(MemHostKeyPolicy::new());
    policy.pin("127.0.0.1", fixture("bravo").fingerprint);

    let config = DialConfig::new(
        double.address.clone(),
        "runner",
        alpha_key_source(),
        Arc::clone(&policy) as Arc<dyn HostKeyPolicy>,
    );
    let connection = dial(config)
        .await
        .expect("the dial succeeds through the fake sshd");

    assert_eq!(connection.address(), double.address);
    assert_eq!(connection.username(), "runner");
    assert_eq!(connection.limits(), &Limits::default());
    assert!(!connection.is_closed(), "the connection is live");

    let decisions = policy.decisions();
    assert_eq!(decisions.len(), 1, "one policy decision rode the dial");
    assert!(decisions[0].accepted, "the pinned pair accepted");
    assert_eq!(decisions[0].presented.host, "127.0.0.1");
    assert_eq!(decisions[0].presented.port, port_of(&double));
    assert_eq!(
        decisions[0].presented.fingerprint,
        fixture("bravo").fingerprint,
        "the presented fingerprint is the crate's single fingerprint path value"
    );

    connection.close().await;
    let outcome = double.next_session_outcome().await;
    assert!(
        matches!(outcome, Some(Ok(()))),
        "the clean close is the double's clean session end, got {outcome:?}"
    );
    double.shutdown();
}

/// `127.0.0.1:9` is deliberately unbound in the no-dial rows: a dial
/// that reached the socket would fail with a wire class; only a
/// refused key load can produce the key-source classes here.
///
/// - *Contract stamp:* ADR-005 decision 2 (decrypt-at-*dial*) +
///   decision 4 (every failure mode fails closed, distinct shapes):
///   every [`KeySourceError`] class is a no-dial outcome, each variant
///   preserved through [`DialError::KeySource`].
#[tokio::test]
async fn every_key_source_posture_is_a_no_dial_outcome() {
    for posture in [
        "locked",
        "absent",
        "decryption",
        "unknown-version",
        "decode",
    ] {
        let (source, expected) = postured_key_source(posture);
        let config = DialConfig::new("127.0.0.1:9", "runner", source, Arc::new(RejectAll));
        let err = dial(config)
            .await
            .expect_err("a refused key source is a no-dial outcome");
        assert!(
            matches!(err, DialError::KeySource(ref e) if std::mem::discriminant(e) == std::mem::discriminant(&expected)),
            "posture {posture}: expected {expected:?} class, got {err:?}"
        );
    }
}

/// - *Contract stamp:* OQ-05 resolution (`host_key.rs`): the crate's
///   built-in default policy connects to nothing; the dial refuses
///   as the decline class (`DialError::HostKeyRejected`), never a
///   wire crash class, never a fallback.
#[tokio::test]
async fn the_crate_default_policy_connects_to_nothing() {
    let (double, _accepts) = spawn_double().await;

    let config = DialConfig::new(
        double.address.clone(),
        "runner",
        alpha_key_source(),
        Arc::new(RejectAll),
    );
    let err = dial(config)
        .await
        .expect_err("the default policy declines every presented key");
    assert!(
        matches!(err, DialError::HostKeyRejected),
        "expected HostKeyRejected, got {err:?}"
    );
    double.shutdown();
}

/// - *Contract stamp:* OQ-05 resolution, case 1 (known beforehand):
///   an assembly pin **hard-declines on mismatch** (the dial aborts);
///   the policy's own log carries the context (the presented key's
///   real fingerprint rode the policy; the crate never second-guesses).
#[tokio::test]
async fn pin_mismatch_hard_declines() {
    let (double, _accepts) = spawn_double().await;

    let policy = Arc::new(MemHostKeyPolicy::new());
    policy.pin("127.0.0.1", fixture("charlie").fingerprint);

    let config = DialConfig::new(
        double.address.clone(),
        "runner",
        alpha_key_source(),
        Arc::clone(&policy) as Arc<dyn HostKeyPolicy>,
    );
    let err = dial(config)
        .await
        .expect_err("a mismatched pin is a hard decline");
    assert!(matches!(err, DialError::HostKeyRejected), "got {err:?}");

    let decisions = policy.decisions();
    assert_eq!(decisions.len(), 1);
    assert!(!decisions[0].accepted);
    assert_eq!(
        decisions[0].presented.fingerprint,
        fixture("bravo").fingerprint,
        "the presented key (not the pin) is what the policy saw"
    );
    double.shutdown();
}

/// - *Contract stamp:* OQ-05 resolution, case 2 (unknown beforehand):
///   a permissive assembly policy decides and the dial proceeds; the
///   crate ships no accept-all of its own (the permissive posture
///   was this row's explicit choice, `MemHostKeyPolicy::
///   with_default(true)`).
#[tokio::test]
async fn case2_policy_decides_and_the_dial_proceeds() {
    let (mut double, accepts) = spawn_double().await;
    accepts.register(fixture("alpha").fingerprint);

    let config = DialConfig::new(
        double.address.clone(),
        "runner",
        alpha_key_source(),
        Arc::new(MemHostKeyPolicy::with_default(true)),
    );
    let connection = dial(config)
        .await
        .expect("the explicit permissive policy accepts");
    assert!(!connection.is_closed());

    connection.close().await;
    let outcome = double.next_session_outcome().await;
    assert!(matches!(outcome, Some(Ok(()))), "clean session end");
    double.shutdown();
}

/// - *Contract stamp:* the publickey-only family contract: the
///   remote end refusing the single attempt is the dial's
///   [`DialError::AuthRejected`]: no retry, no fallback method, the
///   attempt closed behind the refusal.
#[tokio::test]
async fn refused_authentication_is_the_dials_no_connection_outcome() {
    let (double, _accepts) = spawn_double().await;

    let config = DialConfig::new(
        double.address.clone(),
        "runner",
        alpha_key_source(),
        Arc::new(MemHostKeyPolicy::with_default(true)),
    );
    let err = dial(config)
        .await
        .expect_err("the unregistered key is refused by the remote end");
    assert!(matches!(err, DialError::AuthRejected), "got {err:?}");
    double.shutdown();
}

/// - *Contract stamp:* the no-construction-validation contract, the
///   dial-time side: a malformed address refuses *at the dial* (the
///   config carries no validation; the wire is never dialed).
#[tokio::test]
async fn malformed_address_refuses_at_the_dial() {
    let config = DialConfig::new(
        "runner.example",
        "runner",
        alpha_key_source(),
        Arc::new(MemHostKeyPolicy::with_default(true)),
    );
    let err = dial(config).await.expect_err("refuses");
    assert!(
        matches!(err, DialError::Address(_)),
        "expected Address, got {err:?}"
    );
}

/// - *Contract stamp:* the bounded-config wiring (the task's
///   acceptance): the dial's russh `Config` carries the
///   [`Limits`]-mapped knobs (the keepalive 15 s / 3 posture, the
///   always-set wall clock, the window/packet/buffer knobs verbatim)
///   observed from outside the crate through the doc-hidden mapping.
#[test]
fn the_bounded_config_wiring_is_observable() {
    let config = russh_client_config(&Limits::default());

    assert_eq!(config.keepalive_interval, Some(Duration::from_secs(15)));
    assert_eq!(config.keepalive_max, 3);
    assert_eq!(config.inactivity_timeout, Some(Duration::from_secs(600)));
    assert_eq!(config.window_size, 2_097_152);
    assert_eq!(config.maximum_packet_size, 32_768);
    assert_eq!(config.channel_buffer_size, 100);
}

/// - *Contract stamp:* ADR-003 §4 (the client-side fail-closed
///   acceptor): the dialed client's skeleton declines **every**
///   server-initiated open on the wire (the double's initiator
///   observes each open as an unconfirmed refusal, never russh's
///   accept-and-drop default; the never-feature rows (x11, agent)
///   decline too; the declines consulted no host-key policy; and the
///   connection lives through all of it (the clean close is the
///   double's `Ok` session end).
///
///   (`should_accept_unknown_server_channel` has no wire row: russh's
///   server surface exposes no initiator for an unknown type and the
///   `ChannelId` is deliberately unconstructible outside the library
///   while the skeleton's explicit `false` is code-pinned, not
///   row-pinned; recorded in the task.)
#[tokio::test]
async fn server_initiated_opens_decline_cleanly_on_the_wire() {
    event_capture::install();
    let (mut double, accepts) = spawn_double().await;
    accepts.register(fixture("alpha").fingerprint);

    let policy = Arc::new(MemHostKeyPolicy::new());
    policy.pin("127.0.0.1", fixture("bravo").fingerprint);
    let config = DialConfig::new(
        double.address.clone(),
        "runner",
        alpha_key_source(),
        Arc::clone(&policy) as Arc<dyn HostKeyPolicy>,
    );
    let connection = dial(config).await.expect("the dial succeeds");

    let server = double.session_handle().await;
    for outcome in [
        server.channel_open_session().await,
        server
            .channel_open_direct_tcpip("hub.example", 4096, "originator", 0)
            .await,
        server
            .channel_open_forwarded_tcpip("hub.example", 4096, "originator", 0)
            .await,
        server
            .channel_open_forwarded_streamlocal("/run/hub.sock")
            .await,
        server.channel_open_x11("originator", 0).await,
        server.channel_open_agent().await,
    ] {
        match outcome {
            Err(Error::ChannelOpenFailure(reason)) => {
                assert_eq!(
                    reason,
                    ChannelOpenFailure::AdministrativelyProhibited,
                    "the skeleton's clean decline reason"
                );
            }
            other => panic!("every server-initiated open declines cleanly, got {other:?}"),
        }
    }
    assert_eq!(
        policy.decisions().len(),
        1,
        "the declines consulted no host-key policy: one decision, the dial's"
    );

    let forwarded = event_capture::await_event(&[
        ("row", "forwarded-tcpip"),
        ("decline_class", "open"),
        ("reason", "no-acceptor"),
    ])
    .await;
    assert_eq!(
        forwarded.field("message"),
        Some("decline: channel open rejected")
    );
    assert!(forwarded.field("channel_id").is_some());
    assert_eq!(forwarded.level, "DEBUG");
    assert_eq!(
        forwarded.field("feature_build"),
        Some(event_capture::expected_build())
    );

    let never = event_capture::await_event(&[
        ("row", "x11"),
        ("decline_class", "open"),
        ("reason", "never-feature"),
    ])
    .await;
    assert_eq!(
        never.field("message"),
        Some("decline: channel open rejected")
    );

    connection.close().await;
    let outcome = double.next_session_outcome().await;
    assert!(
        matches!(outcome, Some(Ok(()))),
        "the connection stayed usable through the declines: the clean close is a clean session end"
    );
    double.shutdown();
}

/// The smoke posture (`w4-dial`'s decided + recorded item): the
/// alkssh client ↔ alkssh server in-process loopback over a real TCP
/// socket, the cheap real-stack smoke (both russh stacks, the full
/// auth bridge server-side, the full custody + policy chain
/// client-side). A real-`sshd` dial is wave-5 e2e territory.
///
/// - *Contract stamp:* the task's smoke item; `client.md` §The dial
///   surface (the family-runner end: the dial's fingerprint resolves
///   through the injected `IdentityProvider` server-side).
#[tokio::test]
async fn smoke_alkssh_client_dials_the_alkssh_server_direction() {
    let provider = Arc::new(MemIdentityProvider::new());
    provider.register(
        &fingerprint(&fixture("alpha").public_key()),
        identity("smoke-runner", &["relay:connect"]),
    );
    let serve = ServeConfig::new(
        provider,
        Arc::new(MemHostKeySource::from_fixtures(&[fixture("bravo")])),
    );
    let adapter = SshAdapter::new(serve).await.expect("the server constructs");

    let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
        .await
        .expect("the smoke's loopback listener binds");
    let addr = listener.local_addr().expect("the smoke has an address");
    let (outcome_tx, mut outcome_rx) = tokio::sync::mpsc::unbounded_channel();
    tokio::spawn(async move {
        loop {
            let (socket, _) = match listener.accept().await {
                Ok(accepted) => accepted,
                Err(_) => break,
            };
            // The smoke has exactly one connection; the handle call
            // runs it to completion inline (the accept loop makes no
            // next accept until the session ends).
            let _ = outcome_tx.send(adapter.handle(socket).await);
        }
    });

    let policy = Arc::new(MemHostKeyPolicy::new());
    policy.pin("127.0.0.1", fixture("bravo").fingerprint);
    let config = DialConfig::new(
        addr.to_string(),
        "runner",
        alpha_key_source(),
        Arc::clone(&policy) as Arc<dyn HostKeyPolicy>,
    );
    let connection = dial(config)
        .await
        .expect("the alkssh client connects to the alkssh server");
    assert_eq!(policy.decisions().len(), 1);
    assert_eq!(
        policy.decisions()[0].presented.fingerprint,
        fixture("bravo").fingerprint
    );
    assert!(!connection.is_closed());

    connection.close().await;
    let deadline = tokio::time::Instant::now() + Duration::from_secs(10);
    let outcome = tokio::time::timeout_at(deadline, outcome_rx.recv()).await;
    match outcome {
        Ok(Some(Ok(()))) => {}
        Ok(Some(Err(err))) => panic!("the smoke's server session must end clean, got {err:?}"),
        Ok(None) | Err(_) => panic!("the smoke's server session never ended"),
    }
}