tawasuyu/brahman

1//! Servidor admin: emite un `StatusSnapshot` JSON por conexión y cierra.
2
3use std::path::{Path, PathBuf};
4use std::sync::Arc;
5
6use brahman_broker::Broker;
7use tokio::io::AsyncWriteExt;
8use tokio::net::{UnixListener, UnixStream};
9use tokio::sync::Mutex;
10use tracing::warn;
11
12use crate::snapshot::StatusSnapshot;
13
14/// Configuración del servidor admin.
15#[derive(Debug, Clone, Default)]
16pub struct AdminConfig {
17    /// `true` si el Init está atado al servidor que aloja este admin.
18    pub init_attached: bool,
19    /// Contexto operativo del broker, espejado en el snapshot.
20    pub current_context: Option<String>,
21}
22
23/// Servidor admin escuchando en un Unix socket.
24pub struct AdminServer {
25    listener: UnixListener,
26    socket_path: PathBuf,
27    broker: Arc<Mutex<Broker>>,
28    config: AdminConfig,
29}
30
31impl AdminServer {
32    /// Crea el listener. Si `path` existe, lo elimina (asume socket stale).
33    pub fn bind(
34        path: impl Into<PathBuf>,
35        broker: Arc<Mutex<Broker>>,
36        config: AdminConfig,
37    ) -> std::io::Result<Self> {
38        let socket_path = path.into();
39        if socket_path.exists() {
40            std::fs::remove_file(&socket_path)?;
41        }
42        if let Some(parent) = socket_path.parent() {
43            if !parent.as_os_str().is_empty() {
44                std::fs::create_dir_all(parent)?;
45            }
46        }
47        let listener = UnixListener::bind(&socket_path)?;
48        Ok(Self {
49            listener,
50            socket_path,
51            broker,
52            config,
53        })
54    }
55
56    pub fn socket_path(&self) -> &Path {
57        &self.socket_path
58    }
59
60    /// Loop de aceptación: cada conexión recibe un snapshot y se cierra.
61    pub async fn run(self) -> std::io::Result<()> {
62        loop {
63            let (stream, _addr) = self.listener.accept().await?;
64            let broker = self.broker.clone();
65            let config = self.config.clone();
66            tokio::spawn(async move {
67                if let Err(e) = handle_conn(stream, broker, config).await {
68                    warn!(error = %e, "admin conn falló");
69                }
70            });
71        }
72    }
73}
74
75impl Drop for AdminServer {
76    fn drop(&mut self) {
77        if let Err(e) = std::fs::remove_file(&self.socket_path) {
78            if e.kind() != std::io::ErrorKind::NotFound {
79                warn!(path = %self.socket_path.display(), error = %e, "no se pudo limpiar admin socket");
80            }
81        }
82    }
83}
84
85async fn handle_conn(
86    mut stream: UnixStream,
87    broker: Arc<Mutex<Broker>>,
88    config: AdminConfig,
89) -> std::io::Result<()> {
90    let snapshot = build_snapshot(&broker, &config).await;
91    let mut json = serde_json::to_string(&snapshot)?;
92    json.push('\n');
93    stream.write_all(json.as_bytes()).await?;
94    stream.shutdown().await?;
95    Ok(())
96}
97
98async fn build_snapshot(broker: &Arc<Mutex<Broker>>, config: &AdminConfig) -> StatusSnapshot {
99    let b = broker.lock().await;
100    let sessions: Vec<_> = b.cards().cloned().collect();
101    let matches = b.all_matches();
102    StatusSnapshot {
103        server_version: crate::ADMIN_VERSION.to_string(),
104        protocol_version: brahman_card::PROTOCOL_VERSION.to_string(),
105        init_attached: config.init_attached,
106        current_context: config.current_context.clone(),
107        sessions,
108        matches,
109    }
110}