//! The server-direction mount point (ADR-001) — [`SshAdapter`], the
//! constructed russh server stack every mapping composes into.
//!
//! Construction ([`SshAdapter::new`], ADR-005 decision 6): the
//! injected [`HostKeySource`] loads **decrypt-at-construction** — a
//! locked / absent / decryption-failed / unknown-version / undecodable
//! load means **no server is constructed**: the error surfaces as
//! [`AdapterError::Construction`] with the original [`KeySourceError`]
//! variant preserved, never a half-configured adapter, never a keyless
//! server that fails later. An empty key set is the same refusal (the
//! host-key source contract's [`KeySourceError::Absent`] shape).
//!
//! The russh server `Config` (ADR-002 surface notes; ADR-004 decision
//! 4): `methods` is `{publickey}` **exactly** (honest advertisement —
//! `MethodSet::from(&[MethodKind::PublicKey][..])`, never russh's
//! `server_supported()`), the loaded host keys ride `Config::keys`,
//! and the [`Limits`] knobs map one-for-one:
//! `window_size` / `maximum_packet_size` / `channel_buffer_size` /
//! `max_auth_attempts` verbatim, `inactivity_timeout` as **always
//! `Some`** — the bounded-resources posture (the crate never
//! constructs a session with no wall clock). Everything else stays at
//! russh 0.65's own default posture (`Config::default()`, left
//! un-touched deliberately): the rekey `Limits` stay russh-enforced at
//! their RFC-4253 §9 defaults, `auth_rejection_time` stays russh's
//! constant-rejection knob, and the preferred-algorithm set stays
//! russh's — the ed25519-only key-algorithm posture holds by the
//! loaded key set itself (ADR-002 decision 4) and by the auth bridge's
//! explicit non-ed25519 decline (`w2-auth-bridge`).
//!
//! The mount ([`SshAdapter::handle`], ADR-001's contract): a generic
//! byte-duplex stream in, the full russh server stack per stream via
//! [`russh::server::run_stream`], returned **run to completion** —
//! the edge-termination shape only, no ALPN code, no relay leg (the
//! `alk/ssh` pin is a ledger entry; nothing here speaks it). The
//! in-flight framing is the alkhttp door precedent: `handle` is an
//! async per-stream call whose completion is the session's end, so the
//! assembly's accept loop mounts per stream via `tokio::spawn` — the
//! serving loop itself stays spawned inside russh (ADR-006 decision 4,
//! which `run_stream` already satisfies: the session's packet loop
//! leaves on its own task), and the mount's return type stays
//! `Result<(), AdapterError>` — no russh type rides the signature.

use std::sync::Arc;

use russh::MethodKind;
use russh::MethodSet;
use russh::keys::PrivateKey;
use tokio::io::{AsyncRead, AsyncWrite};

use crate::config::ServeConfig;
use crate::error::{AdapterError, KeySourceError};
use crate::handler::SessionHandler;
#[cfg(feature = "tunnel")]
use crate::tunnel_map;

/// The constructed server stack — the substrate every channel-type
/// mapping composes into (ADR-001's `SshAdapter`).
///
/// Built by [`SshAdapter::new`] from the assembly-supplied
/// [`ServeConfig`] (a fail-closed load: see the module docs); mounted
/// per stream by [`SshAdapter::handle`]. Construction is async: the
/// host-key decrypt participates there and nowhere else.
#[derive(Debug)]
pub struct SshAdapter {
    serve: Arc<ServeConfig>,
    config: Arc<russh::server::Config>,
}

impl SshAdapter {
    /// Constructs the adapter — decrypt-at-construction of the host
    /// key set (ADR-005 decision 6): a failed or empty load is no
    /// server, with the [`KeySourceError`] refused through
    /// [`AdapterError::Construction`].
    pub async fn new(serve: ServeConfig) -> Result<Self, AdapterError> {
        let serve = {
            #[cfg(not(feature = "tunnel"))]
            {
                serve
            }
            #[cfg(feature = "tunnel")]
            {
                let mut serve = serve;
                if serve.channel_client.is_some() && serve.tunnel_listen_seam.is_some() {
                    tunnel_map::install_forwarded_surface(&mut serve)
                        .map_err(AdapterError::ListenRegistration)?;
                }
                serve
            }
        };
        let keys = serve.host_key_source.load().await?;
        if keys.is_empty() {
            return Err(AdapterError::Construction(KeySourceError::Absent));
        }
        let config = build_server_config(&serve, keys);
        Ok(Self {
            serve: Arc::new(serve),
            config: Arc::new(config),
        })
    }

    /// The mounted russh server `Config` — in-crate inspection for the
    /// contract tests (advertisement and knob mapping); not public
    /// surface: the mount's behavior is observable on the wire, never
    /// through a russh type leak.
    #[cfg(test)]
    pub(crate) fn config(&self) -> &russh::server::Config {
        &self.config
    }

    /// The mount (ADR-001): mounts the full russh server stack on one
    /// generic duplex byte stream and runs it to completion — an
    /// accepted vanilla-SSH transport is authenticated and served on
    /// this call's task; the assembly's accept loop mounts per stream
    /// with `tokio::spawn(adapter.handle(stream))`.
    ///
    /// The stream is owned edge termination only: no ALPN, no relay
    /// leg, no framing interpretation upstream of russh.
    pub async fn handle<R>(&self, stream: R) -> Result<(), AdapterError>
    where
        R: AsyncRead + AsyncWrite + Unpin + Send + 'static,
    {
        let handler = SessionHandler::new(Arc::clone(&self.serve));
        russh::server::run_stream(Arc::clone(&self.config), stream, handler)
            .await?
            .await
    }
}

/// The adapter's own `Config` construction from the serve-config — the
/// documented subset mapped one-for-one, everything else at russh's
/// default posture (see the module docs; ADR-002 surface notes).
fn build_server_config(serve: &ServeConfig, keys: Vec<PrivateKey>) -> russh::server::Config {
    let limits = serve.limits;
    russh::server::Config {
        methods: MethodSet::from(&[MethodKind::PublicKey][..]),
        keys,
        window_size: limits.window_size,
        maximum_packet_size: limits.maximum_packet_size,
        channel_buffer_size: limits.channel_buffer_size,
        max_auth_attempts: limits.max_auth_attempts,
        inactivity_timeout: Some(limits.inactivity_timeout),
        ..russh::server::Config::default()
    }
}

#[cfg(test)]
mod tests {
    use std::sync::Arc;
    use std::time::Duration;

    use alkvault::CURRENT_KEY_VERSION;
    use russh::client::AuthResult;
    use russh::keys::decode_secret_key;

    use super::{SshAdapter, build_server_config};
    use crate::config::{Limits, ServeConfig};
    use crate::error::{AdapterError, KeySourceError};
    use crate::key_source::{HostKeySource, VaultHostKeySource};

    const NOOP: alkcall::core::auth::NoopIdentityProvider =
        alkcall::core::auth::NoopIdentityProvider;

    /// The committed test-key fixtures — the wave-1 fixture set, read
    /// through `include_str!` per the wave-1 convention (the
    /// `tests/common` double module is the tests/-side owner; src-unit
    /// tests read the fixture files, never its exports).
    const ALPHA_PEM: &str = include_str!("../tests/vectors/alpha.key");
    const BRAVO_PEM: &str = include_str!("../tests/vectors/bravo.key");
    const CHARLIE_PEM: &str = include_str!("../tests/vectors/charlie.key");

    fn unlocked_vault() -> alkvault::VaultServiceHandle {
        let vault = alkvault::VaultServiceHandle::new();
        vault.unlock_new(24).expect("in-memory vault unlocks");
        vault
    }

    fn sealed(vault: &alkvault::VaultServiceHandle, pem: &str) -> alkvault::EncryptedData {
        vault
            .encrypt(pem, CURRENT_KEY_VERSION)
            .expect("seals the test PEM")
    }

    fn serve_config(host_key_source: Arc<dyn HostKeySource>, limits: Limits) -> ServeConfig {
        let mut serve = ServeConfig::new(Arc::new(NOOP), host_key_source);
        serve.limits = limits;
        serve
    }

    fn static_fixture_host_keys(pems: &[&str]) -> Arc<dyn HostKeySource> {
        let keys = pems
            .iter()
            .map(|pem| decode_secret_key(pem, None).expect("fixture PEM decodes"))
            .collect::<Vec<_>>();
        struct Static {
            keys: Vec<russh::keys::PrivateKey>,
        }
        #[async_trait::async_trait]
        impl HostKeySource for Static {
            async fn load(&self) -> Result<Vec<russh::keys::PrivateKey>, KeySourceError> {
                Ok(self.keys.clone())
            }
        }
        Arc::new(Static { keys })
    }

    #[tokio::test]
    async fn methods_advertise_exactly_publickey() {
        let serve = serve_config(static_fixture_host_keys(&[BRAVO_PEM]), Limits::default());
        let adapter = SshAdapter::new(serve).await.expect("constructs");

        let advertised = &adapter.config().methods;
        assert_eq!(advertised.len(), 1);
        assert_eq!(advertised.first(), Some(&russh::MethodKind::PublicKey));
        assert_eq!(
            advertised,
            &russh::MethodSet::from(&[russh::MethodKind::PublicKey][..])
        );
        assert_ne!(advertised, &russh::MethodSet::server_supported());
    }

    #[tokio::test]
    async fn limits_knobs_map_one_for_one_with_inactivity_always_set() {
        let limits = Limits {
            window_size: 1_024,
            maximum_packet_size: 4_096,
            channel_buffer_size: 7,
            max_auth_attempts: 3,
            inactivity_timeout: Duration::from_secs(1_337),
            ..Limits::default()
        };
        let serve = serve_config(static_fixture_host_keys(&[BRAVO_PEM]), limits);

        let adapter = SshAdapter::new(serve).await.expect("constructs");
        let config = adapter.config();

        assert_eq!(config.window_size, 1_024);
        assert_eq!(config.maximum_packet_size, 4_096);
        assert_eq!(config.channel_buffer_size, 7);
        assert_eq!(config.max_auth_attempts, 3);
        assert_eq!(config.inactivity_timeout, Some(Duration::from_secs(1_337)));
    }

    #[tokio::test]
    async fn default_limits_map_to_the_documented_postures() {
        let serve = serve_config(static_fixture_host_keys(&[BRAVO_PEM]), Limits::default());
        let adapter = SshAdapter::new(serve).await.expect("constructs");
        let config = adapter.config();

        assert_eq!(config.window_size, 2_097_152);
        assert_eq!(config.maximum_packet_size, 32_768);
        assert_eq!(config.channel_buffer_size, 100);
        assert_eq!(config.max_auth_attempts, 10);
        assert_eq!(
            config.inactivity_timeout,
            Some(crate::config::DEFAULT_INACTIVITY_TIMEOUT)
        );
    }

    #[tokio::test]
    async fn host_keys_load_decrypt_at_construction_into_the_russh_key_set() {
        let vault = unlocked_vault();
        let blobs = vec![sealed(&vault, BRAVO_PEM), sealed(&vault, CHARLIE_PEM)];
        let serve = serve_config(
            Arc::new(VaultHostKeySource::new(vault, blobs)),
            Limits::default(),
        );

        let adapter = SshAdapter::new(serve).await.expect("constructs");
        let config = adapter.config();

        let expected_bravo = decode_secret_key(BRAVO_PEM, None).expect("decodes");
        let expected_charlie = decode_secret_key(CHARLIE_PEM, None).expect("decodes");

        assert_eq!(config.keys.len(), 2);
        assert_eq!(
            crate::fingerprint::fingerprint(config.keys[0].public_key()),
            crate::fingerprint::fingerprint(expected_bravo.public_key())
        );
        assert_eq!(
            crate::fingerprint::fingerprint(config.keys[1].public_key()),
            crate::fingerprint::fingerprint(expected_charlie.public_key())
        );
    }

    #[tokio::test]
    async fn locked_vault_at_construction_is_no_adapter() {
        let vault = unlocked_vault();
        let blobs = vec![sealed(&vault, BRAVO_PEM)];
        vault.lock();

        let serve = serve_config(
            Arc::new(VaultHostKeySource::new(vault, blobs)),
            Limits::default(),
        );
        let err = SshAdapter::new(serve)
            .await
            .expect_err("locked vault refuses construction");
        assert!(
            matches!(err, AdapterError::Construction(KeySourceError::Locked)),
            "expected Construction(Locked), got {err:?}"
        );
    }

    #[tokio::test]
    async fn absent_host_keys_at_construction_is_no_adapter() {
        let vault = unlocked_vault();
        let serve = serve_config(
            Arc::new(VaultHostKeySource::new(vault, Vec::new())),
            Limits::default(),
        );

        let err = SshAdapter::new(serve)
            .await
            .expect_err("no host keys refuse construction");
        assert!(
            matches!(err, AdapterError::Construction(KeySourceError::Absent)),
            "expected Construction(Absent), got {err:?}"
        );
    }

    #[tokio::test]
    async fn empty_set_from_a_custom_source_refuses_construction() {
        struct Empty;
        #[async_trait::async_trait]
        impl HostKeySource for Empty {
            async fn load(&self) -> Result<Vec<russh::keys::PrivateKey>, KeySourceError> {
                Ok(Vec::new())
            }
        }

        let serve = serve_config(Arc::new(Empty), Limits::default());
        let err = SshAdapter::new(serve)
            .await
            .expect_err("an empty key set serves nothing honestly");
        assert!(matches!(
            err,
            AdapterError::Construction(KeySourceError::Absent)
        ));
    }

    #[tokio::test]
    async fn every_key_source_variant_surfaces_through_construction() {
        let tampered_vault = unlocked_vault();
        let tampered = sealed(&tampered_vault, BRAVO_PEM);
        let mut tampered = tampered;
        tampered.data = "not the ciphertext".into();
        let decrypt_failure = Arc::new(VaultHostKeySource::new(tampered_vault, vec![tampered]));

        let version_vault = unlocked_vault();
        let version_blob = sealed(&version_vault, BRAVO_PEM);
        let mut version_blob = version_blob;
        version_blob.key_version = 1;
        let version_failure = Arc::new(VaultHostKeySource::new(version_vault, vec![version_blob]));

        let decode_vault = unlocked_vault();
        let decode_failure = Arc::new(VaultHostKeySource::new(
            decode_vault.clone(),
            vec![sealed(&decode_vault, "this is not a key")],
        ));

        let stranger_vault = unlocked_vault();
        let stranger = Arc::new(VaultHostKeySource::new(
            unlocked_vault(),
            vec![sealed(&stranger_vault, BRAVO_PEM)],
        ));

        for (source, expected) in [
            (decrypt_failure, KeySourceError::Decryption(String::new())),
            (
                version_failure,
                KeySourceError::UnknownKeyVersion(String::new()),
            ),
            (decode_failure, KeySourceError::Decode(String::new())),
            (stranger, KeySourceError::Decryption(String::new())),
        ] {
            let serve = serve_config(source, Limits::default());
            let err = SshAdapter::new(serve).await.expect_err("refuses");
            let matched = matches!(
                err,
                AdapterError::Construction(ref e) if std::mem::discriminant(e)
                    == std::mem::discriminant(&expected)
            );
            assert!(matched, "expected {expected:?} class, got {err:?}");
        }
    }

    #[tokio::test]
    async fn handle_mounts_the_stack_and_errors_when_the_peer_disconnects() {
        let serve = serve_config(static_fixture_host_keys(&[BRAVO_PEM]), Limits::default());
        let adapter = SshAdapter::new(serve).await.expect("constructs");

        let (client, server) = tokio::io::duplex(65_536);
        drop(client);

        let outcome = adapter.handle(server).await;
        assert!(
            matches!(outcome, Err(AdapterError::Wire(_))),
            "expected a wire-class mount failure, got {outcome:?}"
        );
    }

    struct AcceptAllHostKeys;

    impl russh::client::Handler for AcceptAllHostKeys {
        type Error = russh::Error;

        async fn check_server_key(
            &mut self,
            _server_public_key: &russh::keys::PublicKeyOrCertificate,
        ) -> Result<bool, Self::Error> {
            Ok(true)
        }
    }

    #[tokio::test]
    async fn wire_smoke_full_stack_mounts_and_skeleton_declines_auth_fail_closed() {
        let serve = serve_config(
            static_fixture_host_keys(&[BRAVO_PEM]),
            Limits {
                max_auth_attempts: 3,
                ..Limits::default()
            },
        );
        let adapter = SshAdapter::new(serve).await.expect("constructs");

        let (client_io, server_io) = tokio::io::duplex(65_536);
        let server_task = tokio::spawn(async move { adapter.handle(server_io).await });

        let client_config = Arc::new(russh::client::Config {
            inactivity_timeout: Some(Duration::from_secs(30)),
            ..Default::default()
        });
        let mut handle = russh::client::connect_stream(client_config, client_io, AcceptAllHostKeys)
            .await
            .expect("client mounts against the server stack");

        let alpha = Arc::new(decode_secret_key(ALPHA_PEM, None).expect("fixture decodes"));
        let result = handle
            .authenticate_publickey(
                "anyone",
                russh::keys::PrivateKeyWithHashAlg::new(alpha, None),
            )
            .await
            .expect("auth exchange completes over the wire");

        let remaining = match &result {
            AuthResult::Failure {
                remaining_methods, ..
            } => remaining_methods.clone(),
            other => {
                handle
                    .disconnect(russh::Disconnect::ByApplication, "test done", "en")
                    .await
                    .ok();
                panic!("the skeleton's fail-closed posture must not authenticate: {other:?}")
            }
        };
        assert!(
            !result.success(),
            "the skeleton's every-hook-default posture must not authenticate"
        );
        assert_eq!(
            remaining,
            russh::MethodSet::from(&[russh::MethodKind::PublicKey][..]),
            "the wire carries exactly the publickey-only advertisement"
        );

        handle
            .disconnect(russh::Disconnect::ByApplication, "test done", "en")
            .await
            .expect("client disconnects cleanly");
        server_task
            .await
            .expect("server task ends")
            .expect("clean session end");
    }

    #[test]
    fn server_config_records_the_russh_default_postures_it_keeps() {
        let serve = ServeConfig::new(Arc::new(NOOP), static_fixture_host_keys(&[BRAVO_PEM]));
        let config = build_server_config(&serve, Vec::new());

        assert_eq!(
            config.auth_rejection_time,
            std::time::Duration::from_secs(1),
            "constant-time rejection stays russh-managed"
        );
        assert_eq!(
            config.keepalive_interval, None,
            "the keepalive posture is the client direction's; the server stays russh-default"
        );
    }
}