tawasuyu/brahman

1//! `DaemonEngine` — cliente que implementa `Engine` sobre el wire.
2
3use crate::protocol::{read_frame, write_frame, DaemonRequest, DaemonResponse};
4use async_trait::async_trait;
5use sandokan_core::{Engine, EngineError, ExecHandle, Intent, TelemetryFrame};
6use sandokan_lifecycle::LifecycleState;
7use std::path::PathBuf;
8use std::time::Duration;
9use tokio::net::UnixStream;
10use ulid::Ulid;
11
12/// Engine que delega cada operación a un daemon vía Unix socket.
13///
14/// Stateless: cada llamada abre una conexión, hace un round-trip y la
15/// cierra. Simple y robusto; si el daemon no está, las llamadas fallan
16/// con `EngineError::Transport`.
17pub struct DaemonEngine {
18    socket_path: PathBuf,
19}
20
21impl DaemonEngine {
22    /// Crea un cliente apuntando al socket dado.
23    pub fn new(socket_path: impl Into<PathBuf>) -> Self {
24        Self { socket_path: socket_path.into() }
25    }
26
27    /// `true` si el socket existe y acepta conexiones ahora mismo.
28    pub async fn is_reachable(&self) -> bool {
29        UnixStream::connect(&self.socket_path).await.is_ok()
30    }
31
32    /// Un round-trip: conecta, envía el request, lee el response.
33    async fn roundtrip(&self, req: DaemonRequest) -> Result<DaemonResponse, EngineError> {
34        let mut stream = UnixStream::connect(&self.socket_path)
35            .await
36            .map_err(|e| EngineError::Transport(format!("connect: {e}")))?;
37        write_frame(&mut stream, &req)
38            .await
39            .map_err(|e| EngineError::Transport(format!("send: {e}")))?;
40        read_frame::<_, DaemonResponse>(&mut stream)
41            .await
42            .map_err(|e| EngineError::Transport(format!("recv: {e}")))
43    }
44}
45
46/// Un response que no corresponde al request enviado.
47fn mismatch() -> EngineError {
48    EngineError::Transport("respuesta del daemon no coincide con el request".into())
49}
50
51#[async_trait]
52impl Engine for DaemonEngine {
53    async fn run(&self, intent: Intent) -> Result<ExecHandle, EngineError> {
54        match self.roundtrip(DaemonRequest::Run(intent)).await? {
55            DaemonResponse::Ran(h) => Ok(h),
56            DaemonResponse::Err(e) => Err(e),
57            _ => Err(mismatch()),
58        }
59    }
60
61    async fn stop(&self, card_id: Ulid, grace: Duration) -> Result<(), EngineError> {
62        let req = DaemonRequest::Stop {
63            card_id,
64            grace_ms: grace.as_millis() as u64,
65        };
66        match self.roundtrip(req).await? {
67            DaemonResponse::Stopped => Ok(()),
68            DaemonResponse::Err(e) => Err(e),
69            _ => Err(mismatch()),
70        }
71    }
72
73    async fn list(&self) -> Result<Vec<ExecHandle>, EngineError> {
74        match self.roundtrip(DaemonRequest::List).await? {
75            DaemonResponse::Listed(v) => Ok(v),
76            DaemonResponse::Err(e) => Err(e),
77            _ => Err(mismatch()),
78        }
79    }
80
81    async fn status(&self, card_id: Ulid) -> Result<LifecycleState, EngineError> {
82        match self.roundtrip(DaemonRequest::Status { card_id }).await? {
83            DaemonResponse::Status(s) => Ok(s),
84            DaemonResponse::Err(e) => Err(e),
85            _ => Err(mismatch()),
86        }
87    }
88
89    async fn telemetry(&self, card_id: Ulid) -> Result<TelemetryFrame, EngineError> {
90        match self.roundtrip(DaemonRequest::Telemetry { card_id }).await? {
91            DaemonResponse::Telemetry(t) => Ok(t),
92            DaemonResponse::Err(e) => Err(e),
93            _ => Err(mismatch()),
94        }
95    }
96}