tawasuyu/brahman

1//! Loop servidor: envuelve cualquier `Engine` y lo expone por Unix socket.
2
3use crate::protocol::{read_frame, write_frame, DaemonRequest, DaemonResponse};
4use sandokan_core::Engine;
5use std::path::Path;
6use std::sync::Arc;
7use std::time::Duration;
8use tokio::net::{UnixListener, UnixStream};
9
10/// Sirve `engine` en `socket_path` hasta que el future se cancele.
11///
12/// Si el socket ya existe (daemon previo que no limpió), se borra antes
13/// de bind. Cada conexión se atiende en su propia task; una conexión
14/// puede mandar múltiples requests secuenciales.
15pub async fn serve<E>(engine: Arc<E>, socket_path: &Path) -> std::io::Result<()>
16where
17    E: Engine + 'static,
18{
19    if socket_path.exists() {
20        let _ = std::fs::remove_file(socket_path);
21    }
22    let listener = UnixListener::bind(socket_path)?;
23    tracing::info!(socket = %socket_path.display(), "sandokan-daemon escuchando");
24
25    loop {
26        let (stream, _addr) = listener.accept().await?;
27        let engine = Arc::clone(&engine);
28        tokio::spawn(async move {
29            if let Err(e) = handle_conn(stream, engine).await {
30                tracing::debug!(error = %e, "conexión terminada");
31            }
32        });
33    }
34}
35
36/// Atiende una conexión: lee requests hasta EOF, responde cada uno.
37async fn handle_conn<E>(mut stream: UnixStream, engine: Arc<E>) -> std::io::Result<()>
38where
39    E: Engine,
40{
41    loop {
42        let req: DaemonRequest = match read_frame(&mut stream).await {
43            Ok(r) => r,
44            // EOF limpio = el cliente cerró; no es error.
45            Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => return Ok(()),
46            Err(e) => return Err(e),
47        };
48        let resp = dispatch(&*engine, req).await;
49        write_frame(&mut stream, &resp).await?;
50    }
51}
52
53/// Traduce un request a la llamada `Engine` correspondiente.
54async fn dispatch<E: Engine>(engine: &E, req: DaemonRequest) -> DaemonResponse {
55    match req {
56        DaemonRequest::Run(intent) => match engine.run(intent).await {
57            Ok(h) => DaemonResponse::Ran(h),
58            Err(e) => DaemonResponse::Err(e),
59        },
60        DaemonRequest::Stop { card_id, grace_ms } => {
61            match engine.stop(card_id, Duration::from_millis(grace_ms)).await {
62                Ok(()) => DaemonResponse::Stopped,
63                Err(e) => DaemonResponse::Err(e),
64            }
65        }
66        DaemonRequest::List => match engine.list().await {
67            Ok(v) => DaemonResponse::Listed(v),
68            Err(e) => DaemonResponse::Err(e),
69        },
70        DaemonRequest::Status { card_id } => match engine.status(card_id).await {
71            Ok(s) => DaemonResponse::Status(s),
72            Err(e) => DaemonResponse::Err(e),
73        },
74        DaemonRequest::Telemetry { card_id } => match engine.telemetry(card_id).await {
75            Ok(t) => DaemonResponse::Telemetry(t),
76            Err(e) => DaemonResponse::Err(e),
77        },
78    }
79}
80
81#[cfg(test)]
82mod tests {
83    use super::*;
84    use sandokan_core::{Engine, EngineError};
85    use sandokan_local::LocalEngine;
86    use ulid::Ulid;
87
88    #[tokio::test]
89    async fn roundtrip_list_and_notfound() {
90        let dir = tempfile::tempdir().unwrap();
91        let sock = dir.path().join("sandokan.sock");
92
93        let engine = Arc::new(LocalEngine::new());
94        let sock_srv = sock.clone();
95        let srv = tokio::spawn(async move {
96            let _ = serve(engine, &sock_srv).await;
97        });
98
99        // Espera a que el socket esté listo.
100        for _ in 0..50 {
101            if sock.exists() {
102                break;
103            }
104            tokio::time::sleep(Duration::from_millis(10)).await;
105        }
106
107        let client = crate::DaemonEngine::new(sock.clone());
108        assert!(client.is_reachable().await);
109
110        // list() sobre engine vacío → vacío.
111        let listed = client.list().await.expect("list");
112        assert!(listed.is_empty());
113
114        // status() de un id desconocido → NotFound propagado por el wire.
115        let unknown = Ulid::new();
116        match client.status(unknown).await {
117            Err(EngineError::NotFound(id)) => assert_eq!(id, unknown),
118            other => panic!("esperaba NotFound, fue {other:?}"),
119        }
120
121        srv.abort();
122    }
123}