tawasuyu/brahman

1//! Listener Unix-socket que sirve [`akasha_card::query::QueryRequest`].
2//!
3//! El daemon `akasha` lo monta para que cualquier consumer (UI, CLI,
4//! otro módulo) pueda preguntarle por sus Mónadas sin pasar por
5//! brahman-admin. El path del socket viaja en el `Card.service_socket`
6//! del engine; el broker brahman lo enseña vía MatchEvent::Available
7//! cuando un consumer declara `flow.input = monad-list:json`.
8//!
9//! Wire: line-delimited JSON, single-shot por conexión. Mismo patrón
10//! que `akasha-nous` (mock/real ↔ akasha-core), reutilizado.
11//!
12//! Threading: un thread dedicado, blocking I/O. No vale la pena traer
13//! tokio acá — la frecuencia esperada es muy baja (UI poll cada 2s)
14//! y el handler es trivial (lock db → snapshot → write).
15
16use std::io::{BufRead, BufReader, Write};
17use std::os::unix::net::{UnixListener, UnixStream};
18use std::path::PathBuf;
19use std::sync::{Arc, Mutex};
20
21use akasha_card::query::{
22    EngineInfo, ErrorResponse, ListMonadsResponse, MonadView, QueryRequest,
23};
24use akasha_card::ulid::Ulid;
25
26use crate::db::MonadDb;
27
28/// Configuración del listener.
29pub struct ListenerConfig {
30    pub socket_path: PathBuf,
31    pub engine_id: Ulid,
32    pub engine_label: String,
33    /// Path del directorio que el daemon está observando, para incluir
34    /// en `EngineInfo.watching`. `None` si el daemon no observa nada.
35    pub watching: Option<PathBuf>,
36}
37
38/// Bind del socket + spawn de un thread con accept loop. Devuelve el
39/// path final (útil para confirmar) y un `JoinHandle` para shutdown
40/// explícito (drop = thread sigue, listener queda).
41///
42/// Si el socket ya existe (sesión anterior crasheada), se intenta
43/// removerlo antes del bind. Errores de bind se propagan al caller.
44pub fn spawn_listener(
45    config: ListenerConfig,
46    db: Arc<Mutex<MonadDb>>,
47) -> std::io::Result<std::thread::JoinHandle<()>> {
48    if config.socket_path.exists() {
49        let _ = std::fs::remove_file(&config.socket_path);
50    }
51    if let Some(parent) = config.socket_path.parent() {
52        std::fs::create_dir_all(parent)?;
53    }
54    let listener = UnixListener::bind(&config.socket_path)?;
55
56    let handle = std::thread::Builder::new()
57        .name("akasha-engine-listener".into())
58        .spawn(move || {
59            for conn in listener.incoming() {
60                match conn {
61                    Ok(stream) => {
62                        // Handler sincrónico inline. La frecuencia
63                        // esperada (UI poll cada N segundos) no
64                        // amerita spawn-per-connection; si en el
65                        // futuro hay carga, agregar un threadpool.
66                        if let Err(e) = handle_conn(stream, &db, &config) {
67                            eprintln!("[engine-socket] conn falló: {e}");
68                        }
69                    }
70                    Err(e) => {
71                        eprintln!("[engine-socket] accept falló: {e}");
72                    }
73                }
74            }
75        })?;
76
77    Ok(handle)
78}
79
80fn handle_conn(
81    mut stream: UnixStream,
82    db: &Arc<Mutex<MonadDb>>,
83    config: &ListenerConfig,
84) -> std::io::Result<()> {
85    let mut reader = BufReader::new(stream.try_clone()?);
86    let mut line = String::new();
87    let n = reader.read_line(&mut line)?;
88    if n == 0 {
89        return Ok(());
90    }
91
92    let resp_bytes = match serde_json::from_str::<QueryRequest>(line.trim()) {
93        Ok(QueryRequest::ListMonads) => match handle_list_monads(db, config) {
94            Ok(json) => json,
95            Err(e) => encode_error(format!("list_monads falló: {e}")),
96        },
97        Err(e) => encode_error(format!("JSON inválido: {e}")),
98    };
99
100    stream.write_all(resp_bytes.as_bytes())?;
101    stream.write_all(b"\n")?;
102    stream.flush()?;
103    let _ = stream.shutdown(std::net::Shutdown::Both);
104    Ok(())
105}
106
107fn handle_list_monads(
108    db: &Arc<Mutex<MonadDb>>,
109    config: &ListenerConfig,
110) -> Result<String, String> {
111    let db_lock = db.lock().map_err(|_| "mutex envenenado".to_string())?;
112    let monads: Vec<MonadView> = db_lock.monads().map(MonadView::from_manifest).collect();
113    let resp = ListMonadsResponse {
114        engine: EngineInfo {
115            id: config.engine_id,
116            label: config.engine_label.clone(),
117            watching: config.watching.as_ref().map(|p| p.display().to_string()),
118        },
119        monads,
120    };
121    serde_json::to_string(&resp).map_err(|e| format!("encode: {e}"))
122}
123
124fn encode_error(msg: String) -> String {
125    let err = ErrorResponse { error: msg };
126    serde_json::to_string(&err).unwrap_or_else(|_| "{\"error\":\"encode\"}".into())
127}
128
129// El cliente blocking vive en `akasha_card::query::client` — junto a
130// los wire types — para que un consumer pueda hablar con el daemon
131// importando sólo `akasha-card`, sin arrastrar el peso de
132// `akasha-core` (scanner / db / sled / notify / walkdir / blake3).
133
134#[cfg(test)]
135mod tests {
136    use super::*;
137    use crate::db::MonadDb;
138    use akasha_card::query::client as query_client;
139    use akasha_card::MonadManifest;
140    use std::time::Duration;
141
142    fn fresh_socket_path(name: &str) -> PathBuf {
143        let dir = std::env::temp_dir();
144        let unique = format!("{}-{}-{}.sock", name, std::process::id(), Ulid::new());
145        dir.join(unique)
146    }
147
148    #[test]
149    fn list_monads_roundtrip_empty() {
150        let socket = fresh_socket_path("akasha-engine-test");
151        let db = Arc::new(Mutex::new(MonadDb::new()));
152        let engine_id = Ulid::new();
153        let _h = spawn_listener(
154            ListenerConfig {
155                socket_path: socket.clone(),
156                engine_id,
157                engine_label: "test-engine".into(),
158                watching: Some(PathBuf::from("/tmp/x")),
159            },
160            db.clone(),
161        )
162        .unwrap();
163
164        // Pequeña espera para que el bind se asiente (en práctica el
165        // socket existe inmediatamente tras el bind, pero algunos FS
166        // necesitan un tick). Si esto resulta flaky, agregar un loop
167        // de wait_for(socket.exists()).
168        std::thread::sleep(Duration::from_millis(50));
169
170        let resp = query_client::list_monads(&socket, Duration::from_secs(2)).unwrap();
171        assert_eq!(resp.engine.id, engine_id);
172        assert_eq!(resp.engine.label, "test-engine");
173        assert_eq!(resp.engine.watching.as_deref(), Some("/tmp/x"));
174        assert!(resp.monads.is_empty());
175
176        let _ = std::fs::remove_file(&socket);
177    }
178
179    #[test]
180    fn list_monads_returns_views() {
181        let socket = fresh_socket_path("akasha-engine-test-views");
182        let db = Arc::new(Mutex::new(MonadDb::new()));
183        let m1 = MonadManifest::new("alpha");
184        let m2 = MonadManifest::new("beta");
185        {
186            let mut g = db.lock().unwrap();
187            g.replace_monads(vec![m1.clone(), m2.clone()]);
188        }
189        let _h = spawn_listener(
190            ListenerConfig {
191                socket_path: socket.clone(),
192                engine_id: Ulid::new(),
193                engine_label: "test".into(),
194                watching: None,
195            },
196            db.clone(),
197        )
198        .unwrap();
199        std::thread::sleep(Duration::from_millis(50));
200
201        let resp = query_client::list_monads(&socket, Duration::from_secs(2)).unwrap();
202        assert_eq!(resp.monads.len(), 2);
203        let labels: Vec<_> = resp.monads.iter().map(|m| m.label.as_str()).collect();
204        assert!(labels.contains(&"alpha"));
205        assert!(labels.contains(&"beta"));
206
207        let _ = std::fs::remove_file(&socket);
208    }
209
210    #[test]
211    fn invalid_request_returns_error_response() {
212        let socket = fresh_socket_path("akasha-engine-test-bad");
213        let db = Arc::new(Mutex::new(MonadDb::new()));
214        let _h = spawn_listener(
215            ListenerConfig {
216                socket_path: socket.clone(),
217                engine_id: Ulid::new(),
218                engine_label: "test".into(),
219                watching: None,
220            },
221            db.clone(),
222        )
223        .unwrap();
224        std::thread::sleep(Duration::from_millis(50));
225
226        // Bypass del cliente tipado: mandamos JSON inválido a mano.
227        use std::io::{BufRead, BufReader, Write};
228        let mut stream = UnixStream::connect(&socket).unwrap();
229        stream.write_all(b"not json\n").unwrap();
230        stream.flush().unwrap();
231        let mut reader = BufReader::new(stream);
232        let mut response = String::new();
233        reader.read_line(&mut response).unwrap();
234
235        assert!(
236            response.contains("\"error\""),
237            "esperaba ErrorResponse, got: {response}"
238        );
239
240        let _ = std::fs::remove_file(&socket);
241    }
242}