feat(plex): P2 streaming backend — direct 206 + seek-restart HLS remux

Validated end-to-end on a real remux file (h264+ac3 mkv). The riskiest part of the
Plex epic — on-the-fly playback from the local file — now proven.

- app/plex/stream.py: seek-restart HLS session model (one ffmpeg per item, restarted
  at a seek offset via -ss). Video stream-COPY (I/O-bound, CPU-light — fine on the
  GPU-less prod host) + audio→aac only when not already aac. Maps first video+audio,
  drops subtitles (no VTT rendition clutter). Session cap + idle reaper.
- routes: POST /stream/{rk}/session?start= (direct→raw url, remux→HLS session,
  transcode→501 P3), GET /stream/{rk}/file (206 range, direct), /index.m3u8, /seg_n.ts.
- KEY FINDINGS (hard-won): (1) ffmpeg -hls_playlist_type VOD does NOT write the
  playlist mid-run (only at finalize) → use EVENT (append-only, appears immediately),
  which suits the seek-restart model; (2) naive per-segment copy + keyframe-indexed
  extraction are both unreliable (non-uniform/imprecise segments) — ffmpeg's own HLS
  muxer is the only correct segmenter; (3) dev slowness was the Windows /downloads
  bind-mount write + SMB read → PLEX_HLS_DIR env points scratch at fast container-local
  /var/tmp on dev (prod keeps download_root, fast local disk).

backend/app/config.py

@@ -205,6 +205,11 @@ class Settings(BaseSettings):
     # Max concurrent transcodes for the P3 fallback. Low by default — CPU-only transcode is
     # expensive; direct-serve (browser-compatible files) has no such limit.
     plex_max_transcodes: int = 1
+    # Where on-the-fly HLS remux segments are written (transient, reaped). Empty → a `.plex-hls`
+    # dir under download_root. On localdev the download_root is a slow Windows bind-mount, so point
+    # this at a fast container-local path (e.g. /var/tmp/plex-hls); prod's download_root is fast
+    # local disk, so the default is fine there.
+    plex_hls_dir: str = ""
 
     @property
     def allowed_email_set(self) -> set[str]:

backend/app/plex/stream.py

@@ -0,0 +1,163 @@
+"""On-the-fly HLS remux for Plex playback (seek-restart session model).
+
+Playback is from the LOCAL physical file. Browser-compatible files (playable="direct") are served
+raw with HTTP range requests. Everything else that is h264 video (playable="remux") is remuxed on
+the fly to HLS with **video stream-copy** (cheap, I/O-bound — no video re-encode) and audio to AAC
+when needed. HEVC/VP9 (playable="transcode") needs a full re-encode and is deferred to P3.
+
+Seek model (Jellyfin-style): one ffmpeg session per item, started at a given offset via `-ss`
+(fast keyframe seek). A seek beyond the generated region restarts the session at the new offset,
+so seeking is responsive even on long movies without pre-generating the whole file. ffmpeg's own
+HLS muxer does the segmentation (the only reliable way to cut a stream-copy at keyframes).
+
+Sessions live in this (single) API process; a periodic reaper kills idle ones and frees the temp
+segments. The remux is CPU-light (video copy + a tiny audio transcode), so it runs fine on the
+CPU-only prod host; only P3's full transcode is CPU-heavy.
+"""
+import logging
+import shutil
+import subprocess
+import threading
+import time
+from pathlib import Path
+
+from sqlalchemy.orm import Session as DbSession
+
+from app.config import settings
+from app.models import PlexItem
+from app.plex import paths
+
+log = logging.getLogger("siftlode.plex")
+
+_HLS_ROOT = Path(settings.plex_hls_dir) if settings.plex_hls_dir else Path(settings.download_root) / ".plex-hls"
+_SEG_SECONDS = 6
+_MAX_SESSIONS = 4  # concurrent remux sessions (safety valve for the CPU-only host)
+_SESSION_IDLE_S = 600  # reap a session with no access for this long
+
+_lock = threading.Lock()
+_sessions: dict[str, "HlsSession"] = {}
+
+
+class HlsSession:
+    def __init__(self, key: str, directory: Path, proc: subprocess.Popen, start_s: float):
+        self.key = key
+        self.dir = directory
+        self.proc = proc
+        self.start_s = start_s
+        self.last_access = time.time()
+
+
+def _acodec(item: PlexItem) -> str:
+    # Copy the audio when it's already AAC; otherwise transcode (ac3/dts/… → aac). Video is always
+    # copied for the remux class.
+    return "copy" if (item.codec_audio or "").lower() == "aac" else "aac"
+
+
+def _ffmpeg_cmd(src: Path, start_s: float, out_dir: Path, acodec: str) -> list[str]:
+    args = ["ffmpeg", "-nostdin", "-loglevel", "error"]
+    if start_s > 0:
+        args += ["-ss", f"{start_s:.3f}"]  # fast keyframe seek before -i
+    # Map only the first video + first audio; drop subtitles so ffmpeg's HLS muxer doesn't spawn
+    # WebVTT renditions (subtitle/multi-audio selection is a later refinement served separately).
+    args += ["-i", str(src), "-map", "0:v:0", "-map", "0:a:0?", "-sn", "-c:v", "copy", "-c:a", acodec]
+    if acodec == "aac":
+        args += ["-ac", "2", "-b:a", "192k"]
+    args += [
+        "-f", "hls",
+        "-hls_time", str(_SEG_SECONDS),
+        "-hls_list_size", "0",
+        # EVENT (not VOD): ffmpeg writes/appends the playlist AS segments complete, so playback can
+        # start immediately from this session's offset. VOD only writes the playlist at the end. The
+        # full seekbar comes from our known duration + the seek-restart model, not from the playlist.
+        "-hls_playlist_type", "event",
+        "-hls_segment_type", "mpegts",
+        "-hls_flags", "independent_segments+temp_file",
+        "-hls_segment_filename", str(out_dir / "seg_%d.ts"),
+        str(out_dir / "index.m3u8"),
+    ]
+    return args
+
+
+def _kill(s: HlsSession) -> None:
+    try:
+        s.proc.terminate()
+        try:
+            s.proc.wait(timeout=3)
+        except subprocess.TimeoutExpired:
+            s.proc.kill()
+    except Exception:
+        pass
+    shutil.rmtree(s.dir, ignore_errors=True)
+
+
+def _enforce_cap() -> None:
+    # Caller holds _lock. Drop the least-recently-accessed sessions over the cap.
+    if len(_sessions) <= _MAX_SESSIONS:
+        return
+    for key, s in sorted(_sessions.items(), key=lambda kv: kv[1].last_access)[: len(_sessions) - _MAX_SESSIONS]:
+        _kill(s)
+        _sessions.pop(key, None)
+
+
+def start_session(db: DbSession, item: PlexItem, start_s: float) -> HlsSession | None:
+    """(Re)start the HLS remux for an item at the given offset. Returns None if the local file
+    can't be read."""
+    src = paths.local_media_path(db, item.file_path)
+    if src is None:
+        return None
+    key = item.rating_key
+    start_s = max(0.0, float(start_s))
+    acodec = _acodec(item)
+    with _lock:
+        old = _sessions.pop(key, None)
+        if old is not None:
+            _kill(old)
+        directory = _HLS_ROOT / f"{key}_{int(start_s)}"
+        shutil.rmtree(directory, ignore_errors=True)
+        directory.mkdir(parents=True, exist_ok=True)
+        proc = subprocess.Popen(
+            _ffmpeg_cmd(src, start_s, directory, acodec),
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
+        )
+        s = HlsSession(key, directory, proc, start_s)
+        _sessions[key] = s
+        _enforce_cap()
+        log.info("plex hls session start key=%s start=%.1f acodec=%s", key, start_s, acodec)
+        return s
+
+
+def current_session(key: str) -> HlsSession | None:
+    with _lock:
+        s = _sessions.get(key)
+        if s is not None:
+            s.last_access = time.time()
+        return s
+
+
+def wait_for(path: Path, timeout: float = 20.0) -> bool:
+    """Wait until a session file (playlist / segment) exists and is non-empty. Segments are
+    produced ~faster than realtime, so a segment just ahead of playback appears quickly; a segment
+    far beyond the generated region won't (the frontend restarts the session at a seek instead)."""
+    end = time.time() + timeout
+    while time.time() < end:
+        try:
+            if path.exists() and path.stat().st_size > 0:
+                return True
+        except OSError:
+            pass
+        time.sleep(0.15)
+    return path.exists()
+
+
+def reap_idle() -> int:
+    now = time.time()
+    dropped = 0
+    with _lock:
+        for key, s in list(_sessions.items()):
+            done = s.proc.poll() is not None
+            if now - s.last_access > _SESSION_IDLE_S or (done and now - s.last_access > 30):
+                _kill(s)
+                _sessions.pop(key, None)
+                dropped += 1
+    return dropped

backend/app/routes/plex.py

@@ -8,6 +8,7 @@ catalog sync.
 import logging
 
 from fastapi import APIRouter, Depends, HTTPException, Query, Response
+from fastapi.responses import FileResponse
 from sqlalchemy import and_, func, or_
 from sqlalchemy.orm import Session, aliased
 
@@ -15,6 +16,8 @@ from app import sysconfig
 from app.auth import current_user
 from app.db import get_db
 from app.models import PlexItem, PlexLibrary, PlexSeason, PlexShow, PlexState, User
+from app.plex import paths as plex_paths
+from app.plex import stream as plex_stream
 from app.plex import sync as plex_sync
 from app.plex.client import PlexClient, PlexError, PlexNotConfigured
 from app.routes.admin import admin_user
@@ -279,3 +282,92 @@ def image(
     except PlexError as e:
         raise HTTPException(status_code=502, detail=f"Plex image fetch failed: {e}")
     return Response(content=data, media_type=ct, headers={"Cache-Control": "public, max-age=86400"})
+
+
+# --- Playback (local file → direct 206 or on-the-fly HLS remux; P2) ----------------------------
+
+
+def _item_or_404(db: Session, rating_key: str) -> PlexItem:
+    it = db.query(PlexItem).filter_by(rating_key=str(rating_key)).first()
+    if it is None:
+        raise HTTPException(status_code=404, detail="Unknown Plex item")
+    return it
+
+
+@router.post("/stream/{rating_key}/session")
+def stream_session(
+    rating_key: str,
+    start: float = 0.0,
+    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); transcode (HEVC/VP9) is P3."""
+    plex_stream.reap_idle()  # opportunistic cleanup of idle sessions on each new playback
+    it = _item_or_404(db, rating_key)
+    if it.playable == "direct":
+        return {"mode": "direct", "url": f"/api/plex/stream/{it.rating_key}/file", "start_s": 0.0}
+    if it.playable == "transcode":
+        raise HTTPException(status_code=501, detail="This format needs transcoding (a later phase).")
+    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)
+    if s is None:
+        raise HTTPException(status_code=404, detail="Media file not found on disk")
+    return {"mode": "hls", "start_s": s.start_s, "url": f"/api/plex/stream/{it.rating_key}/index.m3u8"}
+
+
+@router.get("/stream/{rating_key}/file")
+def stream_file(
+    rating_key: str,
+    user: User = Depends(current_user),
+    db: Session = Depends(get_db),
+) -> FileResponse:
+    """Serve the raw local media file with HTTP range support (direct-playable files)."""
+    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")
+    return FileResponse(str(src))  # Starlette honours the Range header (206)
+
+
+@router.get("/stream/{rating_key}/index.m3u8")
+def stream_playlist(
+    rating_key: str,
+    user: User = Depends(current_user),
+    db: Session = Depends(get_db),
+) -> Response:
+    """The current HLS session's playlist (starts a session at 0 if none is running)."""
+    s = plex_stream.current_session(str(rating_key))
+    if s is None:
+        it = _item_or_404(db, rating_key)
+        if it.playable not in ("remux",):
+            raise HTTPException(status_code=400, detail="Not an HLS-remux item")
+        s = plex_stream.start_session(db, it, 0.0)
+        if s is None:
+            raise HTTPException(status_code=404, detail="Media file not found on disk")
+    if not plex_stream.wait_for(s.dir / "index.m3u8", 30):
+        raise HTTPException(status_code=503, detail="Playlist not ready")
+    return Response(
+        content=(s.dir / "index.m3u8").read_bytes(),
+        media_type="application/vnd.apple.mpegurl",
+        headers={"Cache-Control": "no-store"},
+    )
+
+
+@router.get("/stream/{rating_key}/seg_{n}.ts")
+def stream_segment(
+    rating_key: str,
+    n: int,
+    user: User = Depends(current_user),
+) -> Response:
+    """Serve one HLS segment from the current session (waits briefly if it's just ahead)."""
+    s = plex_stream.current_session(str(rating_key))
+    if s is None:
+        raise HTTPException(status_code=404, detail="No active playback session")
+    seg = s.dir / f"seg_{n}.ts"
+    if not plex_stream.wait_for(seg, 20):
+        raise HTTPException(status_code=404, detail="Segment not available")
+    return Response(
+        content=seg.read_bytes(), media_type="video/mp2t", headers={"Cache-Control": "no-store"}
+    )

docker-compose.localdev.yml

@@ -46,6 +46,9 @@ services:
       # Download center: the API serves finished files (it does NOT run the worker loop).
       DOWNLOAD_ROOT: /downloads
       WORKER_ENABLED: "false"
+      # Plex HLS remux scratch on fast container-local storage (the /downloads bind-mount is a slow
+      # Windows Docker Desktop mount in dev; prod uses fast local disk so it keeps the default).
+      PLEX_HLS_DIR: /var/tmp/plex-hls
     depends_on:
       db:
         condition: service_healthy