"""Stage, validate and atomically publish read-only catalogue updates.""" import argparse from contextlib import closing from datetime import datetime, timezone import fcntl import json import os from pathlib import Path import sqlite3 import stat import subprocess import time import uuid class UpdateError(Exception): pass def now(): return datetime.now(timezone.utc).isoformat() def fsync_directory(path): descriptor = os.open(path, os.O_RDONLY | os.O_DIRECTORY) try: os.fsync(descriptor) finally: os.close(descriptor) def atomic_json(path, value): temp = path.with_name('.' + path.name + '.' + uuid.uuid4().hex) try: with temp.open('x', encoding='utf-8') as stream: os.chmod(temp, 0o600) json.dump(value, stream, ensure_ascii=False, indent=2) stream.write('\n') stream.flush() os.fsync(stream.fileno()) os.replace(temp, path) fsync_directory(path.parent) finally: temp.unlink(missing_ok=True) def check_candidate(source, stage): with closing(sqlite3.connect(source.resolve().as_uri() + '?mode=ro', uri=True)) as old: with closing(sqlite3.connect(stage.resolve().as_uri() + '?mode=ro', uri=True)) as candidate: if candidate.execute('PRAGMA quick_check').fetchone()[0] != 'ok': raise UpdateError('Candidate integrity check failed') if candidate.execute('PRAGMA foreign_key_check').fetchone() is not None: raise UpdateError('Candidate relation check failed') for table in ('titles', 'genres', 'countries', 'title_genres', 'title_countries', 'series_fetch_status', 'seasons', 'episodes', 'movie_sources', 'episode_sources'): before = old.execute('SELECT COUNT(*) FROM ' + table).fetchone()[0] after = candidate.execute('SELECT COUNT(*) FROM ' + table).fetchone()[0] if after < before: raise UpdateError('Candidate would remove existing catalogue entries') source_counts = ( 'SELECT title_id, COUNT(*) FROM movie_sources GROUP BY title_id', '''SELECT e.series_id, e.id, CASE WHEN e.id IS NULL THEN e.season_index END, CASE WHEN e.id IS NULL THEN e.episode_index END, COUNT(*) FROM episodes AS e JOIN episode_sources AS s USING (series_id, season_index, episode_index) GROUP BY 1, 2, 3, 4''', ) for query in source_counts: after = {row[:-1]: row[-1] for row in candidate.execute(query)} for row in old.execute(query): if after.get(row[:-1], 0) < row[-1]: raise UpdateError('Candidate would remove existing catalogue sources') row = candidate.execute("SELECT value_json FROM metadata WHERE key='built_at_utc'").fetchone() if row is None or not isinstance(json.loads(row[0]), str): raise UpdateError('Candidate timestamp missing') return json.loads(row[0]) def run_update(database, state_dir, upstream, *, synchronize_fn=None, restart_fn=None, health_fn=None, max_pages=2000, progress=None): database, state_dir = Path(database).resolve(), Path(state_dir).resolve() state_dir.mkdir(mode=0o700, parents=True, exist_ok=True) if synchronize_fn is None: from catalog_sync import synchronize synchronize_fn = synchronize restart_fn = restart_fn or restart_api health_fn = health_fn or wait_for_api with (state_dir / 'sync.lock').open('a') as lock: try: fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) except BlockingIOError: raise UpdateError('A catalogue update is already running') from None original = database.stat() identity = (original.st_ino, original.st_size, original.st_mtime_ns) stage = database.with_name('.catalog-sync-' + uuid.uuid4().hex + '.sqlite') backup = database.with_name('catalog.previous.sqlite') backup_temp = database.with_name('.catalog-backup-' + uuid.uuid4().hex) published = False started = now() atomic_json(state_dir / 'last-result.json', {'status': 'running', 'started_at_utc': started}) try: with closing(sqlite3.connect(database.as_uri() + '?mode=ro', uri=True)) as old: row = old.execute("SELECT value_json FROM metadata WHERE key='built_at_utc'").fetchone() if row is None or not isinstance(json.loads(row[0]), str): raise UpdateError('Previous catalogue timestamp missing') previous_stamp = json.loads(row[0]) report = synchronize_fn(database, stage, upstream, max_pages=max_pages, progress=progress) stamp = check_candidate(database, stage) current = database.stat() if identity != (current.st_ino, current.st_size, current.st_mtime_ns): raise UpdateError('Live catalogue changed during synchronisation') os.chown(stage, original.st_uid, original.st_gid) os.chmod(stage, stat.S_IMODE(original.st_mode)) with stage.open('rb') as stream: os.fsync(stream.fileno()) os.link(database, backup_temp) os.replace(backup_temp, backup) fsync_directory(database.parent) os.replace(stage, database) published = True fsync_directory(database.parent) restart_fn() health_fn(stamp) result = {**report, 'status': 'success', 'started_at_utc': started, 'completed_at_utc': now(), 'api_health_verified': True, 'media_files_downloaded': False} atomic_json(state_dir / 'last-success.json', result) atomic_json(state_dir / 'last-result.json', result) return result except Exception as error: rollback_failed = False if published: try: os.replace(backup, database) fsync_directory(database.parent) restart_fn() health_fn(previous_stamp) except Exception: rollback_failed = True atomic_json(state_dir / 'last-result.json', { 'status': 'failed', 'started_at_utc': started, 'completed_at_utc': now(), 'error_type': type(error).__name__, 'rollback_failed': rollback_failed, 'media_files_downloaded': False, }) if rollback_failed: raise UpdateError('Update failed and API restart needs operator attention') from None raise UpdateError('Update failed; the previous catalogue was retained') from None finally: for suffix in ('', '-journal', '-wal', '-shm'): Path(str(stage) + suffix).unlink(missing_ok=True) backup_temp.unlink(missing_ok=True) def restart_api(): subprocess.run(['/bin/systemctl', 'restart', 'reelbox-api.service'], check=True, capture_output=True, timeout=120) def wait_for_api(expected_stamp): from urllib.request import ProxyHandler, build_opener opener = build_opener(ProxyHandler({})) deadline = time.monotonic() + 90 while time.monotonic() < deadline: try: with opener.open('http://127.0.0.1:18765/api/health', timeout=5) as response: health = json.loads(response.read(65536)) if (health.get('status') == 'ok' and health.get('media_proxy') is False and health.get('snapshot_built_at_utc') == expected_stamp): return except Exception: pass time.sleep(2) raise UpdateError('Updated API did not become healthy in time') def emit_progress(event): fields = {key: event[key] for key in ('phase', 'total') if key in event} if 'completed' in event or 'count' in event: fields['completed'] = event.get('completed', event.get('count')) print(json.dumps({'progress': fields}), flush=True) def probe(upstream): genres = upstream.get('genre/all') movies = upstream.get('movie/by/filtres/0/created/0') series = upstream.get('serie/by/filtres/0/created/0') if not all(isinstance(rows, list) and rows for rows in (genres, movies, series)): raise UpdateError('Upstream metadata probe returned an unexpected shape') movie_id = movies[0]['id'] series_id = series[0]['id'] detail = upstream.get('movie/by/' + str(movie_id)) seasons = upstream.get('season/by/serie/' + str(series_id)) if not isinstance(detail, dict) or not isinstance(seasons, list): raise UpdateError('Upstream detail or season probe failed') return {'status': 'probe_ok', 'movie_id': movie_id, 'series_id': series_id, 'movie_sources': len(detail.get('sources', [])), 'seasons': len(seasons), 'media_files_downloaded': False} def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--database', type=Path, default=Path('/var/lib/reelbox-api/catalog.sqlite')) parser.add_argument('--state-dir', type=Path, default=Path('/var/lib/reelbox-sync')) parser.add_argument('--interval', type=float, default=0.25) parser.add_argument('--max-pages', type=int, default=2000) parser.add_argument('--probe', action='store_true') args = parser.parse_args() from catalog_sync import Upstream key = os.environ.get('REELBOX_UPSTREAM_KEY', '') if not key: print('REELBOX_UPSTREAM_KEY is required; no requests sent', flush=True) return 2 try: upstream = Upstream(key, interval=max(0.25, args.interval)) result = probe(upstream) if args.probe else run_update( args.database, args.state_dir, upstream, max_pages=args.max_pages, progress=emit_progress) print(json.dumps(result, ensure_ascii=False), flush=True) return 0 except Exception as error: print(json.dumps({'status': 'failed', 'error_type': type(error).__name__, 'message': str(error) if isinstance(error, UpdateError) else 'Metadata update failed; see safe status record'}), flush=True) return 1 if __name__ == '__main__': raise SystemExit(main())