aura-gui 0.2.2

//! Esperas de red que no se quedan colgadas. ureq es bloqueante y su `timeout_recv_body` es un tope para el cuerpo
//! entero, no para el silencio entre bytes: una respuesta larga moría a los 90 s y una conexión reutilizada que
//! quedó muerta (tras cambiar de app o suspender el equipo) bloqueaba `send` sin límite. Aquí la operación corre en
//! un hilo auxiliar y quien espera mira la cancelación cada `POLL`: Detener responde al momento y un silencio
//! largo se convierte en un error que se puede reintentar. El hilo abandonado termina solo con los topes de ureq.

use crate::ai::types::AiError;
use crate::core::CancelToken;
use std::io::Read;
use std::sync::mpsc::{Receiver, RecvTimeoutError, sync_channel};
use std::time::{Duration, Instant};

const POLL: Duration = Duration::from_millis(50);
/// Trozos leídos que pueden esperar a que se procesen: acota la memoria si la ventana va lenta.
const QUEUE: usize = 16;
const CHUNK: usize = 8 * 1024;

/// Corre `work` aparte y espera su resultado hasta `limit`, atento a la cancelación.
pub(crate) fn interruptible<T: Send + 'static>(
    cancel: &CancelToken,
    limit: Duration,
    what: &'static str,
    work: impl FnOnce() -> T + Send + 'static,
) -> Result<T, AiError> {
    let (sender, receiver) = sync_channel(1);
    std::thread::Builder::new()
        .name("aura-ai-io".into())
        .spawn(move || {
            let _ = sender.send(work());
        })
        .map_err(|error| AiError::Transport(error.to_string()))?;
    let started = Instant::now();
    loop {
        if cancel.is_cancelled() {
            return Err(AiError::Cancelled);
        }
        match receiver.recv_timeout(POLL) {
            Ok(value) => return Ok(value),
            Err(RecvTimeoutError::Disconnected) => return Err(AiError::Transport(format!("{what}: se interrumpió"))),
            Err(RecvTimeoutError::Timeout) if started.elapsed() >= limit => {
                return Err(AiError::Transport(format!("{what}: sin respuesta en {} s", limit.as_secs())));
            }
            Err(RecvTimeoutError::Timeout) => {}
        }
    }
}

/// Lee el cuerpo en un hilo aparte y entrega sus trozos; el canal se cierra al terminar el cuerpo.
pub(crate) struct ChunkStream {
    receiver: Receiver<std::io::Result<Vec<u8>>>,
}

impl ChunkStream {
    pub(crate) fn spawn(mut reader: impl Read + Send + 'static) -> Result<Self, AiError> {
        let (sender, receiver) = sync_channel(QUEUE);
        std::thread::Builder::new()
            .name("aura-ai-read".into())
            .spawn(move || {
                let mut buffer = vec![0u8; CHUNK];
                loop {
                    let item = match reader.read(&mut buffer) {
                        Ok(0) => return,
                        Ok(read) => Ok(buffer[..read].to_vec()),
                        Err(error) => Err(error),
                    };
                    let failed = item.is_err();
                    // Si nadie escucha ya (se canceló o se dio por colgado), se suelta la conexión.
                    if sender.send(item).is_err() || failed {
                        return;
                    }
                }
            })
            .map_err(|error| AiError::Transport(error.to_string()))?;
        Ok(Self { receiver })
    }

    /// Siguiente trozo; `Ok(None)` al terminar el cuerpo. Falla si pasan `stall` sin bytes o se cancela.
    pub(crate) fn next(&self, cancel: &CancelToken, stall: Duration) -> Result<Option<Vec<u8>>, AiError> {
        let started = Instant::now();
        loop {
            if cancel.is_cancelled() {
                return Err(AiError::Cancelled);
            }
            match self.receiver.recv_timeout(POLL) {
                Ok(Ok(chunk)) => return Ok(Some(chunk)),
                Ok(Err(error)) => return Err(AiError::Transport(error.to_string())),
                Err(RecvTimeoutError::Disconnected) => return Ok(None),
                Err(RecvTimeoutError::Timeout) if started.elapsed() >= stall => {
                    let seconds = stall.as_secs();
                    return Err(AiError::Transport(format!("el servidor dejó de enviar datos durante {seconds} s")));
                }
                Err(RecvTimeoutError::Timeout) => {}
            }
        }
    }
}

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

    #[test]
    fn cancelar_o_agotar_el_limite_no_espera_a_la_operacion() {
        let cancel = CancelToken::default();
        let slow = || std::thread::sleep(Duration::from_secs(5));
        let started = Instant::now();
        let result = interruptible(&cancel, Duration::from_millis(100), "prueba", slow);
        assert!(matches!(result, Err(AiError::Transport(text)) if text.contains("sin respuesta")));
        cancel.cancel();
        assert!(matches!(interruptible(&cancel, Duration::from_secs(5), "prueba", slow), Err(AiError::Cancelled)));
        assert!(started.elapsed() < Duration::from_secs(2));
        assert_eq!(interruptible(&CancelToken::default(), Duration::from_secs(1), "prueba", || 7).ok(), Some(7));
    }

    /// Un lector que entrega un trozo y luego se queda callado, como una conexión muerta.
    struct Silent(bool);

    impl Read for Silent {
        fn read(&mut self, buffer: &mut [u8]) -> std::io::Result<usize> {
            if std::mem::replace(&mut self.0, true) {
                std::thread::sleep(Duration::from_secs(5));
                return Ok(0);
            }
            buffer[..4].copy_from_slice(b"data");
            Ok(4)
        }
    }

    #[test]
    fn un_silencio_largo_es_un_error_y_el_final_del_cuerpo_cierra_el_stream() {
        let cancel = CancelToken::default();
        let stream = ChunkStream::spawn(Silent(false)).unwrap();
        assert_eq!(stream.next(&cancel, Duration::from_secs(1)).unwrap().as_deref(), Some(&b"data"[..]));
        let stalled = stream.next(&cancel, Duration::from_millis(150));
        assert!(matches!(stalled, Err(AiError::Transport(text)) if text.contains("dejó de enviar")));
        let finished = ChunkStream::spawn(std::io::Cursor::new(b"ab".to_vec())).unwrap();
        assert_eq!(finished.next(&cancel, Duration::from_secs(1)).unwrap().as_deref(), Some(&b"ab"[..]));
        assert_eq!(finished.next(&cancel, Duration::from_secs(1)).unwrap(), None);
    }
}