//! Hyperhive dashboard. Lists managed containers (with deep-links to each //! 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; use std::sync::Arc; use anyhow::{Context, Result}; use axum::{ Router, extract::{Path as AxumPath, State}, http::{HeaderMap, StatusCode}, 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; use crate::lifecycle::{self, AGENT_PREFIX, MANAGER_NAME}; const MANAGER_PORT: u16 = 8000; #[derive(Clone)] struct AppState { coord: Arc, } pub async fn serve(port: u16, coord: Arc) -> Result<()> { let app = Router::new() .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) .await .with_context(|| format!("bind dashboard on port {port}"))?; tracing::info!(%port, "dashboard listening"); axum::serve(listener, app).await?; Ok(()) } async fn index(headers: HeaderMap, State(state): State) -> Html { let host = headers .get("host") .and_then(|h| h.to_str().ok()) .unwrap_or("localhost"); let hostname = host.split(':').next().unwrap_or(host).to_owned(); let containers = lifecycle::list().await.unwrap_or_default(); let approvals = state.coord.approvals.pending().unwrap_or_default(); let approvals_html = render_approvals(&approvals).await; Html(format!( "\n\n\n\nhyperhive // h1ve-c0re\n{STYLE}\n\n\n{BANNER}\n{containers}\n{approvals_html}\n{MSG_FLOW}\n{FOOTER}\n{MSG_FLOW_JS}\n\n\n", containers = render_containers(&containers, &hostname), )) } async fn messages_stream( State(state): State, ) -> Sse>> { 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, AxumPath(id): AxumPath) -> Response { match actions::approve(&state.coord, id).await { Ok(()) => Redirect::to("/").into_response(), Err(e) => error_response(&format!("approve {id} failed: {e:#}")), } } async fn post_deny(State(state): State, AxumPath(id): AxumPath) -> Response { match actions::deny(&state.coord, id) { Ok(()) => Redirect::to("/").into_response(), Err(e) => error_response(&format!("deny {id} failed: {e:#}")), } } fn error_response(message: &str) -> Response { ( StatusCode::INTERNAL_SERVER_ERROR, Html(format!( "\n\n{STYLE}\n{BANNER}\n

◆ ERR0R ◆

\n
{message}
\n

← back

\n", message = html_escape(message), )), ) .into_response() } fn render_containers(containers: &[String], hostname: &str) -> String { let mut out = String::from( "

◆ C0NTAINERS ◆

\n
══════════════════════════════════════════════════════════════
\n", ); if containers.is_empty() { out.push_str("

▓ no managed containers ▓

\n"); return out; } out.push_str("
    \n"); for container in containers { if container == MANAGER_NAME { let _ = writeln!( out, "
  • ▓█▓▒░ {container} m1nd :{MANAGER_PORT}
    • ", ); } else if let Some(name) = container.strip_prefix(AGENT_PREFIX) { let port = lifecycle::agent_web_port(name); let _ = writeln!( out, "
  • ▒░▒░░ {name} ag3nt {container} :{port}
    • ", ); } } out.push_str("
\n"); out } async fn render_approvals(approvals: &[Approval]) -> String { let mut out = String::from( "

◆ P3NDING APPR0VALS ◆

\n
══════════════════════════════════════════════════════════════
\n", ); if approvals.is_empty() { out.push_str("

▓ queue empty ▓

\n"); return out; } out.push_str("
    \n"); for a in approvals { let sha_short = &a.commit_ref[..a.commit_ref.len().min(12)]; let diff = approval_diff(&a.agent, &a.commit_ref).await; let _ = writeln!( out, "
  • \n
    → #{id} {agent} {sha_short}\n
    \n
    \n
    \n
    diff vs applied
    {diff}
    \n
    • ", id = a.id, agent = a.agent, diff = html_escape(&diff), ); } out.push_str("
\n"); out } async fn approval_diff(agent: &str, commit_ref: &str) -> String { let applied = Coordinator::agent_applied_dir(agent).join("agent.nix"); let proposed = Coordinator::agent_proposed_dir(agent); let applied_text = std::fs::read_to_string(&applied).unwrap_or_default(); let proposed_text = match git_show(&proposed, commit_ref).await { Ok(s) => s, Err(e) => return format!("(error reading proposed agent.nix: {e:#})"), }; unified_diff(&applied_text, &proposed_text) } async fn git_show(proposed_dir: &Path, commit_ref: &str) -> Result { let out = Command::new("git") .current_dir(proposed_dir) .args(["show", &format!("{commit_ref}:agent.nix")]) .output() .await .context("git show")?; if !out.status.success() { anyhow::bail!( "git show {commit_ref}:agent.nix failed: {}", String::from_utf8_lossy(&out.stderr).trim() ); } Ok(String::from_utf8_lossy(&out.stdout).into_owned()) } fn unified_diff(applied: &str, proposed: &str) -> String { let diff = similar::TextDiff::from_lines(applied, proposed); diff.unified_diff() .context_radius(3) .header("applied", "proposed") .to_string() } fn html_escape(s: &str) -> String { s.replace('&', "&") .replace('<', "<") .replace('>', ">") } const BANNER: &str = r#""#; const MSG_FLOW: &str = r#"

◆ MESS4GE FL0W ◆

══════════════════════════════════════════════════════════════

live tail — newest at the top. tap on every send / recv through the broker.

connecting…
"#; const MSG_FLOW_JS: &str = r#""#; const FOOTER: &str = r#"
══════════════════════════════════════════════════════════════

▲△▲ hyperhive ▲△▲ hive-c0re on this host ▲△▲

"#; const STYLE: &str = r#" "#;