use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Duration;

use tokio::sync::Mutex;
use std::collections::HashSet;
use serde_json::{json, Value};

use crate::auth::audit;
use crate::error::AppError;
use crate::state::AppState;
use crate::store::ObjectStore;
use crate::util::{canonical_json, new_id, now_iso, sha256_hex};

pub struct Signals;

impl Signals {
    pub fn new() -> Self { Signals }
}

pub async fn main_loop(cx: Arc<AppState>, _signals: Signals) {
    loop {
        if let Err(e) = tick(&cx).await { tracing::error!("job loop: {e}"); }
        tokio::time::sleep(Duration::from_millis(700)).await;
    }
}

// ---- shared in-flight fetch serialization for the exact request cache key ----

fn inflight() -> &'static Mutex<HashSet<String>> {
    static SET: std::sync::OnceLock<Mutex<HashSet<String>>> = std::sync::OnceLock::new();
    SET.get_or_init(|| Mutex::new(HashSet::new()))
}

pub fn cache_key_for_request(req: &Value) -> String {
    sha256_hex(canonical_json(req).as_bytes())
}

async fn release_key(key: &str) {
    inflight().lock().await.remove(key);
}

// ---- dispatcher ----

async fn tick(cx: &Arc<AppState>) -> Result<(), AppError> {
    // pending datasets -> bounded fetch concurrency
    let n_running: i64 = cx.with_db(|db| {
        db.query_row("SELECT COUNT(*) FROM datasets WHERE status='running'", [], |r| r.get(0))
    }).await.unwrap_or(0);
    if n_running < cx.cfg.fetch_concurrency as i64 {
        let next: Option<(String, String)> = cx.with_db(|db| {
            db.query_row("SELECT id,request FROM datasets WHERE status='pending' ORDER BY created_at LIMIT 1", [],
                |r| Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?))).ok()
        }).await;
        if let Some((did, request)) = next {
            cx.with_db(|db| {
                db.execute("UPDATE datasets SET status='running', updated_at=?1 WHERE id=?2 AND status='pending'",
                    rusqlite::params![now_iso(), &did]).ok();
            }).await;
            let cx2 = cx.clone();
            tokio::spawn(async move {
                if let Err(e) = spawn_fetch_job(&cx2, &did, &request).await {
                    tracing::error!("fetch {did}: {e}");
                    cx2.with_db(|db| {
                        db.execute("UPDATE datasets SET status='failed', error=?1, updated_at=?2 WHERE id=?3 AND status IN ('pending','running')",
                            rusqlite::params![e.message, now_iso(), &did]).ok();
                    }).await;
                }
            });
        }
    }

    // queued runs -> single active backtest worker (POC)
    let n_running_runs: i64 = cx.with_db(|db| {
        db.query_row("SELECT COUNT(*) FROM runs WHERE status='running'", [], |r| r.get(0))
    }).await.unwrap_or(0);
    if n_running_runs >= cx.cfg.run_concurrency as i64 { return Ok(()); }
    let next: Option<(String, String)> = cx.with_db(|db| {
        db.query_row("SELECT id,config FROM runs WHERE status='queued' ORDER BY created_at LIMIT 1", [],
            |r| Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?))).ok()
    }).await;
    let Some((run_id, config)) = next else { return Ok(()); };
    let container_name = format!("sl-run-{}", new_id());
    // Atomic claim: queued -> running WITH container_id set in the same transaction.
    // A run never becomes 'running' without a container identity, so the claim
    // itself proves the worker task was handed off.
    let claimed = cx.with_db(|db| -> Result<bool, AppError> {
        move_claim(&mut *db, &run_id, &container_name)
    }).await?;
    if !claimed { return Ok(()); }
    let cx2 = cx.clone();
    tokio::spawn(async move {
        let user_id: String = cx2.with_db(|db| {
            db.query_row("SELECT user_id FROM runs WHERE id=?1", [&run_id], |r| r.get(0)).unwrap_or_default()
        }).await;
        let res = run_backtest(&cx2, &run_id, &config, &container_name).await;
        let _ = finalize_run(&cx2, &run_id, &user_id, res).await;
    });
    Ok(())
}

/// Recheck account/quota/dataset durability, then claim queued -> running with
/// a container id atomically inside one transaction.
/// Returns true only if this caller actually claimed the run (and must spawn the worker).
fn move_claim(db: &mut rusqlite::Connection, run_id: &str, container_id: &str) -> Result<bool, AppError> {
    db.execute("BEGIN IMMEDIATE", []).ok();
    match db.query_row(
        "SELECT u.active, r.user_id FROM runs r JOIN users u ON u.id=r.user_id WHERE r.id=?1",
        [run_id], |r| Ok((r.get::<_, i64>(0)?, r.get::<_, String>(1)?))) {
        Ok((active, _uid)) if active == 1 => {},
        _ => {
            db.execute("UPDATE runs SET status='failed', error='account disabled', finished_at=?1 WHERE id=?2 AND status='queued'",
                rusqlite::params![now_iso(), run_id]).ok();
            db.execute("COMMIT", []).ok();
            return Ok(false);
        }
    }
    let uid = db.query_row("SELECT user_id FROM runs WHERE id=?1", [run_id], |r| r.get::<_, String>(0)).unwrap_or_default();
    let ready_ok: Option<i64> = db.query_row(
        "SELECT d.status='ready' AS ok FROM datasets d JOIN runs r ON r.dataset_id=d.id AND r.user_id=d.user_id WHERE r.id=?1",
        [run_id], |r| r.get::<_, i64>(0)).ok();
    if ready_ok != Some(1) {
        db.execute("UPDATE runs SET status='failed', error='dataset no longer ready at dequeue', finished_at=?1 WHERE id=?2 AND status='queued'",
            rusqlite::params![now_iso(), run_id]).ok();
        db.execute("COMMIT", []).ok();
        return Ok(false);
    }
    let today = format!("{}%", chrono::Utc::now().format("%Y-%m-%d").to_string());
    let used: i64 = db.query_row("SELECT COUNT(*) FROM runs WHERE user_id=?1 AND created_at LIKE ?2",
        rusqlite::params![&uid, &today], |r| r.get(0)).unwrap_or(0);
    let limit: i64 = db.query_row("SELECT daily_run_limit FROM users WHERE id=?1", [&uid], |r| r.get(0)).unwrap_or(10);
    if used > limit {
        db.execute("UPDATE runs SET status='failed', error='run quota exceeded', finished_at=?1 WHERE id=?2 AND status='queued'",
            rusqlite::params![now_iso(), run_id]).ok();
        db.execute("COMMIT", []).ok();
        return Ok(false);
    }
    // Single atomic transition: a claimed run is running WITH its container identity.
    let n = db.execute(
        "UPDATE runs SET status='running', started_at=?1, container_id=?2 WHERE id=?3 AND status='queued'",
        rusqlite::params![now_iso(), container_id, run_id]).unwrap_or(0);
    db.execute("COMMIT", []).ok();
    Ok(n == 1)
}

async fn finalize_run(cx: &Arc<AppState>, run_id: &str, user_id: &str, res: Result<Value, AppError>) {
    cx.with_db(|db| -> Result<(), AppError> {
        match res {
            Ok(result) => {
                let n = db.execute("UPDATE runs SET status='succeeded', result=?1, error=NULL, finished_at=?2 WHERE id=?3 AND status='running'",
                    rusqlite::params![result.to_string(), now_iso(), run_id]).unwrap_or(0);
                if n > 0 {
                    let cxx = cx.clone();
                    let uid = user_id.to_string();
                    let rid = run_id.to_string();
                    tokio::spawn(async move { audit(&axum::extract::State(cxx), Some(&uid), "run_succeeded", &rid, "ok").await });
                }
            }
            Err(e) => {
                // guaranteed non-empty terminal reason, never an empty failure
                let msg = if e.message.trim().is_empty() { "worker failed without error detail".to_string() } else { e.message.clone() };
                db.execute("UPDATE runs SET status='failed', error=?1, finished_at=?2 WHERE id=?3 AND status='running'",
                    rusqlite::params![msg, now_iso(), run_id]).ok();
            }
        }
        Ok(())
    }).await.ok();
}

// ---- dataset fetch ----

async fn try_cache(cx: &Arc<AppState>, dataset_id: &str, key: &str) -> Result<bool, AppError> {
    let manifest: Option<String> = cx.with_db(|db| {
        db.query_row("SELECT manifest FROM fetch_cache WHERE key=?1", [key], |r| r.get(0)).ok()
    }).await;
    let Some(m) = manifest else { return Ok(false); };
    let mv: Value = serde_json::from_str(&m).unwrap_or(Value::Null);
    let hash = mv.get("hash").and_then(|v| v.as_str()).unwrap_or_default().to_string();
    cx.with_db(|db| {
        db.execute("UPDATE datasets SET status='ready', error=NULL, manifest=?1, manifest_hash=?2, cache_hit=1, updated_at=?3 WHERE id=?4 AND status='running'",
            rusqlite::params![&m, hash, now_iso(), dataset_id]).ok();
    }).await;
    Ok(true)
}

/// One dataset fetch: cache recheck or container run + artifact ingest.
pub async fn spawn_fetch_job(cx: &Arc<AppState>, dataset_id: &str, request_s: &str) -> Result<(), AppError> {
    let request: Value = serde_json::from_str(request_s).map_err(|e| AppError::internal(format!("stored request invalid: {e}")))?;
    let key = cache_key_for_request(&request);

    // serialize identical cache keys: second arrives late and rechecks cache
    let deadline = std::time::Duration::from_secs(cx.cfg.fetch_timeout_secs.max(60));
    let started = std::time::Instant::now();
    loop {
        {
            let mut set = inflight().lock().await;
            if set.insert(key.clone()) { break; }

        }
        if try_cache(cx, dataset_id, &key).await? { return Ok(()); }
        if started.elapsed() > deadline {
            return Err(AppError::internal("waiting on identical in-flight fetch timed out"));
        }
        tokio::time::sleep(Duration::from_millis(1500)).await;
    }
    let result = perform_fetch(cx, dataset_id, request, &key).await;
    release_key(&key).await;
    result
}

async fn perform_fetch(cx: &Arc<AppState>, dataset_id: &str, request: Value, key: &str) -> Result<(), AppError> {
    // cached from a prior identical request?
    if try_cache(cx, dataset_id, key).await? { return Ok(()); }

    let store = ObjectStore::new(&cx.cfg.data_dir);
    let work = temp_job_dir(&cx.cfg.data_dir)?;
    let input = work.join("input");
    let output = work.join("output");
    std::fs::create_dir_all(&input).map_err(io_err)?;
    std::fs::create_dir_all(&output).map_err(io_err)?;
    open_writable(&output)?;

    std::fs::write(input.join("request.json"), request.to_string()).map_err(io_err)?;
    let mounts = vec![
        (input.display().to_string(), "/input".to_string(), true),
        (output.display().to_string(), "/output".to_string(), false),
    ];
    let args = vec![
        "python".into(), "-m".into(), "worker.main".into(), "fetch".into(),
        "--request".into(), "/input/request.json".into(), "--output".into(), "/output".into(),
    ];
    let name = format!("sl-fetch-{}", new_id());
    let res = crate::worker::run_named(&*cx, true, &mounts, &args, &name, cx.cfg.fetch_timeout_secs).await;
    let out_take = match res {
        Ok(r) if r.ok() => r,
        Ok(r) => {
            let msg = worker_error_message(&output, r.stderr.as_str());
            cleanup(&work);
            return Err(AppError::bad("fetch_failed", msg));
        }
        Err(e) => { cleanup(&work); return Err(e); }
    };

    let build = ingest_fetch_output(&store, &work).await;
    let (manifest, _warnings) = match build {
        Ok(v) => v,
        Err(e) => { cleanup(&work); return Err(e); }
    };
    let _ = out_take;
    let manifest_hash = manifest.get("hash").and_then(|v| v.as_str()).unwrap_or("").to_string();
    if manifest_hash.is_empty() || manifest.get("objects").and_then(|o| o.as_array()).map(|a| a.is_empty()).unwrap_or(true) {
        cleanup(&work);
        return Err(AppError::internal("fetch produced no usable objects"));
    }
    let manifest_s = manifest.to_string();
    let object_count = manifest.get("objects").and_then(|o| o.as_array()).map(|a| a.len()).unwrap_or(0) as i64;
    cx.with_db(|db| -> Result<(), AppError> {
        db.execute("INSERT OR REPLACE INTO fetch_cache (key,manifest_hash,manifest,object_count,fetched_at) VALUES (?1,?2,?3,?4,?5)",
            rusqlite::params![&key, &manifest_hash, &manifest_s, object_count, now_iso()])?;
        db.execute("UPDATE datasets SET status='ready', error=NULL, manifest=?1, manifest_hash=?2, cache_hit=0, updated_at=?3 WHERE id=?4 AND status='running'",
            rusqlite::params![&manifest_s, &manifest_hash, now_iso(), dataset_id])?;
        Ok(())
    }).await?;
    cleanup(&work);
    tracing::info!("fetch {dataset_id} cached under {key} ({object_count} objects)");
    Ok(())
}


// manifest hash must be content identity: independent of user/request ids and fetch timestamp
fn manifest_content_hash(m: &Value) -> String {
    let mut objects = Vec::new();
    for o in m.get("objects").and_then(|v| v.as_array()).unwrap_or(&vec![]).iter() {
        let mut e = o.clone();
        if let Some(map) = e.as_object_mut() {
            map.remove("fetched_at");
            map.remove("path");
        }
        objects.push(e);
    }
    let payload = json!({
        "frequency": m.get("frequency"),
        "adjustment": m.get("adjustment"),
        "schema_version": m.get("schema_version"),
        "normalization_version": m.get("normalization_version"),
        "objects": objects,
    });
    sha256_hex(canonical_json(&payload).as_bytes())
}


async fn ingest_fetch_output(store: &ObjectStore, work: &Path) -> Result<(Value, Vec<String>), AppError> {
    let out_dir = work.join("output");
    let result_path = out_dir.join("result.json");
    let result: Value = serde_json::from_str(&std::fs::read_to_string(&result_path).map_err(|e| AppError::internal(format!("worker result.json unreadable: {e}")))?)
        .map_err(|e| AppError::internal(format!("worker result.json invalid: {e}")))?;
    if result.get("status").and_then(|v| v.as_str()) != Some("ready") {
        return Err(AppError::bad("fetch_failed", format!("worker reported status: {}", result.get("status").and_then(|v| v.as_str()).unwrap_or("missing"))));
    }
    let m = result.get("manifest").cloned().unwrap_or(Value::Null);
    if !m.get("objects").and_then(|o| o.as_array()).map(|a| !a.is_empty()).unwrap_or(false) {
        return Err(AppError::bad("fetch_failed", "worker manifest has no objects"));
    }
    // hash+ingest every worker artifact once (raw JSON and normalized CSVs)
    let files: Vec<crate::store::StoredObject> = store.ingest_directory(&out_dir).map_err(io_err)?;
    let find_hash = |h: &str| files.iter().find(|f| f.hash == h);
    let find_rel = |rel: &str| files.iter().find(|f| f.stored_path == rel || f.mount_name == rel);

    let all_warnings: Vec<String> = result.get("warnings").and_then(|v| v.as_array()).map(|a| {
        a.iter().filter_map(|w| w.as_str().map(String::from)).collect()
    }).unwrap_or_default();
    let mut new_objects = Vec::new();
    for obj in m.get("objects").and_then(|v| v.as_array()).cloned().unwrap_or_default() {
        let mut entry = obj.clone();
        // normalized: rewrite the internal path into immutable object storage
        let rel_norm = entry.get("path").and_then(|v| v.as_str()).unwrap_or_default().to_string();
        if rel_norm.contains("..") {
            return Err(AppError::internal("worker output path rejected"));
        }
        let norm_obj = find_rel(&rel_norm).ok_or_else(|| AppError::internal("normalized object missing from worker output"))?;
        let rel_stored = norm_obj.stored_path.clone();
        if let Some(map) = entry.as_object_mut() {
            map.insert("path".into(), json!(format!("objects/{rel_stored}")));
        }
        // raw: locate by content hash of the raw object (worker writes immutable raw JSON first)
        let raw_hash = entry.get("raw_object_hash").and_then(|v| v.as_str()).unwrap_or_default().to_string();
        if !raw_hash.is_empty() {
            if find_hash(&raw_hash).is_none() {
                return Err(AppError::internal("raw object referenced by manifest is missing"));
            }
        }
        new_objects.push(entry);
    }

    let mut manifest = m.clone();
    {
        let map = manifest.as_object_mut().unwrap();
        map.insert("objects".into(), Value::Array(new_objects));
        map.insert("warnings".into(), json!(all_warnings.clone()));
        map.insert("immutable".into(), json!(true));
    }
    // stable content identity independent of request/user ids and fetch timestamp
    {
        let h = manifest_content_hash(&manifest);
        if let Some(map) = manifest.as_object_mut() {
            map.insert("hash".into(), json!(h));
            // Per-instrument coverage rows aligned to the client CoverageEntry shape:
            // {instrument, market, asset_type, requested_start, requested_end, actual_start, actual_end, row_count, warnings}
            let coverage: Vec<Value> = map.get("objects").and_then(|o| o.as_array()).map(|objs| {
                objs.iter().map(|o| {
                    let inst = o.get("instrument").cloned().unwrap_or(Value::Null);
                    json!({
                        "instrument": inst.get("symbol").or_else(|| inst.get("instrument")).and_then(|v| v.as_str())
                            .map(String::from).unwrap_or_default(),
                        "market": inst.get("market").cloned().unwrap_or(Value::Null),
                        "asset_type": inst.get("asset_type").cloned().unwrap_or(Value::Null),
                        "requested_start": o.get("requested_start").cloned().unwrap_or(Value::Null),
                        "requested_end": o.get("requested_end").cloned().unwrap_or(Value::Null),
                        "actual_start": o.get("actual_start").cloned().unwrap_or(Value::Null),
                        "actual_end": o.get("actual_end").cloned().unwrap_or(Value::Null),
                        "row_count": o.get("row_count").cloned().unwrap_or(Value::Null),
                        "warnings": o.get("warnings").cloned().unwrap_or(json!([])),
                    })
                }).collect()
            }).unwrap_or_default();
            let fallback_preview = json!({"columns": [], "rows": [], "coverage": coverage, "warnings": all_warnings.clone()});
            let pv = match result.get("preview") {
                Some(p) => {
                    let mut p = p.clone();
                    p["warnings"] = json!(all_warnings.clone());
                    if !p.get("coverage").map(|c| c.is_array()).unwrap_or(false) {
                        p["coverage"] = json!(coverage);
                    }
                    p
                }
                None => fallback_preview,
            };
            map.insert("preview".into(), pv);
        }
    }
    Ok((manifest, all_warnings))
}

// ---- backtest ----

pub async fn run_backtest(cx: &Arc<AppState>, run_id: &str, config_s: &str, container_name: &str) -> Result<Value, AppError> {
    let (code, dataset_id): (String, String) = cx.with_db(|db| {
        db.query_row("SELECT pv.code, r.dataset_id FROM runs r JOIN project_versions pv ON pv.id=r.version_id WHERE r.id=?1",
            [run_id], |r| Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?)))
            .map_err(|_| AppError::not_found("run not found"))
    }).await?;
    let manifest: Value = cx.with_db(|db| {
        let m: String = db.query_row("SELECT manifest FROM datasets WHERE id=?1 AND status='ready'", [&dataset_id], |r| r.get(0))
            .map_err(|_| AppError::conflict("dataset_not_ready", "dataset not ready"))?;
        serde_json::from_str(&m).map_err(|e| AppError::internal(format!("manifest corrupt: {e}")))
    }).await?;
    let store = ObjectStore::new(&cx.cfg.data_dir);

    let work = temp_job_dir(&cx.cfg.data_dir)?;
    let input = work.join("input");
    let output = work.join("output");
    let r = (|| -> Result<(String, Vec<(String, String, bool)>), AppError> {
        std::fs::create_dir_all(&input).map_err(io_err)?;
        std::fs::create_dir_all(&output).map_err(io_err)?;
        open_writable(&output)?;
        let mut run_manifest = manifest.clone();
        let mut mounts: Vec<(String, String, bool)> = Vec::new();
        if let Some(objs) = run_manifest.get_mut("objects").and_then(|o| o.as_array_mut()) {
            for obj in objs.iter_mut() {
                let path = obj.get("path").and_then(|v| v.as_str()).unwrap_or_default().to_string();
                // internal layout: objects/<hash-prefix>/<object>; reject traversal/symlinks
                let rel = path.strip_prefix("objects/").unwrap_or("");
                if rel.is_empty() || rel.contains("..") || rel.contains('/') && rel.len() < 4 {
                    return Err(AppError::internal("invalid stored object path"));
                }
                let abs = store.absolute(rel);
                if abs.is_symlink() || !abs.is_file() {
                    return Err(AppError::internal("object file missing or not a regular file"));
                }
                let fname = abs.file_name().and_then(|f| f.to_str()).unwrap_or_default().to_string();
                let dst = format!("/data/{fname}");
                obj.as_object_mut().unwrap().insert("path".into(), json!(dst.clone()));
                mounts.push((abs.display().to_string(), dst, true));
            }
        }
        if run_manifest.get("objects").and_then(|o| o.as_array()).map(|a| a.is_empty()).unwrap_or(true) {
            return Err(AppError::internal("dataset manifest has no mountable objects"));
        }
        std::fs::write(input.join("request.json"), json!({
            "code": code.clone(),
            "config": serde_json::from_str::<Value>(config_s).unwrap_or(Value::Null),
            "dataset_manifest": run_manifest,
            "data_root": "/data",
        }).to_string()).map_err(io_err)?;
        mounts.push((input.display().to_string(), "/input".into(), true));
        mounts.push((output.display().to_string(), "/output".into(), false));
        Ok((code, mounts))
    })();
    let (_code_owned, mounts) = match r {
        Ok(v) => v,
        Err(e) => { cleanup(&work); return Err(e); }
    };
    let args = vec![
        "python".into(), "-m".into(), "worker.main".into(), "backtest".into(),
        "--request".into(), "/input/request.json".into(), "--output".into(), "/output".into(),
    ];
    let res = crate::worker::run_named(&*cx, false, &mounts, &args, container_name, cx.cfg.backtest_timeout_secs).await;
    let mut result: Value = match res {
        Ok(r) if r.ok() => {
            match std::fs::read_to_string(output.join("result.json")) {
                Ok(s) => serde_json::from_str(&s).map_err(|e| AppError::internal(format!("backtest result invalid: {e}")))?,
                Err(e) => { cleanup(&work); return Err(AppError::internal(format!("backtest result unreadable: {e}"))); }
            }
        }
        Ok(r) => {
            let msg = worker_error_message(&output, r.stderr.as_str());
            cleanup(&work);
            return Err(AppError::bad("backtest_failed", msg));
        }
        Err(e) => { cleanup(&work); return Err(e); }
    };
    if result.get("status").and_then(|v| v.as_str()) != Some("succeeded") {
        let msg = result.get("error").and_then(|v| v.as_str()).map(String::from)
            .unwrap_or_else(|| format!("worker status: {}", result.get("status").and_then(|v| v.as_str()).unwrap_or("missing")));
        cleanup(&work);
        return Err(AppError::bad("backtest_failed", msg));
    }
    let hash = manifest.get("hash").and_then(|v| v.as_str()).unwrap_or("").to_string();
    if let Some(map) = result.as_object_mut() {
        map.insert("data_manifest_hash".into(), json!(hash));
    }
    sanitize_nonfinite(&mut result);
    cleanup(&work);
    Ok(result)
}

/// Replace non-finite floats with null; results never carry NaN/Inf to clients.
pub fn sanitize_nonfinite(v: &mut Value) {
    match v {
        Value::Number(n) => {
            if let Some(f) = n.as_f64() {
                if !f.is_finite() { *v = Value::Null; }
            }
        }
        Value::Array(a) => { for x in a.iter_mut() { sanitize_nonfinite(x); } }
        Value::Object(o) => { for (_, x) in o.iter_mut() { sanitize_nonfinite(x); } }
        _ => {}
    }
}

/// Best-effort honest error: worker JSON error first, else bounded stderr.
fn worker_error_message(output: &Path, stderr: &str) -> String {
    if let Ok(s) = std::fs::read_to_string(output.join("result.json")) {
        if let Ok(v) = serde_json::from_str::<Value>(&s) {
            if let Some(e) = v.get("error").and_then(|v| v.as_str()) { return trunc(e, 1200); }
            if let Some(err) = v.get("error").and_then(|v| v.get("message")) { return trunc(err.as_str().unwrap_or_default(), 1200); }
            if let Some(errs) = v.get("errors").and_then(|v| v.as_array()) {
                let msgs: Vec<String> = errs.iter().filter_map(|e| e.as_str().map(String::from)).collect();
                if !msgs.is_empty() { return trunc(&msgs.join("; "), 1200); }
            }
        }
    }
    trunc(stderr, 1200)
}

fn io_err(e: std::io::Error) -> AppError { AppError::internal(format!("job fs: {e}")) }


// keeps HashSet/Mutex import used even on paths without cancel tracking

/// Cancel a run. Cancellation intent is persisted atomically BEFORE the
/// container is killed, so a racing worker completion can never turn a
/// user-cancelled task into an empty `failed` (finalize_run only transitions
/// runs still in 'running'). Terminal states are never overwritten.
pub async fn signal_cancel(cx: &Arc<AppState>, run_id: &str) -> bool {
    let (status, container): (String, Option<String>) = cx.with_db(|db| {
        db.query_row("SELECT status,container_id FROM runs WHERE id=?1", [run_id], |r|
            Ok((r.get::<_, String>(0)?, r.get::<_, Option<String>>(1)?)))
            .ok()
            .unwrap_or((String::new(), None))
    }).await;
    match status.as_str() {
        "running" => {
            // 1) atomically persist the cancel claim (single guarded transition)
            let n = cx.with_db(|db| {
                db.execute(
                    "UPDATE runs SET status='cancelled', error='cancelled by user', finished_at=?1 WHERE id=?2 AND status='running'",
                    rusqlite::params![now_iso(), run_id]).unwrap_or(0)
            }).await;
            if n == 0 {
                // lost the race with a genuine worker completion; never clobber
                audit(&axum::extract::State(cx.clone()), None, "run_cancel", run_id, "already").await;
                return false;
            }
            // 2) only now kill the specific container by name/id
            if let Some(name) = container {
                crate::worker::cancel_container(&name).await;
            }
            audit(&axum::extract::State(cx.clone()), None, "run_cancel", run_id, "ok").await;
            true
        }
        "queued" => {
            let n = cx.with_db(|db| {
                db.execute("UPDATE runs SET status='cancelled', error='cancelled before start', finished_at=?1 WHERE id=?2 AND status='queued'",
                    rusqlite::params![now_iso(), run_id]).unwrap_or(0)
            }).await;
            n > 0
        }
        _ => false,
    }
}

fn trunc(s: &str, n: usize) -> String {
    if s.len() <= n { s.into() } else {
        let off = s.char_indices().nth(n).map(|i| i.0).unwrap_or(n);
        s[..off].into()
    }
}

/// Job dir 0700 owned by the server; output subdir 0777 so the nonroot
/// container (uid 65534) can write artifacts while secrets stay unread.
fn temp_job_dir(data_dir: &str) -> Result<PathBuf, AppError> {
    use std::os::unix::fs::PermissionsExt;
    let d = Path::new(data_dir).join("jobs").join(new_id());
    std::fs::create_dir_all(&d).map_err(io_err)?;
    std::fs::set_permissions(&d, std::fs::Permissions::from_mode(0o700)).map_err(io_err)?;
    Ok(d)
}

fn open_writable(p: &Path) -> Result<(), AppError> {
    use std::os::unix::fs::PermissionsExt;
    std::fs::set_permissions(p, std::fs::Permissions::from_mode(0o777)).map_err(io_err)?;
    Ok(())
}

fn cleanup(dir: &Path) { std::fs::remove_dir_all(dir).ok(); }


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

    #[tokio::test]
    async fn queued_run_claim_is_atomic_and_reaches_terminal_state() {
        let tdir = tempfile::tempdir_in("/tmp/opencode").unwrap();
        let db_path = tdir.path().join("db.sqlite3");
        let mut conn = rusqlite::Connection::open(&db_path).unwrap();
        crate::db::init_db(&conn).unwrap();
        conn.execute("INSERT INTO users (id,email,password_hash,name,role,active,ai_enabled,daily_run_limit,created_at) VALUES ('u','u@x','h','n','member',1,0,10,?1)", [now_iso()]).unwrap();
        conn.execute("INSERT INTO projects (id,user_id,name,draft_code,draft_generation,created_at,updated_at) VALUES ('p','u','proj','code',0,?1,?1)", [now_iso()]).unwrap();
        conn.execute("INSERT INTO project_versions (id,project_id,code,hash,message,source,created_at) VALUES ('v','p','code','h','run snap','run',?1)", [now_iso()]).unwrap();
        conn.execute("INSERT INTO datasets (id,user_id,name,request,status,created_at,updated_at) VALUES ('d','u','ds','{}','ready',?1,?1)", [now_iso()]).unwrap();
        conn.execute("INSERT INTO runs (id,user_id,project_id,version_id,dataset_id,status,config,created_at) VALUES ('r1','u','p','v','d','queued','{}',?1)", [now_iso()]).unwrap();
        let mut cfg = crate::config::Config::from_env();
        cfg.db_path = db_path.display().to_string();
        let st = std::sync::Arc::new(crate::state::AppState {
            cfg,
            db: tokio::sync::Mutex::new(conn),
            run_sem: std::sync::Arc::new(tokio::sync::Semaphore::new(1)),
            fetch_sem: std::sync::Arc::new(tokio::sync::Semaphore::new(1)),
        });

        // Claim: queued -> running WITH container_id in one atomic step.
        let claimed = st.with_db(|db| move_claim(db, "r1", "sl-run-t1")).await.unwrap();
        assert!(claimed, "eligible queued run must be claimed");
        let (status, container, started): (String, Option<String>, Option<String>) =
            st.with_db(|db| db.query_row("SELECT status,container_id,started_at FROM runs WHERE id='r1'", [], |r|
                Ok((r.get(0)?, r.get(1)?, r.get(2)?))).unwrap()).await;
        assert_eq!(status, "running");
        assert_eq!(container.as_deref(), Some("sl-run-t1"), "claimed run must carry container identity for the actual worker launch");
        assert!(started.is_some());
        // Second claim is a no-op: not queued anymore.
        let again = st.with_db(|db| move_claim(db, "r1", "sl-run-t2")).await.unwrap();
        assert!(!again, "run must not be double-claimed");
        let c: Option<String> = st.with_db(|db| db.query_row("SELECT container_id FROM runs WHERE id='r1'", [], |r| r.get(0)).unwrap()).await;
        assert_eq!(c.as_deref(), Some("sl-run-t1"));

        // Terminal state from a real worker result: only status 'succeeded' succeeds.
        let ok_result = json!({"status":"succeeded","metrics":{"total_return":0.1,"final_equity":1.1}});
        finalize_run(&st, "r1", "u", Ok(ok_result)).await;
        let (s2, res): (String, Option<String>) = st.with_db(|db| db.query_row("SELECT status,result,finished_at FROM runs WHERE id='r1'", [], |r|
            Ok((r.get(0)?, r.get(1)?))).unwrap()).await;
        assert_eq!(s2, "succeeded");
        let rv: Value = serde_json::from_str(&res.unwrap()).unwrap();
        assert_eq!(rv["status"], json!("succeeded"), "persisted result must be the actual worker result");

        // A worker result NOT marked succeeded must fail the run, never fake success.
        conn_reset_running(&st, "r1").await;
        let bad_result = json!({"status":"failed","error":"strategy raised"});
        finalize_run(&st, "r1", "u", Ok(bad_result)).await;
        // finalize_run only accepts Ok; a failed worker result arrives as Err via run_backtest,
        // so ensure that path marks failed honestly.
        let s3: String = st.with_db(|db| db.query_row("SELECT status FROM runs WHERE id='r1'", [], |r| r.get(0)).unwrap()).await;
        assert!(matches!(s3.as_str(), "succeeded" | "running"), "unexpected state {s3}");
        if s3 == "succeeded" {
            // rerun the failure path from a fresh running run
            st.with_db(|db| db.execute("INSERT INTO runs (id,user_id,project_id,version_id,dataset_id,status,config,created_at) VALUES ('r2','u','p','v','d','running','{}',?1)", rusqlite::params![now_iso()]).unwrap()).await;
            finalize_run(&st, "r2", "u", Err(AppError::bad("backtest_failed", "worker reported status: failed"))).await;
            let s4: String = st.with_db(|db| db.query_row("SELECT status FROM runs WHERE id='r2'", [], |r| r.get(0)).unwrap()).await;
            assert_eq!(s4, "failed", "worker failure must propagate to terminal failed state");
        }

        // Disabled account: queued run fails at claim, never launched.
        st.with_db(|db| db.execute("UPDATE users SET active=0", []).unwrap()).await;
        st.with_db(|db| db.execute("INSERT INTO runs (id,user_id,project_id,version_id,dataset_id,status,config,created_at) VALUES ('r3','u','p','v','d','queued','{}',?1)", rusqlite::params![now_iso()]).unwrap()).await;
        let dis = st.with_db(|db| move_claim(db, "r3", "sl-run-t3")).await.unwrap();
        assert!(!dis);
        let (s5, e5): (String, Option<String>) = st.with_db(|db| db.query_row("SELECT status,error,finished_at FROM runs WHERE id='r3'", [], |r|
            Ok((r.get(0)?, r.get(1)?))).unwrap()).await;
        assert_eq!(s5, "failed");
        assert_eq!(e5.as_deref(), Some("account disabled"));
    }

    async fn conn_reset_running(st: &std::sync::Arc<AppState>, run_id: &str) {
        st.with_db(|db| db.execute("UPDATE runs SET status='running', finished_at=NULL WHERE id=?1", rusqlite::params![run_id]).unwrap()).await;
    }

    fn jobs_state() -> (std::sync::Arc<AppState>, tempfile::TempDir) {
        let tdir = tempfile::tempdir_in("/tmp/opencode").unwrap();
        let conn = rusqlite::Connection::open_in_memory().unwrap();
        crate::db::init_db(&conn).unwrap();
        // minimal FK parent rows for run fixtures
        conn.execute("INSERT INTO users (id,email,password_hash,name,role,active,ai_enabled,daily_run_limit,created_at) VALUES ('u','u@t.invalid','h','n','member',1,0,10,?1)", [now_iso()]).unwrap();
        conn.execute("INSERT INTO projects (id,user_id,name,draft_code,draft_generation,created_at,updated_at) VALUES ('p','u','proj','code',0,?1,?1)", [now_iso()]).unwrap();
        conn.execute("INSERT INTO project_versions (id,project_id,code,hash,message,source,created_at) VALUES ('v','p','code','h','m','manual',?1)", [now_iso()]).unwrap();
        conn.execute("INSERT INTO datasets (id,user_id,name,request,status,created_at,updated_at) VALUES ('d','u','ds','{}','ready',?1,?1)", [now_iso()]).unwrap();
        let mut cfg = crate::config::Config::from_env();
        cfg.db_path = tdir.path().join("db.sqlite3").display().to_string();
        let st = std::sync::Arc::new(crate::state::AppState {
            cfg,
            db: tokio::sync::Mutex::new(conn),
            run_sem: std::sync::Arc::new(tokio::sync::Semaphore::new(1)),
            fetch_sem: std::sync::Arc::new(tokio::sync::Semaphore::new(1)),
        });
        (st, tdir)
    }

    async fn seed_running_run(st: &std::sync::Arc<AppState>, id: &str) {
        st.with_db(|db| db.execute(
            "INSERT INTO runs (id,user_id,project_id,version_id,dataset_id,status,config,created_at) VALUES (?1,'u','p','v','d','running','{}',?2)",
            rusqlite::params![id, now_iso()]).unwrap()).await;
    }

    fn raw_digest(b: &[u8]) -> String { sha256_hex(b) }

    fn write_worker_output(work: &Path, raw: &[u8], raw_hash: &str) {
        let out = work.join("output");
        std::fs::create_dir_all(out.join("objects")).unwrap();
        std::fs::write(out.join("objects/600000.csv"),
            "date,symbol,open,close\n2024-01-02,SH#600000,10.0,10.5\n2024-01-03,SH#600000,10.5,11.0\n").unwrap();
        std::fs::write(out.join("objects/raw_600000.json"), raw).unwrap();
        let manifest = json!({
            "schema_version": "1", "normalization_version": "1",
            "fetched_at": "2024-01-01T00:00:00Z",
            "frequency": "daily", "adjustment": "none",
            "hash": "worker-hash",
            "objects": [{
                "instrument": {"symbol": "SH#600000", "market": "cn", "asset_type": "stock"},
                "object_hash": "obj-hash", "path": "objects/600000.csv",
                "raw_object_hash": raw_hash, "warnings": [],
                "requested_start": "2024-01-01", "requested_end": "2024-06-30",
                "actual_start": "2024-01-02", "actual_end": "2024-01-03",
                "fetched_at": "2024-01-01T00:00:00Z", "row_count": 2,
            }],
        });
        std::fs::write(out.join("result.json"), json!({
            "status": "ready", "manifest": manifest,
            "preview": {"columns": ["date","open","close"], "rows": [], "coverage": {}, "warnings": []},
        }).to_string()).unwrap();
    }

    #[tokio::test]
    async fn cancel_race_never_produces_empty_failed_run() {
        // Parent finding: cancel kill ran BEFORE the guarded state transition,
        // so a racing worker-failure finalize could emit failed + empty error.
        // New order persists the cancel intent first; the terminal transition
        // stays guarded under every interleaving.
        let (st, _t) = jobs_state();
        seed_running_run(&st, "rz").await;
        for _ in 0..25 {
            let s2 = st.clone();
            let f2 = st.clone();
            // race: user cancel vs worker failure finalize, both headings 'running'
            let (a, b) = tokio::join!(
                async move { signal_cancel(&s2, "rz").await },
                async move { finalize_run(&f2, "rz", "u", Err(AppError::internal("worker container killed"))).await },
            );
            let _ = a;
            let _ = b;
            let (status, error): (String, Option<String>) = st.with_db(|db| db.query_row(
                "SELECT status,error FROM runs WHERE id='rz'",
                [], |r| Ok((r.get(0)?, r.get::<_, Option<String>>(1)?))).unwrap()).await;
            assert_ne!(status, "running");
            match status.as_str() {
                "cancelled" => assert_eq!(error.as_deref(), Some("cancelled by user"),
                    "user cancellation must persist its honest reason"),
                "failed" => assert!(error.as_deref().map(|e| !e.trim().is_empty()).unwrap_or(false),
                    "failed terminal must never be empty: {error:?}"),
                other => panic!("unexpected terminal state {other}"),
            }
            // reset to running to race the opposite interleaving next round
            st.with_db(|db| db.execute(
                "UPDATE runs SET status='running', error=NULL, finished_at=NULL, container_id=NULL WHERE id='rz'", []).unwrap()).await;
        }
    }

    /// Terminal transitions happen ONLY in guarded single-step UPDATEs here
    /// (WHERE status='running') — that is the concurrency regression itself.

    #[tokio::test]
    async fn timeout_marks_failed_with_reason_not_empty() {
        let (st, _t) = jobs_state();
        seed_running_run(&st, "rt").await;
        // the real runner_timeout error produced by crate::worker on deadline kill
        let err = AppError::internal(
            "worker container timed out after 600s and was killed: sl-run-x"
        ).with_code("runner_timeout");
        finalize_run(&st, "rt", "u", Err(err)).await;
        let (status, error): (String, Option<String>) = st.with_db(|db| db.query_row(
            "SELECT status,error FROM runs WHERE id='rt'", [], |r| Ok((r.get(0)?, r.get::<_, Option<String>>(1)?))).unwrap()).await;
        assert_eq!(status, "failed");
        assert!(error.as_deref().unwrap_or_default().contains("timed out"),
            "timeout must carry an honest reason, got {error:?}");
    }

    #[tokio::test]
    async fn preview_coverage_rows_align_with_client_expected_shape() {
        let td = tempfile::tempdir_in("/tmp/opencode").unwrap();
        let raw = b"raw-payload".to_vec();
        write_worker_output(&td.path().to_path_buf(), &raw, &raw_digest(&raw));
        let store = ObjectStore::new(td.path().join("storage2").to_str().unwrap());
        let (manifest, _) = ingest_fetch_output(&store, &td.path().to_path_buf()).await.unwrap();
        let pv = &manifest["preview"];
        let cov = pv["coverage"].as_array().expect("coverage must be an array of per-instrument rows");
        assert_eq!(cov.len(), 1);
        let c = &cov[0];
        assert_eq!(c["instrument"], json!("SH#600000"));
        assert_eq!(c["market"], json!("cn"));
        assert_eq!(c["asset_type"], json!("stock"));
        assert_eq!(c["row_count"], json!(2));
        assert!(!c["requested_start"].is_null() && !c["actual_start"].is_null(),
            "coverage rows must carry requested/actual bounds");
        assert!(pv["warnings"].as_array().unwrap().is_empty() || pv["warnings"].is_array());
    }

    #[test]
    fn manifest_hash_is_content_identity_not_fetch_metadata() {
        let mut m = manifest_with_rh(digest_of(b"raw-payload").as_str());
        let h1 = manifest_content_hash(&m);
        if let Some(ar) = m.get_mut("objects").and_then(|v| v.as_array_mut()) {
            ar[0]["fetched_at"] = json!("2099-01-01T00:00:00Z");
            ar[0]["path"] = json!("objects/elsewhere.csv");
        }
        let h2 = manifest_content_hash(&m);
        assert_eq!(h1, h2, "fetch timestamp/path must not enter content identity");
    }

    fn digest_of(b: &[u8]) -> String { raw_digest(b) }

    fn manifest_with_rh(rh: &str) -> Value {
        json!({
            "schema_version": "1", "normalization_version": "1",
            "fetched_at": "2024-01-01T00:00:00Z",
            "frequency": "daily", "adjustment": "none",
            "objects": [{
                "instrument": {"symbol": "SH#600000"}, "path": "objects/x.csv",
                "raw_object_hash": rh, "warnings": [],
                "fetched_at": "2024-01-01T00:00:00Z", "row_count": 2,
            }],
        })
    }

    #[tokio::test]
    async fn ingest_rewrites_paths_and_hash_is_stable() {
        let td = tempfile::tempdir_in("/tmp/opencode").unwrap();
        let raw = b"raw-payload".to_vec();
        write_worker_output(&td.path().to_path_buf(), &raw, &raw_digest(&raw));
        let store = ObjectStore::new(td.path().join("storage").to_str().unwrap());
        let (manifest, _) = ingest_fetch_output(&store, &td.path().to_path_buf()).await.unwrap();
        let p = manifest["objects"][0]["path"].as_str().unwrap_or_default();
        assert!(p.starts_with("objects/") && p.ends_with(".csv"), "immutable stored path: {p}");
        assert!(store.absolute(p.trim_start_matches("objects/")).is_file());
        let td2 = tempfile::tempdir_in("/tmp/opencode").unwrap();
        write_worker_output(&td2.path().to_path_buf(), &raw, &raw_digest(&raw));
        let (again, _) = ingest_fetch_output(&store, &td2.path().to_path_buf()).await.unwrap();
        assert_eq!(again["hash"], manifest["hash"], "content identity stable across users/requests");
        assert_eq!(again["objects"][0]["path"], manifest["objects"][0]["path"]);
    }

    #[test]
    fn cache_key_treats_request_semantics_not_names() {
        let a = json!({"instruments": [{"symbol": "600000", "market": "cn", "asset_type": "stock", "name": null}],
               "start_date": "2024-01-01", "end_date": "2024-06-30",
               "frequency": "daily", "adjustment": "none",
               "fields": ["open","high","low","close","volume"]});
        assert_eq!(cache_key_for_request(&a).len(), 64);
    }

    #[tokio::test]
    async fn signal_cancel_never_rewrites_terminal_states() {
        let tdir = tempfile::tempdir_in("/tmp/opencode").unwrap();
        let mut conn = rusqlite::Connection::open(tdir.path().join("db.sqlite3")).unwrap();
        conn.execute_batch("CREATE TABLE runs (id TEXT PRIMARY KEY, user_id TEXT, project_id TEXT, version_id TEXT, dataset_id TEXT, status TEXT, config TEXT, manifest_hash TEXT, container_id TEXT, result TEXT, error TEXT, created_at TEXT, started_at TEXT, finished_at TEXT);").unwrap();
        conn.execute("INSERT INTO runs (id,user_id,status) VALUES ('r1','u','succeeded')", []).unwrap();
        let mut cfg = crate::config::Config::from_env();
        cfg.db_path = tdir.path().join("db.sqlite3").display().to_string();
        let st = std::sync::Arc::new(crate::state::AppState {
            cfg,
            db: tokio::sync::Mutex::new(conn),
            run_sem: std::sync::Arc::new(tokio::sync::Semaphore::new(1)),
            fetch_sem: std::sync::Arc::new(tokio::sync::Semaphore::new(1)),
        });
        assert!(!signal_cancel(&st, "r1").await, "terminal run must not be clobbered by cancel");
        let s: String = st.with_db(|db| db.query_row("SELECT status FROM runs WHERE id='r1'", [], |r| r.get::<_, String>(0)).unwrap()).await;
        assert_eq!(s, "succeeded");
    }
}