taktwerk 0.8.0

//! `taktwerk serve`: the workbench API over a project folder and the web UI, on loopback.
//!
//! Endpoints, all JSON, paths relative to the folder:
//!
//! - `GET /api/info`, `GET /api/files` (projects, scenarios, campaigns, models),
//!   `GET /api/file?path=`, `GET /api/check?project=` (`check --json`).
//! - `GET /api/records`, `GET /api/records/run?path=`,
//!   `GET /api/records/series?path=&columns=a,b&max_points=`, `GET /api/campaigns`.
//! - `GET /api/runs`, `POST /api/runs`, `GET /api/runs/<id>`, `POST /api/runs/<id>/stop`, and
//!   the run's own API under `/api/runs/<id>/live/`.
//!
//! Requests that start, stop or write need `Authorization: Bearer <token>`: the token is
//! `TAKTWERK_SERVE_TOKEN` when set, else generated and printed with the URL.

mod files;
mod records;
mod runs;

use std::net::SocketAddr;
use std::path::{Path, PathBuf};
use std::process::ExitCode;
use std::sync::Arc;
use std::time::{SystemTime, UNIX_EPOCH};

use anyhow::{Context as _, bail};
use axum::body::Body;
use axum::extract::{Path as UrlPath, Query, Request, State};
use axum::http::{HeaderMap, Method, StatusCode, header};
use axum::response::{IntoResponse, Response};
use axum::routing::{any, get, post};
use axum::{Json, Router};
use serde::Deserialize;
use serde_json::{Value, json};
use tracing::info;

use runs::{Runs, StartRequest};

/// The environment variable a launcher passes the token in.
pub const TOKEN_ENV: &str = "TAKTWERK_SERVE_TOKEN";

/// The largest file `GET /api/file` returns.
const MAX_FILE: u64 = 1 << 20;
/// Points a series returns unless asked otherwise.
const DEFAULT_POINTS: usize = 4000;

/// Unix milliseconds now.
pub(crate) fn unix_ms() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map_or(0, |d| u64::try_from(d.as_millis()).unwrap_or(u64::MAX))
}

struct Workspace {
    root: PathBuf,
    token: String,
    exe: PathBuf,
    runs: Runs,
}

/// Serve the folder `dir` on `listen` (loopback only) until SIGINT or SIGTERM.
pub fn serve(dir: &Path, listen: &str) -> anyhow::Result<ExitCode> {
    let root = dir
        .canonicalize()
        .with_context(|| format!("folder {}", dir.display()))?;
    if !root.is_dir() {
        bail!("{} is not a folder", root.display());
    }
    let addr: SocketAddr = listen
        .parse()
        .with_context(|| format!("--listen `{listen}`: expected an address and port"))?;
    if !addr.ip().is_loopback() {
        bail!("--listen `{listen}`: serve listens on loopback only");
    }
    let token = match std::env::var(TOKEN_ENV) {
        Ok(t) if !t.trim().is_empty() => t.trim().to_owned(),
        _ => taktwerk_http::token::generate().context("token")?,
    };
    let exe = std::env::current_exe().context("the taktwerk executable")?;
    let rt = tokio::runtime::Builder::new_multi_thread()
        .worker_threads(2)
        .thread_name("taktwerk-serve")
        .enable_all()
        .build()
        .context("cannot start the runtime")?;
    rt.block_on(async move {
        let listener = tokio::net::TcpListener::bind(addr)
            .await
            .with_context(|| format!("listen {addr}"))?;
        let bound = listener.local_addr()?;
        let ws = Arc::new(Workspace {
            runs: Runs::new(exe.clone(), root.clone()),
            root,
            token,
            exe,
        });
        println!("taktwerk workbench: http://{bound}/#token={}", ws.token);
        info!(folder = %ws.root.display(), %bound, ui = taktwerk_http::ui::available(), "serving");
        let app = router(Arc::clone(&ws));
        let mut sigterm =
            tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())?;
        axum::serve(listener, app)
            .with_graceful_shutdown(async move {
                tokio::select! {
                    _ = tokio::signal::ctrl_c() => {}
                    _ = sigterm.recv() => {}
                }
            })
            .await?;
        for run in ws.runs.all() {
            ws.runs.stop(&run).await;
        }
        Ok::<_, anyhow::Error>(())
    })?;
    Ok(ExitCode::SUCCESS)
}

/// An error answer: status and message, as `{"error": message}`.
struct ApiError(StatusCode, String);

impl ApiError {
    fn bad(msg: impl Into<String>) -> Self {
        Self(StatusCode::BAD_REQUEST, msg.into())
    }

    fn not_found(msg: impl Into<String>) -> Self {
        Self(StatusCode::NOT_FOUND, msg.into())
    }
}

impl IntoResponse for ApiError {
    fn into_response(self) -> Response {
        (self.0, Json(json!({ "error": self.1 }))).into_response()
    }
}

impl Workspace {
    fn authorize(&self, headers: &HeaderMap) -> Result<(), ApiError> {
        let given = headers
            .get(header::AUTHORIZATION)
            .and_then(|h| h.to_str().ok())
            .and_then(taktwerk_http::token::bearer);
        match given {
            Some(g) if taktwerk_http::token::matches(&self.token, g) => Ok(()),
            _ => Err(ApiError(
                StatusCode::UNAUTHORIZED,
                "this needs `Authorization: Bearer <token>`".into(),
            )),
        }
    }

    fn file(&self, rel: &str) -> Result<PathBuf, ApiError> {
        files::resolve(&self.root, rel)
            .ok_or_else(|| ApiError::not_found(format!("`{rel}` is not in the folder")))
    }
}

fn router(ws: Arc<Workspace>) -> Router {
    Router::new()
        .route("/api/info", get(info_handler))
        .route("/api/files", get(list_files))
        .route("/api/file", get(read_file))
        .route("/api/check", get(check))
        .route("/api/records", get(list_records))
        .route("/api/records/run", get(record))
        .route("/api/records/series", get(series))
        .route("/api/campaigns", get(campaigns))
        .route("/api/runs", get(list_runs).post(start_run))
        .route("/api/runs/{id}", get(run_detail))
        .route("/api/runs/{id}/stop", post(stop_run))
        .route("/api/runs/{id}/live/{*rest}", any(live))
        .fallback(taktwerk_http::ui::serve)
        .with_state(ws)
}

async fn info_handler(State(ws): State<Arc<Workspace>>) -> Json<Value> {
    let name = ws
        .root
        .file_name()
        .map(|n| n.to_string_lossy().into_owned());
    Json(json!({
        "kind": "workspace",
        "taktwerk": env!("CARGO_PKG_VERSION"),
        "root": ws.root.display().to_string(),
        "name": name,
    }))
}

async fn list_files(State(ws): State<Arc<Workspace>>) -> Json<Value> {
    let root = ws.root.clone();
    let found = tokio::task::spawn_blocking(move || files::scan(&root))
        .await
        .unwrap_or_default();
    Json(json!(found))
}

#[derive(Deserialize)]
struct PathQuery {
    path: String,
}

async fn read_file(
    State(ws): State<Arc<Workspace>>,
    Query(q): Query<PathQuery>,
) -> Result<Json<Value>, ApiError> {
    let path = ws.file(&q.path)?;
    let meta = std::fs::metadata(&path).map_err(|e| ApiError::not_found(e.to_string()))?;
    if !meta.is_file() || meta.len() > MAX_FILE {
        return Err(ApiError::bad(format!(
            "`{}` is not a file of at most 1 MiB",
            q.path
        )));
    }
    let text = std::fs::read_to_string(&path).map_err(|e| ApiError::bad(e.to_string()))?;
    Ok(Json(json!({ "path": q.path, "text": text })))
}

#[derive(Deserialize)]
struct ProjectQuery {
    project: String,
}

async fn check(
    State(ws): State<Arc<Workspace>>,
    Query(q): Query<ProjectQuery>,
) -> Result<Json<Value>, ApiError> {
    let path = ws.file(&q.project)?;
    let out = tokio::process::Command::new(&ws.exe)
        .arg("check")
        .arg(&path)
        .arg("--json")
        .current_dir(&ws.root)
        .output()
        .await
        .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
    serde_json::from_slice(&out.stdout).map(Json).map_err(|_| {
        ApiError(
            StatusCode::INTERNAL_SERVER_ERROR,
            format!("check failed: {}", String::from_utf8_lossy(&out.stderr)),
        )
    })
}

async fn list_records(State(ws): State<Arc<Workspace>>) -> Json<Value> {
    let root = ws.root.clone();
    let list = tokio::task::spawn_blocking(move || records::list(&root))
        .await
        .unwrap_or_default();
    Json(json!(list))
}

async fn record(
    State(ws): State<Arc<Workspace>>,
    Query(q): Query<PathQuery>,
) -> Result<Json<Value>, ApiError> {
    let dir = ws.file(&q.path)?;
    let run = taktwerk_record::read_run_file(&dir).map_err(|e| ApiError::bad(e.to_string()))?;
    let events = taktwerk_record::read_events(&dir).map_err(|e| ApiError::bad(e.to_string()))?;
    Ok(Json(
        json!({ "path": q.path, "run": run, "events": events }),
    ))
}

#[derive(Deserialize)]
struct SeriesQuery {
    path: String,
    columns: String,
    max_points: Option<usize>,
}

async fn series(
    State(ws): State<Arc<Workspace>>,
    Query(q): Query<SeriesQuery>,
) -> Result<Json<Value>, ApiError> {
    let dir = ws.file(&q.path)?;
    let names: Vec<String> = q
        .columns
        .split(',')
        .map(str::trim)
        .filter(|c| !c.is_empty())
        .map(str::to_owned)
        .collect();
    let max = q.max_points.unwrap_or(DEFAULT_POINTS).clamp(4, 200_000);
    let result = tokio::task::spawn_blocking(move || {
        let run = taktwerk_record::read_run_file(&dir)?;
        let cols: Vec<&str> = names.iter().map(String::as_str).collect();
        let s = taktwerk_record::read_series(&dir, &run, &cols)?;
        let rows = s.t_engine.len();
        let s = records::decimate(s, max);
        Ok::<_, taktwerk_record::ReadError>(json!({
            "rows": rows,
            "decimated": s.t_engine.len() < rows,
            "t_engine": s.t_engine,
            "cycle": s.cycle,
            "columns": names,
            "values": s.columns,
        }))
    })
    .await
    .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
    result.map(Json).map_err(|e| ApiError::bad(e.to_string()))
}

async fn campaigns(State(ws): State<Arc<Workspace>>) -> Json<Value> {
    let root = ws.root.clone();
    let list = tokio::task::spawn_blocking(move || records::campaigns(&root))
        .await
        .unwrap_or_default();
    Json(json!(list))
}

async fn list_runs(State(ws): State<Arc<Workspace>>) -> Json<Value> {
    Json(Value::Array(
        ws.runs.all().iter().map(|r| r.to_json(false)).collect(),
    ))
}

async fn start_run(
    State(ws): State<Arc<Workspace>>,
    headers: HeaderMap,
    Json(req): Json<StartRequest>,
) -> Result<(StatusCode, Json<Value>), ApiError> {
    ws.authorize(&headers)?;
    let run = ws.runs.start(&req).map_err(ApiError::bad)?;
    Ok((StatusCode::CREATED, Json(run.to_json(false))))
}

fn find(ws: &Workspace, id: u32) -> Result<Arc<runs::Launched>, ApiError> {
    ws.runs
        .get(id)
        .ok_or_else(|| ApiError::not_found(format!("no run {id}")))
}

async fn run_detail(
    State(ws): State<Arc<Workspace>>,
    UrlPath(id): UrlPath<u32>,
) -> Result<Json<Value>, ApiError> {
    Ok(Json(find(&ws, id)?.to_json(true)))
}

async fn stop_run(
    State(ws): State<Arc<Workspace>>,
    UrlPath(id): UrlPath<u32>,
    headers: HeaderMap,
) -> Result<StatusCode, ApiError> {
    ws.authorize(&headers)?;
    let run = find(&ws, id)?;
    let ws2 = Arc::clone(&ws);
    tokio::spawn(async move { ws2.runs.stop(&run).await });
    Ok(StatusCode::ACCEPTED)
}

async fn live(
    State(ws): State<Arc<Workspace>>,
    UrlPath((id, rest)): UrlPath<(u32, String)>,
    req: Request,
) -> Result<Response, ApiError> {
    let run = find(&ws, id)?;
    let method = req.method().clone();
    if method != Method::GET {
        ws.authorize(req.headers())?;
    }
    let path = match req.uri().query() {
        Some(q) => format!("{rest}?{q}"),
        None => rest,
    };
    let body = Body::new(req.into_body());
    Ok(run.proxy(method, &path, body).await)
}