"""
ExCALIBR FastAPI backend

Run with:
    conda activate excalibr310
    uvicorn api:app --host 127.0.0.1 --port 8000

Pipeline is invoked as a niced subprocess so it yields to other server processes.
Job state is persisted to disk (meta.json per job) so it survives restarts.
"""

import logging
import os
import gzip
import importlib.util
import json
import re
import shutil
import signal
import subprocess
import sys
import time
import traceback
import uuid
import asyncio
from collections import defaultdict
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timedelta
from pathlib import Path
from typing import Dict, Literal
from contextlib import asynccontextmanager

import numpy as np
import pandas as pd
from fastapi import BackgroundTasks, FastAPI, File, Form, HTTPException, Request, UploadFile
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse, Response

# ── Paths ─────────────────────────────────────────────────────────────────────

BASE_DIR        = Path(__file__).parent.parent
JOBS_DIR        = Path(os.environ.get("EXCALIBR_JOBS_DIR", str(BASE_DIR / "jobs")))
PIPELINE_DIR    = BASE_DIR / "ExCALIBR"
PIPELINE_SCRIPT = PIPELINE_DIR / "run_pipeline.py"
PYTHON_BIN      = Path(sys.executable)

FRONTEND_BASE_URL = "https://excalibr.org"

JOBS_DIR.mkdir(exist_ok=True)

logger = logging.getLogger("excalibr.api")
logging.basicConfig(level=logging.INFO)

# ── Limits ────────────────────────────────────────────────────────────────────

MAX_UPLOAD_BYTES         = 100 * 1024 * 1024   # 100 MB
MAX_CONCURRENT_JOBS      = 3
MAX_JOBS_PER_IP_PER_HOUR = 10

# Last-resort fallback only -- used if _load_bootstrap_presets() below fails to
# reach run_pipeline.py's real BOOTSTRAP_PRESETS for any reason. If this fallback
# is ever actually in use, presets may be silently stale relative to the pipeline;
# _load_bootstrap_presets() logs loudly whenever that happens.
_FALLBACK_PRESETS = {
    "light":  (20,   8),
    "medium": (100,  8),
    "large":  (500,  8),
    "xl":     (1000, 8),
    "finest": (1000, 100),
}


def _load_bootstrap_presets() -> dict:
    """
    Fetch BOOTSTRAP_PRESETS directly from run_pipeline.py -- the pipeline's own
    source of truth -- via a short-lived subprocess, rather than importing
    run_pipeline as a module inside this process. Importing it directly would
    pull its full dependency chain (pandas, BootstrapRunner, visualize, and
    whatever those import, likely including JAX) into the API server's own
    memory permanently; this way that cost is paid once, briefly, in a
    throwaway process at startup, not carried for the server's whole lifetime.

    Falls back to _FALLBACK_PRESETS (logged loudly) if this fails for any
    reason, so an unrelated break in the pipeline's import chain can't take
    down job submission entirely -- it just means presets may be stale until
    the underlying issue is fixed.
    """
    code = (
        "import json; "
        "from run_pipeline import BOOTSTRAP_PRESETS; "
        "print(json.dumps(BOOTSTRAP_PRESETS))"
    )
    try:
        result = subprocess.run(
            [str(PYTHON_BIN), "-c", code],
            cwd=str(PIPELINE_DIR),
            capture_output=True,
            text=True,
            timeout=60,
        )
        if result.returncode != 0:
            raise RuntimeError(result.stderr[-2000:])
        loaded = json.loads(result.stdout)
        return {k: tuple(v) for k, v in loaded.items()}
    except Exception:
        logger.error(
            "Could not load BOOTSTRAP_PRESETS from run_pipeline.py -- falling back to "
            "a hardcoded copy, which may be STALE relative to the pipeline:\n%s",
            traceback.format_exc(),
        )
        return dict(_FALLBACK_PRESETS)


PRESETS = _load_bootstrap_presets()

# Presets the website will run on shared infrastructure. xl/finest are far more
# expensive (50x and 625x light's fit count respectively) and are intentionally
# withheld here -- the frontend also hides them behind a "run locally" note, but
# this is the actual enforcement boundary, since a request built directly against
# POST /api/jobs (bypassing the UI) would otherwise still be able to request them.
# This set is a website policy decision, not pipeline data, so (unlike PRESETS
# above) it isn't fetched from the pipeline -- there's nothing to drift out of
# sync with, just a frontend copy that should be kept editorially consistent.
ALLOWED_WEB_PRESETS = {"light", "medium", "large"}

# Worst-case legitimate web job is now bounded by ALLOWED_WEB_PRESETS' most
# expensive entry ("large": 500 * 8 = 4,000 fits) rather than the old unbounded
# granular ceiling (which allowed up to 100,000 fits) -- MAX_RUNTIME should still
# be benchmarked against that actual worst case, but the number to benchmark is
# now much smaller and knowable in advance, rather than a rough extrapolation.
MAX_RUNTIME              = timedelta(days=7)
RETENTION_BUFFER         = timedelta(weeks=1)
RETENTION_PERIOD         = MAX_RUNTIME + RETENTION_BUFFER

# How often the in-app periodic sweep checks for expired job directories.
# Independent of job submissions, unlike the old background_tasks-only cleanup.
CLEANUP_INTERVAL_SECONDS = 3600

# Cores handed to each concurrently-running pipeline invocation. Previously each
# job was launched with --n-jobs -1 (claim all cores) regardless of how many
# jobs were running at once, so MAX_CONCURRENT_JOBS "slots" all fought over the
# same cores instead of sharing them -- this is what produced "waiting for free
# slot" symptoms even when a thread-pool slot was technically free.
N_JOBS_PER_TASK = max(1, (os.cpu_count() or MAX_CONCURRENT_JOBS) // MAX_CONCURRENT_JOBS)

# ── LR+ threshold computation (tauP/tauB) ──────────────────────────────────────
#
# Computes the log-LR+ thresholds (tauP for pathogenic tiers +1..+8, tauB for
# benign tiers -1..-8) at a given prior, for the LR+ panel's dashed threshold
# lines. Loaded directly from evidence_thresholds.py via importlib rather than
# `import assay_calibration...` (which would execute parent package __init__
# files and could pull in the same heavy scientific stack avoided elsewhere in
# this file) or reimplementing the Tavtigian-constant search ourselves (which
# would be a second independent reimplementation of this exact pipeline math,
# same drift risk already eliminated for presets and LR+ curves elsewhere).
# evidence_thresholds.py imports only numpy, so this load is cheap and safe.

_EVIDENCE_THRESHOLDS_PATH = (
    PIPELINE_DIR / "src" / "assay_calibration" / "fit_utils" / "evidence_thresholds.py"
)

# Mirrors PipelineConfig.point_values' default ([1..8]) in
# src/assay_calibration/pipeline/config.py. Only used as a fallback point
# scale for threshold computation -- if that default ever changes, this
# should too.
POINT_VALUES = [1, 2, 3, 4, 5, 6, 7, 8]


def _load_get_tavtigian_constant():
    spec = importlib.util.spec_from_file_location(
        "excalibr_evidence_thresholds", _EVIDENCE_THRESHOLDS_PATH
    )
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module.get_tavtigian_constant


try:
    _get_tavtigian_constant = _load_get_tavtigian_constant()
except Exception:
    logger.error(
        "Could not load get_tavtigian_constant from evidence_thresholds.py -- "
        "the /api/thresholds endpoint will be unavailable until this is fixed:\n%s",
        traceback.format_exc(),
    )
    _get_tavtigian_constant = None


def _compute_log_lr_thresholds(prior: float):
    """tauP/tauB (natural-log LR+ thresholds at each point value) for the
    given prior. Matches thresholds_from_prior(..., acmg_mapping_method=
    'tavtigian') in fit.py exactly -- api.py never sends --acmg-mapping-method,
    so 'tavtigian' (the tool's own default) is always the correct method here.
    Returns (tauP, tauB) as plain float lists, or None if the loader failed."""
    if _get_tavtigian_constant is None:
        return None
    pv = np.array(POINT_VALUES, dtype=float)
    C = _get_tavtigian_constant(prior)
    lrP = C ** (pv / len(pv))
    lrB = 1 / lrP
    return np.log(lrP).tolist(), np.log(lrB).tolist()

# ── Concurrency ───────────────────────────────────────────────────────────────

executor = ThreadPoolExecutor(max_workers=MAX_CONCURRENT_JOBS)

# ── In-memory state ───────────────────────────────────────────────────────────

JobStatus = Literal["pending", "running", "complete", "failed"]
jobs: Dict[str, dict] = {}

# Simple in-memory rate limiting: IP -> list of submission timestamps
_rate_limit: Dict[str, list] = defaultdict(list)

# ── Input validation helpers ──────────────────────────────────────────────────

_EMAIL_RE = re.compile(r'^[a-zA-Z0-9._%+\-]+@[a-zA-Z0-9.\-]+\.[a-zA-Z]{2,}$')


def _validate_email(email: str) -> str:
    """Validate and return a safe email address, or raise HTTPException."""
    email = email.strip()
    if not email:
        return ""
    if not _EMAIL_RE.match(email):
        raise HTTPException(status_code=400, detail="Invalid email address")
    return email


def _sanitize_name(name: str) -> str:
    """Sanitize a name for safe use in filenames and subprocess args."""
    sanitized = re.sub(r'[^a-zA-Z0-9_\-. ]', '_', name)
    return sanitized[:128] or "dataset"


def _validate_params(p: dict):
    """Validate pipeline params and raise HTTPException on invalid input."""
    components = p.get("components", [])
    if not components or not isinstance(components, list):
        raise HTTPException(status_code=400, detail="components must be a non-empty list")
    if not all(c in [2, 3, 4] for c in components):
        raise HTTPException(status_code=400, detail="components must be 2, 3, or 4")

    preset = p.get("preset")
    if preset not in PRESETS:
        raise HTTPException(status_code=400, detail=f"preset must be one of {sorted(PRESETS)}")
    if preset not in ALLOWED_WEB_PRESETS:
        raise HTTPException(
            status_code=400,
            detail=(
                f"preset '{preset}' is too compute-intensive to run on the website -- "
                f"run ExCALIBR locally for this preset. Presets available here: "
                f"{sorted(ALLOWED_WEB_PRESETS)}"
            ),
        )

    benign_method = p.get("benignMethod", "")
    if benign_method not in ("benign", "avg", "synonymous"):
        raise HTTPException(status_code=400, detail="Invalid benignMethod")

    postprocess_mode = p.get("postprocessMode", "standard")
    if postprocess_mode not in ("standard", "conservative", "none"):
        raise HTTPException(
            status_code=400,
            detail="postprocessMode must be 'standard', 'conservative', or 'none'",
        )

    manual_prior = p.get("manualPrior")
    if manual_prior is not None:
        if not isinstance(manual_prior, (int, float)) or not (0 < manual_prior < 1):
            raise HTTPException(status_code=400, detail="manualPrior must be between 0 and 1")

    pathomechanism_prior = p.get("pathomechanismPrior", False)
    if not isinstance(pathomechanism_prior, bool):
        raise HTTPException(status_code=400, detail="pathomechanismPrior must be a boolean")
    if pathomechanism_prior and manual_prior is not None:
        # Mirrors the pipeline's own behavior (visualize.py: "pathomechanism_method
        # is ignored because manual_prior is set") -- enforced here too so a direct
        # API call can't silently create the same landmine the UI prevents.
        raise HTTPException(
            status_code=400,
            detail="pathomechanismPrior and manualPrior are mutually exclusive -- "
                   "manual_prior overrides prior estimation entirely, which pathomechanism "
                   "correction has no effect on.",
        )


def _check_rate_limit(ip: str):
    """Allow at most MAX_JOBS_PER_IP_PER_HOUR submissions per IP per hour."""
    now = datetime.utcnow()
    cutoff = now - timedelta(hours=1)
    _rate_limit[ip] = [t for t in _rate_limit[ip] if t > cutoff]
    if len(_rate_limit[ip]) >= MAX_JOBS_PER_IP_PER_HOUR:
        raise HTTPException(
            status_code=429,
            detail="Too many jobs submitted. Please wait before submitting again."
        )
    _rate_limit[ip].append(now)

# ── Process management ────────────────────────────────────────────────────────
#
# Pipeline subprocesses are launched in their own session (start_new_session=True)
# so they head a fresh process group. run_pipeline.py is invoked with --mode
# parallel, which likely spawns its own worker subprocesses (joblib/loky-style) --
# killing only the parent PID would orphan those workers. Killing the whole
# process group avoids that.


def _kill_process_group(proc: subprocess.Popen, grace_seconds: float = 10.0):
    """Terminate a subprocess and its entire process group. SIGTERM first, then
    SIGKILL after a grace period if it hasn't exited."""
    try:
        pgid = os.getpgid(proc.pid)
    except ProcessLookupError:
        return
    try:
        os.killpg(pgid, signal.SIGTERM)
    except ProcessLookupError:
        return
    try:
        proc.wait(timeout=grace_seconds)
        return
    except subprocess.TimeoutExpired:
        pass
    try:
        os.killpg(pgid, signal.SIGKILL)
    except ProcessLookupError:
        return
    try:
        proc.wait(timeout=grace_seconds)
    except subprocess.TimeoutExpired:
        pass


def _kill_pid_by_number(pid: int, grace_seconds: float = 5.0):
    """Best-effort kill of a process (and its group) by bare PID -- used when we
    no longer hold the original Popen object, e.g. after a server restart."""
    try:
        pgid = os.getpgid(pid)
    except ProcessLookupError:
        return
    try:
        os.killpg(pgid, signal.SIGTERM)
    except ProcessLookupError:
        return
    time.sleep(grace_seconds)
    try:
        os.killpg(pgid, signal.SIGKILL)
    except ProcessLookupError:
        pass

# ── Disk persistence ──────────────────────────────────────────────────────────

def _meta_path(job_id: str) -> Path:
    return JOBS_DIR / job_id / "meta.json"


def _pid_path(job_id: str) -> Path:
    return JOBS_DIR / job_id / "pid"


def _write_meta(job_id: str):
    meta = jobs[job_id]
    payload = {
        "job_id":            job_id,
        "status":            meta["status"],
        "created_at":        meta["created_at"].isoformat(),
        "original_filename": meta["original_filename"],
        "dataset_name":      meta["dataset_name"],
        "email":             meta.get("email"),
        "error":             meta.get("error"),
        "preset":             meta.get("preset"),
        "n_bootstraps":       meta.get("n_bootstraps"),
        "fits_per_bootstrap": meta.get("fits_per_bootstrap"),
        "pid":                meta.get("pid"),
    }
    _meta_path(job_id).write_text(json.dumps(payload, indent=2))


def _load_meta(job_id: str) -> dict | None:
    path = _meta_path(job_id)
    if not path.exists():
        return None
    try:
        payload = json.loads(path.read_text())
        return {
            "status":            payload["status"],
            "created_at":        datetime.fromisoformat(payload["created_at"]),
            "output_dir":        JOBS_DIR / job_id / "output",
            "original_filename": payload.get("original_filename", "input.csv"),
            "dataset_name":      payload.get("dataset_name", job_id),
            "email":             payload.get("email"),
            "error":             payload.get("error"),
            "preset":             payload.get("preset"),
            "n_bootstraps":       payload.get("n_bootstraps"),
            "fits_per_bootstrap": payload.get("fits_per_bootstrap"),
            "pid":                payload.get("pid"),
        }
    except Exception:
        return None


def _recover_jobs_from_disk():
    """
    On startup, rebuild in-memory jobs dict from any job directories on disk.
    Jobs that were 'running' when the server went down are marked 'failed', and
    we make a best-effort attempt to kill any orphaned OS process left over from
    before the restart (identified via the job's pid file), since the in-memory
    Popen handle from the previous process is gone.
    """
    cutoff = datetime.utcnow() - RETENTION_PERIOD
    for job_dir in JOBS_DIR.iterdir():
        if not job_dir.is_dir():
            continue
        job_id = job_dir.name
        meta = _load_meta(job_id)
        if meta is None:
            continue
        if meta["created_at"] < cutoff:
            pid_file = _pid_path(job_id)
            if pid_file.exists():
                try:
                    _kill_pid_by_number(int(pid_file.read_text().strip()))
                except Exception:
                    pass
            shutil.rmtree(job_dir, ignore_errors=True)
            continue
        if meta["status"] == "running":
            pid_file = _pid_path(job_id)
            if pid_file.exists():
                try:
                    _kill_pid_by_number(int(pid_file.read_text().strip()))
                except Exception:
                    pass
                try:
                    pid_file.unlink()
                except OSError:
                    pass
            meta["status"] = "failed"
            meta["error"] = "Server restarted while job was running."
            meta["pid"] = None
        jobs[job_id] = meta

# ── Email ─────────────────────────────────────────────────────────────────────

def _send_email(to: str, job_id: str, status: str):
    if not to:
        return

    results_url = f"{FRONTEND_BASE_URL}/results/{job_id}"

    if status == "complete":
        subject = "ExCALIBR: your calibration job is complete"
        meta = jobs[job_id]
        body = (
            f"Your ExCALIBR calibration job has finished successfully.\n\n"
            f"View your results here:\n{results_url}\n\n"
            f"Job details:\n"
            f"  File:              {meta.get('original_filename', 'unknown')}\n"
            f"  Dataset:           {meta.get('dataset_name', 'unknown')}\n"
            f"  Bootstraps:        {meta.get('n_bootstraps', 'unknown')}\n"
            f"  Fits:  {meta.get('fits_per_bootstrap', 'unknown')}\n"
            f"  Submitted:         {meta['created_at'].strftime('%Y-%m-%d %H:%M UTC')}\n\n"
            f"Results are available for {RETENTION_PERIOD.days} days.\n\n"
            f"Job ID: {job_id}\n"
        )
    else:
        subject = "ExCALIBR: your calibration job failed"
        body = (
            f"Your ExCALIBR calibration job encountered an error.\n\n"
            f"Job ID: {job_id}\n\n"
            f"Please check your input file and try again. "
            f"If the problem persists, contact the ExCALIBR team.\n"
        )

    # Build headers carefully — no user input touches header lines directly
    message = "\r\n".join([
        f"To: {to}",
        "From: ExCALIBR <noreply@excalibr.org>",
        f"Subject: {subject}",
        "Content-Type: text/plain",
        "",
        body,
    ])

    try:
        subprocess.run(
            ["/usr/sbin/sendmail", "-t", "-f", "noreply@excalibr.org"],
            input=message,
            text=True,
            timeout=10,
        )
    except Exception:
        pass

# ── Pipeline runner ───────────────────────────────────────────────────────────

def _run_pipeline(job_id: str, csv_path: Path, output_dir: Path, params: dict, dataset_name: str):
    jobs[job_id]["status"] = "running"
    _write_meta(job_id)

    job_dir = JOBS_DIR / job_id
    progress_file = str(job_dir / "progress.json")
    pid_file = _pid_path(job_id)

    cmd = [
        "nice", "-n", "15",
        str(PYTHON_BIN), str(PIPELINE_SCRIPT),
        "--dataset",            str(csv_path),
        "--name",               dataset_name,
        "--output-dir",         str(output_dir),
        "--n-jobs",             str(N_JOBS_PER_TASK),
        "--progress-file",      progress_file,
        "--components",         *[str(c) for c in params["components"]],
        "--preset",             params["preset"],
        "--benign-method",      params["benignMethod"],
    ]

    postprocess_mode = params.get("postprocessMode", "standard")
    if postprocess_mode == "conservative":
        cmd.append("--conservative-monotonicity")
    elif postprocess_mode == "none":
        cmd.append("--no-postprocess")
    # "standard" -> send neither flag, matches the pipeline's own default

    if params.get("manualPrior") is not None:
        cmd.extend(["--manual-prior", str(params["manualPrior"])])

    if params.get("pathomechanismPrior"):
        cmd.append("--pathomechanism-prior")

    try:
        proc = subprocess.Popen(
            cmd,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.PIPE,
            text=True,
            cwd=str(PIPELINE_DIR),
            start_new_session=True,  # own process group -> can kill the whole tree
        )
        jobs[job_id]["pid"] = proc.pid
        jobs[job_id]["proc"] = proc
        try:
            pid_file.write_text(str(proc.pid))
        except OSError:
            pass

        try:
            _, stderr_output = proc.communicate(timeout=MAX_RUNTIME.total_seconds())
        except subprocess.TimeoutExpired:
            _kill_process_group(proc)
            jobs[job_id]["status"] = "failed"
            jobs[job_id]["error"] = "Job exceeded maximum runtime and was terminated."
            jobs[job_id]["error_internal"] = f"Killed after exceeding MAX_RUNTIME ({MAX_RUNTIME})."
        else:
            if proc.returncode == 0:
                jobs[job_id]["status"] = "complete"
            else:
                jobs[job_id]["status"] = "failed"
                jobs[job_id]["error"] = "Pipeline encountered an error processing your dataset."
                jobs[job_id]["error_internal"] = stderr_output[-4000:]
                logger.error(
                    "job %s: pipeline exited with code %s\n--- stderr ---\n%s",
                    job_id, proc.returncode, stderr_output[-4000:],
                )

    except Exception as e:
        jobs[job_id]["status"] = "failed"
        jobs[job_id]["error"] = "Pipeline encountered an error processing your dataset."
        jobs[job_id]["error_internal"] = str(e)
        logger.error("job %s: exception launching/running pipeline:\n%s", job_id, traceback.format_exc())

    finally:
        jobs[job_id]["pid"] = None
        jobs[job_id].pop("proc", None)
        try:
            pid_file.unlink(missing_ok=True)
        except OSError:
            pass

    _write_meta(job_id)
    _send_email(jobs[job_id].get("email"), job_id, jobs[job_id]["status"])

# ── Cleanup ───────────────────────────────────────────────────────────────────

def _cleanup_old_jobs():
    """
    Remove job directories older than RETENTION_PERIOD.

    Any job still "running" at this point should already have been stopped by
    the MAX_RUNTIME enforcement in _run_pipeline -- but as a safety net, if a
    live Popen handle is still attached to this job, kill its process group
    before removing the directory, so we never orphan a process pointing at
    files that are about to be deleted out from under it.
    """
    cutoff = datetime.utcnow() - RETENTION_PERIOD
    for job_id, meta in list(jobs.items()):
        if meta["created_at"] < cutoff:
            proc = meta.get("proc")
            if proc is not None and proc.poll() is None:
                _kill_process_group(proc)
            pid_file = _pid_path(job_id)
            if pid_file.exists():
                try:
                    _kill_pid_by_number(int(pid_file.read_text().strip()))
                except Exception:
                    pass
            shutil.rmtree(JOBS_DIR / job_id, ignore_errors=True)
            del jobs[job_id]


async def _periodic_cleanup_loop():
    """Runs independently of job submissions, unlike the old approach where
    cleanup only fired via background_tasks inside submit_job -- if submissions
    stopped, so did cleanup, leaving the hourly cron job as the only backstop."""
    while True:
        try:
            _cleanup_old_jobs()
        except Exception:
            pass
        await asyncio.sleep(CLEANUP_INTERVAL_SECONDS)

# ── App ───────────────────────────────────────────────────────────────────────

@asynccontextmanager
async def lifespan(app: FastAPI):
    _recover_jobs_from_disk()
    cleanup_task = asyncio.create_task(_periodic_cleanup_loop())
    try:
        yield
    finally:
        cleanup_task.cancel()
        try:
            await cleanup_task
        except asyncio.CancelledError:
            pass

app = FastAPI(title="ExCALIBR API", docs_url=None, redoc_url=None, openapi_url=None, lifespan=lifespan)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://excalibr.org"],
    allow_methods=["GET", "POST", "OPTIONS"],
    allow_headers=["Content-Type"],
)

# ── Routes ────────────────────────────────────────────────────────────────────

@app.get("/api/thresholds")
def get_thresholds(prior: float):
    """Log-LR+ thresholds (tauP/tauB) for the ±1..±8 evidence tiers at the
    given prior. Job-independent (a pure function of prior + the fixed point
    scale), so this also works for locally-uploaded calibration files that
    have no job_id at all."""
    if not (0 < prior < 1):
        raise HTTPException(status_code=400, detail="prior must be between 0 and 1")

    result = _compute_log_lr_thresholds(prior)
    if result is None:
        raise HTTPException(status_code=500, detail="Threshold computation unavailable")

    tauP, tauB = result
    return {"point_values": POINT_VALUES, "tauP": tauP, "tauB": tauB}


@app.get("/api/presets")
def get_presets():
    return {
        "presets":        {name: {"n_bootstraps": nb, "fits_per_bootstrap": fpb}
                            for name, (nb, fpb) in PRESETS.items()},
        "web_selectable": sorted(ALLOWED_WEB_PRESETS),
    }


@app.post("/api/jobs")
async def submit_job(
    request: Request,
    background_tasks: BackgroundTasks,
    file: UploadFile = File(...),
    params: str = Form(...),
    email: str = Form(default=""),
):
    # Rate limiting — use Cloudflare's real IP header
    client_ip = request.headers.get("CF-Connecting-IP") or request.client.host
    _check_rate_limit(client_ip)

    # Validate email
    safe_email = _validate_email(email)

    # Validate content type
    if file.content_type not in (
        "text/csv", "text/plain", "application/octet-stream", "application/vnd.ms-excel"
    ):
        raise HTTPException(status_code=400, detail="File must be a CSV")

    # Read with size limit
    contents = await file.read(MAX_UPLOAD_BYTES + 1)
    if len(contents) > MAX_UPLOAD_BYTES:
        raise HTTPException(
            status_code=413,
            detail=f"File exceeds {MAX_UPLOAD_BYTES // 1024 // 1024}MB limit"
        )

    # Parse and validate params
    try:
        p = json.loads(params)
    except json.JSONDecodeError:
        raise HTTPException(status_code=400, detail="Invalid params JSON")

    _validate_params(p)

    # Create job directory
    job_id     = str(uuid.uuid4())
    job_dir    = JOBS_DIR / job_id
    output_dir = job_dir / "output"
    job_dir.mkdir()
    output_dir.mkdir()

    # Save CSV
    csv_path = job_dir / "input.csv"
    csv_path.write_bytes(contents)

    # Detect and sanitize dataset name
    try:
        df_head = pd.read_csv(csv_path, nrows=1, low_memory=False)
        if "Dataset" in df_head.columns and not df_head["Dataset"].isna().all():
            dataset_name = _sanitize_name(str(df_head["Dataset"].iloc[0]))
        else:
            dataset_name = job_id
    except Exception:
        dataset_name = job_id

    original_filename = _sanitize_name(file.filename or "input.csv")

    preset = p["preset"]
    preset_n_bootstraps, preset_fits_per_bootstrap = PRESETS[preset]

    jobs[job_id] = {
        "status":            "pending",
        "created_at":        datetime.utcnow(),
        "output_dir":        output_dir,
        "original_filename": original_filename,
        "dataset_name":      dataset_name,
        "email":             safe_email or None,
        "error":             None,
        "error_internal":    None,
        "preset":            preset,
        "n_bootstraps":        preset_n_bootstraps,
        "fits_per_bootstrap":  preset_fits_per_bootstrap,
        "pid":               None,
    }
    _write_meta(job_id)

    executor.submit(_run_pipeline, job_id, csv_path, output_dir, p, dataset_name)
    background_tasks.add_task(_cleanup_old_jobs)

    return {"job_id": job_id}


@app.get("/api/jobs/{job_id}/status")
def get_status(job_id: str):
    if job_id not in jobs:
        raise HTTPException(status_code=404, detail="Job not found")

    meta = jobs[job_id]
    response = {"job_id": job_id, "status": meta["status"]}
    if meta["status"] == "failed" and meta.get("error"):
        response["error"] = meta["error"]  # sanitized message only
    return response


@app.get("/api/jobs/{job_id}/progress")
def get_progress(job_id: str):
    if job_id not in jobs:
        raise HTTPException(status_code=404, detail="Job not found")

    progress_path = JOBS_DIR / job_id / "progress.json"
    if not progress_path.exists():
        return {
            "stage": "pending",
            "stage_label": "Waiting to start…",
            "bootstraps_done": 0,
            "bootstraps_total": 0,
            "percent": 0,
            "elapsed": None,
            "remaining": None,
        }

    try:
        return json.loads(progress_path.read_text())
    except Exception:
        raise HTTPException(status_code=500, detail="Could not read progress")


@app.get("/api/jobs/{job_id}/results")
def get_results(job_id: str):
    if job_id not in jobs:
        raise HTTPException(status_code=404, detail="Job not found")

    meta = jobs[job_id]
    if meta["status"] != "complete":
        raise HTTPException(status_code=409, detail=f"Job is {meta['status']}")

    output_dir   = meta["output_dir"]
    dataset_name = meta["dataset_name"]

    calibration_files = sorted(output_dir.glob(f"{dataset_name}_*c_calibration.json"))
    fits_file = output_dir / f"{dataset_name}_bootstrap_fits.json.gz"
    lr_values_files = sorted(output_dir.glob(f"{dataset_name}_*c_lr_values.json.gz"))

    if not calibration_files or not fits_file.exists():
        raise HTTPException(status_code=500, detail="Expected output files not found")

    return {
        "job_id":      job_id,
        "calibration": [f"/api/jobs/{job_id}/files/{f.name}" for f in calibration_files],
        "fits":        f"/api/jobs/{job_id}/files/{fits_file.name}",
        # Optional -- older jobs run before this field was added won't have it,
        # and the pipeline could in principle skip writing it; frontend should
        # treat an empty list as "no LR+ data available" rather than an error.
        "lr_values":   [f"/api/jobs/{job_id}/lr_values/{f.name}" for f in lr_values_files],
    }


@app.get("/api/jobs/{job_id}/fits")
def get_fits(job_id: str):
    if job_id not in jobs:
        raise HTTPException(status_code=404, detail="Job not found")

    meta = jobs[job_id]
    if meta["status"] != "complete":
        raise HTTPException(status_code=409, detail=f"Job is {meta['status']}")

    dataset_name = meta["dataset_name"]
    fits_file = meta["output_dir"] / f"{dataset_name}_bootstrap_fits.json.gz"

    if not fits_file.exists():
        raise HTTPException(status_code=404, detail="Fits file not found")

    try:
        with gzip.open(fits_file, "rt", encoding="utf-8") as f:
            content = f.read()
    except Exception:
        raise HTTPException(status_code=500, detail="Could not read fits file")

    return Response(content=content, media_type="application/json")


@app.get("/api/jobs/{job_id}/lr_values/{filename}")
def get_lr_values(job_id: str, filename: str):
    """Decompresses one component's *_lr_values.json.gz file, mirroring get_fits.
    Takes a filename (not just job_id) since a job may have one file per fitted
    component (2c/3c/4c) -- caller picks which one via the results manifest."""
    if job_id not in jobs:
        raise HTTPException(status_code=404, detail="Job not found")

    if not re.match(r'^[a-zA-Z0-9_\-. ]+$', filename):
        raise HTTPException(status_code=400, detail="Invalid filename")

    meta = jobs[job_id]
    if meta["status"] != "complete":
        raise HTTPException(status_code=409, detail=f"Job is {meta['status']}")

    lr_file = (meta["output_dir"] / filename).resolve()
    if not str(lr_file).startswith(str(meta["output_dir"].resolve())):
        raise HTTPException(status_code=400, detail="Invalid filename")

    if not lr_file.exists():
        raise HTTPException(status_code=404, detail="LR+ values file not found")

    try:
        with gzip.open(lr_file, "rt", encoding="utf-8") as f:
            content = f.read()
    except Exception:
        raise HTTPException(status_code=500, detail="Could not read LR+ values file")

    return Response(content=content, media_type="application/json")


@app.get("/api/jobs/{job_id}/csv")
def get_csv(job_id: str):
    if job_id not in jobs:
        raise HTTPException(status_code=404, detail="Job not found")

    csv_path = JOBS_DIR / job_id / "input.csv"
    if not csv_path.exists():
        raise HTTPException(status_code=404, detail="CSV not found")

    return FileResponse(path=str(csv_path), filename="input.csv", media_type="text/csv")


@app.get("/api/jobs/{job_id}/files/{filename}")
def get_file(job_id: str, filename: str):
    if job_id not in jobs:
        raise HTTPException(status_code=404, detail="Job not found")

    # Validate filename — only allow safe characters
    if not re.match(r'^[a-zA-Z0-9_\-. ]+$', filename):
        raise HTTPException(status_code=400, detail="Invalid filename")

    output_dir = jobs[job_id]["output_dir"]
    file_path  = (output_dir / filename).resolve()

    # Prevent path traversal
    if not str(file_path).startswith(str(output_dir.resolve())):
        raise HTTPException(status_code=400, detail="Invalid filename")

    if not file_path.exists():
        raise HTTPException(status_code=404, detail="File not found")

    return FileResponse(path=str(file_path), filename=filename)


@app.get("/api/jobs/{job_id}/meta")
def get_meta(job_id: str):
    if job_id not in jobs:
        raise HTTPException(status_code=404, detail="Job not found")
    meta = jobs[job_id]
    return {
        "job_id":            job_id,
        "original_filename": meta.get("original_filename", "input.csv"),
        "dataset_name":      meta["dataset_name"],
        "preset":            meta.get("preset"),
        "n_bootstraps":      meta.get("n_bootstraps"),
        "fits_per_bootstrap": meta.get("fits_per_bootstrap"),
    }


# ── CLI entry point ────────────────────────────────────────────────────────────
#
# Lets the cron-side cleanup script read RETENTION_PERIOD from this file directly,
# so the cron sweep and the in-app enforcement can never drift out of sync.
# Usage: python api.py --retention-minutes

if __name__ == "__main__":
    if len(sys.argv) > 1 and sys.argv[1] == "--retention-minutes":
        print(int(RETENTION_PERIOD.total_seconds() // 60))
        raise SystemExit(0)