tawasuyu/brahman

1//! `brahman-sidecar` — boilerplate del cliente brahman extraído.
2//!
3//! Cualquier módulo que quiera presentarse al Init brahman pero que tenga
4//! su propio runtime (GPUI, current_thread tokio, std-thread loop, etc.)
5//! puede llamar [`spawn`] con su [`brahman_card::Card`]. Eso arma un
6//! thread aparte con un runtime tokio current_thread, conecta al Init,
7//! y mantiene la sesión viva con pings periódicos.
8//!
9//! Si el Init no está disponible, el thread loggea y termina — el módulo
10//! sigue funcionando standalone.
11//!
12//! Errores de conexión / ping se loggean vía `tracing::warn!`. Si querés
13//! capturar la salida del thread (por ejemplo para test), usá
14//! [`spawn_with_handle`] que devuelve un `JoinHandle`.
15
16#![forbid(unsafe_code)]
17#![warn(rust_2018_idioms)]
18
19use std::collections::HashMap;
20use std::sync::{mpsc, Arc, Mutex};
21use std::thread::JoinHandle;
22use std::time::Duration;
23
24use brahman_card::{ulid::Ulid, Card, WitInterface};
25use brahman_handshake::{client::Client, transport};
26use tokio::task::AbortHandle;
27use tracing::{info, warn};
28
29/// Período entre pings al Init.
30pub const DEFAULT_PING_INTERVAL: Duration = Duration::from_secs(30);
31
32/// Configuración del sidecar.
33#[derive(Debug, Clone)]
34pub struct SidecarConfig {
35    /// Card que se presenta al Init.
36    pub card: Card,
37    /// WIT interface opcional. Si es `Some`, el módulo se registra como
38    /// "consciente" (`ResolvedCard::from_conscious`).
39    pub wit: Option<WitInterface>,
40    /// Período entre pings.
41    pub ping_interval: Duration,
42}
43
44impl SidecarConfig {
45    /// Configuración agnóstica con defaults razonables (sin WIT, ping 30s).
46    pub fn new(card: Card) -> Self {
47        Self {
48            card,
49            wit: None,
50            ping_interval: DEFAULT_PING_INTERVAL,
51        }
52    }
53
54    /// Configuración consciente con WIT extraída.
55    pub fn with_wit(mut self, wit: WitInterface) -> Self {
56        self.wit = Some(wit);
57        self
58    }
59}
60
61/// Spawn fire-and-forget agnóstico. Para módulos conscientes usá
62/// [`spawn_conscious`] o [`spawn_with_handle`] con un `SidecarConfig`
63/// personalizado.
64pub fn spawn(card: Card) {
65    if let Err(e) = spawn_with_handle(SidecarConfig::new(card)) {
66        warn!(error = %e, "no se pudo spawnear el sidecar brahman");
67    }
68}
69
70/// Spawn fire-and-forget con WIT. Idéntico a [`spawn`] pero el módulo se
71/// registra como consciente en el broker.
72pub fn spawn_conscious(card: Card, wit: WitInterface) {
73    if let Err(e) = spawn_with_handle(SidecarConfig::new(card).with_wit(wit)) {
74        warn!(error = %e, "no se pudo spawnear el sidecar brahman");
75    }
76}
77
78/// Spawn devolviendo el `JoinHandle` para tests o cleanup explícito.
79pub fn spawn_with_handle(config: SidecarConfig) -> std::io::Result<JoinHandle<()>> {
80    std::thread::Builder::new()
81        .name("brahman-sidecar".into())
82        .spawn(move || run_thread(config))
83}
84
85// =====================================================================
86// SidecarPool — un solo runtime tokio compartido por N sesiones
87// =====================================================================
88
89/// Pool consolidado: un único thread con un runtime tokio
90/// `current_thread` que hostea N tasks de sidecar simultáneas.
91///
92/// Para módulos con muchas sesiones (p. ej. `nouser daemon` publicando
93/// 50+ Mónadas), evita el costo de tener un thread+runtime por cada
94/// sesión.
95///
96/// **API**:
97/// - `SidecarPool::new()` crea el pool (spawn del thread runtime).
98/// - `pool.spawn(card)` añade una sesión sin WIT.
99/// - `pool.spawn_conscious(card, wit)` añade una sesión con WIT.
100/// - `pool.spawn_with_config(config)` para configuración custom.
101///
102/// El pool se mantiene vivo mientras exista. Si el `SidecarPool`
103/// se dropea, el thread interno termina y todas las sesiones cierran.
104pub struct SidecarPool {
105    handle: tokio::runtime::Handle,
106    /// Sesiones vivas indexadas por `Card.id`. Permite que un nuevo
107    /// `spawn` con el mismo id aborte la sesión previa — útil cuando
108    /// un módulo (p. ej. `nouser daemon`) re-publica una Mónada cuya
109    /// composición cambió.
110    sessions: Arc<Mutex<HashMap<Ulid, AbortHandle>>>,
111    _thread: JoinHandle<()>,
112}
113
114impl SidecarPool {
115    /// Crea un pool nuevo. Bloquea hasta que el runtime esté listo.
116    pub fn new() -> std::io::Result<Self> {
117        let (handle_tx, handle_rx) = mpsc::sync_channel::<tokio::runtime::Handle>(0);
118        let thread = std::thread::Builder::new()
119            .name("brahman-sidecar-pool".into())
120            .spawn(move || {
121                let rt = match tokio::runtime::Builder::new_current_thread()
122                    .enable_io()
123                    .enable_time()
124                    .build()
125                {
126                    Ok(rt) => rt,
127                    Err(e) => {
128                        warn!(error = %e, "tokio runtime falló — pool muerto");
129                        return;
130                    }
131                };
132                if handle_tx.send(rt.handle().clone()).is_err() {
133                    return;
134                }
135                // Mantenemos el runtime vivo mientras existan tasks.
136                rt.block_on(std::future::pending::<()>());
137            })?;
138        let handle = handle_rx
139            .recv()
140            .map_err(|_| std::io::Error::other("pool runtime no respondió"))?;
141        Ok(Self {
142            handle,
143            sessions: Arc::new(Mutex::new(HashMap::new())),
144            _thread: thread,
145        })
146    }
147
148    /// Añade una sesión agnóstica al pool (sin WIT).
149    pub fn spawn(&self, card: Card) {
150        self.spawn_with_config(SidecarConfig::new(card));
151    }
152
153    /// Añade una sesión consciente (con WIT) al pool.
154    pub fn spawn_conscious(&self, card: Card, wit: WitInterface) {
155        self.spawn_with_config(SidecarConfig::new(card).with_wit(wit));
156    }
157
158    /// Añade una sesión con configuración custom.
159    ///
160    /// Si ya existía una sesión para el mismo `Card.id`, la previa
161    /// se aborta antes de spawnear la nueva. Esto hace `spawn`
162    /// idempotente respecto al id: re-publicar una Mónada cuya
163    /// composición cambió "refresca" la sesión en el broker.
164    pub fn spawn_with_config(&self, config: SidecarConfig) {
165        let card_id = config.card.id;
166        let join = self.handle.spawn(run_client(config));
167        let abort = join.abort_handle();
168        if let Ok(mut sessions) = self.sessions.lock() {
169            if let Some(prev) = sessions.insert(card_id, abort) {
170                prev.abort();
171            }
172        }
173    }
174
175    /// Cierra explícitamente la sesión asociada a `card_id`. No-op si
176    /// no había sesión registrada.
177    pub fn drop_session(&self, card_id: Ulid) {
178        if let Ok(mut sessions) = self.sessions.lock() {
179            if let Some(abort) = sessions.remove(&card_id) {
180                abort.abort();
181            }
182        }
183    }
184
185    /// Cantidad actual de sesiones vivas (estimada — puede haber
186    /// drift transitorio entre abort y limpieza).
187    pub fn live_sessions(&self) -> usize {
188        self.sessions.lock().map(|s| s.len()).unwrap_or(0)
189    }
190}
191
192impl Default for SidecarPool {
193    fn default() -> Self {
194        Self::new().expect("falló SidecarPool::new")
195    }
196}
197
198fn run_thread(config: SidecarConfig) {
199    let rt = match tokio::runtime::Builder::new_current_thread()
200        .enable_io()
201        .enable_time()
202        .build()
203    {
204        Ok(rt) => rt,
205        Err(e) => {
206            warn!(error = %e, "tokio runtime falló");
207            return;
208        }
209    };
210    rt.block_on(run_client(config));
211}
212
213/// Bucle async del sidecar. Público para que `SidecarPool` lo use vía
214/// `handle.spawn(run_client(...))` desde código externo al runtime.
215pub async fn run_client(config: SidecarConfig) {
216    let path = transport::default_socket_path();
217    let conscious = config.wit.is_some();
218    let mut client = match Client::connect_with(&path, config.card, config.wit).await {
219        Ok(c) => {
220            info!(
221                target: "brahman_sidecar",
222                session = %c.session(),
223                init_attached = c.server_info().init_attached,
224                server = %c.server_info().server_version,
225                conscious,
226                "attached"
227            );
228            c
229        }
230        Err(e) => {
231            warn!(
232                target: "brahman_sidecar",
233                error = %e,
234                socket = %path.display(),
235                "no conectado"
236            );
237            return;
238        }
239    };
240
241    loop {
242        tokio::time::sleep(config.ping_interval).await;
243        if let Err(e) = client.ping().await {
244            warn!(target: "brahman_sidecar", error = %e, "ping falló — terminando sidecar");
245            return;
246        }
247    }
248}