"""Bounded, generation-safe media downloading and canonical slot cleanup."""

from __future__ import annotations

import json
import logging
import os
import re
import signal
import shutil
import stat
import subprocess
import tempfile
import threading
import time
import unicodedata
import uuid
from collections.abc import Iterator
from contextlib import contextmanager
from concurrent.futures import Future, ThreadPoolExecutor
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable

from .config import ControllerConfig
from .contracts import SLOT_IDS
from .models import (
    DeleteToken,
    DownloadReservation,
    ExportReservation,
    SlotProtectedError,
    StaleOperationError,
    UTILITY_EXPORT_MODES,
)
from .store import ControllerStore


LOG = logging.getLogger(__name__)
MEDIA_EXTENSIONS = frozenset({".mp4", ".mkv", ".webm", ".mov", ".m4v"})
VIDEO_EXPORT_INPUT_EXTENSIONS = frozenset(
    {*MEDIA_EXTENSIONS, ".avi", ".flv", ".ts"}
)
EXPORT_AUDIO_EXTENSIONS = {"m4a": ".m4a", "mp3": ".mp3"}
MAX_EXPORT_STEM_BYTES = 180
EXPORT_MANIFEST_NAME = "manifest.json"
_INVALID_FILENAME_CHARACTERS = frozenset('<>:"/\\|?*')
_WINDOWS_RESERVED_NAMES = frozenset(
    {"CON", "PRN", "AUX", "NUL"}
    | {f"COM{index}" for index in range(1, 10)}
    | {f"LPT{index}" for index in range(1, 10)}
)


@dataclass(frozen=True)
class DownloadEvent:
    slot: int
    state: str
    title: str | None = None
    count: int | None = None
    message: str | None = None
    job_id: str | None = None
    generation: int | None = None
    url: str | None = None
    submitter: str | None = None
    note: str | None = None
    reused_slot: bool = False


@dataclass(frozen=True)
class MediaInspection:
    format_names: frozenset[str]
    video_codec: str | None
    audio_codec: str | None
    pixel_format: str | None
    width: int | None
    height: int | None


EventCallback = Callable[[DownloadEvent], None]
Runner = Callable[..., subprocess.CompletedProcess[str]]


def canonical_slot_directory(media_base: Path, slot: int) -> Path:
    if slot not in SLOT_IDS:
        raise ValueError(f"slot must be in 1..10, got {slot!r}")
    return media_base / f"slot{slot}"


def canonical_playlist_path(media_base: Path, slot: int) -> Path:
    return canonical_slot_directory(media_base, slot) / "playlist.m3u"


def player_format_selector(max_height: int) -> str:
    """Select the best Player media without exceeding its configured height."""

    return f"bv*[height<={max_height}]+ba/b[height<={max_height}]"


def mp4_export_format_selector(max_height: int) -> str:
    """Select quality first while making native AVC/AAC the equal-quality preference."""

    return f"bv*[height<={max_height}]+ba/b[height<={max_height}]"


def safe_export_filename(
    stem: str,
    extension: str,
    *,
    used_casefolded: set[str] | None = None,
) -> str:
    """Return a bounded Unicode filename that cannot introduce a path component."""

    if extension not in {".mp4", ".m4a", ".mp3"}:
        raise ValueError("unsupported export filename extension")
    normalized = unicodedata.normalize("NFC", str(stem))
    cleaned = "".join(
        "_"
        if character in _INVALID_FILENAME_CHARACTERS
        or ord(character) < 32
        or ord(character) == 127
        else character
        for character in normalized
    )
    cleaned = re.sub(r"\s+", " ", cleaned).strip(" .") or "untitled"
    if cleaned.split(".", 1)[0].upper() in _WINDOWS_RESERVED_NAMES:
        cleaned = f"_{cleaned}"

    def bounded(value: str, suffix: str) -> str:
        available = MAX_EXPORT_STEM_BYTES - len(suffix.encode("utf-8"))
        candidate = value
        while candidate and len(candidate.encode("utf-8")) > available:
            candidate = candidate[:-1]
        return (candidate.rstrip(" .") or "untitled") + suffix

    used = used_casefolded if used_casefolded is not None else set()
    counter = 1
    while True:
        collision_suffix = "" if counter == 1 else f" ({counter})"
        filename = bounded(cleaned, collision_suffix) + extension
        folded = filename.casefold()
        if folded not in used:
            used.add(folded)
            return filename
        counter += 1


def title_from_filename(path: Path) -> str:
    stem = path.stem
    if " - " in stem:
        stem = stem.split(" - ", 1)[1]
    if " [" in stem:
        stem = stem.rsplit(" [", 1)[0]
    return stem


class DownloadManager:
    """Reserve synchronously, then perform work on a bounded executor."""

    def __init__(
        self,
        config: ControllerConfig,
        store: ControllerStore,
        *,
        on_event: EventCallback | None = None,
        runner: Runner | None = None,
        executor: ThreadPoolExecutor | None = None,
        export_executor: ThreadPoolExecutor | None = None,
    ):
        self.config = config
        self.store = store
        self.on_event = on_event or (lambda event: None)
        self.runner = runner
        self._executor = executor or ThreadPoolExecutor(
            max_workers=config.downloads.max_workers,
            thread_name_prefix="yt-download",
        )
        self._owns_executor = executor is None
        if export_executor is not None:
            self._export_executor = export_executor
            self._owns_export_executor = False
        elif executor is not None:
            # Deterministic callers/tests may inject one executor explicitly.
            self._export_executor = executor
            self._owns_export_executor = False
        else:
            self._export_executor = ThreadPoolExecutor(
                max_workers=config.downloads.export_max_workers,
                thread_name_prefix="yt-export",
            )
            self._owns_export_executor = True
        self._lock = threading.Lock()
        self._futures: set[Future[Any]] = set()
        self._processes: set[subprocess.Popen[str]] = set()
        self._deletions_in_flight: set[tuple[int, int]] = set()
        self._stopping = False

    def enqueue(
        self,
        url: str,
        submitter: str | None,
        note: str | None,
    ) -> DownloadReservation:
        with self._lock:
            if self._stopping:
                return DownloadReservation(status="shutting_down")
        reservation = self.store.reserve_download(url, submitter, note)
        if reservation.status != "reserved":
            return reservation
        assert reservation.slot is not None
        assert reservation.job_id is not None
        assert reservation.generation is not None

        # Publish/record queued after the transaction commits and before a
        # worker can publish downloading or ready.
        self._emit(
            DownloadEvent(
                slot=reservation.slot,
                state="queued",
                job_id=reservation.job_id,
                generation=reservation.generation,
                url=url,
                submitter=submitter,
                note=note,
                reused_slot=reservation.reused_slot,
            )
        )
        try:
            future = self._executor.submit(
                self._download,
                url,
                reservation.slot,
                reservation.generation,
                reservation.job_id,
            )
        except BaseException as exc:
            message = f"could not start download worker: {exc}"
            self.store.fail_download(
                reservation.job_id,
                reservation.slot,
                reservation.generation,
                message,
            )
            self._emit(
                DownloadEvent(
                    slot=reservation.slot,
                    state="error",
                    message=message,
                    job_id=reservation.job_id,
                    generation=reservation.generation,
                )
            )
            return DownloadReservation(
                status="worker_error",
                slot=reservation.slot,
                generation=reservation.generation,
                job_id=reservation.job_id,
                reused_slot=reservation.reused_slot,
            )

        with self._lock:
            self._futures.add(future)
        future.add_done_callback(self._forget_future)
        return reservation

    def enqueue_export(
        self,
        url: str,
        mode: str,
        submitter: str | None,
        note: str | None,
    ) -> ExportReservation:
        """Reserve and queue a utility export without touching Player slots."""

        if mode not in UTILITY_EXPORT_MODES:
            raise ValueError("unsupported utility export mode")
        with self._lock:
            if self._stopping:
                return ExportReservation(status="shutting_down")
        reservation = self.store.reserve_export_job(
            url,
            mode,
            submitter,
            note,
            max_active=self.config.downloads.max_queued_exports,
        )
        if reservation.status != "queued":
            return reservation
        assert reservation.job_id is not None
        try:
            future = self._export_executor.submit(
                self._export,
                url,
                mode,
                reservation.job_id,
            )
        except BaseException as exc:
            message = f"could not start export worker: {exc}"
            self.store.fail_export_job(reservation.job_id, message)
            return ExportReservation(
                status="worker_error",
                job_id=reservation.job_id,
            )
        with self._lock:
            self._futures.add(future)
        future.add_done_callback(self._forget_future)
        return reservation

    def schedule_export_reconciliation(self) -> bool:
        """Queue crash-artifact recovery without delaying core service startup."""

        with self._lock:
            if self._stopping:
                return False
        try:
            future = self._export_executor.submit(self.reconcile_export_artifacts)
        except BaseException:
            LOG.exception("could not schedule utility export reconciliation")
            return False
        with self._lock:
            self._futures.add(future)
        future.add_done_callback(self._forget_future)
        return True

    def _forget_future(self, future: Future[Any]) -> None:
        with self._lock:
            self._futures.discard(future)
        try:
            future.result()
        except BaseException:
            LOG.exception("media worker escaped its failure boundary")

    def _emit(self, event: DownloadEvent) -> None:
        try:
            self.on_event(event)
        except BaseException:
            LOG.exception("download event callback failed")

    def _working_cookie_copy(self) -> Path | None:
        source = self.config.downloads.cookie_path
        if source is None:
            return None
        if source.is_file():
            descriptor, temporary_name = tempfile.mkstemp(
                prefix="ytplayer-cookies-",
                suffix=".txt",
            )
            temporary_path = Path(temporary_name)
            try:
                with os.fdopen(descriptor, "wb") as destination:
                    with source.open("rb") as source_file:
                        shutil.copyfileobj(source_file, destination)
                    destination.flush()
                    os.fsync(destination.fileno())
                temporary_path.chmod(0o600)
                return temporary_path
            except BaseException:
                temporary_path.unlink(missing_ok=True)
                raise
        if self.config.downloads.cookie_required:
            raise FileNotFoundError(
                f"required yt-dlp cookie file is missing: {source}"
            )
        LOG.warning(
            "yt-dlp cookie file is unavailable; continuing without it: %s",
            source,
        )
        return None

    @contextmanager
    def _yt_dlp_arguments(self) -> Iterator[list[str]]:
        working_cookie = self._working_cookie_copy()
        arguments = [
            self.config.downloads.yt_dlp_binary,
            "--ignore-config",
            "--no-js-runtimes",
            "--js-runtimes",
            "node",
            "--cache-dir",
            "/tmp/yt-dlp-cache",
        ]
        if working_cookie is not None:
            arguments.extend(("--cookies", str(working_cookie)))
        try:
            yield arguments
        finally:
            if working_cookie is not None:
                working_cookie.unlink(missing_ok=True)

    def _get_title(self, url: str) -> tuple[str, str | None]:
        try:
            with self._yt_dlp_arguments() as base_arguments:
                arguments = base_arguments + [
                    "--flat-playlist",
                    "--print",
                    "%(playlist_title)s|||%(title)s|||%(id)s",
                    "--playlist-end",
                    "1",
                    url,
                ]
                result = self._run_command(
                    arguments,
                    check=True,
                    capture_output=True,
                    text=True,
                    timeout=min(
                        120.0,
                        self.config.downloads.process_timeout_seconds,
                    ),
                )
            line = (result.stdout or "").strip().splitlines()[0]
            playlist_title, _, remainder = line.partition("|||")
            video_title, _, yt_id = remainder.partition("|||")
            playlist_title = playlist_title.strip()
            video_title = video_title.strip()
            title = (
                playlist_title
                if playlist_title and playlist_title.upper() not in {"NA", "NONE"}
                else video_title
            )
            return title or url, yt_id.strip() or None
        except (IndexError, OSError, subprocess.SubprocessError, ValueError) as exc:
            LOG.warning("could not determine media title for %s: %s", url, exc)
            return url, None

    def _download(
        self,
        url: str,
        slot: int,
        generation: int,
        job_id: str,
    ) -> None:
        if not self.store.mark_download_running(job_id, slot, generation):
            return
        self._emit(
            DownloadEvent(
                slot=slot,
                state="downloading",
                job_id=job_id,
                generation=generation,
            )
        )
        slot_directory = canonical_slot_directory(self.config.storage.media_base, slot)
        temporary_playlist = slot_directory / f".playlist.{job_id}.tmp"
        try:
            slot_directory.mkdir(parents=True, exist_ok=True)
            if slot_directory.is_symlink():
                raise RuntimeError(f"refusing to download through symlink: {slot_directory}")
            title, yt_id = self._get_title(url)
            with self._yt_dlp_arguments() as base_arguments:
                arguments = base_arguments + [
                    "--yes-playlist",
                    "--no-overwrites",
                    "--no-progress",
                    "-o",
                    str(
                        slot_directory
                        / "%(playlist_index)03d - %(title)s [%(id)s].%(ext)s"
                    ),
                    "-f",
                    player_format_selector(self.config.downloads.max_height),
                    url,
                ]
                self._run_command(
                    arguments,
                    check=True,
                    text=True,
                    timeout=self.config.downloads.process_timeout_seconds,
                )
            media_files = sorted(
                path
                for path in slot_directory.iterdir()
                if path.is_file() and path.suffix.lower() in MEDIA_EXTENSIONS
            )
            if not media_files:
                raise RuntimeError("yt-dlp completed without producing media files")

            items = [
                (title_from_filename(path), os.path.relpath(path, slot_directory))
                for path in media_files
            ]
            with temporary_playlist.open("w", encoding="utf-8", newline="\n") as stream:
                for _, relative_path in items:
                    stream.write(relative_path + "\n")
                stream.flush()
                os.fsync(stream.fileno())
            os.replace(
                temporary_playlist,
                canonical_playlist_path(self.config.storage.media_base, slot),
            )
            self._fsync_directory(slot_directory)

            if not self.store.complete_download(
                job_id,
                slot,
                generation,
                title=title,
                items=items,
                yt_id=yt_id,
            ):
                LOG.warning(
                    "ignored stale download completion job=%s slot=%s generation=%s",
                    job_id,
                    slot,
                    generation,
                )
                return
            self._emit(
                DownloadEvent(
                    slot=slot,
                    state="ready",
                    title=title,
                    count=len(items),
                    job_id=job_id,
                    generation=generation,
                )
            )
        except BaseException as exc:
            temporary_playlist.unlink(missing_ok=True)
            message = self._failure_message(exc)
            self.store.fail_download(job_id, slot, generation, message)
            self._emit(
                DownloadEvent(
                    slot=slot,
                    state="error",
                    message=message,
                    job_id=job_id,
                    generation=generation,
                )
            )

    def _export(
        self,
        url: str,
        mode: str,
        job_id: str,
    ) -> None:
        if not self.store.mark_export_running(job_id):
            return
        work_directory: Path | None = None
        published = False
        completed = False
        try:
            work_directory, source_directory, ready_directory, final_parent = (
                self._prepare_export_directories(job_id, mode)
            )
            with self._yt_dlp_arguments() as base_arguments:
                arguments = base_arguments + self._export_download_arguments(
                    mode,
                    source_directory,
                    url,
                )
                self._run_command(
                    arguments,
                    check=True,
                    text=True,
                    timeout=self.config.downloads.process_timeout_seconds,
                )

            sources = self._export_source_files(source_directory, mode)
            if not sources:
                raise RuntimeError("yt-dlp completed without producing export media")
            used_names: set[str] = set()
            ready_outputs: list[Path] = []
            for source in sources:
                extension = ".mp4" if mode == "mp4" else EXPORT_AUDIO_EXTENSIONS[mode]
                output = ready_directory / safe_export_filename(
                    source.stem,
                    extension,
                    used_casefolded=used_names,
                )
                if mode == "mp4":
                    self._produce_compatible_mp4(source, output)
                else:
                    inspection = self._inspect_media(source)
                    self._validate_audio_export(inspection, mode)
                    os.replace(source, output)
                self._fsync_file(output)
                ready_outputs.append(output)

            self._fsync_directory(ready_directory)
            self._write_export_manifest(
                work_directory,
                job_id=job_id,
                mode=mode,
                filenames=[output.name for output in ready_outputs],
            )
            published_directory = final_parent / job_id
            if os.path.lexists(published_directory):
                raise FileExistsError(
                    f"refusing to overwrite existing export directory: {published_directory}"
                )
            os.rename(ready_directory, published_directory)
            self._fsync_directory(final_parent)
            published = True
            media_base = self.config.storage.media_base.resolve()
            output_paths = tuple(
                str((published_directory / output.name).resolve().relative_to(media_base))
                for output in ready_outputs
            )
            if not self.store.complete_export_job(job_id, output_paths):
                raise StaleOperationError(
                    "published export could not transition its durable job"
                )
            completed = True
        except BaseException as exc:
            message = self._failure_message(exc)
            try:
                self.store.fail_export_job(job_id, message)
            except BaseException:
                LOG.exception("could not persist utility export failure for %s", job_id)
            if published:
                LOG.error(
                    "utility export %s was published but needs restart reconciliation: %s",
                    job_id,
                    message,
                )
            else:
                LOG.warning("utility export %s failed: %s", job_id, message)
        finally:
            if work_directory is not None and (not published or completed):
                self._remove_export_work_directory(work_directory)

    def _export_download_arguments(
        self,
        mode: str,
        source_directory: Path,
        url: str,
    ) -> list[str]:
        arguments = [
            "--yes-playlist",
            "--no-overwrites",
            "--no-progress",
            "--windows-filenames",
            "--trim-filenames",
            str(MAX_EXPORT_STEM_BYTES),
            "-o",
            str(
                source_directory
                / "%(playlist_index)03d - %(title)s [%(id)s].%(ext)s"
            ),
        ]
        if mode == "mp4":
            arguments.extend(
                (
                    "-f",
                    mp4_export_format_selector(
                        self.config.downloads.export_max_height
                    ),
                    "-S",
                    (
                        f"res:{self.config.downloads.export_max_height},"
                        "vcodec:h264,acodec:aac"
                    ),
                )
            )
        elif mode == "m4a":
            arguments.extend(
                (
                    "-f",
                    "ba/b",
                    "-S",
                    "abr,acodec:aac",
                    "-x",
                    "--audio-format",
                    "m4a",
                    "--audio-quality",
                    "0",
                )
            )
        elif mode == "mp3":
            arguments.extend(
                (
                    "-f",
                    "ba/b",
                    "-x",
                    "--audio-format",
                    "mp3",
                    "--audio-quality",
                    "0",
                )
            )
        else:
            raise ValueError("unsupported utility export mode")
        arguments.append(url)
        return arguments

    def _prepare_export_directories(
        self,
        job_id: str,
        mode: str,
    ) -> tuple[Path, Path, Path, Path]:
        if not re.fullmatch(r"[0-9a-f]{32}", job_id):
            raise ValueError("invalid export job identity")
        media_base = self.config.storage.media_base
        media_base.mkdir(parents=True, exist_ok=True)
        export_root = media_base
        for part in self.config.downloads.export_directory.parts:
            export_root = export_root / part
            self._ensure_real_directory(export_root)
        work_root = export_root / ".work"
        final_parent = export_root / ("video" if mode == "mp4" else "audio")
        self._ensure_real_directory(work_root)
        self._ensure_real_directory(final_parent)
        work_directory = work_root / job_id
        work_directory.mkdir(mode=0o750, exist_ok=False)
        source_directory = work_directory / "source"
        ready_directory = work_directory / "ready"
        source_directory.mkdir(mode=0o750)
        ready_directory.mkdir(mode=0o750)
        self._fsync_directory(work_root)
        return work_directory, source_directory, ready_directory, final_parent

    @staticmethod
    def _ensure_real_directory(path: Path) -> None:
        try:
            path.mkdir(mode=0o750)
        except FileExistsError:
            pass
        metadata = path.lstat()
        if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISDIR(metadata.st_mode):
            raise RuntimeError(f"refusing unsafe export directory: {path}")

    def _write_export_manifest(
        self,
        work_directory: Path,
        *,
        job_id: str,
        mode: str,
        filenames: list[str],
    ) -> None:
        if not filenames:
            raise ValueError("export manifest must contain output filenames")
        payload = json.dumps(
            {
                "version": 1,
                "job_id": job_id,
                "mode": mode,
                "filenames": filenames,
            },
            ensure_ascii=False,
            sort_keys=True,
        )
        temporary = work_directory / f".{EXPORT_MANIFEST_NAME}.tmp"
        manifest = work_directory / EXPORT_MANIFEST_NAME
        try:
            with temporary.open("x", encoding="utf-8", newline="\n") as stream:
                stream.write(payload + "\n")
                stream.flush()
                os.fsync(stream.fileno())
            os.replace(temporary, manifest)
            self._fsync_directory(work_directory)
        finally:
            temporary.unlink(missing_ok=True)

    @staticmethod
    def _export_source_files(source_directory: Path, mode: str) -> list[Path]:
        allowed = (
            VIDEO_EXPORT_INPUT_EXTENSIONS
            if mode == "mp4"
            else frozenset({EXPORT_AUDIO_EXTENSIONS[mode]})
        )
        media: list[Path] = []
        for candidate in source_directory.iterdir():
            if candidate.is_symlink() or not candidate.is_file():
                raise RuntimeError(
                    f"yt-dlp produced an unsafe export entry: {candidate.name}"
                )
            if candidate.suffix.lower() not in allowed:
                raise RuntimeError(
                    f"yt-dlp produced an unexpected export file: {candidate.name}"
                )
            media.append(candidate)
        return sorted(media, key=lambda path: path.name.casefold())

    def _inspect_media(self, path: Path) -> MediaInspection:
        result = self._run_command(
            [
                "ffprobe",
                "-v",
                "error",
                "-show_streams",
                "-show_format",
                "-of",
                "json",
                str(path),
            ],
            check=True,
            capture_output=True,
            text=True,
            timeout=min(120.0, self.config.downloads.process_timeout_seconds),
        )
        try:
            payload = json.loads(result.stdout or "")
            streams = payload["streams"]
            format_name = payload["format"]["format_name"]
        except (KeyError, TypeError, ValueError, json.JSONDecodeError) as exc:
            raise RuntimeError("ffprobe returned invalid media metadata") from exc
        if not isinstance(streams, list) or not isinstance(format_name, str):
            raise RuntimeError("ffprobe returned invalid media metadata")
        video = next(
            (
                stream
                for stream in streams
                if isinstance(stream, dict) and stream.get("codec_type") == "video"
            ),
            None,
        )
        audio = next(
            (
                stream
                for stream in streams
                if isinstance(stream, dict) and stream.get("codec_type") == "audio"
            ),
            None,
        )

        def positive_integer(value: Any) -> int | None:
            return (
                value
                if isinstance(value, int)
                and not isinstance(value, bool)
                and value > 0
                else None
            )

        return MediaInspection(
            format_names=frozenset(
                value.strip().lower()
                for value in format_name.split(",")
                if value.strip()
            ),
            video_codec=(
                video.get("codec_name") if isinstance(video, dict) else None
            ),
            audio_codec=(
                audio.get("codec_name") if isinstance(audio, dict) else None
            ),
            pixel_format=(
                video.get("pix_fmt") if isinstance(video, dict) else None
            ),
            width=positive_integer(video.get("width") if isinstance(video, dict) else None),
            height=positive_integer(video.get("height") if isinstance(video, dict) else None),
        )

    def _produce_compatible_mp4(self, source: Path, output: Path) -> None:
        source_info = self._inspect_media(source)
        if source_info.video_codec is None or source_info.audio_codec is None:
            raise RuntimeError("MP4 export requires both video and audio")
        if source_info.height is None:
            raise RuntimeError("MP4 export source has no verifiable video height")
        if source_info.height > self.config.downloads.export_max_height:
            raise RuntimeError("MP4 export source exceeds configured maximum height")
        video_compatible = (
            source_info.video_codec == "h264"
            and source_info.pixel_format in {"yuv420p", "yuvj420p"}
        )
        audio_compatible = source_info.audio_codec == "aac"
        # ffprobe reports the same comma-separated demuxer aliases for MOV and
        # MP4. A .mov source therefore needs at least a stream-copy remux; it
        # must never be renamed to .mp4 solely because the alias list includes
        # "mp4".
        container_compatible = (
            source.suffix.lower() == ".mp4"
            and "mp4" in source_info.format_names
        )
        if video_compatible and audio_compatible and container_compatible:
            os.replace(source, output)
        else:
            temporary = output.with_name(
                f".{output.stem}.{uuid.uuid4().hex}.tmp.mp4"
            )
            arguments = [
                "ffmpeg",
                "-nostdin",
                "-hide_banner",
                "-loglevel",
                "error",
                "-n",
                "-i",
                str(source),
                "-map",
                "0:v:0",
                "-map",
                "0:a:0",
                "-map_metadata",
                "0",
            ]
            if video_compatible:
                arguments.extend(("-c:v", "copy"))
            else:
                arguments.extend(
                    (
                        "-c:v",
                        "libx264",
                        "-preset",
                        "medium",
                        "-crf",
                        "18",
                        "-pix_fmt",
                        "yuv420p",
                    )
                )
            if audio_compatible:
                arguments.extend(("-c:a", "copy"))
            else:
                arguments.extend(("-c:a", "aac", "-b:a", "192k"))
            arguments.extend(("-movflags", "+faststart", str(temporary)))
            try:
                self._run_command(
                    arguments,
                    check=True,
                    text=True,
                    timeout=self.config.downloads.process_timeout_seconds,
                )
                converted = self._inspect_media(temporary)
                self._validate_compatible_mp4(converted, source_info.height)
                os.replace(temporary, output)
            finally:
                temporary.unlink(missing_ok=True)
        self._validate_compatible_mp4(
            self._inspect_media(output),
            source_info.height,
        )

    def _validate_compatible_mp4(
        self,
        inspection: MediaInspection,
        source_height: int,
    ) -> None:
        if (
            "mp4" not in inspection.format_names
            or inspection.video_codec != "h264"
            or inspection.audio_codec != "aac"
            or inspection.pixel_format not in {"yuv420p", "yuvj420p"}
        ):
            raise RuntimeError("MP4 compatibility verification failed")
        if (
            inspection.height != source_height
            or inspection.height > self.config.downloads.export_max_height
        ):
            raise RuntimeError("MP4 export resolution verification failed")

    @staticmethod
    def _validate_audio_export(
        inspection: MediaInspection,
        mode: str,
    ) -> None:
        expected_codec = "aac" if mode == "m4a" else "mp3"
        expected_container = "mp4" if mode == "m4a" else "mp3"
        if inspection.video_codec is not None:
            raise RuntimeError("audio export unexpectedly contains video")
        if inspection.audio_codec != expected_codec:
            raise RuntimeError(f"{mode.upper()} export codec verification failed")
        if expected_container not in inspection.format_names:
            raise RuntimeError(f"{mode.upper()} export container verification failed")

    @staticmethod
    def _fsync_file(path: Path) -> None:
        descriptor = os.open(path, os.O_RDONLY)
        try:
            os.fsync(descriptor)
        finally:
            os.close(descriptor)

    def _remove_export_work_directory(self, path: Path) -> None:
        try:
            parent = path.parent
            if os.path.lexists(path):
                if path.is_symlink() or path.is_file():
                    path.unlink()
                else:
                    shutil.rmtree(path)
                self._fsync_directory(parent)
        except OSError:
            LOG.exception("could not remove export work directory %s", path)

    def reconcile_export_artifacts(self) -> None:
        """Recover verified crash-published exports and discard incomplete staging."""

        export_root = self.config.storage.media_base / self.config.downloads.export_directory
        if not os.path.lexists(export_root):
            return
        try:
            self._verify_existing_export_tree(export_root)
        except OSError as exc:
            LOG.error("could not inspect utility export tree: %s", exc)
            return
        except RuntimeError as exc:
            LOG.error("utility export reconciliation refused unsafe storage: %s", exc)
            return

        jobs = {
            job.job_id: job for job in self.store.list_reconcilable_export_jobs()
        }
        for job in jobs.values():
            try:
                self._reconcile_export_job(export_root, job.job_id, job.mode)
            except BaseException:
                LOG.exception("could not reconcile utility export %s", job.job_id)

        work_root = export_root / ".work"
        if not work_root.is_dir() or work_root.is_symlink():
            return
        for candidate in work_root.iterdir():
            if not re.fullmatch(r"[0-9a-f]{32}", candidate.name):
                LOG.warning("leaving unknown utility export staging entry %s", candidate)
                continue
            job = self.store.get_export_job(candidate.name)
            if job is None:
                LOG.warning("leaving untracked utility export staging entry %s", candidate)
                continue
            if job.status == "completed":
                self._remove_export_work_directory(candidate)
            elif job.job_id not in jobs:
                LOG.warning(
                    "leaving utility export staging for active job %s",
                    candidate.name,
                )

    def _verify_existing_export_tree(self, export_root: Path) -> None:
        current = self.config.storage.media_base
        for part in self.config.downloads.export_directory.parts:
            current = current / part
            metadata = current.lstat()
            if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISDIR(metadata.st_mode):
                raise RuntimeError(f"unsafe export directory: {current}")
        for child_name in (".work", "video", "audio"):
            child = export_root / child_name
            if not os.path.lexists(child):
                continue
            metadata = child.lstat()
            if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISDIR(metadata.st_mode):
                raise RuntimeError(f"unsafe export directory: {child}")

    def _reconcile_export_job(
        self,
        export_root: Path,
        job_id: str,
        mode: str,
    ) -> None:
        work_directory = export_root / ".work" / job_id
        final_parent = export_root / ("video" if mode == "mp4" else "audio")
        final_directory = final_parent / job_id
        if os.path.lexists(work_directory):
            work_metadata = work_directory.lstat()
            if stat.S_ISLNK(work_metadata.st_mode) or not stat.S_ISDIR(
                work_metadata.st_mode
            ):
                raise RuntimeError("utility export staging directory is unsafe")
        manifest = self._read_export_manifest(work_directory, job_id, mode)

        if os.path.lexists(final_directory):
            filenames = None if manifest is None else manifest
            outputs = self._validate_export_output_directory(
                final_directory,
                mode,
                expected_filenames=filenames,
            )
        elif manifest is not None:
            ready_directory = work_directory / "ready"
            outputs = self._validate_export_output_directory(
                ready_directory,
                mode,
                expected_filenames=manifest,
            )
            self._ensure_real_directory(final_parent)
            os.rename(ready_directory, final_directory)
            self._fsync_directory(final_parent)
        else:
            if os.path.lexists(work_directory):
                self._remove_export_work_directory(work_directory)
            return

        media_base = self.config.storage.media_base.resolve()
        output_paths = tuple(
            str((final_directory / output.name).resolve().relative_to(media_base))
            for output in outputs
        )
        if self.store.recover_export_job(job_id, output_paths):
            LOG.warning("recovered crash-published utility export %s", job_id)
            if os.path.lexists(work_directory):
                self._remove_export_work_directory(work_directory)

    @staticmethod
    def _read_export_manifest(
        work_directory: Path,
        job_id: str,
        mode: str,
    ) -> list[str] | None:
        manifest = work_directory / EXPORT_MANIFEST_NAME
        if not manifest.is_file() or manifest.is_symlink():
            return None
        if manifest.stat().st_size > 8 * 1024 * 1024:
            raise RuntimeError("utility export manifest is too large")
        try:
            payload = json.loads(manifest.read_text(encoding="utf-8"))
        except (OSError, UnicodeError, json.JSONDecodeError) as exc:
            raise RuntimeError("utility export manifest is invalid") from exc
        if (
            not isinstance(payload, dict)
            or payload.get("version") != 1
            or payload.get("job_id") != job_id
            or payload.get("mode") != mode
        ):
            raise RuntimeError("utility export manifest identity is invalid")
        filenames = payload.get("filenames")
        if not isinstance(filenames, list) or not filenames or len(filenames) > 10000:
            raise RuntimeError("utility export manifest filenames are invalid")
        expected_extension = ".mp4" if mode == "mp4" else EXPORT_AUDIO_EXTENSIONS[mode]
        normalized: list[str] = []
        for filename in filenames:
            if (
                not isinstance(filename, str)
                or not filename
                or Path(filename).name != filename
                or Path(filename).suffix.lower() != expected_extension
            ):
                raise RuntimeError("utility export manifest contains an unsafe filename")
            normalized.append(filename)
        if len({value.casefold() for value in normalized}) != len(normalized):
            raise RuntimeError("utility export manifest contains duplicate filenames")
        return normalized

    def _validate_export_output_directory(
        self,
        directory: Path,
        mode: str,
        *,
        expected_filenames: list[str] | None,
    ) -> list[Path]:
        metadata = directory.lstat()
        if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISDIR(metadata.st_mode):
            raise RuntimeError("utility export output directory is unsafe")
        outputs = sorted(directory.iterdir(), key=lambda path: path.name.casefold())
        if not outputs:
            raise RuntimeError("utility export output directory is empty")
        if expected_filenames is not None and {
            path.name for path in outputs
        } != set(expected_filenames):
            raise RuntimeError("utility export output does not match its manifest")
        expected_extension = ".mp4" if mode == "mp4" else EXPORT_AUDIO_EXTENSIONS[mode]
        for output in outputs:
            if (
                output.is_symlink()
                or not output.is_file()
                or output.suffix.lower() != expected_extension
            ):
                raise RuntimeError("utility export output contains an unsafe entry")
            inspection = self._inspect_media(output)
            if mode == "mp4":
                if inspection.height is None:
                    raise RuntimeError("MP4 export output has no verifiable height")
                self._validate_compatible_mp4(inspection, inspection.height)
            else:
                self._validate_audio_export(inspection, mode)
        return outputs

    @staticmethod
    def _failure_message(exc: BaseException) -> str:
        command = getattr(exc, "cmd", None)
        program = "media command"
        if isinstance(command, (list, tuple)) and command:
            program = Path(str(command[0])).name
        if isinstance(exc, subprocess.TimeoutExpired):
            return f"{program} timed out after {exc.timeout} seconds"
        if isinstance(exc, subprocess.CalledProcessError):
            return f"{program} failed with exit code {exc.returncode}"
        return f"download failed: {exc}"

    def _run_command(
        self,
        arguments: list[str],
        *,
        check: bool,
        text: bool,
        timeout: float,
        capture_output: bool = False,
    ) -> subprocess.CompletedProcess[str]:
        if self.runner is not None:
            return self.runner(
                arguments,
                check=check,
                text=text,
                timeout=timeout,
                capture_output=capture_output,
            )

        with self._lock:
            if self._stopping:
                raise RuntimeError("Controller is shutting down")
            process = subprocess.Popen(
                arguments,
                stdout=subprocess.PIPE if capture_output else None,
                stderr=subprocess.PIPE if capture_output else None,
                text=text,
                start_new_session=True,
            )
            self._processes.add(process)
        try:
            try:
                stdout, stderr = process.communicate(timeout=timeout)
            except subprocess.TimeoutExpired:
                self._terminate_process(process, force=False)
                try:
                    stdout, stderr = process.communicate(timeout=2.0)
                except subprocess.TimeoutExpired:
                    self._terminate_process(process, force=True)
                    stdout, stderr = process.communicate()
                raise subprocess.TimeoutExpired(
                    arguments,
                    timeout,
                    output=stdout,
                    stderr=stderr,
                )
            completed = subprocess.CompletedProcess(
                arguments,
                process.returncode,
                stdout=stdout,
                stderr=stderr,
            )
            if check and process.returncode:
                raise subprocess.CalledProcessError(
                    process.returncode,
                    arguments,
                    output=stdout,
                    stderr=stderr,
                )
            return completed
        finally:
            with self._lock:
                self._processes.discard(process)

    @staticmethod
    def _terminate_process(
        process: subprocess.Popen[str],
        *,
        force: bool,
    ) -> None:
        if process.poll() is not None:
            return
        try:
            os.killpg(
                process.pid,
                signal.SIGKILL if force else signal.SIGTERM,
            )
        except ProcessLookupError:
            return
        except OSError:
            try:
                process.kill() if force else process.terminate()
            except ProcessLookupError:
                pass

    @staticmethod
    def _fsync_directory(path: Path) -> None:
        descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
        try:
            os.fsync(descriptor)
        finally:
            os.close(descriptor)

    def delete_slot(
        self,
        slot: int,
        *,
        reason: str,
        expected_url: str | None = None,
    ) -> bool:
        token = self.store.begin_delete(slot, reason=reason, expected_url=expected_url)
        return self.finish_deletion(token)

    def finish_deletion(self, token: DeleteToken) -> bool:
        slot_directory = canonical_slot_directory(self.config.storage.media_base, token.slot)
        trash_root = self.config.storage.media_base / ".controller-trash"
        trash_path: Path | None = None
        if self._slot_is_guarded(token.slot):
            return False
        if os.path.lexists(slot_directory):
            if slot_directory.is_symlink():
                slot_directory.unlink()
            else:
                trash_root.mkdir(parents=True, exist_ok=True)
                self._fsync_directory(self.config.storage.media_base)
                trash_path = trash_root / (
                    f"slot{token.slot}.generation{token.generation}.{uuid.uuid4().hex}"
                )
                os.replace(slot_directory, trash_path)
                self._fsync_directory(self.config.storage.media_base)
                self._fsync_directory(trash_root)
        if self._slot_is_guarded(token.slot):
            if (
                trash_path is not None
                and not os.path.lexists(slot_directory)
                and os.path.lexists(trash_path)
            ):
                os.replace(trash_path, slot_directory)
                self._fsync_directory(self.config.storage.media_base)
                self._fsync_directory(trash_root)
            return False
        if trash_path is not None:
            shutil.rmtree(trash_path)
            self._fsync_directory(trash_root)
        return self.store.finish_delete(token)

    def _slot_is_guarded(self, slot: int) -> bool:
        return any(
            guard.slot == slot for guard in self.store.list_live_guards()
        )

    def expire_due_slots(self) -> list[int]:
        """Synchronously expire slots for maintenance callers and tests."""

        deleted: list[int] = []
        for token in self.store.begin_expired_deletions():
            try:
                if self.finish_deletion(token):
                    deleted.append(token.slot)
            except OSError:
                LOG.exception("could not finish expired deletion for slot %s", token.slot)
        return deleted

    def schedule_delete_slot(
        self,
        slot: int,
        *,
        reason: str,
        expected_url: str | None = None,
    ) -> DeleteToken:
        """Atomically fence a manual deletion, then finish it off-thread."""

        token = self.store.begin_delete(
            slot,
            reason=reason,
            expected_url=expected_url,
        )
        self._schedule_deletion(token)
        return token

    def schedule_expired_deletions(self) -> list[DeleteToken]:
        """Fence all currently due slots and queue filesystem work."""

        tokens = self.store.begin_expired_deletions()
        for token in tokens:
            self._schedule_deletion(token)
        return tokens

    def recover_pending_deletions(self) -> list[int]:
        recovered: list[int] = []
        for token in self.store.pending_deletions():
            if self._recover_deletion(token):
                recovered.append(token.slot)
        return recovered

    def schedule_pending_deletion_recovery(self) -> bool:
        """Queue every safe pending deletion without blocking the coordinator."""

        with self._lock:
            if self._stopping:
                return False
        scheduled = False
        for token in self.store.pending_deletions():
            scheduled = self._schedule_deletion(token) or scheduled
        return scheduled

    def _schedule_deletion(self, token: DeleteToken) -> bool:
        key = (token.slot, token.generation)
        with self._lock:
            if self._stopping or key in self._deletions_in_flight:
                return False
            self._deletions_in_flight.add(key)
        try:
            future = self._executor.submit(self._deletion_worker, token)
        except BaseException as exc:
            with self._lock:
                self._deletions_in_flight.discard(key)
            self._emit(
                DownloadEvent(
                    slot=token.slot,
                    state="delete_error",
                    message=f"could not start deletion worker: {exc}",
                    generation=token.generation,
                )
            )
            return False
        with self._lock:
            self._futures.add(future)
        future.add_done_callback(self._forget_future)
        return True

    def _deletion_worker(self, token: DeleteToken) -> None:
        key = (token.slot, token.generation)
        try:
            if self._recover_deletion(token):
                self._emit(
                    DownloadEvent(
                        slot=token.slot,
                        state="deleted",
                        generation=token.generation,
                    )
                )
            elif self.store.get_slot(token.slot).media_state == "deleting":
                self._emit(
                    DownloadEvent(
                        slot=token.slot,
                        state="delete_deferred",
                        message="slot deletion was deferred by playback protection",
                        generation=token.generation,
                    )
                )
        except BaseException as exc:
            LOG.exception(
                "could not finish deletion for slot %s generation %s",
                token.slot,
                token.generation,
            )
            self._emit(
                DownloadEvent(
                    slot=token.slot,
                    state="delete_error",
                    message=f"deletion remains pending: {exc}",
                    generation=token.generation,
                )
            )
        finally:
            with self._lock:
                self._deletions_in_flight.discard(key)

    def _recover_deletion(self, token: DeleteToken) -> bool:
        trash_root = self.config.storage.media_base / ".controller-trash"
        slot_directory = canonical_slot_directory(
            self.config.storage.media_base,
            token.slot,
        )
        if self._slot_is_guarded(token.slot):
            LOG.warning(
                "deferring interrupted deletion for protected slot %s",
                token.slot,
            )
            self._restore_guarded_trash(token, trash_root, slot_directory)
            return False
        if trash_root.is_dir():
            prefix = f"slot{token.slot}.generation{token.generation}."
            for candidate in trash_root.iterdir():
                if not candidate.name.startswith(prefix):
                    continue
                if self._slot_is_guarded(token.slot):
                    self._restore_guarded_trash(
                        token,
                        trash_root,
                        slot_directory,
                    )
                    return False
                if candidate.is_symlink() or candidate.is_file():
                    candidate.unlink()
                elif candidate.is_dir():
                    shutil.rmtree(candidate)
                self._fsync_directory(trash_root)
        return self.finish_deletion(token)

    def _restore_guarded_trash(
        self,
        token: DeleteToken,
        trash_root: Path,
        slot_directory: Path,
    ) -> None:
        if not trash_root.is_dir() or os.path.lexists(slot_directory):
            return
        prefix = f"slot{token.slot}.generation{token.generation}."
        candidates = sorted(
            candidate
            for candidate in trash_root.iterdir()
            if candidate.name.startswith(prefix)
        )
        if not candidates:
            return
        os.replace(candidates[0], slot_directory)
        self._fsync_directory(self.config.storage.media_base)
        self._fsync_directory(trash_root)

    def shutdown(self, *, wait: bool = True) -> None:
        with self._lock:
            self._stopping = True
            processes = tuple(self._processes)
        for process in processes:
            self._terminate_process(process, force=False)
        deadline = time.monotonic() + 2.0
        for process in processes:
            if process.poll() is not None:
                continue
            try:
                process.wait(timeout=max(0.0, deadline - time.monotonic()))
            except subprocess.TimeoutExpired:
                self._terminate_process(process, force=True)
        if self._owns_executor:
            self._executor.shutdown(wait=wait, cancel_futures=True)
        if self._owns_export_executor:
            self._export_executor.shutdown(wait=wait, cancel_futures=True)

    def active_future_count(self) -> int:
        with self._lock:
            return len(self._futures)