taxa-server 0.1.0

//! axum web server for a taxa declarative dataset. Reproduces the HTTP contract
//! the D3 frontend speaks and serves the (embedded, unchanged) frontend.
//!
//!   GET  /api/manifest
//!   POST /api/treemap  {axis, filters}           (whole tree; frontend folds client-side)
//!   POST /api/series   {axis, metric, agg, window, resolution, filters, focus}
//!   POST /api/scatter  {x, y, selection?}
//!   GET  /api/search?q&limit
//!   GET  /api/filter-options/{facet}
//!   GET  /api/entity/{id}
//!   GET  /api/entity/{id}/series?metric&window&resolution  (per-entity time chart)
//!   GET  /api/entity/{id}/ohlc?…                            (404 — no price source)
//!   GET  /*  → embedded frontend (SPA fallback)

use std::sync::Arc;

use axum::{
    extract::{Path, Query, State},
    http::{header, HeaderMap, StatusCode, Uri},
    response::{IntoResponse, Response},
    routing::{get, post},
    Json, Router,
};
use serde_json::{Map, Value};
use taxa_core::series::SeriesArgs;
use taxa_core::treemap::TreemapArgs;
use taxa_core::{boot_manifest, frontend_tree, Backend, FrameDataset};

pub mod geo;

#[derive(rust_embed::RustEmbed)]
#[folder = "web"]
#[exclude = "**/*.test.js"]
#[exclude = "package.json"]
struct Assets;

pub struct App {
    pub ds: FrameDataset,
    pub backend: Arc<dyn Backend>,
    pub manifest: Value,
    /// Dedicated series-view dataset + backend (a separate dense timestamped
    /// table). When present, `/api/series` reads from these; otherwise it falls
    /// back to the main `ds`/`backend` (back-compat).
    pub series_ds: Option<FrameDataset>,
    pub series_backend: Option<Arc<dyn Backend>>,
    /// When true (a `branch_set: "treemap"` series view), `/api/series` resolves
    /// the branch set from the TREEMAP (snapshot ranking over the main
    /// `ds`/`backend`) and plots exactly those branches, rather than letting the
    /// series frame rank by its own `size_by`.
    pub series_follows_treemap: bool,
}

pub type Shared = Arc<App>;

pub fn build_router(
    ds: FrameDataset,
    backend: Arc<dyn Backend>,
    series: Option<(FrameDataset, Arc<dyn Backend>)>,
) -> Router {
    build_router_ext(ds, backend, series, false)
}

/// One dataset to serve in a multi-dataset (switcher) deployment.
pub struct ServeDataset {
    pub id: String,
    pub label: String,
    pub app: Shared,
}

fn make_app(
    ds: FrameDataset,
    backend: Arc<dyn Backend>,
    series: Option<(FrameDataset, Arc<dyn Backend>)>,
    series_follows_treemap: bool,
) -> Shared {
    let mut manifest = boot_manifest(&ds);
    // Geo (choropleth) view: offered when the data carries a region-key column.
    // Each map declares the TopoJSON to draw and the column to join on. Behavior
    // (zoom/click/fill) lives in the frontend; values come from POST /api/geo.
    if let Ok(cols) = backend.columns() {
        let mut maps: Vec<serde_json::Value> = Vec::new();
        if cols.contains("state_fips") {
            maps.push(serde_json::json!({"id":"states","label":"States","topojson":"us-states","key_column":"state_fips","key_kind":"fips2"}));
        }
        if cols.contains("county_fips") {
            maps.push(serde_json::json!({"id":"counties","label":"Counties","topojson":"us-counties","key_column":"county_fips","key_kind":"fips5"}));
        }
        if cols.contains("iso3") {
            maps.push(serde_json::json!({"id":"countries","label":"Countries","topojson":"world","key_column":"iso3","key_kind":"iso3"}));
        }
        if !maps.is_empty() {
            let default_map = maps
                .iter()
                .find(|m| m["id"] == "counties")
                .or_else(|| maps.first())
                .map(|m| m["id"].clone())
                .unwrap();
            // Default world maps to the crisper 50m ("med"); US-county maps stay coarse/fast.
            let default_res = if maps.iter().any(|m| m["id"] == "counties") {
                "low"
            } else {
                "med"
            };
            if let Some(views) = manifest.get_mut("views").and_then(|v| v.as_object_mut()) {
                views.insert(
                    "geo".into(),
                    serde_json::json!({
                        "maps": maps,
                        "default_map": default_map,
                        "default_colormap": "Viridis",
                        "default_scale": "log",
                        "default_projection": "naturalEarth",
                        "default_res": default_res,
                        "default_outline": true,
                    }),
                );
            }
        }
    }
    let (series_ds, series_backend) = match series {
        Some((sds, sb)) => (Some(sds), Some(sb)),
        None => (None, None),
    };
    Arc::new(App {
        ds,
        backend,
        manifest,
        series_ds,
        series_backend,
        series_follows_treemap,
    })
}

/// The `/api/*` routes for ONE dataset (no static / fallback / datasets list).
fn api_router(app: Shared) -> Router {
    Router::new()
        .route("/api/manifest", get(h_manifest))
        .route("/api/treemap", post(h_treemap))
        .route("/api/geo", post(h_geo))
        .route("/api/series", post(h_series))
        .route("/api/scatter", post(h_scatter))
        .route("/api/search", get(h_search))
        .route("/api/filter-options/:facet", get(h_filter_options))
        // `*id` (not `:id`) so path-like entity ids ("owner/repo") match — the
        // frontend deliberately keeps slashes as real separators.
        .route("/api/entity/*id", get(h_entity))
        .with_state(app)
}

pub fn build_router_ext(
    ds: FrameDataset,
    backend: Arc<dyn Backend>,
    series: Option<(FrameDataset, Arc<dyn Backend>)>,
    series_follows_treemap: bool,
) -> Router {
    let app = make_app(ds, backend, series, series_follows_treemap);
    let label = app
        .manifest
        .get("title")
        .and_then(|v| v.as_str())
        .unwrap_or("dataset")
        .to_string();
    build_router_multi(vec![ServeDataset {
        id: "main".into(),
        label,
        app,
    }])
}

/// Serve several datasets behind one server: each is mounted at `/d/<id>/api/*`,
/// the FIRST is also mounted at top-level `/api/*` (back-compat / default), plus
/// `/api/datasets` (the switcher list), `/static/*`, and the SPA fallback.
pub fn build_router_multi(entries: Vec<ServeDataset>) -> Router {
    let list = Value::Array(
        entries
            .iter()
            .map(|e| serde_json::json!({"id": e.id, "label": e.label}))
            .collect(),
    );
    let mut r = Router::new();
    for e in &entries {
        r = r.nest(&format!("/d/{}", e.id), api_router(e.app.clone()));
    }
    if let Some(first) = entries.first() {
        r = r.merge(api_router(first.app.clone()));
    }
    let r = r
        .route(
            "/api/datasets",
            get(move || {
                let l = list.clone();
                async move { Json(l) }
            }),
        )
        .route("/static/*path", get(h_asset))
        .fallback(h_index);
    // Behind a shared load balancer the app is mounted at `/<app>` (TAXA_BASE_PATH).
    // Nest the whole router there so `/<app>/api`, `/<app>/static`, `/<app>/d/<id>`,
    // and the SPA fallback all route; index.html is rewritten to match (inject_base).
    let base = base_path();
    if base.is_empty() {
        r
    } else {
        // Outer fallback too: axum's nested fallback misses the bare prefix root
        // (`/<app>/`), so route any unmatched path to the SPA shell.
        Router::new().nest(base, r).fallback(h_index)
    }
}

/// Bind and serve until shutdown.
pub async fn serve(
    ds: FrameDataset,
    backend: Arc<dyn Backend>,
    series: Option<(FrameDataset, Arc<dyn Backend>)>,
    host: &str,
    port: u16,
) -> std::io::Result<()> {
    serve_ext(ds, backend, series, false, host, port).await
}

/// Startup banner; ANSI-colored only when stdout is a terminal (so piped/redirected
/// logs stay clean).
fn serving_banner(host: &str, port: u16) -> String {
    use std::io::IsTerminal;
    let mut banner = if std::io::stdout().is_terminal() {
        // bold-green "taxa", cyan URL.
        format!("\x1b[1;32mtaxa\x1b[0m serving on \x1b[36mhttp://{host}:{port}\x1b[0m")
    } else {
        format!("taxa serving on http://{host}:{port}")
    };
    // Make dev mode unmissable — "am I testing the embed or the disk?" must
    // never be ambiguous (integration post-mortems lost hours to exactly this).
    if let Some(dir) = assets_dir() {
        banner.push_str(&format!(
            "\n  dev assets: serving frontend from {} (embedded bundle bypassed)",
            dir.display()
        ));
    }
    if let Some(dir) = geo::geo_dir_path() {
        match geo::check_registry(dir) {
            Ok(Some(summary)) => {
                banner.push_str(&format!("\n  geo: {} — {}", dir.display(), summary))
            }
            Ok(None) => banner.push_str(&format!(
                "\n  geo: {} (no registry.json — built-in maps only)",
                dir.display()
            )),
            // A broken registry is surfaced loudly but is non-fatal: the geo
            // view degrades to built-in maps rather than the server refusing
            // to boot.
            Err(e) => banner.push_str(&format!("\n  geo: {} — WARNING: {}", dir.display(), e)),
        }
    }
    banner
}

/// `serve` + the `branch_set: "treemap"` flag (see `build_router_ext`).
pub async fn serve_ext(
    ds: FrameDataset,
    backend: Arc<dyn Backend>,
    series: Option<(FrameDataset, Arc<dyn Backend>)>,
    series_follows_treemap: bool,
    host: &str,
    port: u16,
) -> std::io::Result<()> {
    let router = build_router_ext(ds, backend, series, series_follows_treemap);
    let listener = tokio::net::TcpListener::bind((host, port)).await?;
    println!("{}", serving_banner(host, port));
    axum::serve(listener, router).await
}

/// Serve several datasets behind one server (a dataset switcher). See
/// `build_router_multi` for the route layout.
pub async fn serve_multi(entries: Vec<ServeDataset>, host: &str, port: u16) -> std::io::Result<()> {
    let router = build_router_multi(entries);
    let listener = tokio::net::TcpListener::bind((host, port)).await?;
    println!("{}", serving_banner(host, port));
    axum::serve(listener, router).await
}

/// Build a `ServeDataset` (App + id/label) from raw pieces — used by hosts that
/// assemble a multi-dataset server (the CLI).
pub fn serve_dataset(
    id: impl Into<String>,
    label: impl Into<String>,
    ds: FrameDataset,
    backend: Arc<dyn Backend>,
    series: Option<(FrameDataset, Arc<dyn Backend>)>,
    series_follows_treemap: bool,
) -> ServeDataset {
    ServeDataset {
        id: id.into(),
        label: label.into(),
        app: make_app(ds, backend, series, series_follows_treemap),
    }
}

type ApiResult = Result<Json<Value>, (StatusCode, String)>;

fn bad(e: impl std::fmt::Display) -> (StatusCode, String) {
    (StatusCode::BAD_REQUEST, e.to_string())
}

const SERIES_AGGS: &[&str] = &["sum", "mean", "median", "min", "max", "count"];
const RESOLUTIONS: &[&str] = &["d", "w", "m", "q", "y"];
/// Series branch cap — kept in lockstep with the treemap's `MAX_TOP_K`
/// (`treemap.rs`) so the Line tab can't request an unbounded branch set.
const MAX_TOP_K: i64 = 100;

/// Resolve the resolution for a series request against the SERIES FRAME's
/// declared resolutions (a producer may restrict to e.g. `["w","m","y"]`), NOT
/// the global allowlist — so a direct request can't ask `d` against a weekly-
/// pre-bucketed frame (which would re-bucket a stock metric to daily). A present
/// `requested` must be in the frame's allowed set (error otherwise); an absent
/// one defaults to the frame's FIRST allowed resolution (never a hardcoded "d").
fn resolve_resolution(
    sds: &FrameDataset,
    requested: Option<&str>,
) -> Result<String, (StatusCode, String)> {
    let allowed: Vec<&str> = match sds.series_resolutions.as_deref() {
        Some(rs) if !rs.is_empty() => rs.iter().map(String::as_str).collect(),
        _ => RESOLUTIONS.to_vec(),
    };
    match requested {
        Some(r) if !r.is_empty() => {
            if !allowed.contains(&r) {
                return Err(bad(format!("unknown resolution {r:?}")));
            }
            Ok(r.to_string())
        }
        _ => Ok(allowed.first().copied().unwrap_or("d").to_string()),
    }
}

/// Reject filter keys that don't correspond to a declared facet (a `_min`/`_max`
/// suffix targets a range facet) — rather than silently ignoring them.
fn check_filters(
    ds: &FrameDataset,
    filters: &Map<String, Value>,
) -> Result<(), (StatusCode, String)> {
    for key in filters.keys() {
        let base = key
            .strip_suffix("_min")
            .or_else(|| key.strip_suffix("_max"))
            .unwrap_or(key);
        if !ds.filters.iter().any(|f| f.id == base) {
            return Err(bad(format!("unknown filter {key:?}")));
        }
    }
    Ok(())
}

/// Run a blocking engine call off the async runtime. A panic inside the engine
/// task surfaces as a 500 (`JoinError`) rather than taking down the worker.
async fn blocking<T, F>(f: F) -> Result<T, (StatusCode, String)>
where
    F: FnOnce() -> Result<T, (StatusCode, String)> + Send + 'static,
    T: Send + 'static,
{
    match tokio::task::spawn_blocking(f).await {
        Ok(r) => r,
        Err(_) => Err((
            StatusCode::INTERNAL_SERVER_ERROR,
            "engine task panicked".into(),
        )),
    }
}

async fn h_manifest(State(app): State<Shared>) -> Response {
    (
        [(header::CACHE_CONTROL, "public, max-age=60")],
        Json(app.manifest.clone()),
    )
        .into_response()
}

fn obj(body: &Value, key: &str) -> Map<String, Value> {
    body.get(key)
        .and_then(|v| v.as_object())
        .cloned()
        .unwrap_or_default()
}

async fn h_treemap(State(app): State<Shared>, Json(body): Json<Value>) -> ApiResult {
    blocking(move || {
        let axis = body
            .get("axis")
            .and_then(|v| v.as_str())
            .map(str::to_string)
            .or_else(|| app.ds.default_axis.clone())
            .ok_or_else(|| bad("no axis"))?;
        let levels = app
            .ds
            .axis(&axis)
            .map(|a| a.levels.len())
            .ok_or_else(|| bad("unknown axis"))?;
        let mut a = TreemapArgs::new(axis.clone());
        a.filters = obj(&body, "filters");
        check_filters(&app.ds, &a.filters)?;
        // size_by drives the top-K + "Other" ranking; validate it's a metric.
        if let Some(sb) = body
            .get("size_by")
            .and_then(|v| v.as_str())
            .filter(|s| !s.is_empty())
        {
            if app.ds.metric(sb).is_none() {
                return Err(bad(format!("unknown size_by {sb:?}")));
            }
            a.size_by = Some(sb.to_string());
        }
        // Focus-bounded contract: the client sends the current focus path (zoom),
        // how many levels below it to materialize (depth), and the per-parent
        // branch cap (top_k). The engine clamps both and bounds the node count to
        // ≤ Σ(top_k+1)^level. focus arrives as ["root", …]; drop the leading root.
        a.focus = body
            .get("focus")
            .and_then(|v| v.as_array())
            .map(|arr| arr.iter().skip(1).cloned().collect())
            .unwrap_or_default();
        a.depth = body
            .get("depth")
            .and_then(|v| v.as_i64())
            .unwrap_or(levels as i64);
        a.top_k = body.get("top_k").and_then(|v| v.as_i64()).unwrap_or(50);
        let tree = app.backend.treemap(&app.ds, &a).map_err(bad)?;
        frontend_tree(&app.ds, &axis, &tree).ok_or_else(|| bad("unknown axis"))
    })
    .await
    .map(Json)
}

fn window_days(w: &str) -> Option<i64> {
    Some(match w {
        "1w" => 7,
        "1m" => 31,
        "ytd" | "1y" => 365,
        "3y" => 1095,
        "5y" => 1825,
        "max" | "all" => 36500,
        _ => return None,
    })
}

async fn h_series(State(app): State<Shared>, Json(body): Json<Value>) -> ApiResult {
    blocking(move || {
        // Route to the dedicated series backend/dataset when configured; else the
        // main ds/backend (back-compat). All schema checks bind against `sds`.
        let sds = app.series_ds.as_ref().unwrap_or(&app.ds);
        let sbackend = app.series_backend.as_ref().unwrap_or(&app.backend);
        let axis = body
            .get("axis")
            .and_then(|v| v.as_str())
            .ok_or_else(|| bad("no axis"))?;
        let metric = body
            .get("metric")
            .and_then(|v| v.as_str())
            .ok_or_else(|| bad("no metric"))?;
        if sds.axis(axis).is_none() {
            return Err(bad(format!("unknown axis {axis:?}")));
        }
        if sds.metric(metric).is_none() {
            return Err(bad(format!("unknown metric {metric:?}")));
        }
        let mut a = SeriesArgs::new(axis, metric);
        // Validate rather than silently coercing unknown values to a default.
        a.agg = match body.get("agg").and_then(|v| v.as_str()) {
            Some(g) if !g.is_empty() => {
                if !SERIES_AGGS.contains(&g) {
                    return Err(bad(format!("unknown agg {g:?}")));
                }
                Some(g.to_string())
            }
            _ => None,
        };
        // Validate the requested resolution against the SERIES FRAME's declared
        // resolutions (not the global allowlist); absent → the frame's first
        // allowed resolution (never the hardcoded "d" `SeriesArgs::new` seeds).
        a.resolution = resolve_resolution(sds, body.get("resolution").and_then(|v| v.as_str()))?;
        a.filters = obj(&body, "filters");
        check_filters(sds, &a.filters)?;
        a.window_days = match body.get("window").and_then(|v| v.as_str()) {
            Some(w) if !w.is_empty() => {
                Some(window_days(w).ok_or_else(|| bad(format!("unknown window {w:?}")))?)
            }
            _ => None,
        };
        // size_by ranks the branch set (top-K + "Other"), exactly as the treemap
        // does — so the Line tab matches the treemap. The Line tab always forwards
        // the CURRENT treemap size metric, which may not be one of the (reduced)
        // series metrics. Rather than 400, fall back to ranking by the plotted
        // `metric` (always a valid series metric) when `size_by` isn't a series
        // metric. Never 400 on a non-series `size_by`.
        if let Some(sb) = body
            .get("size_by")
            .and_then(|v| v.as_str())
            .filter(|s| !s.is_empty())
        {
            a.size_by = Some(if sds.metric(sb).is_some() {
                sb.to_string()
            } else {
                metric.to_string()
            });
        }
        // Clamp top_k to the same bound the treemap enforces (MAX_TOP_K) so an
        // out-of-range request can't produce an unbounded branch set.
        if let Some(k) = body
            .get("top_k")
            .and_then(|v| v.as_i64())
            .filter(|k| *k > 0)
        {
            a.top_k = k.min(MAX_TOP_K) as usize;
        }
        // focus arrives as ["root", ...]; the engine wants the keys below root.
        a.focus = body
            .get("focus")
            .and_then(|v| v.as_array())
            .map(|arr| arr.iter().skip(1).cloned().collect())
            .unwrap_or_default();

        // `branch_set: "treemap"` — resolve the branch set from the TREEMAP
        // (snapshot ranking over the MAIN ds/backend) one level below focus, at
        // the same axis/focus/size_by/top_k, then plot EXACTLY those branches.
        // This fixes rank-by-history and a size_by the series frame lacks: the
        // ranking happens on the snapshot, the plotting on the series frame.
        if app.series_follows_treemap && app.ds.axis(axis).is_some() {
            let mut ta = TreemapArgs::new(axis);
            ta.filters = a.filters.clone();
            ta.focus = a.focus.clone();
            ta.top_k = a.top_k as i64;
            // The snapshot's size_by: prefer the request's, validated against the
            // MAIN (snapshot) ds; else the snapshot default.
            if let Some(sb) = body
                .get("size_by")
                .and_then(|v| v.as_str())
                .filter(|s| !s.is_empty())
            {
                if app.ds.metric(sb).is_some() {
                    ta.size_by = Some(sb.to_string());
                }
            }
            let bs = app.backend.branch_set(&app.ds, &ta).map_err(bad)?;
            a.branches = Some(bs.keep);
            a.include_other = bs.has_other;
        }
        sbackend.series(sds, &a).map_err(bad)
    })
    .await
    .map(Json)
}

async fn h_geo(State(app): State<Shared>, Json(body): Json<Value>) -> ApiResult {
    blocking(move || {
        let key = body
            .get("key")
            .and_then(|v| v.as_str())
            .ok_or_else(|| bad("no key column"))?
            .to_string();
        let metric = body
            .get("metric")
            .and_then(|v| v.as_str())
            .map(str::to_string)
            .or_else(|| app.ds.default_size_by.clone())
            .ok_or_else(|| bad("no metric"))?;
        if app.ds.metric(&metric).is_none() {
            return Err(bad(format!("unknown metric {metric:?}")));
        }
        let filters = obj(&body, "filters");
        check_filters(&app.ds, &filters)?;
        app.backend
            .geo(&app.ds, &key, &metric, &filters)
            .map_err(bad)
    })
    .await
    .map(Json)
}

async fn h_scatter(State(app): State<Shared>, Json(body): Json<Value>) -> ApiResult {
    blocking(move || {
        let tv = app.manifest["views"]["scatter"].clone();
        let x = body
            .get("x")
            .and_then(|v| v.as_str())
            .or_else(|| tv["default_x"].as_str())
            .ok_or_else(|| bad("no x"))?
            .to_string();
        let y = body
            .get("y")
            .and_then(|v| v.as_str())
            .or_else(|| tv["default_y"].as_str())
            .ok_or_else(|| bad("no y"))?
            .to_string();
        app.backend
            .scatter(&app.ds, &x, &y, &Map::new(), None, None)
            .map_err(bad)
    })
    .await
    .map(Json)
}

async fn h_search(
    State(app): State<Shared>,
    Query(q): Query<std::collections::HashMap<String, String>>,
) -> ApiResult {
    let needle = q.get("q").cloned().unwrap_or_default();
    let limit = q.get("limit").and_then(|s| s.parse().ok()).unwrap_or(20u32);
    let axis = q.get("axis").cloned();
    blocking(move || {
        app.backend
            .search(&app.ds, &needle, axis.as_deref(), limit)
            .map_err(bad)
    })
    .await
    .map(Json)
}

async fn h_filter_options(
    State(app): State<Shared>,
    Path(facet): Path<String>,
    Query(q): Query<std::collections::HashMap<String, String>>,
) -> ApiResult {
    let query = q.get("q").cloned();
    blocking(move || {
        app.backend
            .filter_options(&app.ds, &facet, query.as_deref(), 200)
            .map(Value::Array)
            .map_err(bad)
    })
    .await
    .map(Json)
}

/// `/api/entity/*id` is a single glob, so the entity-detail sub-resources
/// (`…/series`, `…/ohlc`) arrive here with the sub-path glued onto `id`. We
/// peel a trailing `/series` or `/ohlc` segment and dispatch in-handler rather
/// than registering overlapping globs (which axum can't disambiguate). A bare
/// id → the detail record; `…/series` → the per-entity time chart; `…/ohlc` →
/// 404 (no price source yet — the frontend falls back to the line cleanly).
async fn h_entity(
    State(app): State<Shared>,
    Path(id): Path<String>,
    Query(q): Query<std::collections::HashMap<String, String>>,
) -> ApiResult {
    // Only treat a trailing `/series`/`/ohlc` as a subroute when a non-empty id
    // precedes it; otherwise an entity whose id literally ends in those segments
    // (or is exactly "series"/"ohlc") would be unreachable. Such an id falls
    // through to the plain detail lookup below.
    if let Some(eid) = id.strip_suffix("/ohlc").filter(|e| !e.is_empty()) {
        // No price/OHLC source — 404 so entity.js degrades to the metric line.
        return Err((StatusCode::NOT_FOUND, format!("no ohlc for {eid:?}")));
    }
    if let Some(eid) = id.strip_suffix("/series").filter(|e| !e.is_empty()) {
        let eid = eid.to_string();
        return blocking(move || {
            // Read from the dedicated series ds/backend when present (the dense
            // timestamped table); else the main ds/backend (back-compat).
            let sds = app.series_ds.as_ref().unwrap_or(&app.ds);
            let sbackend = app.series_backend.as_ref().unwrap_or(&app.backend);
            let metric = q
                .get("metric")
                .map(String::as_str)
                .ok_or_else(|| bad("no metric"))?;
            if sds.metric(metric).is_none() {
                return Err(bad(format!("unknown metric {metric:?}")));
            }
            // Validate the resolution against the SERIES FRAME's declared
            // resolutions (a producer may restrict to e.g. ["w","m","y"]), not a
            // global list — a global "d" passing here would re-bucket a
            // weekly-pre-bucketed stock metric to daily. Absent → the frame's first
            // allowed resolution, never a hardcoded "d". (Shared with /api/series.)
            let resolution = resolve_resolution(sds, q.get("resolution").map(String::as_str))?;
            let window = match q.get("window").map(String::as_str) {
                Some(w) if !w.is_empty() => {
                    Some(window_days(w).ok_or_else(|| bad(format!("unknown window {w:?}")))?)
                }
                _ => None,
            };
            sbackend
                .entity_series(sds, &eid, metric, window, &resolution)
                .map_err(bad)
        })
        .await
        .map(Json);
    }
    blocking(
        move || match app.backend.detail(&app.ds, &id).map_err(bad)? {
            Some(d) => Ok(d),
            None => Err((StatusCode::NOT_FOUND, format!("no entity {id:?}"))),
        },
    )
    .await
    .map(Json)
}

/// Dev assets dir, resolved at most once: an explicit `set_assets_dir` (the
/// CLI's `--assets-dir`) wins; otherwise the `TAXA_ASSETS_DIR` env var (for
/// daemonized servers, e.g. launchd, where flags are awkward to edit).
static ASSETS_DIR: std::sync::OnceLock<Option<std::path::PathBuf>> = std::sync::OnceLock::new();

fn assets_dir() -> Option<&'static std::path::Path> {
    ASSETS_DIR
        .get_or_init(|| std::env::var_os("TAXA_ASSETS_DIR").map(Into::into))
        .as_deref()
}

/// URL base path the app is mounted under (e.g. `/vc` behind a shared load
/// balancer), from `$TAXA_BASE_PATH`. Normalized to a leading slash with no
/// trailing slash; empty string = served at root (default). The whole router is
/// nested under it and index.html is rewritten so the frontend resolves
/// assets/APIs under the prefix (see `inject_base`).
fn base_path() -> &'static str {
    static BASE: std::sync::OnceLock<String> = std::sync::OnceLock::new();
    BASE.get_or_init(|| {
        let t = std::env::var("TAXA_BASE_PATH")
            .unwrap_or_default()
            .trim()
            .trim_matches('/')
            .to_string();
        if t.is_empty() {
            String::new()
        } else {
            format!("/{t}")
        }
    })
}

/// Rewrite index.html so the frontend works under `base` (`""` = root). Injects,
/// right after `<head>`: a `<base href>` (relative asset URLs resolve under the
/// prefix), an import map remapping absolute `/static/` module specifiers, and a
/// `window.__BASE__` global (api.js/app.js prefix fetch + routing with it).
fn inject_base(html: &[u8], base: &str) -> Vec<u8> {
    let s = String::from_utf8_lossy(html);
    let Some(i) = s.find("<head>") else {
        return html.to_vec();
    };
    let at = i + "<head>".len();
    let js = serde_json::to_string(base).unwrap_or_else(|_| "\"\"".into());
    let snippet = format!(
        "<base href=\"{base}/\">\
         <script>window.__BASE__={js};</script>\
         <script type=\"importmap\">{{\"imports\":{{\"/static/\":\"{base}/static/\"}}}}</script>"
    );
    let mut out = String::with_capacity(s.len() + snippet.len());
    out.push_str(&s[..at]);
    out.push_str(&snippet);
    out.push_str(&s[at..]);
    out.into_bytes()
}

/// Serve frontend assets from `dir` instead of the embedded bundle (dev mode:
/// frontend edits show up on refresh without rebuilding the binary). Call
/// before serving; a later call loses to whatever resolved first.
pub fn set_assets_dir(dir: impl Into<std::path::PathBuf>) {
    let _ = ASSETS_DIR.set(Some(dir.into()));
}

/// Dev mode: read `<assets_dir>/<rel>` from disk. Returns None in production
/// (no dir configured) or for a missing/traversing path → caller falls back to
/// the embedded bundle.
fn dev_asset(rel: &str) -> Option<Vec<u8>> {
    let dir = assets_dir()?;
    if rel.split('/').any(|s| s == ".." || s.is_empty()) {
        return None; // no path traversal
    }
    std::fs::read(dir.join(rel)).ok()
}

fn hex(bytes: &[u8]) -> String {
    bytes.iter().map(|b| format!("{b:02x}")).collect()
}

/// Does an `If-None-Match` header value match this ETag? Weak comparison is
/// fine for a 304: the `W/` prefix is ignored on BOTH sides (so a weak stored
/// validator like the geo module's `W/"<len>-<mtime>"` matches), `*` matches
/// anything. (Shared with the `geo` module's disk-asset serving.)
pub(crate) fn etag_matches(if_none_match: &str, etag: &str) -> bool {
    let want = etag.trim_start_matches("W/");
    if_none_match
        .split(',')
        .map(|t| t.trim().trim_start_matches("W/"))
        .any(|t| t == want || t == "*")
}

/// Serve an embedded asset with a strong ETag (rust_embed's sha256) under
/// `no-cache`: the browser revalidates every time, but a 304 skips the body.
/// So a regenerated vendor file (TopoJSON, geo registry) is picked up on the
/// next refresh, while an unchanged multi-MB map costs only a header
/// round-trip. (Replaces the old fixed `max-age=86400` on `vendor/`, which
/// served stale registries for up to a day.)
fn embedded_response(
    content: rust_embed::EmbeddedFile,
    mime: &str,
    req_headers: &HeaderMap,
) -> Response {
    let etag = format!("\"{}\"", hex(content.metadata.sha256_hash().as_ref()));
    let inm = req_headers
        .get(header::IF_NONE_MATCH)
        .and_then(|v| v.to_str().ok());
    if inm.is_some_and(|v| etag_matches(v, &etag)) {
        return (StatusCode::NOT_MODIFIED, [(header::ETAG, etag)]).into_response();
    }
    (
        [
            (header::CONTENT_TYPE, mime),
            (header::CACHE_CONTROL, "no-cache"),
            (header::ETAG, &etag),
        ],
        content.data,
    )
        .into_response()
}

/// Serve a frontend asset (the frontend references everything under `/static/…`).
/// Dev mode (`--assets-dir` / TAXA_ASSETS_DIR) serves from disk with `no-store`;
/// otherwise the embedded bundle with ETag revalidation.
async fn h_asset(Path(path): Path<String>, req_headers: HeaderMap) -> Response {
    let mime = mime_guess::from_path(&path).first_or_octet_stream();
    if let Some(bytes) = dev_asset(&path) {
        return (
            [
                (header::CONTENT_TYPE, mime.as_ref()),
                (header::CACHE_CONTROL, "no-store"),
            ],
            bytes,
        )
            .into_response();
    }
    // Consumer-supplied geo data (TopoJSON / metric files / registry) under
    // `--geo-dir` — taxa embeds no geo DATA of its own (only the generic
    // us/world atlases that power built-in maps live in the bundle).
    if let Some(resp) = geo::serve_geo_asset(&path, &req_headers) {
        return resp;
    }
    match Assets::get(&path) {
        Some(content) => embedded_response(content, mime.as_ref(), &req_headers),
        None => (StatusCode::NOT_FOUND, "not found").into_response(),
    }
}

#[cfg(test)]
mod etag_tests {
    use super::etag_matches;

    #[test]
    fn strong_and_weak_validators_match() {
        // Strong (embedded sha256) — exact match.
        assert!(etag_matches("\"abc123\"", "\"abc123\""));
        // Weak (geo disk validator) — `W/` ignored on BOTH sides (the bug the
        // boot smoke-test caught: incoming W/ but stored W/ failed to match).
        assert!(etag_matches(
            "W/\"2415-1781732769\"",
            "W/\"2415-1781732769\""
        ));
        // Mixed strength still matches the underlying opaque-tag.
        assert!(etag_matches("\"2415-1\"", "W/\"2415-1\""));
        // Wildcard matches anything; a different tag does not.
        assert!(etag_matches("*", "W/\"whatever\""));
        assert!(!etag_matches("\"other\"", "\"abc123\""));
        // Multi-valued If-None-Match: any member can match.
        assert!(etag_matches(
            "\"x\", W/\"2415-1781732769\"",
            "W/\"2415-1781732769\""
        ));
    }
}

/// Every non-API, non-asset path returns index.html (client-side routing:
/// `/`, `/treemap`, `/scatter`, `/entity/{id}`, …).
async fn h_index(_uri: Uri, req_headers: HeaderMap) -> Response {
    // index.html is rewritten for the base path (inject_base), so its ETag must be
    // computed over the INJECTED bytes (not the embed's hash). Strong, `no-cache` →
    // the browser revalidates and a 304 skips the body.
    let raw =
        dev_asset("index.html").or_else(|| Assets::get("index.html").map(|c| c.data.into_owned()));
    let Some(bytes) = raw else {
        return (StatusCode::NOT_FOUND, "index.html missing from embed").into_response();
    };
    let html = inject_base(&bytes, base_path());
    use std::hash::{Hash, Hasher};
    let mut h = std::collections::hash_map::DefaultHasher::new();
    html.hash(&mut h);
    let etag = format!("\"{:016x}\"", h.finish());
    if let Some(inm) = req_headers
        .get(header::IF_NONE_MATCH)
        .and_then(|v| v.to_str().ok())
    {
        if etag_matches(inm, &etag) {
            return (StatusCode::NOT_MODIFIED, [(header::ETAG, etag)]).into_response();
        }
    }
    (
        [
            (header::CONTENT_TYPE, "text/html"),
            (header::CACHE_CONTROL, "no-cache"),
            (header::ETAG, &etag),
        ],
        html,
    )
        .into_response()
}