tawasuyu/brahman

1//! `brahman-broker-explorer` — probe GUI del broker brahman.
2//!
3//! Cada [`POLL_INTERVAL`] arma un Card observer agnóstico y lo
4//! manda al broker via `brahman_sidecar::await_provider_blocking`
5//! (que internamente abre tokio runtime + Unix socket + handshake).
6//! Reporta 3 estados:
7//!
8//! - **Down**: connect failed (broker no escucha en el socket).
9//! - **Up sin provider**: connect OK, pero el broker no encontró
10//!   productor para el flow probado dentro del timeout.
11//! - **Up con provider**: connect OK + el broker matcheó algo →
12//!   muestra el `producer_service_socket` recibido.
13//!
14//! Configuración via env:
15//! - `BRAHMAN_INIT_SOCKET` — path del socket del broker (default
16//!   resuelto por `brahman_handshake::transport`).
17//! - `BRAHMAN_BROKER_PROBE_FLOW` — nombre del flow probe (default
18//!   `broker-health`).
19//! - `BRAHMAN_BROKER_PROBE_TYPE` — type name del flow probe
20//!   (default `ping`).
21//!
22//! Usá un type name probable (ej. `monad-list:json`,
23//! `event-log:tail`) para detectar productores específicos del
24//! ecosistema.
25
26use std::path::PathBuf;
27use std::time::{Duration, Instant};
28
29use brahman_handshake::transport;
30use brahman_sidecar::{await_provider_blocking, build_consumer_card, ConsumerError};
31use gpui::{
32    div, prelude::*, px, App, Application, Bounds, Context, IntoElement, Render, SharedString,
33    Window, WindowBounds, WindowOptions,
34};
35use yahweh_theme::Theme;
36use yahweh_widget_banner::{banner_themed, Banner};
37use yahweh_widget_stat_card::stat_card;
38use yahweh_widget_app_header::app_header;
39
40const POLL_INTERVAL: Duration = Duration::from_secs(5);
41const PROBE_TIMEOUT: Duration = Duration::from_secs(1);
42
43fn main() {
44    Application::new().run(|cx: &mut App| {
45        Theme::install_default(cx);
46        let bounds = Bounds::centered(None, gpui::size(px(720.), px(480.)), cx);
47        cx.open_window(
48            WindowOptions {
49                window_bounds: Some(WindowBounds::Windowed(bounds)),
50                titlebar: Some(gpui::TitlebarOptions {
51                    title: Some(SharedString::from("Brahman Broker — Probe")),
52                    ..Default::default()
53                }),
54                ..Default::default()
55            },
56            |_w, cx| cx.new(Explorer::new),
57        )
58        .expect("open window");
59        cx.activate(true);
60    });
61}
62
63/// Snapshot de un probe.
64#[derive(Clone, Debug)]
65enum ProbeState {
66    /// Aún no probó (boot, primer ciclo).
67    Pending,
68    /// Connect failed → broker no responde en el path.
69    Down { reason: String },
70    /// Connect OK, sin matching producer dentro del timeout.
71    UpNoProvider { flow: String },
72    /// Connect OK, broker matcheó al menos un producer.
73    UpWithProvider {
74        flow: String,
75        producer_socket: PathBuf,
76    },
77}
78
79struct Explorer {
80    socket_path: PathBuf,
81    flow: String,
82    type_name: String,
83    state: ProbeState,
84    last_probe_ms: u64,
85    last_probe_at: Option<Instant>,
86}
87
88impl Explorer {
89    fn new(cx: &mut Context<Self>) -> Self {
90        let socket_path = transport::default_socket_path();
91        let flow = std::env::var("BRAHMAN_BROKER_PROBE_FLOW")
92            .unwrap_or_else(|_| "broker-health".to_string());
93        let type_name = std::env::var("BRAHMAN_BROKER_PROBE_TYPE")
94            .unwrap_or_else(|_| "ping".to_string());
95
96        let flow_for_loop = flow.clone();
97        let type_for_loop = type_name.clone();
98        cx.spawn(async move |this, cx| {
99            let timer = cx.background_executor().clone();
100            let bg = cx.background_executor().clone();
101            loop {
102                let card = build_consumer_card(
103                    "brahman-broker-explorer",
104                    flow_for_loop.clone(),
105                    type_for_loop.clone(),
106                );
107                let started = Instant::now();
108                // El probe es bloqueante (interno tokio runtime); va
109                // al background executor para no congelar el main.
110                let probe_flow = flow_for_loop.clone();
111                let result = bg
112                    .spawn(async move { await_provider_blocking(card, PROBE_TIMEOUT) })
113                    .await;
114                let elapsed = started.elapsed().as_millis() as u64;
115
116                let new_state = match result {
117                    Ok(socket) => ProbeState::UpWithProvider {
118                        flow: probe_flow.clone(),
119                        producer_socket: socket,
120                    },
121                    Err(ConsumerError::NoProvider { .. }) => ProbeState::UpNoProvider {
122                        flow: probe_flow.clone(),
123                    },
124                    Err(e) => ProbeState::Down {
125                        reason: e.to_string(),
126                    },
127                };
128
129                let _ = this.update(cx, |me, cx| {
130                    me.state = new_state;
131                    me.last_probe_ms = elapsed;
132                    me.last_probe_at = Some(Instant::now());
133                    cx.notify();
134                });
135
136                timer.timer(POLL_INTERVAL).await;
137            }
138        })
139        .detach();
140
141        Self {
142            socket_path,
143            flow,
144            type_name,
145            state: ProbeState::Pending,
146            last_probe_ms: 0,
147            last_probe_at: None,
148        }
149    }
150}
151
152impl Render for Explorer {
153    fn render(&mut self, _w: &mut Window, cx: &mut Context<Self>) -> impl IntoElement {
154        let theme = Theme::global(cx).clone();
155        let bg = theme.bg_app.clone();
156        let text = theme.fg_text;
157        let text_dim = theme.fg_muted;
158        // Acentos por estado: verde = arriba, amber = arriba sin
159        // provider, rojo = abajo, gris = pending.
160        let accent_up = gpui::rgb(0xa3be8c);
161        let accent_partial = gpui::rgb(0xebcb8b);
162        let accent_down = gpui::rgb(0xbf616a);
163        let accent_pending = gpui::rgb(0x6a7280);
164
165        let header_text = format!(
166            "Probe: {}  ·  flow: {}/{}  ·  reload {} ms",
167            self.socket_path.display(),
168            self.flow,
169            self.type_name,
170            self.last_probe_ms,
171        );
172
173        // Header standard via widget compartido.
174        let header = app_header(cx, header_text);
175
176        // Banner permanente debajo del header con el estado actual.
177        // Severidad acorde al kind.
178        let status_banner = match &self.state {
179            ProbeState::Pending => None,
180            ProbeState::Down { reason } => Some(banner_themed(
181                cx,
182                Banner::Error,
183                SharedString::from(format!("Broker DOWN — {reason}")),
184            )),
185            ProbeState::UpNoProvider { .. } => Some(banner_themed(
186                cx,
187                Banner::Warning,
188                SharedString::from("Broker UP, sin provider para el flow"),
189            )),
190            ProbeState::UpWithProvider { .. } => Some(banner_themed(
191                cx,
192                Banner::Success,
193                SharedString::from("Broker UP, provider matcheado"),
194            )),
195        };
196
197        let body = div()
198            .flex()
199            .flex_col()
200            .gap(px(8.))
201            .px(px(16.))
202            .py(px(16.))
203            .child(state_card(cx, &self.state, text, text_dim, accent_up,
204                accent_partial, accent_down, accent_pending));
205
206        div()
207            .flex()
208            .flex_col()
209            .size_full()
210            .bg(bg)
211            .child(header)
212            .when_some(status_banner, |d, b| d.child(b))
213            .child(body)
214    }
215}
216
217/// Wrap del `stat_card` compartido con el mapeo de
218/// `ProbeState` → (label/accent/value/description). Mantenemos
219/// este helper local porque la traducción del enum a strings es
220/// específica del explorer (no es un patrón cross-app).
221#[allow(clippy::too_many_arguments)]
222fn state_card(
223    cx: &mut Context<Explorer>,
224    state: &ProbeState,
225    text: gpui::Hsla,
226    text_dim: gpui::Hsla,
227    accent_up: gpui::Rgba,
228    accent_partial: gpui::Rgba,
229    accent_down: gpui::Rgba,
230    accent_pending: gpui::Rgba,
231) -> impl IntoElement {
232    let (accent, value, description): (gpui::Rgba, String, String) = match state {
233        ProbeState::Pending => (
234            accent_pending,
235            "PENDING".into(),
236            "esperando primer probe…".into(),
237        ),
238        ProbeState::Down { reason } => (
239            accent_down,
240            "DOWN".into(),
241            format!("connect failed: {reason}"),
242        ),
243        ProbeState::UpNoProvider { flow } => (
244            accent_partial,
245            "UP / NO PROVIDER".into(),
246            format!("broker reachable; sin productor para flow `{flow}`"),
247        ),
248        ProbeState::UpWithProvider {
249            flow,
250            producer_socket,
251        } => (
252            accent_up,
253            "UP / PROVIDER".into(),
254            format!(
255                "flow `{flow}` matcheado en producer socket: {}",
256                producer_socket.display()
257            ),
258        ),
259    };
260
261    stat_card(cx, "Estado", value, &description, accent, text, text_dim, &[])
262}
263
264#[cfg(test)]
265mod tests {
266    use super::*;
267
268    #[test]
269    fn pending_is_default_state_at_boot() {
270        // El estado inicial DEBE ser Pending — sino el banner del
271        // header arrancaría con un Error o Warning sin haber probado
272        // (mensaje engañoso).
273        let s = ProbeState::Pending;
274        assert!(matches!(s, ProbeState::Pending));
275    }
276
277    #[test]
278    fn poll_and_probe_constants_are_sane() {
279        // Sanity: el timeout del probe DEBE ser menor que el
280        // intervalo de polling, sino los probes se solapan.
281        assert!(PROBE_TIMEOUT < POLL_INTERVAL);
282        // El intervalo no debería ser tan corto que sature al broker.
283        assert!(POLL_INTERVAL >= Duration::from_secs(2));
284    }
285}