siftlode/backend/app/routes/plex/playback.py

"""Plex playback: image/subtitle proxies, HLS streaming, and watch-state writes."""

import hashlib
import threading
from datetime import datetime, timedelta, timezone

import httpx
from fastapi import (
    APIRouter,
    BackgroundTasks,
    Depends,
    HTTPException,
    Path,
    Query,
    Request,
    Response,
)
from fastapi.responses import FileResponse
from pydantic import BaseModel, ConfigDict, Field
from sqlalchemy import select
from sqlalchemy.orm import Session

from app.auth import current_user, resolved_user_id
from app.db import SessionLocal, get_db
from app.models import (
    POSITION_MAX_S,
    PlexItem,
    PlexSeason,
    PlexShow,
    PlexState,
    User,
)
from app.plex import imgcache, subtitles
from app.plex import paths as plex_paths
from app.plex.lookups import by_rating_key
from app.plex import people as plex_people
from app.plex import stream as plex_stream
from app.plex import watch_sync as plex_watch
from app.plex.client import PlexClient, PlexError, PlexNotConfigured
from app.routes.plex._shared import item_or_404

router = APIRouter(prefix="/api/plex", tags=["plex"])


def _require_session(request: Request) -> int:
    """Lightweight auth for the high-fan-out image proxies: validate the signed session cookie WITHOUT
    a DB user-load, so these endpoints never hold a pool connection for the auth. Poster images are
    low-sensitivity; a valid session is enough."""
    uid, _ = resolved_user_id(request)
    if not uid:
        raise HTTPException(status_code=401, detail="Not authenticated")
    return uid


@router.get("/image/{rating_key}")
def image(rating_key: str, request: Request, variant: str = "thumb") -> Response:
    """Proxy a Plex poster/art image (keeps the admin token server-side; no image duplication). A
    thin on-disk cache serves repeat views without re-hitting Plex.

    ⚠️ This endpoint is called in BURSTS (a collections page fires 100+ concurrent poster requests),
    so it must NOT hold a DB connection during the slow Plex fetch — doing so exhausted the pool
    (QueuePool timeout, broken images on prod). Warm path (cache hit) touches NO DB; cold path opens a
    brief session only to resolve the image key + read the Plex config, then RELEASES it before the
    fetch. Auth is the signed-session check (no DB user-load)."""
    _require_session(request)
    v = "art" if variant == "art" else "thumb"
    w, h = imgcache.SIZES[v]
    # Size is part of the cache key: Plex resizes server-side (see PlexClient.image_bytes), so we cache
    # the small variant. The width tag also means changing a target size just re-caches, never serves a
    # stale differently-sized file (old full-size files under the un-tagged key are simply orphaned).
    cache_key = f"item_{rating_key}_{v}_{w}"
    hit = imgcache.read(cache_key)
    if hit:
        return imgcache.response(*hit)
    # Cold miss: short-lived session for the key + Plex client config, released before the slow fetch.
    try:
        with SessionLocal() as db:
            key = imgcache.key_for(db, str(rating_key), v)
            if not key:
                raise HTTPException(status_code=404, detail="No image")
            plex = PlexClient(db)  # reads base URL + token now; the fetch below needs no DB
    except PlexNotConfigured as e:
        raise HTTPException(status_code=400, detail=str(e))
    try:
        with plex:
            data, ct = plex.image_bytes(key, w, h)
    except PlexError as e:
        raise HTTPException(status_code=502, detail=f"Plex image fetch failed: {e}")
    imgcache.write(cache_key, data, ct)
    return imgcache.response(data, ct)


# Both subtitle routes authenticate with the full `current_user`, not the cookie-only
# `_require_session` the image proxies use. That lighter gate exists for bursts of low-sensitivity
# poster fetches and skips the is_active / is_suspended / session-epoch checks — acceptable for a
# poster, not for a route that starts reads of the media library: a suspended or "log out
# everywhere"-revoked cookie kept scheduling them. They are not bursty (one poll a second, one fetch
# per track), so the DB user-load costs nothing that matters.
#
# Both path parameters are bounded, the same way on both routes: they become a cache FILE name
# (`sub_<rating_key>_<ord>`), and an absurd one — a 250-digit key or ordinal — outgrew the OS's name
# limit and turned the status poll into a 500. A Plex rating_key is a short number; the bound only
# has to be generous.
_RatingKey = Path(max_length=32)
_SubOrd = Path(ge=0, le=9999)


@router.get("/subtitle/{rating_key}/{sub_ord}/status")
def subtitle_status(
    rating_key: str = _RatingKey,
    sub_ord: int = _SubOrd,
    user: User = Depends(current_user),
    # Bounded both ways: a non-finite position used to reach the scheduler's arithmetic and 500, and
    # an absurd one reached ffmpeg's `-ss`, which refused it.
    at: float = Query(0.0, ge=0.0, le=subtitles.MAX_POSITION_S, allow_inf_nan=False),
) -> dict:
    """Is this subtitle track showable yet — and if not, get it moving. Extraction runs in the
    background (see plex/subtitles.py for the measurements that forced it there), so the player polls
    this while it waits instead of hanging on a <track> that will not load for a minute or more.

    `at` is the viewer's absolute position: the extractor windows around it, so what gets read first
    is the part actually about to be watched."""
    return subtitles.ensure(rating_key, sub_ord, at)


@router.get("/subtitle/{rating_key}/{sub_ord}")
def subtitle(
    rating_key: str = _RatingKey,
    sub_ord: int = _SubOrd,
    user: User = Depends(current_user),
    # Finite, like every float this router takes: `-inf` crashed the shift (a 500), `nan` stacked
    # every cue at 0. It can legitimately be negative (a subtitle nudged later than the session start).
    offset: float = Query(0.0, allow_inf_nan=False),
) -> Response:
    """Serve the ord-th subtitle track as WebVTT for the browser's <track> — so choosing a subtitle
    overlays it directly (no HLS re-mux / session restart). Handles EXTERNAL sidecar subs (fetched
    from Plex via the stream `key` — the old `-map 0:s` broke on these since they're not in the file)
    AND embedded text subs (extracted from the local file). Image subs (PGS/VobSub) → 415. Disk-cached.

    A pure READ: it serves whatever is cached and schedules nothing — the status poll owns that,
    because only it knows where the viewer is (see `subtitles.cached_track`). The player mounts the
    <track> only once /status says there is something to show, so in practice this is only ever hit
    warm; on a cache eviction race the 202 sends the player back to polling."""
    # The cached VTT holds ABSOLUTE cue times (per rk+ord); the per-session cue-shift is applied on
    # the way out, so a resume/seek to any offset reuses the same track (no cache-per-offset explosion).
    hit, code = subtitles.cached_track(rating_key, sub_ord)
    if hit is None and code:
        raise HTTPException(
            status_code=subtitles.UNAVAILABLE_STATUS.get(code, 422),
            detail={"code": "subtitle", "reason": code},
        )
    if hit is None:
        return Response(
            status_code=202, media_type="application/json", content='{"state":"preparing"}'
        )
    return Response(subtitles.shift_vtt(hit, offset), media_type="text/vtt")


# Resume/watched thresholds. _PROGRESS_MIN_S matches the feed's PROGRESS_MIN_SECONDS (5), but
# _FINISH_MARGIN_S is DELIBERATELY larger than the feed's FINISH_MARGIN_SECONDS (30 vs 10): this is
# long-form content (movies/episodes with credits) and crossing it MARKS THE ITEM WATCHED, whereas
# the feed's margin only clears a short video's resume bar — different domains, intentional divergence,
# not drift. Keep them separate on purpose (see routes/feed.py).
_PROGRESS_MIN_S = 5

_FINISH_MARGIN_S = 30

# S2 write-back throttle: the resume position is now pushed to Plex on the periodic (non-final)
# checkpoint too — not only on the final pause/unmount flush — so Plex (the source of truth) stays
# current in near-real-time. To keep a single watch from spraying `/:/timeline` calls, a push is
# rate-limited to at most one per item per user every _PUSH_THROTTLE_S seconds; a `final` checkpoint
# (pause / seek / stop) always pushes immediately and resets the window. This is an in-memory, single-
# process rate-limiter (the deployment is one FastAPI process): after a restart the map is empty, so at
# most one extra push happens on the first checkpoint — harmless. It is NOT durable domain state.
_PUSH_THROTTLE_S = 10
_last_push_at: dict[tuple[int, int], datetime] = {}
_push_throttle_lock = threading.Lock()


def _should_push_resume(user_id: int, item_id: int, now: datetime, final: bool) -> bool:
    """Whether to mirror this resume checkpoint to Plex now. `final` always pushes; a periodic tick
    pushes only when ≥_PUSH_THROTTLE_S has elapsed since the last push for this (user, item). Records
    `now` as the last-push time whenever it returns True so the window is measured from the real push
    (and so `_resume_push_superseded` can drop an older, still-pending push — see there)."""
    with _push_throttle_lock:
        last = _last_push_at.get((user_id, item_id))
        if final or last is None or (now - last).total_seconds() >= _PUSH_THROTTLE_S:
            _last_push_at[(user_id, item_id)] = now
            # Opportunistic eviction so the map can't grow unbounded over a long-lived process (one
            # entry per item ever checkpointed). Anything older than a few windows is dead — a push
            # completes in seconds, so a stale entry protects nothing.
            if len(_last_push_at) > 2048:
                cutoff = now - timedelta(seconds=_PUSH_THROTTLE_S * 6)
                for k in [k for k, t in _last_push_at.items() if t < cutoff]:
                    del _last_push_at[k]
            return True
        return False


def _resume_push_superseded(user_id: int, item_id: int, intent: datetime) -> bool:
    """True if a newer resume push for this (user, item) was scheduled after `intent`. Called at the
    HTTP boundary of a background resume push: because periodic and seek/pause pushes are independent
    background tasks with no ordering guard, an older still-pending periodic push could otherwise land
    at Plex AFTER a newer backward-seek push and — with the Plex-authoritative pull — durably revive the
    stale position. Skipping a superseded push keeps the last-scheduled position the one Plex settles on."""
    with _push_throttle_lock:
        return _last_push_at.get((user_id, item_id), intent) > intent


@router.get("/person-image")
def person_image(u: str, request: Request) -> Response:
    """Proxy a cast/crew photo from Plex's public metadata CDN (host-whitelisted; no token needed).
    Signed-session auth (no DB) so it, like /image, never holds a pool connection during the fetch."""
    _require_session(request)
    if not u.startswith(plex_people.PERSON_IMG_HOST):
        raise HTTPException(status_code=400, detail="Unsupported image host")
    cache_key = "person_" + hashlib.sha1(u.encode()).hexdigest()
    hit = imgcache.read(cache_key)
    if hit:
        return imgcache.response(*hit)
    try:
        with httpx.Client(timeout=10.0, follow_redirects=True) as c:
            r = c.get(u)
            r.raise_for_status()
    except httpx.HTTPError:
        raise HTTPException(status_code=502, detail="Image fetch failed")
    ct = r.headers.get("content-type", "image/jpeg")
    imgcache.write(cache_key, r.content, ct)
    return imgcache.response(r.content, ct)


def _push_resume_guarded(
    user_id: int,
    item_id: int,
    rating_key: str,
    time_ms: int,
    duration_ms: int,
    expect_ts: datetime | None,
    intent: datetime,
) -> None:
    """Background entrypoint for a resume push that drops itself if a newer resume push for the same
    (user, item) was scheduled after it (see `_resume_push_superseded`), then delegates to the real
    push. This narrows the periodic-vs-seek write-ordering race to the (rare) case where the older push
    has already left the process before the newer one is scheduled."""
    if _resume_push_superseded(user_id, item_id, intent):
        return
    plex_watch.push_state_to_plex(
        user_id, item_id, rating_key, "resume", time_ms, duration_ms, expect_ts
    )


def _push_watch(
    background: BackgroundTasks,
    db: Session,
    user_id: int,
    item_id: int,
    rating_key: str,
    action: str,
    time_ms: int = 0,
    duration_ms: int = 0,
    intent: datetime | None = None,
) -> None:
    """Schedule a best-effort Siftlode→Plex watch push, but only for a user whose owner link has
    push enabled — non-owner users never touch Plex. Guarding here (with the request session) avoids
    spinning up a throwaway background session on every state change for the common no-link case.

    A `resume` push with an `intent` timestamp is routed through `_push_resume_guarded` so a superseded
    (older) push can't land out of order; `watched`/`unwatched` scrobbles go straight through."""
    if plex_watch.link_for_push(db, user_id) is None:
        return
    # Capture the row's freshest timestamp NOW so the background push can tell if a newer local edit
    # lands before it settles synced_to_plex (which would otherwise strand that newer value).
    st = (
        db.execute(
            select(PlexState).where(PlexState.user_id == user_id, PlexState.item_id == item_id)
        )
        .scalars()
        .first()
    )
    marker = plex_watch._state_marker(st)
    if action == "resume" and intent is not None:
        background.add_task(
            _push_resume_guarded, user_id, item_id, rating_key, time_ms, duration_ms, marker, intent
        )
        return
    background.add_task(
        plex_watch.push_state_to_plex,
        user_id,
        item_id,
        rating_key,
        action,
        time_ms,
        duration_ms,
        marker,
    )


class ItemProgressIn(BaseModel):
    # Finite only: JSON parsing here accepts `Infinity`/`NaN`, and the handler's `int()` turned
    # one into a 500.
    model_config = ConfigDict(allow_inf_nan=False)

    # float, truncated in the handler — the player may post a raw currentTime (a float); a plain
    # `int` field would 422 it on the periodic progress checkpoint (the old `int(...)` truncated).
    # Bounded above by the int4 column (see POSITION_MAX_S): past it the flush was a 500.
    position_seconds: float = Field(0, le=POSITION_MAX_S)
    duration_seconds: float = Field(0, le=POSITION_MAX_S)
    final: bool = False


@router.post("/item/{rating_key}/progress")
def item_progress(
    rating_key: str,
    body: ItemProgressIn,
    background: BackgroundTasks,
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    """Checkpoint the resume position while playing (near the end → mark watched + clear it).

    S2 write-back: the resume position is mirrored to Plex (the source of truth) on the periodic
    checkpoint too, not only on the final flush — throttled to ≥_PUSH_THROTTLE_S per item (see
    `_should_push_resume`). `final` (sent on pause / seek / pagehide / unmount) pushes immediately and
    resets the throttle window, so Plex sees a settled position the instant the user pauses or seeks."""
    it = item_or_404(db, rating_key)
    # An open player checkpoints every 10 s whether or not it is playing, so this is the signal that a
    # PAUSED viewer still wants their HLS session — the one thing `last_access` could not see, since
    # it moved only on a file fetch and a paused player fetches nothing (see stream.reap_idle).
    plex_stream.touch_session(str(rating_key))
    position = max(0, int(body.position_seconds))
    duration = int(body.duration_seconds)
    final = body.final
    now = datetime.now(timezone.utc)
    st = (
        db.execute(
            select(PlexState).where(PlexState.user_id == user.id, PlexState.item_id == it.id)
        )
        .scalars()
        .first()
    )
    was_watched = st is not None and st.status == "watched"

    near_end = duration > 0 and position > duration - _FINISH_MARGIN_S
    if near_end:
        if st is None:
            st = PlexState(user_id=user.id, item_id=it.id)
            db.add(st)
        st.status = "watched"
        st.watched_at = now
        st.position_seconds = 0
        st.progress_updated_at = now
        st.synced_to_plex = False
        # Terminal state for resume write-back: drop the throttle/ordering entry so the map doesn't
        # retain a dead key for every finished item.
        with _push_throttle_lock:
            _last_push_at.pop((user.id, it.id), None)
        db.commit()
        # Immediate on the watched transition; skip a repeat scrobble if it was already watched
        # (a later checkpoint of the same finished item would otherwise bump Plex's viewCount again).
        if not was_watched:
            _push_watch(background, db, user.id, it.id, rating_key, "watched")
        return {"status": "watched", "position_seconds": 0}

    if position < _PROGRESS_MIN_S:
        if st is not None and st.status == "new":
            db.delete(st)
        elif st is not None:
            st.position_seconds = 0
            st.progress_updated_at = now
        db.commit()
        return {"position_seconds": 0}

    if st is None:
        st = PlexState(user_id=user.id, item_id=it.id)
        db.add(st)
    st.position_seconds = position
    st.progress_updated_at = now
    push_now = _should_push_resume(user.id, it.id, now, final)
    if push_now:
        # Mark dirty before the (background) push so the reconcile knows to re-push if it fails, and so
        # `push_state_to_plex` can settle synced_to_plex once Plex confirms.
        st.synced_to_plex = False
    db.commit()
    if push_now:
        _push_watch(
            background,
            db,
            user.id,
            it.id,
            rating_key,
            "resume",
            position * 1000,
            duration * 1000,
            intent=now,
        )
    return {"position_seconds": position}


class ItemStateIn(BaseModel):
    status: str = ""


@router.post("/item/{rating_key}/state")
def item_state(
    rating_key: str,
    body: ItemStateIn,
    background: BackgroundTasks,
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    """Set the per-user watch status (new|watched|hidden). watched→scrobble / →new→unscrobble push
    to a linked Plex account immediately; `hidden` is a Siftlode-only concept with no Plex analogue,
    so it never touches Plex."""
    it = item_or_404(db, rating_key)
    status = body.status
    if status not in ("new", "watched", "hidden"):
        raise HTTPException(status_code=400, detail="Invalid status")
    st = (
        db.execute(
            select(PlexState).where(PlexState.user_id == user.id, PlexState.item_id == it.id)
        )
        .scalars()
        .first()
    )
    prev_status = st.status if st is not None else "new"
    if status == "new":
        # Does Plex hold anything for this item that the un-mark has to clear? Two things qualify: a
        # view count, and a resume point. An IN-PROGRESS item carries status "new" WITH a position —
        # that is how the importer maps a bare `viewOffset` (`_plex_watch_to_state`) — so keying the
        # push on the status alone silently skipped the single case "Clear resume point" exists for:
        # the un-mark never reached Plex, the offset stayed, and the next pull imported it straight
        # back (Protector, cleared twice by the user and back at 92s both times). `hidden` is
        # Siftlode-only and still pushes nothing, exactly as this endpoint's docstring says.
        mirrored_on_plex = st is not None and (
            st.status == "watched" or (st.position_seconds or 0) > 0
        )
        if st is not None:
            db.delete(st)
            db.commit()
        if mirrored_on_plex:
            _push_watch(background, db, user.id, it.id, rating_key, "unwatched")
        return {"status": "new"}
    if st is None:
        st = PlexState(user_id=user.id, item_id=it.id)
        db.add(st)
    st.status = status
    if status == "watched":
        st.watched_at = datetime.now(timezone.utc)
        st.position_seconds = 0
    st.synced_to_plex = False
    db.commit()
    if status == "watched" and prev_status != "watched":
        _push_watch(background, db, user.id, it.id, rating_key, "watched")
    return {"status": status}


def _bulk_state(
    background: BackgroundTasks,
    db: Session,
    user_id: int,
    eps: list[PlexItem],
    watched: bool,
) -> int:
    """Mark every episode in `eps` watched (or unwatched) for a user, mirroring each change to a
    linked Plex account in the background (best-effort). Returns how many rows actually changed."""
    now = datetime.now(timezone.utc)
    existing = {
        s.item_id: s
        for s in db.execute(
            select(PlexState).where(
                PlexState.user_id == user_id,
                PlexState.item_id.in_([e.id for e in eps] or [0]),
            )
        ).scalars()
    }
    pushable = plex_watch.link_for_push(db, user_id) is not None
    to_push: list[tuple[int, str, str]] = []
    for e in eps:
        st = existing.get(e.id)
        prev = st.status if st is not None else "new"
        if watched:
            if st is None:
                st = PlexState(user_id=user_id, item_id=e.id)
                db.add(st)
            st.status = "watched"
            st.watched_at = now
            st.position_seconds = 0
            st.synced_to_plex = False
            if prev != "watched":
                to_push.append((e.id, e.rating_key, "watched"))
        else:  # → new: drop any state (but keep the user's private "hidden" marks)
            if st is not None and prev != "hidden":
                db.delete(st)
                if prev != "new":
                    to_push.append((e.id, e.rating_key, "unwatched"))
    db.commit()
    if pushable and to_push:
        # One coalesced background task for the whole batch (reuses a single session + Plex client),
        # not one task per episode — a 200-ep show mark is 1 task, not 200.
        background.add_task(plex_watch.push_bulk_state_to_plex, user_id, to_push)
    return len(to_push)


class WatchedIn(BaseModel):
    watched: bool = False


@router.post("/show/{rating_key}/state")
def show_state(
    rating_key: str,
    body: WatchedIn,
    background: BackgroundTasks,
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    """Mark a whole show watched/unwatched (all its episodes) for the current user + push to Plex."""
    sh = by_rating_key(db, PlexShow, rating_key)
    if sh is None:
        raise HTTPException(status_code=404, detail="Unknown Plex show")
    watched = body.watched
    eps = (
        db.execute(select(PlexItem).where(PlexItem.show_id == sh.id, PlexItem.kind == "episode"))
        .scalars()
        .all()
    )
    changed = _bulk_state(background, db, user.id, eps, watched)
    return {"changed": changed, "watched": watched}


@router.post("/season/{rating_key}/state")
def season_state(
    rating_key: str,
    body: WatchedIn,
    background: BackgroundTasks,
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    """Mark a whole season watched/unwatched (all its episodes) for the current user + push to Plex."""
    se = by_rating_key(db, PlexSeason, rating_key)
    if se is None:
        raise HTTPException(status_code=404, detail="Unknown Plex season")
    watched = body.watched
    eps = (
        db.execute(select(PlexItem).where(PlexItem.season_id == se.id, PlexItem.kind == "episode"))
        .scalars()
        .all()
    )
    changed = _bulk_state(background, db, user.id, eps, watched)
    return {"changed": changed, "watched": watched}


@router.post("/stream/{rating_key}/session")
def stream_session(
    rating_key: str,
    # Finite: `start=inf` reached `int()` in the session builder and came back as a 500.
    start: float = Query(0.0, allow_inf_nan=False),
    audio: int | None = None,
    aoff: float = Query(0.0, allow_inf_nan=False),
    multi: bool = False,
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    """Prepare playback from `start` seconds. Direct-playable files are served raw; remux files
    (re)start an HLS session at the offset (seek-restart, video stream-copied); transcode files (non-h264:
    HEVC/VP9/AV1/…) take the same HLS path but with the video re-encoded to H.264 (VAAPI or libx264).
    Selecting a non-default audio track (`audio` = stream ordinal) forces the HLS path so the chosen audio
    can be muxed in. Subtitles are independent WebVTT tracks (GET /subtitle) — no session restart."""
    it = item_or_404(db, rating_key)
    transcode = it.playable == "transcode"
    # `multi` (item has ≥2 audio tracks) forces the HLS path so ALL audio tracks ship as renditions and
    # the client can switch audio without a restart — even for otherwise direct-playable files.
    if it.playable == "direct" and audio is None and not aoff and not multi:
        return {
            "mode": "direct",
            "url": f"/api/plex/stream/{it.rating_key}/file",
            "start_s": 0.0,
        }
    if plex_paths.local_media_path(db, it.file_path) is None:
        raise HTTPException(status_code=404, detail="Media file not found on disk")
    s = plex_stream.start_session(db, it, start, audio, aoff, multi, transcode=transcode)
    if s is None:
        raise HTTPException(status_code=404, detail="Media file not found on disk")
    # Report the REAL media start (the keyframe ffmpeg landed on, which hls.js zero-bases to), not the
    # requested offset — the client's absolute clock and the subtitle cue-shift both key off this.
    # `transcode`/`hw`/`tonemap` let the player show an honest cold-start ("Preparing stream…") and
    # surface the active mode + encoder in the stats overlay (`tonemap` = the HW HDR→SDR path); both
    # segment streams are otherwise identical.
    return {
        "mode": "hls",
        "start_s": s.media_start_s,
        # The session id is part of the path (not a query param) so the playlist's RELATIVE segment
        # names resolve back into the same session — see `stream_hls_file` for why it has to.
        "url": f"/api/plex/stream/{it.rating_key}/hls/{s.id}/{s.entry}",
        "transcode": s.transcode,
        "hw": s.hw,
        "tonemap": s.tonemap,
    }


@router.get("/stream/{rating_key}/file")
def stream_file(
    rating_key: str,
    download: bool = False,
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> FileResponse:
    """Serve the raw local media file with HTTP range support. Used both for direct-play (inline)
    and for the player's "download original" button (`?download=1` → attachment). The download is
    the untouched physical file — no re-encode/repackage."""
    it = item_or_404(db, rating_key)
    src = plex_paths.local_media_path(db, it.file_path)
    if src is None:
        raise HTTPException(status_code=404, detail="Media file not found on disk")
    if download:
        # A clean, safe download name from the title + the real file extension.
        ext = src.suffix or (f".{it.container}" if it.container else "")
        base = (
            "".join(c for c in (it.title or "video") if c.isalnum() or c in " -_.").strip()
            or "video"
        )
        return FileResponse(str(src), filename=f"{base}{ext}")
    return FileResponse(str(src))  # Starlette honours the Range header (206)


_HLS_CT = {
    ".m3u8": "application/vnd.apple.mpegurl",
    ".ts": "video/mp2t",
    ".vtt": "text/vtt",
}


@router.get("/stream/{rating_key}/hls/{session_id}/{filename}")
def stream_hls_file(
    rating_key: str,
    session_id: str,
    filename: str,
    user: User = Depends(current_user),
) -> Response:
    """Serve a file (playlist / segment / subtitle) from ONE HLS session's directory. The playlist
    references its segments/subtitles by relative name, all resolving back here.

    The session id in the path is what makes "back here" mean the same session. Sessions are keyed by
    rating_key alone and a seek-restart or a second viewer of the same item replaces the registered
    one, while the DIRECTORY is per (key, offset, tag) — so resolving by rating_key handed the old
    viewer the new session's `seg_12.ts`, produced from a different offset under `-copyts`: wrong
    bytes with a 200, which the player has no way to notice (deep-review round 5). A request for a
    session that is no longer the registered one is a 404, and the player reclaims at the playhead.

    That reclaim is also why a request with no live session is no longer auto-started here: the
    auto-start could only ever begin at 0, silently throwing away the viewer's position."""
    # Filename must be a plain hls artifact — no path separators, only the expected extensions.
    ext = "." + filename.rsplit(".", 1)[-1] if "." in filename else ""
    if ext not in _HLS_CT or "/" in filename or "\\" in filename or ".." in filename:
        raise HTTPException(status_code=404, detail="Not found")
    s = plex_stream.current_session(str(rating_key))
    if s is None or s.id != session_id:
        raise HTTPException(status_code=404, detail="No active playback session")
    path = s.dir / filename
    # A segment beyond the generated region can never arrive: ffmpeg writes them in order, so if the
    # predecessor is not there this one is not next. Answering now instead of waiting it out is what
    # keeps a seek past the produced region — or a deliberate flood of hopeless names — from parking
    # threadpool workers and their DB connections (see `segment_may_arrive`).
    if not plex_stream.segment_may_arrive(s.dir, filename):
        raise HTTPException(status_code=404, detail="Not available yet")
    timeout = 30 if ext == ".m3u8" else 20
    # `proc` so a session whose ffmpeg has already exited answers at once instead of polling out the
    # whole timeout: this is a SYNC route on the ~40-slot threadpool, holding a pooled DB connection
    # with it, so every such wait is a worker (and a connection) parked for 20 s — an ordinary seek
    # past the produced region was enough to do it (round 5).
    if not plex_stream.wait_for(path, timeout, proc=s.proc):
        raise HTTPException(status_code=404, detail="Not available yet")
    return Response(
        content=path.read_bytes(),
        media_type=_HLS_CT[ext],
        headers={"Cache-Control": "no-store"},
    )