diff --git a/docs/API.md b/docs/API.md index 41b83b8..cf114f4 100644 --- a/docs/API.md +++ b/docs/API.md @@ -11,7 +11,6 @@ All media paths are scoped to a **root**, identified by a stable `root_id`. | `GET` | `/api/config` | Returns the current root configuration. | | `GET` | `/api/roots` | List all active roots with status. | | `PUT` | `/api/roots` | Atomically replace the full root set. | -| `GET` | `/api/roots/{root_id}/index` | Returns the media index for one root. | | `GET` | `/api/roots/{root_id}/status` | Returns scanner and library status for one root. | | `POST` | `/api/roots/{root_id}/scan` | Triggers a new scan for one root. | | `POST` | `/api/roots/{root_id}/play` | Opens a media file with the system player. | diff --git a/docs/multi-index-plan.md b/docs/multi-index-plan.md index a7cc325..30a1b19 100644 --- a/docs/multi-index-plan.md +++ b/docs/multi-index-plan.md @@ -45,7 +45,6 @@ Every `Movie.id` and `Series.id` is namespaced with its `root_id`: |----------|-------------| | `GET /api/roots` | List all roots (name, path, root_id, status) | | `PUT /api/roots` | Atomically replace full root map `{name: path}` | -| `GET /api/roots/{root_id}/index` | Full index for one root | | `GET /api/roots/{root_id}/status` | Per-root scanning/loading/error state | | `POST /api/roots/{root_id}/scan` | Trigger scan for one root | | `WS /api/roots/{root_id}/ws` | Per-root WebSocket (init/upsert/remove/task) | diff --git a/frontend/src/App.vue b/frontend/src/App.vue index 2d9b637..a518b92 100644 --- a/frontend/src/App.vue +++ b/frontend/src/App.vue @@ -273,7 +273,7 @@ async function refreshRoots() { path: r.path, status: r.status, }) - if (r.status === "ready" || r.status === "scanning" || r.status === "loading") { + if (r.status === "ready" || r.status === "scanning") { activeIds.push(r.root_id) } } diff --git a/frontend/src/composables/useMediaWebSocket.ts b/frontend/src/composables/useMediaWebSocket.ts index c623482..0ba637a 100644 --- a/frontend/src/composables/useMediaWebSocket.ts +++ b/frontend/src/composables/useMediaWebSocket.ts @@ -16,6 +16,8 @@ interface RootState { movieMap: Map seriesMap: Map connected: boolean + initialized: boolean + pendingMessages: WsMessage[] reconnectTimer: ReturnType | null } @@ -252,30 +254,48 @@ export function useMediaWebSocket() { function updateMergedState() { mediaIndex.value = buildIndex() - // Loading is done when at least one root has connected and sent init - let anyConnected = false + // Consider a root "connected" only after init is received. + let anyInitialized = false for (const state of roots.value.values()) { - if (state.connected) { - anyConnected = true + if (state.connected && state.initialized) { + anyInitialized = true break } } - if (anyConnected) { + if (anyInitialized) { loading.value = false error.value = null } - connected.value = anyConnected + connected.value = anyInitialized } function processJson(state: RootState, text: string) { const msg = JSON.parse(text) as WsMessage + // Prevent out-of-order corruption: buffer delta messages until we receive + // the initial full-state payload. + if (msg.type !== "init" && !state.initialized) { + state.pendingMessages.push(msg) + return + } + switch (msg.type) { case "init": { state.movieMap.clear() state.seriesMap.clear() for (const m of msg.data.movies) state.movieMap.set(m.id, m) for (const s of msg.data.series) state.seriesMap.set(s.id, s) + state.initialized = true + + // Replay any deltas that arrived before init completed. + if (state.pendingMessages.length > 0) { + const queued = state.pendingMessages + state.pendingMessages = [] + for (const queuedMsg of queued) { + processJson(state, JSON.stringify(queuedMsg)) + } + } + updateMergedState() console.log( `[WS ${state.rootId}] init: ${state.movieMap.size} movies, ${state.seriesMap.size} series`, @@ -347,6 +367,8 @@ export function useMediaWebSocket() { movieMap: new Map(), seriesMap: new Map(), connected: false, + initialized: false, + pendingMessages: [], reconnectTimer: null, } roots.value.set(rootId, state) @@ -359,6 +381,8 @@ export function useMediaWebSocket() { ws.onopen = () => { state.connected = true + state.initialized = false + state.pendingMessages = [] updateMergedState() console.log(`[WS ${rootId}] Connected`) } @@ -367,6 +391,8 @@ export function useMediaWebSocket() { ws.onclose = (ev) => { state.connected = false + state.initialized = false + state.pendingMessages = [] state.ws = null updateMergedState() console.log(`[WS ${rootId}] Closed (code=${ev.code})`) diff --git a/mediahive/hivescan/__main__.py b/mediahive/hivescan/__main__.py index 2dd0d1b..beba38f 100644 --- a/mediahive/hivescan/__main__.py +++ b/mediahive/hivescan/__main__.py @@ -37,7 +37,6 @@ The server exposes per-root endpoints: WS /api/roots/{root_id}/ws Live index updates & task progress POST /api/roots/{root_id}/scan Trigger a new scan GET /api/roots/{root_id}/status Current root status - GET /api/roots/{root_id}/index Full index as JSON (HTTP fallback) """, ) parser.add_argument( diff --git a/mediahive/index_store.py b/mediahive/index_store.py index 80e0b47..afaafca 100644 --- a/mediahive/index_store.py +++ b/mediahive/index_store.py @@ -33,6 +33,8 @@ logger = logging.getLogger("mediahive.index_store") # Debounce interval for writing snapshots to disk (seconds) SNAPSHOT_DEBOUNCE = 5.0 +# Debounce interval for rebuilding the in-memory API snapshot (seconds) +SNAPSHOT_CACHE_DEBOUNCE = 0.25 class IndexStore: @@ -67,6 +69,17 @@ class IndexStore: self._snapshot_dirty = False self._snapshot_task: asyncio.Task | None = None + # In-memory API snapshot cache (served by get_full_index) + self._snapshot_cache_dirty = True + self._snapshot_cache_task: asyncio.Task | None = None + self._cached_snapshot = IndexSnapshot( + generated_at=datetime.now().isoformat(), + media_root=self.media_root, + stats=MediaStats(), + movies=[], + series=[], + ) + # ------------------------------------------------------------------ # Persistence # ------------------------------------------------------------------ @@ -82,90 +95,55 @@ class IndexStore: ap = AsyncPath(self.snapshot_path) if not await ap.exists(): logger.info("No snapshot found at %s, starting fresh", self.snapshot_path) + self._schedule_snapshot_cache_refresh() return try: raw = await ap.read_bytes() - await asyncio.to_thread(self._load_snapshot_sync, raw) + loaded_movies, loaded_series = await asyncio.to_thread( + self._load_snapshot_sync, + raw, + ) + self._merge_loaded_snapshot( + loaded_movies, + loaded_series, + ) + + self._schedule_snapshot_cache_refresh() except Exception: logger.exception("Failed to load snapshot from %s", self.snapshot_path) - def _load_snapshot_sync(self, raw: bytes) -> None: + def _load_snapshot_sync(self, raw: bytes) -> tuple[list[Movie], list[Series]]: """Parse snapshot bytes in a thread-pool context.""" data = msgspec.json.decode(raw, type=IndexSnapshot) + loaded_movies: list[Movie] = [] + loaded_series: list[Series] = [] for m in data.movies: - if m.showreel_source_sets: - filtered_source_sets = [] - for source_set in m.showreel_source_sets: - filtered_sources = [ - p - for p in source_set - if ( - Path(self.media_root, p).exists() - if self.media_root - else Path(p).exists() - ) - ] - if filtered_sources: - filtered_source_sets.append(filtered_sources) - m.showreel_source_sets = filtered_source_sets or None - m.showreel_images = ( - [source_set[0] for source_set in filtered_source_sets] - if filtered_source_sets - else None - ) - elif m.showreel_images: - filtered_images = [ - p - for p in m.showreel_images - if ( - Path(self.media_root, p).exists() - if self.media_root - else Path(p).exists() - ) - ] - m.showreel_images = filtered_images or None - m.showreel_source_sets = ( - [[p] for p in filtered_images] if filtered_images else None - ) m.id = self._maybe_migrate_id(m.id) m.root_id = self.root_id - self.movies[m.id] = m + loaded_movies.append(m) for s in data.series: - for season in s.seasons: - for ep in season.episodes: - if ep.reel_sources: - filtered_sources = [ - p - for p in ep.reel_sources - if ( - Path(self.media_root, p).exists() - if self.media_root - else Path(p).exists() - ) - ] - ep.reel_sources = filtered_sources or None - ep.reel_image = ( - filtered_sources[0] if filtered_sources else None - ) - elif ep.reel_image: - full = ( - Path(self.media_root, ep.reel_image) - if self.media_root - else Path(ep.reel_image) - ) - if full.exists(): - ep.reel_sources = [ep.reel_image] - else: - ep.reel_image = None - ep.reel_sources = None s.id = self._maybe_migrate_id(s.id) s.root_id = self.root_id - self.series[s.id] = s + loaded_series.append(s) logger.info( "Loaded snapshot: %d movies, %d series", - len(self.movies), - len(self.series), + len(loaded_movies), + len(loaded_series), ) + return loaded_movies, loaded_series + + def _merge_loaded_snapshot( + self, + movies: list[Movie], + series: list[Series], + ) -> None: + """Merge loaded snapshot items without overriding newer in-memory updates.""" + for movie in movies: + if movie.id not in self.movies: + self.movies[movie.id] = movie + for show in series: + if show.id not in self.series: + self.series[show.id] = show async def _write_snapshot(self) -> None: """Write current index to disk (called from debounce task).""" @@ -190,6 +168,35 @@ class IndexStore: self._snapshot_dirty = True if self._snapshot_task is None or self._snapshot_task.done(): self._snapshot_task = asyncio.create_task(self._snapshot_writer()) + self._schedule_snapshot_cache_refresh() + + def _schedule_snapshot_cache_refresh(self) -> None: + """Schedule a debounced rebuild of the in-memory API snapshot cache.""" + self._snapshot_cache_dirty = True + if self._snapshot_cache_task is None or self._snapshot_cache_task.done(): + self._snapshot_cache_task = asyncio.create_task( + self._snapshot_cache_writer() + ) + + async def _refresh_snapshot_cache_once(self) -> None: + """Rebuild cached snapshot once using copied store values.""" + movies = list(self.movies.values()) + series = list(self.series.values()) + self._cached_snapshot = await asyncio.to_thread( + self._build_snapshot_from_lists, + movies, + series, + ) + + async def _snapshot_cache_writer(self) -> None: + """Refresh the in-memory snapshot cache while mutations are pending.""" + while True: + await asyncio.sleep(SNAPSHOT_CACHE_DEBOUNCE) + if self._snapshot_cache_dirty: + self._snapshot_cache_dirty = False + await self._refresh_snapshot_cache_once() + else: + break async def _snapshot_writer(self) -> None: """Flush to disk every SNAPSHOT_DEBOUNCE seconds while dirty.""" @@ -203,6 +210,12 @@ class IndexStore: async def flush_snapshot(self) -> None: """Force-write a snapshot immediately (e.g. on shutdown).""" + if self._snapshot_cache_task and not self._snapshot_cache_task.done(): + self._snapshot_cache_task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await self._snapshot_cache_task + await self._refresh_snapshot_cache_once() + if self._snapshot_task and not self._snapshot_task.done(): self._snapshot_task.cancel() with contextlib.suppress(asyncio.CancelledError): @@ -310,8 +323,11 @@ class IndexStore: series: list[Series], ) -> IndexSnapshot: """Build a sorted IndexSnapshot with computed stats from list copies.""" - movies_list = sorted(movies, key=lambda x: (x.title.lower(), x.year or 0)) - series_list = sorted(series, key=lambda x: x.title.lower()) + movies_list = sorted( + movies, + key=lambda x: ((x.title or "").lower(), x.year or 0), + ) + series_list = sorted(series, key=lambda x: (x.title or "").lower()) total_movie_versions = sum(len(m.torrents) for m in movies_list) total_series_episodes = sum( @@ -339,5 +355,5 @@ class IndexStore: ) def get_full_index(self) -> IndexSnapshot: - """Return the full index as an IndexSnapshot.""" - return self._build_snapshot() + """Return the latest in-memory IndexSnapshot cache.""" + return self._cached_snapshot diff --git a/mediahive/root_registry.py b/mediahive/root_registry.py index dc232f1..55f4cfa 100644 --- a/mediahive/root_registry.py +++ b/mediahive/root_registry.py @@ -105,9 +105,21 @@ class RootContext: # Event queue and consumer self._events: asyncio.Queue[ScanEvent] = asyncio.Queue() self._consumer_task: asyncio.Task | None = None + self._startup_task: asyncio.Task | None = None async def start(self) -> None: - """Load snapshot and start event consumer.""" + """Start consumer immediately and load snapshot in the background.""" + self.status = "loading" + self.error = None + + if self._consumer_task is None or self._consumer_task.done(): + self._consumer_task = asyncio.create_task(self._consume_events()) + + if self._startup_task is None or self._startup_task.done(): + self._startup_task = asyncio.create_task(self._load_snapshot_background()) + + async def _load_snapshot_background(self) -> None: + """Load snapshot without blocking root activation paths.""" try: await self.store.load_snapshot() self.status = "ready" @@ -117,15 +129,20 @@ class RootContext: len(self.store.movies), len(self.store.series), ) + except asyncio.CancelledError: + raise except Exception as exc: self.status = "error" self.error = str(exc) logger.exception("Root %s failed to load snapshot", self.root_id) - self._consumer_task = asyncio.create_task(self._consume_events()) - async def stop(self) -> None: """Stop consumer, flush snapshot, stop scanner.""" + if self._startup_task and not self._startup_task.done(): + self._startup_task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await self._startup_task + if self.scanner is not None: try: await self.scanner.stop() diff --git a/mediahive/server.py b/mediahive/server.py index a262e4e..8c9f0c6 100644 --- a/mediahive/server.py +++ b/mediahive/server.py @@ -34,7 +34,6 @@ from mediahive.hivescan.images import close_image_client from mediahive.hivescan.scanner import RootScanner from mediahive.hivescan.tmdb_client import close_http_client from mediahive.models.protocol import ( - MsgspecResponse, OpenFolderRequest, PlayMediaRequest, RootsRequest, @@ -285,7 +284,7 @@ async def _attach_scanners() -> None: """Ensure every active root context has a running scanner.""" async with _attach_scanners_lock: for ctx in supervisor.all_contexts().values(): - if ctx.scanner is None and ctx.status in {"ready", "loading"}: + if ctx.scanner is None and ctx.status == "ready": try: scanner = RootScanner(ctx.root_id, ctx.root_path, ctx.send_event) await scanner.start() @@ -296,6 +295,13 @@ async def _attach_scanners() -> None: ) +async def _attach_scanners_loop() -> None: + """Periodically attach scanners as roots transition to ready.""" + while True: + await _attach_scanners() + await asyncio.sleep(1) + + async def _activate_all_roots() -> None: """Background task: validate and activate all configured roots. @@ -359,6 +365,7 @@ async def lifespan(_app: FastAPI): # Defer root activation to a background task so the server starts # immediately and macOS permission dialogs do not block startup. activation_task = asyncio.create_task(_activate_all_roots()) + scanner_attach_task = asyncio.create_task(_attach_scanners_loop()) logger.info("Server ready; waiting for root activation") @@ -369,6 +376,10 @@ async def lifespan(_app: FastAPI): with suppress(asyncio.CancelledError): await activation_task + scanner_attach_task.cancel() + with suppress(asyncio.CancelledError): + await scanner_attach_task + await supervisor.shutdown() with suppress(Exception): @@ -434,13 +445,6 @@ async def put_roots(request: Request): } -@app.get("/api/roots/{root_id}/index") -async def get_root_index(root_id: str): - """Return the full index for a single root.""" - ctx = _get_context(root_id) - return MsgspecResponse(ctx.store.get_full_index()) - - @app.get("/api/roots/{root_id}/status") async def get_root_status(root_id: str): """Return status for a single root."""