//! Client connector to the centralized `huddle-server`.
//!
//! huddle's primary transport is libp2p (mDNS on the LAN, gossipsub
//! across direct/relayed connections). This module adds a *second* path:
//! a WebSocket to a single canonical server that the operator hosts. The
//! server is reachable only as a **Tor v3 onion**, so `.onion` URLs are
//! dialed through Tor's local SOCKS5 proxy; plain `ws://host:port` URLs
//! (used in tests) are dialed directly.
//!
//! The server is a dumb ciphertext mover: we hand it the same opaque
//! huddle wire bytes we would have published on a gossipsub topic,
//! tagged with the cleartext `room` id, base64-encoded. It fans them out
//! to the room's other members and queues them for offline ones. All
//! encryption/authentication stays in the layers above — this module
//! never inspects the payload.

use std::sync::Arc;

use base64::engine::general_purpose::STANDARD as B64;
use base64::Engine;
use futures::{SinkExt, StreamExt};
use tokio::sync::mpsc;
use tokio_tungstenite::tungstenite::Message as WsMessage;
use tokio_tungstenite::WebSocketStream;
use tracing::warn;

use crate::error::{HuddleError, Result};
use crate::identity::{relay_auth_msg, Identity};
// huddle 2.0.4 (WS1.1): the relay control messages are now the single shared
// definition in `huddle_protocol::relay`, so the client and relay can't drift.
// We serialize `ClientMsg` (send) and deserialize `ServerMsg` (receive).
use huddle_protocol::relay::{ClientMsg, ServerMsg};

/// huddle 1.3.4: WebSocket frame/message ceiling for the relay **client**.
/// The relay was hardened in 1.3.1 to cap inbound frames at 512 KiB, but the
/// client connecting *to* a relay used tungstenite's 64 MiB default, so a
/// malicious (or compromised) relay could blast 64 MiB frames at it. Mirror
/// the server's 512 KiB ceiling so the WS layer is the real per-frame bound.
const RELAY_WS_FRAME_CAP: usize = 512 * 1024;

/// huddle 1.3.4: post-parse guard on a `ServerMsg::Message` payload, matching
/// the relay's own `MAX_PAYLOAD_B64`. Even under the frame cap, reject an
/// over-large base64 blob before allocating its decoded bytes.
const MAX_PAYLOAD_B64: usize = 256 * 1024;

/// huddle 1.3.4: the relay's auth nonce is a fixed 32 bytes (~44 base64
/// chars). Cap the accepted `nonce_b64` length generously so a hostile relay
/// can't make us allocate/concatenate a huge "nonce" in `relay_auth_msg`.
const MAX_NONCE_B64: usize = 256;

/// huddle 1.3.4: cap on app messages buffered before the authenticated `Hello`
/// goes out. A relay that accepts the socket but never sends its `Challenge`
/// (the server still closes it after its 20 s pre-auth timeout) can't make the
/// writer's pre-auth `pending` queue grow without bound in the meantime.
const MAX_PENDING_PREAUTH: usize = 1024;
/// huddle 2.0.3 (audit N-M7): how often to send an application keepalive
/// (`ClientMsg::Ping`) once authenticated. The relay reaps a socket that sends
/// no frame for ~150 s, so a quiet-but-healthy connection must speak up well
/// inside that window; 60 s gives a comfortable margin.
const RELAY_KEEPALIVE_SECS: u64 = 60;

/// Shared WebSocket config (frame caps) for every relay-client door.
fn relay_ws_config() -> tokio_tungstenite::tungstenite::protocol::WebSocketConfig {
    tokio_tungstenite::tungstenite::protocol::WebSocketConfig {
        max_message_size: Some(RELAY_WS_FRAME_CAP),
        max_frame_size: Some(RELAY_WS_FRAME_CAP),
        ..Default::default()
    }
}

// huddle 2.0.4 (WS1.1): `ClientMsg` / `ServerMsg` are now the single shared
// definition in `huddle_protocol::relay` (imported above), so the client and
// relay can never drift. We serialize `ClientMsg` (send) and deserialize
// `ServerMsg` (receive); the unified types are byte-compatible with every prior
// client and relay (the relay's historical serde attributes are authoritative).

/// What the connector surfaces to the rest of huddle-core. The caller
/// drives these into the same path that handles a received gossipsub
/// message (decode → decrypt → `AppEvent`).
#[derive(Debug, Clone)]
pub enum ServerEvent {
    /// Handshake complete; the mailbox (if any) will follow as `Message`s.
    Ready,
    /// Delivery receipt for one of our `publish` calls: how many of the
    /// room's other members received it live vs. were queued because they
    /// were offline. Lets the UI mark a message delivered/pending.
    Sent {
        id: String,
        delivered: usize,
        queued: usize,
    },
    /// A room message delivered (live or from the offline mailbox). huddle 2.0:
    /// `mailbox_id` is `Some(row_id)` when a 2.0 relay used at-least-once
    /// delivery and the consumer should `ack_mailbox(row_id)` after durably
    /// handling the message; `None` when delivered live or by a pre-2.0 relay
    /// (classical delete-on-deliver — no ACK needed).
    Message {
        room: String,
        id: String,
        payload: Vec<u8>,
        mailbox_id: Option<i64>,
    },
    /// huddle 1.2.1: the relay minted a connect code for us (with its TTL).
    ConnectToken { token: String, ttl_secs: u64 },
    /// huddle 1.2.1: the relay resolved a connect code we redeemed.
    /// `fingerprint`/`pubkey` are `None` when the code was unknown or expired.
    ConnectTokenResolved {
        fingerprint: Option<String>,
        pubkey_b64: Option<String>,
    },
    /// The socket closed; the caller may choose to reconnect.
    Disconnected,
}

/// A live connection to the server. Cloneable handle; cloning shares the
/// same underlying socket.
#[derive(Clone)]
pub struct ServerClient {
    out_tx: mpsc::UnboundedSender<ClientMsg>,
}

impl ServerClient {
    /// Open a connection, send the initial `hello`, and return the client
    /// plus a stream of [`ServerEvent`]s.
    ///
    /// - `url`: `ws://<onion>:80/ws` (onion), `wss://relay/ws` (clearnet TLS),
    ///   or `ws://host:port/ws` (clearnet plain / tests).
    /// - `dial`: how to physically reach it — one of the transport "doors"
    ///   (`Socks5` for onion via Tor, `Tls` for `wss://`, `Direct` for `ws://`).
    /// - `identity`: our identity, used to answer the relay's auth `Challenge`
    ///   (huddle 1.1.4). The connector signs the challenge nonce and sends the
    ///   pubkey + signature in `Hello`; the relay rejects us otherwise.
    pub async fn connect(
        url: &str,
        dial: &crate::network::transport::DialMode,
        identity: Arc<Identity>,
        rooms: Vec<String>,
    ) -> Result<(Self, mpsc::UnboundedReceiver<ServerEvent>)> {
        use crate::network::transport::DialMode;
        match dial {
            DialMode::Socks5 { proxy } => {
                let proxy: std::net::SocketAddr = proxy
                    .parse()
                    .map_err(|e| HuddleError::Network(format!("bad socks address: {e}")))?;
                let target = host_port_from_ws_url(url)?;
                let stream = tokio_socks::tcp::Socks5Stream::connect(proxy, target.as_str())
                    .await
                    .map_err(|e| HuddleError::Network(format!("tor socks connect: {e}")))?;
                let (ws, _resp) =
                    tokio_tungstenite::client_async_with_config(url, stream, Some(relay_ws_config()))
                        .await
                        .map_err(|e| HuddleError::Network(format!("ws handshake: {e}")))?;
                Ok(Self::spawn(ws, identity, rooms))
            }
            // Plain `ws://` and `wss://` with the system trust store both go
            // through `connect_async`, which negotiates TLS from the URL
            // scheme (tokio-tungstenite's rustls-tls-native-roots feature).
            DialMode::Direct | DialMode::Tls { pinned_cert_der: None } => {
                let (ws, _resp) = tokio_tungstenite::connect_async_with_config(
                    url,
                    Some(relay_ws_config()),
                    false,
                )
                .await
                .map_err(|e| HuddleError::Network(format!("ws connect: {e}")))?;
                Ok(Self::spawn(ws, identity, rooms))
            }
            // Self-signed cert pinning is structured but not wired in this
            // build — the recommended clearnet-TLS path uses a real cert
            // (Caddy / Let's Encrypt / Cloudflare), which the arm above
            // handles. Onion doors remain available for stronger privacy.
            DialMode::Tls {
                pinned_cert_der: Some(_),
            } => Err(HuddleError::Network(
                "pinned-certificate wss is not supported in this build — use a real cert (Caddy/Let's Encrypt) or an onion door".into(),
            )),
            // huddle 1.0: in-process Tor via Arti. Bootstraps (once) an
            // embedded Tor client and opens the stream to the onion through
            // it, then speaks WebSocket over that stream — `spawn` is reused.
            #[cfg(feature = "arti")]
            DialMode::Arti { bridge } => {
                let client =
                    crate::network::transport::arti_client(bridge.as_deref()).await?;
                let hp = host_port_from_ws_url(url)?;
                let (host, port_s) = hp.rsplit_once(':').ok_or_else(|| {
                    HuddleError::Network(format!("bad host:port from {url}"))
                })?;
                let port: u16 = port_s
                    .parse()
                    .map_err(|_| HuddleError::Network(format!("bad port in {url}")))?;
                let stream = client
                    .connect((host, port))
                    .await
                    .map_err(|e| HuddleError::Network(format!("arti connect: {e}")))?;
                let (ws, _resp) =
                    tokio_tungstenite::client_async_with_config(url, stream, Some(relay_ws_config()))
                        .await
                        .map_err(|e| HuddleError::Network(format!("ws handshake: {e}")))?;
                Ok(Self::spawn(ws, identity, rooms))
            }
        }
    }

    /// Spawn the read/write pumps for an established socket. Generic over
    /// the inner stream so the Tor-SOCKS and direct paths (different
    /// stream types) share one implementation.
    fn spawn<S>(
        ws: WebSocketStream<S>,
        identity: Arc<Identity>,
        rooms: Vec<String>,
    ) -> (Self, mpsc::UnboundedReceiver<ServerEvent>)
    where
        S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send + 'static,
    {
        let (mut sink, mut stream) = ws.split();
        let (out_tx, mut out_rx) = mpsc::unbounded_channel::<ClientMsg>();
        let (ev_tx, ev_rx) = mpsc::unbounded_channel::<ServerEvent>();

        // huddle 1.1.4: we do NOT send `Hello` up front anymore. The relay
        // opens with a `Challenge`; the reader pump (below) signs that nonce
        // and queues the authenticated `Hello`. Because the relay rejects
        // anything sent before a valid `Hello`, the writer pump holds back
        // any other outgoing frame (a publish/subscribe the app issues during
        // the handshake window) until the `Hello` has actually gone out.
        tokio::spawn(async move {
            let mut hello_sent = false;
            let mut pending: Vec<ClientMsg> = Vec::new();
            // huddle 2.0.3 (audit N-M7): application keepalive. The relay now reaps
            // an authenticated socket that sends no frame for ~150 s, so send a
            // ClientMsg::Ping on a 60 s cadence to keep a healthy-but-quiet
            // connection alive. Folded into the writer rather than a separate task
            // holding an `out_tx` clone: a permanent clone would keep `out_rx` from
            // ending when every public handle drops, so the socket would never
            // close and the server would never mark us offline (breaking mailboxing).
            let mut keepalive =
                tokio::time::interval(std::time::Duration::from_secs(RELAY_KEEPALIVE_SECS));
            keepalive.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
            loop {
                tokio::select! {
                    // Prefer real app messages over the keepalive when both are ready.
                    biased;
                    recv = out_rx.recv() => {
                        // When `out_rx` ends (every `ServerClient` handle dropped)
                        // break to close the socket so the server marks us offline
                        // and starts mailboxing.
                        let Some(msg) = recv else { break; };
                        let is_hello = matches!(msg, ClientMsg::Hello { .. });
                        if !hello_sent && !is_hello {
                            // huddle 1.3.4: bound the pre-auth backlog. If the relay
                            // accepts the socket but never sends its Challenge (the
                            // server still closes it after its 20 s pre-auth timeout),
                            // this keeps `pending` from growing without limit.
                            if pending.len() >= MAX_PENDING_PREAUTH {
                                warn!(
                                    cap = MAX_PENDING_PREAUTH,
                                    "relay has not authenticated us yet; dropping a queued message"
                                );
                                continue;
                            }
                            pending.push(msg);
                            continue;
                        }
                        let json = match serde_json::to_string(&msg) {
                            Ok(j) => j,
                            Err(_) => continue,
                        };
                        if sink.send(WsMessage::Text(json.into())).await.is_err() {
                            return;
                        }
                        if is_hello {
                            hello_sent = true;
                            // Flush anything the app queued while we waited for the
                            // challenge, preserving its order after the Hello.
                            for m in pending.drain(..) {
                                let json = match serde_json::to_string(&m) {
                                    Ok(j) => j,
                                    Err(_) => continue,
                                };
                                if sink.send(WsMessage::Text(json.into())).await.is_err() {
                                    return;
                                }
                            }
                        }
                    }
                    _ = keepalive.tick(), if hello_sent => {
                        let json = match serde_json::to_string(&ClientMsg::Ping) {
                            Ok(j) => j,
                            Err(_) => continue,
                        };
                        if sink.send(WsMessage::Text(json.into())).await.is_err() {
                            return;
                        }
                    }
                }
            }
            let _ = sink.close().await;
        });

        // Reader pump: parse server messages into ServerEvents. On the opening
        // `Challenge`, prove our identity by signing the nonce and sending the
        // authenticated `Hello` through the writer.
        // Held only long enough to send the one `Hello` in response to the
        // challenge, then dropped. Crucially it must NOT outlive that: if the
        // reader kept a permanent `out_tx` clone, dropping every public
        // `ServerClient` handle would no longer end the writer's `out_rx`, the
        // socket would never close, and the server would never mark us offline
        // (breaking offline mailboxing). `Option::take()` releases it after use.
        let mut hello_tx = Some(out_tx.clone());
        tokio::spawn(async move {
            while let Some(frame) = stream.next().await {
                let frame = match frame {
                    Ok(f) => f,
                    Err(_) => break,
                };
                let text = match frame {
                    WsMessage::Text(t) => t.as_str().to_string(),
                    WsMessage::Binary(b) => String::from_utf8_lossy(&b).into_owned(),
                    WsMessage::Close(_) => break,
                    _ => continue,
                };
                match serde_json::from_str::<ServerMsg>(&text) {
                    Ok(ServerMsg::Challenge { nonce_b64 }) => {
                        if nonce_b64.len() > MAX_NONCE_B64 {
                            warn!(
                                len = nonce_b64.len(),
                                "relay sent an oversized challenge nonce — dropping connection"
                            );
                            break;
                        }
                        if let Some(tx) = hello_tx.take() {
                            match B64.decode(nonce_b64.as_bytes()) {
                                Ok(nonce) => {
                                    let sig = identity.sign(&relay_auth_msg(&nonce));
                                    let hello = ClientMsg::Hello {
                                        fingerprint: identity.fingerprint().to_string(),
                                        pubkey_b64: B64.encode(identity.public_bytes()),
                                        signature_b64: B64.encode(sig),
                                        rooms: rooms.clone(),
                                        // huddle 2.0: announce at-least-once ACK
                                        // support so a 2.0 relay holds each mailbox
                                        // row until we ACK it. Ignored by pre-2.0
                                        // relays (unknown field).
                                        acks: true,
                                    };
                                    // If the writer is gone the connection is dead anyway.
                                    let _ = tx.send(hello);
                                }
                                Err(e) => {
                                    warn!(error = %e, "relay sent an undecodable challenge nonce");
                                    break;
                                }
                            }
                        }
                        // `tx` dropped here — the reader no longer pins the
                        // outgoing channel open.
                    }
                    // huddle 2.0.4: the unified `Ready` carries the relay-echoed
                    // fingerprint; we already know our own identity, so ignore it.
                    Ok(ServerMsg::Ready { .. }) => {
                        let _ = ev_tx.send(ServerEvent::Ready);
                    }
                    Ok(ServerMsg::Sent {
                        id,
                        delivered,
                        queued,
                    }) => {
                        let _ = ev_tx.send(ServerEvent::Sent {
                            id,
                            delivered,
                            queued,
                        });
                    }
                    Ok(ServerMsg::ConnectToken { token, ttl_secs }) => {
                        let _ = ev_tx.send(ServerEvent::ConnectToken { token, ttl_secs });
                    }
                    Ok(ServerMsg::ConnectTokenResolved {
                        fingerprint,
                        pubkey_b64,
                        // huddle 2.0.4: the unified variant also carries the
                        // echoed `token`; the client doesn't need it.
                        ..
                    }) => {
                        let _ = ev_tx.send(ServerEvent::ConnectTokenResolved {
                            fingerprint,
                            pubkey_b64,
                        });
                    }
                    Ok(ServerMsg::Message {
                        room,
                        id,
                        payload_b64,
                        mailbox_id,
                    }) => {
                        if payload_b64.len() > MAX_PAYLOAD_B64 {
                            warn!(
                                len = payload_b64.len(),
                                "server sent an oversized payload — dropping"
                            );
                            continue;
                        }
                        match B64.decode(payload_b64.as_bytes()) {
                            Ok(payload) => {
                                // huddle 2.0: pass the relay's mailbox row id (if
                                // any) through so the consumer can ACK durable
                                // receipt and let the relay delete the row.
                                let _ = ev_tx.send(ServerEvent::Message {
                                    room,
                                    id,
                                    payload,
                                    mailbox_id,
                                });
                            }
                            Err(e) => warn!(error = %e, "server sent undecodable payload"),
                        }
                    }
                    Ok(ServerMsg::Error { message }) => warn!(%message, "huddle-server error"),
                    Ok(ServerMsg::Pong) => {}
                    Err(e) => warn!(error = %e, "unparseable server message"),
                }
            }
            let _ = ev_tx.send(ServerEvent::Disconnected);
        });

        (Self { out_tx }, ev_rx)
    }

    /// Send a room's opaque wire bytes to the server for fan-out.
    pub fn publish(&self, room: &str, id: &str, payload: &[u8]) -> Result<()> {
        self.send(ClientMsg::Publish {
            room: room.to_string(),
            id: id.to_string(),
            payload_b64: B64.encode(payload),
        })
    }

    /// huddle 1.2: deliver `payload` straight to recipient `to`'s
    /// fingerprint, independent of room membership (1:1 DMs, friend requests).
    /// The server delivers it live to every connection `to` has open, or
    /// queues it in their mailbox when they're offline. `room` is the opaque
    /// tag the recipient files it under.
    pub fn send_direct(&self, to: &str, room: &str, id: &str, payload: &[u8]) -> Result<()> {
        self.send(ClientMsg::SendDirect {
            to: to.to_string(),
            room: room.to_string(),
            id: id.to_string(),
            payload_b64: B64.encode(payload),
        })
    }

    /// Assert membership of a room so the server mailboxes us when offline.
    pub fn subscribe(&self, room: &str) -> Result<()> {
        self.send(ClientMsg::Subscribe {
            room: room.to_string(),
        })
    }

    pub fn unsubscribe(&self, room: &str) -> Result<()> {
        self.send(ClientMsg::Unsubscribe {
            room: room.to_string(),
        })
    }

    /// huddle 1.2.1: ask the relay to mint a short-lived connect code bound to
    /// our identity. The reply arrives as `ServerEvent::ConnectToken`.
    pub fn create_connect_token(&self) -> Result<()> {
        self.send(ClientMsg::CreateConnectToken)
    }

    /// huddle 1.2.1: ask the relay to resolve a connect code to its owner.
    /// The reply arrives as `ServerEvent::ConnectTokenResolved`.
    pub fn redeem_connect_token(&self, token: &str) -> Result<()> {
        self.send(ClientMsg::RedeemConnectToken {
            token: token.to_string(),
        })
    }

    /// Ask the server to re-drain our mailbox.
    pub fn fetch(&self) -> Result<()> {
        self.send(ClientMsg::Fetch)
    }

    /// huddle 2.0 (F7): acknowledge durable receipt of a mailbox-delivered
    /// message so the relay deletes that row (at-least-once delivery). Call this
    /// only after the message has been persisted, using the `mailbox_id` carried
    /// on the corresponding [`ServerEvent::Message`] (skip it when that id is
    /// `None` — a live or pre-2.0 delivery the relay already dropped on deliver).
    pub fn ack_mailbox(&self, mailbox_id: i64) -> Result<()> {
        self.send(ClientMsg::Ack { mailbox_id })
    }

    pub fn ping(&self) -> Result<()> {
        self.send(ClientMsg::Ping)
    }

    fn send(&self, msg: ClientMsg) -> Result<()> {
        self.out_tx
            .send(msg)
            .map_err(|_| HuddleError::Network("server connection closed".to_string()))
    }
}

/// Extract `host:port` from a `ws://`/`wss://` URL for the SOCKS target.
/// Defaults to port 80 for `ws://` (matches the onion's `HiddenServicePort
/// 80`) and 443 for `wss://` when no explicit port is given.
fn host_port_from_ws_url(url: &str) -> Result<String> {
    let (rest, default_port) = if let Some(r) = url.strip_prefix("wss://") {
        (r, 443)
    } else if let Some(r) = url.strip_prefix("ws://") {
        (r, 80)
    } else {
        return Err(HuddleError::Network(format!(
            "expected ws:// url, got {url}"
        )));
    };
    let authority = rest.split('/').next().unwrap_or(rest);
    if authority.is_empty() {
        return Err(HuddleError::Network(format!("no host in url: {url}")));
    }
    if authority.contains(':') {
        Ok(authority.to_string())
    } else {
        Ok(format!("{authority}:{default_port}"))
    }
}

#[cfg(test)]
mod tests {
    use super::host_port_from_ws_url;

    #[test]
    fn parses_host_port() {
        assert_eq!(
            host_port_from_ws_url("ws://abc.onion/ws").unwrap(),
            "abc.onion:80"
        );
        assert_eq!(
            host_port_from_ws_url("ws://127.0.0.1:8787/ws").unwrap(),
            "127.0.0.1:8787"
        );
        assert_eq!(host_port_from_ws_url("wss://h:443").unwrap(), "h:443");
        // huddle 1.0: bare wss:// defaults to 443, not 80.
        assert_eq!(
            host_port_from_ws_url("wss://relay.example/ws").unwrap(),
            "relay.example:443"
        );
        assert!(host_port_from_ws_url("http://x").is_err());
    }
}
