api/routes/deployments.py

"""
Production deployment jobs for the Data explorer React UI.

A second *kind* of batch job alongside the YouTube pipeline (``/api/batch-jobs``):
a multi-step **prod deployment** — database (Neon prod) then web (HuggingFace).

The actual work runs in a detached orchestrator
(``packages/hosting/src/hosting/deploy/run_deployment.py``) launched here; it
writes per-step status + logs under ``data/cache/batch_jobs/deployments/`` which
these endpoints read back. The orchestrator is launched **by file path** (not
``-m``) so it works even though the hosting package isn't installed in the API
venv.

Safety: real (non-dry-run) deploys are gated behind ``DEPLOYMENTS_ALLOW_LAUNCH``
(default off). Dry-run launches — which never execute a command — are always
allowed so the dashboard can be exercised without touching prod.
"""
from __future__ import annotations

import json
import os
import signal
import subprocess
import sys
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, List, Optional

from fastapi import APIRouter, HTTPException, Query
from loguru import logger
from pydantic import BaseModel, Field

router = APIRouter(prefix="/deployments", tags=["deployments"])

# Step metadata is duplicated (lightly) from hosting.deploy.run_deployment so the
# API need not import the hosting package (not installed in the API venv). The
# orchestrator remains the source of truth for the argv that actually runs.
# Keep in sync with hosting.deploy.run_deployment.STEP_DEFS (source of truth for
# the argv). A prod deployment is just a DATA COPY of local public serving objects
# into Neon prod public (runtime/auth tables excluded). ``web`` (HuggingFace) is a
# separate, optional step — selectable in the panel but not run by default.
_STEP_DEFS: List[Dict[str, str]] = [
    {
        "key": "database",
        "label": "Database — serving copy (Neon prod)",
        "description": "Copy local public serving objects → Neon prod public (excludes runtime/auth tables).",
        "target": "neon-prod",
    },
    {
        "key": "web",
        "label": "Web (HuggingFace)",
        "description": "Build & deploy the web app + docs to HuggingFace Spaces.",
        "target": "huggingface",
    },
]
_STEP_KEYS = [d["key"] for d in _STEP_DEFS]
_DEFAULT_STEPS = ["database"]


def _repo_root() -> Path:
    # api/routes/deployments.py -> repo root
    return Path(__file__).resolve().parents[2]


def _status_dir() -> Path:
    return _repo_root() / "data" / "cache" / "batch_jobs" / "deployments"


def _orchestrator_path() -> Path:
    return (
        _repo_root()
        / "packages"
        / "hosting"
        / "src"
        / "hosting"
        / "deploy"
        / "run_deployment.py"
    )


def _deploy_enabled() -> bool:
    # Opt-IN switch for real deploys — prod is destructive/outward-facing.
    return os.getenv("DEPLOYMENTS_ALLOW_LAUNCH", "0").strip().lower() in (
        "1",
        "true",
        "yes",
        "on",
    )


def _pid_alive(pid: Any) -> bool:
    try:
        p = int(pid)
    except (TypeError, ValueError):
        return False
    try:
        os.kill(p, 0)
    except OSError:
        return False
    try:
        with open(f"/proc/{p}/stat") as fh:
            state = fh.read().split(") ", 1)[1][:1]
        if state in ("Z", "X"):
            try:
                os.waitpid(p, os.WNOHANG)
            except (ChildProcessError, OSError):
                pass
            return False
    except (FileNotFoundError, IndexError):
        pass
    return True


# --------------------------------------------------------------------------- #
# Models
# --------------------------------------------------------------------------- #
class DeploymentStepModel(BaseModel):
    key: str
    label: str = ""
    description: str = ""
    target: str = ""
    status: str = "pending"
    started_at: Optional[str] = None
    finished_at: Optional[str] = None
    exit_code: Optional[int] = None
    note: Optional[str] = None
    log: str = ""
    cmd: str = ""


class DeploymentJobModel(BaseModel):
    job_id: str
    job_type: str = "deployment"
    label: str = ""
    dry_run: bool = False
    pid: Optional[int] = None
    status: str = "running"
    started_at: Optional[str] = None
    updated_at: Optional[str] = None
    finished_at: Optional[str] = None
    steps: List[DeploymentStepModel] = Field(default_factory=list)
    # ``True`` while the orchestrator process is still alive (computed, not stored).
    live: bool = False


class StepDefModel(BaseModel):
    key: str
    label: str
    description: str
    target: str


class DeploymentsListResponse(BaseModel):
    jobs: List[DeploymentJobModel] = Field(default_factory=list)
    available_steps: List[StepDefModel] = Field(default_factory=list)
    enabled: bool = False


class LaunchDeploymentRequest(BaseModel):
    steps: List[str] = Field(default_factory=lambda: list(_DEFAULT_STEPS))
    # Default to a dry run — a real prod deploy must be opted into explicitly.
    dry_run: bool = True


class LaunchDeploymentResponse(BaseModel):
    launched: bool
    job_id: str = ""
    pid: Optional[int] = None
    dry_run: bool = True
    steps: List[str] = Field(default_factory=list)
    detail: str = ""


class StopDeploymentResponse(BaseModel):
    stopped: bool = False
    detail: str = ""


class DeploymentLogResponse(BaseModel):
    job_id: str
    step: str
    path: str = ""
    lines: List[str] = Field(default_factory=list)


# --------------------------------------------------------------------------- #
# Helpers
# --------------------------------------------------------------------------- #
def _read_job(path: Path) -> Optional[DeploymentJobModel]:
    try:
        data = json.loads(path.read_text())
    except (OSError, ValueError):
        return None
    if not isinstance(data, dict) or not data.get("job_id"):
        return None
    pid = data.get("pid")
    live = _pid_alive(pid)
    # A job whose orchestrator died without finishing is stale — surface it as
    # failed so the dashboard doesn't pin it at "running" forever.
    status = str(data.get("status") or "")
    if status == "running" and not live:
        status = "failed"
        data["status"] = status
    try:
        job = DeploymentJobModel(**data)
    except Exception:
        return None
    job.live = live
    return job


def _list_jobs() -> List[DeploymentJobModel]:
    d = _status_dir()
    if not d.is_dir():
        return []
    jobs: List[DeploymentJobModel] = []
    for path in d.glob("*.json"):
        job = _read_job(path)
        if job is not None:
            jobs.append(job)
    jobs.sort(key=lambda j: j.started_at or "", reverse=True)
    return jobs


def _job_path(job_id: str) -> Path:
    return _status_dir() / f"{job_id}.json"


def _write_initial_status(
    status_dir: Path, job_id: str, steps: List[str], dry_run: bool, pid: int
) -> None:
    """Seed a ``running`` status file so the job is visible before the detached
    orchestrator boots and writes its own (which then takes over)."""
    by_key = {d["key"]: d for d in _STEP_DEFS}
    now = datetime.now(timezone.utc).isoformat()
    data = {
        "job_id": job_id,
        "job_type": "deployment",
        "label": "Prod deployment" + (" (dry run)" if dry_run else ""),
        "dry_run": dry_run,
        "pid": pid,
        "status": "running",
        "started_at": now,
        "updated_at": now,
        "finished_at": None,
        "steps": [
            {
                "key": k,
                "label": by_key.get(k, {}).get("label", k),
                "description": by_key.get(k, {}).get("description", ""),
                "target": by_key.get(k, {}).get("target", ""),
                "status": "pending",
                "started_at": None,
                "finished_at": None,
                "exit_code": None,
                "log": f"data/cache/batch_jobs/deployments/{job_id}_{k}.log",
                "cmd": "",
            }
            for k in steps
        ],
    }
    try:
        dest = status_dir / f"{job_id}.json"
        tmp = dest.with_suffix(".json.tmp")
        tmp.write_text(json.dumps(data, indent=2))
        os.replace(tmp, dest)
    except Exception:
        pass


# --------------------------------------------------------------------------- #
# Endpoints
# --------------------------------------------------------------------------- #
@router.get("/", response_model=DeploymentsListResponse)
async def list_deployments(limit: int = Query(25, ge=1, le=200)) -> DeploymentsListResponse:
    jobs = _list_jobs()[:limit]
    return DeploymentsListResponse(
        jobs=jobs,
        available_steps=[StepDefModel(**d) for d in _STEP_DEFS],
        enabled=_deploy_enabled(),
    )


@router.get("/{job_id}", response_model=DeploymentJobModel)
async def get_deployment(job_id: str) -> DeploymentJobModel:
    job = _read_job(_job_path(job_id))
    if job is None:
        raise HTTPException(status_code=404, detail=f"deployment job not found: {job_id}")
    return job


@router.post("/launch", response_model=LaunchDeploymentResponse)
async def launch_deployment(req: LaunchDeploymentRequest) -> LaunchDeploymentResponse:
    steps = [s.strip().lower() for s in (req.steps or []) if s and s.strip()]
    if not steps:
        steps = list(_DEFAULT_STEPS)
    bad = [s for s in steps if s not in _STEP_KEYS]
    if bad:
        raise HTTPException(status_code=400, detail=f"unknown step(s): {bad}; known: {_STEP_KEYS}")

    dry_run = bool(req.dry_run)
    if not dry_run and not _deploy_enabled():
        raise HTTPException(
            status_code=403,
            detail="Real deploys are disabled. Set DEPLOYMENTS_ALLOW_LAUNCH=1 to enable, "
            "or run with dry_run=true.",
        )

    # Refuse to start while another deployment is still live.
    for j in _list_jobs():
        if j.live and j.status == "running":
            raise HTTPException(
                status_code=409,
                detail=f"A deployment is already running ({j.job_id}). Stop it first.",
            )

    script = _orchestrator_path()
    if not script.is_file():
        raise HTTPException(status_code=500, detail=f"orchestrator missing: {script}")

    stamp = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")
    job_id = f"deploy_{stamp}"
    status_dir = _status_dir()
    status_dir.mkdir(parents=True, exist_ok=True)

    argv = [
        sys.executable,
        str(script),
        "--job-id",
        job_id,
        "--steps",
        ",".join(steps),
        "--status-dir",
        str(status_dir),
    ]
    if dry_run:
        argv.append("--dry-run")

    # The orchestrator owns its own per-step logs; capture its own stdout/stderr
    # (launch/argparse errors) to a bootstrap log next to the status file.
    boot_log = status_dir / f"{job_id}_orchestrator.log"
    try:
        logf = open(boot_log, "ab")
        proc = subprocess.Popen(
            argv,
            cwd=str(_repo_root()),
            env=dict(os.environ),
            stdout=logf,
            stderr=subprocess.STDOUT,
            stdin=subprocess.DEVNULL,
            start_new_session=True,  # detach so it outlives the request
        )
    except Exception as exc:
        logger.exception("deployment launch failed")
        raise HTTPException(status_code=500, detail=f"launch failed: {exc}") from exc

    # Write an initial status file so the job appears immediately (the
    # orchestrator atomically overwrites it once it boots — last writer wins).
    _write_initial_status(status_dir, job_id, steps, dry_run, proc.pid)

    logger.info(
        f"launched deployment job_id={job_id} steps={steps} dry_run={dry_run} pid={proc.pid}"
    )
    detail = (
        f"Started dry-run deployment '{job_id}' (pid {proc.pid}). No commands were executed."
        if dry_run
        else f"Started PROD deployment '{job_id}' (pid {proc.pid}). The dashboard updates as it runs."
    )
    return LaunchDeploymentResponse(
        launched=True,
        job_id=job_id,
        pid=proc.pid,
        dry_run=dry_run,
        steps=steps,
        detail=detail,
    )


@router.post("/{job_id}/stop", response_model=StopDeploymentResponse)
async def stop_deployment(job_id: str, force: bool = Query(False)) -> StopDeploymentResponse:
    job = _read_job(_job_path(job_id))
    if job is None:
        raise HTTPException(status_code=404, detail=f"deployment job not found: {job_id}")
    if not job.live or job.pid is None:
        return StopDeploymentResponse(detail=f"Deployment '{job_id}' is not running.")
    sig = signal.SIGKILL if force else signal.SIGTERM
    try:
        os.killpg(os.getpgid(int(job.pid)), sig)
    except Exception:
        try:
            os.kill(int(job.pid), sig)
        except Exception as exc:
            raise HTTPException(status_code=500, detail=f"stop failed: {exc}") from exc
    how = "SIGKILL" if force else "SIGTERM"
    logger.info(f"stopped deployment {job_id} via {how} (pid {job.pid})")
    return StopDeploymentResponse(
        stopped=True,
        detail=f"Stopped deployment '{job_id}'. The dashboard will update as it exits.",
    )


@router.get("/{job_id}/log", response_model=DeploymentLogResponse)
async def deployment_log(
    job_id: str,
    step: str = Query(..., description="Step key whose log to tail."),
    lines: int = Query(200, ge=1, le=2000),
) -> DeploymentLogResponse:
    step = (step or "").strip().lower()
    if step not in _STEP_KEYS:
        raise HTTPException(status_code=400, detail=f"step must be one of {_STEP_KEYS}")
    path = _status_dir() / f"{job_id}_{step}.log"
    if not path.is_file():
        return DeploymentLogResponse(job_id=job_id, step=step)
    try:
        size = path.stat().st_size
        with open(path, "rb") as fh:
            if size > 262144:
                fh.seek(-262144, 2)
            data = fh.read()
        tail = data.decode("utf-8", "replace").splitlines()[-lines:]
    except Exception:
        tail = []
    return DeploymentLogResponse(
        job_id=job_id,
        step=step,
        path=str(path.relative_to(_repo_root())),
        lines=tail,
    )