siftlode/backend/app/downloads/gc.py

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

Five passes each run:
  1. Pre-expiry warning — a done file whose `expires_at` (its folder's retention, `folders`) falls
     within the grace window, and that its user did not keep, warns its user once (`expiry_warned`).
  2. Expiry — such a file past its `expires_at` leaves the library: the job is deleted and its hold
     released, so the shared file goes with its last holder.
  3. LRU eviction — if a total cache cap is set and exceeded, delete least-recently-accessed ready
     assets (never an upload, never one a user kept) until back under it; each holder's file leaves
     their library, with a notice.
  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, or an upload's stranded file).
     It skips the dot-prefixed system trees (all but `.uploads`) and anything newer than an hour,
     so an in-flight download is never touched.

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

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

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

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

log = logs.get_logger(__name__)
# The one dot-prefixed top-level tree the orphan sweep enters. The others are not part of the asset
# layout (user clips, Plex HLS/img caches, in-progress staging, editor storyboards) and are skipped.
_UPLOAD_TREE = ".uploads"
_ORPHAN_MIN_AGE_S = 3600  # only sweep files older than this, so an in-flight download is never hit


def _notify(db: Session, job: DownloadJob, asset: MediaAsset | None, ntype: str) -> None:
    title = job.display_name or (asset.title if asset else None) or "download"
    create_notification(
        db,
        user_id=job.user_id,
        type=ntype,
        title=title,  # English fallback; the client renders a translated message from data
        data={
            "kind": ntype,
            "title": title,
            "video_id": asset.source_ref if asset else None,
            "job_id": job.id,
        },
        commit=False,
    )


def _relock_expired(
    db: Session, job_id: int, asset_id: int | None, now: datetime
) -> DownloadJob | None:
    """Re-read an expired job under the lock order the user routes take (`own_job_for_update`: the
    asset, then the job) and say whether it is still to go. Between the pass's SELECT and here a
    user may have deleted or re-downloaded it (dropping it again would release its asset a second
    time, freeing a file another job still holds) or kept or moved it (it is no longer expired)."""
    if asset_id is not None:
        db.execute(select(MediaAsset.id).where(MediaAsset.id == asset_id).with_for_update())
    job = db.execute(
        select(DownloadJob)
        .where(DownloadJob.id == job_id)
        .with_for_update()
        .execution_options(populate_existing=True)
    ).scalar_one_or_none()
    still = (
        job is not None
        and job.asset_id == asset_id
        and job.status == JobStatus.DONE
        and not job.kept
        and job.expires_at is not None
        and job.expires_at <= now
    )
    return job if still else None


def _mark_warned(db: Session, job_id: int, expires_at: datetime) -> DownloadJob | None:
    """Mark one file warned if it still goes when the pass saw it go — not deleted, kept, moved or
    re-timed since (each of those sets a fresh expiry, and its own warning to come) — and return it.
    The UPDATE's row lock holds it so until the commit."""
    hit = db.execute(
        update(DownloadJob)
        .where(
            DownloadJob.id == job_id,
            DownloadJob.status == JobStatus.DONE,
            DownloadJob.kept.is_(False),
            DownloadJob.expiry_warned.is_(False),
            DownloadJob.expires_at == expires_at,
        )
        .values(expiry_warned=True)
        .returning(DownloadJob.id)
        .execution_options(synchronize_session=False)
    ).scalar_one_or_none()
    return db.get(DownloadJob, job_id) if hit is not None else None


def _on_disk(db: Session) -> int:
    """The bytes the cache holds now (every asset whose file is on disk)."""
    return (
        db.execute(
            select(func.coalesce(func.sum(MediaAsset.size_bytes), 0)).where(quota.ON_DISK)
        ).scalar()
        or 0
    )


def _relock_evictable(db: Session, asset_id: int) -> tuple[MediaAsset | None, list[DownloadJob]]:
    """Re-read an eviction candidate and its holders under their locks (the asset, then its jobs —
    the order the user routes take): between the pass's SELECT and here a holder may have kept it,
    or every holder deleted it."""
    asset = db.execute(
        select(MediaAsset)
        .where(MediaAsset.id == asset_id)
        .with_for_update()
        .execution_options(populate_existing=True)
    ).scalar_one_or_none()
    holders = (
        db.execute(
            select(DownloadJob)
            .where(DownloadJob.asset_id == asset_id)
            .with_for_update()
            .execution_options(populate_existing=True)
        )
        .scalars()
        .all()
    )
    return asset, list(holders)


def _drop_job(db: Session, job: DownloadJob, ntype: str) -> None:
    """Take a file out of its user's library: tell them, release the job's hold on the asset (the
    file goes with the last holder), delete the row (its shares and links cascade)."""
    _notify(db, job, db.get(MediaAsset, job.asset_id) if job.asset_id else None, ntype)
    service.release_asset(db, job)
    db.delete(job)


def _sweep_orphan_files(db: Session, files_root: str) -> int:
    """Delete media/sidecar files (and now-empty dirs) under FILES_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) — or, under `.uploads`, that no upload
    row names at all. Skips the other 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(files_root)
    if not root.is_dir():
        return 0
    keep: set[Path] = set()
    # An upload's file is on disk from ingest on, whatever its state (a queued conversion's original,
    # a failed one kept for Retry), so every upload row keeps its file — not only a ready one.
    for rel in db.execute(
        select(MediaAsset.rel_path).where(
            or_(MediaAsset.status == "ready", MediaAsset.source_kind == "upload"),
            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():
        # .edits / .plex-* / .staging / .storyboards + any dotfile — but `.uploads` is swept: a
        # conversion stopped between placing the new file and the commit that points at it (or the
        # unlink of the original after it) leaves a file no row names, and nothing else finds it.
        if entry.name.startswith(".") and entry.name != _UPLOAD_TREE:
            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 — one per file, the first run inside the grace window. Each mark is
    #    conditional on the expiry the SELECT saw and committed on its own: a file deleted since is
    #    passed over (a write to its gone row would fail the whole run), and one kept, moved or re-timed
    #    since keeps its fresh expiry unwarned.
    warn_before = now + timedelta(days=grace_days)
    soon = db.execute(
        select(DownloadJob.id, DownloadJob.expires_at).where(
            DownloadJob.status == JobStatus.DONE,
            DownloadJob.kept.is_(False),
            DownloadJob.expiry_warned.is_(False),
            DownloadJob.expires_at.is_not(None),
            DownloadJob.expires_at > now,
            DownloadJob.expires_at <= warn_before,
        )
    ).all()
    db.rollback()  # the pass-1 snapshot holds no locks; each mark takes its own below
    for job_id, expires_at in soon:
        job = _mark_warned(db, job_id, expires_at)
        if job is not None:
            asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None
            _notify(db, job, asset, "download_expiring")
            warned += 1
        db.commit()

    # 2. Expiry — a file past its folder's retention leaves the library (warned in pass 1).
    #    Each file is re-checked under its locks and committed on its own: the release unlinks the
    #    file at once, so it must never act on what the SELECT saw a while ago.
    dead = db.execute(
        select(DownloadJob.id, DownloadJob.asset_id).where(
            DownloadJob.status == JobStatus.DONE,
            DownloadJob.kept.is_(False),
            DownloadJob.expires_at.is_not(None),
            DownloadJob.expires_at <= now,
        )
    ).all()
    db.rollback()  # the pass-2 snapshot holds no locks; each file takes its own below
    for job_id, asset_id in dead:
        job = _relock_expired(db, job_id, asset_id, now)
        if job is not None:
            _drop_job(db, job, "download_expired")
            expired += 1
        db.commit()

    # 3. LRU eviction over the total cache cap: the disk's safety net. Never an upload (no source to
    #    fetch again) and never a file anyone kept; an evicted file leaves every holder's library.
    cap = sysconfig.get_int(db, "download_total_max_bytes")
    if cap > 0:
        if _on_disk(db) > cap:
            kept_assets = select(DownloadJob.asset_id).where(
                DownloadJob.kept.is_(True), DownloadJob.asset_id.is_not(None)
            )
            candidate_ids = (
                db.execute(
                    select(MediaAsset.id)
                    .where(
                        MediaAsset.status == "ready",
                        MediaAsset.source_kind != "upload",
                        MediaAsset.id.not_in(kept_assets),
                    )
                    .order_by(MediaAsset.last_access_at.asc().nulls_first())
                )
                .scalars()
                .all()
            )
            db.rollback()
            for asset_id in candidate_ids:
                # Counted afresh before each: a file deleted since (this one, an upload, a candidate
                # further down) has freed its space, and no other user's file goes for that space.
                if _on_disk(db) <= cap:
                    break
                # Locked and re-read like an expiry: since the SELECT a holder may have kept it,
                # every holder deleted it, or someone asked for it again (a fresh cache hit).
                asset, holders = _relock_evictable(db, asset_id)
                if (
                    asset is None
                    or asset.status != "ready"
                    or any(j.kept for j in holders)
                    or (asset.last_access_at is not None and asset.last_access_at > now)
                ):
                    db.commit()
                    continue
                for job in holders:
                    _notify(db, job, asset, "download_evicted")
                    db.delete(job)
                if asset.rel_path:
                    storage.delete_asset_files(settings.files_root, asset.rel_path)
                db.delete(asset)
                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, is never evicted and 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, is 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.files_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.files_root)

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