siftlode/backend/app/downloads/gc.py

"""Download retention garbage collector (a scheduler job).

Five passes each run:
  1. Pre-expiry warning — for ready assets expiring within the grace window, notify each holder
     once (gc_notified flag) so a download isn't silently deleted.
  2. Expiry deletion — delete assets past their TTL (files + row). The FK SET NULL nulls the
     holders' jobs' asset_id, so those jobs surface as "expired" (re-downloadable) in the UI.
  3. LRU eviction — if a total cache cap is set and exceeded, delete least-recently-accessed
     ready assets until back under it.
  4. Orphan reap — delete errored, fileless, unreferenced asset rows (they carry no TTL, so
     passes 1-3 skip them; benign row-count hygiene).
  5. Orphan FILE sweep — delete on-disk files/dirs no surviving asset owns (cruft left when a
     re-download renames an asset, e.g. old accented channel dirs), skipping system trees and
     anything newer than an hour so an in-flight download is never touched.

Runs in the API process (which mounts DOWNLOAD_ROOT). Pure disk/DB work — no YouTube quota.
"""

import time
from datetime import datetime, timedelta, timezone
from pathlib import Path

from sqlalchemy import func, select
from sqlalchemy.orm import Session

from app import logs, sysconfig
from app.config import settings
from app.downloads import storage
from app.models import DownloadJob, MediaAsset
from app.notifications import create_notification

log = logs.get_logger(__name__)
# Dot-prefixed top-level trees that are NOT part of the Plex/flat asset layout — the orphan sweep
# never touches them (user clips, Plex HLS/img caches, in-progress staging, editor storyboards).
_SYSTEM_TREE = {".edits", ".plex-hls", ".plex-img-cache", ".staging", ".storyboards"}
_ORPHAN_MIN_AGE_S = 3600  # only sweep files older than this, so an in-flight download is never hit


def _holders(db: Session, asset_id: int) -> list[int]:
    return list(
        db.execute(
            select(DownloadJob.user_id).where(DownloadJob.asset_id == asset_id).distinct()
        ).scalars()
    )


def _notify_holders(db: Session, asset: MediaAsset, ntype: str) -> None:
    title = asset.title or "download"
    for uid in _holders(db, asset.id):
        create_notification(
            db,
            user_id=uid,
            type=ntype,
            title=title,  # English fallback; the client renders a translated message from data
            data={"kind": ntype, "title": asset.title, "video_id": asset.source_ref},
            commit=False,
        )


def _delete_asset(db: Session, asset: MediaAsset, ntype: str) -> None:
    if asset.rel_path:
        storage.delete_asset_files(settings.download_root, asset.rel_path)
    _notify_holders(db, asset, ntype)
    db.delete(asset)  # jobs.asset_id -> NULL via FK ON DELETE SET NULL


def _sweep_orphan_files(db: Session, download_root: str) -> int:
    """Delete media/sidecar files (and now-empty dirs) under DOWNLOAD_ROOT that no `ready` asset
    owns — cruft left when a re-download changes an asset's on-disk name (e.g. an old accented
    channel dir, or a poster from a superseded naming scheme). Skips the dot-prefixed system trees
    and only removes files older than `_ORPHAN_MIN_AGE_S`, so a just-completed or in-flight download
    is never deleted, and only files NOT owned by a surviving asset."""
    root = Path(download_root)
    if not root.is_dir():
        return 0
    keep: set[Path] = set()
    for rel in db.execute(
        select(MediaAsset.rel_path).where(
            MediaAsset.status == "ready", MediaAsset.rel_path.is_not(None)
        )
    ).scalars():
        media = (root / rel).resolve()
        keep |= {media, media.with_suffix(".nfo"), media.with_suffix(".jpg")}

    cutoff = time.time() - _ORPHAN_MIN_AGE_S
    removed = 0
    # Throttle the whole-tree walk: this runs in the API process, and an unpaced rglob+stat over a
    # large download root monopolizes disk IO long enough to make web requests + playback stutter.
    # Yield the IO channel after every `batch` files touched so foreground work interleaves.
    pause_s = max(0, settings.download_gc_sweep_pause_ms) / 1000.0
    batch = max(1, settings.download_gc_sweep_batch)
    touched = 0

    def _throttle() -> None:
        nonlocal touched
        touched += 1
        if pause_s and touched % batch == 0:
            time.sleep(pause_s)

    unreclaimed = 0
    for entry in root.iterdir():
        if entry.name.startswith("."):  # .edits / .plex-* / .staging / .storyboards + any dotfile
            continue
        files = [entry] if entry.is_file() else [p for p in entry.rglob("*") if p.is_file()]
        for f in files:
            _throttle()
            try:
                if f.resolve() in keep or f.stat().st_mtime > cutoff:
                    continue
                f.unlink()
                removed += 1
            except OSError:
                # silent-ok: counted, not logged per file — a broken mount would put one line
                # under every orphan. The total is reported once at the end of the sweep.
                unreclaimed += 1
        if entry.is_dir():  # prune emptied dirs (deepest first), then the channel dir itself
            for d in sorted(
                (p for p in entry.rglob("*") if p.is_dir()),
                key=lambda p: len(p.parts),
                reverse=True,
            ):
                try:
                    d.rmdir()
                # silent-ok: `rmdir` refusing means the directory still holds files, which is
                # the ordinary answer here — this prune only takes the ones that emptied.
                except OSError:
                    pass
            try:
                entry.rmdir()
            # silent-ok: the same prune one level up, at the channel directory.
            except OSError:
                pass
    if unreclaimed:
        log.warning(
            "some orphaned download files could not be reclaimed",
            extra=logs.event("download.gc.unreclaimed", files=unreclaimed, removed=removed),
        )
    return removed


def run_download_gc(db: Session) -> dict:
    now = datetime.now(timezone.utc)
    grace_days = sysconfig.get_int(db, "download_gc_grace_days")
    warned = expired = evicted = reaped = 0

    # 1. Pre-expiry warnings.
    warn_before = now + timedelta(days=grace_days)
    soon = (
        db.execute(
            select(MediaAsset).where(
                MediaAsset.status == "ready",
                MediaAsset.gc_notified.is_(False),
                MediaAsset.expires_at.is_not(None),
                MediaAsset.expires_at > now,
                MediaAsset.expires_at <= warn_before,
            )
        )
        .scalars()
        .all()
    )
    for asset in soon:
        _notify_holders(db, asset, "download_expiring")
        asset.gc_notified = True
        warned += 1
    if soon:
        db.commit()

    # 2. Expiry deletion (ready assets past TTL, and any errored leftovers with a TTL).
    dead = (
        db.execute(
            select(MediaAsset).where(
                MediaAsset.status.in_(("ready", "error")),
                MediaAsset.expires_at.is_not(None),
                MediaAsset.expires_at <= now,
            )
        )
        .scalars()
        .all()
    )
    for asset in dead:
        _delete_asset(db, asset, "download_expired")
        expired += 1
    if dead:
        db.commit()

    # 3. LRU eviction over the total cache cap.
    cap = sysconfig.get_int(db, "download_total_max_bytes")
    if cap > 0:
        total = (
            db.execute(
                select(func.coalesce(func.sum(MediaAsset.size_bytes), 0)).where(
                    MediaAsset.status == "ready"
                )
            ).scalar()
            or 0
        )
        if total > cap:
            candidates = (
                db.execute(
                    select(MediaAsset)
                    .where(MediaAsset.status == "ready")
                    .order_by(MediaAsset.last_access_at.asc().nulls_first())
                )
                .scalars()
                .all()
            )
            for asset in candidates:
                if total <= cap:
                    break
                total -= asset.size_bytes or 0
                _delete_asset(db, asset, "download_evicted")
                evicted += 1
            db.commit()

    # 4. Reap orphaned errored rows (no file, no holder). get_or_create_asset resets a still-
    #    referenced errored row to pending on re-enqueue, so these only accumulate once every
    #    holder is gone; they carry no expires_at, so passes 1-3 never touch them.
    orphans = (
        db.execute(
            select(MediaAsset).where(
                MediaAsset.status == "error",
                MediaAsset.ref_count == 0,
                MediaAsset.rel_path.is_(None),
            )
        )
        .scalars()
        .all()
    )
    for asset in orphans:
        db.delete(asset)  # no file to unlink; FK SET NULL clears any stray job.asset_id
        reaped += 1
    if orphans:
        db.commit()

    # 4b. Reap unheld uploads. An upload carries no expires_at and is never re-enqueued, so the passes
    #     above never free one that lost its last holder outside `release_asset` — a conversion that
    #     finished after its job was cancelled, or one cancelled mid-run (the asset was `downloading`
    #     then). Its file, ready or still the unconverted original, sits at `rel_path`.
    unheld = (
        db.execute(
            select(MediaAsset).where(
                MediaAsset.source_kind == "upload",
                MediaAsset.ref_count == 0,
                MediaAsset.status != "downloading",
            )
        )
        .scalars()
        .all()
    )
    for asset in unheld:
        if asset.rel_path:  # no holder is left to tell, so no notice — just the file and the row
            storage.delete_asset_files(settings.download_root, asset.rel_path)
        db.delete(asset)
        reaped += 1
    if unheld:
        db.commit()

    # 5. Orphan FILE sweep — remove on-disk cruft no surviving asset owns (old accented dirs, stale
    #    posters). Runs last so the keep-set reflects the assets that survived passes 2-4.
    swept = _sweep_orphan_files(db, settings.download_root)

    result = {
        "warned": warned,
        "expired": expired,
        "evicted": evicted,
        "reaped": reaped,
        "swept": swept,
    }
    log.info("download_gc %s", result)
    return result