"""FC-3h: first-class backup/restore service for FC.

Two independent backup kinds:
  - 'db'     — pg_dump only; fast; nightly via Beat (settings-gated)
  - 'images' — tar+zstd of /images; slow; manual trigger only

Files live under <images_root>/_backups/. Each backup writes:
  fc_<kind>_<ts>.{sql|tar.zst}        — the artifact
  fc_<kind>_<ts>.json                 — manifest (kind/tag/triggered_by)

Service functions are sync (subprocess-bound). Celery tasks in
backend.app.tasks.backup wrap each one with task_run-tracked
lifecycle + soft/hard time limits + retention bookkeeping.
"""
from __future__ import annotations

import json
import os
import shutil
import subprocess
import tempfile
from datetime import UTC, datetime
from pathlib import Path

_BACKUPS_DIRNAME = "_backups"

# Subprocess-level guardrails BEYOND the Celery soft_time_limit. The Celery
# soft limit signals the Python process; subprocess.Popen in a blocking syscall
# ignores that signal, so these bound the worst case directly. Each sits just
# UNDER its task's Celery soft_time_limit so the bounded-kill (_run_bounded) is
# the primary guard and fires cleanly before Celery's soft/hard limits — which
# matters because an NFS D-state hang defeats even Celery's SIGKILL (the failure
# that wedged the maintenance lane for hours, #739).
#   backup_db_task:     soft=1800s / hard=2100s  → 1700s
#   backup_images_task: soft=21600s / hard=23400s → 21000s
_DB_SUBPROCESS_TIMEOUT_S = 1700            # ~28 min, under the 30-min DB soft limit
_IMAGES_SUBPROCESS_TIMEOUT_S = 21000       # ~5.8 hr, under the 6-hr images soft limit
# Grace after SIGKILL to reap the child. If it can't be reaped in this window
# (an uninterruptible NFS D-state — the failure mode that wedged the
# concurrency-1 maintenance lane for hours, operator-flagged 2026-06-07), we
# STOP waiting and fail fast, freeing the worker slot. The orphan is reaped by
# the OS once its blocking syscall clears.
_KILL_REAP_GRACE_S = 10


def _run_bounded(cmd: list[str], timeout: int) -> None:
    """subprocess.run(check=True, timeout) whose reaper can't itself hang.

    subprocess.run's timeout path SIGKILLs the child then blocks in wait() to
    reap it — but a process stuck in uninterruptible I/O (NFS) can't be reaped,
    so wait() blocks for hours. Here we bound the post-kill reap and re-raise
    TimeoutExpired regardless, so the caller fails fast instead of wedging."""
    proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    try:
        out, err = proc.communicate(timeout=timeout)
    except subprocess.TimeoutExpired:
        proc.kill()
        try:
            proc.communicate(timeout=_KILL_REAP_GRACE_S)
        except subprocess.TimeoutExpired:
            pass  # unkillable (D-state) — abandon the reap, fail fast
        raise
    if proc.returncode != 0:
        raise subprocess.CalledProcessError(
            proc.returncode, cmd, output=out, stderr=err
        )


def _libpq_url(sa_url: str) -> str:
    """Strip SQLAlchemy +psycopg/+asyncpg driver suffix for pg_dump/psql."""
    for driver in (
        "postgresql+psycopg",
        "postgresql+asyncpg",
        "postgresql+psycopg2",
    ):
        if sa_url.startswith(driver + "://"):
            return "postgresql://" + sa_url[len(driver) + 3:]
    return sa_url


def _backups_dir(images_root: Path) -> Path:
    p = images_root / _BACKUPS_DIRNAME
    p.mkdir(parents=True, exist_ok=True)
    return p


def _now_ts() -> str:
    return datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ")


def _file_size_or_none(path: Path) -> int | None:
    try:
        return path.stat().st_size
    except OSError:
        return None


def _write_manifest(
    out_dir: Path, *, kind: str, ts: str,
    tag: str | None, triggered_by: str,
    artifact_path: Path,
) -> Path:
    manifest = {
        "kind": kind,
        "backup_id": f"fc_{kind}_{ts}",
        "tag": tag,
        "triggered_by": triggered_by,
        "created_at": datetime.now(UTC).isoformat(),
        "artifact_path": str(artifact_path),
    }
    mf = out_dir / f"fc_{kind}_{ts}.json"
    mf.write_text(json.dumps(manifest, indent=2))
    return mf


def backup_db(
    *, db_url: str, images_root: Path,
    tag: str | None = None, triggered_by: str = "manual",
) -> dict:
    """Run pg_dump; write .sql + manifest; return dict for the caller
    to persist into BackupRun. Raises on subprocess failure."""
    ts = _now_ts()
    out_dir = _backups_dir(images_root)
    # Custom format (-Fc): compressed (much smaller on NFS) and restored with
    # pg_restore. The .dump extension marks it as non-SQL. The BackupRun field
    # is still named sql_path — it's just "the db artifact path".
    sql_path = out_dir / f"fc_db_{ts}.dump"
    # Dump to LOCAL disk first, then move the finished file to the (NFS) backups
    # dir. pg_dump's long phase is then a DB-socket wait + local writes — both
    # killable — instead of an NFS write that can hang uninterruptibly. Only the
    # final move touches NFS, and it's a bounded single-file step.
    fd, tmp_name = tempfile.mkstemp(prefix="fc_db_", suffix=".dump")
    os.close(fd)
    tmp_path = Path(tmp_name)
    try:
        _run_bounded(
            [
                "pg_dump", "--no-owner", "--no-acl", "-Fc",
                "-f", str(tmp_path), _libpq_url(db_url),
            ],
            _DB_SUBPROCESS_TIMEOUT_S,
        )
        shutil.move(str(tmp_path), str(sql_path))
    finally:
        if tmp_path.exists():
            tmp_path.unlink(missing_ok=True)
    manifest_path = _write_manifest(
        out_dir, kind="db", ts=ts, tag=tag, triggered_by=triggered_by,
        artifact_path=sql_path,
    )
    return {
        "kind": "db",
        "ts": ts,
        "sql_path": str(sql_path),
        "tar_path": None,
        "manifest_path": str(manifest_path),
        "size_bytes": _file_size_or_none(sql_path),
    }


def backup_images(
    *, images_root: Path,
    tag: str | None = None, triggered_by: str = "manual",
) -> dict:
    """Run tar --zstd over images_root; write .tar.zst + manifest."""
    ts = _now_ts()
    out_dir = _backups_dir(images_root)
    tar_path = out_dir / f"fc_images_{ts}.tar.zst"
    # No local-temp here (the archive is hundreds of GB — it can't stage in
    # /tmp), but bounded-kill still applies so a tar wedged on NFS fails fast
    # rather than holding the lane for hours.
    _run_bounded(
        [
            "tar", "--zstd", "-cf", str(tar_path),
            "-C", str(images_root.parent), images_root.name,
            f"--exclude={images_root.name}/_backups",
            f"--exclude={images_root.name}/_quarantine",
        ],
        _IMAGES_SUBPROCESS_TIMEOUT_S,
    )
    manifest_path = _write_manifest(
        out_dir, kind="images", ts=ts, tag=tag, triggered_by=triggered_by,
        artifact_path=tar_path,
    )
    return {
        "kind": "images",
        "ts": ts,
        "sql_path": None,
        "tar_path": str(tar_path),
        "manifest_path": str(manifest_path),
        "size_bytes": _file_size_or_none(tar_path),
    }


def restore_db(*, db_url: str, sql_path: Path) -> None:
    """Wipe public schema, then load from the custom-format dump. Raises on
    subprocess failure; partial-restore state is the caller's concern."""
    libpq = _libpq_url(db_url)
    subprocess.run(
        [
            "psql", libpq, "-c",
            "DROP SCHEMA IF EXISTS public CASCADE; CREATE SCHEMA public;",
        ],
        capture_output=True, check=True, timeout=120,
    )
    # Custom-format (-Fc) dumps are restored with pg_restore, not psql.
    subprocess.run(
        ["pg_restore", "--no-owner", "--no-acl", "-d", libpq, str(sql_path)],
        capture_output=True, check=True,
        timeout=_DB_SUBPROCESS_TIMEOUT_S,
    )


def restore_images(*, images_root: Path, tar_path: Path) -> None:
    """Untar over images_root.parent. Additive — files NOT in the
    tarball are NOT removed. Caller wipes first if a clean restore
    is needed."""
    subprocess.run(
        [
            "tar", "--zstd", "-xf", str(tar_path),
            "-C", str(images_root.parent),
        ],
        capture_output=True, check=True,
        timeout=_IMAGES_SUBPROCESS_TIMEOUT_S,
    )


def unlink_artifact_files(
    *,
    sql_path: str | None,
    tar_path: str | None,
    manifest_path: str | None,
) -> dict:
    """Best-effort unlink of all on-disk files for a BackupRun row.
    Returns dict keyed by label with True/False per file. Missing
    files count as success (missing_ok semantics)."""
    deleted: dict = {}
    for label, p in (
        ("sql", sql_path),
        ("tar", tar_path),
        ("manifest", manifest_path),
    ):
        if not p:
            continue
        path = Path(p)
        try:
            path.unlink(missing_ok=True)
            deleted[label] = True
        except OSError:
            deleted[label] = False
    return deleted