siftlode/backend/app/downloads/service.py

"""Enqueue + cache resolution for the download center (the per-user layer over MediaAsset).

`enqueue` is the single entry point used by the API: it resolves (or creates) the shared
MediaAsset for the requested source+format, links a per-user DownloadJob to it, and — on a
cache hit (asset already ready) — marks the job done instantly with no re-download. The worker
handles everything else. State transitions used by the routes (pause/resume/cancel/delete) and
the worker live in `worker.py` / the routes; this module owns enqueue + asset bookkeeping.
"""
from datetime import datetime, timezone

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

from app.downloads import edit as editmod
from app.downloads import formats, quota
from app.models import Channel, DownloadJob, MediaAsset, Video


class EditError(Exception):
    """Bad edit request (empty spec / source not ready); `reason` is an i18n-mappable key."""

    def __init__(self, reason: str):
        self.reason = reason
        super().__init__(reason)


def source_url(source_kind: str, source_ref: str) -> str:
    """The URL yt-dlp should fetch. YouTube stores the 11-char id; ad-hoc stores the URL."""
    if source_kind == "youtube":
        return f"https://www.youtube.com/watch?v={source_ref}"
    return source_ref


def reference_url(job, asset) -> str | None:
    """The clean "downloaded from" URL for a job — shared by the private Downloads page and the public
    watch page so both surfaces always agree. None for edit derivatives (a clip's source is the user's
    own earlier download, not a web page); else the canonical page URL yt-dlp recorded; else one derived
    from source_kind+source_ref (covers queued/older rows before the worker fills it)."""
    if job is None or job.job_kind == "edit":
        return None
    if asset is not None and asset.source_webpage_url:
        return asset.source_webpage_url
    if job.source_kind in ("youtube", "url"):
        return source_url(job.source_kind, job.source_ref)
    return None


def get_or_create_asset(
    db: Session, source_kind: str, source_ref: str, format_sig: str
) -> MediaAsset:
    """Fetch the cache row for this (source, format) or create a fresh `pending` one.

    Races on the unique cache key are resolved by re-selecting after an IntegrityError, so two
    simultaneous enqueues of the same video+format converge on one shared asset."""
    stmt = select(MediaAsset).where(
        MediaAsset.source_kind == source_kind,
        MediaAsset.source_ref == source_ref,
        MediaAsset.format_sig == format_sig,
    )
    asset = db.execute(stmt).scalar_one_or_none()
    if asset is not None:
        return asset
    asset = MediaAsset(
        source_kind=source_kind,
        source_ref=source_ref,
        format_sig=format_sig,
        status="pending",
        ref_count=0,
    )
    db.add(asset)
    try:
        db.flush()
    except IntegrityError:
        db.rollback()
        asset = db.execute(stmt).scalar_one()
    return asset


def populate_from_catalog(db: Session, asset: MediaAsset) -> bool:
    """Fill an asset's display metadata (title/uploader/thumbnail/date/duration) from our own
    catalog when the source is a known YouTube video — so a queued/downloading row shows the
    real title + thumbnail immediately instead of the bare id. 0 network cost. Returns True if
    it found a catalog match."""
    if asset.source_kind != "youtube" or asset.title:
        return False
    video = db.get(Video, asset.source_ref)
    if video is None:
        return False
    asset.title = video.title
    asset.thumbnail_url = video.thumbnail_url
    asset.duration_s = video.duration_seconds
    if video.published_at:
        asset.upload_date = video.published_at.date()
    channel = db.get(Channel, video.channel_id)
    if channel is not None:
        asset.uploader = channel.title
    return True


def _next_queue_pos(db: Session) -> int:
    return (db.execute(select(func.coalesce(func.max(DownloadJob.queue_pos), 0))).scalar() or 0) + 1


def enqueue(
    db: Session,
    user_id: int,
    source_kind: str,
    source_ref: str,
    spec: dict,
    profile_id: int | None = None,
    display_name: str | None = None,
) -> DownloadJob:
    """Queue a download for `user_id`. Returns the DownloadJob (already `done` on a cache hit).

    Raises quota.QuotaExceeded if the user is at their job-count or footprint cap."""
    quota.check_enqueue(db, user_id)
    spec = formats.normalize(spec)
    sig = formats.format_sig(spec)
    asset = get_or_create_asset(db, source_kind, source_ref, sig)
    # Show a real title + thumbnail the moment it's queued (feed downloads are catalog videos).
    populate_from_catalog(db, asset)

    job = DownloadJob(
        user_id=user_id,
        asset_id=asset.id,
        source_kind=source_kind,
        source_ref=source_ref,
        profile_id=profile_id,
        profile_snapshot=spec,
        display_name=display_name or (asset.title[:255] if asset.status == "ready" and asset.title else None),
        status="queued",
        queue_pos=_next_queue_pos(db),
    )
    db.add(job)
    asset.ref_count = (asset.ref_count or 0) + 1

    # Cache hit: the file already exists — no worker round-trip needed.
    if asset.status == "ready":
        job.status = "done"
        job.progress = 100
        asset.last_access_at = datetime.now(timezone.utc)

    db.commit()
    db.refresh(job)
    return job


def _clip_title(source_title: str | None, display_name: str | None) -> str:
    if display_name:
        return display_name[:255]
    base = source_title or "Clip"
    return f"{base} (clip)"[:255]


def enqueue_edit(
    db: Session,
    user_id: int,
    source_job: DownloadJob,
    edit_spec: dict,
    display_name: str | None = None,
) -> DownloadJob:
    """Queue a phase-2 editor derivative (trim/crop) of `source_job` for `user_id`.

    The derivative is per-user (never shared-cached): its asset uses `source_kind='edit'` with a
    cache key that embeds the user id, so an identical re-edit by the same user is a cache hit but
    it never dedups across users. Reuses the same worker/quota/GC path as a download.

    Raises quota.QuotaExceeded (holdings cap) or EditError (empty spec / source not ready)."""
    src_asset = db.get(MediaAsset, source_job.asset_id) if source_job.asset_id else None
    if src_asset is None or src_asset.status != "ready" or not src_asset.rel_path:
        raise EditError("source_not_ready")

    spec = editmod.normalize_edit_spec(edit_spec)
    if not spec:
        raise EditError("empty_edit")

    quota.check_enqueue(db, user_id)

    sig = editmod.edit_sig(spec)
    ref = f"{user_id}:{src_asset.id}"[:512]
    asset = get_or_create_asset(db, "edit", ref, sig)
    if not asset.title:
        asset.title = _clip_title(src_asset.title, display_name)
        asset.uploader = src_asset.uploader
        asset.thumbnail_url = src_asset.thumbnail_url
        asset.upload_date = src_asset.upload_date
        asset.duration_s = editmod.clip_duration(spec, src_asset.duration_s)

    job = DownloadJob(
        user_id=user_id,
        asset_id=asset.id,
        job_kind="edit",
        source_kind="edit",
        source_ref=ref,
        source_job_id=source_job.id,
        source_asset_id=src_asset.id,
        edit_spec=spec,
        profile_snapshot={},
        display_name=display_name or asset.title,
        status="queued",
        queue_pos=_next_queue_pos(db),
    )
    db.add(job)
    asset.ref_count = (asset.ref_count or 0) + 1

    if asset.status == "ready":  # identical clip already produced for this user → instant
        job.status = "done"
        job.progress = 100
        asset.last_access_at = datetime.now(timezone.utc)

    db.commit()
    db.refresh(job)
    return job