ligbox-ops-platform/projects/ops-desk/api/app/vm112_purge_jobs.py

"""Purge assíncrono com polling + persistência SQLite (Spec 017 Fase 2b/3)."""

from __future__ import annotations

import json
import sqlite3
import threading
import time
import traceback
import uuid
from datetime import datetime, timezone
from typing import Any, Callable

from app import auth, vm112_domains

_lock = threading.Lock()
_schema_ready = False


class PurgeJobConflictError(Exception):
    """Já existe purge queued/running para o domínio."""


def _now() -> str:
    return datetime.now(timezone.utc).isoformat()


def init_purge_jobs_schema(conn) -> None:
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS vm112_purge_jobs (
            id TEXT PRIMARY KEY,
            domain TEXT NOT NULL,
            status TEXT NOT NULL,
            timeline_json TEXT NOT NULL DEFAULT '[]',
            elapsed_vm112 INTEGER NOT NULL DEFAULT 0,
            desk_json TEXT NOT NULL DEFAULT '{}',
            vm112_json TEXT NOT NULL DEFAULT '{}',
            error TEXT,
            by_user TEXT,
            created_at TEXT NOT NULL,
            updated_at TEXT NOT NULL
        )
        """
    )
    conn.execute(
        """
        CREATE INDEX IF NOT EXISTS idx_vm112_purge_jobs_domain_status
        ON vm112_purge_jobs(domain, status)
        """
    )
    conn.commit()


def _ensure_schema() -> None:
    global _schema_ready
    if _schema_ready:
        return
    conn = auth.db()
    try:
        init_purge_jobs_schema(conn)
        _schema_ready = True
    finally:
        conn.close()


def _sqlite_retry(fn: Callable[[], Any], *, attempts: int = 8) -> Any:
    last_err: Exception | None = None
    for attempt in range(attempts):
        try:
            return fn()
        except sqlite3.OperationalError as exc:
            last_err = exc
            if "locked" not in str(exc).lower() or attempt >= attempts - 1:
                raise
            time.sleep(min(0.05 * (2**attempt), 1.0))
    if last_err:
        raise last_err
    return None


def _row_to_job(row) -> dict[str, Any]:
    return {
        "id": row["id"],
        "job_id": row["id"],
        "domain": row["domain"],
        "status": row["status"],
        "timeline": json.loads(row["timeline_json"] or "[]"),
        "elapsed_vm112": int(row["elapsed_vm112"] or 0),
        "desk": json.loads(row["desk_json"] or "{}"),
        "vm112": json.loads(row["vm112_json"] or "{}"),
        "error": row["error"],
        "by": row["by_user"],
        "created_at": row["created_at"],
        "updated_at": row["updated_at"],
    }


def _persist_job(job: dict[str, Any]) -> None:
    _ensure_schema()

    def _write() -> None:
        conn = auth.db()
        try:
            job["updated_at"] = _now()
            conn.execute(
                """
                INSERT INTO vm112_purge_jobs (
                    id, domain, status, timeline_json, elapsed_vm112,
                    desk_json, vm112_json, error, by_user, created_at, updated_at
                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                ON CONFLICT(id) DO UPDATE SET
                    status = excluded.status,
                    timeline_json = excluded.timeline_json,
                    elapsed_vm112 = excluded.elapsed_vm112,
                    desk_json = excluded.desk_json,
                    vm112_json = excluded.vm112_json,
                    error = excluded.error,
                    by_user = excluded.by_user,
                    updated_at = excluded.updated_at
                """,
                (
                    job["id"],
                    job["domain"],
                    job["status"],
                    json.dumps(job.get("timeline") or [], ensure_ascii=False),
                    int(job.get("elapsed_vm112") or 0),
                    json.dumps(job.get("desk") or {}, ensure_ascii=False),
                    json.dumps(job.get("vm112") or {}, ensure_ascii=False),
                    job.get("error"),
                    job.get("by"),
                    job.get("created_at") or _now(),
                    job["updated_at"],
                ),
            )
            conn.commit()
        finally:
            conn.close()

    _sqlite_retry(_write)


def _load_job(job_id: str) -> dict[str, Any] | None:
    _ensure_schema()

    def _read() -> dict[str, Any] | None:
        conn = auth.db()
        try:
            row = conn.execute(
                "SELECT * FROM vm112_purge_jobs WHERE id = ?", (job_id,)
            ).fetchone()
            return _row_to_job(row) if row else None
        finally:
            conn.close()

    return _sqlite_retry(_read)


def _find_active_job_locked(conn: sqlite3.Connection, domain: str) -> sqlite3.Row | None:
    return conn.execute(
        """
        SELECT id, domain, status, created_at FROM vm112_purge_jobs
        WHERE domain = ? AND status IN ('queued', 'running')
        ORDER BY created_at DESC LIMIT 1
        """,
        (domain.lower().strip(),),
    ).fetchone()


def _mutate_job(job_id: str, fn: Callable[[dict[str, Any]], None]) -> dict[str, Any] | None:
    with _lock:
        def _mutate() -> dict[str, Any] | None:
            conn = auth.db()
            try:
                row = conn.execute(
                    "SELECT * FROM vm112_purge_jobs WHERE id = ?", (job_id,)
                ).fetchone()
                if not row:
                    return None
                job = _row_to_job(row)
                fn(job)
                job["updated_at"] = _now()
                conn.execute(
                    """
                    INSERT INTO vm112_purge_jobs (
                        id, domain, status, timeline_json, elapsed_vm112,
                        desk_json, vm112_json, error, by_user, created_at, updated_at
                    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                    ON CONFLICT(id) DO UPDATE SET
                        status = excluded.status,
                        timeline_json = excluded.timeline_json,
                        elapsed_vm112 = excluded.elapsed_vm112,
                        desk_json = excluded.desk_json,
                        vm112_json = excluded.vm112_json,
                        error = excluded.error,
                        by_user = excluded.by_user,
                        updated_at = excluded.updated_at
                    """,
                    (
                        job["id"],
                        job["domain"],
                        job["status"],
                        json.dumps(job.get("timeline") or [], ensure_ascii=False),
                        int(job.get("elapsed_vm112") or 0),
                        json.dumps(job.get("desk") or {}, ensure_ascii=False),
                        json.dumps(job.get("vm112") or {}, ensure_ascii=False),
                        job.get("error"),
                        job.get("by"),
                        job.get("created_at") or _now(),
                        job["updated_at"],
                    ),
                )
                conn.commit()
                return dict(job)
            finally:
                conn.close()

        return _sqlite_retry(_mutate)


def _upsert_step(job_id: str, step: dict[str, str]) -> None:
    def _apply(job: dict[str, Any]) -> None:
        timeline: list[dict[str, str]] = job.setdefault("timeline", [])
        for i, existing in enumerate(timeline):
            if existing.get("label") == step.get("label"):
                timeline[i] = step
                return
        timeline.append(step)

    _mutate_job(job_id, _apply)


def _set_job(job_id: str, **fields: Any) -> None:
    _mutate_job(job_id, lambda job: job.update(fields))


def create_job(domain: str, username: str) -> str:
    domain = domain.lower().strip()
    job_id = uuid.uuid4().hex[:16]
    now = _now()
    job = {
        "id": job_id,
        "job_id": job_id,
        "domain": domain,
        "status": "queued",
        "timeline": [],
        "elapsed_vm112": 0,
        "desk": {},
        "vm112": {},
        "error": None,
        "by": username,
        "created_at": now,
        "updated_at": now,
    }

    def _create() -> None:
        conn = auth.db()
        try:
            active = _find_active_job_locked(conn, domain)
            if active:
                raise PurgeJobConflictError(
                    f"Purge já em curso para {domain} (job {active['id']}, status {active['status']})"
                )
            conn.execute(
                """
                INSERT INTO vm112_purge_jobs (
                    id, domain, status, timeline_json, elapsed_vm112,
                    desk_json, vm112_json, error, by_user, created_at, updated_at
                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                """,
                (
                    job["id"],
                    job["domain"],
                    job["status"],
                    "[]",
                    0,
                    "{}",
                    "{}",
                    None,
                    job["by"],
                    job["created_at"],
                    job["updated_at"],
                ),
            )
            conn.commit()
        finally:
            conn.close()

    with _lock:
        _ensure_schema()
        _sqlite_retry(_create)
    return job_id


def start_job(domain: str, root_password: str, username: str) -> str:
    job_id = create_job(domain, username)
    thread = threading.Thread(
        target=_execute_job,
        args=(job_id, domain, root_password, username),
        daemon=True,
    )
    thread.start()
    return job_id


def _desk_already_done(job: dict[str, Any]) -> bool:
    for step in job.get("timeline") or []:
        if str(step.get("label") or "") == "Purge concluído" and step.get("status") == "ok":
            return True
    return False


def _finish_desk_phase(job_id: str) -> dict[str, Any] | None:
    job = _load_job(job_id)
    if not job:
        return None
    if _desk_already_done(job):
        if job["status"] != "done":
            _set_job(job_id, status="done")
        return _load_job(job_id)

    domain = job["domain"]
    conn = auth.db()
    try:
        desk_counts, desk_timeline = vm112_domains.purge_desk_timeline(conn, domain)
    finally:
        conn.close()

    for step in desk_timeline:
        _upsert_step(job_id, step)

    total_desk = sum(desk_counts.values())
    _upsert_step(
        job_id,
        vm112_domains._timeline_entry("Purge concluído", "ok", f"Desk: {total_desk} registo(s)"),
    )
    _set_job(job_id, status="done", desk=desk_counts)
    return _load_job(job_id)


def recover_job(job_id: str, domain: str | None = None) -> dict[str, Any] | None:
    """Finaliza job quando VM112 já removeu o domínio (ex.: API reiniciada)."""
    job = _load_job(job_id)
    if not job:
        if not domain:
            return None
        domain = domain.lower().strip()
        if vm112_domains.domain_exists_on_vm112(domain):
            return None
        conn = auth.db()
        try:
            desk_counts, desk_timeline = vm112_domains.purge_desk_timeline(conn, domain)
        finally:
            conn.close()
        total_desk = sum(desk_counts.values())
        timeline = [
            vm112_domains._timeline_entry(
                "Purge recuperado",
                "ok",
                "Domínio já ausente na VM112",
            ),
            *desk_timeline,
            vm112_domains._timeline_entry("Purge concluído", "ok", f"Desk: {total_desk} registo(s)"),
        ]
        return {
            "id": job_id,
            "job_id": job_id,
            "domain": domain,
            "status": "done",
            "timeline": timeline,
            "elapsed_vm112": 0,
            "desk": desk_counts,
            "vm112": {"ok": True, "recovered": True},
            "error": None,
            "by": None,
        }

    if job["status"] in ("done", "error"):
        return job

    domain = (domain or job["domain"]).lower().strip()
    if vm112_domains.domain_exists_on_vm112(domain):
        return job

    _upsert_step(
        job_id,
        vm112_domains._timeline_entry(
            "Purge VM112",
            "ok",
            "Domínio já removido na VM112 (recuperação)",
        ),
    )
    return _finish_desk_phase(job_id)


def _execute_job(job_id: str, domain: str, root_password: str, username: str) -> None:
    domain = domain.lower().strip()
    try:
        _set_job(job_id, status="running")

        conn = auth.db()
        try:
            if not vm112_domains.verify_root_password(conn, root_password):
                step = vm112_domains._timeline_entry("Validação Root", "fail", "Senha Root incorrecta")
                _upsert_step(job_id, step)
                _set_job(job_id, status="error", error="Senha Root incorrecta")
                return
        finally:
            conn.close()

        _upsert_step(job_id, vm112_domains._timeline_entry("Validação Root + confirmação", "ok"))
        _upsert_step(
            job_id,
            vm112_domains._timeline_entry(
                "Purge VM112 — em execução",
                "running",
                "Carbonio, site, portal, Cloudflare, Traefik…",
            ),
        )

        vm112_result: dict[str, Any] = {"ok": False}
        vm112_banner_marked = False
        for kind, payload in vm112_domains.purge_vm112_with_poll(domain):
            if kind == "step":
                if not vm112_banner_marked:
                    _upsert_step(
                        job_id,
                        vm112_domains._timeline_entry(
                            "Purge VM112 — em execução", "ok", "Passos abaixo",
                        ),
                    )
                    vm112_banner_marked = True
                _upsert_step(job_id, payload)
            elif kind == "heartbeat":
                _set_job(job_id, elapsed_vm112=int(payload.get("elapsed") or 0))
            elif kind == "final":
                vm112_result = payload
                break

        if not vm112_result.get("ok", False):
            step = vm112_domains._timeline_entry(
                "Purge VM112",
                "fail",
                str(vm112_result.get("error") or "falhou"),
            )
            _upsert_step(job_id, step)
            _set_job(job_id, status="error", error=str(vm112_result.get("error") or "falhou"))
            return

        _set_job(job_id, vm112=vm112_result)
        _finish_desk_phase(job_id)
    except Exception as exc:
        err = str(exc) or "erro inesperado"
        _upsert_step(
            job_id,
            vm112_domains._timeline_entry("Purge VM112", "fail", err),
        )
        _set_job(job_id, status="error", error=err)
        traceback.print_exc()


def get_job_public(job_id: str) -> dict[str, Any] | None:
    job = _load_job(job_id)
    if not job:
        return None
    if job["status"] == "running":
        try:
            if not vm112_domains.domain_exists_on_vm112(job["domain"]):
                job = recover_job(job_id) or job
        except Exception:
            pass
    return job


def list_jobs(limit: int = 100, offset: int = 0) -> dict[str, Any]:
    _ensure_schema()
    limit = max(1, min(int(limit), 500))
    offset = max(0, int(offset))
    conn = auth.db()
    try:
        total = conn.execute("SELECT COUNT(*) FROM vm112_purge_jobs").fetchone()[0]
        rows = conn.execute(
            """
            SELECT id, domain, status, by_user, created_at, updated_at,
                   elapsed_vm112, desk_json, error
            FROM vm112_purge_jobs
            ORDER BY created_at DESC
            LIMIT ? OFFSET ?
            """,
            (limit, offset),
        ).fetchall()
        jobs = []
        for row in rows:
            desk = json.loads(row["desk_json"] or "{}")
            desk_total = sum(int(v or 0) for v in desk.values())
            jobs.append(
                {
                    "id": row["id"],
                    "job_id": row["id"],
                    "domain": row["domain"],
                    "status": row["status"],
                    "by": row["by_user"],
                    "created_at": row["created_at"],
                    "updated_at": row["updated_at"],
                    "elapsed_vm112": int(row["elapsed_vm112"] or 0),
                    "desk": desk,
                    "desk_removed_total": desk_total,
                    "error": row["error"],
                }
            )
        return {"jobs": jobs, "total": int(total), "limit": limit, "offset": offset}
    finally:
        conn.close()