path: root/server/src/runs.rs

use axum::extract::{Path, Query};
use axum::http::StatusCode;
use axum::Json;
use serde_json::{json, Value};

use crate::auth::{audit, AuthUser};
use crate::config::MAX_CODE_LEN;
use crate::error::{AppError, AppResult};
use crate::util::new_id;
use crate::util::now_iso;

pub use crate::state::Cx;

const RUN_COLS: &str = "id,project_id,version_id,config,dataset_id,status,error,manifest_hash,created_at,result,started_at,finished_at";

fn row_run(r: &rusqlite::Row) -> rusqlite::Result<Value> {
    let config: String = r.get(3)?;
    let result: Option<String> = r.get(9)?;
    Ok(json!({
        "id": r.get::<_, String>(0)?,
        "project_id": r.get::<_, String>(1)?,
        "version_id": r.get::<_, String>(2)?,
        "config": serde_json::from_str::<Value>(&config).unwrap_or(Value::Null),
        "dataset_id": r.get::<_, String>(4)?,
        "status": r.get::<_, String>(5)?,
        "error": r.get::<_, Option<String>>(6)?,
        "manifest_hash": r.get::<_, Option<String>>(7)?,
        "data_manifest_hash": r.get::<_, Option<String>>(7)?,
        "created_at": r.get::<_, String>(8)?,
        "result": result.and_then(|s| serde_json::from_str::<Value>(&s).ok()).unwrap_or(Value::Null),
        "started_at": r.get::<_, Option<String>>(10)?,
        "finished_at": r.get::<_, Option<String>>(11)?,
    }))
}

fn select_run(db: &mut rusqlite::Connection, run_id: &str) -> AppResult<Value> {
    let cols = RUN_COLS;
    db.query_row(format!("SELECT {cols} FROM runs WHERE id=?1").as_str(), [run_id], row_run)
        .map_err(|_| AppError::not_found("run not found"))
}

async fn assert_run_owner(cx: &Cx, user_id: &str, run_id: &str) -> AppResult<()> {
    let owner: Option<String> = cx.with_db(|db| {
        db.query_row("SELECT user_id FROM runs WHERE id=?1", [run_id], |r| r.get(0)).ok()
    }).await;
    match owner {
        Some(u) if u == user_id => Ok(()),
        _ => Err(AppError::not_found("run not found")),
    }
}

pub async fn get(cx: Cx, auth: AuthUser, Path(id): Path<String>) -> AppResult<Json<Value>> {
    assert_run_owner(&cx, &auth.id, &id).await?;
    let v = cx.with_db(|db| select_run(db, &id)).await?;
    Ok(Json(v))
}

pub async fn list(cx: Cx, auth: AuthUser, q: Query<std::collections::HashMap<String, String>>) -> AppResult<Json<Value>> {
    let project = q.get("project_id").cloned();
    let cols = RUN_COLS;
    let items: Vec<Value> = cx.with_db(|db| -> AppResult<Vec<Value>> {
        let owned = project.as_deref();
        let sql = if owned.is_some() {
            format!("SELECT {cols} FROM runs WHERE user_id=?1 AND project_id=?2 ORDER BY created_at DESC LIMIT 200")
        } else {
            format!("SELECT {cols} FROM runs WHERE user_id=?1 ORDER BY created_at DESC LIMIT 200")
        };
        let mut st = db.prepare(&sql)?;
        let mut rows = if let Some(p) = owned {
            st.query(rusqlite::params![auth.id, p])?
        } else {
            st.query([auth.id.clone()])?
        };
        let mut out = Vec::new();
        while let Some(r) = rows.next()? { out.push(row_run(r)?); }
        Ok(out)
    }).await?;
    Ok(Json(json!({ "items": items })))
}

fn build_config(j: &Value) -> AppResult<Value> {
    let g = |k: &str, d: f64| j.get(k).and_then(|v| v.as_f64()).unwrap_or(d);
    let capital = j.get("capital").and_then(|v| v.as_f64());
    let config = json!({
        "capital": capital.unwrap_or(1_000_000.0),
        "commission": g("commission", 0.0003),
        "slippage": g("slippage", 0.001),
        "benchmark_symbol": j.get("benchmark_symbol").cloned().unwrap_or(Value::Null),
        "parameters": j.get("parameters").cloned().unwrap_or(json!({})),
        "seed": j.get("seed").cloned().unwrap_or(Value::Null),
    });
    let cap = config.get("capital").and_then(|v| v.as_f64()).unwrap();
    if !cap.is_finite() || !(1.0..=1e12).contains(&cap) {
        return Err(AppError::bad("validation", "capital must be between 1 and 1e12"));
    }
    for k in ["commission", "slippage"] {
        let v = config.get(k).and_then(|v| v.as_f64()).unwrap();
        if !v.is_finite() || !(0.0..=0.05).contains(&v) {
            return Err(AppError::bad("validation", format!("{k} must be between 0 and 0.05")));
        }
    }
    if config.get("parameters").and_then(|p| p.as_object()).map(|o| !o.is_empty()).unwrap_or(false) {
        if serde_json::to_string(config.get("parameters").unwrap()).unwrap_or_default().len() > MAX_CODE_LEN {
            return Err(AppError::bad("validation", "parameters too large"));
        }
    }
    Ok(config)
}

/// Insert a queued run inside one transaction with an atomic per-user daily quota check.
async fn insert_run(cx: &Cx, auth: &AuthUser, project_id: &str, dataset_id: &str, version_id: &str, config: &Value, manifest_hash: &Option<String>) -> AppResult<String> {
    let uid = auth.id.clone();
    let cfg_s = config.to_string();
    let limit = auth.daily_run_limit;
    let today: String = chrono::Utc::now().format("%Y-%m-%d").to_string();
    let pattern = format!("{today}%");
    let id = new_id();
    let now = now_iso();
    cx.with_db(|db| -> AppResult<String> {
        db.execute("BEGIN IMMEDIATE", []).ok();
        let used: i64 = db.query_row(
            "SELECT COUNT(*) FROM runs WHERE user_id=?1 AND created_at LIKE ?2",
            rusqlite::params![&uid, &pattern], |r| r.get(0)).unwrap_or(0);
        if used >= limit {
            db.execute("ROLLBACK", []).ok();
            return Err(AppError::new(axum::http::StatusCode::TOO_MANY_REQUESTS, "run_limit", "daily run quota reached"));
        }
        db.execute("INSERT INTO runs (id,user_id,project_id,version_id,dataset_id,status,config,manifest_hash,created_at) VALUES (?1,?2,?3,?4,?5,'queued',?6,?7,?8)",
            rusqlite::params![&id, &uid, project_id, version_id, dataset_id, &cfg_s, manifest_hash, &now])?;
        db.execute("COMMIT", []).map_err(AppError::from)?;
        Ok(id)
    }).await
}

pub async fn enqueue(cx: Cx, auth: AuthUser, body: Option<Json<serde_json::Value>>) -> AppResult<(StatusCode, Json<Value>)> {
    let Json(j) = body.ok_or_else(|| AppError::bad("invalid_body", "JSON body required"))?;
    let project_id = j.get("project_id").and_then(|v| v.as_str()).ok_or_else(|| AppError::bad("validation", "project_id required"))?.to_string();
    let dataset_id = j.get("dataset_id").and_then(|v| v.as_str()).ok_or_else(|| AppError::bad("validation", "dataset_id required"))?.to_string();

    // dataset must exist, belong to the caller and be ready
    let (ready, manifest_hash): (bool, Option<String>) = cx.with_db(|db| {
        db.query_row("SELECT status='ready', manifest_hash FROM datasets WHERE id=?1 AND user_id=?2",
            rusqlite::params![&dataset_id, &auth.id], |r| Ok((r.get::<_, i64>(0)? != 0, r.get(1)?)))
            .map_err(|_| AppError::not_found("dataset not found"))
    }).await?;
    if !ready { return Err(AppError::conflict("dataset_not_ready", "dataset pending or failed; cannot run")); }

    // snapshot + pin draft
    let draft: String = cx.with_db(|db| {
        db.query_row("SELECT draft_code FROM projects WHERE id=?1 AND user_id=?2", rusqlite::params![&project_id, &auth.id], |r| r.get(0))
            .map_err(|_| AppError::not_found("project not found"))
    }).await?;
    if draft.trim().is_empty() {
        return Err(AppError::bad("validation", "draft is empty; cannot run empty strategy"));
    }

    let config = build_config(&j)?;
    let version_id = push_run_version(&cx, &auth.id, &project_id, &draft).await?;

    // warnings need explicit acknowledgement
    let warnings: Value = cx.with_db(|db| {
        let m: String = db.query_row("SELECT manifest FROM datasets WHERE id=?1", [&dataset_id], |r| r.get(0)).unwrap_or_default();
        let mj: Value = serde_json::from_str(&m).unwrap_or(Value::Null);
        Ok::<_, AppError>(mj.get("warnings").cloned().unwrap_or(json!([])))
    }).await?;
    let mut warnings = warnings;
    if let Some(objs) = warnings.as_array_mut() {
        // explicit unsupported-restriction surfacing for index feeds
        let has_index: bool = cx.with_db(|db| {
            let m: String = db.query_row("SELECT manifest FROM datasets WHERE id=?1", [&dataset_id], |r| r.get(0)).unwrap_or_default();
            let mj: Value = serde_json::from_str(&m).unwrap_or(Value::Null);
            Ok::<_, AppError>(mj.get("objects").and_then(|o| o.as_array()).map(|a| {
                a.iter().any(|o| o.get("instrument").and_then(|i| i.get("asset_type")).and_then(|t| t.as_str()) == Some("index"))
            }).unwrap_or(false))
        }).await?;
        if has_index && !objs.iter().any(|w| w.as_str().map(|s| s.contains("index feeds are nontradable")).unwrap_or(false)) {
            objs.push(json!("index feeds are nontradable research proxies; direct index orders are rejected by the engine"));
        }
    }
    let has_warns = warnings.as_array().map(|a| !a.is_empty()).unwrap_or(false);
    if has_warns && !j.get("acknowledge_warnings").and_then(|v| v.as_bool()).unwrap_or(false) {
        return Err(AppError::conflict("warnings_unacknowledged", "dataset has warnings; acknowledge to run")
            .with_details(json!({"warnings": warnings})));
    }

    let run_id = insert_run(&cx, &auth, &project_id, &dataset_id, &version_id, &config, &manifest_hash).await?;
    audit(&cx, Some(&auth.id), "run_create", &run_id, "ok").await;
    let v = cx.with_db(|db| select_run(db, &run_id)).await?;
    Ok((StatusCode::ACCEPTED, Json(v)))
}

/// Snapshot the draft as a pinned run version, reusing an identical hash version.
async fn push_run_version(cx: &Cx, _user_id: &str, project_id: &str, code: &str) -> AppResult<String> {
    let hash = crate::util::sha256_hex(code.as_bytes());
    let code = code.to_string();
    cx.with_db(|db| -> AppResult<String> {
        let vid: Option<String> = db.query_row("SELECT id FROM project_versions WHERE project_id=?1 AND hash=?2",
            rusqlite::params![project_id, &hash], |r| r.get(0)).ok();
        if let Some(v) = vid { return Ok(v); }
        let id = new_id();
        db.execute("INSERT INTO project_versions (id,project_id,code,hash,message,source,created_at) VALUES (?1,?2,?3,?4,'run snapshot','run',?5)",
            rusqlite::params![&id, project_id, &code, &hash, now_iso()])?;
        Ok(id)
    }).await
}

pub async fn cancel(cx: Cx, auth: AuthUser, Path(id): Path<String>) -> AppResult<Json<Value>> {
    assert_run_owner(&cx, &auth.id, &id).await?;
    let status: Option<String> = cx.with_db(|db| {
        db.query_row("SELECT status FROM runs WHERE id=?1", [&id], |r| r.get(0)).ok()
    }).await;
    if matches!(status.as_deref(), Some("queued") | Some("running")) {
        let res = crate::jobs::signal_cancel(&cx, &id).await;
        audit(&cx, Some(&auth.id), "run_cancel", &id, if res { "ok" } else { "not_running" }).await;
    }
    let v = cx.with_db(|db| select_run(db, &id)).await?;
    Ok(Json(v))
}

/// Fresh run pinned to the original code+config+dataset; never refetches data.
pub async fn rerun(cx: Cx, auth: AuthUser, Path(id): Path<String>, body: Option<Json<serde_json::Value>>) -> AppResult<(StatusCode, Json<Value>)> {
    assert_run_owner(&cx, &auth.id, &id).await?;
    let j = body.map(|b| b.0).unwrap_or(json!({}));
    if j.get("use_original_data").and_then(|v| v.as_bool()) != Some(true) {
        return Err(AppError::bad("validation", "use_original_data must be true; rerun is pinned to original data"));
    }
    let (project_id, version_id, dataset_id, config, manifest_hash): (String, String, String, String, Option<String>) = cx.with_db(|db| {
        db.query_row("SELECT project_id,version_id,dataset_id,config,manifest_hash FROM runs WHERE id=?1", [&id], |r|
            Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?, r.get::<_, String>(2)?, r.get::<_, String>(3)?, r.get::<_, Option<String>>(4)?)))
            .map_err(|_| AppError::not_found("run not found"))
    }).await?;
    // dataset must still be ready and owned
    let ready: bool = cx.with_db(|db| {
        db.query_row("SELECT status='ready' FROM datasets WHERE id=?1 AND user_id=?2",
            rusqlite::params![&dataset_id, &auth.id], |r| r.get(0))
            .map_err(|_| AppError::not_found("dataset not found"))
    }).await?;
    if !ready { return Err(AppError::conflict("dataset_not_ready", "original dataset is missing or not ready")); }
    let cfg_val: Value = serde_json::from_str(&config).unwrap_or(json!({}));
    let new_id_r = insert_run(&cx, &auth, &project_id, &dataset_id, &version_id, &cfg_val, &manifest_hash).await?;
    audit(&cx, Some(&auth.id), "run_rerun", &new_id_r, "ok").await;
    let v = cx.with_db(|db| select_run(db, &new_id_r)).await?;
    Ok((StatusCode::ACCEPTED, Json(v)))
}

/// Mark running runs failed after a server restart (queued stays resumable).
pub async fn cleanup_interrupted(cx: &Cx) {
    cx.with_db(|db| {
        db.execute("UPDATE runs SET status='failed', error='interrupted by server restart', finished_at=?1 WHERE status='running'",
            rusqlite::params![now_iso()]).ok();
    }).await;
}


#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn config_bounds_enforced() {
        let ok = json!({"capital": 100000, "commission": 0.0003, "slippage": 0.001});
        assert!(build_config(&ok).is_ok());
        let bad = json!({"capital": 0.0});
        assert_eq!(build_config(&bad).unwrap_err().code, "validation");
        let bad = json!({"capital": 5000, "commission": 0.5});
        assert_eq!(build_config(&bad).unwrap_err().code, "validation");
        let big = json!({"capital": 100000, "parameters": {"huge": "x".repeat(MAX_CODE_LEN + 1)}});
        assert_eq!(build_config(&big).unwrap_err().code, "validation");
        let defaults = build_config(&json!({})).unwrap();
        assert_eq!(defaults["commission"], json!(0.0003), "defaults are visible to users");
    }

    #[test]
    fn nonfinite_sanitized() {
        let mut v = json!({"metrics": {"sharpe": f64::NAN, "total_return": 1.0}, "equity": [1.0, f64::INFINITY]});
        crate::jobs::sanitize_nonfinite(&mut v);
        let s = serde_json::to_string(&v).unwrap();
        assert!(!s.contains("NaN") && !s.contains("Infinity"), "results must never carry NaN/Inf");
        assert_eq!(v["metrics"]["sharpe"], Value::Null);
    }
}