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}
20
21/// Servidor admin escuchando en un Unix socket.
22pub struct AdminServer {
23    listener: UnixListener,
24    socket_path: PathBuf,
25    broker: Arc<Mutex<Broker>>,
26    config: AdminConfig,
27}
28
29impl AdminServer {
30    /// Crea el listener. Si `path` existe, lo elimina (asume socket stale).
31    pub fn bind(
32        path: impl Into<PathBuf>,
33        broker: Arc<Mutex<Broker>>,
34        config: AdminConfig,
35    ) -> std::io::Result<Self> {
36        let socket_path = path.into();
37        if socket_path.exists() {
38            std::fs::remove_file(&socket_path)?;
39        }
40        if let Some(parent) = socket_path.parent() {
41            if !parent.as_os_str().is_empty() {
42                std::fs::create_dir_all(parent)?;
43            }
44        }
45        let listener = UnixListener::bind(&socket_path)?;
46        Ok(Self {
47            listener,
48            socket_path,
49            broker,
50            config,
51        })
52    }
53
54    pub fn socket_path(&self) -> &Path {
55        &self.socket_path
56    }
57
58    /// Loop de aceptación: cada conexión recibe un snapshot y se cierra.
59    pub async fn run(self) -> std::io::Result<()> {
60        loop {
61            let (stream, _addr) = self.listener.accept().await?;
62            let broker = self.broker.clone();
63            let config = self.config.clone();
64            tokio::spawn(async move {
65                if let Err(e) = handle_conn(stream, broker, config).await {
66                    warn!(error = %e, "admin conn falló");
67                }
68            });
69        }
70    }
71}
72
73impl Drop for AdminServer {
74    fn drop(&mut self) {
75        if let Err(e) = std::fs::remove_file(&self.socket_path) {
76            if e.kind() != std::io::ErrorKind::NotFound {
77                warn!(path = %self.socket_path.display(), error = %e, "no se pudo limpiar admin socket");
78            }
79        }
80    }
81}
82
83async fn handle_conn(
84    mut stream: UnixStream,
85    broker: Arc<Mutex<Broker>>,
86    config: AdminConfig,
87) -> std::io::Result<()> {
88    let snapshot = build_snapshot(&broker, &config).await;
89    let mut json = serde_json::to_string(&snapshot)?;
90    json.push('\n');
91    stream.write_all(json.as_bytes()).await?;
92    stream.shutdown().await?;
93    Ok(())
94}
95
96async fn build_snapshot(broker: &Arc<Mutex<Broker>>, config: &AdminConfig) -> StatusSnapshot {
97    let b = broker.lock().await;
98    let sessions: Vec<_> = b.cards().cloned().collect();
99    let matches = b.all_matches();
100    StatusSnapshot {
101        server_version: crate::ADMIN_VERSION.to_string(),
102        protocol_version: brahman_card::PROTOCOL_VERSION.to_string(),
103        init_attached: config.init_attached,
104        sessions,
105        matches,
106    }
107}