tawasuyu/brahman

1//! `brahman-sidecar::discovery` — API reusable para que un módulo
2//! consumer encuentre proveedores vivos vía broker, sin hardcodear
3//! sockets ni reimplementar el patrón a mano.
4//!
5//! Es la generalización de `discover_producer_socket` del CLI
6//! `nouser attract --remote`: declarás el `TypeRef` que querés
7//! consumir y el broker te empuja un `MatchEvent::Available` con el
8//! `producer_service_socket` del primer proveedor matched.
9//!
10//! Pipeline:
11//! 1. `build_consumer_card(label, flow_name, type_name)` arma una
12//!    Card mínima (Ente, Oneshot, Virtual) con un input flow.
13//! 2. `await_provider(card, timeout)` se conecta al brahman-init,
14//!    espera hasta `timeout` por `MatchEvent::Available`, devuelve
15//!    el socket del proveedor electo, y envía Farewell.
16//! 3. Para mundos blocking (CLIs, tests, std-thread loops) hay
17//!    `await_provider_blocking` que arma su propio runtime
18//!    `current_thread`.
19//!
20//! Quién elige al proveedor es el broker, no este módulo. Si el
21//! broker tiene `priority_contexts` activo, podés cambiar de
22//! proveedor sin tocar el consumer; el matching dinámico se respeta.
23
24use std::path::PathBuf;
25use std::time::{Duration, Instant};
26
27use brahman_card::{
28    Card, CardKind, Flow, Flows, Lifecycle, Payload, Priority, Supervision, TypeRef,
29};
30use brahman_handshake::client::{Client, ClientError};
31use brahman_handshake::messages::MatchEventKind;
32use brahman_handshake::transport;
33
34#[derive(Debug, thiserror::Error)]
35pub enum ConsumerError {
36    #[error("no se pudo conectar al init en {socket}: {source}")]
37    Connect {
38        socket: PathBuf,
39        #[source]
40        source: ClientError,
41    },
42    #[error("error en cliente brahman: {0}")]
43    Client(#[from] ClientError),
44    #[error("timeout {timeout:?} sin proveedor disponible para flow '{flow}' (type '{type_ref}')")]
45    NoProvider {
46        flow: String,
47        type_ref: String,
48        timeout: Duration,
49    },
50    #[error("no se pudo crear runtime tokio: {0}")]
51    Runtime(String),
52}
53
54/// Construye una Card mínima de consumer que declara un input flow
55/// con el `TypeRef::Primitive { name }` solicitado. Usá esto para
56/// el caso común; si necesitás algo más rico (output flows,
57/// permissions, references) construí la Card a mano y pasala a
58/// [`await_provider`] directamente.
59pub fn build_consumer_card(
60    consumer_label: impl Into<String>,
61    flow_name: impl Into<String>,
62    type_name: impl Into<String>,
63) -> Card {
64    Card {
65        payload: Payload::Virtual,
66        supervision: Supervision::OneShot,
67        lifecycle: Lifecycle::Oneshot,
68        priority: Priority::Normal,
69        kind: CardKind::Ente,
70        flow: Flows {
71            input: vec![Flow {
72                name: flow_name.into(),
73                ty: TypeRef::Primitive {
74                    name: type_name.into(),
75                },
76                pin_to: None,
77            }],
78            output: vec![],
79        },
80        ..Card::new(consumer_label)
81    }
82}
83
84/// Conecta al brahman-init, registra `consumer_card`, espera el
85/// primer `MatchEvent::Available` y devuelve el `producer_service_socket`
86/// que el broker emitió. Cierra la sesión con Farewell antes de
87/// retornar (best-effort).
88///
89/// La `consumer_card` debe declarar al menos un `flow.input`; si no,
90/// el broker no puede hacer matching y el await siempre dará timeout.
91pub async fn await_provider(
92    consumer_card: Card,
93    timeout: Duration,
94) -> Result<PathBuf, ConsumerError> {
95    let init_path = transport::default_socket_path();
96
97    // Capturamos descriptor para el mensaje de error antes de mover
98    // la card al cliente.
99    let (flow_name, type_ref_name) = describe_first_input(&consumer_card);
100
101    let mut client = Client::connect(&init_path, consumer_card)
102        .await
103        .map_err(|source| ConsumerError::Connect {
104            socket: init_path.clone(),
105            source,
106        })?;
107
108    let deadline = Instant::now() + timeout;
109    let socket = loop {
110        let remaining = deadline.saturating_duration_since(Instant::now());
111        if remaining.is_zero() {
112            break None;
113        }
114        match client.await_event(remaining).await? {
115            Some(ev) if ev.kind == MatchEventKind::Available => {
116                break ev.producer_service_socket;
117            }
118            Some(_) => continue, // Lost u otros: seguir esperando hasta el deadline
119            None => break None,
120        }
121    };
122
123    let _ = client.farewell().await; // best-effort cleanup
124
125    socket.ok_or(ConsumerError::NoProvider {
126        flow: flow_name,
127        type_ref: type_ref_name,
128        timeout,
129    })
130}
131
132/// Wrapper bloqueante de [`await_provider`]. Crea un runtime tokio
133/// `current_thread` efímero y bloquea el thread llamador. Útil para
134/// CLIs, tests y módulos std-thread (p. ej. el frontend GPUI antes
135/// de tener su propio runtime async).
136pub fn await_provider_blocking(
137    consumer_card: Card,
138    timeout: Duration,
139) -> Result<PathBuf, ConsumerError> {
140    let rt = tokio::runtime::Builder::new_current_thread()
141        .enable_io()
142        .enable_time()
143        .build()
144        .map_err(|e| ConsumerError::Runtime(e.to_string()))?;
145
146    rt.block_on(await_provider(consumer_card, timeout))
147}
148
149/// Conecta al brahman-init con una Card observer (sin inputs ni
150/// outputs) y pide la lista de sesiones activas. Útil para
151/// herramientas de observabilidad (broker-explorer, CLIs).
152///
153/// El observer se identifica con `observer_label`. La sesión se
154/// cierra con Farewell antes de retornar (best-effort).
155pub async fn list_sessions(
156    observer_label: impl Into<String>,
157) -> Result<brahman_handshake::messages::SessionList, ConsumerError> {
158    let init_path = transport::default_socket_path();
159    // Card mínima sin flow.input/output: el observer no participa en
160    // matching, sólo establece sesión para poder consultar.
161    let card = Card {
162        payload: Payload::Virtual,
163        supervision: Supervision::OneShot,
164        lifecycle: Lifecycle::Oneshot,
165        priority: Priority::Normal,
166        kind: CardKind::Ente,
167        flow: Flows {
168            input: vec![],
169            output: vec![],
170        },
171        ..Card::new(observer_label)
172    };
173
174    let mut client = Client::connect(&init_path, card)
175        .await
176        .map_err(|source| ConsumerError::Connect {
177            socket: init_path.clone(),
178            source,
179        })?;
180
181    let list = client.list_sessions().await?;
182    let _ = client.farewell().await;
183    Ok(list)
184}
185
186/// Wrapper bloqueante de [`list_sessions`]. Idéntico patrón a
187/// `await_provider_blocking`: runtime current_thread efímero.
188pub fn list_sessions_blocking(
189    observer_label: impl Into<String>,
190) -> Result<brahman_handshake::messages::SessionList, ConsumerError> {
191    let label = observer_label.into();
192    let rt = tokio::runtime::Builder::new_current_thread()
193        .enable_io()
194        .enable_time()
195        .build()
196        .map_err(|e| ConsumerError::Runtime(e.to_string()))?;
197    rt.block_on(list_sessions(label))
198}
199
200fn describe_first_input(card: &Card) -> (String, String) {
201    match card.flow.input.first() {
202        Some(flow) => {
203            let type_name = match &flow.ty {
204                TypeRef::Primitive { name } => name.clone(),
205                TypeRef::Wit { package, name, .. } => format!("{package}#{name}"),
206            };
207            (flow.name.clone(), type_name)
208        }
209        None => ("(sin input)".into(), "(sin tipo)".into()),
210    }
211}
212
213#[cfg(test)]
214mod tests {
215    use super::*;
216    use brahman_card::ulid::Ulid;
217
218    #[test]
219    fn builder_sets_input_flow_with_primitive_type() {
220        let c = build_consumer_card("nouser.cli", "embed-result", "json");
221        assert_eq!(c.label, "nouser.cli");
222        assert_eq!(c.kind, CardKind::Ente);
223        assert!(matches!(c.lifecycle, Lifecycle::Oneshot));
224        assert!(matches!(c.supervision, Supervision::OneShot));
225        assert_eq!(c.flow.input.len(), 1);
226        let f = &c.flow.input[0];
227        assert_eq!(f.name, "embed-result");
228        match &f.ty {
229            TypeRef::Primitive { name } => assert_eq!(name, "json"),
230            _ => panic!("expected primitive type"),
231        }
232        assert!(c.flow.output.is_empty());
233        // El builder asigna un id real (no nil) — fundamental para que
234        // el broker no colisione con otros consumers.
235        assert!(c.id != Ulid::nil(), "consumer card id no debe ser nil");
236    }
237
238    #[test]
239    fn builder_assigns_distinct_ids_per_call() {
240        let a = build_consumer_card("a", "f", "t");
241        let b = build_consumer_card("a", "f", "t");
242        assert_ne!(a.id, b.id, "cada Card debería tener id propio");
243    }
244
245    #[test]
246    fn describe_falls_back_when_no_input_flow() {
247        let mut c = build_consumer_card("x", "f", "t");
248        c.flow.input.clear();
249        let (flow, ty) = describe_first_input(&c);
250        assert_eq!(flow, "(sin input)");
251        assert_eq!(ty, "(sin tipo)");
252    }
253
254    #[test]
255    fn describe_formats_wit_type() {
256        let mut c = build_consumer_card("x", "f", "t");
257        c.flow.input[0].ty = TypeRef::Wit {
258            package: "brahman:dht".into(),
259            interface: None,
260            name: "entity-result".into(),
261        };
262        let (_, ty) = describe_first_input(&c);
263        assert_eq!(ty, "brahman:dht#entity-result");
264    }
265}