FabledCurator/backend/app/tasks/maintenance.py

"""Periodic maintenance: recover stuck import tasks, garbage-collect old finished tasks."""

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

from PIL import Image
from sqlalchemy import delete, select, update

from ..celery_app import celery
from ..models import DownloadEvent, ImageRecord, ImportSettings, ImportTask, TaskRun
from ..utils.phash import compute_phash
from ._sync_engine import sync_session_factory as _sync_session_factory

log = logging.getLogger(__name__)

STUCK_THRESHOLD_MINUTES = 5
ORPHAN_PENDING_THRESHOLD_MINUTES = 30
OLD_TASK_DAYS = 7
PHASH_PAGE = 500
VERIFY_PAGE = 200
FFPROBE_TIMEOUT_SECONDS = 10
TASK_RUN_KEEP_OK_SECONDS = 24 * 3600       # 24 h
TASK_RUN_KEEP_FAILURE_SECONDS = 7 * 24 * 3600  # 7 days

# Per-queue overrides for recover_stalled_task_runs. Queues whose
# tasks can legitimately run longer than the default 5-min threshold
# need their own larger value, otherwise the sweep marks in-flight
# tasks 'error' before they get a chance to finish. The dict's value
# MUST be ≥ the longest task.time_limit on the queue + a small buffer.
#
# ml: tag_and_embed video branch samples 10 frames, runs tagger +
#     embedder on each — soft_time_limit=900 / time_limit=1200; sweep
#     at 25 min gives a 5-min buffer past the hard kill.
#     Operator-flagged 2026-05-28 (image 6288, an mp4, marked failed
#     at the 5-min sweep tick while still processing).
QUEUE_STUCK_THRESHOLD_MINUTES: dict[str, int] = {
    "ml": 25,
}


@celery.task(name="backend.app.tasks.maintenance.recover_interrupted_tasks")
def recover_interrupted_tasks() -> int:
    """Recover stuck ImportTask rows. Two distinct stuck states:

    1. 'processing' > 5 min — worker crash mid-import. Re-queue via
       .delay() and let the import retry. Was 30 min historically;
       tightened 2026-05-24 after operator hit a 2224-row zombie pile.
       import_media_file is sub-second for the vast majority of files and
       capped at the per-task soft_time_limit (5 min), so anything still
       'processing' after that window is a confirmed crash.

    2. 'pending' or 'queued' > 30 min — enqueue-phase crash. scan_directory
       creates rows with status='pending' (commit), then in a second pass
       transitions to 'queued' and calls .delay() (commit). If the scanner
       crashes between those two commits, rows are orphaned in 'pending'
       (never enqueued) with no recovery path — invisible to the
       'processing' sweep above. Flagged 2026-05-25 by operator hitting a
       5490-row orphan pile. Flip these to 'failed' (not re-enqueue) so
       the operator drains them via /api/import/retry-failed at their own
       pace; bulk-re-enqueueing 5000+ rows would thundering-herd the
       import worker.

    Returns total rows touched (recovered + marked failed).
    """
    SessionLocal = _sync_session_factory()
    now = datetime.now(UTC)
    processing_cutoff = now - timedelta(minutes=STUCK_THRESHOLD_MINUTES)
    orphan_cutoff = now - timedelta(minutes=ORPHAN_PENDING_THRESHOLD_MINUTES)
    with SessionLocal() as session:
        # Both sweeps used to be SELECT ids → UPDATE WHERE id IN (...) which
        # blew past psycopg's 65535-parameter ceiling once a sweep covered
        # tens of thousands of rows (operator hit it 2026-05-26 after the
        # /import deep scan piled up orphans). Folding the SELECT into the
        # UPDATE eliminates the IN-list entirely. RETURNING gives us back
        # exactly the ids that flipped so the stuck sweep can still
        # .delay() each one.
        stuck_result = session.execute(
            update(ImportTask)
            .where(ImportTask.status == "processing")
            .where(ImportTask.started_at < processing_cutoff)
            .values(
                status="queued",
                started_at=None,
                error="recovered from stuck state",
            )
            .returning(ImportTask.id)
        )
        stuck_ids = [row[0] for row in stuck_result.all()]

        orphan_result = session.execute(
            update(ImportTask)
            .where(ImportTask.status.in_(["pending", "queued"]))
            .where(ImportTask.created_at < orphan_cutoff)
            .values(
                status="failed",
                error=(
                    "orphan pending/queued swept by recover_interrupted_tasks "
                    "(scanner likely crashed mid-enqueue); retry via "
                    "/api/import/retry-failed"
                ),
            )
        )
        orphan_count = orphan_result.rowcount or 0

        session.commit()

        if stuck_ids:
            from .import_file import import_media_file
            for tid in stuck_ids:
                import_media_file.delay(tid)

        return len(stuck_ids) + orphan_count


@celery.task(name="backend.app.tasks.maintenance.cleanup_old_tasks")
def cleanup_old_tasks() -> int:
    """Delete completed/skipped/failed ImportTask rows older than 7 days.

    Why 7 days: long enough to debug an issue an operator only notices days
    later; short enough that the task table stays a useful operational view
    rather than an archive. Matches IR's default.
    """
    SessionLocal = _sync_session_factory()
    cutoff = datetime.now(UTC) - timedelta(days=OLD_TASK_DAYS)
    with SessionLocal() as session:
        result = session.execute(
            delete(ImportTask)
            .where(ImportTask.status.in_(["complete", "skipped", "failed"]))
            .where(ImportTask.finished_at < cutoff)
        )
        session.commit()
        return result.rowcount or 0


@celery.task(name="backend.app.tasks.maintenance.recover_stalled_task_runs")
def recover_stalled_task_runs() -> int:
    """Flip task_run rows stuck in 'running' past their queue-specific
    threshold to 'error'. FC-3i.

    A row gets stuck when the worker dies without emitting
    task_postrun / task_failure (e.g. OOM, container restart between
    signals, signal handler raised+logged). The default 5-min threshold
    fits short-lived queues (import/thumbnail/download); queues that
    legitimately run longer tasks (ml-video, deep scans) get their
    own larger threshold via QUEUE_STUCK_THRESHOLD_MINUTES so the
    sweep doesn't preempt them.

    Runs once per distinct threshold value: each pass updates rows
    whose queue maps to that threshold.
    """
    SessionLocal = _sync_session_factory()
    now = datetime.now(UTC)
    # Group queues by their threshold value so we issue one UPDATE
    # per distinct threshold. Queues NOT in the override dict use the
    # default; their UPDATE excludes the override queues so each row
    # is touched at most once.
    override_queues = set(QUEUE_STUCK_THRESHOLD_MINUTES.keys())
    total = 0
    with SessionLocal() as session:
        # Default-threshold pass — all queues except the overridden ones.
        default_cutoff = now - timedelta(minutes=STUCK_THRESHOLD_MINUTES)
        default_stmt = (
            update(TaskRun)
            .where(TaskRun.status == "running")
            .where(TaskRun.started_at < default_cutoff)
            .values(
                status="error",
                error_type="RecoverySweep",
                error_message=(
                    f"no completion signal received within "
                    f"{STUCK_THRESHOLD_MINUTES} min"
                ),
                finished_at=now,
            )
        )
        if override_queues:
            default_stmt = default_stmt.where(
                TaskRun.queue.notin_(override_queues)
            )
        total += session.execute(default_stmt).rowcount or 0

        # Per-queue override passes.
        for queue, minutes in QUEUE_STUCK_THRESHOLD_MINUTES.items():
            cutoff = now - timedelta(minutes=minutes)
            stmt = (
                update(TaskRun)
                .where(TaskRun.status == "running")
                .where(TaskRun.queue == queue)
                .where(TaskRun.started_at < cutoff)
                .values(
                    status="error",
                    error_type="RecoverySweep",
                    error_message=(
                        f"no completion signal received within "
                        f"{minutes} min"
                    ),
                    finished_at=now,
                )
            )
            total += session.execute(stmt).rowcount or 0

        session.commit()
        return total


@celery.task(name="backend.app.tasks.maintenance.prune_task_runs")
def prune_task_runs() -> dict:
    """Daily retention for task_run rows. FC-3i.

    - 'ok' rows: deleted after TASK_RUN_KEEP_OK_SECONDS (24h default).
      Success is high-volume, not interesting after a day.
    - 'error' / 'timeout' rows: deleted after TASK_RUN_KEEP_FAILURE_SECONDS
      (7 days default). Failures are operationally interesting longer.
    - 'running' rows: NEVER deleted by this task. The recovery sweep
      (recover_stalled_task_runs) is the mechanism that flips them to
      terminal state; prune doesn't touch in-flight state.
    - 'retry' rows: treated as failures (>7d).

    Returns dict of how many rows were deleted in each bucket.
    """
    SessionLocal = _sync_session_factory()
    now = datetime.now(UTC)
    ok_cutoff = now - timedelta(seconds=TASK_RUN_KEEP_OK_SECONDS)
    fail_cutoff = now - timedelta(seconds=TASK_RUN_KEEP_FAILURE_SECONDS)
    with SessionLocal() as session:
        ok_deleted = session.execute(
            delete(TaskRun)
            .where(TaskRun.status == "ok")
            .where(TaskRun.finished_at < ok_cutoff)
        ).rowcount or 0
        fail_deleted = session.execute(
            delete(TaskRun)
            .where(TaskRun.status.in_(["error", "timeout", "retry"]))
            .where(TaskRun.finished_at < fail_cutoff)
        ).rowcount or 0
        session.commit()
        return {"ok_deleted": ok_deleted, "failures_deleted": fail_deleted}


@celery.task(name="backend.app.tasks.maintenance.backfill_phash")
def backfill_phash() -> int:
    """Recompute phash for stored images that have none (imported before
    FC-2d-i+ii). Keyset-paginated by id (restart-safe), NULL-only fill,
    idempotent. Videos legitimately keep phash NULL. A missing/unreadable
    file is logged and left NULL — never fails the task."""
    SessionLocal = _sync_session_factory()
    updated = 0
    last_id = 0
    with SessionLocal() as session:
        while True:
            rows = session.execute(
                select(ImageRecord)
                .where(ImageRecord.id > last_id)
                .where(ImageRecord.phash.is_(None))
                .where(ImageRecord.mime.like("image/%"))
                .order_by(ImageRecord.id.asc())
                .limit(PHASH_PAGE)
            ).scalars().all()
            if not rows:
                break
            for rec in rows:
                try:
                    with Image.open(rec.path) as im:
                        ph = compute_phash(im)
                except Exception as exc:
                    log.warning(
                        "backfill_phash: unreadable %s: %s", rec.path, exc
                    )
                    ph = None
                if ph is not None and rec.phash is None:
                    rec.phash = ph
                    updated += 1
            session.commit()
            last_id = rows[-1].id
    return updated


def _verify_one(path: Path, expected_sha: str, mime: str, sha_fn) -> str:
    """Compute the integrity verdict for one file. Status precedence:
    failed_verification (can't run) > corrupt (sha mismatch / decode
    fails) > ok (passes both). Never raises."""
    try:
        if not path.is_file():
            return "failed_verification"
        try:
            actual_sha = sha_fn(path)
        except (OSError, PermissionError):
            return "failed_verification"
        if actual_sha != expected_sha:
            return "corrupt"
        if mime and mime.startswith("image/"):
            try:
                with Image.open(path) as im:
                    im.verify()
            except Exception:
                return "corrupt"
            return "ok"
        if mime and mime.startswith("video/"):
            try:
                proc = subprocess.run(
                    ["ffprobe", "-v", "error", "-i", str(path)],
                    capture_output=True,
                    timeout=FFPROBE_TIMEOUT_SECONDS,
                )
            except FileNotFoundError:
                # ffprobe binary missing — environment problem, not file.
                return "failed_verification"
            except subprocess.TimeoutExpired:
                return "corrupt"
            return "ok" if proc.returncode == 0 else "corrupt"
        # Unknown mime — sha matched already; trust that.
        return "ok"
    except Exception as exc:
        log.warning("verify_integrity unexpected error for %s: %s", path, exc)
        return "failed_verification"


@celery.task(name="backend.app.tasks.maintenance.verify_integrity")
def verify_integrity() -> int:
    """Verify every ImageRecord file: sha256 recompute + decode/probe
    (PIL for images; ffprobe for videos). Writes integrity_status
    (always — the column reflects the most recent verdict).
    Keyset-paginated, fail-soft per row, idempotent. Returns the total
    count verified."""
    from ..services.importer import _sha256_of  # reuse the importer's helper

    SessionLocal = _sync_session_factory()
    total = 0
    counts = {"ok": 0, "corrupt": 0, "failed_verification": 0}
    last_id = 0
    with SessionLocal() as session:
        while True:
            rows = session.execute(
                select(ImageRecord)
                .where(ImageRecord.id > last_id)
                .order_by(ImageRecord.id.asc())
                .limit(VERIFY_PAGE)
            ).scalars().all()
            if not rows:
                break
            for rec in rows:
                rec.integrity_status = _verify_one(
                    Path(rec.path), rec.sha256, rec.mime, _sha256_of
                )
                counts[rec.integrity_status] = (
                    counts.get(rec.integrity_status, 0) + 1
                )
                total += 1
            session.commit()
            last_id = rows[-1].id
    log.info("verify_integrity verdicts: %s (total %d)", counts, total)
    return total


@celery.task(name="backend.app.tasks.maintenance.cleanup_old_download_events")
def cleanup_old_download_events() -> int:
    """FC-3d: delete terminal DownloadEvent rows older than the configured
    retention window. Never touches pending/running rows.

    Why terminal-only: pending/running rows represent in-flight work whose
    owning task may still be alive; deleting them would orphan the task.
    Retention days comes from ImportSettings.download_event_retention_days
    so the operator can tune without a code change.
    """
    SessionLocal = _sync_session_factory()
    with SessionLocal() as session:
        settings = session.execute(
            select(ImportSettings).where(ImportSettings.id == 1)
        ).scalar_one()
        retention_days = settings.download_event_retention_days
        cutoff = datetime.now(UTC) - timedelta(days=retention_days)
        result = session.execute(
            delete(DownloadEvent)
            .where(DownloadEvent.status.in_(["ok", "error", "skipped"]))
            .where(DownloadEvent.started_at < cutoff)
        )
        session.commit()
        return result.rowcount or 0