pretranscode/pretranscode.py

#!/usr/bin/env python3
"""Immutable Jellyfin H.264 cache backed by an atomic symlink view."""

import hashlib
import json
import os
import signal
import subprocess
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path, PurePosixPath

VIDEO_EXTENSIONS = {".mkv", ".mp4", ".m4v", ".avi", ".mov", ".ts", ".m2ts", ".webm"}
BLACKLISTED_CODECS = {"hevc", "h265", "mpeg2video", "vc1", "vc-1"}


def env_int(name, default):
    return int(os.environ.get(name, default))


SOURCE_ROOT = Path(os.environ.get("SOURCE_ROOT", "/source"))
VIEW_ROOT = Path(os.environ.get("VIEW_ROOT", "/view"))
CACHE_ROOT = Path(os.environ.get("CACHE_ROOT", "/cache"))
STATE_ROOT = Path(os.environ.get("STATE_ROOT", "/state"))
STATE_FILE = STATE_ROOT / "state.json"
JELLYFIN_URL = os.environ.get("JELLYFIN_URL", "http://jellyfin:8096").rstrip("/")
JELLYFIN_LIBRARY_ROOT = os.environ.get("JELLYFIN_LIBRARY_ROOT", "/data/library").rstrip("/")
JELLYFIN_ORIGINAL_ROOT = os.environ.get("JELLYFIN_ORIGINAL_ROOT", "/data/original").rstrip("/")
JELLYFIN_CACHE_ROOT = os.environ.get("JELLYFIN_CACHE_ROOT", "/data/cache").rstrip("/")
JELLYFIN_API_KEY = os.environ.get("JELLYFIN_API_KEY", "")
CACHE_MAX_BYTES = env_int("CACHE_MAX_BYTES", 1_000_000_000_000)
SERIES_COOLDOWN_SECONDS = env_int("SERIES_COOLDOWN_SECONDS", 3600)
POLL_SECONDS = env_int("POLL_SECONDS", 5)
RESCAN_SECONDS = env_int("RESCAN_SECONDS", 300)
H264_CRF = os.environ.get("H264_CRF", "20")
H264_PRESET = os.environ.get("H264_PRESET", "medium")
H264_MAXRATE = os.environ.get("H264_MAXRATE", "12M")
H264_BUFSIZE = os.environ.get("H264_BUFSIZE", "24M")
FFMPEG = os.environ.get("FFMPEG", "/usr/bin/ffmpeg")
FFPROBE = os.environ.get("FFPROBE", "/usr/bin/ffprobe")


def log(message):
    print(time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), message, flush=True)


def atomic_json(path, value):
    path.parent.mkdir(parents=True, exist_ok=True)
    temporary = path.with_name(path.name + ".tmp")
    with temporary.open("w") as handle:
        json.dump(value, handle, sort_keys=True, separators=(",", ":"))
        handle.flush()
        os.fsync(handle.fileno())
    os.replace(temporary, path)


def load_json(path, default):
    try:
        with path.open() as handle:
            return json.load(handle)
    except (FileNotFoundError, json.JSONDecodeError, OSError):
        return default


def fingerprint(path):
    stat = path.stat()
    return f"{stat.st_size}:{stat.st_mtime_ns}"


def relative_source(path):
    value = str(path)
    for root in (JELLYFIN_LIBRARY_ROOT, str(SOURCE_ROOT)):
        prefix = root.rstrip("/") + "/"
        if value.startswith(prefix):
            return value[len(prefix):]
    return None


def item_key(relative):
    parts = PurePosixPath(relative).parts
    if len(parts) >= 2 and parts[0].lower() == "tv":
        return "series:" + parts[1]
    return "item:" + relative


def cache_basename(relative):
    digest = hashlib.sha256(relative.encode()).hexdigest()
    suffix = ".mp4" if Path(relative).suffix.lower() in {".mp4", ".m4v"} else ".mkv"
    return digest + suffix


def source_target(relative):
    return JELLYFIN_ORIGINAL_ROOT + "/" + PurePosixPath(relative).as_posix()


def cache_target(relative):
    return JELLYFIN_CACHE_ROOT + "/files/" + cache_basename(relative)


class Manager:
    def __init__(self):
        STATE_ROOT.mkdir(parents=True, exist_ok=True)
        if not (CACHE_ROOT / ".pretranscode-cache").is_file():
            raise RuntimeError(f"cache filesystem is not mounted at {CACHE_ROOT}")
        files_root = CACHE_ROOT / "files"
        files_root.mkdir(parents=True, exist_ok=True)
        for partial in files_root.glob("*.partial"):
            partial.unlink(missing_ok=True)
        for output in list(files_root.glob("*.mkv")) + list(files_root.glob("*.mp4")):
            if not output.with_suffix(output.suffix + ".json").is_file():
                output.unlink(missing_ok=True)
        for manifest in files_root.glob("*.json"):
            if not manifest.with_suffix("").is_file():
                manifest.unlink(missing_ok=True)
        VIEW_ROOT.mkdir(parents=True, exist_ok=True)
        self.state = load_json(STATE_FILE, {})
        self.state.setdefault("inventory", {})
        self.state.setdefault("priorities", {})
        self.state.setdefault("cooldowns", {})
        self.state.setdefault("retry_after", {})
        self.active_items = set()
        self.active_keys = set()
        self.previous_active_items = set()
        self.previous_active_keys = set()
        self.sessions_healthy = False
        self.running = True
        self.process = None
        self.last_scan = 0
        self.last_activity_save = 0
        signal.signal(signal.SIGTERM, self.stop)
        signal.signal(signal.SIGINT, self.stop)

    def stop(self, *_):
        self.running = False
        self.cancel_process("service stopping")

    def save(self):
        atomic_json(STATE_FILE, self.state)

    def api(self, path, method="GET"):
        request = urllib.request.Request(
            JELLYFIN_URL + path,
            method=method,
            headers={"X-Emby-Token": JELLYFIN_API_KEY, "Content-Length": "0"},
        )
        with urllib.request.urlopen(request, timeout=10) as response:
            data = response.read()
            return json.loads(data) if data else None

    def read_sessions(self):
        try:
            sessions = self.api("/Sessions")
        except Exception as error:
            if self.sessions_healthy:
                log(f"Jellyfin sessions unavailable; stopping cache work: {error}")
            self.sessions_healthy = False
            self.active_items = set()
            self.active_keys = set()
            return False

        now = time.time()
        current_items = set()
        current_keys = set()
        for session in sessions or []:
            item = session.get("NowPlayingItem") or {}
            path = item.get("Path")
            relative = relative_source(path) if path else None
            if not relative and item.get("Id"):
                try:
                    detail = self.api("/Items/" + item["Id"] + "?Fields=Path") or {}
                    relative = relative_source(detail.get("Path", ""))
                except Exception:
                    relative = None
            if not relative:
                continue
            current_items.add(relative)
            current_keys.add(item_key(relative))
            self.touch_cache(relative)

        ended_items = self.previous_active_items - current_items
        ended_keys = self.previous_active_keys - current_keys
        changed = current_items != self.previous_active_items or current_keys != self.previous_active_keys
        for relative in current_items | ended_items:
            self.state["priorities"][relative] = now
        for key in current_keys | ended_keys:
            if key.startswith("series:"):
                self.state["cooldowns"][key] = now + SERIES_COOLDOWN_SECONDS
        if changed or (current_items and now - self.last_activity_save >= 60):
            self.save()
            self.last_activity_save = now

        self.previous_active_items = current_items
        self.previous_active_keys = current_keys
        self.active_items = current_items
        self.active_keys = current_keys
        self.sessions_healthy = True
        return True

    def locked_keys(self):
        now = time.time()
        expired = [key for key, until in self.state["cooldowns"].items() if float(until) <= now]
        if expired:
            for key in expired:
                self.state["cooldowns"].pop(key, None)
            self.save()
        return self.active_keys | {
            key for key, until in self.state["cooldowns"].items() if float(until) > now
        }

    def is_locked(self, relative):
        return item_key(relative) in self.locked_keys()

    def cache_paths(self, relative):
        output = CACHE_ROOT / "files" / cache_basename(relative)
        manifest = output.with_suffix(output.suffix + ".json")
        return output, manifest

    def cache_valid(self, relative):
        source = SOURCE_ROOT / relative
        output, manifest_path = self.cache_paths(relative)
        if not source.is_file() or not output.is_file():
            return False
        manifest = load_json(manifest_path, {})
        try:
            return manifest.get("source_fingerprint") == fingerprint(source)
        except OSError:
            return False

    def touch_cache(self, relative):
        if not self.cache_valid(relative):
            return
        output, manifest = self.cache_paths(relative)
        now = time.time()
        for path in (output, manifest):
            try:
                os.utime(path, (now, now), follow_symlinks=False)
            except OSError:
                pass

    def ensure_link(self, relative, target, locked=False):
        link = VIEW_ROOT / relative
        link.parent.mkdir(parents=True, exist_ok=True)
        try:
            if link.is_symlink() and os.readlink(link) == target:
                return False
        except OSError:
            pass
        if locked and (link.exists() or link.is_symlink()):
            return False
        temporary = link.with_name(link.name + ".pretranscode-tmp")
        try:
            temporary.unlink(missing_ok=True)
            os.symlink(target, temporary)
            os.replace(temporary, link)
            return True
        finally:
            temporary.unlink(missing_ok=True)

    def swap_to_original(self, relative):
        return self.ensure_link(relative, source_target(relative), locked=self.is_locked(relative))

    def swap_to_cache(self, relative):
        if self.is_locked(relative) or not self.cache_valid(relative):
            return False
        output, manifest_path = self.cache_paths(relative)
        manifest = load_json(manifest_path, {})
        changed = self.ensure_link(relative, cache_target(relative), locked=False)
        if changed or manifest.get("refresh_version") != 2:
            self.refresh_item(relative)
            manifest["refresh_version"] = 2
            atomic_json(manifest_path, manifest)
        return changed

    def refresh_item(self, relative):
        item_id = (self.state["inventory"].get(relative) or {}).get("item_id")
        if not item_id:
            return
        query = urllib.parse.urlencode({
            "metadataRefreshMode": "FullRefresh",
            "imageRefreshMode": "None",
            "replaceAllMetadata": "false",
            "replaceAllImages": "false",
        })
        try:
            self.api(f"/Items/{item_id}/Refresh?{query}", method="POST")
        except Exception as error:
            log(f"Unable to refresh Jellyfin item {relative}: {error}")

    def probe(self, path):
        command = [
            FFPROBE, "-v", "error", "-select_streams", "v:0",
            "-show_entries", "stream=codec_name,bits_per_raw_sample,pix_fmt,profile",
            "-show_entries", "format=duration", "-of", "json", str(path),
        ]
        result = subprocess.run(command, capture_output=True, text=True, timeout=120)
        if result.returncode:
            raise RuntimeError(result.stderr.strip() or "ffprobe failed")
        data = json.loads(result.stdout)
        stream = (data.get("streams") or [{}])[0]
        raw_depth = stream.get("bits_per_raw_sample")
        try:
            depth = int(raw_depth or 0)
        except ValueError:
            depth = 0
        pixel = (stream.get("pix_fmt") or "").lower()
        profile = (stream.get("profile") or "").lower()
        if not depth:
            if "12" in pixel or "12" in profile:
                depth = 12
            elif "10" in pixel or "10" in profile:
                depth = 10
            else:
                depth = 8
        return {
            "codec": (stream.get("codec_name") or "").lower(),
            "bit_depth": depth,
            "profile": profile,
            "duration": float((data.get("format") or {}).get("duration") or 0),
        }

    def jellyfin_items(self):
        values = {}
        start = 0
        limit = 1000
        try:
            while True:
                query = urllib.parse.urlencode({
                    "Recursive": "true",
                    "IncludeItemTypes": "Movie,Episode",
                    "Fields": "Path,MediaStreams",
                    "StartIndex": start,
                    "Limit": limit,
                })
                page = self.api("/Items?" + query) or {}
                items = page.get("Items") or []
                for item in items:
                    relative = relative_source(item.get("Path", ""))
                    if not relative:
                        continue
                    video = next(
                        (stream for stream in item.get("MediaStreams", [])
                         if (stream.get("Type") or "").lower() == "video"),
                        {},
                    )
                    values[relative] = {
                        "item_id": item.get("Id"),
                        "codec": (video.get("Codec") or "").lower(),
                        "bit_depth": int(video.get("BitDepth") or 8),
                        "profile": (video.get("Profile") or "").lower(),
                        "duration": float(item.get("RunTimeTicks") or 0) / 10_000_000,
                    }
                start += len(items)
                if not items or start >= int(page.get("TotalRecordCount") or start):
                    break
        except Exception as error:
            log(f"Unable to read Jellyfin inventory: {error}")
        return values

    def scan_inventory(self):
        catalog = self.jellyfin_items()
        inventory = self.state["inventory"]
        seen = set()
        added = False
        for root, directories, files in os.walk(SOURCE_ROOT):
            directories.sort()
            files.sort()
            for filename in files:
                if Path(filename).suffix.lower() not in VIDEO_EXTENSIONS:
                    continue
                source = Path(root) / filename
                relative = source.relative_to(SOURCE_ROOT).as_posix()
                seen.add(relative)
                try:
                    current_fingerprint = fingerprint(source)
                except OSError:
                    continue
                old = inventory.get(relative) or {}
                if old.get("fingerprint") == current_fingerprint:
                    if relative in catalog and catalog[relative].get("item_id"):
                        old["item_id"] = catalog[relative]["item_id"]
                    continue
                details = catalog.get(relative)
                view = VIEW_ROOT / relative
                view_uses_cache = view.is_symlink() and os.readlink(view) == cache_target(relative)
                if details and not old and not view_uses_cache:
                    details = dict(details)
                else:
                    try:
                        details = self.probe(source)
                    except Exception as error:
                        log(f"Probe failed for {relative}: {error}")
                        self.state["retry_after"][relative] = time.time() + 21600
                        continue
                    if relative in catalog:
                        details["item_id"] = catalog[relative].get("item_id")
                details["fingerprint"] = current_fingerprint
                details["size"] = source.stat().st_size
                details["mtime"] = source.stat().st_mtime
                inventory[relative] = details
                self.state["retry_after"].pop(relative, None)
                added = True

        removed = set(inventory) - seen
        for relative in removed:
            if self.is_locked(relative):
                continue
            self.remove_cache(relative)
            inventory.pop(relative, None)
            self.state["priorities"].pop(relative, None)
            self.state["retry_after"].pop(relative, None)
            added = True
        if added:
            self.save()
        return added

    def sync_view(self):
        changed = False
        source_files = set()
        for root, directories, files in os.walk(SOURCE_ROOT):
            relative_root = Path(root).relative_to(SOURCE_ROOT)
            (VIEW_ROOT / relative_root).mkdir(parents=True, exist_ok=True)
            directories.sort()
            files.sort()
            for filename in files:
                relative = (relative_root / filename).as_posix()
                source_files.add(relative)
                is_video = Path(filename).suffix.lower() in VIDEO_EXTENSIONS
                valid_cache = is_video and self.cache_valid(relative)
                if valid_cache and not self.is_locked(relative):
                    changed |= self.swap_to_cache(relative)
                else:
                    changed |= self.ensure_link(
                        relative, source_target(relative), locked=is_video and self.is_locked(relative)
                    )

        for root, directories, files in os.walk(VIEW_ROOT, topdown=False):
            for filename in files:
                path = Path(root) / filename
                relative = path.relative_to(VIEW_ROOT).as_posix()
                if relative.endswith(".pretranscode-tmp"):
                    path.unlink(missing_ok=True)
                elif relative not in source_files and path.is_symlink() and not self.is_locked(relative):
                    path.unlink(missing_ok=True)
                    changed = True
            for directory in directories:
                path = Path(root) / directory
                try:
                    path.rmdir()
                except OSError:
                    pass
        return changed

    def incompatible(self, info):
        codec = (info.get("codec") or "").lower()
        depth = int(info.get("bit_depth") or 8)
        profile = (info.get("profile") or "").lower()
        return codec in BLACKLISTED_CODECS or (codec == "h264" and (depth > 8 or "10" in profile))

    def candidate_list(self):
        now = time.time()
        values = []
        for relative, info in self.state["inventory"].items():
            if not self.incompatible(info) or self.cache_valid(relative) or self.is_locked(relative):
                continue
            if float(self.state["retry_after"].get(relative, 0)) > now:
                continue
            priority = self.state["priorities"].get(relative)
            values.append((0 if priority is not None else 1, -float(priority or info.get("mtime", 0)), relative))
        values.sort()
        return values

    def cache_entries(self):
        entries = []
        for manifest_path in (CACHE_ROOT / "files").glob("*.json"):
            manifest = load_json(manifest_path, {})
            relative = manifest.get("relative")
            if not relative:
                continue
            output = manifest_path.with_suffix("")
            if not output.is_file():
                manifest_path.unlink(missing_ok=True)
                continue
            entries.append((output.stat().st_mtime, relative, output, manifest_path))
        return sorted(entries)

    def remove_cache(self, relative):
        if self.is_locked(relative):
            return False
        output, manifest = self.cache_paths(relative)
        link = VIEW_ROOT / relative
        if link.is_symlink():
            try:
                if os.readlink(link) == cache_target(relative):
                    self.ensure_link(relative, source_target(relative), locked=False)
                    self.refresh_item(relative)
            except OSError:
                pass
        output.unlink(missing_ok=True)
        manifest.unlink(missing_ok=True)
        return True

    def cache_usage(self):
        return sum(
            path.stat().st_size
            for path in (CACHE_ROOT / "files").iterdir()
            if path.is_file()
        )

    def cache_available(self):
        stat = os.statvfs(CACHE_ROOT)
        filesystem_free = stat.f_bavail * stat.f_frsize
        configured_free = max(0, CACHE_MAX_BYTES - self.cache_usage())
        return min(filesystem_free, configured_free)

    def enforce_cache_limit(self):
        usage = self.cache_usage()
        for _, relative, output, manifest in self.cache_entries():
            if usage <= CACHE_MAX_BYTES:
                break
            if self.is_locked(relative):
                continue
            size = output.stat().st_size + manifest.stat().st_size
            log(f"Evicting {relative} after cache resize")
            if self.remove_cache(relative):
                usage -= size

    def ensure_space(self, required, may_evict):
        if self.cache_available() >= required:
            return True
        if not may_evict:
            return False
        for _, relative, _, _ in self.cache_entries():
            if self.is_locked(relative):
                continue
            log(f"Evicting {relative}")
            self.remove_cache(relative)
            if self.cache_available() >= required:
                return True
        return False

    def cancel_process(self, reason):
        if not self.process or self.process.poll() is not None:
            return
        log(f"Stopping encode: {reason}")
        try:
            os.killpg(self.process.pid, signal.SIGTERM)
            self.process.wait(timeout=10)
        except (ProcessLookupError, subprocess.TimeoutExpired):
            try:
                os.killpg(self.process.pid, signal.SIGKILL)
            except ProcessLookupError:
                pass
        self.process = None

    def ffmpeg_command(self, source, partial):
        common = [
            FFMPEG, "-nostdin", "-hide_banner", "-loglevel", "warning", "-i", str(source),
            "-map", "0:v:0", "-map", "0:a?", "-map", "0:s?", "-map_metadata", "0", "-map_chapters", "0",
            "-c:v", "libx264", "-preset", H264_PRESET, "-crf", H264_CRF,
            "-maxrate", H264_MAXRATE, "-bufsize", H264_BUFSIZE,
            "-pix_fmt", "yuv420p", "-profile:v", "high", "-level:v", "4.1",
            "-c:a", "copy", "-c:s", "copy",
        ]
        if partial.name.endswith(".mp4.partial"):
            common += ["-movflags", "+faststart", "-f", "mp4"]
        else:
            common += ["-map", "0:t?", "-c:t", "copy", "-f", "matroska"]
        return common + [str(partial)]

    def validate_output(self, source, output, expected_duration):
        details = self.probe(output)
        if details.get("codec") != "h264" or int(details.get("bit_depth") or 0) > 8:
            raise RuntimeError("output is not 8-bit H.264")
        source_duration = float(expected_duration or 0)
        output_duration = float(details.get("duration") or 0)
        if source_duration and abs(source_duration - output_duration) > max(3, source_duration * 0.01):
            raise RuntimeError(f"duration mismatch: source={source_duration}, output={output_duration}")
        if output.stat().st_size < 1_000_000:
            raise RuntimeError("output is unexpectedly small")

    def encode(self, relative, priority):
        source = SOURCE_ROOT / relative
        if not source.is_file() or self.is_locked(relative):
            return "skipped"
        output, manifest_path = self.cache_paths(relative)
        partial = output.with_name(output.name + ".partial")
        partial.unlink(missing_ok=True)
        info = self.state["inventory"].get(relative) or {}
        estimate = min(max(int(source.stat().st_size * 2.2), 1_000_000_000), 100_000_000_000)
        if not self.ensure_space(estimate, may_evict=priority):
            return "full"

        log(f"Encoding {relative}")
        command = self.ffmpeg_command(source, partial)
        log_path = STATE_ROOT / "ffmpeg.log"
        with log_path.open("ab", buffering=0) as ffmpeg_log:
            self.process = subprocess.Popen(
                command, stdout=ffmpeg_log, stderr=ffmpeg_log, start_new_session=True
            )
            last_mirror = 0
            while self.running and self.process.poll() is None:
                time.sleep(POLL_SECONDS)
                if not self.read_sessions() or self.is_locked(relative):
                    self.cancel_process("item or series is being watched")
                    partial.unlink(missing_ok=True)
                    return "cancelled"
                if time.time() - last_mirror >= 60:
                    if self.sync_view():
                        try:
                            self.api("/Library/Refresh", method="POST")
                        except Exception as error:
                            log(f"Unable to request library refresh: {error}")
                    last_mirror = time.time()
            returncode = self.process.wait() if self.process else -1
            self.process = None

        if not self.running:
            partial.unlink(missing_ok=True)
            return "cancelled"
        if returncode != 0:
            partial.unlink(missing_ok=True)
            raise RuntimeError(f"ffmpeg exited with status {returncode}")
        if self.is_locked(relative):
            partial.unlink(missing_ok=True)
            return "cancelled"

        self.validate_output(source, partial, info.get("duration"))
        os.replace(partial, output)
        atomic_json(manifest_path, {
            "relative": relative,
            "source_fingerprint": fingerprint(source),
            "created_at": time.time(),
        })
        if self.is_locked(relative):
            output.unlink(missing_ok=True)
            manifest_path.unlink(missing_ok=True)
            return "cancelled"
        self.swap_to_cache(relative)
        self.state["priorities"].pop(relative, None)
        self.state["retry_after"].pop(relative, None)
        self.save()
        log(f"Completed {relative}")
        return "completed"

    def run(self):
        if not JELLYFIN_API_KEY:
            raise RuntimeError("JELLYFIN_API_KEY is required")
        self.read_sessions()
        self.enforce_cache_limit()
        self.scan_inventory()
        self.sync_view()
        self.save()
        while self.running:
            if not self.read_sessions():
                time.sleep(POLL_SECONDS)
                continue
            self.enforce_cache_limit()
            now = time.time()
            if now - self.last_scan >= RESCAN_SECONDS:
                changed = self.scan_inventory()
                changed |= self.sync_view()
                if changed:
                    try:
                        self.api("/Library/Refresh", method="POST")
                    except Exception as error:
                        log(f"Unable to request library refresh: {error}")
                self.last_scan = now

            candidates = self.candidate_list()
            if not candidates:
                time.sleep(POLL_SECONDS)
                continue
            selected = None
            for rank, _, relative in candidates:
                source = SOURCE_ROOT / relative
                estimate = min(max(int(source.stat().st_size * 2.2), 1_000_000_000), 100_000_000_000)
                if rank == 0 or self.cache_available() >= estimate:
                    selected = (rank == 0, relative)
                    break
            if not selected:
                time.sleep(60)
                continue
            priority, relative = selected
            try:
                result = self.encode(relative, priority)
                if result == "full":
                    time.sleep(60)
            except Exception as error:
                log(f"Encode failed for {relative}: {error}")
                self.state["retry_after"][relative] = time.time() + 21600
                self.save()
                output, manifest = self.cache_paths(relative)
                output.with_name(output.name + ".partial").unlink(missing_ok=True)
                time.sleep(POLL_SECONDS)

    def sync_once(self):
        self.scan_inventory()
        self.sync_view()
        self.save()

    def active(self):
        if not self.read_sessions():
            return True
        return bool(self.active_items)

    def trim(self, target_bytes):
        if self.active():
            raise RuntimeError("cache resize refused while media is playing")
        target = max(0, int(target_bytes) - 5_000_000_000)
        entries = self.cache_entries()
        used = sum(output.stat().st_size + manifest.stat().st_size for _, _, output, manifest in entries)
        for _, relative, output, manifest in entries:
            if used <= target:
                break
            size = output.stat().st_size + manifest.stat().st_size
            if self.remove_cache(relative):
                log(f"Trimmed {relative}")
                used -= size
        if used > target:
            raise RuntimeError("unable to trim cache below target because content is locked")
        self.sync_view()
        self.save()


def main():
    manager = Manager()
    command = sys.argv[1] if len(sys.argv) > 1 else "run"
    if command == "run":
        manager.run()
    elif command == "sync":
        manager.sync_once()
    elif command == "active":
        raise SystemExit(0 if not manager.active() else 1)
    elif command == "trim" and len(sys.argv) == 3:
        manager.trim(int(sys.argv[2]))
    else:
        raise SystemExit(f"unknown command: {command}")


if __name__ == "__main__":
    main()