//! QUIC transport for peer RPC with fleet-CA mTLS.
//!
//! Runs on a UDP port distinct from chitchat gossip (see
//! [`ClusterConfig::rpc_lan`](crate::config::ClusterConfig::rpc_lan) —
//! defaults to gossip port + 1) so the two UDP-based protocols don't
//! collide.
//!
//! # Trust model
//!
//! A single fleet root CA signs every node's leaf certificate. Each
//! node loads its own cert + private key + the CA public cert into a
//! [`NodeIdentity`]. Both server and client verifiers require peer
//! certs to chain to the fleet CA — that's mTLS. A node without a
//! signed cert cannot join RPC.
//!
//! # Test model
//!
//! The [`NodeIdentity::generate_test_pair`] helper cuts an ephemeral
//! CA + two leaf certs in-memory for two named nodes. Both nodes share
//! the CA (so they trust each other) but have distinct leaves. This is
//! the same code path used in production; there is no test-only side
//! door in the transport itself.
//!
//! # ALPN
//!
//! The single protocol identifier `clawstor-rpc/1` is offered on every
//! connection; peers advertising anything else are rejected during the
//! TLS handshake.

use crate::config::{ClusterConfig, ClusterTlsConfig};
use anyhow::{bail, Context, Result};
use quinn::{ClientConfig, Endpoint, ServerConfig, VarInt};
use rustls::pki_types::{CertificateDer, PrivateKeyDer};
use std::fs::File;
use std::io::{BufReader, Write};
use std::net::SocketAddr;
use std::path::Path;
use std::sync::Arc;
use std::time::Duration;

/// ALPN identifier all fleet RPC endpoints advertise + require.
pub const CLAWSTOR_RPC_ALPN: &[u8] = b"clawstor-rpc/1";

/// Idle timeout on connection — if no data for this long the connection
/// dies. Short enough to notice partitions, long enough to survive a
/// paused laptop.
/// Field finding 2026-07-12: raised from 30s to 10min because a long
/// `cargo build` between the initial connect and the follow-up upload
/// would silently kill the QUIC connection on `open_bi`. Cargo builds
/// on real workspaces routinely run for minutes; the idle timeout is
/// there to detect crashed peers, not to enforce interaction cadence.
const IDLE_TIMEOUT: Duration = Duration::from_secs(600);

/// Field finding 2026-07-12: keep the connection warm with a ping
/// every KEEP_ALIVE_INTERVAL — cheap belt-and-braces on top of the
/// larger idle window so cargo runs longer than the idle timeout
/// still stay dialed.
const KEEP_ALIVE_INTERVAL: Duration = Duration::from_secs(15);

/// Cap on any single RPC message payload. Ping/pong is tiny; other RPCs
/// stream larger payloads via streams-of-many-messages. Prevents an
/// adversary from allocating unbounded memory on a single message.
const MAX_MESSAGE_BYTES: usize = 16 * 1024;

/// Cryptographic identity a node presents on both sides of the RPC
/// endpoint. Fully specifies the mTLS setup: our own cert chain + key,
/// plus the CA cert we require the peer's cert to chain to.
///
/// [`PrivateKeyDer`] is not `Clone`, so `NodeIdentity` isn't derive-Clone
/// either. Use [`NodeIdentity::clone_id`] when a second owner is required.
#[derive(Debug)]
pub struct NodeIdentity {
    /// This node's certificate chain (leaf + optional intermediates).
    pub cert_chain: Vec<CertificateDer<'static>>,
    /// This node's private key matching the leaf cert above.
    pub key: PrivateKeyDer<'static>,
    /// The fleet root CA. Both server verifier and client verifier
    /// require the peer's chain to end at this cert.
    pub trusted_ca: CertificateDer<'static>,
}

impl NodeIdentity {
    /// Build a fresh fleet CA and a pair of signed leaf certs named
    /// after `node_a` and `node_b`. Returns the two independent
    /// identities — each carries its own leaf but shares the CA so
    /// they trust each other's certs.
    ///
    /// Used by tests + as a starting point for the production
    /// bootstrap flow (write PEMs to `/etc/claw-store/tls/`).
    pub fn generate_test_pair(node_a: &str, node_b: &str) -> Result<(Self, Self)> {
        let ca_key = rcgen::KeyPair::generate().context("generating CA key")?;
        let mut ca_params = rcgen::CertificateParams::new(vec![])
            .context("building CA params")?;
        ca_params.is_ca = rcgen::IsCa::Ca(rcgen::BasicConstraints::Unconstrained);
        ca_params
            .distinguished_name
            .push(rcgen::DnType::CommonName, "clawstor fleet CA");
        let ca_cert = ca_params
            .self_signed(&ca_key)
            .context("self-signing CA cert")?;
        let ca_der = ca_cert.der().clone();

        let id_a = build_leaf(node_a, &ca_cert, &ca_key, ca_der.clone())?;
        let id_b = build_leaf(node_b, &ca_cert, &ca_key, ca_der)?;
        Ok((id_a, id_b))
    }

    /// Load a persistent identity from PEM files. Production path — reads
    /// paths configured in `[cluster.tls]`:
    ///   * `ca_path` — PEM of the fleet root CA cert
    ///   * `cert_path` — PEM of this node's leaf cert (signed by the CA)
    ///   * `key_path` — PEM of this node's private key
    ///
    /// The three files are always kept together. Convention:
    /// `/etc/claw-store/tls/{ca.crt, node.crt, node.key}` with node.key mode
    /// 0o600. The fleet CA private key stays on whichever host cut it
    /// (typically the primary); every other node needs only the public
    /// CA cert.
    pub fn from_pem_files(
        ca_path: &Path,
        cert_path: &Path,
        key_path: &Path,
    ) -> Result<Self> {
        let trusted_ca = read_single_cert(ca_path)
            .with_context(|| format!("loading CA cert from {}", ca_path.display()))?;
        let leaf = read_single_cert(cert_path)
            .with_context(|| format!("loading node cert from {}", cert_path.display()))?;
        let key = read_private_key(key_path)
            .with_context(|| format!("loading node private key from {}", key_path.display()))?;
        Ok(Self {
            cert_chain: vec![leaf],
            key,
            trusted_ca,
        })
    }

    /// Convenience: read all three PEM files from a single directory using
    /// the canonical filenames `ca.crt`, `node.crt`, `node.key`.
    pub fn from_pem_dir(dir: &Path) -> Result<Self> {
        Self::from_pem_files(
            &dir.join("ca.crt"),
            &dir.join("node.crt"),
            &dir.join("node.key"),
        )
    }

    /// Load persistent identity from the paths in a `[cluster.tls]` block.
    /// Errors when the config doesn't have a TLS section.
    pub fn from_cluster_config(cfg: &ClusterConfig) -> Result<Self> {
        let tls = cfg
            .tls
            .as_ref()
            .context("cluster.tls section not configured; cannot load node identity")?;
        Self::from_pem_files(&tls.ca_cert, &tls.node_cert, &tls.node_key)
    }
}

/// Read a single X.509 cert from a PEM file, DER form.
fn read_single_cert(path: &Path) -> Result<CertificateDer<'static>> {
    let mut reader = BufReader::new(File::open(path)?);
    let mut iter = rustls_pemfile::certs(&mut reader);
    let first = iter
        .next()
        .context("PEM contained no CERTIFICATE block")?
        .context("parsing CERTIFICATE block")?;
    Ok(first)
}

/// Read a PKCS#8 or SEC1 or RSA private key from a PEM file. `rustls-pemfile`
/// handles all three formats transparently — we don't care which one the
/// operator's `openssl`/`rcgen`/whatever tool produced.
fn read_private_key(path: &Path) -> Result<PrivateKeyDer<'static>> {
    let mut reader = BufReader::new(File::open(path)?);
    rustls_pemfile::private_key(&mut reader)?
        .context("PEM contained no PRIVATE KEY block")
}

/// The fleet certificate authority. Held only by the node that cut it (or
/// nodes that hold a copy of the CA private key). Signs per-node leaf certs
/// via [`FleetCa::sign_leaf`]. Persisted to disk as `ca.crt` + `ca.key`.
///
/// [`Debug`] is implemented manually to elide the private key. Accidentally
/// printing the CA key would let anyone forge fleet identities.
pub struct FleetCa {
    /// The self-signed CA certificate.
    cert: rcgen::Certificate,
    /// The CA's private key. NEVER logged, NEVER printed.
    key: rcgen::KeyPair,
    /// PEM copy of `cert`, cached to avoid re-serializing on every use.
    cert_pem: String,
    /// Owned DER of `cert`, cached for `NodeIdentity::trusted_ca`.
    ca_der: CertificateDer<'static>,
}

impl std::fmt::Debug for FleetCa {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("FleetCa")
            .field("cert_pem_len", &self.cert_pem.len())
            .field("ca_der_len", &self.ca_der.len())
            .field("key", &"<redacted>")
            .finish()
    }
}

impl FleetCa {
    /// Generate a new fleet CA with the given common name. Used by the
    /// `fleet-ca init` CLI to bootstrap a fresh cluster.
    pub fn generate(cn: &str) -> Result<Self> {
        if cn.is_empty() {
            bail!("fleet CA common name cannot be empty");
        }
        let key = rcgen::KeyPair::generate().context("generating CA key")?;
        let mut params = rcgen::CertificateParams::new(vec![])
            .context("building CA cert params")?;
        params.is_ca = rcgen::IsCa::Ca(rcgen::BasicConstraints::Unconstrained);
        params
            .distinguished_name
            .push(rcgen::DnType::CommonName, cn);
        let cert = params.self_signed(&key).context("self-signing CA cert")?;
        let cert_pem = cert.pem();
        let ca_der = cert.der().clone();
        Ok(Self {
            cert,
            key,
            cert_pem,
            ca_der,
        })
    }

    /// Persist the CA to a directory as `ca.crt` (public) and `ca.key`
    /// (private, mode 0o600 on Unix). Creates the directory if missing.
    pub fn save(&self, dir: &Path) -> Result<()> {
        std::fs::create_dir_all(dir)
            .with_context(|| format!("creating CA dir {}", dir.display()))?;
        let cert_path = dir.join("ca.crt");
        std::fs::write(&cert_path, &self.cert_pem)
            .with_context(|| format!("writing {}", cert_path.display()))?;
        let key_path = dir.join("ca.key");
        write_secret_file(&key_path, self.key.serialize_pem().as_bytes())
            .with_context(|| format!("writing {}", key_path.display()))?;
        Ok(())
    }

    /// Load a previously-saved CA from `ca.crt` + `ca.key` in `dir`.
    pub fn load(dir: &Path) -> Result<Self> {
        let cert_path = dir.join("ca.crt");
        let key_path = dir.join("ca.key");
        let cert_pem = std::fs::read_to_string(&cert_path)
            .with_context(|| format!("reading {}", cert_path.display()))?;
        let key_pem = std::fs::read_to_string(&key_path)
            .with_context(|| format!("reading {}", key_path.display()))?;
        // Reconstruct the rcgen types from PEM. `KeyPair::from_pem` handles
        // PKCS#8 output the way we serialized it; `params_from_ca_cert_pem`
        // rebuilds the params so subsequent signing operations produce
        // certs that chain correctly.
        let key = rcgen::KeyPair::from_pem(&key_pem)
            .context("parsing CA private key from PEM")?;
        let params = rcgen::CertificateParams::from_ca_cert_pem(&cert_pem)
            .context("parsing CA cert PEM")?;
        let cert = params.self_signed(&key).context("re-binding CA cert to key")?;
        // Use the ORIGINAL cert bytes from disk as our authoritative DER.
        // Rebuilding via self_signed above yields a byte-identical cert
        // when the SPKI matches, but reading from disk is deterministic.
        let ca_der = read_single_cert(&cert_path)?;
        let cert_pem_owned = cert_pem;
        Ok(Self {
            cert,
            key,
            cert_pem: cert_pem_owned,
            ca_der,
        })
    }

    /// Sign a leaf cert for `node_name` and return an in-memory
    /// [`NodeIdentity`] ready to hand to `QuicServer::bind` /
    /// `QuicClient::new`.
    pub fn sign_leaf(&self, node_name: &str) -> Result<NodeIdentity> {
        if node_name.is_empty() {
            bail!("node name cannot be empty when signing a leaf");
        }
        let (leaf_key, leaf_cert) = self.mint_leaf(node_name)?;
        let leaf_der = leaf_cert.der().clone();
        let key_der = PrivateKeyDer::try_from(leaf_key.serialize_der())
            .map_err(|e| anyhow::anyhow!("converting leaf key to rustls form: {e}"))?;
        Ok(NodeIdentity {
            cert_chain: vec![leaf_der],
            key: key_der,
            trusted_ca: self.ca_der.clone(),
        })
    }

    /// Sign a leaf cert and write PEM files (`node.crt`, `node.key`,
    /// `ca.crt`) into `out_dir`. Used by `fleet-ca sign`.
    pub fn sign_leaf_to_pem(&self, node_name: &str, out_dir: &Path) -> Result<()> {
        self.sign_leaf_to_pem_with_sans(node_name, &[], out_dir)
    }

    /// Phase 8 (2026-07-14): sign a leaf with extra SANs alongside the
    /// primary `node_name`. Used by `fleet-ca-tailscale-sign` so a
    /// laptop's leaf cert works whether peers dial by its LAN
    /// hostname *or* its Tailscale MagicDNS name.
    ///
    /// `extra_sans` are dropped in as DNS SANs. Empty entries are
    /// skipped so callers can conditionally include a value without
    /// pre-filtering.
    pub fn sign_leaf_to_pem_with_sans(
        &self,
        node_name: &str,
        extra_sans: &[String],
        out_dir: &Path,
    ) -> Result<()> {
        if node_name.is_empty() {
            bail!("node name cannot be empty when signing a leaf");
        }
        let (leaf_key, leaf_cert) = self.mint_leaf_with_sans(node_name, extra_sans)?;
        std::fs::create_dir_all(out_dir)
            .with_context(|| format!("creating output dir {}", out_dir.display()))?;

        let node_crt_path = out_dir.join("node.crt");
        std::fs::write(&node_crt_path, leaf_cert.pem())
            .with_context(|| format!("writing {}", node_crt_path.display()))?;

        let node_key_path = out_dir.join("node.key");
        write_secret_file(&node_key_path, leaf_key.serialize_pem().as_bytes())
            .with_context(|| format!("writing {}", node_key_path.display()))?;

        let ca_out_path = out_dir.join("ca.crt");
        std::fs::write(&ca_out_path, &self.cert_pem)
            .with_context(|| format!("writing {}", ca_out_path.display()))?;
        Ok(())
    }

    /// Shared leaf-minting: generate a key, sign the SAN=[node_name] cert.
    fn mint_leaf(
        &self,
        node_name: &str,
    ) -> Result<(rcgen::KeyPair, rcgen::Certificate)> {
        self.mint_leaf_with_sans(node_name, &[])
    }

    /// Phase 8: like [`mint_leaf`] but tacks on additional DNS SANs.
    /// Order: `[node_name, ...extra_sans]`. Empty extras are dropped
    /// so callers can conditionally pass values.
    fn mint_leaf_with_sans(
        &self,
        node_name: &str,
        extra_sans: &[String],
    ) -> Result<(rcgen::KeyPair, rcgen::Certificate)> {
        let leaf_key = rcgen::KeyPair::generate().context("generating leaf key")?;
        let mut sans = vec![node_name.to_string()];
        for s in extra_sans {
            let s = s.trim();
            if s.is_empty() || sans.iter().any(|existing| existing == s) {
                continue;
            }
            sans.push(s.to_string());
        }
        let mut leaf_params =
            rcgen::CertificateParams::new(sans).context("building leaf params")?;
        leaf_params
            .distinguished_name
            .push(rcgen::DnType::CommonName, node_name);
        let leaf_cert = leaf_params
            .signed_by(&leaf_key, &self.cert, &self.key)
            .context("signing leaf cert with CA")?;
        Ok((leaf_key, leaf_cert))
    }
}

/// Write a file with restrictive permissions (0o600 on Unix). Existing
/// mode-independent writes go through `std::fs::write`; this is for
/// private keys where broader read access is a real vulnerability.
fn write_secret_file(path: &Path, contents: &[u8]) -> Result<()> {
    let mut file = File::create(path)
        .with_context(|| format!("creating {}", path.display()))?;
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600))
            .with_context(|| format!("chmod 0o600 {}", path.display()))?;
    }
    file.write_all(contents)?;
    file.sync_all()?;
    Ok(())
}

/// Sign a single leaf cert for `name` under the given CA and package it
/// with the CA cert into a [`NodeIdentity`].
fn build_leaf(
    name: &str,
    ca_cert: &rcgen::Certificate,
    ca_key: &rcgen::KeyPair,
    ca_der: CertificateDer<'static>,
) -> Result<NodeIdentity> {
    let leaf_key = rcgen::KeyPair::generate().context("generating leaf key")?;
    // Include the node name as a DNS SAN so a client connecting with
    // server_name=name passes rustls's cert-name verification.
    let mut leaf_params = rcgen::CertificateParams::new(vec![name.to_string()])
        .context("building leaf params")?;
    leaf_params
        .distinguished_name
        .push(rcgen::DnType::CommonName, name);
    let leaf_cert = leaf_params
        .signed_by(&leaf_key, ca_cert, ca_key)
        .context("signing leaf cert with CA")?;
    let leaf_der = leaf_cert.der().clone();
    // rustls wants the private key in DER form.
    let key_der = PrivateKeyDer::try_from(leaf_key.serialize_der())
        .map_err(|e| anyhow::anyhow!("converting leaf key to rustls form: {e}"))?;
    Ok(NodeIdentity {
        cert_chain: vec![leaf_der],
        key: key_der,
        trusted_ca: ca_der,
    })
}

/// A running QUIC RPC server. Owns the quinn endpoint; drop stops it.
pub struct QuicServer {
    endpoint: Endpoint,
}

impl QuicServer {
    /// Bind a QUIC listener on `bind` with mTLS enforced against the
    /// fleet CA in `identity`. `bind` accepts port 0 to let the OS
    /// assign one (useful in tests).
    pub fn bind(bind: SocketAddr, identity: NodeIdentity) -> Result<Self> {
        install_default_crypto_provider();
        let server_crypto = build_server_crypto(&identity)?;
        let quic_crypto = quinn::crypto::rustls::QuicServerConfig::try_from(server_crypto)
            .context("wrapping rustls ServerConfig for quinn")?;
        let mut server_config = ServerConfig::with_crypto(Arc::new(quic_crypto));
        Arc::get_mut(&mut server_config.transport)
            .expect("fresh transport config is unique")
            .max_concurrent_uni_streams(0u8.into())
            .max_idle_timeout(Some(
                VarInt::from_u64(IDLE_TIMEOUT.as_millis() as u64)
                    .expect("idle timeout fits u64")
                    .into(),
            ));
        let endpoint =
            Endpoint::server(server_config, bind).context("binding quinn server endpoint")?;
        Ok(Self { endpoint })
    }

    /// The actual bound socket address (resolves port 0 to the assigned port).
    pub fn local_addr(&self) -> Result<SocketAddr> {
        self.endpoint.local_addr().map_err(Into::into)
    }

    /// Accept the next incoming connection. Returns `None` when the
    /// endpoint is closed. The returned [`quinn::Connection`] is
    /// already handshaked; call `ping_handler_loop` (or any other
    /// server-side handler) to serve requests on it.
    pub async fn accept(&self) -> Option<Result<quinn::Connection>> {
        let incoming = self.endpoint.accept().await?;
        // Two failure modes matter here:
        //   1. `incoming.accept()` returns Err → convert to a Result inside
        //      the returned Option so callers see the accept failure.
        //   2. The Connecting future fails → same shape.
        let connecting = match incoming.accept() {
            Ok(c) => c,
            Err(e) => return Some(Err(anyhow::Error::from(e))),
        };
        Some(connecting.await.map_err(anyhow::Error::from))
    }

    /// Graceful shutdown.
    pub async fn shutdown(&self) {
        self.endpoint
            .close(VarInt::from_u32(0), b"server shutdown");
        self.endpoint.wait_idle().await;
    }
}

/// A QUIC client for opening outbound peer connections.
pub struct QuicClient {
    endpoint: Endpoint,
    client_config: ClientConfig,
}

impl QuicClient {
    /// Build a client bound to `bind_addr` (usually `0.0.0.0:0` — OS
    /// picks the port) that verifies peer certs against the fleet CA
    /// carried in `identity`, and presents `identity`'s leaf on mTLS
    /// challenge.
    pub fn new(bind_addr: SocketAddr, identity: NodeIdentity) -> Result<Self> {
        install_default_crypto_provider();
        let client_crypto = build_client_crypto(&identity)?;
        let quic_crypto = quinn::crypto::rustls::QuicClientConfig::try_from(client_crypto)
            .context("wrapping rustls ClientConfig for quinn")?;
        let mut client_config = ClientConfig::new(Arc::new(quic_crypto));
        // Field finding 2026-07-12: apply the raised idle timeout + a
        // keep-alive so a long cargo build between the initial connect
        // and a follow-up upload doesn't kill the connection.
        let mut transport = quinn::TransportConfig::default();
        transport
            .max_idle_timeout(Some(
                VarInt::from_u64(IDLE_TIMEOUT.as_millis() as u64)
                    .expect("idle timeout fits u64")
                    .into(),
            ))
            .keep_alive_interval(Some(KEEP_ALIVE_INTERVAL));
        client_config.transport_config(Arc::new(transport));
        let endpoint = Endpoint::client(bind_addr).context("binding quinn client endpoint")?;
        Ok(Self {
            endpoint,
            client_config,
        })
    }

    /// Connect to `addr` and validate the peer cert's SAN matches
    /// `expected_server_name`. Returns the completed connection.
    pub async fn connect(
        &self,
        addr: SocketAddr,
        expected_server_name: &str,
    ) -> Result<quinn::Connection> {
        let connecting = self
            .endpoint
            .connect_with(self.client_config.clone(), addr, expected_server_name)
            .with_context(|| format!("dialing {addr} for name {expected_server_name}"))?;
        connecting
            .await
            .with_context(|| format!("completing handshake with {addr}"))
    }

    /// Phase 8b (2026-07-14): LAN-first probe with a tailnet
    /// fallback. Attempts `lan` (when given) under a tight budget;
    /// if the handshake doesn't complete in `lan_probe`, falls
    /// back to `tailscale` (when given) under a longer budget.
    ///
    /// Rationale: on the fleet's 1G LAN a handshake typically
    /// finishes in single-digit ms. Tailscale (WireGuard over WAN
    /// for roaming clients) can take 100-500 ms. A short LAN
    /// probe lets in-office nodes take the fast path without
    /// starving roaming nodes when LAN isn't reachable.
    ///
    /// If both address slots are `None` the call errors
    /// immediately rather than hanging.
    pub async fn connect_lan_first(
        &self,
        expected_server_name: &str,
        lan: Option<SocketAddr>,
        tailscale: Option<SocketAddr>,
        lan_probe: std::time::Duration,
    ) -> Result<(quinn::Connection, ConnectRoute)> {
        if let Some(lan_addr) = lan {
            // No fallback path → don't apply the probe deadline.
            // Otherwise a slow-but-fine LAN handshake can spuriously
            // fail when the operator never opted into a tailnet
            // fallback in the first place.
            if tailscale.is_none() {
                let conn = self
                    .connect(lan_addr, expected_server_name)
                    .await
                    .with_context(|| format!("dialing LAN addr {lan_addr}"))?;
                return Ok((conn, ConnectRoute::Lan(lan_addr)));
            }
            match tokio::time::timeout(
                lan_probe,
                self.connect(lan_addr, expected_server_name),
            )
            .await
            {
                Ok(Ok(conn)) => return Ok((conn, ConnectRoute::Lan(lan_addr))),
                Ok(Err(e)) => tracing::debug!(
                    peer = expected_server_name,
                    lan = %lan_addr,
                    error = %e,
                    "LAN dial failed; trying tailnet if configured"
                ),
                Err(_) => tracing::debug!(
                    peer = expected_server_name,
                    lan = %lan_addr,
                    "LAN probe hit deadline; falling back to tailnet"
                ),
            }
        }
        if let Some(ts_addr) = tailscale {
            let conn = self
                .connect(ts_addr, expected_server_name)
                .await
                .with_context(|| format!("tailnet fallback to {ts_addr}"))?;
            return Ok((conn, ConnectRoute::Tailscale(ts_addr)));
        }
        bail!(
            "no reachable address for {expected_server_name}: LAN {} + tailnet {} both failed or absent",
            lan.map(|a| a.to_string()).unwrap_or_else(|| "-".into()),
            tailscale.map(|a| a.to_string()).unwrap_or_else(|| "-".into())
        )
    }

    /// Graceful shutdown.
    pub async fn shutdown(&self) {
        self.endpoint
            .close(VarInt::from_u32(0), b"client shutdown");
        self.endpoint.wait_idle().await;
    }
}

/// Phase 8b (2026-07-14): which route won the LAN-first probe.
/// Returned from [`QuicClient::connect_lan_first`] so operators
/// (and telemetry) can see which side of the network was chosen
/// per connection.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ConnectRoute {
    Lan(SocketAddr),
    Tailscale(SocketAddr),
}

/// Open a bidi stream on `conn`, send `payload`, read back the peer's
/// response (bounded by [`MAX_MESSAGE_BYTES`]). This is the client
/// side of the ping RPC.
pub async fn ping(conn: &quinn::Connection, payload: &[u8]) -> Result<Vec<u8>> {
    if payload.len() > MAX_MESSAGE_BYTES {
        bail!(
            "ping payload {} bytes exceeds cap {}",
            payload.len(),
            MAX_MESSAGE_BYTES
        );
    }
    let (mut send, mut recv) = conn
        .open_bi()
        .await
        .context("opening bidi stream for ping")?;
    send.write_all(payload).await.context("writing ping payload")?;
    send.finish().context("finishing ping send stream")?;
    let response = recv
        .read_to_end(MAX_MESSAGE_BYTES)
        .await
        .context("reading ping response")?;
    Ok(response)
}

/// Server side of the ping RPC. Loops accepting bidi streams on `conn`
/// and echoing every payload back prefixed with `b"pong:"`. Runs until
/// the connection closes.
pub async fn ping_handler_loop(conn: quinn::Connection) -> Result<()> {
    loop {
        let (mut send, mut recv) = match conn.accept_bi().await {
            Ok(pair) => pair,
            Err(quinn::ConnectionError::ApplicationClosed(_))
            | Err(quinn::ConnectionError::ConnectionClosed(_))
            | Err(quinn::ConnectionError::LocallyClosed) => return Ok(()),
            Err(e) => return Err(anyhow::Error::from(e)),
        };
        let request = recv
            .read_to_end(MAX_MESSAGE_BYTES)
            .await
            .context("reading ping request")?;
        let mut response = b"pong:".to_vec();
        response.extend_from_slice(&request);
        send.write_all(&response)
            .await
            .context("writing pong response")?;
        send.finish().context("finishing pong send stream")?;
    }
}

/// Build the server-side rustls config: require client certs, chain to
/// our fleet CA, present our own leaf on handshake.
fn build_server_crypto(identity: &NodeIdentity) -> Result<rustls::ServerConfig> {
    let mut ca_store = rustls::RootCertStore::empty();
    ca_store
        .add(identity.trusted_ca.clone())
        .context("installing fleet CA in server root store")?;
    let client_verifier =
        rustls::server::WebPkiClientVerifier::builder(Arc::new(ca_store))
            .build()
            .context("building mTLS client verifier")?;
    let mut cfg = rustls::ServerConfig::builder()
        .with_client_cert_verifier(client_verifier)
        .with_single_cert(identity.cert_chain.clone(), identity.key.clone_key())
        .context("installing server leaf cert + key")?;
    cfg.alpn_protocols = vec![CLAWSTOR_RPC_ALPN.to_vec()];
    Ok(cfg)
}

/// Build the client-side rustls config: verify server cert chains to
/// our fleet CA, present our own leaf when the server challenges.
fn build_client_crypto(identity: &NodeIdentity) -> Result<rustls::ClientConfig> {
    let mut ca_store = rustls::RootCertStore::empty();
    ca_store
        .add(identity.trusted_ca.clone())
        .context("installing fleet CA in client root store")?;
    let mut cfg = rustls::ClientConfig::builder()
        .with_root_certificates(ca_store)
        .with_client_auth_cert(identity.cert_chain.clone(), identity.key.clone_key())
        .context("installing client leaf cert + key")?;
    cfg.alpn_protocols = vec![CLAWSTOR_RPC_ALPN.to_vec()];
    Ok(cfg)
}

/// Install the ring-backed CryptoProvider once per process. Called from
/// both server and client bootstrap. Idempotent — ignores the "already
/// installed" error so multiple endpoints in the same process (or tests)
/// coexist.
fn install_default_crypto_provider() {
    // `install_default` returns Err if a provider is already set — that's
    // fine; we just want SOME provider present when rustls builders run.
    let _ = rustls::crypto::ring::default_provider().install_default();
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::atomic::{AtomicU16, Ordering};

    /// Port allocator separate from the gossip tests (which use 41000+)
    /// so parallel test execution never conflicts.
    static NEXT_PORT: AtomicU16 = AtomicU16::new(42001);
    fn next_port() -> u16 {
        NEXT_PORT.fetch_add(1, Ordering::Relaxed)
    }

    fn loopback(port: u16) -> SocketAddr {
        format!("127.0.0.1:{port}").parse().unwrap()
    }

    fn wildcard(port: u16) -> SocketAddr {
        format!("0.0.0.0:{port}").parse().unwrap()
    }

    #[test]
    fn generate_test_pair_produces_two_distinct_leaves_sharing_a_ca() {
        let (a, b) = NodeIdentity::generate_test_pair("a", "b").unwrap();
        assert_ne!(
            a.cert_chain[0], b.cert_chain[0],
            "each node must have a distinct leaf"
        );
        assert_eq!(
            a.trusted_ca, b.trusted_ca,
            "both nodes must trust the same CA to talk"
        );
        assert!(!a.cert_chain.is_empty());
        assert!(!b.cert_chain.is_empty());
    }

    #[tokio::test]
    async fn ping_pong_between_two_mtls_peers() {
        let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();

        // B binds a server on an OS-assigned port.
        let server = QuicServer::bind(loopback(0), id_b).unwrap();
        let server_addr = server.local_addr().unwrap();

        // Spawn the accept loop: for every incoming connection, serve
        // the ping RPC.
        let accept_task = tokio::spawn(async move {
            if let Some(res) = server.accept().await {
                let conn = res.expect("accept succeeded");
                let _ = ping_handler_loop(conn).await;
            }
            server.shutdown().await;
        });

        // A connects as a client and calls ping.
        let client = QuicClient::new(loopback(0), id_a).unwrap();
        let conn = client
            .connect(server_addr, "b")
            .await
            .expect("A→B connect");
        let response = ping(&conn, b"hello").await.expect("ping");
        assert_eq!(response, b"pong:hello");

        // Cleanup.
        conn.close(VarInt::from_u32(0), b"done");
        client.shutdown().await;
        // Give the accept task a moment to finish, then abort to be safe.
        tokio::time::sleep(Duration::from_millis(50)).await;
        accept_task.abort();
    }

    #[tokio::test]
    async fn connect_lan_first_takes_lan_when_reachable() {
        // Live LAN server, valid tailscale would just be a decoy —
        // we should never dial it. Assert the returned route.
        let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();
        let server = QuicServer::bind(loopback(0), id_b).unwrap();
        let server_addr = server.local_addr().unwrap();
        let accept_task = tokio::spawn(async move {
            if let Some(res) = server.accept().await {
                let conn = res.expect("accept");
                let _ = ping_handler_loop(conn).await;
            }
            server.shutdown().await;
        });
        let client = QuicClient::new(loopback(0), id_a).unwrap();
        // Fake tailscale addr = a port nothing binds. Must not be
        // dialed since LAN succeeded first.
        let fake_ts: SocketAddr = "127.0.0.1:1".parse().unwrap();
        let (conn, route) = client
            .connect_lan_first(
                "b",
                Some(server_addr),
                Some(fake_ts),
                Duration::from_secs(2),
            )
            .await
            .expect("connect_lan_first");
        assert!(matches!(route, ConnectRoute::Lan(a) if a == server_addr));
        let response = ping(&conn, b"hi").await.unwrap();
        assert_eq!(response, b"pong:hi");
        conn.close(VarInt::from_u32(0), b"done");
        client.shutdown().await;
        tokio::time::sleep(Duration::from_millis(50)).await;
        accept_task.abort();
    }

    #[tokio::test]
    async fn connect_lan_first_falls_through_to_tailscale_on_lan_deadline() {
        // LAN addr is a black hole (drops SYNs). Probe budget = 100ms.
        // Tailscale addr = real server. Must fall through and pick
        // the tailscale route.
        let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();
        let server = QuicServer::bind(loopback(0), id_b).unwrap();
        let server_addr = server.local_addr().unwrap();
        let accept_task = tokio::spawn(async move {
            if let Some(res) = server.accept().await {
                let conn = res.expect("accept");
                let _ = ping_handler_loop(conn).await;
            }
            server.shutdown().await;
        });
        // 240.x.x.x is RFC1112 unroutable — SYN just times out.
        // We don't need a live server; the deadline must fire on
        // its own.
        let black_hole: SocketAddr = "240.0.0.1:1".parse().unwrap();
        let client = QuicClient::new(loopback(0), id_a).unwrap();
        let started = std::time::Instant::now();
        let (conn, route) = client
            .connect_lan_first(
                "b",
                Some(black_hole),
                Some(server_addr),
                Duration::from_millis(150),
            )
            .await
            .expect("connect_lan_first");
        assert!(matches!(route, ConnectRoute::Tailscale(a) if a == server_addr));
        // Sanity: we shouldn't have waited far beyond the probe
        // budget before starting the tailscale attempt.
        assert!(
            started.elapsed() < Duration::from_secs(3),
            "took too long: {:?}",
            started.elapsed()
        );
        conn.close(VarInt::from_u32(0), b"done");
        client.shutdown().await;
        tokio::time::sleep(Duration::from_millis(50)).await;
        accept_task.abort();
    }

    #[tokio::test]
    async fn connect_lan_first_lan_only_ignores_probe_deadline() {
        // Phase 8c hotfix: when no fallback exists, LAN dial gets
        // unlimited time — otherwise a slow-but-fine handshake
        // fails a caller that never opted into a fallback path.
        let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();
        let server = QuicServer::bind(loopback(0), id_b).unwrap();
        let server_addr = server.local_addr().unwrap();
        let accept_task = tokio::spawn(async move {
            if let Some(res) = server.accept().await {
                let conn = res.expect("accept");
                let _ = ping_handler_loop(conn).await;
            }
            server.shutdown().await;
        });
        let client = QuicClient::new(loopback(0), id_a).unwrap();
        // Absurdly short probe. Would fail if the deadline applied.
        let (conn, route) = client
            .connect_lan_first("b", Some(server_addr), None, Duration::from_nanos(1))
            .await
            .expect("connect_lan_first with no fallback ignores deadline");
        assert!(matches!(route, ConnectRoute::Lan(a) if a == server_addr));
        conn.close(VarInt::from_u32(0), b"done");
        client.shutdown().await;
        tokio::time::sleep(Duration::from_millis(50)).await;
        accept_task.abort();
    }

    #[tokio::test]
    async fn connect_lan_first_errors_when_both_addrs_absent() {
        let (id_a, _id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();
        let client = QuicClient::new(loopback(0), id_a).unwrap();
        let err = client
            .connect_lan_first("b", None, None, Duration::from_millis(50))
            .await
            .unwrap_err();
        assert!(err.to_string().contains("no reachable address"));
        client.shutdown().await;
    }

    #[tokio::test]
    async fn client_rejects_peer_with_wrong_ca() {
        // Two CAs, A and X. A pair (id_a, id_b) share CA_A; a rogue id_x
        // has its own CA_X. When A tries to talk to X, TLS handshake
        // must fail — X's leaf doesn't chain to CA_A.
        let (id_a, _id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();
        let (_id_x_peer, id_x) = NodeIdentity::generate_test_pair("x-peer", "x").unwrap();

        let server = QuicServer::bind(loopback(0), id_x).unwrap();
        let server_addr = server.local_addr().unwrap();

        let _accept_task = tokio::spawn(async move {
            // Server will fail the handshake because the client's cert
            // doesn't chain to CA_X. We just want to be here to accept
            // the SYN; the handshake failure is what the client asserts.
            let _ = server.accept().await;
        });

        let client = QuicClient::new(loopback(0), id_a).unwrap();
        let err = client
            .connect(server_addr, "x")
            .await
            .err()
            .expect("connect must fail — wrong CA");
        let msg = format!("{err:#}");
        // The exact error text varies by rustls version; look for the
        // token that always appears in a chain-verification failure.
        assert!(
            msg.to_lowercase().contains("certificate")
                || msg.to_lowercase().contains("verify")
                || msg.to_lowercase().contains("closed")
                || msg.to_lowercase().contains("handshake"),
            "unexpected error text: {msg}"
        );
    }

    #[tokio::test]
    async fn client_rejects_wrong_server_name() {
        // A and B share a CA; B's cert has SAN=["b"]. A connects
        // asking for server_name="c" — rustls must reject on SAN
        // mismatch even though the cert chain is valid.
        let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();
        let server = QuicServer::bind(loopback(0), id_b).unwrap();
        let server_addr = server.local_addr().unwrap();

        let _accept_task = tokio::spawn(async move {
            let _ = server.accept().await;
        });

        let client = QuicClient::new(loopback(0), id_a).unwrap();
        let err = client
            .connect(server_addr, "c")
            .await
            .err()
            .expect("connect must fail — SAN mismatch");
        let msg = format!("{err:#}").to_lowercase();
        assert!(
            msg.contains("name") || msg.contains("certificate") || msg.contains("handshake"),
            "unexpected error text: {msg}"
        );
    }

    #[tokio::test]
    async fn server_binds_wildcard_and_reports_concrete_local_addr() {
        let (_id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();
        let server = QuicServer::bind(wildcard(0), id_b).unwrap();
        let addr = server.local_addr().unwrap();
        assert_ne!(addr.port(), 0, "OS must assign a real port");
        server.shutdown().await;
    }

    #[tokio::test]
    async fn ping_rejects_oversize_payload() {
        let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();
        let server = QuicServer::bind(loopback(0), id_b).unwrap();
        let server_addr = server.local_addr().unwrap();
        let _accept = tokio::spawn(async move {
            if let Some(Ok(conn)) = server.accept().await {
                let _ = ping_handler_loop(conn).await;
            }
        });
        let client = QuicClient::new(loopback(0), id_a).unwrap();
        let conn = client.connect(server_addr, "b").await.unwrap();
        let oversized = vec![0u8; MAX_MESSAGE_BYTES + 1];
        let err = ping(&conn, &oversized)
            .await
            .err()
            .expect("must reject oversize payload before send");
        assert!(err.to_string().contains("exceeds cap"));
        conn.close(VarInt::from_u32(0), b"done");
        client.shutdown().await;
    }

    // ── Phase 1d: persistent identity + FleetCa ────────────────────────

    #[test]
    fn fleet_ca_rejects_empty_common_name() {
        let err = FleetCa::generate("").unwrap_err().to_string();
        assert!(err.contains("common name cannot be empty"));
    }

    #[test]
    fn fleet_ca_save_and_load_round_trip_preserves_signing() {
        // Generate a CA, save it, wipe it from memory, reload, sign a
        // leaf, verify that leaf still validates against the reloaded
        // CA's advertised trust root.
        let tmp = tempfile::TempDir::new().unwrap();
        let dir = tmp.path();

        let ca = FleetCa::generate("clawstor test CA").unwrap();
        ca.save(dir).unwrap();
        assert!(dir.join("ca.crt").exists(), "ca.crt should be written");
        assert!(dir.join("ca.key").exists(), "ca.key should be written");

        // Confirm the CA private key file is chmod 0o600 on Unix — leaking
        // it would let anyone forge fleet identities.
        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            let key_mode = std::fs::metadata(dir.join("ca.key"))
                .unwrap()
                .permissions()
                .mode();
            assert_eq!(
                key_mode & 0o777,
                0o600,
                "ca.key must be mode 0o600, got {:o}",
                key_mode & 0o777
            );
        }

        drop(ca);
        let ca_reloaded = FleetCa::load(dir).unwrap();
        // The reloaded CA must produce leaves whose `trusted_ca` field
        // matches what was originally written to disk.
        let ident = ca_reloaded.sign_leaf("test-node").unwrap();
        let expected_ca_der = read_single_cert(&dir.join("ca.crt")).unwrap();
        assert_eq!(
            ident.trusted_ca, expected_ca_der,
            "leaf must carry the same CA DER as on disk"
        );
    }

    #[test]
    fn fleet_ca_load_errors_when_files_missing() {
        let tmp = tempfile::TempDir::new().unwrap();
        let err = FleetCa::load(tmp.path()).unwrap_err().to_string();
        // Either the .crt or .key path — we don't over-specify which
        // one fails first, just that a filesystem/parse error surfaces.
        assert!(
            err.contains("ca.crt") || err.contains("ca.key"),
            "expected ca.crt/ca.key in error, got: {err}"
        );
    }

    #[test]
    fn sign_leaf_to_pem_writes_all_three_files_with_correct_permissions() {
        let tmp = tempfile::TempDir::new().unwrap();
        let ca = FleetCa::generate("clawstor test CA").unwrap();
        let node_dir = tmp.path().join("nodes/architect");
        ca.sign_leaf_to_pem("architect", &node_dir).unwrap();
        for name in &["ca.crt", "node.crt", "node.key"] {
            assert!(
                node_dir.join(name).exists(),
                "{name} must be written"
            );
        }
        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            let mode = std::fs::metadata(node_dir.join("node.key"))
                .unwrap()
                .permissions()
                .mode();
            assert_eq!(mode & 0o777, 0o600, "node.key must be 0o600");
        }
    }

    #[tokio::test]
    async fn persistent_identity_round_trips_through_disk_and_pings() {
        // End-to-end proof: cut a CA on disk, sign leaves for two nodes
        // via `sign_leaf_to_pem`, reload both via `NodeIdentity::from_pem_dir`,
        // run a real QUIC ping/pong between them. This is the flow a
        // production operator follows:
        //   1. `fleet-ca init --dir /etc/claw-store/tls`  (server side)
        //   2. `fleet-ca sign --node architect --out-dir …` (per node)
        //   3. daemon reads the resulting dir and starts RPC
        let tmp = tempfile::TempDir::new().unwrap();
        let ca_dir = tmp.path().join("ca");
        let arch_dir = tmp.path().join("architect-tls");
        let tank_dir = tmp.path().join("tank-tls");

        let ca = FleetCa::generate("clawstor test CA").unwrap();
        ca.save(&ca_dir).unwrap();

        // Reload the CA from disk before signing — proves save/load
        // preserves signing capability, not just the in-memory instance.
        let ca_reloaded = FleetCa::load(&ca_dir).unwrap();
        ca_reloaded
            .sign_leaf_to_pem("architect", &arch_dir)
            .unwrap();
        ca_reloaded.sign_leaf_to_pem("tank", &tank_dir).unwrap();

        let id_arch = NodeIdentity::from_pem_dir(&arch_dir).unwrap();
        let id_tank = NodeIdentity::from_pem_dir(&tank_dir).unwrap();

        let server = QuicServer::bind(loopback(0), id_tank).unwrap();
        let server_addr = server.local_addr().unwrap();
        let accept_task = tokio::spawn(async move {
            if let Some(Ok(conn)) = server.accept().await {
                let _ = ping_handler_loop(conn).await;
            }
        });

        let client = QuicClient::new(loopback(0), id_arch).unwrap();
        let conn = client
            .connect(server_addr, "tank")
            .await
            .expect("architect → tank ping using persisted mTLS");
        let response = ping(&conn, b"persistent hello").await.unwrap();
        assert_eq!(response, b"pong:persistent hello");

        conn.close(VarInt::from_u32(0), b"done");
        client.shutdown().await;
        tokio::time::sleep(Duration::from_millis(50)).await;
        accept_task.abort();
    }

    #[test]
    fn node_identity_from_cluster_config_errors_without_tls_section() {
        use crate::config::ClusterConfig;
        let cfg = ClusterConfig {
            zone: "test".into(),
            bind_lan: Some(loopback(next_port())),
            tls: None,
            ..Default::default()
        };
        let err = NodeIdentity::from_cluster_config(&cfg)
            .unwrap_err()
            .to_string();
        assert!(err.contains("cluster.tls"), "err: {err}");
    }

    #[test]
    fn node_identity_from_cluster_config_loads_pem_paths() {
        let tmp = tempfile::TempDir::new().unwrap();
        let ca = FleetCa::generate("cfg test CA").unwrap();
        let node_dir = tmp.path().join("node-tls");
        ca.sign_leaf_to_pem("cfg-test", &node_dir).unwrap();

        use crate::config::{ClusterConfig, ClusterTlsConfig};
        let cfg = ClusterConfig {
            zone: "test".into(),
            bind_lan: Some(loopback(next_port())),
            tls: Some(ClusterTlsConfig {
                ca_cert: node_dir.join("ca.crt"),
                node_cert: node_dir.join("node.crt"),
                node_key: node_dir.join("node.key"),
            }),
            ..Default::default()
        };
        let ident = NodeIdentity::from_cluster_config(&cfg).unwrap();
        assert_eq!(ident.cert_chain.len(), 1);
        assert!(!ident.trusted_ca.is_empty());
    }

    #[test]
    fn from_pem_files_errors_on_missing_ca_file() {
        let tmp = tempfile::TempDir::new().unwrap();
        let ca = FleetCa::generate("test").unwrap();
        let node_dir = tmp.path().join("n");
        ca.sign_leaf_to_pem("n", &node_dir).unwrap();
        // Delete the CA cert to simulate a broken deployment.
        std::fs::remove_file(node_dir.join("ca.crt")).unwrap();
        let err = NodeIdentity::from_pem_dir(&node_dir).unwrap_err().to_string();
        assert!(err.contains("CA cert") || err.contains("ca.crt"), "err: {err}");
    }
}