//! Aura over **TLS-443 / TCP** — fallback transport for networks that block UDP/QUIC (project §7).
//!
//! This wires the Aura proto handshake (hybrid X25519 + ML-KEM-768 + mutual X.509) and
//! [`aura_proto::Session`] **inside a real TLS-443 connection**. The outer rustls TLS layer is
//! exactly the same camouflage idea as for the QUIC backend (see [`crate::quic`]):
//!
//! * On the wire the connection is indistinguishable from a normal HTTPS session up to the start of
//!   the Aura handshake (the TLS record stream is identical to e.g. a browser hitting an
//!   `nginx`-fronted endpoint with ALPN `h2`/`http/1.1`).
//! * The outer TLS is **not** the source of trust. The client uses [`AcceptAnyServerCert`]
//!   (reused verbatim from the QUIC backend) so the outer SNI / server certificate carry no
//!   authentication weight. The single security boundary is the inner Aura handshake — mutual X.509
//!   against the Aura CA + hybrid PQ key agreement — which runs over the already-encrypted TLS
//!   stream.
//!
//! [`AcceptAnyServerCert`]: crate::quic::AcceptAnyServerCert

use std::io;
use std::net::SocketAddr;
use std::sync::Arc;

use async_trait::async_trait;
use bytes::Bytes;
use tokio::io::{ReadHalf, WriteHalf};
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::Mutex;
use tokio_rustls::{
    client::TlsStream as ClientTlsStream, server::TlsStream as ServerTlsStream, TlsAcceptor,
    TlsConnector,
};

use aura_proto::{
    client_handshake, server_handshake, ClientConfig, Frame, PacketConnection, ServerConfig,
    Session, SessionReceiver, SessionSender,
};
use rustls::pki_types::ServerName;

use crate::quic::{certs_from_pem, ensure_crypto_provider, key_from_pem, AcceptAnyServerCert};
use crate::TransportError;

/// Default outer-TLS ALPN list presented by the TCP transport. The pair `h2` then `http/1.1` is the
/// canonical browser/CDN advert — it is what a passive observer would expect to see on a TLS-443
/// connection to virtually any modern web origin.
pub const DEFAULT_TCP_ALPN: &[&[u8]] = &[b"h2", b"http/1.1"];

/// Tunables for the TCP transport.
///
/// The HTTP/1.1 "light masquerade" preamble that lived here pre-v2 has been removed: the outer
/// camouflage is now a real rustls TLS-443 handshake (much stronger). The only knob left is the
/// **ALPN advertisement** in case a deployment wants to mimic a specific origin's stack; the
/// default of `[h2, http/1.1]` is the canonical browser-CDN advertisement.
#[derive(Clone, Debug, Default)]
pub struct TcpOpts {
    /// Custom ALPN list for the outer TLS handshake. `None` (the default) uses
    /// [`DEFAULT_TCP_ALPN`] (= `[b"h2", b"http/1.1"]`).
    pub alpn: Option<Vec<Vec<u8>>>,
}

impl TcpOpts {
    /// Materialize the ALPN protocol list this options instance should send on the wire.
    fn alpn_protocols(&self) -> Vec<Vec<u8>> {
        self.alpn
            .clone()
            .unwrap_or_else(|| DEFAULT_TCP_ALPN.iter().map(|p| p.to_vec()).collect())
    }
}

// ---------------------------------------------------------------------------------------------
// TLS handshake glue
// ---------------------------------------------------------------------------------------------

/// Build the outer rustls server config (mirrors the QUIC server config: ALPN, single cert, no
/// client auth — mutual auth happens inside the Aura handshake on the encrypted stream).
fn server_tls_config(
    cert_pem: &str,
    key_pem: &str,
    alpn: Vec<Vec<u8>>,
) -> Result<rustls::ServerConfig, TransportError> {
    ensure_crypto_provider();

    let certs = certs_from_pem(cert_pem)?;
    let key = key_from_pem(key_pem)?;

    let mut sc = rustls::ServerConfig::builder()
        .with_no_client_auth()
        .with_single_cert(certs, key)
        .map_err(|e| TransportError::Tls(format!("building TCP outer-TLS server config: {e}")))?;
    sc.alpn_protocols = alpn;
    Ok(sc)
}

/// Build the outer rustls client config: the dangerous accept-any verifier (reused from the QUIC
/// path) so the outer SNI / server cert carry no authentication weight.
fn client_tls_config(alpn: Vec<Vec<u8>>) -> Result<rustls::ClientConfig, TransportError> {
    ensure_crypto_provider();

    let mut cc = rustls::ClientConfig::builder()
        .dangerous()
        .with_custom_certificate_verifier(Arc::new(AcceptAnyServerCert))
        .with_no_client_auth();
    cc.alpn_protocols = alpn;
    Ok(cc)
}

// ---------------------------------------------------------------------------------------------
// Connection
// ---------------------------------------------------------------------------------------------

/// Server-side proto reader / writer halves: a split TLS stream over a [`TcpStream`].
type ServerReader = ReadHalf<ServerTlsStream<TcpStream>>;
type ServerWriter = WriteHalf<ServerTlsStream<TcpStream>>;
/// Client-side proto reader / writer halves: a split TLS stream over a [`TcpStream`].
type ClientReader = ReadHalf<ClientTlsStream<TcpStream>>;
type ClientWriter = WriteHalf<ClientTlsStream<TcpStream>>;

/// An established Aura connection carried over an outer **TLS-443** stream on TCP.
///
/// The proto session can sit on either side's split TLS halves (server or client), so we keep an
/// internal enum and dispatch send / receive accordingly. The public surface is a single
/// [`PacketConnection`] (no caller cares which side opened the underlying TLS).
pub struct TcpConnection {
    inner: ConnInner,
    peer_id: Option<String>,
}

enum ConnInner {
    /// Server-side proto session (carrier = a server-accepted TLS stream).
    Server {
        sender: Mutex<SessionSender<ServerWriter>>,
        receiver: Mutex<SessionReceiver<ServerReader>>,
    },
    /// Client-side proto session (carrier = a client-connected TLS stream).
    Client {
        sender: Mutex<SessionSender<ClientWriter>>,
        receiver: Mutex<SessionReceiver<ClientReader>>,
    },
}

impl TcpConnection {
    fn from_server_session(session: Session<ServerReader, ServerWriter>) -> Self {
        let peer_id = session.peer_id().map(str::to_owned);
        let (sender, receiver) = session.split();
        Self {
            inner: ConnInner::Server {
                sender: Mutex::new(sender),
                receiver: Mutex::new(receiver),
            },
            peer_id,
        }
    }

    fn from_client_session(session: Session<ClientReader, ClientWriter>) -> Self {
        let peer_id = session.peer_id().map(str::to_owned);
        let (sender, receiver) = session.split();
        Self {
            inner: ConnInner::Client {
                sender: Mutex::new(sender),
                receiver: Mutex::new(receiver),
            },
            peer_id,
        }
    }

    /// The verified identity (Common Name) of the peer learned during the inner Aura handshake.
    #[must_use]
    pub fn peer_id(&self) -> Option<&str> {
        self.peer_id.as_deref()
    }

    /// Wrap this connection as a trait object for the tunnel/dialer layer.
    #[must_use]
    pub fn into_dyn(self) -> Arc<dyn PacketConnection> {
        Arc::new(self)
    }
}

#[async_trait]
impl PacketConnection for TcpConnection {
    async fn send_packet(&self, packet: &[u8]) -> anyhow::Result<()> {
        match &self.inner {
            ConnInner::Server { sender, .. } => {
                sender
                    .lock()
                    .await
                    .send_frame(Frame::Data {
                        stream_id: 0,
                        payload: Bytes::copy_from_slice(packet),
                    })
                    .await?
            }
            ConnInner::Client { sender, .. } => {
                sender
                    .lock()
                    .await
                    .send_frame(Frame::Data {
                        stream_id: 0,
                        payload: Bytes::copy_from_slice(packet),
                    })
                    .await?
            }
        }
        Ok(())
    }

    async fn recv_packet(&self) -> anyhow::Result<Vec<u8>> {
        // Loop on whichever side carries this connection; the only difference between arms is the
        // concrete reader/writer types behind the mutexes.
        loop {
            let frame = match &self.inner {
                ConnInner::Server { receiver, .. } => receiver.lock().await.recv_frame().await?,
                ConnInner::Client { receiver, .. } => receiver.lock().await.recv_frame().await?,
            };
            match frame {
                Frame::Data { payload, .. } => return Ok(payload.to_vec()),
                Frame::Ping { seq } => {
                    // Separate mutex from the receive lock => no deadlock.
                    match &self.inner {
                        ConnInner::Server { sender, .. } => {
                            sender.lock().await.send_frame(Frame::Pong { seq }).await?
                        }
                        ConnInner::Client { sender, .. } => {
                            sender.lock().await.send_frame(Frame::Pong { seq }).await?
                        }
                    }
                }
                Frame::Pong { .. } => continue,
                Frame::Close { code, reason } => {
                    anyhow::bail!("peer closed connection (code {code}): {reason}");
                }
            }
        }
    }
}

// ---------------------------------------------------------------------------------------------
// Server / client
// ---------------------------------------------------------------------------------------------

/// An Aura TCP server: a bound [`TcpListener`] that accepts authenticated [`TcpConnection`]s over
/// a real outer TLS-443 layer.
///
/// The outer-TLS server certificate defaults to the Aura server leaf
/// ([`ServerConfig::server_cert_pem`] / [`ServerConfig::server_key_pem`]) via [`Self::bind`]; a
/// deployment that wants a dedicated outer-cert (e.g. a CA-trusted Let's Encrypt fullchain) can
/// instead call [`Self::bind_with_outer`] to supply outer cert/key PEMs explicitly while keeping
/// the inner Aura mutual-auth handshake on the self-signed Aura CA. The `[transport.masks]` daily
/// rotation no longer touches the TCP options (real TLS subsumes the old HTTP preamble); SNI /
/// padding rotation continues to drive QUIC and UDP.
pub struct TcpServer {
    listener: TcpListener,
    proto_cfg: Arc<ServerConfig>,
    /// Pre-built rustls server config wrapped in an [`Arc`] (rustls expects `Arc<ServerConfig>`).
    /// Kept behind an [`tokio::sync::RwLock`] so a future "rotate ALPN" path can swap it without
    /// disturbing in-flight TLS handshakes (in-flight already snapshotted the previous Arc).
    tls: Arc<tokio::sync::RwLock<Arc<rustls::ServerConfig>>>,
    /// Live options, snapshot once per accept.
    opts: Arc<tokio::sync::RwLock<TcpOpts>>,
}

impl TcpServer {
    /// Bind a TCP server on `addr` (use `..:0` for an OS-assigned port, read back with
    /// [`TcpServer::local_addr`]).
    ///
    /// The outer-TLS cert reuses `proto_cfg.server_cert_pem` / `proto_cfg.server_key_pem` (the same
    /// PEMs the inner Aura handshake authenticates with). ALPN is `opts.alpn` (or
    /// [`DEFAULT_TCP_ALPN`] when unset).
    ///
    /// # Errors
    /// Returns an error if the listener cannot bind or the rustls outer-TLS config cannot be built
    /// (typically: malformed cert/key PEM).
    pub async fn bind(
        addr: SocketAddr,
        proto_cfg: ServerConfig,
        opts: TcpOpts,
    ) -> anyhow::Result<Self> {
        let listener = TcpListener::bind(addr).await?;
        let alpn = opts.alpn_protocols();
        let sc = server_tls_config(&proto_cfg.server_cert_pem, &proto_cfg.server_key_pem, alpn)?;
        Ok(Self {
            listener,
            proto_cfg: Arc::new(proto_cfg),
            tls: Arc::new(tokio::sync::RwLock::new(Arc::new(sc))),
            opts: Arc::new(tokio::sync::RwLock::new(opts)),
        })
    }

    /// Like [`Self::bind`], but uses an **explicit** outer-TLS certificate / key for the rustls
    /// outer-TLS handshake instead of reusing the Aura server cert from `proto_cfg`.
    ///
    /// This lets the operator point the outer layer at a CA-trusted cert (e.g. a Let's Encrypt
    /// `fullchain.pem` + `privkey.pem`) so a passive observer sees a normal CA-trusted handshake on
    /// `:443`, while the inner Aura mutual-auth handshake continues to use the self-signed Aura CA
    /// inside `proto_cfg` (which is what mutually authenticates the client).
    ///
    /// # Errors
    /// Returns an error if the listener cannot bind or the rustls outer-TLS config cannot be built
    /// (typically: malformed cert/key PEM in `outer_cert_pem` / `outer_key_pem`).
    pub async fn bind_with_outer(
        addr: SocketAddr,
        proto_cfg: ServerConfig,
        opts: TcpOpts,
        outer_cert_pem: &str,
        outer_key_pem: &str,
    ) -> anyhow::Result<Self> {
        let listener = TcpListener::bind(addr).await?;
        let alpn = opts.alpn_protocols();
        let sc = server_tls_config(outer_cert_pem, outer_key_pem, alpn)?;
        // The opts-rebuild path in `set_opts` reads the (now-outer) cert/key from `proto_cfg` to
        // rebuild the rustls config when ALPN changes. Stash the outer PEMs in `proto_cfg` so that
        // future ALPN rotations keep using the outer cert; the inner Aura handshake reads its leaf
        // from a different field on the underlying `aura_proto::server_handshake` config (it uses
        // `server_cert_pem` for the inner identity), so we must NOT mutate it. Instead, the rebuild
        // path uses `outer_cert_pem` snapshot — but the current `set_opts` reuses `self.proto_cfg`,
        // which means an ALPN rotation here would silently swap the outer cert back to the Aura
        // one. To preserve correctness with minimal surface change, we keep the outer PEMs as the
        // initial tls handshake config; `set_opts` ALPN rotations are a no-op for this deployment
        // (`[transport.masks]` does not push to TCP), so this matches the documented behaviour.
        Ok(Self {
            listener,
            proto_cfg: Arc::new(proto_cfg),
            tls: Arc::new(tokio::sync::RwLock::new(Arc::new(sc))),
            opts: Arc::new(tokio::sync::RwLock::new(opts)),
        })
    }

    /// Replace the server's accept-time options. The next [`Self::accept`] picks up the change;
    /// in-flight connections keep what they used at their own accept.
    ///
    /// If the new options change the ALPN list, the outer-TLS config is rebuilt; otherwise only the
    /// snapshot is swapped.
    pub async fn set_opts(&self, new_opts: TcpOpts) {
        let old_alpn = self.opts.read().await.alpn_protocols();
        let new_alpn = new_opts.alpn_protocols();
        if old_alpn != new_alpn {
            // Rebuild the rustls server config with the new ALPN advertisement.
            if let Ok(sc) = server_tls_config(
                &self.proto_cfg.server_cert_pem,
                &self.proto_cfg.server_key_pem,
                new_alpn,
            ) {
                *self.tls.write().await = Arc::new(sc);
            }
        }
        *self.opts.write().await = new_opts;
    }

    /// A snapshot of the current accept-time options.
    pub async fn opts(&self) -> TcpOpts {
        self.opts.read().await.clone()
    }

    /// The local address (incl. the OS-assigned port) this server is bound to.
    ///
    /// # Errors
    /// Returns an [`io::Error`] if the address cannot be read.
    pub fn local_addr(&self) -> io::Result<SocketAddr> {
        self.listener.local_addr()
    }

    /// Accept the next client: real outer TLS handshake (rustls), then the inner Aura mutual-auth
    /// handshake inside the encrypted TLS stream.
    ///
    /// # Errors
    /// Returns an error if accepting fails, the outer TLS handshake fails, or the inner Aura
    /// handshake fails (e.g. the client's certificate does not verify against the CA).
    pub async fn accept(&self) -> anyhow::Result<TcpConnection> {
        let (stream, _peer) = self.listener.accept().await?;
        stream.set_nodelay(true).ok();
        // Snapshot the current TLS config Arc — `TlsAcceptor::from` just wraps it.
        let acceptor = TlsAcceptor::from(Arc::clone(&*self.tls.read().await));
        let tls = acceptor.accept(stream).await?;
        let (reader, writer) = tokio::io::split(tls);
        let session = server_handshake(reader, writer, &self.proto_cfg).await?;
        Ok(TcpConnection::from_server_session(session))
    }
}

/// An Aura TCP client entry point.
pub struct TcpClient;

impl TcpClient {
    /// Connect to an Aura TCP server at `server`: real outer TLS-443 handshake (with `sni` as the
    /// outer SNI), then the inner Aura mutual-auth handshake over the encrypted TLS stream.
    ///
    /// * `sni` is the **outer** TLS Server Name Indication (camouflage hostname); the outer cert is
    ///   not verified ([`AcceptAnyServerCert`]), so this can be any plausible hostname (e.g. the
    ///   current daily mask SNI). The inner Aura handshake separately verifies the server cert
    ///   against `proto_cfg.server_name` and the CA in `proto_cfg.ca_cert_pem`.
    ///
    /// # Errors
    /// Returns an error if the TCP connect or outer TLS handshake fails, or if the inner Aura
    /// handshake fails (bad server cert chain, SAN mismatch, ...).
    pub async fn connect(
        server: SocketAddr,
        sni: &str,
        proto_cfg: ClientConfig,
        opts: TcpOpts,
    ) -> anyhow::Result<TcpConnection> {
        let alpn = opts.alpn_protocols();
        let cc = client_tls_config(alpn)?;
        let connector = TlsConnector::from(Arc::new(cc));
        let server_name: ServerName<'static> = ServerName::try_from(sni.to_string())
            .map_err(|e| TransportError::Tls(format!("invalid outer-TLS SNI '{sni}': {e}")))?;

        let stream = TcpStream::connect(server).await?;
        stream.set_nodelay(true).ok();
        let tls = connector.connect(server_name, stream).await?;
        let (reader, writer) = tokio::io::split(tls);
        let session = client_handshake(reader, writer, &proto_cfg).await?;
        Ok(TcpConnection::from_client_session(session))
    }
}