use anyhow::{anyhow, Result};
use axum::{
    body::Body,
    extract::{Path, Query, State},
    http::{header, HeaderValue, StatusCode, Uri},
    response::{IntoResponse, Response},
    routing::get,
    Json, Router,
};
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use std::collections::HashMap;
use std::io::ErrorKind;
use std::net::SocketAddr;
use std::path::{Path as FsPath, PathBuf};
use std::time::{SystemTime, UNIX_EPOCH};
use tokio::fs;
use tokio::time::Instant;

use moraine_clickhouse::ClickHouseClient;
use moraine_config::AppConfig;

#[derive(Clone)]
struct AppState {
    clickhouse: ClickHouseClient,
    static_dir: PathBuf,
}

#[derive(Deserialize)]
struct LimitQuery {
    limit: Option<u32>,
}

#[derive(Deserialize)]
struct AnalyticsQuery {
    range: Option<String>,
}

#[derive(Serialize)]
struct TableSummary {
    name: String,
    engine: String,
    is_temporary: u8,
    rows: u64,
}

pub async fn run_server(
    cfg: AppConfig,
    host: String,
    port: u16,
    static_dir: PathBuf,
) -> Result<()> {
    validate_static_dir(&static_dir)?;

    let clickhouse = ClickHouseClient::new(cfg.clickhouse)?;

    let state = AppState {
        clickhouse,
        static_dir,
    };

    let app = Router::new()
        .route("/api/health", get(api_health))
        .route("/api/status", get(api_status))
        .route("/api/analytics", get(api_analytics))
        .route("/api/tables", get(api_tables))
        .route("/api/web-searches", get(api_web_searches))
        .route("/api/tables/:table", get(api_table_rows))
        .route("/api/sessions", get(api_sessions))
        .fallback(get(static_fallback))
        .with_state(state.clone());

    let bind = format!("{}:{}", host, port)
        .parse::<SocketAddr>()
        .map_err(|err| anyhow!("invalid bind address: {err}"))?;

    println!("moraine-monitor running at http://{}", bind);
    println!("serving UI from {}", state.static_dir.display());

    let listener = tokio::net::TcpListener::bind(bind).await.map_err(|error| {
        if error.kind() == ErrorKind::AddrInUse {
            anyhow!(
                "failed to bind {bind}: address already in use. another monitor may already be running (including legacy cortex-monitor). stop it or rerun with `moraine run monitor -- --port <free-port>`"
            )
        } else {
            anyhow!("failed to bind {bind}: {error}")
        }
    })?;
    axum::serve(listener, app).await?;
    Ok(())
}

fn validate_static_dir(static_dir: &FsPath) -> Result<()> {
    let metadata = std::fs::metadata(static_dir).map_err(|error| {
        anyhow!(
            "monitor static directory `{}` is unavailable: {error}. if running from source, build UI assets with `(cd web/monitor && bun install --frozen-lockfile && bun run build)`; otherwise ensure packaged `web/monitor/dist` assets are installed or pass `--static-dir <path>`",
            static_dir.display()
        )
    })?;

    if !metadata.is_dir() {
        return Err(anyhow!(
            "monitor static directory `{}` is not a directory; pass `--static-dir <path>` pointing to a built monitor dist directory",
            static_dir.display()
        ));
    }

    let index_path = static_dir.join("index.html");
    if !index_path.is_file() {
        return Err(anyhow!(
            "monitor static directory `{}` does not contain `index.html`; build monitor assets or pass `--static-dir <path>`",
            static_dir.display()
        ));
    }

    Ok(())
}

fn json_response<T: Serialize>(payload: T, status: StatusCode) -> Response {
    let mut response = Json(payload).into_response();
    *response.status_mut() = status;
    response
}

fn clickhouse_ping_payload(version: String, ping_result: Result<()>, ping_ms: f64) -> Value {
    match ping_result {
        Ok(()) => json!({
            "healthy": true,
            "version": version,
            "ping_ms": ping_ms,
            "error": Value::Null,
        }),
        Err(error) => json!({
            "healthy": false,
            "version": version,
            "ping_ms": ping_ms,
            "error": error.to_string(),
        }),
    }
}

async fn api_health(State(state): State<AppState>) -> Response {
    let start = Instant::now();
    let connection_stats = query_clickhouse_connections(&state)
        .await
        .unwrap_or_else(|error| json!({"total": Value::Null, "error": error.to_string()}));

    match state.clickhouse.ping().await {
        Ok(()) => match state.clickhouse.version().await {
            Ok(version) => {
                let body = json!({
                    "ok": true,
                    "url": state.clickhouse.config().url,
                    "database": state.clickhouse.config().database,
                    "version": version,
                    "ping_ms": (Instant::elapsed(&start).as_secs_f64() * 1000.0),
                    "connections": connection_stats,
                });
                json_response(body, StatusCode::OK)
            }
            Err(error) => json_response(
                json!({
                    "ok": false,
                    "url": state.clickhouse.config().url,
                    "database": state.clickhouse.config().database,
                    "error": error.to_string(),
                    "connections": connection_stats,
                }),
                StatusCode::SERVICE_UNAVAILABLE,
            ),
        },
        Err(error) => json_response(
            json!({
                "ok": false,
                "url": state.clickhouse.config().url,
                "database": state.clickhouse.config().database,
                "error": error.to_string(),
                "connections": connection_stats,
            }),
            StatusCode::SERVICE_UNAVAILABLE,
        ),
    }
}

async fn api_status(State(state): State<AppState>) -> Response {
    let db_exists = match state
        .clickhouse
        .query_rows::<Value>(
            &format!(
                "SELECT name FROM system.databases WHERE name = '{}'",
                escape_literal(&state.clickhouse.config().database)
            ),
            None,
        )
        .await
    {
        Ok(rows) => !rows.is_empty(),
        Err(_) => false,
    };

    let tables = if db_exists {
        match query_table_summaries(&state).await {
            Ok(value) => value,
            Err(error) => {
                return json_response(
                    json!({"ok": false, "error": error.to_string()}),
                    StatusCode::SERVICE_UNAVAILABLE,
                );
            }
        }
    } else {
        Vec::new()
    };

    let estimated_total_rows = tables.iter().map(|table| table.rows).sum::<u64>();
    let heartbeat = if db_exists {
        query_heartbeat(&state).await.unwrap_or_else(|_| {
            json!({"present": false, "alive": false, "latest": Value::Null, "age_seconds": Value::Null})
        })
    } else {
        json!({"present": false, "alive": false, "latest": Value::Null, "age_seconds": Value::Null})
    };

    let clickhouse_ping = if db_exists {
        match state.clickhouse.version().await {
            Ok(version) => {
                let start = Instant::now();
                let ping_result = state.clickhouse.ping().await;
                let ping_ms = Instant::elapsed(&start).as_secs_f64() * 1000.0;
                clickhouse_ping_payload(version, ping_result, ping_ms)
            }
            Err(error) => {
                json!({
                    "healthy": false,
                    "version": Value::Null,
                    "ping_ms": Value::Null,
                    "error": error.to_string(),
                })
            }
        }
    } else {
        json!({
            "healthy": false,
            "version": Value::Null,
            "ping_ms": Value::Null,
            "error": "database not found",
        })
    };

    json_response(
        json!({
            "ok": true,
            "clickhouse": {
                "url": state.clickhouse.config().url,
            "database": state.clickhouse.config().database,
            "healthy": clickhouse_ping["healthy"],
            "version": clickhouse_ping["version"],
            "ping_ms": clickhouse_ping["ping_ms"],
            "error": clickhouse_ping["error"],
            "connections": if db_exists {
                query_clickhouse_connections(&state)
                    .await
                    .unwrap_or_else(|error| json!({"total": Value::Null, "error": error.to_string()}))
            } else {
                json!({"total": Value::Null, "error": "database not found"})
            },
        },
        "database": {
            "exists": db_exists,
            "table_count": tables.len(),
            "estimated_total_rows": estimated_total_rows,
                "tables": tables,
            },
            "ingestor": heartbeat,
        }),
        StatusCode::OK,
    )
}

async fn api_tables(State(state): State<AppState>) -> Response {
    match query_table_summaries(&state).await {
        Ok(tables) => json_response(json!({"ok": true, "tables": tables}), StatusCode::OK),
        Err(error) => json_response(
            json!({"ok": false, "error": error.to_string()}),
            StatusCode::SERVICE_UNAVAILABLE,
        ),
    }
}

async fn api_web_searches(
    Query(params): Query<LimitQuery>,
    State(state): State<AppState>,
) -> Response {
    #[derive(serde::Deserialize, serde::Serialize)]
    struct WebSearchRow {
        event_time: String,
        harness: String,
        source_name: String,
        session_id: String,
        model: String,
        action: String,
        search_query: String,
        result_url: String,
        source_ref: String,
    }

    let limit = params.limit.unwrap_or(100).clamp(1, 1000);
    let database = &state.clickhouse.config().database;
    let table = format!(
        "{}.{}",
        escape_identifier(database),
        escape_identifier("events")
    );

    let query = format!(
        "SELECT \
            toString(event_ts) AS event_time, \
            harness, \
            source_name, \
            session_id, \
            lowerUTF8(trim(BOTH ' ' FROM model)) AS model, \
            if(payload_type = 'web_search_call', op_kind, if(tool_name = 'WebFetch', 'open_page', if(tool_name = 'WebSearch', 'search', payload_type))) AS action, \
            if(length(JSONExtractString(payload_json, 'action', 'query')) > 0, \
               JSONExtractString(payload_json, 'action', 'query'), \
               if(length(JSONExtractString(payload_json, 'input', 'query')) > 0, \
                  JSONExtractString(payload_json, 'input', 'query'), \
                  if(length(JSONExtractString(payload_json, 'data', 'query')) > 0, \
                     JSONExtractString(payload_json, 'data', 'query'), \
                     text_content))) AS search_query, \
            if(length(JSONExtractString(payload_json, 'action', 'url')) > 0, \
               JSONExtractString(payload_json, 'action', 'url'), \
               JSONExtractString(payload_json, 'input', 'url')) AS result_url, \
            source_ref \
         FROM {table} \
         WHERE payload_type = 'web_search_call' \
            OR (payload_type = 'tool_use' AND tool_name IN ('WebSearch', 'WebFetch')) \
            OR payload_type = 'search_results_received' \
         ORDER BY event_ts DESC \
         LIMIT {limit}",
        table = table,
        limit = limit,
    );

    let rows = match state
        .clickhouse
        .query_rows::<WebSearchRow>(&query, None)
        .await
    {
        Ok(rows) => rows,
        Err(error) => {
            return json_response(
                json!({"ok": false, "error": format!("web search query failed: {error}")}),
                StatusCode::SERVICE_UNAVAILABLE,
            );
        }
    };

    json_response(
        json!({
            "ok": true,
            "table": "web_searches",
            "limit": limit,
            "schema": [
                {"name": "event_time", "type": "String", "default_expression": ""},
                {"name": "harness", "type": "String", "default_expression": ""},
                {"name": "source_name", "type": "String", "default_expression": ""},
                {"name": "session_id", "type": "String", "default_expression": ""},
                {"name": "model", "type": "String", "default_expression": ""},
                {"name": "action", "type": "String", "default_expression": ""},
                {"name": "search_query", "type": "String", "default_expression": ""},
                {"name": "result_url", "type": "String", "default_expression": ""},
                {"name": "source_ref", "type": "String", "default_expression": ""}
            ],
            "rows": rows,
        }),
        StatusCode::OK,
    )
}

#[derive(Clone, Copy)]
struct AnalyticsRange {
    key: &'static str,
    label: &'static str,
    window_seconds: u32,
    bucket_seconds: u32,
}

fn resolve_analytics_range(value: Option<&str>) -> AnalyticsRange {
    match value.unwrap_or("24h") {
        "15m" => AnalyticsRange {
            key: "15m",
            label: "Last 15m",
            window_seconds: 15 * 60,
            bucket_seconds: 60,
        },
        "1h" => AnalyticsRange {
            key: "1h",
            label: "Last 1h",
            window_seconds: 60 * 60,
            bucket_seconds: 5 * 60,
        },
        "6h" => AnalyticsRange {
            key: "6h",
            label: "Last 6h",
            window_seconds: 6 * 60 * 60,
            bucket_seconds: 15 * 60,
        },
        "24h" => AnalyticsRange {
            key: "24h",
            label: "Last 24h",
            window_seconds: 24 * 60 * 60,
            bucket_seconds: 60 * 60,
        },
        "7d" => AnalyticsRange {
            key: "7d",
            label: "Last 7d",
            window_seconds: 7 * 24 * 60 * 60,
            bucket_seconds: 6 * 60 * 60,
        },
        "30d" => AnalyticsRange {
            key: "30d",
            label: "Last 30d",
            window_seconds: 30 * 24 * 60 * 60,
            bucket_seconds: 24 * 60 * 60,
        },
        _ => AnalyticsRange {
            key: "24h",
            label: "Last 24h",
            window_seconds: 24 * 60 * 60,
            bucket_seconds: 60 * 60,
        },
    }
}

fn analytics_window_query(table: &str, window_filter: &str, window_seconds: u32) -> String {
    format!(
        "SELECT \
           toUInt64(anchor_unix) AS now_unix, \
           toUInt64(greatest(anchor_unix - toInt64({window_seconds}), toInt64(0))) AS from_unix \
         FROM ( \
           SELECT \
             if( \
               count() = 0, \
               toInt64(toUnixTimestamp(now())), \
               toInt64(intDiv(toUnixTimestamp64Milli(max(event_ts)), 1000)) \
             ) AS anchor_unix \
           FROM {table} \
           WHERE {window_filter} \
         )",
    )
}

async fn api_analytics(
    Query(params): Query<AnalyticsQuery>,
    State(state): State<AppState>,
) -> Response {
    #[derive(serde::Deserialize, serde::Serialize)]
    struct TokenRow {
        bucket_unix: u64,
        model: String,
        endpoint_kind: String,
        bucket: String,
        tokens: u64,
    }

    #[derive(serde::Deserialize, serde::Serialize)]
    struct TurnRow {
        bucket_unix: u64,
        model: String,
        turns: u64,
    }

    #[derive(serde::Deserialize, serde::Serialize)]
    struct ConcurrentRow {
        bucket_unix: u64,
        concurrent_sessions: u64,
    }

    let range = resolve_analytics_range(params.range.as_deref());
    let database = &state.clickhouse.config().database;
    let table = format!(
        "{}.{}",
        escape_identifier(database),
        escape_identifier("events")
    );
    let model_expr = "if(lowerUTF8(trim(BOTH ' ' FROM model)) = 'codex', 'gpt-5.3-codex-xhigh', lowerUTF8(trim(BOTH ' ' FROM model)))";
    let model_expr_latest = "if(lowerUTF8(trim(BOTH ' ' FROM model_latest)) = 'codex', 'gpt-5.3-codex-xhigh', lowerUTF8(trim(BOTH ' ' FROM model_latest)))";
    let window_filter = format!(
        "event_ts >= now() - INTERVAL {} SECOND AND length(trim(BOTH ' ' FROM model)) > 0 AND lowerUTF8(trim(BOTH ' ' FROM model)) != '<synthetic>'",
        range.window_seconds
    );
    let token_latest_filter = format!(
        "event_ts_latest >= now() - INTERVAL {} SECOND AND length(trim(BOTH ' ' FROM model_latest)) > 0 AND lowerUTF8(trim(BOTH ' ' FROM model_latest)) != '<synthetic>' AND arraySum(mapValues(token_usage_buckets_latest)) > 0",
        range.window_seconds
    );
    let turn_filter = format!(
        "{} AND length(trim(BOTH ' ' FROM request_id)) > 0",
        window_filter
    );
    let concurrent_filter = format!(
        "event_ts >= now() - INTERVAL {} SECOND AND length(trim(BOTH ' ' FROM session_id)) > 0 AND arraySum(mapValues(token_usage_buckets)) > 0",
        range.window_seconds
    );

    let token_query = format!(
        "SELECT bucket_unix, model, endpoint_kind, bucket, toUInt64(sum(tokens)) AS tokens \
         FROM ( \
           SELECT bucket_unix, model, endpoint_kind, bucket, toUInt64(max(tokens_latest)) AS tokens \
           FROM ( \
             SELECT \
               toUInt64(toUnixTimestamp(toStartOfInterval(event_ts_latest, INTERVAL {bucket_seconds} SECOND))) AS bucket_unix, \
               {model_expr_latest} AS model, \
               endpoint_kind_latest AS endpoint_kind, \
               harness_latest, \
               session_id_latest, \
               request_id_latest, \
               bucket, \
               toUInt64(tokens_latest) AS tokens_latest \
             FROM ( \
               SELECT \
                 event_uid, \
                 argMax(event_ts, event_version) AS event_ts_latest, \
                 argMax(model, event_version) AS model_latest, \
                 argMax(endpoint_kind, event_version) AS endpoint_kind_latest, \
                 argMax(harness, event_version) AS harness_latest, \
                 argMax(session_id, event_version) AS session_id_latest, \
                 argMax(request_id, event_version) AS request_id_latest, \
                 argMax(token_usage_buckets, event_version) AS token_usage_buckets_latest \
               FROM {table} \
               GROUP BY event_uid \
             ) ARRAY JOIN mapKeys(token_usage_buckets_latest) AS bucket, mapValues(token_usage_buckets_latest) AS tokens_latest \
             WHERE {token_latest_filter} AND tokens_latest > 0 \
           ) \
           WHERE harness_latest = 'claude-code' AND length(trim(BOTH ' ' FROM request_id_latest)) > 0 \
           GROUP BY bucket_unix, model, endpoint_kind, session_id_latest, request_id_latest, bucket \
           UNION ALL \
           SELECT \
             toUInt64(toUnixTimestamp(toStartOfInterval(event_ts_latest, INTERVAL {bucket_seconds} SECOND))) AS bucket_unix, \
             {model_expr_latest} AS model, \
             endpoint_kind_latest AS endpoint_kind, \
             bucket, \
             toUInt64(tokens_latest) AS tokens \
           FROM ( \
             SELECT \
               event_uid, \
               argMax(event_ts, event_version) AS event_ts_latest, \
               argMax(model, event_version) AS model_latest, \
               argMax(endpoint_kind, event_version) AS endpoint_kind_latest, \
               argMax(harness, event_version) AS harness_latest, \
               argMax(session_id, event_version) AS session_id_latest, \
               argMax(request_id, event_version) AS request_id_latest, \
               argMax(token_usage_buckets, event_version) AS token_usage_buckets_latest \
             FROM {table} \
             GROUP BY event_uid \
           ) ARRAY JOIN mapKeys(token_usage_buckets_latest) AS bucket, mapValues(token_usage_buckets_latest) AS tokens_latest \
           WHERE {token_latest_filter} AND tokens_latest > 0 \
           AND NOT (harness_latest = 'claude-code' AND length(trim(BOTH ' ' FROM request_id_latest)) > 0) \
         ) \
         GROUP BY bucket_unix, model, endpoint_kind, bucket \
         ORDER BY bucket_unix ASC, model ASC, endpoint_kind ASC, bucket ASC",
        bucket_seconds = range.bucket_seconds,
        model_expr_latest = model_expr_latest,
        table = table,
        token_latest_filter = token_latest_filter,
    );

    let turns_query = format!(
        "SELECT \
            toUInt64(toUnixTimestamp(toStartOfInterval(event_ts, INTERVAL {bucket_seconds} SECOND))) AS bucket_unix, \
            {model_expr} AS model, \
            toUInt64(uniqExact(tuple(session_id, request_id))) AS turns \
         FROM {table} \
         WHERE {turn_filter} \
         GROUP BY bucket_unix, model \
         ORDER BY bucket_unix ASC, model ASC",
        bucket_seconds = range.bucket_seconds,
        model_expr = model_expr,
        table = table,
        turn_filter = turn_filter,
    );

    let concurrent_query = format!(
        "SELECT bucket_unix, toUInt64(uniqExact(session_stream_key)) AS concurrent_sessions \
         FROM ( \
           SELECT \
             toUInt64(toUnixTimestamp(toStartOfInterval(event_ts, INTERVAL {bucket_seconds} SECOND))) AS bucket_unix, \
             if(harness = 'claude-code' AND length(trim(BOTH ' ' FROM agent_run_id)) > 0, concat(session_id, '::', agent_run_id), session_id) AS session_stream_key \
           FROM {table} \
           WHERE {concurrent_filter} \
         ) \
         GROUP BY bucket_unix \
         ORDER BY bucket_unix ASC",
        bucket_seconds = range.bucket_seconds,
        table = table,
        concurrent_filter = concurrent_filter,
    );

    let token_rows = match state
        .clickhouse
        .query_rows::<TokenRow>(&token_query, None)
        .await
    {
        Ok(rows) => rows,
        Err(error) => {
            return json_response(
                json!({"ok": false, "error": format!("analytics token query failed: {error}")}),
                StatusCode::SERVICE_UNAVAILABLE,
            );
        }
    };

    let turn_rows = match state
        .clickhouse
        .query_rows::<TurnRow>(&turns_query, None)
        .await
    {
        Ok(rows) => rows,
        Err(error) => {
            return json_response(
                json!({"ok": false, "error": format!("analytics turns query failed: {error}")}),
                StatusCode::SERVICE_UNAVAILABLE,
            );
        }
    };

    let concurrent_rows = match state
        .clickhouse
        .query_rows::<ConcurrentRow>(&concurrent_query, None)
        .await
    {
        Ok(rows) => rows,
        Err(error) => {
            return json_response(
                json!({"ok": false, "error": format!("analytics concurrent-session query failed: {error}")}),
                StatusCode::SERVICE_UNAVAILABLE,
            );
        }
    };

    #[derive(serde::Deserialize)]
    struct WindowRow {
        now_unix: u64,
        from_unix: u64,
    }

    let window_query = analytics_window_query(&table, &window_filter, range.window_seconds);

    let (now_unix, from_unix) = match state
        .clickhouse
        .query_rows::<WindowRow>(&window_query, None)
        .await
    {
        Ok(rows) if !rows.is_empty() => {
            let row = &rows[0];
            (row.now_unix, row.from_unix)
        }
        _ => {
            let now = SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .map(|d| d.as_secs())
                .unwrap_or(0);
            (now, now.saturating_sub(range.window_seconds as u64))
        }
    };

    json_response(
        json!({
            "ok": true,
            "range": {
                "key": range.key,
                "label": range.label,
                "window_seconds": range.window_seconds,
                "bucket_seconds": range.bucket_seconds,
                "from_unix": from_unix,
                "to_unix": now_unix,
            },
            "series": {
                "tokens": token_rows,
                "turns": turn_rows,
                "concurrent_sessions": concurrent_rows,
            }
        }),
        StatusCode::OK,
    )
}

#[derive(Deserialize)]
struct SessionsQuery {
    limit: Option<u32>,
    since: Option<String>,
}

async fn api_sessions(
    Query(params): Query<SessionsQuery>,
    State(state): State<AppState>,
) -> Response {
    let limit = params.limit.unwrap_or(50).clamp(1, 200);
    let since = params.since.as_deref().unwrap_or("30d");

    let since_seconds: u64 = match since {
        "1h" => 60 * 60,
        "6h" => 6 * 60 * 60,
        "24h" => 24 * 60 * 60,
        "7d" => 7 * 24 * 60 * 60,
        "30d" => 30 * 24 * 60 * 60,
        "90d" => 90 * 24 * 60 * 60,
        "all" => 0,
        _ => 30 * 24 * 60 * 60,
    };

    match build_sessions_payload(&state, limit, since_seconds).await {
        Ok(payload) => json_response(payload, StatusCode::OK),
        Err(error) => json_response(
            json!({"ok": false, "error": format!("sessions query failed: {error}")}),
            StatusCode::SERVICE_UNAVAILABLE,
        ),
    }
}

#[derive(serde::Deserialize)]
struct SessionSummaryRow {
    session_id: String,
    harness: String,
    source_name: String,
    started_at_ms: i64,
    ended_at_ms: i64,
    models_blob: String,
    trace_id: String,
    first_user_text: String,
}

#[derive(serde::Deserialize)]
struct SessionEventRow {
    session_id: String,
    event_ts_ms: i64,
    event_kind: String,
    actor_kind: String,
    payload_type: String,
    tool_name: String,
    tool_call_id: String,
    tool_error: u8,
    latency_ms: u32,
    model: String,
    endpoint_kind: String,
    input_tokens: u32,
    output_tokens: u32,
    cache_read_tokens: u32,
    cache_write_tokens: u32,
    #[serde(default)]
    token_usage_buckets: HashMap<String, u64>,
    #[serde(default)]
    token_usage_native_units: HashMap<String, f64>,
    text_preview: String,
    text_content: String,
    tool_args_json: String,
}

fn events_table_ref(database: &str) -> String {
    format!(
        "{}.{}",
        escape_identifier(database),
        escape_identifier("events")
    )
}

fn deduped_events_source(events_table: &str, filter: &str) -> String {
    let filter = filter.trim();
    debug_assert!(!filter.is_empty());

    format!(
        "(SELECT * FROM {events_table} WHERE {filter} \
         ORDER BY event_uid ASC, event_version DESC, ingested_at DESC \
         LIMIT 1 BY event_uid)"
    )
}

fn session_summary_query(events_table: &str, limit: u32, since_seconds: u64) -> String {
    let recency_filter = if since_seconds == 0 {
        String::new()
    } else {
        format!(" AND event_ts >= now() - INTERVAL {} SECOND", since_seconds)
    };
    let source_filter = format!("length(trim(BOTH ' ' FROM session_id)) > 0{recency_filter}");
    let events_source = deduped_events_source(events_table, &source_filter);

    format!(
        "SELECT \
            session_id, \
            any(harness) AS harness, \
            any(source_name) AS source_name, \
            toInt64(toUnixTimestamp64Milli(min(event_ts))) AS started_at_ms, \
            toInt64(toUnixTimestamp64Milli(max(event_ts))) AS ended_at_ms, \
            arrayStringConcat( \
                arrayFilter(m -> length(m) > 0, \
                    groupUniqArray(lowerUTF8(trim(BOTH ' ' FROM model)))), \
                '\u{1f}' \
            ) AS models_blob, \
            any(if(length(trim(BOTH ' ' FROM trace_id)) > 0, trace_id, '')) AS trace_id, \
            argMin( \
                if(length(text_content) > 0, substring(text_content, 1, 400), substring(text_preview, 1, 400)), \
                if(event_kind = 'message' AND actor_kind = 'user', event_ts, toDateTime64('2099-01-01', 3)) \
            ) AS first_user_text \
         FROM {events_source} AS deduped_events \
         GROUP BY session_id \
         ORDER BY ended_at_ms DESC \
         LIMIT {limit}"
    )
}

fn session_events_query(events_table: &str, session_ids: &[String]) -> String {
    debug_assert!(!session_ids.is_empty());
    let id_list = session_ids
        .iter()
        .map(|id| format!("'{}'", escape_literal(id)))
        .collect::<Vec<_>>()
        .join(",");
    let source_filter = format!("session_id IN ({id_list})");
    let events_source = deduped_events_source(events_table, &source_filter);

    format!(
        "SELECT \
            session_id, \
            toInt64(toUnixTimestamp64Milli(event_ts)) AS event_ts_ms, \
            event_kind, \
            actor_kind, \
            payload_type, \
            tool_name, \
            tool_call_id, \
            tool_error, \
            latency_ms, \
            model, \
            endpoint_kind, \
            input_tokens, \
            output_tokens, \
            cache_read_tokens, \
            cache_write_tokens, \
            token_usage_buckets, \
            token_usage_native_units, \
            substring(text_preview, 1, 2000) AS text_preview, \
            substring(text_content, 1, 2000) AS text_content, \
            if(event_kind = 'tool_call', substring(JSONExtractRaw(payload_json, 'input'), 1, 2000), '') AS tool_args_json \
         FROM {events_source} AS deduped_events \
         ORDER BY session_id, event_ts, source_file, source_generation, source_offset, source_line_no, event_uid"
    )
}

async fn build_sessions_payload(state: &AppState, limit: u32, since_seconds: u64) -> Result<Value> {
    let database = &state.clickhouse.config().database;
    let events_table = events_table_ref(database);
    let summary_query = session_summary_query(&events_table, limit, since_seconds);

    let summary_rows = state
        .clickhouse
        .query_rows::<SessionSummaryRow>(&summary_query, None)
        .await?;

    if summary_rows.is_empty() {
        return Ok(json!({"ok": true, "sessions": []}));
    }

    let session_ids: Vec<String> = summary_rows.iter().map(|r| r.session_id.clone()).collect();
    let events_query = session_events_query(&events_table, &session_ids);

    let event_rows = state
        .clickhouse
        .query_rows::<SessionEventRow>(&events_query, None)
        .await?;

    let mut events_by_session: HashMap<String, Vec<SessionEventRow>> = HashMap::new();
    for row in event_rows {
        events_by_session
            .entry(row.session_id.clone())
            .or_default()
            .push(row);
    }

    let now_ms = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_millis() as i64)
        .unwrap_or(0);

    let sessions: Vec<Value> = summary_rows
        .into_iter()
        .map(|summary| {
            let events = events_by_session
                .remove(&summary.session_id)
                .unwrap_or_default();
            build_session_json(summary, events, now_ms)
        })
        .collect();

    Ok(json!({"ok": true, "sessions": sessions}))
}

fn build_session_json(
    summary: SessionSummaryRow,
    events: Vec<SessionEventRow>,
    now_ms: i64,
) -> Value {
    let duration_ms = (summary.ended_at_ms - summary.started_at_ms).max(0);

    let status = if now_ms - summary.ended_at_ms < 60_000 {
        "active"
    } else {
        "completed"
    };

    let models: Vec<String> = if summary.models_blob.is_empty() {
        Vec::new()
    } else {
        summary
            .models_blob
            .split('\u{1f}')
            .filter(|s| !s.is_empty())
            .map(|s| s.to_string())
            .collect()
    };

    let title = derive_title(&summary.first_user_text);

    let turns = build_turns(&events);
    let total_tokens: u64 = turns
        .iter()
        .map(|t| t.get("totalTokens").and_then(|v| v.as_u64()).unwrap_or(0))
        .sum();
    let total_tool_calls: u64 = turns
        .iter()
        .map(|t| t.get("toolCalls").and_then(|v| v.as_u64()).unwrap_or(0))
        .sum();

    let harness = harness_descriptor(&summary.harness, &summary.source_name);

    json!({
        "id": summary.session_id,
        "title": title,
        "harness": harness,
        "startedAt": summary.started_at_ms,
        "endedAt": summary.ended_at_ms,
        "durationMs": duration_ms,
        "status": status,
        "models": models,
        "turns": turns,
        "totalTokens": total_tokens,
        "totalToolCalls": total_tool_calls,
        "tags": Vec::<String>::new(),
        "traceId": summary.trace_id,
    })
}

fn derive_title(first_user_text: &str) -> String {
    let trimmed = first_user_text.trim();
    if trimmed.is_empty() {
        return "(untitled session)".to_string();
    }

    let first_line = trimmed.lines().next().unwrap_or(trimmed);
    let clean = first_line.trim();
    let max_chars = 120;
    let char_count = clean.chars().count();
    if char_count <= max_chars {
        clean.to_string()
    } else {
        let truncated: String = clean.chars().take(max_chars).collect();
        format!("{truncated}\u{2026}")
    }
}

fn harness_descriptor(harness_id: &str, source_name: &str) -> Value {
    let id = if harness_id.trim().is_empty() {
        source_name.trim()
    } else {
        harness_id.trim()
    };
    let id = if id.is_empty() { "unknown" } else { id };
    let label = id.to_string();
    let short: String = id
        .split(|c: char| !c.is_ascii_alphanumeric())
        .filter(|part| !part.is_empty())
        .take(2)
        .filter_map(|part| part.chars().next())
        .collect::<String>()
        .to_uppercase();
    let short = if short.is_empty() {
        id.chars().take(2).collect::<String>().to_uppercase()
    } else {
        short
    };

    let hue = hue_for_label(id);

    json!({
        "id": id,
        "label": label,
        "short": short,
        "hue": hue,
    })
}

fn hue_for_label(label: &str) -> u32 {
    match label {
        "claude-code" => 25,
        "codex" => 150,
        "hermes" => 265,
        "cursor" => 200,
        "aider" => 340,
        "continue" => 100,
        "cli" => 60,
        _ => {
            let mut hash: u32 = 0;
            for byte in label.bytes() {
                hash = hash.wrapping_mul(31).wrapping_add(byte as u32);
            }
            hash % 360
        }
    }
}

fn token_bucket(event: &SessionEventRow, bucket: &str) -> u64 {
    event.token_usage_buckets.get(bucket).copied().unwrap_or(0)
}

fn sum_token_buckets(event: &SessionEventRow, buckets: &[&str]) -> u64 {
    buckets
        .iter()
        .map(|bucket| token_bucket(event, bucket))
        .sum()
}

fn event_token_total(event: &SessionEventRow) -> u64 {
    let canonical_total: u64 = event.token_usage_buckets.values().copied().sum();
    if canonical_total > 0 {
        canonical_total
    } else {
        (event.input_tokens
            + event.output_tokens
            + event.cache_read_tokens
            + event.cache_write_tokens) as u64
    }
}

fn event_prompt_tokens(event: &SessionEventRow) -> u64 {
    let canonical_prompt = sum_token_buckets(
        event,
        &[
            "input_text",
            "input_cache_read",
            "input_cache_write",
            "input_image",
            "input_audio",
            "embedding_input_text",
            "embedding_input_image",
        ],
    );
    if canonical_prompt > 0 {
        canonical_prompt
    } else {
        (event.input_tokens + event.cache_read_tokens + event.cache_write_tokens) as u64
    }
}

fn event_output_tokens(event: &SessionEventRow) -> u64 {
    let canonical_output = sum_token_buckets(
        event,
        &[
            "output_text",
            "output_image",
            "output_audio",
            "reasoning",
            "server_tool_use",
        ],
    );
    if canonical_output > 0 {
        canonical_output
    } else {
        event.output_tokens as u64
    }
}

fn nonzero_native_units(event: &SessionEventRow) -> Value {
    let units = event
        .token_usage_native_units
        .iter()
        .filter(|(_, value)| **value > 0.0)
        .map(|(key, value)| (key.clone(), json!(value)))
        .collect::<serde_json::Map<_, _>>();
    Value::Object(units)
}

fn build_turns(events: &[SessionEventRow]) -> Vec<Value> {
    struct TurnBuilder {
        idx: u32,
        model: String,
        started_at: i64,
        ended_at: i64,
        steps: Vec<Value>,
        prompt_tokens: u64,
        completion_tokens: u64,
        tool_calls: u64,
    }

    if events.is_empty() {
        return Vec::new();
    }

    let mut turns: Vec<TurnBuilder> = Vec::new();
    let mut current: Option<TurnBuilder> = None;
    let mut open_tool_calls: HashMap<String, usize> = HashMap::new();

    for event in events {
        let starts_new_turn = event.event_kind == "message" && event.actor_kind == "user";

        if starts_new_turn {
            if let Some(finished) = current.take() {
                turns.push(finished);
            }
            current = Some(TurnBuilder {
                idx: turns.len() as u32,
                model: String::new(),
                started_at: event.event_ts_ms,
                ended_at: event.event_ts_ms,
                steps: Vec::new(),
                prompt_tokens: 0,
                completion_tokens: 0,
                tool_calls: 0,
            });
            open_tool_calls.clear();
        }

        let turn = match current.as_mut() {
            Some(t) => t,
            None => continue,
        };

        turn.ended_at = turn.ended_at.max(event.event_ts_ms);
        let event_prompt = event_prompt_tokens(event);
        let event_total = event_token_total(event);
        turn.prompt_tokens += event_prompt;
        turn.completion_tokens += event_total.saturating_sub(event_prompt);
        if !event.model.trim().is_empty() && turn.model.is_empty() {
            turn.model = event.model.trim().to_string();
        }

        match (
            event.event_kind.as_str(),
            event.actor_kind.as_str(),
            event.payload_type.as_str(),
        ) {
            ("message", "user", _) => {
                turn.steps.push(json!({
                    "kind": "user",
                    "at": event.event_ts_ms,
                    "text": preferred_text(event),
                }));
            }
            ("message", "assistant", _) | (_, "assistant", "text") => {
                let mut step = json!({
                    "kind": "assistant",
                    "at": event.event_ts_ms,
                    "text": preferred_text(event),
                    "tokens": event_output_tokens(event),
                    "endpointKind": event.endpoint_kind,
                    "nativeTokenUnits": nonzero_native_units(event),
                });
                if event.latency_ms > 0 {
                    if let Some(obj) = step.as_object_mut() {
                        obj.insert("durationMs".into(), json!(event.latency_ms));
                    }
                }
                turn.steps.push(step);
            }
            ("reasoning", _, _) | (_, _, "thinking") => {
                let text = preferred_text(event);
                if !text.is_empty() {
                    turn.steps.push(json!({
                        "kind": "thinking",
                        "at": event.event_ts_ms,
                        "text": text,
                    }));
                }
            }
            ("tool_call", _, _) | (_, _, "tool_use") => {
                turn.tool_calls += 1;
                let step_index = turn.steps.len();
                if !event.tool_call_id.is_empty() {
                    open_tool_calls.insert(event.tool_call_id.clone(), step_index);
                }
                let args = if event.tool_args_json.trim().is_empty() {
                    Value::Object(Default::default())
                } else {
                    serde_json::from_str::<Value>(&event.tool_args_json)
                        .unwrap_or_else(|_| Value::Object(Default::default()))
                };
                turn.steps.push(json!({
                    "kind": "tool_call",
                    "at": event.event_ts_ms,
                    "tool": if event.tool_name.is_empty() { "tool".to_string() } else { event.tool_name.clone() },
                    "args": args,
                    "latencyMs": event.latency_ms,
                    "result": "",
                    "resultAt": event.event_ts_ms,
                    "status": if event.tool_error != 0 { "error" } else { "ok" },
                    "callId": event.tool_call_id,
                }));
            }
            ("tool_result", _, _) | (_, "tool", "tool_result") => {
                let result_text = preferred_text(event);
                if let Some(&step_index) = open_tool_calls.get(&event.tool_call_id) {
                    if let Some(step) = turn.steps.get_mut(step_index) {
                        if let Some(obj) = step.as_object_mut() {
                            let call_at = obj
                                .get("at")
                                .and_then(|v| v.as_i64())
                                .unwrap_or(event.event_ts_ms);
                            obj.insert("resultAt".into(), json!(event.event_ts_ms));
                            obj.insert(
                                "latencyMs".into(),
                                json!((event.event_ts_ms - call_at).max(0) as u32),
                            );
                            obj.insert("result".into(), json!(result_text));
                            if event.tool_error != 0 {
                                obj.insert("status".into(), json!("error"));
                            }
                        }
                    }
                    open_tool_calls.remove(&event.tool_call_id);
                }
            }
            _ => {}
        }
    }

    if let Some(finished) = current.take() {
        turns.push(finished);
    }

    turns
        .into_iter()
        .map(|builder| {
            let duration = (builder.ended_at - builder.started_at).max(0);
            let total = builder.prompt_tokens + builder.completion_tokens;
            json!({
                "idx": builder.idx,
                "model": builder.model,
                "startedAt": builder.started_at,
                "endedAt": builder.ended_at,
                "durationMs": duration,
                "promptTokens": builder.prompt_tokens,
                "completionTokens": builder.completion_tokens,
                "totalTokens": total,
                "toolCalls": builder.tool_calls,
                "steps": builder.steps,
            })
        })
        .collect()
}

fn preferred_text(event: &SessionEventRow) -> String {
    if !event.text_content.trim().is_empty() {
        event.text_content.trim().to_string()
    } else {
        event.text_preview.trim().to_string()
    }
}

async fn query_table_summaries(state: &AppState) -> Result<Vec<TableSummary>> {
    #[derive(serde::Deserialize)]
    struct TableRow {
        name: String,
        engine: String,
        is_temporary: u8,
    }

    #[derive(serde::Deserialize)]
    struct PartsRow {
        table: String,
        rows: u64,
    }

    let database = &state.clickhouse.config().database;
    let tables = state
        .clickhouse
        .query_rows::<TableRow>(&format!(
            "SELECT name, engine, is_temporary FROM system.tables WHERE database = '{}' ORDER BY name",
            escape_literal(database)
        ), None)
        .await?;

    let parts = state
        .clickhouse
        .query_rows::<PartsRow>(&format!(
            "SELECT table, SUM(rows) AS rows FROM system.parts WHERE database = '{}' AND active GROUP BY table",
            escape_literal(database)
        ), None)
        .await
        .unwrap_or_default();

    let counts: HashMap<String, u64> = parts.into_iter().map(|row| (row.table, row.rows)).collect();

    let values = tables
        .into_iter()
        .map(|row| TableSummary {
            name: row.name.clone(),
            engine: row.engine,
            is_temporary: row.is_temporary,
            rows: counts.get(&row.name).copied().unwrap_or(0),
        })
        .collect();

    Ok(values)
}

async fn query_heartbeat(state: &AppState) -> Result<Value> {
    let database = &state.clickhouse.config().database;
    // Column-level probe: presence of any column means the table exists, and
    // the optional backend_sinks column (migration 017) is only selected when
    // the database actually carries it.
    #[derive(serde::Deserialize)]
    struct ColumnRow {
        name: String,
    }
    let columns = state
        .clickhouse
        .query_rows::<ColumnRow>(
            &format!(
            "SELECT name FROM system.columns WHERE database = '{}' AND table = 'ingest_heartbeats'",
            escape_literal(database)
        ),
            None,
        )
        .await?;

    if columns.is_empty() {
        return Ok(json!({
            "present": false,
            "alive": false,
            "latest": Value::Null,
            "age_seconds": Value::Null,
        }));
    }

    let backend_sinks_select = if columns.iter().any(|col| col.name == "backend_sinks") {
        ", backend_sinks"
    } else {
        ""
    };
    let query = format!(
        "SELECT ts, toUnixTimestamp64Milli(ts) AS ts_unix_ms, host, service_version, queue_depth, files_active, files_watched, rows_raw_written, rows_events_written, rows_errors_written, flush_latency_ms, append_to_visible_p50_ms, append_to_visible_p95_ms, last_error{} FROM {}.ingest_heartbeats ORDER BY ts DESC LIMIT 1",
        backend_sinks_select,
        escape_identifier(database)
    );

    let latest = state
        .clickhouse
        .query_rows::<Value>(&query, None)
        .await?
        .into_iter()
        .next();

    let Some(mut latest) = latest else {
        return Ok(
            json!({"present": false, "alive": false, "latest": Value::Null, "age_seconds": Value::Null}),
        );
    };

    // The sink stores backend_sinks as a JSON-encoded string; decode it so
    // /api/health consumers see a structured `{backend: status}` map.
    if let Some(obj) = latest.as_object_mut() {
        if let Some(Value::String(raw)) = obj.get("backend_sinks") {
            let decoded = if raw.trim().is_empty() {
                json!({})
            } else {
                serde_json::from_str::<Value>(raw).unwrap_or_else(|_| Value::String(raw.clone()))
            };
            obj.insert("backend_sinks".to_string(), decoded);
        }
    }

    let age_seconds = latest
        .get("ts_unix_ms")
        .and_then(value_to_i64)
        .and_then(|ts_ms| {
            if ts_ms < 0 {
                return None;
            }

            let now_ms = SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .ok()?
                .as_millis() as i64;
            if now_ms < ts_ms {
                return Some(0);
            }

            Some(((now_ms - ts_ms) / 1000) as u64)
        });

    Ok(json!({
        "present": true,
        "latest": latest,
        "age_seconds": age_seconds,
        "alive": age_seconds.map(|age| age <= 30).unwrap_or(false),
    }))
}

async fn query_clickhouse_connections(state: &AppState) -> Result<Value> {
    #[derive(serde::Deserialize)]
    struct MetricRow {
        metric: String,
        value: u64,
    }

    let metrics = state
        .clickhouse
        .query_rows::<MetricRow>(
            "SELECT metric, value FROM system.metrics WHERE metric IN ('TCPConnection','HTTPConnection','MySQLConnection','PostgreSQLConnection','InterserverConnection')",
            None,
        )
        .await?;

    let mut tcp = 0_u64;
    let mut http = 0_u64;
    let mut mysql = 0_u64;
    let mut postgres = 0_u64;
    let mut interserver = 0_u64;

    for row in metrics {
        match row.metric.as_str() {
            "TCPConnection" => tcp = tcp.saturating_add(row.value),
            "HTTPConnection" => http = http.saturating_add(row.value),
            "MySQLConnection" => mysql = mysql.saturating_add(row.value),
            "PostgreSQLConnection" => postgres = postgres.saturating_add(row.value),
            "InterserverConnection" => interserver = interserver.saturating_add(row.value),
            _ => {}
        }
    }

    let total = tcp
        .saturating_add(http)
        .saturating_add(mysql)
        .saturating_add(postgres)
        .saturating_add(interserver);

    Ok(json!({
        "total": total,
        "tcp": tcp,
        "http": http,
        "mysql": mysql,
        "postgres": postgres,
        "interserver": interserver,
    }))
}

async fn api_table_rows(
    Path(table): Path<String>,
    Query(params): Query<LimitQuery>,
    State(state): State<AppState>,
) -> Response {
    if !is_safe_identifier(&table) {
        return json_response(
            json!({"ok": false, "error": "invalid table name"}),
            StatusCode::BAD_REQUEST,
        );
    }

    let limit = params.limit.unwrap_or(25).clamp(1, 500);
    let database = &state.clickhouse.config().database;

    #[derive(serde::Deserialize)]
    struct SchemaRow {
        name: String,
        #[serde(rename = "type")]
        type_name: String,
        default_expression: Option<String>,
    }

    let schema = match state
        .clickhouse
        .query_rows::<SchemaRow>(&format!(
            "SELECT name, type, default_expression FROM system.columns WHERE database = '{}' AND table = '{}' ORDER BY position",
            escape_literal(database),
            escape_literal(&table)
        ), None)
        .await
    {
        Ok(value) => value,
        Err(error) => {
            return json_response(
                json!({
                    "ok": false,
                    "error": format!("unable to read schema for table {table}: {error}"),
                }),
                StatusCode::SERVICE_UNAVAILABLE,
            );
        }
    };

    let preview_columns: Vec<String> = schema.iter().map(|entry| entry.name.clone()).collect();
    let rows_query = table_preview_rows_query(database, &table, &preview_columns, limit);

    let rows = match state
        .clickhouse
        .query_rows::<Value>(&rows_query, None)
        .await
    {
        Ok(value) => value,
        Err(error) => {
            return json_response(
                json!({
                    "ok": false,
                    "error": format!("unable to read table {table}: {error}"),
                }),
                StatusCode::SERVICE_UNAVAILABLE,
            );
        }
    };

    let schema_payload: Vec<Value> = schema
        .into_iter()
        .map(|entry| {
            json!({
                "name": entry.name,
                "type": entry.type_name,
                "default_expression": entry.default_expression.unwrap_or_default(),
            })
        })
        .collect();

    json_response(
        json!({
            "ok": true,
            "table": table,
            "limit": limit,
            "schema": schema_payload,
            "rows": rows,
        }),
        StatusCode::OK,
    )
}

async fn static_fallback(State(state): State<AppState>, uri: Uri) -> Response {
    let requested = uri.path();
    if requested.contains("..") {
        return json_response(
            json!({"ok": false, "error": "forbidden"}),
            StatusCode::FORBIDDEN,
        );
    }

    let file_path = if requested == "/" || requested.is_empty() {
        state.static_dir.join("index.html")
    } else {
        let mut target = state.static_dir.join(requested.trim_start_matches('/'));
        if target.is_dir() {
            target.push("index.html");
        }
        target
    };

    let canonical_root = match fs::canonicalize(&state.static_dir).await {
        Ok(path) => path,
        Err(error) => {
            return json_response(
                json!({"ok": false, "error": format!("static directory unavailable: {error}")}),
                StatusCode::INTERNAL_SERVER_ERROR,
            );
        }
    };

    let canonical_file = match fs::canonicalize(&file_path).await {
        Ok(path) => path,
        Err(_) => {
            return json_response(
                json!({"ok": false, "error": "not found"}),
                StatusCode::NOT_FOUND,
            );
        }
    };

    if !canonical_file.starts_with(&canonical_root) {
        return json_response(
            json!({"ok": false, "error": "forbidden"}),
            StatusCode::FORBIDDEN,
        );
    }

    let bytes = match fs::read(&canonical_file).await {
        Ok(value) => value,
        Err(error) => {
            return json_response(
                json!({"ok": false, "error": format!("failed to read file: {error}")}),
                StatusCode::INTERNAL_SERVER_ERROR,
            );
        }
    };

    let content_type = mime_guess::from_path(&canonical_file)
        .first_or_octet_stream()
        .essence_str()
        .to_string();

    let mut response = Response::new(Body::from(bytes));
    *response.status_mut() = StatusCode::OK;
    response.headers_mut().insert(
        header::CONTENT_TYPE,
        HeaderValue::from_str(&content_type)
            .unwrap_or_else(|_| HeaderValue::from_static("application/octet-stream")),
    );
    response
}

fn is_safe_identifier(value: &str) -> bool {
    let mut chars = value.chars();
    match chars.next() {
        Some(first) if first == '_' || first.is_ascii_alphabetic() => {}
        _ => return false,
    }

    chars.all(|ch| ch == '_' || ch.is_ascii_alphanumeric())
}

fn escape_literal(value: &str) -> String {
    value.replace('\'', "''")
}

fn escape_identifier(value: &str) -> String {
    format!("`{}`", value.replace('`', "``"))
}

fn table_preview_rows_query(
    database: &str,
    table: &str,
    column_names: &[String],
    limit: u32,
) -> String {
    let database = escape_identifier(database);
    let table = escape_identifier(table);

    if column_names.is_empty() {
        return format!("SELECT * FROM {database}.{table} LIMIT {limit}");
    }

    let order_by = column_names
        .iter()
        .map(|name| escape_identifier(name))
        .collect::<Vec<_>>()
        .join(", ");

    format!("SELECT * FROM {database}.{table} ORDER BY {order_by} LIMIT {limit}")
}

fn value_to_i64(value: &Value) -> Option<i64> {
    if let Some(n) = value.as_i64() {
        return Some(n);
    }
    if let Some(n) = value.as_u64() {
        return i64::try_from(n).ok();
    }
    value.as_str().and_then(|raw| raw.parse::<i64>().ok())
}

#[cfg(test)]
mod tests {
    use super::*;
    use anyhow::anyhow;
    use std::fs;
    use std::path::PathBuf;
    use std::time::{SystemTime, UNIX_EPOCH};

    fn temp_path(suffix: &str) -> PathBuf {
        let stamp = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .expect("clock")
            .as_nanos();
        std::env::temp_dir().join(format!(
            "moraine-monitor-core-{suffix}-{}-{stamp}",
            std::process::id()
        ))
    }

    fn session_event(
        event_kind: &str,
        actor_kind: &str,
        payload_type: &str,
        at: i64,
    ) -> SessionEventRow {
        SessionEventRow {
            session_id: "session-1".to_string(),
            event_ts_ms: at,
            event_kind: event_kind.to_string(),
            actor_kind: actor_kind.to_string(),
            payload_type: payload_type.to_string(),
            tool_name: String::new(),
            tool_call_id: String::new(),
            tool_error: 0,
            latency_ms: 0,
            model: String::new(),
            endpoint_kind: String::new(),
            input_tokens: 0,
            output_tokens: 0,
            cache_read_tokens: 0,
            cache_write_tokens: 0,
            token_usage_buckets: HashMap::new(),
            token_usage_native_units: HashMap::new(),
            text_preview: String::new(),
            text_content: String::new(),
            tool_args_json: String::new(),
        }
    }

    fn compact_sql(query: &str) -> String {
        query.split_whitespace().collect::<Vec<_>>().join(" ")
    }

    #[test]
    fn identifier_safety_helper() {
        assert!(is_safe_identifier("events"));
        assert!(is_safe_identifier("_tmp_1"));
        assert!(!is_safe_identifier("1events"));
        assert!(!is_safe_identifier("events;drop"));
    }

    #[test]
    fn build_turns_emits_assistant_latency_duration() {
        let mut user = session_event("message", "user", "text", 1_000);
        user.text_content = "run the tool".to_string();
        let mut assistant = session_event("message", "assistant", "text", 5_500);
        assistant.text_content = "done".to_string();
        assistant.output_tokens = 42;
        assistant.latency_ms = 4_500;

        let turns = build_turns(&[user, assistant]);
        let steps = turns[0]["steps"].as_array().expect("steps");

        assert_eq!(steps[1]["durationMs"], json!(4_500));
        assert_eq!(steps[1]["tokens"], json!(42));
    }

    #[test]
    fn build_turns_omits_zero_assistant_latency_duration() {
        let mut user = session_event("message", "user", "text", 1_000);
        user.text_content = "fresh prompt".to_string();
        let mut assistant = session_event("message", "assistant", "text", 1_600);
        assistant.text_content = "first reply".to_string();

        let turns = build_turns(&[user, assistant]);
        let steps = turns[0]["steps"].as_array().expect("steps");

        assert!(steps[1].get("durationMs").is_none());
    }

    #[test]
    fn escaping_helpers() {
        assert_eq!(escape_literal("a'b"), "a''b");
        assert_eq!(escape_identifier("ev`ents"), "`ev``ents`");
    }

    #[test]
    fn events_table_ref_escapes_database_identifier() {
        assert_eq!(events_table_ref("mor`aine"), "`mor``aine`.`events`");
    }

    #[test]
    fn session_summary_query_dedups_by_latest_event_uid_with_recency() {
        let query = compact_sql(&session_summary_query("`moraine`.`events`", 50, 86_400));

        assert!(query.contains(
            "FROM (SELECT * FROM `moraine`.`events` WHERE length(trim(BOTH ' ' FROM session_id)) > 0 AND event_ts >= now() - INTERVAL 86400 SECOND ORDER BY event_uid ASC, event_version DESC, ingested_at DESC LIMIT 1 BY event_uid) AS deduped_events"
        ));
        assert!(query.contains("GROUP BY session_id ORDER BY ended_at_ms DESC LIMIT 50"));
    }

    #[test]
    fn session_summary_query_since_all_omits_recency_filter() {
        let query = compact_sql(&session_summary_query("`moraine`.`events`", 25, 0));

        assert!(query.contains(
            "FROM (SELECT * FROM `moraine`.`events` WHERE length(trim(BOTH ' ' FROM session_id)) > 0 ORDER BY event_uid ASC, event_version DESC, ingested_at DESC LIMIT 1 BY event_uid) AS deduped_events"
        ));
        assert!(!query.contains("INTERVAL"));
    }

    #[test]
    fn session_events_query_dedups_and_escapes_session_ids() {
        let query = compact_sql(&session_events_query(
            "`moraine`.`events`",
            &["session-1".to_string(), "quote's".to_string()],
        ));

        assert!(query.contains(
            "FROM (SELECT * FROM `moraine`.`events` WHERE session_id IN ('session-1','quote''s') ORDER BY event_uid ASC, event_version DESC, ingested_at DESC LIMIT 1 BY event_uid) AS deduped_events"
        ));
        assert!(query.contains(
            "ORDER BY session_id, event_ts, source_file, source_generation, source_offset, source_line_no, event_uid"
        ));
    }

    #[test]
    fn table_preview_query_orders_by_schema_columns() {
        let query = table_preview_rows_query(
            "analytics",
            "events",
            &[
                "event_ts".to_string(),
                "session_id".to_string(),
                "event_id".to_string(),
            ],
            25,
        );
        assert_eq!(
            query,
            "SELECT * FROM `analytics`.`events` ORDER BY `event_ts`, `session_id`, `event_id` LIMIT 25"
        );
    }

    #[test]
    fn table_preview_query_escapes_identifiers() {
        let query = table_preview_rows_query("ana`lytics", "ev`ents", &["co`l".to_string()], 10);
        assert_eq!(
            query,
            "SELECT * FROM `ana``lytics`.`ev``ents` ORDER BY `co``l` LIMIT 10"
        );
    }

    #[test]
    fn table_preview_query_handles_empty_schema() {
        let columns: Vec<String> = Vec::new();
        let query = table_preview_rows_query("analytics", "events", &columns, 5);
        assert_eq!(query, "SELECT * FROM `analytics`.`events` LIMIT 5");
    }

    #[test]
    fn analytics_window_query_guards_empty_recent_window() {
        let query = analytics_window_query(
            "`moraine`.`events`",
            "event_ts >= now() - INTERVAL 86400 SECOND",
            86_400,
        );

        assert!(query.contains("count() = 0"));
        assert!(query.contains("toUnixTimestamp64Milli(max(event_ts))"));
        assert!(query.contains("greatest(anchor_unix - toInt64(86400), toInt64(0))"));
        assert!(!query.contains("max(event_ts) - INTERVAL"));
    }

    #[test]
    fn clickhouse_ping_payload_marks_healthy_when_ping_succeeds() {
        let payload = clickhouse_ping_payload("24.8".to_string(), Ok(()), 3.5);
        assert_eq!(payload["healthy"], json!(true));
        assert_eq!(payload["version"], json!("24.8"));
        assert_eq!(payload["ping_ms"], json!(3.5));
        assert_eq!(payload["error"], Value::Null);
    }

    #[test]
    fn clickhouse_ping_payload_marks_unhealthy_when_ping_fails() {
        let payload =
            clickhouse_ping_payload("24.8".to_string(), Err(anyhow!("ping failed")), 8.25);
        assert_eq!(payload["healthy"], json!(false));
        assert_eq!(payload["version"], json!("24.8"));
        assert_eq!(payload["ping_ms"], json!(8.25));
        assert_eq!(payload["error"], json!("ping failed"));
    }

    #[test]
    fn validate_static_dir_accepts_built_directory() {
        let root = temp_path("static-valid");
        fs::create_dir_all(&root).expect("create root");
        fs::write(root.join("index.html"), "<!doctype html>").expect("write index");

        validate_static_dir(&root).expect("valid static dir");

        let _ = fs::remove_dir_all(root);
    }

    #[test]
    fn validate_static_dir_rejects_missing_directory() {
        let missing = temp_path("static-missing");
        let err = validate_static_dir(&missing).expect_err("missing static dir should fail");
        assert!(err.to_string().contains("is unavailable"));
    }

    #[test]
    fn validate_static_dir_rejects_non_directory() {
        let root = temp_path("static-file");
        fs::create_dir_all(&root).expect("create root");
        let path = root.join("dist");
        fs::write(&path, "not a dir").expect("write file");

        let err = validate_static_dir(&path).expect_err("file should fail");
        assert!(err.to_string().contains("is not a directory"));

        let _ = fs::remove_dir_all(root);
    }

    #[test]
    fn validate_static_dir_requires_index_html() {
        let root = temp_path("static-no-index");
        fs::create_dir_all(&root).expect("create root");

        let err = validate_static_dir(&root).expect_err("missing index should fail");
        assert!(err.to_string().contains("does not contain `index.html`"));

        let _ = fs::remove_dir_all(root);
    }
}
