//! A client for the daemon over its Unix socket (spec §8.5), signing every

//! request with the caller's key: the shims and the CLI use it, and so do

//! the tests.

use anyhow::{Context, Result};

use http_body_util::{BodyExt, Full};

use hyper::body::Bytes;

use spectra_core::keys::KeyPair;

use spectra_core::proof::Proof;

use std::path::{Path, PathBuf};

/// A response, its body read whole.

#[derive(Debug)]

pub struct Response {

    pub status: u16,

    pub headers: Vec<(String, String)>,

    pub body: Vec<u8>,

}

impl Response {

    pub fn json(&self) -> Result<serde_json::Value> {

        serde_json::from_slice(&self.body).with_context(|| format!("not JSON: {}", String::from_utf8_lossy(&self.body)))

    }

    pub fn header(&self, name: &str) -> Option<&str> {

        self.headers.iter().find(|(k, _)| k.eq_ignore_ascii_case(name)).map(|(_, v)| v.as_str())

    }

}

/// Where the daemon is: its Unix socket, or a server over TLS with a

/// pinned certificate (spec §8.6).

#[derive(Clone)]

pub enum Endpoint {

    Unix(PathBuf),

    Tls { addr: String, config: std::sync::Arc<rustls::ClientConfig> },

}

impl Endpoint {

    /// `https://HOST:PORT` accepting exactly the certificate pinned by `pin`.

    pub fn tls(url: &str, pin: [u8; 32]) -> Result<Endpoint> {

        Ok(Endpoint::Tls { addr: host_port(url)?, config: crate::tls::Pinned::new(Some(pin)).client_config()? })

    }

    pub fn describe(&self) -> String {

        match self {

            Endpoint::Unix(p) => p.display().to_string(),

            Endpoint::Tls { addr, .. } => format!("https://{addr}"),

        }

    }

}

/// `host:port` of an `https://` URL.

pub fn host_port(url: &str) -> Result<String> {

    let rest = url.strip_prefix("https://").with_context(|| format!("{url} is not an https:// URL"))?;

    let hp = rest.split('/').next().unwrap_or("");

    if hp.is_empty() {

        anyhow::bail!("{url} names no host");

    }

    // The port defaults to Spectra's.

    let has_port = if hp.starts_with('[') { hp.contains("]:") } else { hp.contains(':') };

    Ok(if has_port { hp.to_string() } else { format!("{hp}:7350") })

}

trait Io: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send {}

impl<T: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send> Io for T {}

pub struct Client {

    endpoint: Endpoint,

    signer: Option<KeyPair>,

}

impl Client {

    pub fn new(socket: &Path, signer: Option<KeyPair>) -> Client {

        Client { endpoint: Endpoint::Unix(socket.to_path_buf()), signer }

    }

    pub fn at(endpoint: Endpoint, signer: Option<KeyPair>) -> Client {

        Client { endpoint, signer }

    }

    pub fn endpoint(&self) -> &Endpoint {

        &self.endpoint

    }

    /// The key this client signs with, if any.

    pub fn signer(&self) -> Option<&KeyPair> {

        self.signer.as_ref()

    }

    async fn io(&self) -> Result<Box<dyn Io>> {

        Ok(match &self.endpoint {

            Endpoint::Unix(socket) => Box::new(

                tokio::net::UnixStream::connect(socket)

                    .await

                    .with_context(|| format!("connecting to {} (is `spectra serve` running?)", socket.display()))?,

            ),

            Endpoint::Tls { addr, config } => Box::new(crate::tls::connect(addr, config.clone()).await?),

        })

    }

    /// Whether the daemon answers at the endpoint.

    pub async fn reachable(&self) -> bool {

        self.io().await.is_ok()

    }

    /// Whether a daemon answers on the socket.

    pub async fn alive(socket: &Path) -> bool {

        tokio::net::UnixStream::connect(socket).await.is_ok()

    }

    /// Send `method path` with `body` and extra headers, signed.

    pub async fn send(&self, method: &str, path: &str, headers: &[(&str, String)], body: Vec<u8>) -> Result<Response> {

        let stream = self.io().await?;

        let (mut sender, conn) = hyper::client::conn::http1::handshake(hyper_util::rt::TokioIo::new(stream)).await?;

        tokio::spawn(async move {

            let _ = conn.await;

        });

        // One request a connection: the server closes it after answering, so

        // nothing lingers to hold up its shutdown.

        let mut req = hyper::Request::builder().method(method).uri(path).header("host", "spectra").header("connection", "close");

        // JSON unless the caller says otherwise (a form, say).

        if !headers.iter().any(|(k, _)| k.eq_ignore_ascii_case("content-type")) {

            req = req.header("content-type", "application/json");

        }

        for (k, v) in headers {

            req = req.header(*k, v);

        }

        if let Some(pair) = &self.signer {

            for (k, v) in sign_headers(pair, &format!("{method} {path}"), &body) {

                req = req.header(k, v);

            }

        }

        let resp = sender.send_request(req.body(Full::new(Bytes::from(body)))?).await?;

        let status = resp.status().as_u16();

        let headers = resp.headers().iter().map(|(k, v)| (k.to_string(), v.to_str().unwrap_or("").to_string())).collect();

        let body = resp.into_body().collect().await?.to_bytes().to_vec();

        Ok(Response { status, headers, body })

    }

    /// A JSON-RPC request to `/mcp` with the protocol headers of `version`.

    pub async fn mcp(&self, version: &str, request: &serde_json::Value) -> Result<Response> {

        let method = request.get("method").and_then(|m| m.as_str()).unwrap_or("").to_string();

        let mut headers = vec![

            ("accept", "application/json, text/event-stream".to_string()),

            ("mcp-protocol-version", version.to_string()),

            ("mcp-method", method.clone()),

        ];

        if let Some(name) = request.pointer("/params/name").or_else(|| request.pointer("/params/uri")).and_then(|n| n.as_str()) {

            headers.push(("mcp-name", name.to_string()));

        }

        self.send("POST", "/mcp", &headers, serde_json::to_vec(request)?).await

    }

}

impl Client {

    /// Pass a request through as it is, unsigned, and hand back the

    /// response with its body still streaming (Radar's proxy, spec §8.6).

    pub async fn forward(&self, req: hyper::Request<Full<Bytes>>) -> Result<hyper::Response<hyper::body::Incoming>> {

        let stream = self.io().await?;

        let (mut sender, conn) = hyper::client::conn::http1::handshake(hyper_util::rt::TokioIo::new(stream)).await?;

        tokio::spawn(async move {

            let _ = conn.await;

        });

        Ok(sender.send_request(req).await?)

    }

    /// Send a signed GET and hand each line of the streamed body to `line`

    /// as it arrives (an SSE stream, say); stop when it returns false.

    pub async fn stream(&self, path: &str, mut line: impl FnMut(&str) -> bool) -> Result<u16> {

        let stream = self.io().await?;

        let (mut sender, conn) = hyper::client::conn::http1::handshake(hyper_util::rt::TokioIo::new(stream)).await?;

        tokio::spawn(async move {

            let _ = conn.await;

        });

        let mut req = hyper::Request::builder().method("GET").uri(path).header("host", "spectra").header("connection", "close");

        if let Some(pair) = &self.signer {

            for (k, v) in sign_headers(pair, &format!("GET {path}"), b"") {

                req = req.header(k, v);

            }

        }

        let resp = sender.send_request(req.body(Full::new(Bytes::new()))?).await?;

        let status = resp.status().as_u16();

        let mut body = resp.into_body();

        let mut buf = String::new();

        while let Some(frame) = body.frame().await {

            if let Ok(data) = frame?.into_data() {

                buf.push_str(&String::from_utf8_lossy(&data));

                while let Some(i) = buf.find('\n') {

                    let l: String = buf.drain(..=i).collect();

                    if !line(l.trim_end()) {

                        return Ok(status);

                    }

                }

            }

        }

        Ok(status)

    }

}

/// The signature headers of spec §8.5 for `method_path` and `body`.

pub fn sign_headers(pair: &KeyPair, method_path: &str, body: &[u8]) -> Vec<(&'static str, String)> {

    let p = Proof::sign(pair, method_path, body);

    vec![

        ("spectra-key", p.sig.key.to_hex()),

        ("spectra-time", p.time_ms.to_string()),

        ("spectra-nonce", hex::encode(p.nonce)),

        ("spectra-signature", hex::encode(spectra_core::cbor::encode(&p.sig.to_value()))),

    ]

}

/// The JSON-RPC result inside a response, whether it came as JSON or as

/// the last `data:` of an SSE stream.

pub fn rpc_result(r: &Response) -> Result<serde_json::Value> {

    let is_sse = r.header("content-type").is_some_and(|c| c.starts_with("text/event-stream"));

    if !is_sse {

        return r.json();

    }

    let text = String::from_utf8_lossy(&r.body);

    let last = text

        .lines()

        .rev()

        .filter_map(|l| l.strip_prefix("data:"))

        .map(str::trim)

        .find(|l| !l.is_empty())

        .context("an SSE response with no data")?;

    Ok(serde_json::from_str(last)?)

}