path: root/crates/asist-core/src/http.rs

//! Cliente HTTP/1.1 mínimo para hablar con los servidores locales.
//!
//! Los tres motores (llama-server, tts-server) escuchan en localhost sobre HTTP
//! plano, así que no hace falta TLS ni un runtime asíncrono. A cambio de las
//! ~300 líneas de aquí se gana lo único que ninguna librería genérica da
//! cómodo y que este pipeline necesita de verdad: **abortar una respuesta a
//! media descarga**. Cuando el usuario interrumpe al asistente hay que dejar de
//! leer audio ya generado y cerrar el socket en ese mismo instante; con un
//! cliente que sólo expone `read_to_end` eso no se puede hacer.

use std::io::{BufRead, BufReader, Read, Write};
use std::net::TcpStream;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::Duration;

use crate::error::{Error, Result};

/// Bandera compartida que aborta una lectura en curso.
///
/// Se comprueba entre bloques, así que el corte ocurre como muy tarde un
/// `read` después de activarla —del orden de milisegundos con los tamaños de
/// bloque que usan los servidores.
#[derive(Clone, Default)]
pub struct Cancel(Arc<AtomicBool>);

impl Cancel {
    pub fn new() -> Self {
        Self::default()
    }

    pub fn cancel(&self) {
        self.0.store(true, Ordering::SeqCst);
    }

    pub fn is_cancelled(&self) -> bool {
        self.0.load(Ordering::SeqCst)
    }

    pub fn reset(&self) {
        self.0.store(false, Ordering::SeqCst);
    }
}

#[derive(Clone, Debug)]
pub struct HttpClient {
    /// `host:puerto` del servidor local.
    pub authority: String,
    pub connect_timeout: Duration,
    /// Tiempo máximo sin recibir un solo byte. No es el tiempo total: una
    /// síntesis larga puede tardar minutos siempre que siga fluyendo.
    pub read_timeout: Duration,
}

impl HttpClient {
    pub fn new(authority: impl Into<String>) -> Self {
        Self {
            authority: authority.into(),
            connect_timeout: Duration::from_secs(5),
            read_timeout: Duration::from_secs(120),
        }
    }

    pub fn with_read_timeout(mut self, timeout: Duration) -> Self {
        self.read_timeout = timeout;
        self
    }

    fn connect(&self, path: &str) -> Result<TcpStream> {
        let addr = resolve(&self.authority, path)?;
        let stream = TcpStream::connect_timeout(&addr, self.connect_timeout).map_err(|e| {
            Error::Transport {
                url: format!("http://{}{}", self.authority, path),
                source: e,
            }
        })?;
        stream.set_read_timeout(Some(self.read_timeout))?;
        // Los cuerpos son pequeños y la latencia manda: nada de Nagle.
        let _ = stream.set_nodelay(true);
        Ok(stream)
    }

    fn send(&self, method: &str, path: &str, body: Option<&[u8]>) -> Result<Response> {
        let mut stream = self.connect(path)?;
        let mut head = format!(
            "{method} {path} HTTP/1.1\r\nHost: {}\r\nConnection: close\r\nAccept: */*\r\n",
            self.authority
        );
        if let Some(body) = body {
            head.push_str("Content-Type: application/json\r\n");
            head.push_str(&format!("Content-Length: {}\r\n", body.len()));
        }
        head.push_str("\r\n");

        let url = format!("http://{}{}", self.authority, path);
        let io = |e: std::io::Error| Error::Transport {
            url: url.clone(),
            source: e,
        };
        stream.write_all(head.as_bytes()).map_err(io)?;
        if let Some(body) = body {
            stream.write_all(body).map_err(io)?;
        }
        stream.flush().map_err(io)?;

        Response::read_head(BufReader::new(stream), url)
    }

    /// GET que devuelve el cuerpo completo ya decodificado.
    pub fn get(&self, path: &str) -> Result<Vec<u8>> {
        self.send("GET", path, None)?.into_body()
    }

    /// POST de JSON que devuelve el cuerpo completo.
    pub fn post_json(&self, path: &str, body: &serde_json::Value) -> Result<Vec<u8>> {
        let payload = serde_json::to_vec(body)?;
        self.send("POST", path, Some(&payload))?.into_body()
    }

    /// POST de JSON cuya respuesta se consume a trozos según llega.
    ///
    /// `on_chunk` recibe cada bloque en cuanto está disponible y devuelve
    /// `false` para cortar; junto a `cancel` son las dos vías por las que el
    /// barge-in detiene una síntesis a medias.
    pub fn post_json_streaming(
        &self,
        path: &str,
        body: &serde_json::Value,
        cancel: &Cancel,
        mut on_chunk: impl FnMut(&[u8]) -> bool,
    ) -> Result<()> {
        let payload = serde_json::to_vec(body)?;
        let mut response = self.send("POST", path, Some(&payload))?;
        response.error_for_status()?;
        response.stream(cancel, &mut on_chunk)
    }

    /// POST de JSON que entrega la respuesta línea a línea (SSE de llama-server).
    pub fn post_json_lines(
        &self,
        path: &str,
        body: &serde_json::Value,
        cancel: &Cancel,
        mut on_line: impl FnMut(&str) -> bool,
    ) -> Result<()> {
        let mut pending = Vec::new();
        self.post_json_streaming(path, body, cancel, |chunk| {
            pending.extend_from_slice(chunk);
            while let Some(nl) = pending.iter().position(|b| *b == b'\n') {
                let line: Vec<u8> = pending.drain(..=nl).collect();
                let line = String::from_utf8_lossy(&line);
                if !on_line(line.trim_end_matches(['\r', '\n'])) {
                    return false;
                }
            }
            true
        })
    }

    /// Sondea `/health` y devuelve `true` si el servidor está listo.
    pub fn healthy(&self, path: &str) -> bool {
        self.get(path).is_ok()
    }
}

fn resolve(authority: &str, path: &str) -> Result<std::net::SocketAddr> {
    use std::net::ToSocketAddrs;
    authority
        .to_socket_addrs()
        .map_err(|e| Error::Transport {
            url: format!("http://{authority}{path}"),
            source: e,
        })?
        .next()
        .ok_or_else(|| Error::Config(format!("no se pudo resolver «{authority}»")))
}

/// Codificación del cuerpo anunciada por el servidor.
enum Body {
    Chunked,
    Length(usize),
    /// Sin longitud ni chunked: el cuerpo acaba cuando cierra el socket.
    UntilClose,
}

pub struct Response {
    pub status: u16,
    url: String,
    reader: BufReader<TcpStream>,
    body: Body,
}

impl Response {
    fn read_head(mut reader: BufReader<TcpStream>, url: String) -> Result<Self> {
        let io = |e: std::io::Error| Error::Transport {
            url: url.clone(),
            source: e,
        };

        let mut line = String::new();
        reader.read_line(&mut line).map_err(io)?;
        let status = line
            .split_whitespace()
            .nth(1)
            .and_then(|s| s.parse::<u16>().ok())
            .ok_or_else(|| Error::Config(format!("respuesta HTTP ilegible de {url}: {line:?}")))?;

        let mut body = Body::UntilClose;
        loop {
            let mut header = String::new();
            if reader.read_line(&mut header).map_err(io)? == 0 {
                break;
            }
            let header = header.trim_end();
            if header.is_empty() {
                break;
            }
            let Some((name, value)) = header.split_once(':') else {
                continue;
            };
            let (name, value) = (name.trim().to_ascii_lowercase(), value.trim());
            match name.as_str() {
                "transfer-encoding" if value.eq_ignore_ascii_case("chunked") => {
                    body = Body::Chunked;
                }
                "content-length" => {
                    if let Ok(n) = value.parse() {
                        body = Body::Length(n);
                    }
                }
                _ => {}
            }
        }

        Ok(Self {
            status,
            url,
            reader,
            body,
        })