aura-gui 0.2.2

//! Transporte HTTP: envía la petición y lee el stream SSE en el hilo que llama, sin async.

use crate::ai::provider::Decoder;
use crate::ai::sse::SseParser;
use crate::ai::types::{AiError, Event};
use crate::core::CancelToken;
use serde_json::Value;
use std::io::Read;
use std::time::Duration;

const CONNECT_TIMEOUT: Duration = Duration::from_secs(15);
/// Cabeceras de la respuesta: una conexión reutilizada que quedó muerta no responde nunca.
const RESPONSE_TIMEOUT: Duration = Duration::from_secs(90);
/// Un stream sin bytes durante este tiempo se da por colgado (ver `watchdog`).
const STALL_TIMEOUT: Duration = Duration::from_secs(120);
/// Tope absoluto del cuerpo: solo libera el hilo lector de una conexión ya abandonada, nunca corta una respuesta.
const BODY_CAP: Duration = Duration::from_secs(30 * 60);
/// Cuánto del cuerpo de un error se conserva para mostrarlo.
const ERROR_BODY_LIMIT: u64 = 2_000;

/// Cuántas veces se reintenta y cuánto se espera: solo antes de recibir nada del stream, nunca a mitad.
#[derive(Clone, Copy, Debug)]
pub struct RetryPolicy {
    pub retries: u32,
    /// Espera del primer reintento; se duplica en cada uno.
    pub base_delay: Duration,
    /// Tope de cualquier espera, también si el servidor pide más con `retry-after`.
    pub max_delay: Duration,
}

impl Default for RetryPolicy {
    fn default() -> Self {
        Self { retries: 3, base_delay: Duration::from_secs(1), max_delay: Duration::from_secs(30) }
    }
}

/// Saturación (429, 529) y fallos pasajeros del servidor.
fn retryable_status(code: u16) -> bool {
    matches!(code, 429 | 500 | 502 | 503 | 504 | 529)
}

/// Espera en tramos cortos para poder cancelar; `false` si se canceló.
fn wait(delay: Duration, cancel: &CancelToken) -> bool {
    let end = std::time::Instant::now() + delay;
    while std::time::Instant::now() < end {
        if cancel.is_cancelled() {
            return false;
        }
        std::thread::sleep(Duration::from_millis(20).min(delay));
    }
    !cancel.is_cancelled()
}

/// Un solo cliente para todo el proceso: reutiliza las conexiones (keep-alive), así cada vuelta del agente se
/// ahorra DNS, TCP y TLS. Con la función `gzip` de ureq pide y descomprime las respuestas comprimidas.
fn shared_agent() -> &'static ureq::Agent {
    static AGENT: std::sync::OnceLock<ureq::Agent> = std::sync::OnceLock::new();
    AGENT.get_or_init(|| {
        ureq::Agent::config_builder()
            .timeout_connect(Some(CONNECT_TIMEOUT))
            .timeout_send_request(Some(CONNECT_TIMEOUT))
            .timeout_recv_response(Some(RESPONSE_TIMEOUT))
            .timeout_recv_body(Some(BODY_CAP))
            .http_status_as_error(false)
            .build()
            .into()
    })
}

/// Petición JSON sin streaming (servicios de autocompletado que responden de una vez), sin reintentos: la
/// respuesta solo sirve en el momento. La cancelación se comprueba antes de enviar y al terminar.
pub fn post_json(
    url: &str,
    headers: &[(String, String)],
    body: &Value,
    cancel: &CancelToken,
) -> Result<Value, AiError> {
    if cancel.is_cancelled() {
        return Err(AiError::Cancelled);
    }
    let payload = serde_json::to_vec(body).map_err(|error| AiError::Decode(error.to_string()))?;
    let mut request = shared_agent().post(url).header("content-type", "application/json");
    for (name, value) in headers {
        request = request.header(name, value);
    }
    let response = request.send(&payload).map_err(|error| AiError::Transport(error.to_string()))?;
    let code = response.status().as_u16();
    let mut text = String::new();
    let limit = if code < 400 { 4_000_000 } else { ERROR_BODY_LIMIT };
    response
        .into_body()
        .into_reader()
        .take(limit)
        .read_to_string(&mut text)
        .map_err(|error| AiError::Transport(error.to_string()))?;
    if code >= 400 {
        return Err(AiError::Status { code, body: text });
    }
    if cancel.is_cancelled() {
        return Err(AiError::Cancelled);
    }
    serde_json::from_str(&text).map_err(|error| AiError::Decode(error.to_string()))
}

pub fn stream_sse(
    url: &str,
    headers: &[(String, String)],
    body: &Value,
    cancel: &CancelToken,
    decoder: &mut dyn Decoder,
    emit: &mut dyn FnMut(Event),
) -> Result<(), AiError> {
    stream_sse_with(RetryPolicy::default(), url, headers, body, cancel, decoder, emit)
}

pub fn stream_sse_with(
    policy: RetryPolicy,
    url: &str,
    headers: &[(String, String)],
    body: &Value,
    cancel: &CancelToken,
    decoder: &mut dyn Decoder,
    emit: &mut dyn FnMut(Event),
) -> Result<(), AiError> {
    let agent = shared_agent();
    let payload: std::sync::Arc<[u8]> =
        serde_json::to_vec(body).map_err(|error| AiError::Decode(error.to_string()))?.into();
    let mut attempt = 0;
    let reader = loop {
        if cancel.is_cancelled() {
            return Err(AiError::Cancelled);
        }
        let mut request =
            agent.post(url).header("content-type", "application/json").header("accept", "text/event-stream");
        for (name, value) in headers {
            request = request.header(name, value);
        }
        let body = payload.clone();
        let sent =
            crate::ai::watchdog::interruptible(cancel, RESPONSE_TIMEOUT, "petición", move || request.send(&body[..]));
        let (error, server_delay) = match sent? {
            Err(error) => (AiError::Transport(error.to_string()), None),
            Ok(response) => {
                let code = response.status().as_u16();
                let server_delay = response
                    .headers()
                    .get("retry-after")
                    .and_then(|value| value.to_str().ok())
                    .and_then(|value| value.trim().parse::<u64>().ok())
                    .map(Duration::from_secs);
                let reader = response.into_body().into_reader();
                if code < 400 {
                    break reader;
                }
                let mut text = String::new();
                if let Err(error) = reader.take(ERROR_BODY_LIMIT).read_to_string(&mut text) {
                    text = format!("(no se pudo leer el cuerpo: {error})");
                }
                (AiError::Status { code, body: text }, server_delay)
            }
        };
        let retry = match &error {
            AiError::Status { code, .. } => retryable_status(*code),
            _ => true,
        };
        if !retry || attempt >= policy.retries {
            return Err(error);
        }
        let delay = server_delay.unwrap_or_else(|| policy.base_delay * 2u32.saturating_pow(attempt));
        attempt += 1;
        if !wait(delay.min(policy.max_delay), cancel) {
            return Err(AiError::Cancelled);
        }
    };
    let (mut parser, mut events, mut decoded) = (SseParser::default(), Vec::new(), Vec::new());
    let chunks = crate::ai::watchdog::ChunkStream::spawn(reader)?;
    while let Some(chunk) = chunks.next(cancel, STALL_TIMEOUT)? {
        parser.push(&chunk, &mut events);
        for event in events.drain(..) {
            decoder.feed(&event, &mut decoded)?;
            decoded.drain(..).for_each(&mut *emit);
        }
    }
    if !decoder.is_complete() {
        // Sin marca de final, lo recibido es parcial: no puede pasar por una respuesta completa.
        return Err(AiError::Transport("la conexión se cerró antes de terminar la respuesta".into()));
    }
    decoder.finish(&mut decoded);
    decoded.drain(..).for_each(emit);
    Ok(())
}

#[cfg(test)]
#[path = "http_tests.rs"]
mod tests;