hprof-analyzer 0.2.0

//! Unified HTTP server: OQL queries + report sections (`server` subcommand).
//! Extends the OQL server with lazy full-analysis and per-section endpoints.
//! Loopback only, sync tiny_http.

pub const DEFAULT_PORT: u16 = 7070;

use std::io::{self, Read as _};
use std::sync::{Arc, Mutex};

use tiny_http::{Response, Server};

use crate::AnalyzeOptions;
use crate::query::server::ServerState as OqlState;
use crate::report::Report;

/// The full analysis pipeline state — transitions once through the cycle.
pub enum AnalysisState {
    NotStarted,
    Running,
    Done(Arc<Report>),
    Failed(String),
}

pub struct ServeState {
    path: String,
    opts: AnalyzeOptions,
    oql: Arc<OqlState>,
    pub state: Arc<Mutex<AnalysisState>>,
    /// Per-object retained-size array from the full analysis pipeline.
    /// Populated once `AnalysisState::Done` is reached; shared with the OQL
    /// engine so `@retainedHeapSize` queries skip the full re-scan.
    retained: Arc<Mutex<Option<Arc<Vec<u64>>>>>,
}

impl ServeState {
    pub fn new(path: &str, opts: AnalyzeOptions) -> io::Result<Self> {
        let oql = Arc::new(OqlState::load(path, opts.query_path_depth, true)?);
        Ok(ServeState {
            path: path.to_string(),
            opts,
            oql,
            state: Arc::new(Mutex::new(AnalysisState::NotStarted)),
            retained: Arc::new(Mutex::new(None)),
        })
    }

    /// If analysis is not yet started, launch it in a background thread.
    /// Returns true if a new thread was spawned, false if already running/done/failed.
    fn ensure_analysis(&self) -> bool {
        let mut guard = self.state.lock().unwrap_or_else(|e| e.into_inner());
        if matches!(*guard, AnalysisState::NotStarted) {
            *guard = AnalysisState::Running;
            drop(guard); // release lock before spawning

            let state_arc = Arc::clone(&self.state);
            let oql_arc = Arc::clone(&self.oql);
            let retained_arc = Arc::clone(&self.retained);
            let path = self.path.clone();
            let opts = self.opts.clone();
            std::thread::spawn(move || {
                let result = crate::analyze_to_report_with_retained(
                    &crate::source::HprofSource::from(path.as_str()),
                    &opts,
                );
                let mut g = state_arc.lock().unwrap_or_else(|e| e.into_inner());
                match result {
                    Ok((r, retained_vec)) => {
                        *g = AnalysisState::Done(Arc::new(r));
                        drop(g);
                        // Store the retained-size array so OQL queries can use
                        // @retainedHeapSize without re-scanning the dump.
                        let retained_shared = Arc::new(retained_vec);
                        {
                            let mut rg = retained_arc.lock().unwrap_or_else(|e| e.into_inner());
                            *rg = Some(Arc::clone(&retained_shared));
                        }
                        oql_arc.set_full_analysis_with_retained(retained_shared);
                    }
                    Err(e) => {
                        *g = AnalysisState::Failed(e.to_string());
                        drop(g);
                        // Still unlock OQL (analysis failed but partial pipeline ran).
                        oql_arc.set_full_analysis_with_retained(Arc::new(Vec::new()));
                    }
                }
            });
            true
        } else {
            false
        }
    }

    /// Read the cached report, if available. Returns None while running/not started.
    fn report(&self) -> Option<Arc<Report>> {
        let guard = self.state.lock().unwrap_or_else(|e| e.into_inner());
        if let AnalysisState::Done(ref r) = *guard {
            Some(Arc::clone(r))
        } else {
            None
        }
    }

    /// Returns true when analysis is Done.
    #[cfg(test)]
    pub fn report_ready(&self) -> bool {
        let guard = self.state.lock().unwrap_or_else(|e| e.into_inner());
        matches!(*guard, AnalysisState::Done(_))
    }

    fn status_json(&self) -> serde_json::Value {
        let guard = self.state.lock().unwrap_or_else(|e| e.into_inner());
        match *guard {
            AnalysisState::NotStarted => serde_json::json!({"status": "not_started"}),
            AnalysisState::Running => serde_json::json!({"status": "analyzing"}),
            AnalysisState::Done(_) => serde_json::json!({"status": "ready"}),
            AnalysisState::Failed(ref e) => serde_json::json!({"status": "failed", "error": e}),
        }
    }

    /// Build the offline HTML download response for `GET /download/offline`.
    ///
    /// Returns `(html_bytes, filename_stem)` when the report is ready, or
    /// an error string + 503 status when the analysis hasn't finished yet.
    fn offline_html(&self) -> Result<(Vec<u8>, String), (u16, String)> {
        match self.report() {
            Some(report) => {
                let html = crate::html::render_html(&report);
                let stem = std::path::Path::new(&self.path)
                    .file_stem()
                    .and_then(|s| s.to_str())
                    .unwrap_or("report")
                    .to_string();
                Ok((html.into_bytes(), stem))
            }
            None => {
                let guard = self.state.lock().unwrap_or_else(|e| e.into_inner());
                let msg = match *guard {
                    AnalysisState::Failed(ref e) => format!("analysis failed: {e}"),
                    _ => "analysis not complete yet — retry in a few seconds".to_string(),
                };
                Err((503, msg))
            }
        }
    }

    /// Route (method, url, body) → (http_status, body_string, content_type).
    pub fn route(&self, method: &str, url: &str, body: &str) -> (u16, String, &'static str) {
        let (path, query) = split_path_query(url);
        let want_md = query.contains("format=md");

        match (method, path) {
            // ── Status / trigger ──────────────────────────────────────────────
            ("GET", "/status") => {
                (200, self.status_json().to_string(), "application/json")
            }
            ("POST", "/analyze") => {
                let spawned = self.ensure_analysis();
                let status_str = if spawned {
                    "started"
                } else {
                    let g = self.state.lock().unwrap_or_else(|e| e.into_inner());
                    match *g {
                        AnalysisState::Running => "already_running",
                        AnalysisState::Done(_) => "already_done",
                        AnalysisState::Failed(_) => "failed",
                        AnalysisState::NotStarted => "started",
                    }
                };
                (200, serde_json::json!({"ok": true, "status": status_str}).to_string(), "application/json")
            }

            // ── Report sections ───────────────────────────────────────────────
            ("GET", p) if p == "/report" || p.starts_with("/report/") => {
                self.ensure_analysis();

                match self.report() {
                    None => {
                        let guard = self.state.lock().unwrap_or_else(|e| e.into_inner());
                        if let AnalysisState::Failed(ref e) = *guard {
                            let b = serde_json::json!({"ok":false,"error":{"kind":"analysis","message":e}}).to_string();
                            return (500, b, "application/json");
                        }
                        drop(guard);
                        let b = serde_json::json!({
                            "ok": false,
                            "status": "analyzing",
                            "message": "analysis running, retry in a few seconds"
                        }).to_string();
                        (202, b, "application/json")
                    }
                    Some(report) => match p {
                        "/report" | "/report/" => {
                            if want_md {
                                let md = crate::report::render_markdown(&report);
                                (200, md, "text/markdown; charset=utf-8")
                            } else {
                                match serde_json::to_string_pretty(&*report) {
                                    Ok(j) => (200, j, "application/json"),
                                    Err(e) => (500, serde_json::json!({"ok":false,"error":{"kind":"serialization","message":e.to_string()}}).to_string(), "application/json"),
                                }
                            }
                        }
                        "/report/overview" => {
                            let off_heap_cap = report.leak_indicators.direct_byte_buffer_capacity_sum;
                            section_response(&report.overview, want_md, move |r, out| {
                                crate::report::render_system_overview(r, off_heap_cap, out);
                            })
                        }
                        "/report/leaks" => {
                            let limit = parse_limit(query);
                            if let Some(n) = limit.filter(|&n| n < report.leaks.suspects.len()) {
                                let mut trimmed = report.leaks.clone();
                                trimmed.suspects.truncate(n);
                                section_response(&trimmed, want_md, |r, out| {
                                    crate::report::render_leak_suspects(r, out);
                                })
                            } else {
                                section_response(&report.leaks, want_md, |r, out| {
                                    crate::report::render_leak_suspects(r, out);
                                })
                            }
                        }
                        "/report/top" => {
                            let limit = parse_limit(query);
                            if let Some(n) = limit {
                                let mut trimmed = report.top.clone();
                                if n < trimmed.biggest_objects.len() {
                                    trimmed.biggest_objects.truncate(n);
                                }
                                if n < trimmed.biggest_classes.len() {
                                    trimmed.biggest_classes.truncate(n);
                                }
                                let total_shallow = report.leaks.total_shallow;
                                section_response(&trimmed, want_md, |t, out| {
                                    crate::report::render_top_consumers(t, total_shallow, out);
                                })
                            } else {
                                section_response(&report.top, want_md, |t, out| {
                                    crate::report::render_top_consumers(t, report.leaks.total_shallow, out);
                                })
                            }
                        }
                        "/report/threads" => section_response(&report.threads, want_md, |r, out| {
                            crate::report::render_threads(r, false, out);
                        }),
                        _ => not_found(path),
                    },
                }
            }

            // ── OQL ───────────────────────────────────────────────────────────
            ("POST", "/") | ("POST", "/query") | ("POST", "/stream")
            | ("GET", "/help") | ("GET", "/schema") => {
                self.oql.route_guarded(method, url, body)
            }

            // ── Meta ──────────────────────────────────────────────────────────
            ("GET", "/") | ("GET", "/version") => {
                (200, version_json().to_string(), "application/json")
            }

            // ── 404 / 405 ─────────────────────────────────────────────────────
            (_, p) if KNOWN_PATHS.contains(&p) => (
                405,
                serde_json::json!({
                    "ok": false,
                    "error": {"kind": "method", "message": format!("method {method} not allowed on {p}")}
                })
                .to_string(),
                "application/json",
            ),
            _ => not_found(path),
        }
    }
}

const KNOWN_PATHS: &[&str] = &[
    "/",
    "/version",
    "/status",
    "/analyze",
    "/report",
    "/report/",
    "/report/overview",
    "/report/leaks",
    "/report/top",
    "/report/threads",
    "/help",
    "/schema",
    "/download/offline",
];

fn not_found(path: &str) -> (u16, String, &'static str) {
    (
        404,
        serde_json::json!({"ok":false,"error":{"kind":"route","message":format!("no route {path}")}})
            .to_string(),
        "application/json",
    )
}

fn section_response<T: serde::Serialize>(
    section: &T,
    want_md: bool,
    render_fn: impl FnOnce(&T, &mut String),
) -> (u16, String, &'static str) {
    if want_md {
        let mut out = String::new();
        render_fn(section, &mut out);
        (200, out, "text/markdown; charset=utf-8")
    } else {
        match serde_json::to_string_pretty(section) {
            Ok(j) => (200, j, "application/json"),
            Err(e) => (500, serde_json::json!({"ok":false,"error":{"kind":"serialization","message":e.to_string()}}).to_string(), "application/json"),
        }
    }
}

fn split_path_query(url: &str) -> (&str, &str) {
    match url.split_once('?') {
        Some((p, q)) => (p, q),
        None => (url, ""),
    }
}

fn parse_limit(query: &str) -> Option<usize> {
    query.split('&').find_map(|part| {
        let (k, v) = part.split_once('=')?;
        if k == "limit" { v.parse().ok() } else { None }
    })
}

pub fn version_json() -> serde_json::Value {
    serde_json::json!({
        "name": "hprof-analyzer server",
        "version": env!("CARGO_PKG_VERSION"),
        "endpoints": [
            {"method":"GET",  "path":"/status",          "desc":"analysis status: not_started|analyzing|ready|failed"},
            {"method":"POST", "path":"/analyze",         "desc":"trigger full analysis"},
            {"method":"GET",  "path":"/report",          "desc":"full Report JSON (add ?format=md for Markdown)"},
            {"method":"GET",  "path":"/report/overview", "desc":"SystemOverview JSON or ?format=md"},
            {"method":"GET",  "path":"/report/leaks",    "desc":"LeakSuspects JSON or ?format=md"},
            {"method":"GET",  "path":"/report/top",      "desc":"TopConsumers JSON or ?format=md"},
            {"method":"GET",  "path":"/report/threads",  "desc":"ThreadOverview JSON or ?format=md"},
            {"method":"POST", "path":"/",                "desc":"run OQL (raw body or {\"query\":\"...\"}), JSON QueryResult back"},
            {"method":"POST", "path":"/query",           "desc":"alias of /"},
            {"method":"POST", "path":"/stream",          "desc":"run OQL, NDJSON rows"},
            {"method":"GET",  "path":"/help",            "desc":"OQL language reference JSON"},
            {"method":"GET",  "path":"/schema",          "desc":"JSON Schema for QueryResult"},
            {"method":"GET",  "path":"/download/offline", "desc":"download a self-contained offline HTML report (no WASM)"},
            {"method":"GET",  "path":"/version",         "desc":"this document"}
        ]
    })
}

pub fn run_server(path: &str, port: u16, opts: AnalyzeOptions) -> io::Result<()> {
    let addr = format!("127.0.0.1:{port}");
    let server =
        Server::http(&addr).map_err(|e| io::Error::other(format!("bind {addr} failed: {e}")))?;
    let bound = server.server_addr();

    println!("hprof-analyzer server listening on http://{bound}");
    println!();
    println!("Analysis:");
    println!("  GET  /status                    → not_started | analyzing | ready | failed");
    println!("  POST /analyze                   → trigger full analysis (run first!)");
    println!();
    println!("Report sections (add ?format=md for Markdown):");
    println!("  GET  /report                    → full report JSON");
    println!("  GET  /report/overview           → SystemOverview JSON");
    println!("  GET  /report/leaks              → LeakSuspects JSON");
    println!("  GET  /report/top                → TopConsumers JSON");
    println!("  GET  /report/threads            → ThreadOverview JSON");
    println!("  GET  /download/offline          → self-contained offline HTML file (no WASM)");
    println!();
    println!("OQL queries:");
    println!("  POST /         -d 'SELECT …'    → JSON QueryResult");
    println!("  POST /query    -d 'SELECT …'    → alias of POST /");
    println!("  POST /stream   -d 'SELECT …'    → NDJSON rows");
    println!("  GET  /help                      → OQL language reference");
    println!("  GET  /schema                    → QueryResult JSON schema");
    println!("  GET  /version                   → endpoint catalog");
    println!();
    println!("After POST /analyze completes, OQL queries can use @retainedHeapSize,");
    println!("dominators(), and @inbounds.");
    println!();
    println!("examples:");
    println!("  curl -s http://{bound}/status");
    println!("  curl -s -X POST http://{bound}/analyze");
    println!("  curl -s 'http://{bound}/report?format=md'");
    println!("  curl -s http://{bound}/ -d 'SELECT @displayName FROM java.lang.Thread'");
    println!("  curl -s http://{bound}/ -d 'SELECT @retainedHeapSize FROM java.lang.Thread'");
    println!("(loopback only; Ctrl-C to stop)");

    let state = Arc::new(ServeState::new(path, opts)?);
    // Prewarm the OQL ReplCache in the background so the first query is fast.
    state.oql.prewarm();
    let server = Arc::new(server);
    let n_workers = std::thread::available_parallelism()
        .map(|n| n.get())
        .unwrap_or(4);
    let mut handles = Vec::with_capacity(n_workers);
    for _ in 0..n_workers {
        let server = Arc::clone(&server);
        let state = Arc::clone(&state);
        handles.push(std::thread::spawn(move || {
            loop {
                let mut request = match server.recv() {
                    Ok(r) => r,
                    Err(_) => break,
                };
                let method = request.method().as_str().to_string();
                let url = request.url().to_string();
                const MAX_BODY: usize = 64 * 1024; // 64 KB
                let mut body = String::new();
                let mut limited = request.as_reader().take(MAX_BODY as u64 + 1);
                let _ = limited.read_to_string(&mut body);
                if body.len() > MAX_BODY {
                    let resp = Response::from_string(
                        serde_json::json!({"ok":false,"error":{"kind":"body_too_large","message":"request body exceeds 64 KB limit"}}).to_string()
                    )
                    .with_status_code(413)
                    .with_header(tiny_http::Header::from_bytes("Content-Type", "application/json").unwrap());
                    let _ = request.respond(resp);
                    continue;
                }

                // ── Offline download — needs Content-Disposition header ────────
                let (path, _) = split_path_query(&url);
                if method == "GET" && path == "/download/offline" {
                    state.ensure_analysis();
                    let resp = match state.offline_html() {
                        Ok((bytes, stem)) => {
                            let disposition = format!(
                                "attachment; filename=\"{stem}-report.html\""
                            );
                            Response::from_data(bytes)
                                .with_status_code(200)
                                .with_header(tiny_http::Header::from_bytes(
                                    "Content-Type", "text/html; charset=utf-8",
                                ).unwrap())
                                .with_header(tiny_http::Header::from_bytes(
                                    "Content-Disposition", disposition.as_bytes(),
                                ).unwrap())
                        }
                        Err((status, msg)) => {
                            Response::from_string(msg)
                                .with_status_code(status)
                                .with_header(tiny_http::Header::from_bytes(
                                    "Content-Type", "text/plain; charset=utf-8",
                                ).unwrap())
                        }
                    };
                    let _ = request.respond(resp);
                    continue;
                }

                let (status, resp_body, ctype) = state.route(&method, &url, &body);
                let resp = Response::from_string(resp_body)
                    .with_status_code(status)
                    .with_header(
                        format!("Content-Type: {ctype}")
                            .parse::<tiny_http::Header>()
                            .unwrap(),
                    );
                let _ = request.respond(resp);
            }
        }));
    }
    for h in handles {
        let _ = h.join();
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    const FIXTURE: &str = "tests/fixtures/dump_4_philosophers.hprof";

    fn make_state() -> ServeState {
        ServeState::new(FIXTURE, AnalyzeOptions::default()).expect("ServeState::new")
    }

    /// Shared, lazily-initialized state that has already completed analysis.
    /// All tests that need a ready report use this to avoid running 7+ concurrent
    /// analyses (which races against the 30 s timeout on slow/loaded CI hosts).
    fn ready_state() -> &'static ServeState {
        static STATE: std::sync::OnceLock<ServeState> = std::sync::OnceLock::new();
        STATE.get_or_init(|| {
            let s = make_state();
            s.route("POST", "/analyze", "");
            wait_for_ready(&s);
            s
        })
    }

    fn wait_for_ready(state: &ServeState) {
        for _ in 0..120 {
            if state.report_ready() {
                return;
            }
            std::thread::sleep(std::time::Duration::from_millis(500));
        }
        panic!("analysis did not complete within 60s");
    }

    #[test]
    fn status_before_analyze_is_not_started() {
        let s = make_state();
        let (status, body, _) = s.route("GET", "/status", "");
        assert_eq!(status, 200, "body: {body}");
        let v: serde_json::Value = serde_json::from_str(&body).unwrap();
        assert_eq!(v["status"], "not_started", "got: {v}");
    }

    #[test]
    fn analyze_starts_and_returns_started() {
        let s = make_state();
        let (status, body, _) = s.route("POST", "/analyze", "");
        assert_eq!(status, 200, "body: {body}");
        let v: serde_json::Value = serde_json::from_str(&body).unwrap();
        assert!(v["ok"] == true, "ok flag: {v}");
        let st = v["status"].as_str().unwrap_or("");
        assert!(
            matches!(st, "started" | "already_running" | "already_done"),
            "status: {v}"
        );
    }

    #[test]
    fn status_ready_after_analysis() {
        let s = ready_state();
        let (_, body, _) = s.route("GET", "/status", "");
        let v: serde_json::Value = serde_json::from_str(&body).unwrap();
        assert_eq!(v["status"], "ready", "got: {v}");
    }

    #[test]
    fn report_json_has_schema_version() {
        let s = ready_state();
        let (status, body, _) = s.route("GET", "/report", "");
        assert_eq!(status, 200, "body: {body}");
        let v: serde_json::Value = serde_json::from_str(&body).unwrap();
        assert!(v["schema_version"].is_u64(), "schema_version missing: {v}");
    }

    #[test]
    fn report_overview_md_contains_heap() {
        let s = ready_state();
        let (status, body, ctype) = s.route("GET", "/report/overview?format=md", "");
        assert_eq!(status, 200, "body: {body}");
        assert!(ctype.starts_with("text/markdown"), "wrong ctype: {ctype}");
        assert!(
            body.to_lowercase().contains("heap"),
            "no 'heap' in md: {}",
            &body[..body.len().min(500)]
        );
    }

    #[test]
    fn report_leaks_json_has_expected_keys() {
        let s = ready_state();
        let (status, body, _) = s.route("GET", "/report/leaks", "");
        assert_eq!(status, 200, "body: {body}");
        let v: serde_json::Value = serde_json::from_str(&body).unwrap();
        assert!(
            v.get("suspects").is_some() || v.get("total_shallow").is_some(),
            "expected suspects or total_shallow key: {}",
            &body[..body.len().min(300)]
        );
    }

    #[test]
    fn report_top_json_is_valid() {
        let s = ready_state();
        let (status, body, _) = s.route("GET", "/report/top", "");
        assert_eq!(status, 200, "body: {body}");
        assert!(
            serde_json::from_str::<serde_json::Value>(&body).is_ok(),
            "not JSON: {body}"
        );
    }

    #[test]
    fn report_threads_json_is_valid() {
        let s = ready_state();
        let (status, body, _) = s.route("GET", "/report/threads", "");
        assert_eq!(status, 200, "body: {body}");
        assert!(
            serde_json::from_str::<serde_json::Value>(&body).is_ok(),
            "not JSON: {body}"
        );
    }

    #[test]
    fn oql_still_works_via_server() {
        let s = make_state();
        let (status, body, _) = s.route("POST", "/", "SELECT @objectAddress FROM java.lang.Thread");
        assert_eq!(status, 200, "body: {body}");
        let v: serde_json::Value = serde_json::from_str(&body).unwrap();
        assert_eq!(v["ok"], true, "oql failed: {v}");
    }

    #[test]
    fn version_lists_report_endpoints() {
        let s = make_state();
        let (_, body, _) = s.route("GET", "/version", "");
        let v: serde_json::Value = serde_json::from_str(&body).unwrap();
        let paths: Vec<&str> = v["endpoints"]
            .as_array()
            .unwrap()
            .iter()
            .map(|e| e["path"].as_str().unwrap())
            .collect();
        for p in [
            "/report",
            "/report/overview",
            "/report/leaks",
            "/report/top",
            "/report/threads",
        ] {
            assert!(paths.contains(&p), "missing {p}: {v}");
        }
    }

    #[test]
    fn offline_html_returns_valid_html_after_analysis() {
        let s = ready_state();
        let result = s.offline_html();
        assert!(result.is_ok(), "expected Ok, got: {:?}", result.err());
        let (bytes, stem) = result.unwrap();
        let html = String::from_utf8(bytes).expect("valid UTF-8");
        assert!(
            html.starts_with("<!DOCTYPE html>"),
            "expected HTML, got: {}",
            &html[..html.len().min(100)]
        );
        assert!(html.contains("<title>"), "expected <title> in HTML");
        assert!(!stem.is_empty(), "expected non-empty stem");
    }

    #[test]
    fn offline_html_returns_503_when_not_ready() {
        let s = make_state();
        // Don't trigger analysis; report() returns None → error path
        let result = s.offline_html();
        assert!(result.is_err(), "expected Err when analysis not started");
        let (status, _) = result.unwrap_err();
        assert_eq!(status, 503, "expected 503 status");
    }

    #[test]
    fn unknown_route_is_404() {
        let s = make_state();
        let (status, _, _) = s.route("GET", "/bogus", "");
        assert_eq!(status, 404);
    }

    #[test]
    fn wrong_method_is_405() {
        let s = make_state();
        let (status, body, _) = s.route("DELETE", "/analyze", "");
        assert_eq!(status, 405, "body: {body}");
        let v: serde_json::Value = serde_json::from_str(&body).unwrap();
        assert_eq!(v["error"]["kind"], "method", "got: {v}");
    }

    #[test]
    fn report_section_while_analyzing_is_202() {
        let s = make_state();
        {
            let mut g = s.state.lock().unwrap();
            *g = AnalysisState::Running;
        }
        let (status, body, _) = s.route("GET", "/report/overview", "");
        assert_eq!(status, 202, "expected 202 while running, body: {body}");
        let v: serde_json::Value = serde_json::from_str(&body).unwrap();
        assert_eq!(v["status"], "analyzing", "got: {v}");
    }
}