tawasuyu/brahman

1//! Servidor de handshake. Listener Unix socket → sesiones por conexión.
2
3use std::collections::HashMap;
4use std::path::{Path, PathBuf};
5use std::sync::Arc;
6use std::time::{SystemTime, UNIX_EPOCH};
7
8use brahman_card::{ResolvedCard, CARD_SCHEMA_VERSION};
9use tokio::net::{UnixListener, UnixStream};
10use tokio::sync::Mutex;
11use tracing::{debug, warn};
12use ulid::Ulid;
13
14use crate::codec::{read_frame, write_frame};
15use crate::messages::{Farewell, Frame, HandshakeError, Hello, HelloAck, Ping, Pong, SessionId};
16
17/// Tabla de sesiones vivas indexada por `SessionId`.
18pub type SessionRegistry = Arc<Mutex<HashMap<SessionId, ResolvedCard>>>;
19
20/// Configuración del servidor.
21#[derive(Debug, Clone)]
22pub struct ServerConfig {
23    /// `true` si el Init está atado al servidor (se reporta en `HelloAck`).
24    pub init_attached: bool,
25}
26
27impl Default for ServerConfig {
28    fn default() -> Self {
29        Self {
30            init_attached: false,
31        }
32    }
33}
34
35/// Servidor de handshake escuchando en un Unix socket.
36pub struct Server {
37    listener: UnixListener,
38    socket_path: PathBuf,
39    sessions: SessionRegistry,
40    config: ServerConfig,
41}
42
43impl Server {
44    /// Crea el listener en `path`. Si el archivo existe, lo elimina (asume
45    /// que es un socket stale de una sesión previa).
46    pub fn bind(path: impl Into<PathBuf>, config: ServerConfig) -> std::io::Result<Self> {
47        let socket_path = path.into();
48        if socket_path.exists() {
49            std::fs::remove_file(&socket_path)?;
50        }
51        if let Some(parent) = socket_path.parent() {
52            if !parent.as_os_str().is_empty() {
53                std::fs::create_dir_all(parent)?;
54            }
55        }
56        let listener = UnixListener::bind(&socket_path)?;
57        Ok(Self {
58            listener,
59            socket_path,
60            sessions: Arc::new(Mutex::new(HashMap::new())),
61            config,
62        })
63    }
64
65    /// Devuelve la ruta del socket (útil para clientes en el mismo proceso).
66    pub fn socket_path(&self) -> &Path {
67        &self.socket_path
68    }
69
70    /// Vista compartida del registro de sesiones — útil para el Init/Admin
71    /// para inspeccionar quién está conectado.
72    pub fn sessions(&self) -> SessionRegistry {
73        self.sessions.clone()
74    }
75
76    /// Acepta UNA conexión, devuelve la `Session` lista para `handle()`.
77    /// No corre el handler — eso es responsabilidad del llamante.
78    pub async fn accept_one(&self) -> std::io::Result<Session> {
79        let (stream, _addr) = self.listener.accept().await?;
80        Ok(Session {
81            stream,
82            sessions: self.sessions.clone(),
83            config: self.config.clone(),
84        })
85    }
86
87    /// Loop de aceptación: cada conexión se despacha en una task separada.
88    /// Vive hasta que el listener falle o el caller drop el future.
89    pub async fn run(self) -> std::io::Result<()> {
90        loop {
91            let session = self.accept_one().await?;
92            tokio::spawn(async move {
93                if let Err(e) = session.handle().await {
94                    warn!(error = %e, "session terminó con error");
95                }
96            });
97        }
98    }
99}
100
101impl Drop for Server {
102    fn drop(&mut self) {
103        // Limpieza best-effort del socket. Si falla, log y seguir.
104        if let Err(e) = std::fs::remove_file(&self.socket_path) {
105            if e.kind() != std::io::ErrorKind::NotFound {
106                warn!(path = %self.socket_path.display(), error = %e, "no se pudo limpiar socket");
107            }
108        }
109    }
110}
111
112/// Conexión individual aceptada por el servidor.
113pub struct Session {
114    stream: UnixStream,
115    sessions: SessionRegistry,
116    config: ServerConfig,
117}
118
119impl Session {
120    /// Procesa la conexión hasta `Farewell` o EOF: handshake + loop de pings.
121    pub async fn handle(mut self) -> std::io::Result<()> {
122        let session_id = match self.do_handshake().await? {
123            Some(id) => id,
124            None => return Ok(()), // hello rechazado, conexión cerrada
125        };
126
127        loop {
128            let frame = match read_frame(&mut self.stream).await {
129                Ok(f) => f,
130                Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => {
131                    debug!(session = %session_id, "cliente cerró conexión sin Farewell");
132                    self.sessions.lock().await.remove(&session_id);
133                    return Ok(());
134                }
135                Err(e) => return Err(e),
136            };
137            match frame {
138                Frame::Ping(Ping { session }) if session == session_id => {
139                    let pong = Pong {
140                        timestamp_ms: now_ms(),
141                    };
142                    write_frame(&mut self.stream, &Frame::Pong(pong)).await?;
143                }
144                Frame::Ping(_) => {
145                    write_frame(
146                        &mut self.stream,
147                        &Frame::Error(HandshakeError::Unauthorized(
148                            "session-id no coincide".into(),
149                        )),
150                    )
151                    .await?;
152                }
153                Frame::Farewell(Farewell { session }) => {
154                    if session == session_id {
155                        self.sessions.lock().await.remove(&session_id);
156                    }
157                    return Ok(());
158                }
159                _ => {
160                    // Frame inesperado en estado post-handshake.
161                    write_frame(
162                        &mut self.stream,
163                        &Frame::Error(HandshakeError::Rejected(
164                            "frame inesperado tras handshake".into(),
165                        )),
166                    )
167                    .await?;
168                }
169            }
170        }
171    }
172
173    /// Lee el Hello, valida, registra la sesión y emite HelloAck.
174    /// Devuelve `Some(session_id)` si el handshake fue exitoso.
175    async fn do_handshake(&mut self) -> std::io::Result<Option<SessionId>> {
176        let frame = read_frame(&mut self.stream).await?;
177        let hello = match frame {
178            Frame::Hello(h) => h,
179            _ => {
180                write_frame(
181                    &mut self.stream,
182                    &Frame::Error(HandshakeError::Rejected(
183                        "primer frame debe ser Hello".into(),
184                    )),
185                )
186                .await?;
187                return Ok(None);
188            }
189        };
190
191        if let Some(err) = self.validate_hello(&hello) {
192            write_frame(&mut self.stream, &Frame::Error(err)).await?;
193            return Ok(None);
194        }
195
196        let resolved = ResolvedCard::from_agnostic(hello.card);
197        let session_id = Ulid::new();
198        self.sessions
199            .lock()
200            .await
201            .insert(session_id, resolved);
202
203        let ack = HelloAck {
204            server_version: crate::HANDSHAKE_VERSION.to_string(),
205            protocol_version: brahman_card::PROTOCOL_VERSION.to_string(),
206            session: session_id,
207            init_attached: self.config.init_attached,
208        };
209        write_frame(&mut self.stream, &Frame::HelloAck(ack)).await?;
210        debug!(session = %session_id, "handshake completado");
211        Ok(Some(session_id))
212    }
213
214    /// Validaciones que el servidor aplica al Hello del cliente.
215    fn validate_hello(&self, hello: &Hello) -> Option<HandshakeError> {
216        if hello.schema_version != CARD_SCHEMA_VERSION {
217            return Some(HandshakeError::SchemaMismatch {
218                client: hello.schema_version,
219                server: CARD_SCHEMA_VERSION,
220            });
221        }
222        if hello.protocol_version != brahman_card::PROTOCOL_VERSION {
223            return Some(HandshakeError::ProtocolMismatch(format!(
224                "cliente={}, servidor={}",
225                hello.protocol_version,
226                brahman_card::PROTOCOL_VERSION
227            )));
228        }
229        if let Err(e) = hello.card.validate() {
230            return Some(HandshakeError::InvalidCard(e.to_string()));
231        }
232        None
233    }
234}
235
236fn now_ms() -> u64 {
237    SystemTime::now()
238        .duration_since(UNIX_EPOCH)
239        .map(|d| d.as_millis() as u64)
240        .unwrap_or(0)
241}