FabledCurator/backend/app/tasks/backup.py

"""FC-3h: backup/restore Celery tasks.

All tasks live on the maintenance queue (per celery_app.task_routes).
task_run lifecycle tracking is automatic via FC-3i signals — these
tasks just record the operator-facing artifact metadata into
BackupRun.
"""
from __future__ import annotations

import logging
from datetime import UTC, datetime
from pathlib import Path

from celery.exceptions import SoftTimeLimitExceeded
from sqlalchemy import select
from sqlalchemy.exc import DBAPIError, OperationalError

from ..celery_app import celery
from ..config import get_config
from ..models import BackupRun, ImportSettings
from ..services import backup_service
from ._sync_engine import sync_session_factory as _sync_session_factory

log = logging.getLogger(__name__)
IMAGES_ROOT = Path("/images")


def _mark_failed(session, row: BackupRun, exc: BaseException) -> None:
    """Flip a BackupRun row from running/restoring to error with a
    truncated error message and finished_at. Caller already holds the
    session open."""
    row.status = "error"
    row.error = f"{type(exc).__name__}: {exc}"[:2000]
    row.finished_at = datetime.now(UTC)
    session.add(row)
    session.commit()


@celery.task(
    name="backend.app.tasks.backup.backup_db_task",
    bind=True,
    autoretry_for=(OperationalError, DBAPIError),
    retry_backoff=10, retry_backoff_max=120, max_retries=2,
    soft_time_limit=600, time_limit=720,
)
def backup_db_task(self, *, tag: str | None = None,
                   triggered_by: str = "manual") -> dict:
    """Create one DB backup. Returns {'backup_run_id': N}."""
    SessionLocal = _sync_session_factory()
    cfg = get_config()
    now = datetime.now(UTC)
    with SessionLocal() as session:
        row = BackupRun(
            kind="db", status="running", tag=tag,
            triggered_by=triggered_by, started_at=now, manifest={},
        )
        session.add(row)
        session.commit()
        session.refresh(row)
        run_id = row.id

    try:
        result = backup_service.backup_db(
            db_url=cfg.database_url_sync, images_root=IMAGES_ROOT,
            tag=tag, triggered_by=triggered_by,
        )
    except (SoftTimeLimitExceeded, Exception) as exc:
        with SessionLocal() as session:
            row = session.get(BackupRun, run_id)
            if row is not None:
                _mark_failed(session, row, exc)
        raise

    with SessionLocal() as session:
        row = session.get(BackupRun, run_id)
        row.status = "ok"
        row.finished_at = datetime.now(UTC)
        row.sql_path = result["sql_path"]
        row.size_bytes = result["size_bytes"]
        row.manifest = {
            "manifest_path": result["manifest_path"],
            "ts": result["ts"],
        }
        session.commit()
    return {"backup_run_id": run_id}


@celery.task(
    name="backend.app.tasks.backup.backup_images_task",
    bind=True,
    autoretry_for=(OperationalError, DBAPIError),
    retry_backoff=30, retry_backoff_max=300, max_retries=1,
    soft_time_limit=21600, time_limit=23400,
)
def backup_images_task(self, *, tag: str | None = None,
                       triggered_by: str = "manual") -> dict:
    """Create one images backup. Same shape as backup_db_task; uses
    tar_path instead of sql_path."""
    SessionLocal = _sync_session_factory()
    now = datetime.now(UTC)
    with SessionLocal() as session:
        row = BackupRun(
            kind="images", status="running", tag=tag,
            triggered_by=triggered_by, started_at=now, manifest={},
        )
        session.add(row)
        session.commit()
        session.refresh(row)
        run_id = row.id

    try:
        result = backup_service.backup_images(
            images_root=IMAGES_ROOT,
            tag=tag, triggered_by=triggered_by,
        )
    except (SoftTimeLimitExceeded, Exception) as exc:
        with SessionLocal() as session:
            row = session.get(BackupRun, run_id)
            if row is not None:
                _mark_failed(session, row, exc)
        raise

    with SessionLocal() as session:
        row = session.get(BackupRun, run_id)
        row.status = "ok"
        row.finished_at = datetime.now(UTC)
        row.tar_path = result["tar_path"]
        row.size_bytes = result["size_bytes"]
        row.manifest = {
            "manifest_path": result["manifest_path"],
            "ts": result["ts"],
        }
        session.commit()
    return {"backup_run_id": run_id}


@celery.task(
    name="backend.app.tasks.backup.restore_db_task",
    bind=True,
    max_retries=0,  # NEVER auto-retry a half-applied restore.
    soft_time_limit=1200, time_limit=1800,
)
def restore_db_task(self, *, source_backup_run_id: int) -> dict:
    """Restore from a previous DB backup. Inserts a NEW BackupRun row
    (kind='db', status='restoring') linked to the source via
    restored_from_id; flips to 'restored' on success or 'error' on
    failure. Operator sees the restore as a row in the dashboard."""
    SessionLocal = _sync_session_factory()
    cfg = get_config()
    now = datetime.now(UTC)
    with SessionLocal() as session:
        src = session.get(BackupRun, source_backup_run_id)
        if src is None or src.kind != "db" or not src.sql_path:
            raise ValueError(
                f"BackupRun id={source_backup_run_id} is not a valid DB backup"
            )
        marker = BackupRun(
            kind="db", status="restoring",
            triggered_by="restore", started_at=now,
            restored_from_id=src.id,
            manifest={"source_sql_path": src.sql_path},
        )
        session.add(marker)
        session.commit()
        session.refresh(marker)
        marker_id = marker.id
        sql_path = src.sql_path

    try:
        backup_service.restore_db(
            db_url=cfg.database_url_sync, sql_path=Path(sql_path),
        )
    except (SoftTimeLimitExceeded, Exception) as exc:
        with SessionLocal() as session:
            row = session.get(BackupRun, marker_id)
            if row is not None:
                _mark_failed(session, row, exc)
        raise

    with SessionLocal() as session:
        row = session.get(BackupRun, marker_id)
        row.status = "restored"
        row.finished_at = datetime.now(UTC)
        session.commit()
    return {"backup_run_id": marker_id}


@celery.task(
    name="backend.app.tasks.backup.restore_images_task",
    bind=True, max_retries=0,
    soft_time_limit=21600, time_limit=23400,
)
def restore_images_task(self, *, source_backup_run_id: int) -> dict:
    """Mirrors restore_db_task; uses backup_service.restore_images."""
    SessionLocal = _sync_session_factory()
    now = datetime.now(UTC)
    with SessionLocal() as session:
        src = session.get(BackupRun, source_backup_run_id)
        if src is None or src.kind != "images" or not src.tar_path:
            raise ValueError(
                f"BackupRun id={source_backup_run_id} is not a valid images backup"
            )
        marker = BackupRun(
            kind="images", status="restoring",
            triggered_by="restore", started_at=now,
            restored_from_id=src.id,
            manifest={"source_tar_path": src.tar_path},
        )
        session.add(marker)
        session.commit()
        session.refresh(marker)
        marker_id = marker.id
        tar_path = src.tar_path

    try:
        backup_service.restore_images(
            images_root=IMAGES_ROOT, tar_path=Path(tar_path),
        )
    except (SoftTimeLimitExceeded, Exception) as exc:
        with SessionLocal() as session:
            row = session.get(BackupRun, marker_id)
            if row is not None:
                _mark_failed(session, row, exc)
        raise

    with SessionLocal() as session:
        row = session.get(BackupRun, marker_id)
        row.status = "restored"
        row.finished_at = datetime.now(UTC)
        session.commit()
    return {"backup_run_id": marker_id}


@celery.task(
    name="backend.app.tasks.backup.prune_backups",
    soft_time_limit=300, time_limit=600,
)
def prune_backups() -> dict:
    """Daily Beat. Per-kind retention from ImportSettings.

    Returns {"db_deleted": N, "images_deleted": M, "files_unlinked": K}.
    Tagged rows (tag IS NOT NULL) are never pruned.
    Status='running' / 'restoring' rows are never pruned — the
    recover_stalled_backup_runs sweep flips truly-stuck ones to
    'error' first. (Earlier docstring claimed the FC-3i TaskRun sweep
    handled those, but TaskRun cleanup never touched BackupRun rows.
    Audit 2026-06-02 added the dedicated sweep.)
    """
    SessionLocal = _sync_session_factory()
    counts = {"db_deleted": 0, "images_deleted": 0, "files_unlinked": 0}
    with SessionLocal() as session:
        s = ImportSettings.load_sync(session)
        for kind, keep in (
            ("db", s.backup_db_keep_last_n),
            ("images", s.backup_images_keep_last_n),
        ):
            candidates = session.execute(
                select(BackupRun)
                .where(BackupRun.kind == kind)
                .where(BackupRun.tag.is_(None))
                .where(BackupRun.status.in_(["ok", "error"]))
                .order_by(BackupRun.started_at.desc())
                .offset(keep)
            ).scalars().all()
            for row in candidates:
                result = backup_service.unlink_artifact_files(
                    sql_path=row.sql_path,
                    tar_path=row.tar_path,
                    manifest_path=(row.manifest or {}).get("manifest_path"),
                )
                counts["files_unlinked"] += sum(
                    1 for v in result.values() if v
                )
                session.delete(row)
                counts[f"{kind}_deleted"] += 1
        session.commit()
    return counts


@celery.task(
    name="backend.app.tasks.backup.backup_db_nightly",
    soft_time_limit=60, time_limit=120,
)
def backup_db_nightly() -> dict:
    """Hourly tick. Dispatches a real backup ONLY if the configured
    UTC hour matches and the nightly setting is enabled. Returns
    either {'skipped': '<reason>'} or {'dispatched': '<task_id>'}."""
    SessionLocal = _sync_session_factory()
    with SessionLocal() as session:
        s = ImportSettings.load_sync(session)
        nightly_enabled = s.backup_db_nightly_enabled
        configured_hour = s.backup_db_nightly_hour_utc
    if not nightly_enabled:
        return {"skipped": "nightly disabled"}
    now_hour = datetime.now(UTC).hour
    if now_hour != configured_hour:
        return {"skipped": f"hour={now_hour} != configured={configured_hour}"}
    res = backup_db_task.delay(triggered_by="nightly")
    return {"dispatched": res.id}