diff --git a/controller/models.py b/controller/models.py new file mode 100644 index 0000000..c06467f --- /dev/null +++ b/controller/models.py @@ -0,0 +1,113 @@ +"""Controller domain records shared by persistence and orchestration.""" + +from __future__ import annotations + +from dataclasses import dataclass + + +MEDIA_STATES = frozenset( + {"empty", "queued", "downloading", "ready", "error", "deleting"} +) +LIVE_JOB_STATES = frozenset({"reserved", "running"}) +LIVE_GUARD_STATES = frozenset({"active", "review_required"}) + + +@dataclass(frozen=True) +class SlotRecord: + slot: int + generation: int + media_state: str + url: str | None + yt_id: str | None + title: str + item_count: int + published_item_count: int + created_at: float | None + ready_at: float | None + expires_at: float | None + last_error: str | None + updated_at: float + + +@dataclass(frozen=True) +class DownloadReservation: + status: str + slot: int | None = None + generation: int | None = None + job_id: str | None = None + reused_slot: bool = False + oldest_expires_at: float | None = None + + +@dataclass(frozen=True) +class DownloadJob: + job_id: str + slot: int + generation: int + url: str + status: str + submitter: str | None + note: str | None + requested_at: float + started_at: float | None + finished_at: float | None + error: str | None + + +@dataclass(frozen=True) +class DeleteToken: + slot: int + generation: int + reason: str + + +@dataclass(frozen=True) +class SlotGuard: + guard_id: str + slot: int + kind: str + state: str + created_at: float + review_after: float | None + cleared_at: float | None + cleared_by: str | None + clear_reason: str | None + + +@dataclass(frozen=True) +class PlayerObservation: + availability: str + phase: str + active_slot: int | None + active_path: str | None + paused: bool | None + generation: int + reconciliation_state: str + reconciliation_due_at: float | None + last_status_at: float | None + last_telemetry_at: float | None + + +@dataclass(frozen=True) +class MigrationReport: + migrated: bool + source_path: str | None = None + source_sha256: str | None = None + backup_path: str | None = None + warnings: tuple[str, ...] = () + + +class StoreError(RuntimeError): + """Base persistence error.""" + + +class MigrationError(StoreError): + """Legacy migration was rejected or could not complete safely.""" + + +class SlotProtectedError(StoreError): + """A slot cannot be modified because live activity protects it.""" + + +class StaleOperationError(StoreError): + """An operation token no longer identifies the slot occupant.""" diff --git a/controller/store.py b/controller/store.py new file mode 100644 index 0000000..473c9e7 --- /dev/null +++ b/controller/store.py @@ -0,0 +1,1311 @@ +"""Transactional SQLite persistence and non-destructive v1 YAML migration.""" + +from __future__ import annotations + +import contextlib +import datetime as dt +import hashlib +import json +import os +import sqlite3 +import tempfile +import threading +import time +import uuid +from pathlib import Path +from typing import Any, Iterable, Iterator, Mapping, Sequence + +import yaml + +from .config import StorageConfig +from .contracts import SLOT_IDS +from .models import ( + DeleteToken, + DownloadJob, + DownloadReservation, + MigrationError, + MigrationReport, + PlayerObservation, + SlotGuard, + SlotProtectedError, + SlotRecord, + StaleOperationError, +) + + +SCHEMA_VERSION = 1 +LEGACY_MIGRATION_NAME = "legacy_playlists_yaml_v1" +MAX_LEGACY_YAML_BYTES = 8 * 1024 * 1024 + + +SCHEMA_SQL = """ +CREATE TABLE IF NOT EXISTS schema_migrations ( + version INTEGER PRIMARY KEY, + name TEXT NOT NULL UNIQUE, + applied_at REAL NOT NULL +); + +CREATE TABLE IF NOT EXISTS data_migrations ( + name TEXT PRIMARY KEY, + applied_at REAL NOT NULL, + source_path TEXT, + source_sha256 TEXT, + backup_path TEXT, + details_json TEXT NOT NULL DEFAULT '{}' +); + +CREATE TABLE IF NOT EXISTS slots ( + slot_number INTEGER PRIMARY KEY CHECK (slot_number BETWEEN 1 AND 10), + generation INTEGER NOT NULL DEFAULT 0 CHECK (generation >= 0), + media_state TEXT NOT NULL DEFAULT 'empty' + CHECK (media_state IN ('empty','queued','downloading','ready','error','deleting')), + url TEXT, + youtube_id TEXT, + title TEXT NOT NULL DEFAULT '', + item_count INTEGER NOT NULL DEFAULT 0 CHECK (item_count >= 0), + published_item_count INTEGER NOT NULL DEFAULT 0 CHECK (published_item_count >= 0), + created_at REAL, + ready_at REAL, + expires_at REAL, + last_error TEXT, + updated_at REAL NOT NULL, + CHECK (media_state <> 'empty' OR (url IS NULL AND item_count = 0)) +); + +CREATE TABLE IF NOT EXISTS slot_items ( + slot_number INTEGER NOT NULL REFERENCES slots(slot_number), + position INTEGER NOT NULL CHECK (position >= 1), + generation INTEGER NOT NULL CHECK (generation >= 0), + title TEXT NOT NULL, + relative_path TEXT NOT NULL, + PRIMARY KEY (slot_number, position) +); + +CREATE TABLE IF NOT EXISTS download_jobs ( + job_id TEXT PRIMARY KEY, + slot_number INTEGER NOT NULL REFERENCES slots(slot_number), + generation INTEGER NOT NULL CHECK (generation >= 1), + url TEXT NOT NULL, + status TEXT NOT NULL + CHECK (status IN ('reserved','running','succeeded','failed','cancelled','interrupted','stale')), + submitter TEXT, + note TEXT, + requested_at REAL NOT NULL, + started_at REAL, + finished_at REAL, + error TEXT, + UNIQUE (slot_number, generation) +); + +CREATE UNIQUE INDEX IF NOT EXISTS one_live_download_per_slot +ON download_jobs(slot_number) +WHERE status IN ('reserved','running'); + +CREATE TABLE IF NOT EXISTS slot_guards ( + guard_id TEXT PRIMARY KEY, + slot_number INTEGER NOT NULL REFERENCES slots(slot_number), + kind TEXT NOT NULL CHECK (kind IN + ('player_active','pending_play','restart_uncertain','legacy_uncertain')), + state TEXT NOT NULL CHECK (state IN ('active','review_required','cleared')), + created_at REAL NOT NULL, + review_after REAL, + cleared_at REAL, + cleared_by TEXT, + clear_reason TEXT, + details_json TEXT NOT NULL DEFAULT '{}' +); + +CREATE UNIQUE INDEX IF NOT EXISTS one_live_guard_per_slot +ON slot_guards(slot_number) +WHERE state IN ('active','review_required'); + +CREATE TABLE IF NOT EXISTS player_observation ( + singleton INTEGER PRIMARY KEY CHECK (singleton = 1), + availability TEXT NOT NULL CHECK (availability IN ('unknown','online','offline')), + phase TEXT NOT NULL CHECK (phase IN ('unknown','idle','playing','paused','error')), + active_slot INTEGER REFERENCES slots(slot_number), + active_path TEXT, + paused INTEGER CHECK (paused IS NULL OR paused IN (0,1)), + generation INTEGER NOT NULL DEFAULT 0, + reconciliation_state TEXT NOT NULL CHECK (reconciliation_state IN + ('clear','confirmed','awaiting_evidence','operator_review','admin_cleared','unknown')), + reconciliation_due_at REAL, + last_status_at REAL, + last_telemetry_at REAL, + last_status_json TEXT, + last_telemetry_json TEXT +); + +CREATE TABLE IF NOT EXISTS audio_router_observation ( + singleton INTEGER PRIMARY KEY CHECK (singleton = 1), + mode TEXT, + freshness TEXT NOT NULL CHECK (freshness IN ('unknown','current','stale')), + observed_at REAL, + last_warning TEXT +); + +CREATE TABLE IF NOT EXISTS audio_leases ( + lease_id TEXT PRIMARY KEY, + operation_id TEXT, + playback_generation INTEGER, + state TEXT NOT NULL CHECK (state IN + ('acquiring','held','unconfirmed','ownership_lost','restore_pending','released','abandoned')), + previous_mode TEXT, + target_mode TEXT NOT NULL, + restoration_mode TEXT, + restoration_source TEXT NOT NULL CHECK (restoration_source IN + ('observed_previous','configured_fallback','none')), + controller_changed INTEGER NOT NULL DEFAULT 0 CHECK (controller_changed IN (0,1)), + confirmation_deadline REAL, + created_at REAL NOT NULL, + acquired_at REAL, + released_at REAL, + warning TEXT, + error TEXT +); + +CREATE UNIQUE INDEX IF NOT EXISTS one_live_audio_lease +ON audio_leases((1)) +WHERE state IN ('acquiring','held','unconfirmed','restore_pending'); + +CREATE TABLE IF NOT EXISTS audit_events ( + event_id INTEGER PRIMARY KEY AUTOINCREMENT, + occurred_at REAL NOT NULL, + event_type TEXT NOT NULL, + slot_number INTEGER REFERENCES slots(slot_number), + actor TEXT, + details_json TEXT NOT NULL DEFAULT '{}' +); +""" + + +class _UniqueKeyLoader(yaml.SafeLoader): + pass + + +def _construct_unique_mapping( + loader: _UniqueKeyLoader, + node: yaml.MappingNode, + deep: bool = False, +) -> dict[Any, Any]: + result: dict[Any, Any] = {} + for key_node, value_node in node.value: + key = loader.construct_object(key_node, deep=deep) + try: + duplicate = key in result + except TypeError as exc: + raise MigrationError("legacy YAML contains an unhashable mapping key") from exc + if duplicate: + raise MigrationError(f"legacy YAML contains duplicate key {key!r}") + result[key] = loader.construct_object(value_node, deep=deep) + return result + + +_UniqueKeyLoader.add_constructor( + yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG, + _construct_unique_mapping, +) + + +def _utc_timestamp(value: Any, field: str, warnings: list[str]) -> float | None: + if value in (None, ""): + return None + if isinstance(value, (int, float)) and not isinstance(value, bool): + return float(value) + if isinstance(value, dt.datetime): + if value.tzinfo is None: + value = value.replace(tzinfo=dt.timezone.utc) + return value.timestamp() + if isinstance(value, dt.date): + return dt.datetime.combine(value, dt.time(), tzinfo=dt.timezone.utc).timestamp() + if isinstance(value, str): + try: + parsed = dt.datetime.fromisoformat(value.replace("Z", "+00:00")) + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=dt.timezone.utc) + return parsed.timestamp() + except ValueError: + pass + warnings.append(f"ignored invalid {field}: {value!r}") + return None + + +def _coerce_count(value: Any, slot: int, warnings: list[str]) -> int: + if isinstance(value, bool): + warnings.append(f"slot {slot}: invalid item_count {value!r}; used 0") + return 0 + try: + count = int(value) + except (TypeError, ValueError, OverflowError): + count = -1 + if count < 0: + warnings.append(f"slot {slot}: invalid item_count {value!r}; used 0") + return 0 + return count + + +def _row_to_slot(row: sqlite3.Row) -> SlotRecord: + return SlotRecord( + slot=row["slot_number"], + generation=row["generation"], + media_state=row["media_state"], + url=row["url"], + yt_id=row["youtube_id"], + title=row["title"], + item_count=row["item_count"], + published_item_count=row["published_item_count"], + created_at=row["created_at"], + ready_at=row["ready_at"], + expires_at=row["expires_at"], + last_error=row["last_error"], + updated_at=row["updated_at"], + ) + + +class ControllerStore: + """Thread-safe transactional store. + + Controller v2's coordinator is the intended sole writer. The re-entrant + lock also makes direct calls safe in HTTP/tests and SQLite transactions are + the final cross-connection reservation boundary. + """ + + def __init__(self, config: StorageConfig, *, now=time.time): + self.config = config + self.path = config.database_path + self.legacy_path = config.legacy_yaml_path + self._now = now + self._lock = threading.RLock() + self._connection: sqlite3.Connection | None = None + self.migration_report = MigrationReport(migrated=False) + + def open(self) -> "ControllerStore": + with self._lock: + if self._connection is not None: + return self + self.config.data_base.mkdir(parents=True, exist_ok=True) + if not self.path.exists(): + if self.legacy_path.exists(): + self.migration_report = self._migrate_legacy_yaml() + else: + self._create_new_database() + self._connection = self._connect(self.path) + self._validate_database(self._connection) + self._recover_interrupted_jobs() + return self + + def close(self) -> None: + with self._lock: + if self._connection is not None: + self._connection.close() + self._connection = None + + def __enter__(self) -> "ControllerStore": + return self.open() + + def __exit__(self, exc_type, exc, traceback) -> None: + self.close() + + @property + def connection(self) -> sqlite3.Connection: + if self._connection is None: + raise RuntimeError("ControllerStore is not open") + return self._connection + + @staticmethod + def _connect(path: Path, *, runtime: bool = True) -> sqlite3.Connection: + connection = sqlite3.connect( + path, + timeout=5.0, + isolation_level=None, + check_same_thread=False, + ) + connection.row_factory = sqlite3.Row + connection.execute("PRAGMA foreign_keys = ON") + connection.execute("PRAGMA busy_timeout = 5000") + connection.execute("PRAGMA synchronous = FULL") + if runtime: + connection.execute("PRAGMA journal_mode = WAL") + else: + connection.execute("PRAGMA journal_mode = DELETE") + return connection + + @contextlib.contextmanager + def _transaction(self) -> Iterator[sqlite3.Connection]: + with self._lock: + connection = self.connection + connection.execute("BEGIN IMMEDIATE") + try: + yield connection + except BaseException: + connection.rollback() + raise + else: + connection.commit() + + def _initialize_schema(self, connection: sqlite3.Connection) -> None: + connection.executescript(SCHEMA_SQL) + now = self._now() + connection.executemany( + """ + INSERT OR IGNORE INTO slots + (slot_number, generation, media_state, updated_at) + VALUES (?, 0, 'empty', ?) + """, + ((slot, now) for slot in SLOT_IDS), + ) + connection.execute( + """ + INSERT OR IGNORE INTO player_observation + (singleton, availability, phase, generation, reconciliation_state) + VALUES (1, 'unknown', 'unknown', 0, 'clear') + """ + ) + connection.execute( + """ + INSERT OR IGNORE INTO audio_router_observation + (singleton, freshness) + VALUES (1, 'unknown') + """ + ) + connection.execute( + """ + INSERT OR IGNORE INTO schema_migrations(version, name, applied_at) + VALUES (?, ?, ?) + """, + (SCHEMA_VERSION, "controller_v2_initial", now), + ) + connection.execute(f"PRAGMA user_version = {SCHEMA_VERSION}") + + def _create_new_database(self) -> None: + staging = self._staging_path() + try: + connection = self._connect(staging, runtime=False) + try: + self._initialize_schema(connection) + connection.execute( + """ + INSERT INTO data_migrations + (name, applied_at, details_json) + VALUES ('fresh_database_v1', ?, '{}') + """, + (self._now(),), + ) + self._validate_database(connection) + finally: + connection.close() + self._fsync_file(staging) + os.replace(staging, self.path) + self._fsync_directory(self.path.parent) + except BaseException: + staging.unlink(missing_ok=True) + raise + + def _staging_path(self) -> Path: + descriptor, raw_path = tempfile.mkstemp( + prefix=f".{self.path.name}.", + suffix=".staging", + dir=self.path.parent, + ) + os.close(descriptor) + return Path(raw_path) + + @staticmethod + def _fsync_file(path: Path) -> None: + descriptor = os.open(path, os.O_RDONLY) + try: + os.fsync(descriptor) + finally: + os.close(descriptor) + + @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 _create_backup(self, source: bytes, digest: str) -> Path: + backup = self.legacy_path.with_name( + f"{self.legacy_path.name}.pre-v2.{digest}.bak" + ) + if backup.exists(): + if hashlib.sha256(backup.read_bytes()).hexdigest() != digest: + raise MigrationError(f"existing migration backup is invalid: {backup}") + return backup + + descriptor, raw_temp = tempfile.mkstemp( + prefix=f".{backup.name}.", + suffix=".tmp", + dir=backup.parent, + ) + temp = Path(raw_temp) + try: + with os.fdopen(descriptor, "wb") as stream: + stream.write(source) + stream.flush() + os.fsync(stream.fileno()) + try: + os.link(temp, backup) + except FileExistsError: + if hashlib.sha256(backup.read_bytes()).hexdigest() != digest: + raise MigrationError(f"existing migration backup is invalid: {backup}") + self._fsync_directory(backup.parent) + finally: + temp.unlink(missing_ok=True) + + if hashlib.sha256(backup.read_bytes()).hexdigest() != digest: + raise MigrationError("migration backup verification failed") + return backup + + def _parse_legacy(self, source: bytes) -> tuple[dict[int, Mapping[str, Any]], list[str]]: + try: + text = source.decode("utf-8") + except UnicodeDecodeError as exc: + raise MigrationError("legacy YAML is not valid UTF-8") from exc + try: + raw = yaml.load(text, Loader=_UniqueKeyLoader) + except (yaml.YAMLError, MigrationError) as exc: + if isinstance(exc, MigrationError): + raise + raise MigrationError(f"legacy YAML cannot be parsed: {exc}") from exc + if raw is None: + raw = {} + if not isinstance(raw, Mapping): + raise MigrationError("legacy YAML root must be a mapping") + raw_slots = raw.get("slots", {}) + if not isinstance(raw_slots, Mapping): + raise MigrationError("legacy YAML slots must be a mapping") + + normalized: dict[int, Mapping[str, Any]] = {} + original_keys: dict[int, Any] = {} + for key, value in raw_slots.items(): + if isinstance(key, bool): + raise MigrationError(f"invalid legacy slot key {key!r}") + try: + slot = int(key) + except (TypeError, ValueError, OverflowError) as exc: + raise MigrationError(f"invalid legacy slot key {key!r}") from exc + if slot not in SLOT_IDS: + raise MigrationError(f"legacy slot {slot} is outside stable range 1..10") + if slot in normalized: + raise MigrationError( + f"legacy slot keys {original_keys[slot]!r} and {key!r} normalize to {slot}" + ) + if not isinstance(value, Mapping): + raise MigrationError(f"legacy slot {slot} must be a mapping") + embedded = value.get("slot", slot) + if isinstance(embedded, bool): + raise MigrationError(f"legacy slot {slot} has invalid embedded identity") + try: + embedded_slot = int(embedded) + except (TypeError, ValueError, OverflowError) as exc: + raise MigrationError(f"legacy slot {slot} has invalid embedded identity") from exc + if embedded_slot != slot: + raise MigrationError( + f"legacy slot key {slot} conflicts with embedded slot {embedded_slot}" + ) + normalized[slot] = value + original_keys[slot] = key + return normalized, [] + + def _migrate_legacy_yaml(self) -> MigrationReport: + source = self.legacy_path.read_bytes() + if len(source) > MAX_LEGACY_YAML_BYTES: + raise MigrationError("legacy YAML exceeds migration size limit") + digest = hashlib.sha256(source).hexdigest() + backup = self._create_backup(source, digest) + slots, warnings = self._parse_legacy(source) + staging = self._staging_path() + try: + connection = self._connect(staging, runtime=False) + try: + self._initialize_schema(connection) + connection.execute("BEGIN IMMEDIATE") + try: + now = self._now() + for slot_number, legacy in sorted(slots.items()): + self._import_legacy_slot( + connection, + slot_number, + legacy, + warnings, + now, + ) + details = json.dumps({"warnings": warnings}, sort_keys=True) + connection.execute( + """ + INSERT INTO data_migrations + (name, applied_at, source_path, source_sha256, + backup_path, details_json) + VALUES (?, ?, ?, ?, ?, ?) + """, + ( + LEGACY_MIGRATION_NAME, + now, + str(self.legacy_path), + digest, + str(backup), + details, + ), + ) + connection.commit() + except BaseException: + connection.rollback() + raise + self._validate_database(connection) + finally: + connection.close() + self._fsync_file(staging) + os.replace(staging, self.path) + self._fsync_directory(self.path.parent) + except BaseException: + staging.unlink(missing_ok=True) + raise + return MigrationReport( + migrated=True, + source_path=str(self.legacy_path), + source_sha256=digest, + backup_path=str(backup), + warnings=tuple(warnings), + ) + + def _import_legacy_slot( + self, + connection: sqlite3.Connection, + slot: int, + legacy: Mapping[str, Any], + warnings: list[str], + now: float, + ) -> None: + raw_state = str(legacy.get("state", "ready")).strip().lower() + error: str | None = None + guarded = raw_state in {"playing", "paused", "busy"} + interrupted = raw_state in {"queued", "downloading"} + if raw_state in {"ready", "idle", "finished", "playing", "paused", "busy"}: + media_state = "ready" + elif raw_state == "error": + media_state = "error" + error = str(legacy.get("last_error") or "legacy download error") + elif interrupted: + media_state = "error" + error = "download interrupted during Controller v2 migration" + warnings.append(f"slot {slot}: legacy {raw_state} download marked interrupted") + else: + media_state = "error" + error = f"unknown legacy state: {raw_state}" + warnings.append(f"slot {slot}: unknown legacy state {raw_state!r}") + + url_value = legacy.get("url") + url = str(url_value) if url_value not in (None, "") else None + if url is None: + media_state = "error" + error = error or "legacy slot has no URL" + warnings.append(f"slot {slot}: missing URL") + title = str(legacy.get("title") or "") + count = _coerce_count(legacy.get("item_count", 0), slot, warnings) + created_at = _utc_timestamp(legacy.get("created_at"), f"slot {slot} created_at", warnings) + expires_at = _utc_timestamp(legacy.get("expires_at"), f"slot {slot} expires_at", warnings) + generation = 1 + connection.execute( + """ + UPDATE slots + SET generation = ?, media_state = ?, url = ?, youtube_id = ?, + title = ?, item_count = ?, published_item_count = ?, + created_at = ?, ready_at = ?, expires_at = ?, last_error = ?, + updated_at = ? + WHERE slot_number = ? + """, + ( + generation, + media_state, + url, + legacy.get("yt_id"), + title, + count, + count, + created_at, + created_at if media_state == "ready" else None, + expires_at, + error, + now, + slot, + ), + ) + if guarded: + guard_id = f"legacy-{slot}-{generation}" + connection.execute( + """ + INSERT INTO slot_guards + (guard_id, slot_number, kind, state, created_at, + review_after, details_json) + VALUES (?, ?, 'legacy_uncertain', 'review_required', ?, NULL, ?) + """, + ( + guard_id, + slot, + now, + json.dumps({"legacy_state": raw_state}, sort_keys=True), + ), + ) + if interrupted and url is not None: + connection.execute( + """ + INSERT INTO download_jobs + (job_id, slot_number, generation, url, status, + requested_at, finished_at, error) + VALUES (?, ?, ?, ?, 'interrupted', ?, ?, ?) + """, + ( + f"legacy-interrupted-{slot}-{generation}", + slot, + generation, + url, + created_at or now, + now, + error, + ), + ) + connection.execute( + """ + INSERT INTO audit_events + (occurred_at, event_type, slot_number, actor, details_json) + VALUES (?, 'legacy_slot_migrated', ?, 'migration', ?) + """, + (now, slot, json.dumps({"legacy_state": raw_state}, sort_keys=True)), + ) + + @staticmethod + def _validate_database(connection: sqlite3.Connection) -> None: + version = connection.execute("PRAGMA user_version").fetchone()[0] + if version != SCHEMA_VERSION: + raise MigrationError( + f"unsupported Controller database schema {version}; expected {SCHEMA_VERSION}" + ) + violations = connection.execute("PRAGMA foreign_key_check").fetchall() + if violations: + raise MigrationError(f"Controller database foreign-key failure: {violations!r}") + integrity = connection.execute("PRAGMA integrity_check").fetchone()[0] + if integrity != "ok": + raise MigrationError(f"Controller database integrity failure: {integrity}") + slots = connection.execute( + "SELECT slot_number FROM slots ORDER BY slot_number" + ).fetchall() + if [row[0] for row in slots] != list(SLOT_IDS): + raise MigrationError("Controller database must contain stable slots 1..10") + + def _recover_interrupted_jobs(self) -> None: + with self._transaction() as connection: + now = self._now() + live = connection.execute( + """ + SELECT job_id, slot_number, generation + FROM download_jobs + WHERE status IN ('reserved','running') + """ + ).fetchall() + for row in live: + message = "download interrupted by Controller restart" + connection.execute( + """ + UPDATE download_jobs + SET status = 'interrupted', finished_at = ?, error = ? + WHERE job_id = ? + """, + (now, message, row["job_id"]), + ) + connection.execute( + """ + UPDATE slots + SET media_state = 'error', last_error = ?, updated_at = ? + WHERE slot_number = ? AND generation = ? + """, + (message, now, row["slot_number"], row["generation"]), + ) + + def list_slots(self) -> list[SlotRecord]: + with self._lock: + rows = self.connection.execute( + "SELECT * FROM slots ORDER BY slot_number" + ).fetchall() + return [_row_to_slot(row) for row in rows] + + def get_slot(self, slot: int) -> SlotRecord: + if slot not in SLOT_IDS: + raise ValueError(f"slot must be in 1..10, got {slot!r}") + with self._lock: + row = self.connection.execute( + "SELECT * FROM slots WHERE slot_number = ?", (slot,) + ).fetchone() + assert row is not None + return _row_to_slot(row) + + def get_slot_items(self, slot: int) -> list[tuple[str, str]]: + with self._lock: + rows = self.connection.execute( + """ + SELECT title, relative_path FROM slot_items + WHERE slot_number = ? ORDER BY position + """, + (slot,), + ).fetchall() + return [(row["title"], row["relative_path"]) for row in rows] + + def reserve_download( + self, + url: str, + submitter: str | None, + note: str | None, + *, + job_id: str | None = None, + ) -> DownloadReservation: + if not isinstance(url, str) or not url.strip(): + raise ValueError("download URL must be non-empty") + url = url.strip() + job_id = job_id or uuid.uuid4().hex + with self._transaction() as connection: + existing = connection.execute( + """ + SELECT s.* FROM slots s + WHERE s.url = ? AND s.media_state <> 'empty' + ORDER BY s.slot_number LIMIT 1 + """, + (url,), + ).fetchone() + reused = existing is not None + if existing is not None: + slot = existing["slot_number"] + if existing["media_state"] in {"queued", "downloading", "deleting"}: + return DownloadReservation(status="slot_busy", slot=slot, reused_slot=True) + if self._slot_has_live_guard(connection, slot): + return DownloadReservation(status="slot_protected", slot=slot, reused_slot=True) + else: + free = connection.execute( + """ + SELECT s.slot_number FROM slots s + WHERE s.media_state = 'empty' + AND NOT EXISTS ( + SELECT 1 FROM slot_guards g + WHERE g.slot_number = s.slot_number + AND g.state IN ('active','review_required') + ) + ORDER BY s.slot_number LIMIT 1 + """ + ).fetchone() + if free is None: + oldest = connection.execute( + """ + SELECT MIN(expires_at) FROM slots + WHERE expires_at IS NOT NULL + """ + ).fetchone()[0] + return DownloadReservation( + status="queue_full", + oldest_expires_at=oldest, + ) + slot = free["slot_number"] + + current = connection.execute( + "SELECT generation FROM slots WHERE slot_number = ?", (slot,) + ).fetchone() + generation = current["generation"] + 1 + now = self._now() + connection.execute( + """ + UPDATE slots + SET generation = ?, media_state = 'queued', url = ?, + created_at = COALESCE(created_at, ?), last_error = NULL, + updated_at = ? + WHERE slot_number = ? + """, + (generation, url, now, now, slot), + ) + connection.execute( + """ + INSERT INTO download_jobs + (job_id, slot_number, generation, url, status, + submitter, note, requested_at) + VALUES (?, ?, ?, ?, 'reserved', ?, ?, ?) + """, + (job_id, slot, generation, url, submitter, note, now), + ) + return DownloadReservation( + status="reserved", + slot=slot, + generation=generation, + job_id=job_id, + reused_slot=reused, + ) + + @staticmethod + def _slot_has_live_guard(connection: sqlite3.Connection, slot: int) -> bool: + return connection.execute( + """ + SELECT 1 FROM slot_guards + WHERE slot_number = ? AND state IN ('active','review_required') + """, + (slot,), + ).fetchone() is not None + + def mark_download_running(self, job_id: str, slot: int, generation: int) -> bool: + with self._transaction() as connection: + now = self._now() + updated = connection.execute( + """ + UPDATE download_jobs SET status = 'running', started_at = ? + WHERE job_id = ? AND slot_number = ? AND generation = ? + AND status = 'reserved' + """, + (now, job_id, slot, generation), + ).rowcount + if not updated: + return False + slot_updated = connection.execute( + """ + UPDATE slots SET media_state = 'downloading', updated_at = ? + WHERE slot_number = ? AND generation = ? AND media_state = 'queued' + """, + (now, slot, generation), + ).rowcount + if not slot_updated: + raise StaleOperationError("download reservation no longer owns the slot") + return True + + def complete_download( + self, + job_id: str, + slot: int, + generation: int, + *, + title: str, + items: Sequence[tuple[str, str]], + yt_id: str | None = None, + ) -> bool: + with self._transaction() as connection: + job = connection.execute( + """ + SELECT status FROM download_jobs + WHERE job_id = ? AND slot_number = ? AND generation = ? + """, + (job_id, slot, generation), + ).fetchone() + if job is None or job["status"] not in {"reserved", "running"}: + return False + current = connection.execute( + "SELECT generation FROM slots WHERE slot_number = ?", (slot,) + ).fetchone() + if current["generation"] != generation: + connection.execute( + """ + UPDATE download_jobs SET status = 'stale', finished_at = ?, + error = 'stale worker completion' + WHERE job_id = ? + """, + (self._now(), job_id), + ) + return False + now = self._now() + expires = now + self.config.retention_seconds + connection.execute("DELETE FROM slot_items WHERE slot_number = ?", (slot,)) + connection.executemany( + """ + INSERT INTO slot_items + (slot_number, position, generation, title, relative_path) + VALUES (?, ?, ?, ?, ?) + """, + ( + (slot, position, generation, item_title, relative_path) + for position, (item_title, relative_path) in enumerate(items, start=1) + ), + ) + connection.execute( + """ + UPDATE slots + SET media_state = 'ready', youtube_id = ?, title = ?, + item_count = ?, ready_at = ?, expires_at = ?, + last_error = NULL, updated_at = ? + WHERE slot_number = ? AND generation = ? + """, + (yt_id, title, len(items), now, expires, now, slot, generation), + ) + connection.execute( + """ + UPDATE download_jobs + SET status = 'succeeded', finished_at = ?, error = NULL + WHERE job_id = ? + """, + (now, job_id), + ) + return True + + def fail_download( + self, + job_id: str, + slot: int, + generation: int, + error: str, + ) -> bool: + with self._transaction() as connection: + now = self._now() + updated = connection.execute( + """ + UPDATE download_jobs + SET status = 'failed', finished_at = ?, error = ? + WHERE job_id = ? AND slot_number = ? AND generation = ? + AND status IN ('reserved','running') + """, + (now, error, job_id, slot, generation), + ).rowcount + if not updated: + return False + connection.execute( + """ + UPDATE slots + SET media_state = 'error', last_error = ?, updated_at = ? + WHERE slot_number = ? AND generation = ? + """, + (error, now, slot, generation), + ) + return True + + def get_job(self, job_id: str) -> DownloadJob | None: + with self._lock: + row = self.connection.execute( + "SELECT * FROM download_jobs WHERE job_id = ?", (job_id,) + ).fetchone() + if row is None: + return None + return DownloadJob( + job_id=row["job_id"], + slot=row["slot_number"], + generation=row["generation"], + url=row["url"], + status=row["status"], + submitter=row["submitter"], + note=row["note"], + requested_at=row["requested_at"], + started_at=row["started_at"], + finished_at=row["finished_at"], + error=row["error"], + ) + + def protect_slot( + self, + slot: int, + kind: str, + *, + guard_id: str | None = None, + review_after: float | None = None, + details: Mapping[str, Any] | None = None, + ) -> SlotGuard: + if slot not in SLOT_IDS: + raise ValueError("slot must be in 1..10") + guard_id = guard_id or uuid.uuid4().hex + with self._transaction() as connection: + existing = connection.execute( + """ + SELECT * FROM slot_guards + WHERE slot_number = ? AND state IN ('active','review_required') + """, + (slot,), + ).fetchone() + if existing is not None: + return self._row_to_guard(existing) + now = self._now() + connection.execute( + """ + INSERT INTO slot_guards + (guard_id, slot_number, kind, state, created_at, + review_after, details_json) + VALUES (?, ?, ?, 'active', ?, ?, ?) + """, + ( + guard_id, + slot, + kind, + now, + review_after, + json.dumps(details or {}, sort_keys=True), + ), + ) + row = connection.execute( + "SELECT * FROM slot_guards WHERE guard_id = ?", (guard_id,) + ).fetchone() + assert row is not None + return self._row_to_guard(row) + + @staticmethod + def _row_to_guard(row: sqlite3.Row) -> SlotGuard: + return SlotGuard( + guard_id=row["guard_id"], + slot=row["slot_number"], + kind=row["kind"], + state=row["state"], + created_at=row["created_at"], + review_after=row["review_after"], + cleared_at=row["cleared_at"], + cleared_by=row["cleared_by"], + clear_reason=row["clear_reason"], + ) + + def list_live_guards(self) -> list[SlotGuard]: + with self._lock: + rows = self.connection.execute( + """ + SELECT * FROM slot_guards + WHERE state IN ('active','review_required') + ORDER BY slot_number + """ + ).fetchall() + return [self._row_to_guard(row) for row in rows] + + def mark_due_guards_for_review(self, now: float | None = None) -> list[SlotGuard]: + now = self._now() if now is None else now + with self._transaction() as connection: + connection.execute( + """ + UPDATE slot_guards SET state = 'review_required' + WHERE state = 'active' AND review_after IS NOT NULL + AND review_after <= ? + """, + (now,), + ) + rows = connection.execute( + """ + SELECT * FROM slot_guards + WHERE state = 'review_required' ORDER BY slot_number + """ + ).fetchall() + return [self._row_to_guard(row) for row in rows] + + def clear_guard( + self, + slot: int, + guard_id: str, + *, + actor: str, + reason: str, + ) -> bool: + if not actor.strip() or not reason.strip(): + raise ValueError("administrative guard clear requires actor and reason") + with self._transaction() as connection: + now = self._now() + updated = connection.execute( + """ + UPDATE slot_guards + SET state = 'cleared', cleared_at = ?, cleared_by = ?, + clear_reason = ? + WHERE guard_id = ? AND slot_number = ? + AND state IN ('active','review_required') + """, + (now, actor.strip(), reason.strip(), guard_id, slot), + ).rowcount + if not updated: + return False + connection.execute( + """ + UPDATE player_observation + SET phase = 'unknown', active_slot = NULL, active_path = NULL, + paused = NULL, reconciliation_state = 'admin_cleared', + reconciliation_due_at = NULL + WHERE singleton = 1 + """ + ) + connection.execute( + """ + INSERT INTO audit_events + (occurred_at, event_type, slot_number, actor, details_json) + VALUES (?, 'playback_guard_admin_cleared', ?, ?, ?) + """, + (now, slot, actor.strip(), json.dumps({"reason": reason.strip()})), + ) + return True + + def begin_delete( + self, + slot: int, + *, + reason: str, + expected_url: str | None = None, + ) -> DeleteToken: + if slot not in SLOT_IDS: + raise ValueError("slot must be in 1..10") + with self._transaction() as connection: + row = connection.execute( + "SELECT * FROM slots WHERE slot_number = ?", (slot,) + ).fetchone() + assert row is not None + if row["media_state"] == "empty": + raise StaleOperationError(f"slot {slot} is already empty") + if expected_url is not None and row["url"] != expected_url: + raise StaleOperationError(f"slot {slot} URL no longer matches") + if row["media_state"] in {"queued", "downloading", "deleting"}: + raise SlotProtectedError(f"slot {slot} is {row['media_state']}") + if self._slot_has_live_guard(connection, slot): + raise SlotProtectedError(f"slot {slot} has active playback protection") + live_job = connection.execute( + """ + SELECT 1 FROM download_jobs + WHERE slot_number = ? AND status IN ('reserved','running') + """, + (slot,), + ).fetchone() + if live_job is not None: + raise SlotProtectedError(f"slot {slot} has an active download") + generation = row["generation"] + 1 + now = self._now() + connection.execute( + """ + UPDATE slots + SET generation = ?, media_state = 'deleting', updated_at = ? + WHERE slot_number = ? + """, + (generation, now, slot), + ) + return DeleteToken(slot=slot, generation=generation, reason=reason) + + def finish_delete(self, token: DeleteToken) -> bool: + with self._transaction() as connection: + now = self._now() + updated = connection.execute( + """ + UPDATE slots + SET media_state = 'empty', url = NULL, youtube_id = NULL, + title = '', item_count = 0, created_at = NULL, + ready_at = NULL, expires_at = NULL, last_error = NULL, + updated_at = ? + WHERE slot_number = ? AND generation = ? + AND media_state = 'deleting' + """, + (now, token.slot, token.generation), + ).rowcount + if not updated: + return False + connection.execute( + "DELETE FROM slot_items WHERE slot_number = ?", (token.slot,) + ) + connection.execute( + """ + INSERT INTO audit_events + (occurred_at, event_type, slot_number, actor, details_json) + VALUES (?, 'slot_deleted', ?, 'controller', ?) + """, + (now, token.slot, json.dumps({"reason": token.reason})), + ) + return True + + def begin_expired_deletions(self, now: float | None = None) -> list[DeleteToken]: + now = self._now() if now is None else now + tokens: list[DeleteToken] = [] + for slot in SLOT_IDS: + record = self.get_slot(slot) + if ( + record.media_state in {"ready", "error"} + and record.expires_at is not None + and record.expires_at <= now + ): + try: + tokens.append(self.begin_delete(slot, reason="expired")) + except SlotProtectedError: + continue + return tokens + + def player_observation(self) -> PlayerObservation: + with self._lock: + row = self.connection.execute( + "SELECT * FROM player_observation WHERE singleton = 1" + ).fetchone() + assert row is not None + return PlayerObservation( + availability=row["availability"], + phase=row["phase"], + active_slot=row["active_slot"], + active_path=row["active_path"], + paused=None if row["paused"] is None else bool(row["paused"]), + generation=row["generation"], + reconciliation_state=row["reconciliation_state"], + reconciliation_due_at=row["reconciliation_due_at"], + last_status_at=row["last_status_at"], + last_telemetry_at=row["last_telemetry_at"], + ) + + def set_player_observation( + self, + *, + availability: str, + phase: str, + active_slot: int | None, + active_path: str | None, + paused: bool | None, + reconciliation_state: str, + reconciliation_due_at: float | None, + status_payload: Mapping[str, Any] | None = None, + telemetry_payload: Mapping[str, Any] | None = None, + ) -> PlayerObservation: + now = self._now() + with self._transaction() as connection: + connection.execute( + """ + UPDATE player_observation + SET availability = ?, phase = ?, active_slot = ?, + active_path = ?, paused = ?, generation = generation + 1, + reconciliation_state = ?, reconciliation_due_at = ?, + last_status_at = CASE WHEN ? IS NULL THEN last_status_at ELSE ? END, + last_telemetry_at = CASE WHEN ? IS NULL THEN last_telemetry_at ELSE ? END, + last_status_json = COALESCE(?, last_status_json), + last_telemetry_json = COALESCE(?, last_telemetry_json) + WHERE singleton = 1 + """, + ( + availability, + phase, + active_slot, + active_path, + None if paused is None else int(paused), + reconciliation_state, + reconciliation_due_at, + None if status_payload is None else 1, + now, + None if telemetry_payload is None else 1, + now, + None if status_payload is None else json.dumps(status_payload, sort_keys=True), + None if telemetry_payload is None else json.dumps(telemetry_payload, sort_keys=True), + ), + ) + return self.player_observation() + + def mark_restart_uncertain(self) -> SlotGuard | None: + observation = self.player_observation() + if observation.active_slot is None: + return None + due = None + if self.config.uncertain_review_after_seconds > 0: + due = self._now() + self.config.uncertain_review_after_seconds + guard = self.protect_slot( + observation.active_slot, + "restart_uncertain", + review_after=due, + details={"previous_phase": observation.phase}, + ) + due = guard.review_after + self.set_player_observation( + availability="unknown", + phase="unknown", + active_slot=observation.active_slot, + active_path=observation.active_path, + paused=None, + reconciliation_state="awaiting_evidence", + reconciliation_due_at=due, + ) + return guard + + def set_published_item_count(self, slot: int, count: int) -> None: + with self._transaction() as connection: + connection.execute( + """ + UPDATE slots SET published_item_count = ?, updated_at = ? + WHERE slot_number = ? + """, + (count, self._now(), slot), + ) diff --git a/tests/test_store.py b/tests/test_store.py new file mode 100644 index 0000000..e898127 --- /dev/null +++ b/tests/test_store.py @@ -0,0 +1,315 @@ +import concurrent.futures +import hashlib +import os +import tempfile +import unittest +from pathlib import Path + +import yaml + +from controller.config import StorageConfig +from controller.models import MigrationError, SlotProtectedError +from controller.store import ControllerStore, LEGACY_MIGRATION_NAME + + +class MutableClock: + def __init__(self, value=1_700_000_000.0): + self.value = value + + def __call__(self): + return self.value + + +class StoreTestCase(unittest.TestCase): + def setUp(self): + self.tempdir = tempfile.TemporaryDirectory() + self.base = Path(self.tempdir.name) + self.media = self.base / "media" + self.data = self.base / "data" + self.clock = MutableClock() + self.config = StorageConfig( + media_base=self.media, + data_base=self.data, + retention_seconds=100, + uncertain_review_after_seconds=30, + ) + + def tearDown(self): + self.tempdir.cleanup() + + def open_store(self): + return ControllerStore(self.config, now=self.clock).open() + + +class FreshStoreTests(StoreTestCase): + def test_fresh_database_seeds_exact_stable_slots(self): + with self.open_store() as store: + slots = store.list_slots() + self.assertEqual([slot.slot for slot in slots], list(range(1, 11))) + self.assertTrue(all(slot.media_state == "empty" for slot in slots)) + self.assertEqual(store.player_observation().phase, "unknown") + + def test_slot_reservation_is_committed_before_worker_transition(self): + with self.open_store() as store: + reservation = store.reserve_download( + "https://www.youtube.com/playlist?list=one", + "Alice", + "Service", + job_id="job-one", + ) + self.assertEqual(reservation.status, "reserved") + self.assertEqual(reservation.slot, 1) + self.assertEqual(store.get_slot(1).media_state, "queued") + job = store.get_job("job-one") + self.assertIsNotNone(job) + self.assertEqual(job.status, "reserved") + self.assertTrue(store.mark_download_running("job-one", 1, 1)) + self.assertEqual(store.get_slot(1).media_state, "downloading") + + def test_concurrent_requests_get_only_one_remaining_slot(self): + with self.open_store() as store: + for index in range(1, 10): + result = store.reserve_download( + f"https://example.test/{index}", + None, + None, + job_id=f"seed-{index}", + ) + self.assertEqual(result.slot, index) + + def reserve(index): + return store.reserve_download( + f"https://example.test/concurrent-{index}", + None, + None, + job_id=f"concurrent-{index}", + ) + + with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool: + results = list(pool.map(reserve, range(8))) + winners = [result for result in results if result.status == "reserved"] + self.assertEqual(len(winners), 1) + self.assertEqual(winners[0].slot, 10) + self.assertEqual( + sum(result.status == "queue_full" for result in results), + 7, + ) + + def test_stale_worker_cannot_overwrite_reused_slot(self): + with self.open_store() as store: + first = store.reserve_download("https://example.test/one", None, None, job_id="old") + store.mark_download_running("old", 1, first.generation) + store.fail_download("old", 1, first.generation, "failed") + deletion = store.begin_delete(1, reason="test") + self.assertTrue(store.finish_delete(deletion)) + second = store.reserve_download("https://example.test/two", None, None, job_id="new") + self.assertEqual(second.slot, 1) + self.assertFalse( + store.complete_download( + "old", + 1, + first.generation, + title="stale", + items=(("Stale", "stale.mp4"),), + ) + ) + self.assertEqual(store.get_slot(1).url, "https://example.test/two") + + def test_deleting_slot_does_not_renumber_other_slots(self): + with self.open_store() as store: + first = store.reserve_download("https://example.test/one", None, None, job_id="one") + second = store.reserve_download("https://example.test/two", None, None, job_id="two") + store.fail_download("one", 1, first.generation, "failed") + store.fail_download("two", 2, second.generation, "failed") + token = store.begin_delete(1, reason="manual") + store.finish_delete(token) + self.assertEqual(store.get_slot(1).media_state, "empty") + self.assertEqual(store.get_slot(2).slot, 2) + self.assertEqual(store.get_slot(2).url, "https://example.test/two") + + +class GuardAndExpiryTests(StoreTestCase): + def _ready_slot(self, store, *, slot_expected=1): + reservation = store.reserve_download("https://example.test/ready", None, None, job_id="ready") + self.assertEqual(reservation.slot, slot_expected) + store.mark_download_running("ready", reservation.slot, reservation.generation) + store.complete_download( + "ready", + reservation.slot, + reservation.generation, + title="Ready", + items=(("One", "one.mp4"),), + ) + return reservation + + def test_active_and_review_guards_block_delete_and_expiry(self): + with self.open_store() as store: + reservation = self._ready_slot(store) + guard = store.protect_slot( + 1, + "restart_uncertain", + guard_id="review-me", + review_after=self.clock.value + 10, + ) + self.clock.value += 200 + due = store.mark_due_guards_for_review() + self.assertEqual(due[0].state, "review_required") + with self.assertRaises(SlotProtectedError): + store.begin_delete(1, reason="manual") + self.assertEqual(store.begin_expired_deletions(), []) + self.assertTrue( + store.clear_guard( + 1, + guard.guard_id, + actor="operator", + reason="Player host replaced", + ) + ) + self.assertEqual(store.player_observation().phase, "unknown") + expired = store.begin_expired_deletions() + self.assertEqual([token.slot for token in expired], [reservation.slot]) + + def test_admin_clear_requires_current_token_actor_and_reason(self): + with self.open_store() as store: + self._ready_slot(store) + guard = store.protect_slot(1, "player_active", guard_id="current") + with self.assertRaises(ValueError): + store.clear_guard(1, guard.guard_id, actor="", reason="because") + self.assertFalse( + store.clear_guard( + 1, + "stale-token", + actor="operator", + reason="confirmed", + ) + ) + self.assertEqual(store.list_live_guards()[0].guard_id, "current") + + def test_restart_marks_last_active_slot_uncertain_without_claiming_idle(self): + with self.open_store() as store: + self._ready_slot(store) + store.set_player_observation( + availability="online", + phase="playing", + active_slot=1, + active_path="/mnt/Media/YTPlayer/slot1/playlist.m3u", + paused=False, + reconciliation_state="confirmed", + reconciliation_due_at=None, + ) + guard = store.mark_restart_uncertain() + self.assertEqual(guard.slot, 1) + observation = store.player_observation() + self.assertEqual(observation.phase, "unknown") + self.assertEqual(observation.reconciliation_state, "awaiting_evidence") + self.assertEqual(observation.active_slot, 1) + + +class LegacyMigrationTests(StoreTestCase): + def write_legacy(self, data): + self.data.mkdir(parents=True, exist_ok=True) + raw = yaml.safe_dump(data, sort_keys=True).encode("utf-8") + self.config.legacy_yaml_path.write_bytes(raw) + return raw + + def test_migration_preserves_holes_timestamps_original_and_backup(self): + raw = self.write_legacy( + { + "slots": { + "2": { + "slot": 2, + "url": "https://example.test/two", + "yt_id": "abc", + "title": "Two", + "path": "/unsafe/ignored", + "m3u": "/unsafe/ignored/playlist.m3u", + "created_at": "2025-01-02T03:04:05Z", + "expires_at": "2025-01-04T03:04:05Z", + "item_count": 3, + "state": "playing", + }, + "7": { + "slot": 7, + "url": "https://example.test/seven", + "title": "Seven", + "path": "/unsafe/seven", + "m3u": "/unsafe/seven/playlist.m3u", + "created_at": "2025-02-01T00:00:00Z", + "expires_at": "2025-02-03T00:00:00Z", + "item_count": 1, + "state": "ready", + }, + } + } + ) + digest = hashlib.sha256(raw).hexdigest() + with self.open_store() as store: + self.assertTrue(store.migration_report.migrated) + self.assertEqual(store.migration_report.source_sha256, digest) + self.assertEqual(store.get_slot(1).media_state, "empty") + self.assertEqual(store.get_slot(2).title, "Two") + self.assertEqual(store.get_slot(7).slot, 7) + self.assertEqual(store.get_slot(2).created_at, 1_735_787_045.0) + guards = store.list_live_guards() + self.assertEqual([(guard.slot, guard.state) for guard in guards], [(2, "review_required")]) + marker = store.connection.execute( + "SELECT * FROM data_migrations WHERE name = ?", + (LEGACY_MIGRATION_NAME,), + ).fetchone() + self.assertEqual(marker["source_sha256"], digest) + backup = Path(marker["backup_path"]) + self.assertEqual(backup.read_bytes(), raw) + self.assertEqual(self.config.legacy_yaml_path.read_bytes(), raw) + + def test_completed_migration_is_idempotent_and_ignores_changed_yaml(self): + original = self.write_legacy( + { + "slots": { + "3": { + "slot": 3, + "url": "https://example.test/original", + "title": "Original", + "path": "/ignored", + "m3u": "/ignored/playlist.m3u", + "created_at": "2025-01-01T00:00:00Z", + "expires_at": "2025-01-03T00:00:00Z", + "item_count": 1, + "state": "ready", + } + } + } + ) + with self.open_store() as store: + generation = store.get_slot(3).generation + backup_path = Path(store.migration_report.backup_path) + self.config.legacy_yaml_path.write_text("not: [the, original]\n", encoding="utf-8") + with self.open_store() as store: + self.assertFalse(store.migration_report.migrated) + self.assertEqual(store.get_slot(3).title, "Original") + self.assertEqual(store.get_slot(3).generation, generation) + self.assertEqual(backup_path.read_bytes(), original) + + def test_failed_migration_keeps_original_and_verified_backup_recoverable(self): + self.data.mkdir(parents=True, exist_ok=True) + raw = b"slots:\n 1: [truncated\n" + self.config.legacy_yaml_path.write_bytes(raw) + store = ControllerStore(self.config, now=self.clock) + with self.assertRaises(MigrationError): + store.open() + self.assertFalse(self.config.database_path.exists()) + self.assertEqual(self.config.legacy_yaml_path.read_bytes(), raw) + backups = list(self.data.glob("playlists.yaml.pre-v2.*.bak")) + self.assertEqual(len(backups), 1) + self.assertEqual(backups[0].read_bytes(), raw) + + def test_normalized_duplicate_slot_keys_fail_all_or_nothing(self): + self.data.mkdir(parents=True, exist_ok=True) + raw = b"slots:\n 1:\n slot: 1\n '01':\n slot: 1\n" + self.config.legacy_yaml_path.write_bytes(raw) + with self.assertRaisesRegex(MigrationError, "normalize to 1"): + self.open_store() + self.assertFalse(self.config.database_path.exists()) + + +if __name__ == "__main__": + unittest.main()