diff --git a/docs/API.md b/docs/API.md index c26d6a1..0a65870 100644 --- a/docs/API.md +++ b/docs/API.md @@ -10,20 +10,20 @@ All media paths are scoped to a **root**, identified by a friendly `root_id` | --- | --- | --- | | `GET` | `/api/health` | Lightweight health check. | | `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. | +| `PUT` | `/api/config/roots` | Atomically replace the full root set. | | `POST` | `/api/play/{root_id}` | Opens a media file with the system player. | | `POST` | `/api/open-folder/{root_id}` | Opens a folder in the system file explorer. | | `GET` | `/api/meta/{root_id}/{meta_key}` | Returns allowed metadata from `/.mediahive`. | +| `GET` | `/api/meta/playback-state` | Returns merged resume positions across all roots. | | `GET` | `/api/player/status` | Returns whether remote player control is currently available. | | `GET` | `/api/mpcbe/status` | Reports whether MPC-BE's local web interface is reachable. | | `GET` | `/api/media/{root_id}/{file_path:path}` | Serves files from the specified root. | | `GET` | `/api/assets/{root_id}/{asset_type}/{asset_path:path}` | Serves typed assets from `/.mediahive`. | -| `WS` | `/api/ws/{root_id}` | Streams live index updates and task progress for one root. | +| `WS` | `/api/ws` | Streams roots, index updates, and task progress for all roots. | ## Notes -- `PUT /api/roots` accepts `{ "roots": { "name": "/absolute/path", ... } }`, validates paths, and atomically swaps the active set. +- `PUT /api/config/roots` accepts `{ "roots": { "name": "/absolute/path", ... } }`, validates paths, and atomically swaps the active set. - `POST /api/play/{root_id}` and `POST /api/open-folder/{root_id}` expect JSON request bodies with `file_path` / `folder_path` relative to the root. - `GET /api/media/{root_id}/{file_path:path}` is constrained to the specified root; path traversal outside the root is rejected. - `GET /api/assets/{root_id}/{asset_type}/{asset_path:path}` is constrained to `/.mediahive/{asset_type}` where `asset_type` is one of `movies`, `series`, `people`. diff --git a/docs/multi-index-plan.md b/docs/multi-index-plan.md deleted file mode 100644 index 301de48..0000000 --- a/docs/multi-index-plan.md +++ /dev/null @@ -1,94 +0,0 @@ -# Multi-Root Implementation Notes - -## Overview - -MediaHive now supports multiple independent media roots. Each root is a filesystem directory with its own index, scanner, and WebSocket stream. The frontend merges per-root state into a single reactive view. - -## Architecture - -### Root Identity - -- **Root ID**: friendly root name derived from configured path basename. -- **Name/ID collision handling**: suffixes `2`, `3`, … are appended to keep each root ID unique. -- **Path normalization**: lower-case Windows drive letter, strip trailing slashes, forward slashes only (`as_posix()`). - -### Per-Root Runtime (`RootContext`) - -Each active root gets an isolated `RootContext` managed by the `Supervisor`: - -- `root_id`, `root_path` — stable identifiers -- `IndexStore` — owns snapshot at `/.mediahive/index.json` -- `RootScanner` — per-root scanning instance (replaced legacy global scanner) -- `asyncio.Queue` + consumer task — bridges scanner events to WebSocket -- `status`: `idle` | `loading` | `ready` | `scanning` | `error` - -### Supervisor - -- Holds `dict[str, RootContext]` keyed by `root_id`. -- `replace_roots(new_roots)` atomically swaps the active set: - 1. Validate & canonicalize paths. - 2. Derive unique friendly `root_id` for each. - 3. Prepare new `RootContext`s (load snapshots). - 4. Swap dict atomically. - 5. Stop removed contexts in background with bounded timeout. -- Exposes merged read helpers (`merged_index`, `all_statuses`). - -### Item IDs - -`root_id` is stored separately on each item. - -- `Movie.id` uses a slug built from the movie title and year, for example `spider-man-no-way-home-2021`. -- `Series.id` uses a slug built from the series title, for example `lost`. -- Legacy snapshot migrations are handled by `scripts/indexmigr.py`, not during app startup. - -## API - -| Endpoint | Description | -|----------|-------------| -| `GET /api/roots` | List all roots (name, path, root_id, status) | -| `PUT /api/roots` | Atomically replace full root map `{name: path}` | -| `WS /api/ws/{root_id}` | Per-root WebSocket (init/upsert/remove/task + status/task events) | -| `GET /api/media/{root_id}/{path:path}` | Serve media file scoped to root | -| `GET /api/assets/{root_id}/{asset_type}/{asset_path:path}` | Serve `.mediahive/{asset_type}` assets (`movies`, `series`, `people`) | -| `POST /api/play/{root_id}` | Play file within root | -| `POST /api/open-folder/{root_id}` | Open folder within root | -| `GET /api/meta/{root_id}/{meta_key}` | Per-root metadata (for example `playback-state`) | -| `POST /api/ui/pick-folder` | Native OS folder picker (returns path) | - -> **Removed legacy endpoints**: `/api/change-folder`, `/api/index`, `/api/scan`, `/api/status`, `/api/playback/resume-positions`. No backwards compatibility is maintained. - -## macOS Startup Safety - -The server **must not** touch the filesystem during startup, because macOS may show permission dialogs that block the event loop and prevent the HTTP server from accepting requests. - -- `lifespan()` creates a background task (`_activate_all_roots()`) and immediately yields. -- All filesystem validation (`exists()`, `is_dir()`, `resolve()`) runs in a thread pool via `asyncio.to_thread()`. -- CLI entry points (`__main__.py`, `winmain.py`, `hivescan/__main__.py`) pass raw paths via the `MEDIAHIVE_ROOTS` environment variable; they do **not** validate paths before starting the server. - -## POSIX Path Enforcement - -All stored and transmitted paths use forward slashes exclusively: - -- `_normalize_path()` always returns POSIX paths. -- Config stores `p.as_posix()`. -- URLs use `/` separators. -- `Path(root_path) / relative_path` works correctly on Windows because `Path` accepts POSIX separators. - -## Config Migration - -- Old `media_folder` string is auto-migrated to `roots: {basename: path}` on load. -- `roots` is persisted back to TOML config. - -## Scanner - -- Legacy global module-level scanner API was removed from `hivescan/scanner.py`. -- `RootScanner` is the only scanning interface. -- Each `RootScanner` owns its own `showreel_queue`, `scan_task`, `rescan_worker_task`, and `_seen_mtimes`. - -## Frontend - -- `useMediaWebSocket.ts` manages one WebSocket per active root. -- `App.vue` merges per-root `movieMap`/`seriesMap` into a single `mediaIndex`. -- `Header.vue` provides add/remove root UI via `PUT /api/roots`. -- Playback URLs are root-qualified (`/api/media/{root_id}/...`). -- Metadata cache assets use typed root paths (`/api/assets/{root_id}/{asset_type}/...`) rather than exposing `.mediahive` in URLs. diff --git a/frontend/src/App.vue b/frontend/src/App.vue index 4351ffb..464f6e6 100644 --- a/frontend/src/App.vue +++ b/frontend/src/App.vue @@ -37,6 +37,7 @@
(() => Array.from(tasks.value.values())) -// Poll for active roots and connect WS to them -const rootStatuses = ref< - Map ->(new Map()) - function getRootName(rootId: string | null | undefined): string | null { if (!rootId) return null - return rootStatuses.value.get(rootId)?.name || null + return rootStatuses.value.get(rootId)?.root_id || null } function normalizePosixPath(value: string): string { @@ -427,10 +422,14 @@ const hasLibraryItems = computed(() => { return mediaIndex.value.movies.length > 0 || mediaIndex.value.series.length > 0 }) const hasAnySnapshotLoaded = computed(() => - Array.from(rootStatuses.value.values()).some((root) => root.snapshotLoaded), + Array.from(rootStatuses.value.values()).some((root) => root.snapshot_loaded), ) const isInitialScanMode = computed(() => !hasLibraryItems.value && !hasAnySnapshotLoaded.value) +const headerRoots = computed(() => + Array.from(rootStatuses.value.values()).sort((a, b) => a.root_id.localeCompare(b.root_id)), +) + const showProgressPanel = computed(() => { if (isInitialScanMode.value) { return !wsConnected.value || progressRoots.value.length > 0 @@ -480,51 +479,7 @@ watch( { deep: false }, ) -async function refreshRoots() { - try { - const roots = await fetchRoots() - const newMap = new Map< - string, - { name: string; path: string; status: string; snapshotLoaded: boolean } - >() - const activeIds: string[] = [] - for (const r of roots) { - newMap.set(r.root_id, { - name: r.root_id, - path: r.path, - status: r.status, - snapshotLoaded: Boolean(r.snapshot_loaded), - }) - if (r.status === "ready" || r.status === "scanning") { - activeIds.push(r.root_id) - } - } - rootStatuses.value = newMap - setActiveRoots(activeIds) - } catch (e) { - console.error("Failed to fetch roots:", e) - } -} - -let rootsPollTimer: number | null = null -function startRootsPolling() { - if (rootsPollTimer !== null) return - void refreshRoots() - rootsPollTimer = window.setInterval(refreshRoots, 5000) -} -function stopRootsPolling() { - if (rootsPollTimer !== null) { - window.clearInterval(rootsPollTimer) - rootsPollTimer = null - } -} - -onMounted(() => { - startRootsPolling() -}) - onUnmounted(() => { - stopRootsPolling() if (libraryUpdateToastTimer !== null) { window.clearTimeout(libraryUpdateToastTimer) libraryUpdateToastTimer = null diff --git a/frontend/src/api.ts b/frontend/src/api.ts index 213e808..b24a57e 100644 --- a/frontend/src/api.ts +++ b/frontend/src/api.ts @@ -9,18 +9,9 @@ export interface PlayerInfo { path: string | null } -export interface RootStatus { +export interface RootEntry { root_id: string path: string - status: string - error: string | null - snapshot_loaded: boolean - movies: number - series: number -} - -export interface RootsResponse { - roots: RootStatus[] } export function normalizeMediaPath(input: string): string { @@ -123,37 +114,19 @@ function splitAssetTypePath(assetPath: string): { assetType: string; relativePat return { assetType: assetType.toLowerCase(), relativePath: rest.join("/") } } -/** - * Fetch active roots and their statuses - */ -export async function fetchRoots(): Promise { - const response = await fetch("/api/roots") - if (!response.ok) { - throw new Error(`Failed to load roots: ${response.statusText}`) - } - const data = await response.json() - return data.roots || [] -} - /** * Fetch merged resume positions from all roots. */ export async function fetchResumePositions(): Promise> { try { - const roots = await fetchRoots() - const merged: Record = {} - await Promise.all( - roots.map(async (root) => { - const response = await fetch(`/api/meta/${encodeURIComponent(root.root_id)}/playback-state`) - if (!response.ok) return - const data = await response.json().catch(() => ({})) - const positions = data?.data?.resume_positions - if (positions && typeof positions === "object") { - Object.assign(merged, positions) - } - }), - ) - return merged + const response = await fetch("/api/meta/playback-state") + if (!response.ok) return {} + const data = await response.json().catch(() => ({})) + const positions = data?.data?.resume_positions + if (!positions || typeof positions !== "object") { + return {} + } + return positions as Record } catch { return {} } @@ -164,8 +137,8 @@ export async function fetchResumePositions(): Promise> { */ export async function replaceRoots( roots: Record, -): Promise<{ accepted: RootStatus[]; failed: unknown[] }> { - const response = await fetch("/api/roots", { +): Promise<{ accepted: RootEntry[]; failed: unknown[] }> { + const response = await fetch("/api/config/roots", { method: "PUT", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ roots }), diff --git a/frontend/src/components/Header.vue b/frontend/src/components/Header.vue index 1dd3939..44e434e 100644 --- a/frontend/src/components/Header.vue +++ b/frontend/src/components/Header.vue @@ -306,7 +306,7 @@ import { ref, watch, computed, onMounted, onUnmounted } from "vue" import { useRouter, useRoute } from "vue-router" import { navAttrs } from "../composables/useKeyboardNavigation" import logoUrl from "../assets/mediahive.webp" -import { fetchRoots, replaceRoots, pickFolderAndAddRoot, fetchPlayers } from "../api" +import { replaceRoots, pickFolderAndAddRoot, fetchPlayers } from "../api" import type { PlayerInfo } from "../api" import HexKeyboard from "./HexKeyboard.vue" import { @@ -334,6 +334,7 @@ interface RootEntry { const props = defineProps<{ currentView: "movies" | "series" | "search" searchQuery: string + roots: RootEntry[] mpcBeConnected: boolean navRow: number position: "top" | "after-hero" | "after-movie-header" | "after-series-hero" @@ -357,7 +358,7 @@ window.addEventListener("pywebviewready", _onPywebviewReady, { once: true }) onUnmounted(() => window.removeEventListener("pywebviewready", _onPywebviewReady)) const showSettings = computed(() => route.path === "/settings") -const roots = ref([]) +const roots = computed(() => props.roots) function openSettings() { if (showSettings.value) return @@ -408,25 +409,11 @@ async function refreshPlayers() { } } -async function refreshRoots() { - try { - const data = await fetchRoots() - roots.value = data.map((r) => ({ - root_id: r.root_id, - path: r.path, - status: r.status, - })) - } catch (e) { - console.error("Failed to fetch roots:", e) - } -} - async function removeRoot(rootId: string) { const filtered = roots.value.filter((r) => r.root_id !== rootId) const newRoots = Object.fromEntries(filtered.map((r) => [r.root_id, r.path])) try { await replaceRoots(newRoots) - await refreshRoots() } catch (e) { console.error("Failed to remove root:", e) alert("Failed to remove root") @@ -441,7 +428,6 @@ async function addRoot() { newRoots[suggestedId] = folder try { await replaceRoots(newRoots) - await refreshRoots() closeSettings() } catch (e) { console.error("Failed to add root:", e) @@ -451,7 +437,6 @@ async function addRoot() { watch(showSettings, (visible) => { if (visible) { - void refreshRoots() void refreshPlayers() } }) diff --git a/frontend/src/composables/useMediaWebSocket.ts b/frontend/src/composables/useMediaWebSocket.ts index e812542..413b3ee 100644 --- a/frontend/src/composables/useMediaWebSocket.ts +++ b/frontend/src/composables/useMediaWebSocket.ts @@ -13,31 +13,24 @@ import type { MediaIndex, TaskInfo, WsMessage, + WsRootStatus, } from "../types" interface RootState { - rootId: string - ws: WebSocket | null movieMap: Map seriesMap: Map peopleMap: Map - connected: boolean initialized: boolean pendingMessages: WsMessage[] - reconnectTimer: ReturnType | null } +export interface RootStatusEntry extends WsRootStatus {} + const MERGED_KEY_DELIMITER = "::" /** - * Composable that connects to per-root MediaHive WebSockets and keeps + * Composable that connects to one all-roots MediaHive WebSocket and keeps * a merged media index updated in real time. - * - * The server sends per-root: - * - "init" → full index (movies + series) on connect - * - "upsert" → single item inserted or updated - * - "remove" → single item removed - * - "task" → background task progress */ export function useMediaWebSocket() { type RootTaskInfo = TaskInfo & { root_id: string } @@ -47,8 +40,11 @@ export function useMediaWebSocket() { const error = shallowRef(null) const connected = shallowRef(false) const tasks = shallowRef>(new Map()) + const roots = shallowRef>(new Map()) - const roots = shallowRef>(new Map()) + const rootStates = shallowRef>(new Map()) + const wsRef = shallowRef(null) + let reconnectTimer: ReturnType | null = null let disposed = false // Single periodic sweep for completed tasks instead of one timeout per task @@ -373,10 +369,41 @@ export function useMediaWebSocket() { return merged } + function ensureRootState(rootId: string): RootState { + const existing = rootStates.value.get(rootId) + if (existing) { + return existing + } + const created: RootState = { + movieMap: new Map(), + seriesMap: new Map(), + peopleMap: new Map(), + initialized: false, + pendingMessages: [], + } + rootStates.value.set(rootId, created) + return created + } + + function pruneMissingRoots(nextRoots: Map) { + for (const rootId of rootStates.value.keys()) { + if (!nextRoots.has(rootId)) { + rootStates.value.delete(rootId) + } + } + + for (const [taskKey, task] of tasks.value.entries()) { + if (!nextRoots.has(task.root_id)) { + tasks.value.delete(taskKey) + } + } + tasks.value = new Map(tasks.value) + } + function buildIndex(): MediaIndex { const movies: MovieUi[] = [] const series: SeriesUi[] = [] - for (const state of roots.value.values()) { + for (const state of rootStates.value.values()) { movies.push(...state.movieMap.values()) series.push(...state.seriesMap.values()) } @@ -392,68 +419,84 @@ export function useMediaWebSocket() { function updateMergedState() { mediaIndex.value = buildIndex() - // Consider a root "connected" only after init is received. + let anyInitialized = false - for (const state of roots.value.values()) { - if (state.connected && state.initialized) { + for (const state of rootStates.value.values()) { + if (state.initialized) { anyInitialized = true break } } - if (anyInitialized) { + + if (anyInitialized || roots.value.size === 0) { loading.value = false error.value = null } - connected.value = anyInitialized + + connected.value = wsRef.value?.readyState === WebSocket.OPEN } - function processJson(state: RootState, text: string) { - const msg = JSON.parse(text) as WsMessage + function applyRootInit(rootId: string, rootData: { movies: Record; series: Record; people?: Record }) { + const state = ensureRootState(rootId) - // 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 + state.peopleMap.clear() + for (const [id, person] of Object.entries(rootData.people || {})) { + const parsed = Number(id) + const normalized = normalizePerson(person) + if (Number.isFinite(parsed)) { + state.peopleMap.set(parsed, normalized || { name: "", profile_path: null, gender: null }) + } } - switch (msg.type) { - case "init": { - state.peopleMap.clear() - for (const [id, person] of Object.entries(msg.data.people || {})) { - const parsed = Number(id) - const normalized = normalizePerson(person) - if (Number.isFinite(parsed)) { - state.peopleMap.set(parsed, normalized || { name: "", profile_path: null, gender: null }) - } - } + state.movieMap.clear() + state.seriesMap.clear() + for (const [id, m] of Object.entries(rootData.movies || {})) { + state.movieMap.set(id, withMovieIdentity(id, m, rootId, state.peopleMap)) + } + for (const [id, s] of Object.entries(rootData.series || {})) { + state.seriesMap.set(id, withSeriesIdentity(id, s, rootId, state.peopleMap)) + } - state.movieMap.clear() - state.seriesMap.clear() - for (const [id, m] of Object.entries(msg.data.movies || {})) { - state.movieMap.set(id, withMovieIdentity(id, m, state.rootId, state.peopleMap)) - } - for (const [id, s] of Object.entries(msg.data.series || {})) { - state.seriesMap.set(id, withSeriesIdentity(id, s, state.rootId, state.peopleMap)) - } - state.initialized = true + 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`, - ) - break + if (state.pendingMessages.length > 0) { + const queued = state.pendingMessages + state.pendingMessages = [] + for (const queuedMsg of queued) { + processMessage(queuedMsg) } + } + } + + function processMessage(msg: WsMessage) { + switch (msg.type) { + case "roots": { + const next = new Map() + for (const root of msg.roots || []) { + next.set(root.root_id, { ...root }) + ensureRootState(root.root_id) + } + roots.value = next + pruneMissingRoots(next) + updateMergedState() + return + } + + case "init": { + for (const [rootId, rootData] of Object.entries(msg.roots || {})) { + applyRootInit(rootId, rootData) + } + updateMergedState() + return + } + case "upsert": { + const state = ensureRootState(msg.root_id) + if (!state.initialized) { + state.pendingMessages.push(msg) + return + } + if (msg.people) { for (const [id, person] of Object.entries(msg.people)) { const parsed = Number(id) @@ -467,160 +510,109 @@ export function useMediaWebSocket() { if (msg.kind === "movie") { state.movieMap.set( msg.id, - withMovieIdentity(msg.id, msg.item as Movie, state.rootId, state.peopleMap), + withMovieIdentity(msg.id, msg.item as Movie, msg.root_id, state.peopleMap), ) } else { state.seriesMap.set( msg.id, - withSeriesIdentity(msg.id, msg.item as Series, state.rootId, state.peopleMap), + withSeriesIdentity(msg.id, msg.item as Series, msg.root_id, state.peopleMap), ) } updateMergedState() - break + return } + case "remove": { + const state = ensureRootState(msg.root_id) + if (!state.initialized) { + state.pendingMessages.push(msg) + return + } + if (msg.kind === "movie") { state.movieMap.delete(msg.id) } else { state.seriesMap.delete(msg.id) } updateMergedState() - break + return } + case "task": { const info = msg.data - const taskKey = `${state.rootId}:${info.id}` - tasks.value.set(taskKey, { ...info, root_id: state.rootId }) + const taskKey = `${msg.root_id}:${info.id}` + tasks.value.set(taskKey, { ...info, root_id: msg.root_id }) tasks.value = new Map(tasks.value) if (info.status === "completed" || info.status === "cancelled" || info.status === "error") { completedTaskIds.add(taskKey) startTaskSweep() } - break - } - } - } - - function handleMessage(state: RootState, event: MessageEvent) { - try { - let text: string - if (event.data instanceof Blob) { - event.data.text().then((t) => processJson(state, t)) return - } else if (event.data instanceof ArrayBuffer) { - text = new TextDecoder().decode(event.data) - } else { - text = event.data as string } - processJson(state, text) - } catch (e) { - console.error(`[WS ${state.rootId}] Failed to handle message:`, e) } } - function connectRoot(rootId: string) { - if (disposed) return - const existing = roots.value.get(rootId) - if (existing?.ws) { - // Already connecting or connected + function handleRawMessage(event: MessageEvent) { + const processText = (text: string) => { + try { + processMessage(JSON.parse(text) as WsMessage) + } catch (e) { + console.error("[WS] Failed to handle message:", e) + } + } + + if (event.data instanceof Blob) { + void event.data.text().then(processText) return } + if (event.data instanceof ArrayBuffer) { + processText(new TextDecoder().decode(event.data)) + return + } + processText(event.data as string) + } + + function scheduleReconnect() { + if (disposed) return + if (reconnectTimer) clearTimeout(reconnectTimer) + reconnectTimer = setTimeout(() => { + reconnectTimer = null + connect() + }, 2000) + } + + function connect() { + if (disposed) return + if (wsRef.value && wsRef.value.readyState <= WebSocket.OPEN) return const proto = location.protocol === "https:" ? "wss:" : "ws:" - const url = `${proto}//${location.host}/api/ws/${encodeURIComponent(rootId)}` + const url = `${proto}//${location.host}/api/ws` - const state: RootState = { - rootId, - ws: null, - movieMap: new Map(), - seriesMap: new Map(), - peopleMap: new Map(), - connected: false, - initialized: false, - pendingMessages: [], - reconnectTimer: null, + console.log(`[WS] Connecting to ${url}...`) + const ws = new WebSocket(url) + wsRef.value = ws + + ws.onopen = () => { + connected.value = true + error.value = null + console.log("[WS] Connected") } - roots.value.set(rootId, state) - function doConnect() { - if (disposed) return - console.log(`[WS ${rootId}] Connecting to ${url}...`) - const ws = new WebSocket(url) - state.ws = ws + ws.onmessage = (ev) => handleRawMessage(ev) - ws.onopen = () => { - state.connected = true - state.initialized = false - state.pendingMessages = [] - updateMergedState() - console.log(`[WS ${rootId}] Connected`) - } - - ws.onmessage = (ev) => handleMessage(state, ev) - - ws.onclose = (ev) => { - state.connected = false - state.initialized = false - state.pendingMessages = [] - state.ws = null - updateMergedState() - console.log(`[WS ${rootId}] Closed (code=${ev.code})`) - scheduleReconnect() - } - - ws.onerror = (ev) => { - console.error(`[WS ${rootId}] Error:`, ev) - if (!mediaIndex.value) { - error.value = "WebSocket connection failed" - } + ws.onclose = (ev) => { + if (wsRef.value === ws) { + wsRef.value = null } + connected.value = false + console.log(`[WS] Closed (code=${ev.code})`) + scheduleReconnect() } - function scheduleReconnect() { - if (disposed) return - if (state.reconnectTimer) clearTimeout(state.reconnectTimer) - state.reconnectTimer = setTimeout(() => { - console.log(`[WS ${rootId}] Reconnecting...`) - doConnect() - }, 2000) - } - - doConnect() - } - - function disconnectRoot(rootId: string) { - const state = roots.value.get(rootId) - if (!state) return - if (state.reconnectTimer) { - clearTimeout(state.reconnectTimer) - state.reconnectTimer = null - } - if (state.ws) { - state.ws.onclose = null - state.ws.close() - state.ws = null - } - state.connected = false - roots.value.delete(rootId) - updateMergedState() - } - - function setActiveRoots(rootIds: string[]) { - if (disposed) return - const desired = new Set(rootIds) - const current = new Set(roots.value.keys()) - - // Add new roots - for (const rid of desired) { - if (!current.has(rid)) { - connectRoot(rid) - } - } - - // Remove old roots - for (const rid of current) { - if (!desired.has(rid)) { - disconnectRoot(rid) + ws.onerror = (ev) => { + console.error("[WS] Error:", ev) + if (!mediaIndex.value) { + error.value = "WebSocket connection failed" } } } @@ -628,18 +620,24 @@ export function useMediaWebSocket() { function disconnect() { disposed = true stopTaskSweep() - for (const state of roots.value.values()) { - if (state.reconnectTimer) { - clearTimeout(state.reconnectTimer) - } - if (state.ws) { - state.ws.onclose = null - state.ws.close() - } + + if (reconnectTimer) { + clearTimeout(reconnectTimer) + reconnectTimer = null } + + if (wsRef.value) { + wsRef.value.onclose = null + wsRef.value.close() + wsRef.value = null + } + + connected.value = false roots.value.clear() + rootStates.value.clear() } + connect() onUnmounted(disconnect) return { @@ -648,7 +646,7 @@ export function useMediaWebSocket() { error: readonly(error), connected: readonly(connected), tasks: readonly(tasks), - setActiveRoots, + roots: readonly(roots), disconnect, } } diff --git a/frontend/src/types.ts b/frontend/src/types.ts index ca10260..d543795 100644 --- a/frontend/src/types.ts +++ b/frontend/src/types.ts @@ -189,17 +189,35 @@ export interface TaskInfo { } // WebSocket message types (matching server msgspec tagged structs) +export interface WsRootStatus { + root_id: string + path: string + status: string + error: string | null + snapshot_loaded: boolean + movies: number + series: number +} + +export interface WsRootInitData { + movies: Record + series: Record + people?: Record +} + +export interface WsRootsMessage { + type: "roots" + roots: WsRootStatus[] +} + export interface WsInitMessage { type: "init" - data: { - movies: Record - series: Record - people?: Record - } + roots: Record } export interface WsUpsertMessage { type: "upsert" + root_id: string kind: "movie" | "series" id: string item: Movie | Series @@ -208,13 +226,20 @@ export interface WsUpsertMessage { export interface WsRemoveMessage { type: "remove" + root_id: string kind: "movie" | "series" id: string } export interface WsTaskMessage { type: "task" + root_id: string data: TaskInfo } -export type WsMessage = WsInitMessage | WsUpsertMessage | WsRemoveMessage | WsTaskMessage +export type WsMessage = + | WsRootsMessage + | WsInitMessage + | WsUpsertMessage + | WsRemoveMessage + | WsTaskMessage diff --git a/mediahive/hivescan/__main__.py b/mediahive/hivescan/__main__.py index 1d92628..974961f 100644 --- a/mediahive/hivescan/__main__.py +++ b/mediahive/hivescan/__main__.py @@ -33,8 +33,8 @@ Examples: Exclude paths by creating .mediahive/scanignore (gitignore syntax). -The server exposes per-root endpoints: - WS /api/ws/{root_id} Live index updates & task progress +The server exposes a unified endpoint: + WS /api/ws Live index updates, task progress, and root status changes """, ) parser.add_argument( diff --git a/mediahive/index_store.py b/mediahive/index_store.py index e37581d..2d1ca3b 100644 --- a/mediahive/index_store.py +++ b/mediahive/index_store.py @@ -9,6 +9,7 @@ debounced background task. import asyncio import contextlib import logging +from collections.abc import Callable from datetime import datetime from pathlib import Path @@ -23,10 +24,6 @@ from mediahive.models.data import ( TaskInfo, ) from mediahive.models.events import Remove, Task, Upsert -from mediahive.models.protocol import ( - WsInit, - WsInitData, -) from mediahive.models.tmdb import Person logger = logging.getLogger("mediahive.index_store") @@ -64,6 +61,8 @@ class IndexStore: # Connected WebSocket clients self._clients: set[WebSocket] = set() + # Passive listeners for broadcast events (used by server-level WS fan-in) + self._listeners: set[Callable[[object], None]] = set() # Snapshot debounce state self._snapshot_dirty = False @@ -388,19 +387,30 @@ class IndexStore: # WebSocket management # ------------------------------------------------------------------ + def add_listener(self, listener: Callable[[object], None]) -> None: + """Register a listener called for each broadcast message.""" + self._listeners.add(listener) + + def remove_listener(self, listener: Callable[[object], None]) -> None: + """Unregister a previously registered broadcast listener.""" + self._listeners.discard(listener) + async def connect(self, ws: WebSocket) -> None: """Accept a WS client and send the full index as init.""" await ws.accept() self._clients.add(ws) logger.info("WS client connected (%d total)", len(self._clients)) # Send full current state - msg = WsInit( - data=WsInitData( - movies=dict(self.movies), - series=dict(self.series), - people=dict(self.people), - ) - ) + msg = { + "type": "init", + "roots": { + "": { + "movies": dict(self.movies), + "series": dict(self.series), + "people": dict(self.people), + } + }, + } await ws.send_bytes(msgspec.json.encode(msg)) def disconnect(self, ws: WebSocket) -> None: @@ -410,6 +420,12 @@ class IndexStore: def _broadcast(self, msg: object) -> None: """Broadcast a message to all connected WS clients (non-blocking).""" + for listener in tuple(self._listeners): + try: + listener(msg) + except Exception: + logger.exception("IndexStore listener failed") + data = msgspec.json.encode(msg) dead: list[WebSocket] = [] for ws in self._clients: diff --git a/mediahive/models/protocol.py b/mediahive/models/protocol.py index 4c83350..e4f7b3e 100644 --- a/mediahive/models/protocol.py +++ b/mediahive/models/protocol.py @@ -8,8 +8,8 @@ from __future__ import annotations import msgspec from fastapi.responses import Response -from .data import Movie, Series -from .events import Remove, ScanEvent, Task, Upsert +from .data import Movie, Series, TaskInfo +from .events import ScanEvent from .tmdb import Person # --------------------------------------------------------------------------- @@ -17,33 +17,78 @@ from .tmdb import Person # --------------------------------------------------------------------------- -class WsInitData(msgspec.Struct): - """Payload of the init message.""" +class WsRootStatus(msgspec.Struct): + """Current status for one configured root.""" + + root_id: str + path: str + status: str + error: str | None = None + snapshot_loaded: bool = False + movies: int = 0 + series: int = 0 + + +class WsRootInitData(msgspec.Struct): + """Initial full index payload for one root.""" movies: dict[str, Movie] series: dict[str, Series] people: dict[int, Person] -class WsInit(msgspec.Struct, tag="init"): - """Full index sent on WS connect.""" +class WsRoots(msgspec.Struct, tag="roots"): + """Root list and status update.""" - data: WsInitData + roots: list[WsRootStatus] + + +class WsInit(msgspec.Struct, tag="init"): + """Full index payload keyed by root_id.""" + + roots: dict[str, WsRootInitData] + + +class WsUpsert(msgspec.Struct, tag="upsert"): + """Single item inserted or updated for one root.""" + + root_id: str + kind: str # "movie" or "series" + id: str + item: Movie | Series + people: dict[int, Person] | None = None + + +class WsRemove(msgspec.Struct, tag="remove"): + """Single item removed for one root.""" + + root_id: str + kind: str + id: str + + +class WsTask(msgspec.Struct, tag="task"): + """Task progress update for one root.""" + + root_id: str + data: TaskInfo # Union of all outbound WS messages (for documentation / future decoding) -WsMessage = WsInit | Upsert | Remove | Task +WsMessage = WsRoots | WsInit | WsUpsert | WsRemove | WsTask # Re-export unified types for backward compatibility __all__ = [ - "Remove", "ScanEvent", - "Task", - "Upsert", "WsInit", - "WsInitData", "WsMessage", + "WsRemove", + "WsRootInitData", + "WsRootStatus", + "WsRoots", + "WsTask", + "WsUpsert", ] @@ -67,7 +112,7 @@ class OpenFolderRequest(msgspec.Struct): class RootsRequest(msgspec.Struct): - """PUT /api/roots body.""" + """PUT /api/config/roots body.""" roots: dict[str, str] @@ -80,7 +125,7 @@ class RootEntryResponse(msgspec.Struct): class RootStatusResponse(msgspec.Struct): - """Per-root status in GET /api/roots.""" + """Legacy per-root status shape kept for non-WS callers.""" root_id: str path: str diff --git a/mediahive/root_registry.py b/mediahive/root_registry.py index cee08d9..302cb4b 100644 --- a/mediahive/root_registry.py +++ b/mediahive/root_registry.py @@ -280,7 +280,7 @@ class Supervisor: base_name = _derive_root_name(configured_path) unique_name = base_name suffix = 2 - existing_names = {e.name for e in candidates} + existing_names = {e.root_id for e in candidates} while unique_name in existing_names: unique_name = f"{base_name}{suffix}" suffix += 1 diff --git a/mediahive/server.py b/mediahive/server.py index 6994276..17c4c52 100644 --- a/mediahive/server.py +++ b/mediahive/server.py @@ -40,10 +40,18 @@ from mediahive.config import load_config 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.events import Remove, Task, Upsert from mediahive.models.protocol import ( OpenFolderRequest, PlayMediaRequest, RootsRequest, + WsInit, + WsRemove, + WsRootInitData, + WsRoots, + WsRootStatus, + WsTask, + WsUpsert, ) from mediahive.players import detect_players, launch_player from mediahive.root_registry import Supervisor @@ -114,6 +122,53 @@ def _load_root_metadata(root_path: Path, meta_key: str): ) +def _all_root_statuses() -> list[WsRootStatus]: + """Return root statuses in WebSocket wire format.""" + return [ + WsRootStatus( + root_id=s["root_id"], + path=s["path"], + status=s["status"], + error=s.get("error"), + snapshot_loaded=bool(s.get("snapshot_loaded")), + movies=int(s.get("movies", 0)), + series=int(s.get("series", 0)), + ) + for s in supervisor.all_statuses() + ] + + +def _full_ws_init(root_ids: set[str] | None = None) -> WsInit: + """Build an init payload for all roots or only selected root_ids.""" + roots: dict[str, WsRootInitData] = {} + for rid, ctx in supervisor.all_contexts().items(): + if root_ids is not None and rid not in root_ids: + continue + roots[rid] = WsRootInitData( + movies=dict(ctx.store.movies), + series=dict(ctx.store.series), + people=dict(ctx.store.people), + ) + return WsInit(roots=roots) + + +def _translate_store_event(root_id: str, event: object): + """Convert per-root store events into unified websocket messages.""" + if isinstance(event, Upsert): + return WsUpsert( + root_id=root_id, + kind=event.kind, + id=event.id, + item=event.item, + people=event.people, + ) + if isinstance(event, Remove): + return WsRemove(root_id=root_id, kind=event.kind, id=event.id) + if isinstance(event, Task): + return WsTask(root_id=root_id, data=event.data) + return None + + def _open_with_default_app(path: Path) -> None: if sys.platform == "win32": os.startfile(str(path)) @@ -352,7 +407,7 @@ async def _activate_all_roots() -> None: desired.update(cfg.roots) if not desired: - logger.info("No roots configured; waiting for PUT /api/roots") + logger.info("No roots configured; waiting for PUT /api/config/roots") return # Validate paths in a thread pool (macOS permission-dialog safe) @@ -445,13 +500,7 @@ async def get_config(): # --- Root management --- -@app.get("/api/roots") -async def get_roots(): - """List all active roots with their status.""" - return {"roots": supervisor.all_statuses()} - - -@app.put("/api/roots") +@app.put("/api/config/roots") async def put_roots(request: Request): """Atomically replace the full root set.""" body = msgspec.json.decode(await request.body(), type=RootsRequest) @@ -467,22 +516,102 @@ async def put_roots(request: Request): } -# --- Per-root WebSocket --- +# --- Unified WebSocket --- -@app.websocket("/api/ws/{root_id}") -async def ws_endpoint(ws: WebSocket, root_id: str) -> None: - """Live index updates and task progress for a single root.""" - ctx = supervisor.get(root_id) - if ctx is None: - await ws.close(code=1008, reason="Unknown root") - return +@app.websocket("/api/ws") +async def ws_endpoint(ws: WebSocket) -> None: + """Live updates stream for all roots and all connected clients.""" + listeners: dict[str, object] = {} + attached_contexts = supervisor.all_contexts() + outbound: asyncio.Queue[bytes] = asyncio.Queue() start = time.perf_counter() ws_id = log_ws_open(ws) close_code: int | None = None + prev_root_ids: set[str] = set() + prev_meta: dict[str, tuple[str, str, str | None, bool]] = {} + + def _sync_listeners() -> tuple[ + set[str], dict[str, tuple[str, str, str | None, bool]] + ]: + nonlocal attached_contexts + current_contexts = supervisor.all_contexts() + current_ids = set(current_contexts.keys()) + + for rid in list(listeners.keys()): + if rid in current_ids: + continue + old_ctx = attached_contexts.get(rid) + listener = listeners.pop(rid) + if old_ctx is not None: + old_ctx.store.remove_listener(listener) + + for rid, ctx in current_contexts.items(): + if rid in listeners: + continue + + def _listener(event: object, *, _rid=rid) -> None: + translated = _translate_store_event(_rid, event) + if translated is None: + return + outbound.put_nowait(msgspec.json.encode(translated)) + + listeners[rid] = _listener + ctx.store.add_listener(_listener) + + attached_contexts = current_contexts + meta = { + s.root_id: (s.path, s.status, s.error, s.snapshot_loaded) + for s in _all_root_statuses() + } + return current_ids, meta + + async def _send_outbound() -> None: + while True: + payload = await outbound.get() + await ws.send_bytes(payload) + + async def _watch_roots() -> None: + nonlocal prev_root_ids, prev_meta + while True: + current_ids, current_meta = _sync_listeners() + root_set_changed = current_ids != prev_root_ids + meta_changed = current_meta != prev_meta + + if root_set_changed or meta_changed: + outbound.put_nowait( + msgspec.json.encode(WsRoots(roots=_all_root_statuses())) + ) + + if root_set_changed: + outbound.put_nowait(msgspec.json.encode(_full_ws_init())) + else: + became_loaded = { + rid + for rid, meta in current_meta.items() + if rid in prev_meta and not prev_meta[rid][3] and meta[3] + } + if became_loaded: + outbound.put_nowait( + msgspec.json.encode(_full_ws_init(became_loaded)) + ) + + prev_root_ids = current_ids + prev_meta = current_meta + await asyncio.sleep(1.0) + + await ws.accept() + + current_ids, current_meta = _sync_listeners() + prev_root_ids = current_ids + prev_meta = current_meta + await ws.send_bytes(msgspec.json.encode(WsRoots(roots=_all_root_statuses()))) + await ws.send_bytes(msgspec.json.encode(_full_ws_init())) + + sender_task = asyncio.create_task(_send_outbound()) + watcher_task = asyncio.create_task(_watch_roots()) - await ctx.store.connect(ws) try: while True: await ws.receive_text() @@ -491,7 +620,18 @@ async def ws_endpoint(ws: WebSocket, root_id: str) -> None: except OSError, RuntimeError: pass finally: - ctx.store.disconnect(ws) + sender_task.cancel() + watcher_task.cancel() + with suppress(asyncio.CancelledError): + await sender_task + with suppress(asyncio.CancelledError): + await watcher_task + + for rid, listener in list(listeners.items()): + ctx = attached_contexts.get(rid) + if ctx is not None: + ctx.store.remove_listener(listener) + log_ws_close(ws_id, close_code, time.perf_counter() - start) @@ -581,6 +721,30 @@ async def root_metadata(root_id: str, meta_key: str): return {"key": meta_key, "data": _load_root_metadata(ctx.root_path, meta_key)} +@app.get("/api/meta/playback-state") +async def merged_playback_state(): + """Return merged playback-state resume positions from all active roots.""" + merged: dict[str, float] = {} + for ctx in supervisor.all_contexts().values(): + try: + data = _load_root_metadata(ctx.root_path, "playback-state") + except HTTPException as exc: + if exc.status_code == 404: + continue + raise + + if not isinstance(data, dict): + continue + positions = data.get("resume_positions") + if not isinstance(positions, dict): + continue + for key, value in positions.items(): + if isinstance(value, int | float): + merged[str(key)] = float(value) + + return {"key": "playback-state", "data": {"resume_positions": merged}} + + # --- MPC-BE / Player status --- diff --git a/mediahive/winmain.py b/mediahive/winmain.py index 5ea9d23..b07c892 100644 --- a/mediahive/winmain.py +++ b/mediahive/winmain.py @@ -872,7 +872,7 @@ def winmain() -> None: def _activate_initial_roots() -> None: body = json.dumps({"roots": initial_roots}).encode("utf-8") req = urllib.request.Request( - url=f"{backend_url}/api/roots", + url=f"{backend_url}/api/config/roots", data=body, method="PUT", headers={"Content-Type": "application/json"},