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