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_card::card_themed;
38use yahweh_widget_theme_switcher::theme_switcher;
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        let header = div()
174            .flex()
175            .flex_row()
176            .items_center()
177            .px(px(16.))
178            .py(px(12.))
179            .bg(theme.bg_panel.clone())
180            .border_b_1()
181            .border_color(theme.border)
182            .text_color(text)
183            .text_size(px(14.))
184            .child(div().flex_grow().child(header_text))
185            .child(theme_switcher(cx));
186
187        // Banner permanente debajo del header con el estado actual.
188        // Severidad acorde al kind.
189        let status_banner = match &self.state {
190            ProbeState::Pending => None,
191            ProbeState::Down { reason } => Some(banner_themed(
192                cx,
193                Banner::Error,
194                SharedString::from(format!("Broker DOWN — {reason}")),
195            )),
196            ProbeState::UpNoProvider { .. } => Some(banner_themed(
197                cx,
198                Banner::Warning,
199                SharedString::from("Broker UP, sin provider para el flow"),
200            )),
201            ProbeState::UpWithProvider { .. } => Some(banner_themed(
202                cx,
203                Banner::Success,
204                SharedString::from("Broker UP, provider matcheado"),
205            )),
206        };
207
208        let body = div()
209            .flex()
210            .flex_col()
211            .gap(px(8.))
212            .px(px(16.))
213            .py(px(16.))
214            .child(state_card(cx, &self.state, text, text_dim, accent_up,
215                accent_partial, accent_down, accent_pending));
216
217        div()
218            .flex()
219            .flex_col()
220            .size_full()
221            .bg(bg)
222            .child(header)
223            .when_some(status_banner, |d, b| d.child(b))
224            .child(body)
225    }
226}
227
228#[allow(clippy::too_many_arguments)]
229fn state_card(
230    cx: &mut Context<Explorer>,
231    state: &ProbeState,
232    text: gpui::Hsla,
233    text_dim: gpui::Hsla,
234    accent_up: gpui::Rgba,
235    accent_partial: gpui::Rgba,
236    accent_down: gpui::Rgba,
237    accent_pending: gpui::Rgba,
238) -> impl IntoElement {
239    let (label, accent, value, description): (&str, gpui::Rgba, String, String) = match state {
240        ProbeState::Pending => (
241            "Estado",
242            accent_pending,
243            "PENDING".into(),
244            "esperando primer probe…".into(),
245        ),
246        ProbeState::Down { reason } => (
247            "Estado",
248            accent_down,
249            "DOWN".into(),
250            format!("connect failed: {reason}"),
251        ),
252        ProbeState::UpNoProvider { flow } => (
253            "Estado",
254            accent_partial,
255            "UP / NO PROVIDER".into(),
256            format!("broker reachable; sin productor para flow `{flow}`"),
257        ),
258        ProbeState::UpWithProvider {
259            flow,
260            producer_socket,
261        } => (
262            "Estado",
263            accent_up,
264            "UP / PROVIDER".into(),
265            format!(
266                "flow `{flow}` matcheado en producer socket: {}",
267                producer_socket.display()
268            ),
269        ),
270    };
271
272    card_themed(cx)
273        .border_l_4()
274        .border_color(accent)
275        .child(
276            div()
277                .text_color(accent)
278                .text_size(px(11.))
279                .child(SharedString::from(label.to_string())),
280        )
281        .child(
282            div()
283                .text_color(text)
284                .text_size(px(28.))
285                .child(SharedString::from(value)),
286        )
287        .child(
288            div()
289                .text_color(text_dim)
290                .text_size(px(11.))
291                .child(SharedString::from(description)),
292        )
293}
294
295#[cfg(test)]
296mod tests {
297    use super::*;
298
299    #[test]
300    fn pending_is_default_state_at_boot() {
301        // El estado inicial DEBE ser Pending — sino el banner del
302        // header arrancaría con un Error o Warning sin haber probado
303        // (mensaje engañoso).
304        let s = ProbeState::Pending;
305        assert!(matches!(s, ProbeState::Pending));
306    }
307
308    #[test]
309    fn poll_and_probe_constants_are_sane() {
310        // Sanity: el timeout del probe DEBE ser menor que el
311        // intervalo de polling, sino los probes se solapan.
312        assert!(PROBE_TIMEOUT < POLL_INTERVAL);
313        // El intervalo no debería ser tan corto que sature al broker.
314        assert!(POLL_INTERVAL >= Duration::from_secs(2));
315    }
316}