tawasuyu/brahman

1//! `sandokan-remote` — `RemoteEngine`: orquesta en un host remoto.
2//!
3//! Misma técnica que `DaemonEngine` pero el transporte es un canal SSH
4//! `direct-streamlocal` hacia el `sandokan.sock` del host remoto. El wire
5//! es idéntico (postcard length-prefixed) — sólo cambia el túnel, así
6//! que el código de protocolo se reusa tal cual.
7//!
8//! Requiere que el host remoto corra `sandokan daemon` escuchando en
9//! `remote_socket`.
10
11#![forbid(unsafe_code)]
12
13use async_trait::async_trait;
14use brahman_ssh_multiplex::{SshConfig, SshSession};
15use sandokan_core::{Engine, EngineError, ExecHandle, Intent, TelemetryFrame};
16use sandokan_daemon::{read_frame, write_frame, DaemonRequest, DaemonResponse};
17use sandokan_lifecycle::LifecycleState;
18use std::time::Duration;
19use ulid::Ulid;
20
21/// Engine que delega a un daemon sandokan en un host remoto, tunelando
22/// el wire sobre SSH. La sesión SSH maestra se mantiene; cada operación
23/// abre un canal `direct-streamlocal` nuevo (multiplexado, barato).
24pub struct RemoteEngine {
25    session: SshSession,
26    remote_socket: String,
27}
28
29impl RemoteEngine {
30    /// Conecta por SSH al host y prepara el túnel al socket del daemon.
31    pub async fn connect(
32        ssh: &SshConfig,
33        remote_socket: impl Into<String>,
34    ) -> Result<Self, EngineError> {
35        let session = SshSession::connect(ssh)
36            .await
37            .map_err(|e| EngineError::Transport(format!("ssh connect: {e}")))?;
38        Ok(Self { session, remote_socket: remote_socket.into() })
39    }
40
41    /// Construye un `RemoteEngine` sobre una `SshSession` ya establecida
42    /// (permite compartir la conexión maestra con otros consumidores).
43    pub fn with_session(session: SshSession, remote_socket: impl Into<String>) -> Self {
44        Self { session, remote_socket: remote_socket.into() }
45    }
46
47    async fn roundtrip(&self, req: DaemonRequest) -> Result<DaemonResponse, EngineError> {
48        let mut stream = self
49            .session
50            .forward_unix(&self.remote_socket)
51            .await
52            .map_err(|e| EngineError::Transport(format!("ssh forward: {e}")))?;
53        write_frame(&mut stream, &req)
54            .await
55            .map_err(|e| EngineError::Transport(format!("send: {e}")))?;
56        read_frame::<_, DaemonResponse>(&mut stream)
57            .await
58            .map_err(|e| EngineError::Transport(format!("recv: {e}")))
59    }
60}
61
62/// Un response que no corresponde al request enviado.
63fn mismatch() -> EngineError {
64    EngineError::Transport("respuesta remota no coincide con el request".into())
65}
66
67#[async_trait]
68impl Engine for RemoteEngine {
69    async fn run(&self, intent: Intent) -> Result<ExecHandle, EngineError> {
70        match self.roundtrip(DaemonRequest::Run(intent)).await? {
71            DaemonResponse::Ran(h) => Ok(h),
72            DaemonResponse::Err(e) => Err(e),
73            _ => Err(mismatch()),
74        }
75    }
76
77    async fn stop(&self, card_id: Ulid, grace: Duration) -> Result<(), EngineError> {
78        let req = DaemonRequest::Stop {
79            card_id,
80            grace_ms: grace.as_millis() as u64,
81        };
82        match self.roundtrip(req).await? {
83            DaemonResponse::Stopped => Ok(()),
84            DaemonResponse::Err(e) => Err(e),
85            _ => Err(mismatch()),
86        }
87    }
88
89    async fn list(&self) -> Result<Vec<ExecHandle>, EngineError> {
90        match self.roundtrip(DaemonRequest::List).await? {
91            DaemonResponse::Listed(v) => Ok(v),
92            DaemonResponse::Err(e) => Err(e),
93            _ => Err(mismatch()),
94        }
95    }
96
97    async fn status(&self, card_id: Ulid) -> Result<LifecycleState, EngineError> {
98        match self.roundtrip(DaemonRequest::Status { card_id }).await? {
99            DaemonResponse::Status(s) => Ok(s),
100            DaemonResponse::Err(e) => Err(e),
101            _ => Err(mismatch()),
102        }
103    }
104
105    async fn telemetry(&self, card_id: Ulid) -> Result<TelemetryFrame, EngineError> {
106        match self.roundtrip(DaemonRequest::Telemetry { card_id }).await? {
107            DaemonResponse::Telemetry(t) => Ok(t),
108            DaemonResponse::Err(e) => Err(e),
109            _ => Err(mismatch()),
110        }
111    }
112}