tawasuyu/chasqui

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