//! The `w2-bounded-resources` contract rows: the bounded-resources
//! policy enforced through the real wire via the loopback harness: the
//! in-crate concurrent-channel cap at every `channel_open_*` accept
//! path, its recovery on close, its `Limits` tunability, and the
//! behavioral verification of the wired `max_auth_attempts` russh knob
//! (the ceiling disconnects an exceeding client). The size/window/
//! buffer knobs (`window_size`, `maximum_packet_size`,
//! `channel_buffer_size`), the inactivity wall clock and the remaining
//! auth-counting posture ride the config-mapping tests
//! (`adapter.rs` `#[cfg(test)]` + the harness's own
//! `client_config_maps_the_limits_postures`); recorded, not
//! re-proven on the wire here.
//!
//! The cap rows are feature-gated to the builds whose opens can be
//! accepted (session channels need `session` or `git`; the
//! tunnel-gate ordering row additionally needs `tunnel`), per the
//! `dispatch_core.rs` feature-build convention.

#![allow(dead_code)]

mod common;

use std::sync::Arc;

use alkcall::core::auth::IdentityProvider;
use common::harness::{
    DUPLEX_CAPACITY, HarnessClientHandler, LoopbackSession, authenticate_publickey, client_config,
    drive, serve_config,
};
use common::{MemHostKeyPolicy, MemIdentityProvider, fixture, identity};
use russh::{ChannelOpenFailure, Error, MethodSet};

use alkssh::{Limits, SshAdapter};

fn registered_provider(fingerprints: &[&str]) -> Arc<MemIdentityProvider> {
    let provider = Arc::new(MemIdentityProvider::new());
    for (index, registered) in fingerprints.iter().enumerate() {
        provider.register(registered, identity(&format!("user-{index}"), &["ssh"]));
    }
    provider
}

/// Mounts an authenticated loopback session over the doubles with the
/// row's [`Limits`], the bounded-resources rows' driver.
async fn authenticated_with(limits: Limits) -> LoopbackSession {
    let provider = registered_provider(&[fixture("alpha").fingerprint]);
    let adapter = SshAdapter::new(serve_config(
        provider as Arc<dyn IdentityProvider>,
        &[fixture("bravo")],
        limits,
    ))
    .await
    .expect("adapter constructs");

    let mut session = drive(
        adapter,
        Arc::new(MemHostKeyPolicy::with_default(true)),
        limits,
    )
    .await
    .expect("the loopback mounts");
    let result = authenticate_publickey(&mut session.client, "anyone", fixture("alpha")).await;
    assert!(
        result.success(),
        "the fixture identity authenticates before every bounded-resources row"
    );
    session
}

/// Matches an open failure's SSH reason, the wire decline reason the
/// cap-gate rows assert.
fn open_failure(err: Error, expected: ChannelOpenFailure, context: &str) {
    assert!(
        matches!(
            &err,
            Error::ChannelOpenFailure(reason) if *reason == expected
        ),
        "{context}: expected open decline with {expected:?}, got {err:?}"
    );
}

/// - *Contract stamp:* `overview.md` §Bounded-resources policy (the
///   auth-attempts row; the `max_auth_attempts` knob is behavioral,
///   verified where cheap: the ceiling disconnects an exceeding
///   client) + ADR-004 decision 6 (the mapping in `adapter.rs` makes
///   the knob the counting ceiling; within the ceiling, failures stay
///   ordinary `AuthResult::Failure` riding the publickey-only
///   advertisement).
///
/// Drive: `max_auth_attempts: 2` (Limits-tunable, not the 10 default),
/// three unregistered keys. The first two rejections stay ordinary
/// failures with the advertisement intact; the third request past the
/// ceiling is answered with the server's
/// `Disconnect::NoMoreAuthMethodsAvailable`. Client-side, the
/// in-flight attempt sees russh's synthesized `Failure` with the
/// **empty** remaining-methods set (no reply ever came; the
/// advertisement is gone). The session ends cleanly server-side.
#[tokio::test]
async fn auth_attempt_ceiling_disconnects_an_exceeding_client() {
    use russh::MethodKind;
    use russh::keys::PrivateKeyWithHashAlg;

    let provider = registered_provider(&[fixture("alpha").fingerprint]);
    let limits = Limits {
        max_auth_attempts: 2,
        ..Limits::default()
    };
    let adapter = SshAdapter::new(serve_config(
        provider as Arc<dyn IdentityProvider>,
        &[fixture("bravo")],
        limits,
    ))
    .await
    .expect("adapter constructs");

    let (client_io, server_io) = tokio::io::duplex(DUPLEX_CAPACITY);
    let server_task = tokio::spawn(async move { adapter.handle(server_io).await });
    let mut client = russh::client::connect_stream(
        client_config(limits),
        client_io,
        HarnessClientHandler::new(Arc::new(MemHostKeyPolicy::with_default(true))),
    )
    .await
    .expect("the in-process pair connects");

    let normal_advertisement = MethodSet::from(&[MethodKind::PublicKey][..]);
    let within = authenticate_publickey(&mut client, "anyone", fixture("delta")).await;
    assert!(
        !within.success(),
        "within the ceiling, a rejection stays an ordinary failure: {within:?}"
    );
    assert_eq!(
        remaining_ads(&within),
        Some(normal_advertisement.clone()),
        "the within-ceiling failure still carries the advertisement"
    );
    let last_within = authenticate_publickey(&mut client, "anyone", fixture("charlie")).await;
    assert!(
        !last_within.success(),
        "the ceiling's last rejection still rides as an ordinary failure: {last_within:?}"
    );
    assert_eq!(
        remaining_ads(&last_within),
        Some(normal_advertisement),
        "the advertisement survives every within-ceiling rejection"
    );

    let exceeding = async {
        let key = Arc::new(fixture("bravo").private_key());
        client
            .authenticate_publickey("anyone", PrivateKeyWithHashAlg::new(key, None))
            .await
    }
    .await;
    match &exceeding {
        Ok(result) => {
            assert!(
                !result.success(),
                "the attempt past the ceiling never re-answers auth: {result:?}"
            );
            assert_eq!(
                remaining_ads(result),
                Some(MethodSet::empty()),
                "the past-ceiling answer is the torn-down shape: no advertisement rides"
            );
        }
        Err(_) => {
            // The client loop already exited on the disconnect, the
            // equally-valid torn-down shape (the send side is gone).
        }
    }

    drop(client);
    server_task
        .await
        .expect("server task ends")
        .expect("the ceiling disconnect ends the session cleanly");
}

/// The remaining-methods set of an [`AuthResult`]; `None` on success
/// (the within-ceiling vs past-ceiling observation shape).
fn remaining_ads(result: &russh::client::AuthResult) -> Option<russh::MethodSet> {
    match result {
        russh::client::AuthResult::Failure {
            remaining_methods, ..
        } => Some(remaining_methods.clone()),
        _ => None,
    }
}

#[cfg(any(feature = "session", feature = "git"))]
#[tokio::test]
async fn concurrent_channel_cap_declines_and_recovers_over_the_wire() {
    use common::harness::open_session_channel;

    let session = authenticated_with(Limits {
        max_concurrent_channels: 2,
        ..Limits::default()
    })
    .await;

    let first = open_session_channel(&session.client)
        .await
        .expect("the first open admits under the cap");
    let _second = open_session_channel(&session.client)
        .await
        .expect("the second open admits under the cap");

    let err = open_session_channel(&session.client)
        .await
        .expect_err("the third open is refused at the cap, never queued");
    open_failure(
        err,
        ChannelOpenFailure::AdministrativelyProhibited,
        "the cap-exceeded open",
    );

    first.close().await.expect("the close rides the wire");

    // The close reply is not observable client-side at this pin: russh
    // removes a locally-closed channel from its map before the
    // server's CHANNEL_CLOSE arrives back; no message, and no
    // `None`, is ever delivered (the harness's `await_close` only
    // observes *server-initiated* closes). The recovery observation is
    // the open itself, and it is deterministic: close and open ride
    // the client loop's single FIFO event queue, they are written in
    // that order on the wire, and the server's packet loop processes
    // them in order; the counted close releases before the recovery
    // open is admitted. Had the release not happened, this open
    // declines and this expect fails.
    let _third = open_session_channel(&session.client)
        .await
        .expect("a close recovers the cap; the next open succeeds again");

    session.unmount().await;
}

#[cfg(any(feature = "session", feature = "git"))]
#[tokio::test]
async fn zero_cap_refuses_every_open_and_poisons_nothing() {
    use common::harness::open_session_channel;

    let session = authenticated_with(Limits {
        max_concurrent_channels: 0,
        ..Limits::default()
    })
    .await;

    for attempt in 1..=2 {
        let err = open_session_channel(&session.client)
            .await
            .expect_err("a zero cap refuses the very first open: no half-open, no queue");
        open_failure(
            err,
            ChannelOpenFailure::AdministrativelyProhibited,
            &format!("the zero-cap open #{attempt}"),
        );
    }

    session.unmount().await;
}

/// - *Contract stamp:* `w3-tunnel-direct`'s recorded admission
///   mechanics (decline-first-alloc-never applies to the cap budget
///   too): the tunnel row's pre-admit classes (here the **absent
///   fabric handle** under the feature) answer before the admission
///   is taken, so under an exhausted cap the absent-handle class
///   (`ConnectFailed`, the cannot-bridge wire class) wins over the cap
///   reason. The cap class on a *configured* direct row rides
///   tests/tunnel_mapping.rs's fabric double.
#[cfg(all(feature = "tunnel", any(feature = "session", feature = "git")))]
#[tokio::test]
async fn the_tunnel_rows_pre_admit_class_answers_before_the_cap_gate() {
    use common::harness::{open_direct_tcpip, open_session_channel};

    let session = authenticated_with(Limits {
        max_concurrent_channels: 1,
        ..Limits::default()
    })
    .await;

    let _first = open_session_channel(&session.client)
        .await
        .expect("the first open admits, so the cap is now exhausted");

    let err = open_direct_tcpip(&session.client, "internal.example", 5432)
        .await
        .expect_err("the pre-admit class answers before the cap gate is even consulted");
    open_failure(
        err,
        ChannelOpenFailure::ConnectFailed,
        "the absent-handle decline precedes the admission",
    );

    session.unmount().await;
}