bctx 0.1.39

//! `bctx dashboard`: a local web view of ~/.bctx/executions.db, served to this machine only.

use anyhow::{Context, Result};
use axum::extract::{Path, State};
use axum::http::{HeaderMap, StatusCode};
use axum::response::{Html, IntoResponse, Response};
use axum::routing::get;
use axum::{Json, Router};
use serde_json::{json, Value};
use std::io::{BufRead, Write};
use std::path::PathBuf;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use std::time::Duration;

const PAGE: &str = include_str!("local_dashboard.html");

#[derive(Clone)]
struct Ctx {
    token: Arc<String>,
    port: u16,
    db_path: Arc<PathBuf>,
    /// Milliseconds since the Unix epoch of the last allowed request.
    last_visit: Arc<AtomicU64>,
}

/// Only our own origin may talk to us (blocks DNS-rebinding pages in the user's browser).
fn allowed(ctx: &Ctx, token: &str, headers: &HeaderMap) -> Result<(), StatusCode> {
    let host = headers
        .get("host")
        .and_then(|h| h.to_str().ok())
        .unwrap_or("");
    if host != format!("127.0.0.1:{}", ctx.port) && host != format!("localhost:{}", ctx.port) {
        return Err(StatusCode::FORBIDDEN);
    }
    if token != ctx.token.as_str() {
        return Err(StatusCode::NOT_FOUND);
    }
    ctx.last_visit.store(now_ms(), Ordering::Relaxed);
    Ok(())
}

async fn page(State(ctx): State<Ctx>, Path(token): Path<String>, headers: HeaderMap) -> Response {
    match allowed(&ctx, &token, &headers) {
        Ok(()) => Html(PAGE).into_response(),
        Err(code) => code.into_response(),
    }
}

async fn summary(
    State(ctx): State<Ctx>,
    Path(token): Path<String>,
    headers: HeaderMap,
) -> Response {
    if let Err(code) = allowed(&ctx, &token, &headers) {
        return code.into_response();
    }
    match summary_json(&ctx.db_path) {
        Ok(v) => Json(v).into_response(),
        Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
    }
}

fn summary_json(db_path: &PathBuf) -> Result<Value> {
    let store = forge::tracker::store::ExecutionStore::open(db_path)?;
    let (before, after, count) = store.totals()?;
    let saved = before.saturating_sub(after);
    let pct = if before == 0 {
        0.0
    } else {
        saved as f64 * 100.0 / before as f64
    };
    let by_day: Vec<Value> = store
        .by_day(30)?
        .into_iter()
        .map(|(day, b, a)| json!({ "day": day, "tokens_before": b, "tokens_after": a }))
        .collect();
    let by_program: Vec<Value> = store
        .summary()?
        .into_iter()
        .map(|(p, b, a, avg)| json!({ "program": p, "tokens_before": b, "tokens_after": a, "avg_pct": avg }))
        .collect();
    let recent: Vec<Value> = store
        .recent(50)?
        .into_iter()
        .map(|r| {
            json!({
                "program": r.program, "args": r.args,
                "tokens_before": r.tokens_before, "tokens_after": r.tokens_after,
                "exit_code": r.exit_code, "duration_ms": r.duration_ms, "timestamp": r.timestamp,
            })
        })
        .collect();
    Ok(json!({
        "totals": { "commands": count, "tokens_before": before, "tokens_after": after, "saved": saved, "saved_pct": pct },
        "by_day": by_day,
        "by_program": by_program,
        "recent": recent,
    }))
}

/// `bctx dashboard`: starts the server in the background, prints its URL and returns, so
/// an agent that runs it is not left waiting. The server stops `idle_secs` after the last visit.
pub fn start(port: u16, idle_secs: u64, open_browser: bool) -> Result<()> {
    let exe = std::env::current_exe()?;
    let (port, idle) = (port.to_string(), idle_secs.to_string());
    let mut server = crate::detach::spawn(|| {
        let mut cmd = std::process::Command::new(&exe);
        cmd.args([
            "dashboard",
            "--serve",
            "--port",
            &port,
            "--idle-secs",
            &idle,
        ])
        .stdin(std::process::Stdio::null())
        .stdout(std::process::Stdio::piped())
        .stderr(std::process::Stdio::null());
        cmd
    })?;
    let stdout = server
        .stdout
        .take()
        .context("no pipe to the dashboard server")?;
    let (tx, rx) = std::sync::mpsc::channel();
    std::thread::spawn(move || {
        let mut line = String::new();
        let _ = BufRead::read_line(&mut std::io::BufReader::new(stdout), &mut line);
        let _ = tx.send(line);
    });
    let url = rx
        .recv_timeout(Duration::from_secs(10))
        .ok()
        .map(|l| l.trim().to_string())
        .filter(|l| l.starts_with("http://"))
        .context("the dashboard server did not start (is the port in use?)")?;
    println!("bctx dashboard: {url}");
    println!(
        "Served to this machine only, in the background; it stops {} after the page was last open.",
        human_duration(idle_secs)
    );
    if open_browser {
        let _ = open::that(&url);
    }
    Ok(())
}

fn human_duration(secs: u64) -> String {
    match secs {
        60 => "1 minute".into(),
        s if s % 60 == 0 => format!("{} minutes", s / 60),
        1 => "1 second".into(),
        s => format!("{s} seconds"),
    }
}

fn now_ms() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map_or(0, |d| d.as_millis() as u64)
}

/// Resolves once nobody has loaded the page or its data for `idle`.
async fn idle_for(last_visit: Arc<AtomicU64>, idle: Duration) {
    let check = (idle / 4).clamp(Duration::from_millis(250), Duration::from_secs(30));
    loop {
        tokio::time::sleep(check).await;
        let since = now_ms().saturating_sub(last_visit.load(Ordering::Relaxed));
        if since >= idle.as_millis() as u64 {
            return;
        }
    }
}

/// The server itself (`bctx dashboard --serve`): prints its URL as one line, then serves
/// until `idle_secs` pass without a visit.
pub fn serve(port: u16, idle_secs: u64) -> Result<()> {
    // A server outlives whoever reads its output (an agent, `| head -1`): with the CLI's
    // default SIGPIPE, the first print after that reader goes away would kill it.
    // SAFETY: signal() with SIG_IGN only sets this process's disposition for SIGPIPE.
    #[cfg(unix)]
    unsafe {
        libc::signal(libc::SIGPIPE, libc::SIG_IGN);
    }
    let db_path = PathBuf::from(crate::commands::home_dir())
        .join(".bctx")
        .join("executions.db");
    let token = format!(
        "{}{}",
        uuid::Uuid::new_v4().simple(),
        uuid::Uuid::new_v4().simple()
    );
    let rt = tokio::runtime::Builder::new_multi_thread()
        .worker_threads(2)
        .enable_all()
        .build()?;
    rt.block_on(async move {
        let listener = tokio::net::TcpListener::bind(("127.0.0.1", port)).await?;
        let port = listener.local_addr()?.port();
        let last_visit = Arc::new(AtomicU64::new(now_ms()));
        let ctx = Ctx {
            token: Arc::new(token.clone()),
            port,
            db_path: Arc::new(db_path),
            last_visit: last_visit.clone(),
        };
        let app = Router::new()
            .route("/:token/", get(page))
            .route("/:token/api/summary", get(summary))
            .with_state(ctx);
        let url = format!("http://127.0.0.1:{port}/{token}/");
        let mut out = std::io::stdout();
        let _ = writeln!(out, "{url}");
        let _ = out.flush();
        axum::serve(listener, app)
            .with_graceful_shutdown(idle_for(last_visit, Duration::from_secs(idle_secs)))
            .await?;
        Ok::<(), anyhow::Error>(())
    })
}