//! The dial surface (`w4-dial`) — the door *outward* (`client.md`
//! §The dial surface), the client-direction mirror of
//! [`crate::adapter`]'s mount: the russh client stack every wave-4 op
//! row composes into.
//!
//! The entry ([`dial`]) takes the wave-1 [`DialConfig`] and runs the
//! whole authenticated-dial sequence, **fail-closed at every step**:
//!
//! 1. **The decrypt-at-dial key load** (ADR-005 decision 2): the
//!    injected [`KeySource::load`] runs *here and nowhere else* —
//!    locked / absent / decryption / unknown-version / decode are all
//!    **no-dial outcomes** ([`DialError::KeySource`], the variant
//!    preserved). The decoded key lives for this dial; the vault is
//!    needed only while the load runs (locking mid-session keeps
//!    established connections streaming, POC-2 §2).
//! 2. **The bounded russh client `Config`** ([`russh_client_config`])
//!    from the [`Limits`]: the keepalive 15 s / 3 posture
//!    (`reverse-runner.md` lesson 3 — OpenSSH
//!    `ServerAliveInterval=15`/`CountMax=3` parity), the wall clock
//!    **always `Some`** (`inactivity_timeout` — the crate never
//!    constructs a session with no clock), and the window/packet/
//!    channel-buffer knobs verbatim. Everything else stays russh's
//!    own default posture (the ADR-002 surface notes, mirrored from
//!    [`crate::adapter`]'s server mapping: rekey `Limits`
//!    russh-enforced, preferred algorithms russh's — the ed25519-only
//!    posture holds by the loaded key set itself). No client-side
//!    auth-attempt knob exists at this pin and none is needed: the
//!    family contract is exactly **one publickey attempt**.
//! 3. **The policy call at `check_server_key`** (OQ-05) — the
//!    wave-1 [`HostKeyPolicy`](crate::HostKeyPolicy) seam consumed by
//!    the handler ([`crate::client_handler`]): the crate default
//!    [`RejectAll`](crate::RejectAll) connects to nothing; a
//!    pin-mismatch hard-declines; a case-2 assembly policy decides.
//!    Declines surface as [`DialError::HostKeyRejected`].
//! 4. **Publickey-only authentication** (the family contract):
//!    exactly one `authenticate_publickey` attempt from the loaded
//!    key — no password, no keyboard-interactive, no cert, no retry,
//!    no fallback method (russh's handle methods for those are never
//!    exposed through [`SshConnection`]); a refusal is
//!    [`DialError::AuthRejected`] and the attempt is closed cleanly.
//!
//! **The connection handle** ([`SshConnection`]) is the type
//! everything else in the wave rides (the op rows open channels on
//! it; the `-R` leg installs the acceptor registry over it) — the
//! russh `client::Handle` wrapped with the crate's surface, the
//! policy, limits and endpoint labels carried. The connection handle's
//! clean-close path is [`SshConnection::close`]; the listener handles'
//! surface is [`crate::client_accept`]'s.
//!
//! ## Lifecycle posture (`w4-lifecycle` — the OQ-12 resolution in code)
//!
//! The handle surface is **primitives, not policy** (reverse-runner
//! fact 4): reconnect-and-restore (rebinding persisted tunnel ports —
//! the alkminer flow) is *assembly* composition over these
//! primitives; the crate provides bound ports, handles, cancellation,
//! clean close, and never owns a reconnect/restore policy. Two
//! recorded facts pin the explicit-vs-drop boundary:
//!
//! - **Drop is never the clean path.** At this russh pin dropping the
//!   last wire-stack handle ends the runner and the transport goes
//!   down **abruptly** — the peer observes a broken session (its
//!   serving outcome is an error wire class), never the clean
//!   disconnect message. So the surface never relies on drop timing
//!   for clean state: only [`SshConnection::close`] sends
//!   `Disconnect::ByApplication` — the peer observes a clean
//!   disconnect, no half-open channels (row-pinned through the
//!   server double, both directions of the boundary: abrupt-on-drop,
//!   clean-on-close).
//!   And a cancellation is async — it cannot hide inside a sync
//!   `Drop` hook, so no handle hides a `Drop` side effect: the
//!   listener handles' drop frees *nothing* on their wire state.
//! - **Every listener/handle is explicitly cancellable** — the `-R`
//!   listens return the [`ReverseListener`](crate::client_accept::
//!   ReverseListener) handles (the actually-bound endpoint carried;
//!   `cancel()` the wire-honest teardown), the connection handle
//!   carries [`SshConnection::close`]; the stateless wire forms
//!   ([`SshConnection::reverse_listen_cancel`] + the unix twin)
//!   mirror sshd's global-request cancels 1:1.
//!
//! ## Test surface
//!
//! Two mechanics are re-exported `#[doc(hidden)]` (the `serve.rs`
//! §Test-surface pattern) for the wire contract rows:
//! [`russh_client_config`] (the bounded-knob mapping, asserted
//! verbatim) and [`ClientHandler`](crate::client_handler::ClientHandler)
//! (the handler skeleton; `client_handler.rs`). The wire vehicle is
//! the in-process fake `sshd` in `tests/common/sshd_double.rs` — the
//! client direction dials vanilla servers, so the loopback harness
//! (the *server*-direction vehicle) only hosts the crate-to-crate
//! smoke (see the task's smoke-posture record).

use std::sync::Arc;

use russh::Disconnect;
use russh::keys::PrivateKeyWithHashAlg;

#[cfg(feature = "tunnel")]
use crate::channel_cap::ChannelCap;
#[cfg(feature = "tunnel")]
use crate::client_accept::AcceptorRegistry;
use crate::client_handler::ClientHandler;
use crate::config::{DialConfig, Limits};
use crate::error::DialError;

/// The authenticated dial — the client direction's entry point
/// (`client.md` §The dial surface): see the module docs for the
/// four-step sequence and their fail-closed outcomes.
pub async fn dial(config: DialConfig) -> Result<SshConnection, DialError> {
    let key = config.key_source.load().await?;

    let (host, port) = parse_dial_address(&config.address)?;
    let handler = ClientHandler::new(Arc::clone(&config.host_key_policy), host, port);
    #[cfg(feature = "tunnel")]
    let (handler, acceptors) = {
        let acceptors = Arc::new(AcceptorRegistry::new());
        (
            handler.with_tunnel_surface(Arc::clone(&acceptors), config.limits),
            acceptors,
        )
    };

    let russh_config = Arc::new(build_client_config(&config.limits));
    let mut handle = russh::client::connect(russh_config, config.address.as_str(), handler).await?;

    let result = handle
        .authenticate_publickey(
            config.username.as_str(),
            PrivateKeyWithHashAlg::new(Arc::new(key), None),
        )
        .await?;
    if !result.success() {
        let _ = handle
            .disconnect(Disconnect::ByApplication, "authentication refused", "en")
            .await;
        return Err(DialError::AuthRejected);
    }

    Ok(SshConnection {
        shared: Arc::new(ConnectionShared {
            handle,
            limits: config.limits,
            address: config.address,
            username: config.username,
            #[cfg(feature = "tunnel")]
            acceptors,
            #[cfg(feature = "tunnel")]
            listener_budget: Arc::new(ChannelCap::default()),
        }),
    })
}

/// Parses the dial address into its endpoint labels — the
/// [`PresentedHostKey`](crate::host_key::PresentedHostKey)'s
/// `host`/`port` decision context. Malformed targets refuse before
/// any connection ([`DialError::Address`]): plain `host:port` and the
/// bracket form `[host]:port` (literal IPv6) are the grammar.
fn parse_dial_address(address: &str) -> Result<(String, u16), DialError> {
    let (host, port) = if let Some(rest) = address.strip_prefix('[') {
        let (host, rest) = rest
            .split_once(']')
            .ok_or_else(|| DialError::Address(address.to_string()))?;
        let port = rest
            .strip_prefix(':')
            .ok_or_else(|| DialError::Address(address.to_string()))?;
        (host, port)
    } else {
        address
            .rsplit_once(':')
            .ok_or_else(|| DialError::Address(address.to_string()))?
    };
    if host.is_empty() {
        return Err(DialError::Address(address.to_string()));
    }
    let port = port
        .parse::<u16>()
        .map_err(|_| DialError::Address(address.to_string()))?;
    Ok((host.to_string(), port))
}

/// The [`Limits`] → russh client `Config` mapping — the bounded knobs
/// verbatim, the wall clock always on, everything else at russh's own
/// default posture (see the module docs). The mapping is the dial
/// task's `bounded-config wiring observable`;
/// `#[doc(hidden)]` re-export from the crate root.
pub(crate) fn build_client_config(limits: &Limits) -> russh::client::Config {
    russh::client::Config {
        window_size: limits.window_size,
        maximum_packet_size: limits.maximum_packet_size,
        channel_buffer_size: limits.channel_buffer_size,
        inactivity_timeout: Some(limits.inactivity_timeout),
        keepalive_interval: limits.keepalive_interval,
        keepalive_max: limits.keepalive_max,
        ..russh::client::Config::default()
    }
}

#[doc(hidden)]
pub fn russh_client_config(limits: &Limits) -> russh::client::Config {
    build_client_config(limits)
}

/// The authenticated connection handle — the type every later
/// wave-4 op rides (`client.md`: "the authenticated connection
/// handle"): the russh `client::Handle` wrapped with the crate's
/// surface. The remote end is authenticated (publickey-only, single
/// attempt); the dial key is gone (its lifetime is the dial's —
/// everything after rides the wire, never the key).
///
/// The russh handle, budgets and registry ride one shared inner
/// (behind the `Arc`) — the sharing the `client_accept` module's
/// listener handles need for their independent `cancel()` calls; the
/// wire-visible teardowns stay explicit calls on top of that shared
/// stack (the module docs' §Lifecycle posture: drop is never a
/// wire-visible act).
pub struct SshConnection {
    shared: Arc<ConnectionShared>,
}

/// The connection handle's internals — the one behind the
/// [`SshConnection`]'s `Arc`, every field private to the crate: the
/// wire surface is the wrapped methods, never the russh stack itself
/// (`serve.rs`'s wrapped-surface pattern mirrored client-side).
///
/// The arc's meaning for the lifecycle boundary: the russh stack
/// (the runner) lives while *any* handle holds the arc — the
/// connection handle's drop is the abrupt teardown only when it is
/// the last holder (row-pinned), and a live `ReverseListener` (its
/// `cancel()` composes on this stack) keeps the wire serving until
/// its own explicit cancel. The clean teardown is
/// [`SshConnection::close`]'s, whatever holders exist.
pub(crate) struct ConnectionShared {
    /// The russh handle the connection and the listener handles
    /// compose into (its own request methods take `&self` at this
    /// pin — the shared arc is the concurrent-sharing shape).
    pub(crate) handle: russh::client::Handle<ClientHandler>,
    pub(crate) limits: Limits,
    pub(crate) address: String,
    pub(crate) username: String,
    /// The `-R` leg's acceptor registry (the dial constructs it empty;
    /// the assembly installs/uninstalls across the live connection —
    /// see [`SshConnection::acceptors`] and
    /// [`crate::client_accept`]`'s recorded decisions.
    #[cfg(feature = "tunnel")]
    pub(crate) acceptors: Arc<AcceptorRegistry>,
    /// The reverse-listen count's dedicated bound — admitted at the
    /// listen rows, released at the confirmed cancels and the refused/
    /// wire-error paths (the cap relation recorded in
    /// [`crate::client_accept`]; [`crate::channel_cap`]'s mutex makes
    /// the counts sound under the caller's task). Behind the arc so
    /// the listener handles release it from their own tasks too.
    #[cfg(feature = "tunnel")]
    pub(crate) listener_budget: Arc<ChannelCap>,
}

impl SshConnection {
    /// The bounded-resource budget the dial ran under (carried; the
    /// later op rows' admission surface).
    pub fn limits(&self) -> &Limits {
        &self.shared.limits
    }

    /// The dialed address as the config carried it.
    pub fn address(&self) -> &str {
        &self.shared.address
    }

    /// The user the connection authenticated as.
    pub fn username(&self) -> &str {
        &self.shared.username
    }

    /// The `-R` leg's acceptor registry (the
    /// `w4-reverse-listen` surface, `client_accept`'s module docs
    /// carry the recorded decisions) — install/uninstall the
    /// per-type acceptors across the live connection (the
    /// late-install dynamic). The dial constructs it empty: nothing
    /// server-initiated is served until a slot holds, fail-closed at
    /// arrival time.
    #[cfg(feature = "tunnel")]
    pub fn acceptors(&self) -> &AcceptorRegistry {
        &self.shared.acceptors
    }

    /// The reverse-listen count's shared handle — the op rows
    /// ([`crate::client_accept`]) read it; private to the crate.
    #[cfg(feature = "tunnel")]
    pub(crate) fn listener_budget(&self) -> &ChannelCap {
        &self.shared.listener_budget
    }

    /// The shared inner — the listener handles' compose-into point
    /// (their cancels run against the same russh stack); private to
    /// the crate.
    #[cfg(feature = "tunnel")]
    pub(crate) fn shared(&self) -> &Arc<ConnectionShared> {
        &self.shared
    }

    /// The russh handle the op rows compose into (`w4-op-rows` — the
    /// method bodies live in [`crate::client_ops`], a sibling module,
    /// so the field stays private to this module).
    #[cfg(any(feature = "session", feature = "git", feature = "tunnel"))]
    pub(crate) fn russh_handle(
        &self,
    ) -> &russh::client::Handle<crate::client_handler::ClientHandler> {
        &self.shared.handle
    }

    /// The dial's bounded-resource budget, verbatim — the op rows
    /// read the knobs they consume (the exec-command length cap
    /// drives the command-carrying rows' client-mirror gate).
    #[cfg(any(feature = "session", feature = "git"))]
    pub(crate) fn budgets(&self) -> &Limits {
        &self.shared.limits
    }

    /// Whether the underlying russh session is closed.
    pub fn is_closed(&self) -> bool {
        self.shared.handle.is_closed()
    }

    /// The explicit clean-close path (the lifecycle primitive,
    /// [`russh::Disconnect::ByApplication`] — the honest close): the
    /// peer observes a clean disconnect, no half-open channels —
    /// every open channel is torn down with the session (row-pinned
    /// through the server double). **Drop is not this path** (the
    /// module docs' §Lifecycle posture): a dropped handle releases
    /// nothing on the wire — the connection outlives the handle
    /// until the peer or the wall clock ends it.
    pub async fn close(self) {
        let _ = self
            .shared
            .handle
            .disconnect(Disconnect::ByApplication, "closed by application", "en")
            .await;
    }
}

impl std::fmt::Debug for SshConnection {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let mut debug = f.debug_struct("SshConnection");
        debug
            .field("address", &self.shared.address)
            .field("username", &self.shared.username)
            .field("limits", &self.shared.limits)
            .field("closed", &self.shared.handle.is_closed());
        #[cfg(feature = "tunnel")]
        debug.field("acceptors", &self.shared.acceptors);
        debug.finish()
    }
}

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

    use russh::keys::PrivateKey;

    use super::{build_client_config, parse_dial_address};
    use crate::config::{DialConfig, Limits};
    use crate::error::{DialError, KeySourceError};
    use crate::host_key::RejectAll;
    use crate::key_source::KeySource;

    struct NoKeySource;

    #[async_trait::async_trait]
    impl KeySource for NoKeySource {
        async fn load(&self) -> Result<PrivateKey, KeySourceError> {
            Err(KeySourceError::Absent)
        }
    }

    /// - *Contract stamp:* `client.md` §Correctness facts (the
    ///   keepalive posture) + `overview.md` §Bounded-resources table —
    ///   the defaults land in the russh mapping verbatim.
    #[test]
    fn bounded_config_mapping_carries_the_default_postures() {
        let config = build_client_config(&Limits::default());

        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.inactivity_timeout,
            Some(crate::config::DEFAULT_INACTIVITY_TIMEOUT),
            "the wall clock is always set — no unbounded session"
        );
        assert_eq!(
            config.keepalive_interval,
            Some(Duration::from_secs(15)),
            "the 15 s keepalive posture is the dial's default"
        );
        assert_eq!(config.keepalive_max, 3);
    }

    /// - *Contract stamp:* the [`Limits`] tunability contract (every
    ///   knob is the assembly's to tune; nothing is crate-fixed).
    #[test]
    fn bounded_config_mapping_is_tunable_one_for_one() {
        let limits = Limits {
            window_size: 1_024,
            maximum_packet_size: 4_096,
            channel_buffer_size: 7,
            inactivity_timeout: Duration::from_secs(1_337),
            keepalive_interval: None,
            keepalive_max: 9,
            ..Limits::default()
        };
        let config = build_client_config(&limits);

        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.inactivity_timeout, Some(Duration::from_secs(1_337)));
        assert_eq!(config.keepalive_interval, None);
        assert_eq!(config.keepalive_max, 9);
    }

    #[test]
    fn dial_address_plain_form_parses() {
        let (host, port) = parse_dial_address("runner.example:22").expect("parses");
        assert_eq!(host, "runner.example");
        assert_eq!(port, 22);

        let (host, port) = parse_dial_address("127.0.0.1:4022").expect("parses");
        assert_eq!(host, "127.0.0.1");
        assert_eq!(port, 4022);
    }

    #[test]
    fn dial_address_bracket_form_parses() {
        let (host, port) = parse_dial_address("[2001:db8::1]:2222").expect("parses");
        assert_eq!(host, "2001:db8::1");
        assert_eq!(port, 2222);
    }

    #[test]
    fn dial_address_malformed_forms_refuse() {
        for bad in [
            "runner.example",
            ":22",
            "[]:22",
            "host:port",
            "[::1:2]x",
            "",
        ] {
            let err = parse_dial_address(bad).expect_err("refuses");
            assert!(
                matches!(err, DialError::Address(ref given) if *given == bad),
                "expected Address, got {err:?}"
            );
        }
    }

    #[test]
    fn dial_address_out_of_range_port_refuses() {
        let err = parse_dial_address("host:99999").expect_err("refuses");
        assert!(matches!(err, DialError::Address(_)));
    }

    #[tokio::test]
    async fn key_source_refusal_surfaces_the_variant_through_the_config() {
        let config = DialConfig::new(
            "runner.example:22",
            "runner",
            Arc::new(NoKeySource),
            Arc::new(RejectAll),
        );
        assert!(
            matches!(config.key_source.load().await, Err(KeySourceError::Absent)),
            "the no-dial outcome's variant is preserved through the config"
        );
    }
}