Cargo.lock

@@ -257,12 +257,6 @@ version = "0.3.32"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7e3450815272ef58cec6d564423f6e755e25379b217b0bc688e295ba24df6b1d"
 
-[[package]]
-name = "futures-sink"
-version = "0.3.32"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c39754e157331b013978ec91992bde1ac089843443c49cbc7f46150b0fad0893"
-
 [[package]]
 name = "futures-task"
 version = "0.3.32"
@@ -333,7 +327,6 @@ dependencies = [
  "serde_json",
  "similar",
  "tokio",
- "tokio-stream",
  "tracing",
  "tracing-subscriber",
 ]
@@ -776,31 +769,6 @@ dependencies = [
  "syn",
 ]
 
-[[package]]
-name = "tokio-stream"
-version = "0.1.18"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "32da49809aab5c3bc678af03902d4ccddea2a87d028d86392a4b1560c6906c70"
-dependencies = [
- "futures-core",
- "pin-project-lite",
- "tokio",
- "tokio-util",
-]
-
-[[package]]
-name = "tokio-util"
-version = "0.7.18"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9ae9cec805b01e8fc3fd2fe289f89149a9b66dd16786abd8b19cfa7b48cb0098"
-dependencies = [
- "bytes",
- "futures-core",
- "futures-sink",
- "pin-project-lite",
- "tokio",
-]
-
 [[package]]
 name = "tower"
 version = "0.5.3"

Cargo.toml

@@ -30,9 +30,7 @@ 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,6 +16,5 @@ 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,5 +1,4 @@
-//! Sqlite-backed message broker. Survives `hive-c0re` restart, and taps every
+//! Sqlite-backed message broker. Survives `hive-c0re` restart.
-//! send/recv onto a broadcast channel so the dashboard can stream it.
 
 use std::path::Path;
 use std::sync::Mutex;
@@ -8,8 +7,6 @@ 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 (
@@ -24,30 +21,8 @@ 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 {
@@ -56,33 +31,20 @@ impl Broker {
             std::fs::create_dir_all(parent)
                 .with_context(|| format!("create db parent {}", parent.display()))?;
         }
-        let conn = Connection::open(path)
+        let conn =
-            .with_context(|| format!("open broker db {}", path.display()))?;
+            Connection::open(path).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(())
     }
 
@@ -106,13 +68,6 @@ 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,7 +2,6 @@
 //! 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;
@@ -13,16 +12,11 @@ use axum::{
     Router,
     extract::{Path as AxumPath, State},
     http::{HeaderMap, StatusCode},
-    response::{
+    response::{Html, IntoResponse, Redirect, 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;
@@ -40,7 +34,6 @@ 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)
@@ -63,24 +56,11 @@ 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{MSG_FLOW}\n{FOOTER}\n{MSG_FLOW_JS}\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{FOOTER}\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>,
@@ -205,44 +185,6 @@ 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>
@@ -367,24 +309,6 @@ 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,10 +34,7 @@ in
   };
 
   config = lib.mkIf cfg.enable {
-    environment.systemPackages = [
+    environment.systemPackages = [ cfg.package ];
-      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).