Phase 7b: broker broadcast + dashboard SSE message-flow tail; pkgs.git in module

Cargo.toml

@@ -30,7 +30,9 @@ tokio = { version = "1", features = [
   "process",
   "rt-multi-thread",
   "signal",
+  "sync",
   "time",
 ] }
+tokio-stream = { version = "0.1", features = ["sync"] }
 tracing = "0.1"
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }

hive-c0re/Cargo.toml

@@ -16,5 +16,6 @@ serde.workspace = true
 serde_json.workspace = true
 similar.workspace = true
 tokio.workspace = true
+tokio-stream.workspace = true
 tracing.workspace = true
 tracing-subscriber.workspace = true

hive-c0re/src/broker.rs

@@ -1,4 +1,5 @@
-//! Sqlite-backed message broker. Survives `hive-c0re` restart.
+//! Sqlite-backed message broker. Survives `hive-c0re` restart, and taps every
+//! send/recv onto a broadcast channel so the dashboard can stream it.
 
 use std::path::Path;
 use std::sync::Mutex;
@@ -7,6 +8,8 @@ use std::time::{SystemTime, UNIX_EPOCH};
 use anyhow::{Context, Result};
 use hive_sh4re::Message;
 use rusqlite::{Connection, OptionalExtension, params};
+use serde::Serialize;
+use tokio::sync::broadcast;
 
 const SCHEMA: &str = r"
 CREATE TABLE IF NOT EXISTS messages (
@@ -21,8 +24,30 @@ CREATE INDEX IF NOT EXISTS idx_messages_undelivered
     ON messages (recipient, id) WHERE delivered_at IS NULL;
 ";
 
+/// Capacity of the live event channel. Slow subscribers (e.g. an idle browser)
+/// may drop events past this; we send a `lagged` notice in their stream.
+const EVENT_CHANNEL: usize = 256;
+
+#[derive(Debug, Clone, Serialize)]
+#[serde(rename_all = "snake_case", tag = "kind")]
+pub enum MessageEvent {
+    Sent {
+        from: String,
+        to: String,
+        body: String,
+        at: i64,
+    },
+    Delivered {
+        from: String,
+        to: String,
+        body: String,
+        at: i64,
+    },
+}
+
 pub struct Broker {
     conn: Mutex<Connection>,
+    events: broadcast::Sender<MessageEvent>,
 }
 
 impl Broker {
@@ -31,20 +56,33 @@ impl Broker {
             std::fs::create_dir_all(parent)
                 .with_context(|| format!("create db parent {}", parent.display()))?;
         }
-        let conn =
+        let conn = Connection::open(path)
-            Connection::open(path).with_context(|| format!("open broker db {}", path.display()))?;
+            .with_context(|| format!("open broker db {}", path.display()))?;
         conn.execute_batch(SCHEMA).context("apply broker schema")?;
+        let (events, _) = broadcast::channel(EVENT_CHANNEL);
         Ok(Self {
             conn: Mutex::new(conn),
+            events,
         })
     }
 
+    pub fn subscribe(&self) -> broadcast::Receiver<MessageEvent> {
+        self.events.subscribe()
+    }
+
     pub fn send(&self, message: &Message) -> Result<()> {
         let conn = self.conn.lock().unwrap();
         conn.execute(
             "INSERT INTO messages (sender, recipient, body, sent_at) VALUES (?1, ?2, ?3, ?4)",
             params![message.from, message.to, message.body, now_unix()],
         )?;
+        drop(conn);
+        let _ = self.events.send(MessageEvent::Sent {
+            from: message.from.clone(),
+            to: message.to.clone(),
+            body: message.body.clone(),
+            at: now_unix(),
+        });
         Ok(())
     }
 
@@ -68,6 +106,13 @@ impl Broker {
             "UPDATE messages SET delivered_at = ?1 WHERE id = ?2",
             params![now_unix(), id],
         )?;
+        drop(conn);
+        let _ = self.events.send(MessageEvent::Delivered {
+            from: from.clone(),
+            to: to.clone(),
+            body: body.clone(),
+            at: now_unix(),
+        });
         Ok(Some(Message { from, to, body }))
     }
 }

hive-c0re/src/dashboard.rs

@@ -2,6 +2,7 @@
 //! container's web UI), pending approvals (with unified diff vs the applied
 //! repo, plus approve/deny buttons), and the manager.
 
+use std::convert::Infallible;
 use std::fmt::Write as _;
 use std::net::SocketAddr;
 use std::path::Path;
@@ -12,11 +13,16 @@ use axum::{
     Router,
     extract::{Path as AxumPath, State},
     http::{HeaderMap, StatusCode},
-    response::{Html, IntoResponse, Redirect, Response},
+    response::{
+        Html, IntoResponse, Redirect, Response,
+        sse::{Event, KeepAlive, Sse},
+    },
     routing::{get, post},
 };
 use hive_sh4re::Approval;
 use tokio::process::Command;
+use tokio_stream::wrappers::BroadcastStream;
+use tokio_stream::{Stream, StreamExt};
 
 use crate::actions;
 use crate::coordinator::Coordinator;
@@ -34,6 +40,7 @@ pub async fn serve(port: u16, coord: Arc<Coordinator>) -> Result<()> {
         .route("/", get(index))
         .route("/approve/{id}", post(post_approve))
         .route("/deny/{id}", post(post_deny))
+        .route("/messages/stream", get(messages_stream))
         .with_state(AppState { coord });
     let addr = SocketAddr::from(([0, 0, 0, 0], port));
     let listener = tokio::net::TcpListener::bind(addr)
@@ -56,11 +63,24 @@ async fn index(headers: HeaderMap, State(state): State<AppState>) -> Html<String
     let approvals_html = render_approvals(&approvals).await;
 
     Html(format!(
-        "<!doctype html>\n<html lang=\"en\">\n<head>\n<meta charset=\"utf-8\">\n<title>hyperhive // h1ve-c0re</title>\n{STYLE}\n</head>\n<body>\n{BANNER}\n{containers}\n{approvals_html}\n{FOOTER}\n</body>\n</html>\n",
+        "<!doctype html>\n<html lang=\"en\">\n<head>\n<meta charset=\"utf-8\">\n<title>hyperhive // h1ve-c0re</title>\n{STYLE}\n</head>\n<body>\n{BANNER}\n{containers}\n{approvals_html}\n{MSG_FLOW}\n{FOOTER}\n{MSG_FLOW_JS}\n</body>\n</html>\n",
         containers = render_containers(&containers, &hostname),
     ))
 }
 
+async fn messages_stream(
+    State(state): State<AppState>,
+) -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
+    let rx = state.coord.broker.subscribe();
+    let stream = BroadcastStream::new(rx).filter_map(|res| {
+        // Drop lagged events. Browsers reconnect; nothing to do here.
+        let event = res.ok()?;
+        let json = serde_json::to_string(&event).ok()?;
+        Some(Ok(Event::default().data(json)))
+    });
+    Sse::new(stream).keep_alive(KeepAlive::default())
+}
+
 async fn post_approve(
     State(state): State<AppState>,
     AxumPath(id): AxumPath<i64>,
@@ -185,6 +205,44 @@ const BANNER: &str = r#"<pre class="banner">
 ░▒▓█▓▒░  HYPERHIVE  ░▒▓█▓▒░  HIVE-C0RE  ░▒▓█▓▒░  WE ARE THE WIRED  ░▒▓█▓▒░
 </pre>"#;
 
+const MSG_FLOW: &str = r#"<h2>◆ MESS4GE FL0W ◆</h2>
+<div class="divider">══════════════════════════════════════════════════════════════</div>
+<p class="meta">live tail — newest at the top. tap on every <code>send</code> / <code>recv</code> through the broker.</p>
+<div id="msgflow" class="msgflow"><span class="meta">connecting…</span></div>"#;
+
+const MSG_FLOW_JS: &str = r#"<script>
+(() => {
+  const flow = document.getElementById('msgflow');
+  if (!flow) return;
+  flow.innerHTML = '';
+  const es = new EventSource('/messages/stream');
+  const MAX_ROWS = 200;
+  const tsFmt = (n) => new Date(n * 1000).toISOString().slice(11, 19);
+  const esc = (s) => s.replace(/[&<>]/g, (c) => ({'&':'&amp;','<':'&lt;','>':'&gt;'}[c]));
+  es.onmessage = (e) => {
+    let m;
+    try { m = JSON.parse(e.data); } catch { return; }
+    const row = document.createElement('div');
+    row.className = 'msgrow ' + m.kind;
+    const kind = m.kind === 'sent' ? '→' : '✓';
+    row.innerHTML =
+      '<span class="msg-ts">' + tsFmt(m.at) + '</span>' +
+      '<span class="msg-arrow">' + kind + '</span>' +
+      '<span class="msg-from">' + esc(m.from) + '</span>' +
+      '<span class="msg-sep">→</span>' +
+      '<span class="msg-to">' + esc(m.to) + '</span>' +
+      '<span class="msg-body">' + esc(m.body) + '</span>';
+    flow.insertBefore(row, flow.firstChild);
+    while (flow.childNodes.length > MAX_ROWS) flow.removeChild(flow.lastChild);
+  };
+  es.onerror = () => {
+    flow.insertBefore(Object.assign(document.createElement('div'), {
+      className: 'msgrow meta', textContent: '[connection lost — retrying]'
+    }), flow.firstChild);
+  };
+})();
+</script>"#;
+
 const FOOTER: &str = r#"<footer>
 <div class="divider">══════════════════════════════════════════════════════════════</div>
 <p>▲△▲ <a href="https://git.berlin.ccc.de/vinzenz/hyperhive">hyperhive</a> ▲△▲ hive-c0re on this host ▲△▲</p>
@@ -309,6 +367,24 @@ const STYLE: &str = r#"
     color: var(--fg);
     white-space: pre;
   }
+  .msgflow {
+    background: var(--bg-elev);
+    border: 1px solid var(--border);
+    padding: 0.8em;
+    font-size: 0.85em;
+    line-height: 1.5;
+    max-height: 32em;
+    overflow-y: auto;
+  }
+  .msgrow { display: grid; grid-template-columns: auto auto auto auto auto 1fr; gap: 0.6em; align-items: baseline; padding: 0.1em 0; }
+  .msgrow.sent .msg-arrow { color: var(--cyan); }
+  .msgrow.delivered .msg-arrow { color: var(--green); }
+  .msg-ts { color: var(--muted); font-size: 0.85em; }
+  .msg-arrow { font-weight: bold; }
+  .msg-from { color: var(--amber); }
+  .msg-sep { color: var(--muted); }
+  .msg-to { color: var(--pink); }
+  .msg-body { color: var(--fg); white-space: pre-wrap; word-break: break-word; }
   footer {
     margin-top: 4em;
     text-align: center;

nix/modules/hive-c0re.nix

@@ -34,7 +34,10 @@ in
   };
 
   config = lib.mkIf cfg.enable {
-    environment.systemPackages = [ cfg.package ];
+    environment.systemPackages = [
+      cfg.package
+      pkgs.git
+    ];
 
     # Dashboard + per-container web UIs share the host's network namespace and
     # need their ports reachable. Dashboard: `cfg.dashboardPort` (default 7000).