feat(service-storage-harness): slice 3 — migrate MediaIndex onto harness (scoped replace_items, +service_id)
Register MediaIndex as a harness concern with ALTER TABLE migration to add service_id column (idempotent). Scope replace_items by service_id (FIXES latent global-clear bug where building for one Jellyfin wiped another's rows). Scope query by service_id (empty-string = all rows, backward-compat). Thread service_id through build_media_index + worker + query_media router. New regression test proves scoped replace preserves other services' rows. Backend: 321 pytest pass, ruff clean. Frontend: build green.
This commit is contained in:
@@ -282,6 +282,7 @@ def query_media(
|
|||||||
sort_order: str = Query("Ascending", description="Ascending or Descending"),
|
sort_order: str = Query("Ascending", description="Ascending or Descending"),
|
||||||
limit: int = Query(100, ge=1, le=1000),
|
limit: int = Query(100, ge=1, le=1000),
|
||||||
offset: int = Query(0, ge=0),
|
offset: int = Query(0, ge=0),
|
||||||
|
jellyfin_service_id: str | None = None,
|
||||||
client: JellyfinClient = Depends(get_jellyfin_client),
|
client: JellyfinClient = Depends(get_jellyfin_client),
|
||||||
user_id: str = Depends(get_user_id),
|
user_id: str = Depends(get_user_id),
|
||||||
index: MediaIndex = Depends(get_media_index),
|
index: MediaIndex = Depends(get_media_index),
|
||||||
@@ -316,6 +317,7 @@ def query_media(
|
|||||||
sort_order=sort_order,
|
sort_order=sort_order,
|
||||||
limit=limit,
|
limit=limit,
|
||||||
offset=offset,
|
offset=offset,
|
||||||
|
service_id=jellyfin_service_id or "",
|
||||||
)
|
)
|
||||||
|
|
||||||
logger.info("Media query returned total=%s rows=%s", total, len(rows))
|
logger.info("Media query returned total=%s rows=%s", total, len(rows))
|
||||||
|
|||||||
@@ -17,6 +17,7 @@ from typing import Any, Callable, Iterable
|
|||||||
from media_library_viewer_api.clients.jellyfin import JellyfinClient
|
from media_library_viewer_api.clients.jellyfin import JellyfinClient
|
||||||
from media_library_viewer_api.domain.media import display_media_row, normalize_media_item
|
from media_library_viewer_api.domain.media import display_media_row, normalize_media_item
|
||||||
from media_library_viewer_api.path_utils import resolve_remote_media_path
|
from media_library_viewer_api.path_utils import resolve_remote_media_path
|
||||||
|
from media_library_viewer_api.services.service_data import StorageConcern
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
@@ -25,6 +26,20 @@ logger = logging.getLogger(__name__)
|
|||||||
DEFAULT_INDEX_PATH = Path(".cache/media_library_viewer/media_index.sqlite")
|
DEFAULT_INDEX_PATH = Path(".cache/media_library_viewer/media_index.sqlite")
|
||||||
MEDIA_TYPES = "Movie,Episode,Video"
|
MEDIA_TYPES = "Movie,Episode,Video"
|
||||||
|
|
||||||
|
# Harness concern registration: the media_index DB is scoped by service_id so
|
||||||
|
# multiple Jellyfin instances can coexist. The ALTER TABLE migration adds the
|
||||||
|
# service_id column to existing DBs; init_schema adds it for fresh installs.
|
||||||
|
# The harness run_migrations catches "duplicate column name" on re-runs.
|
||||||
|
MEDIA_INDEX_CONCERN = StorageConcern(
|
||||||
|
concern_key="media_index",
|
||||||
|
db_filename=DEFAULT_INDEX_PATH.name,
|
||||||
|
migrations=[
|
||||||
|
"ALTER TABLE media_items ADD COLUMN service_id TEXT NOT NULL DEFAULT ''",
|
||||||
|
],
|
||||||
|
tables=["media_items"],
|
||||||
|
service_id_column="service_id",
|
||||||
|
)
|
||||||
|
|
||||||
# Only values from this whitelist are interpolated into ORDER BY. User-selected
|
# Only values from this whitelist are interpolated into ORDER BY. User-selected
|
||||||
# sort keys map to these known SQL snippets to avoid SQL injection.
|
# sort keys map to these known SQL snippets to avoid SQL injection.
|
||||||
SORT_COLUMNS = {
|
SORT_COLUMNS = {
|
||||||
@@ -135,7 +150,8 @@ class MediaIndex:
|
|||||||
date_added_ts INTEGER,
|
date_added_ts INTEGER,
|
||||||
path TEXT,
|
path TEXT,
|
||||||
library_id TEXT,
|
library_id TEXT,
|
||||||
library_name TEXT
|
library_name TEXT,
|
||||||
|
service_id TEXT NOT NULL DEFAULT ''
|
||||||
);
|
);
|
||||||
CREATE TABLE IF NOT EXISTS index_metadata (
|
CREATE TABLE IF NOT EXISTS index_metadata (
|
||||||
key TEXT PRIMARY KEY,
|
key TEXT PRIMARY KEY,
|
||||||
@@ -167,8 +183,13 @@ class MediaIndex:
|
|||||||
(key, str(value)),
|
(key, str(value)),
|
||||||
)
|
)
|
||||||
|
|
||||||
def replace_items(self, rows: Iterable[dict[str, Any]]) -> int:
|
def replace_items(self, rows: Iterable[dict[str, Any]], service_id: str = "") -> int:
|
||||||
"""Atomically replace indexed media rows with a freshly built set."""
|
"""Atomically replace indexed media rows with a freshly built set.
|
||||||
|
|
||||||
|
Scoped by ``service_id``: only rows belonging to this service are
|
||||||
|
deleted before the new batch is inserted. This means building for one
|
||||||
|
Jellyfin instance no longer wipes another instance's rows.
|
||||||
|
"""
|
||||||
self.init_schema()
|
self.init_schema()
|
||||||
row_list = list(rows)
|
row_list = list(rows)
|
||||||
columns = [
|
columns = [
|
||||||
@@ -194,13 +215,14 @@ class MediaIndex:
|
|||||||
"path",
|
"path",
|
||||||
"library_id",
|
"library_id",
|
||||||
"library_name",
|
"library_name",
|
||||||
|
"service_id",
|
||||||
]
|
]
|
||||||
placeholders = ",".join(["?"] * len(columns))
|
placeholders = ",".join(["?"] * len(columns))
|
||||||
with self.connect() as conn:
|
with self.connect() as conn:
|
||||||
conn.execute("DELETE FROM media_items")
|
conn.execute("DELETE FROM media_items WHERE service_id = ?", (service_id,))
|
||||||
conn.executemany(
|
conn.executemany(
|
||||||
f"INSERT OR REPLACE INTO media_items ({','.join(columns)}) VALUES ({placeholders})",
|
f"INSERT OR REPLACE INTO media_items ({','.join(columns)}) VALUES ({placeholders})",
|
||||||
[[row.get(column) for column in columns] for row in row_list],
|
[[row.get(column) if column != "service_id" else service_id for column in columns] for row in row_list],
|
||||||
)
|
)
|
||||||
conn.execute(
|
conn.execute(
|
||||||
"INSERT OR REPLACE INTO index_metadata (key, value) VALUES ('updated_at', ?)",
|
"INSERT OR REPLACE INTO index_metadata (key, value) VALUES ('updated_at', ?)",
|
||||||
@@ -287,8 +309,14 @@ class MediaIndex:
|
|||||||
sort_order: str = "Ascending",
|
sort_order: str = "Ascending",
|
||||||
limit: int = 100,
|
limit: int = 100,
|
||||||
offset: int = 0,
|
offset: int = 0,
|
||||||
|
service_id: str = "",
|
||||||
) -> tuple[list[dict[str, Any]], int]:
|
) -> tuple[list[dict[str, Any]], int]:
|
||||||
"""Query indexed media with full-index filters, sorting, and pagination."""
|
"""Query indexed media with full-index filters, sorting, and pagination.
|
||||||
|
|
||||||
|
When ``service_id`` is non-empty, only rows matching that service are
|
||||||
|
returned. When empty (the default), all rows are returned (backward-
|
||||||
|
compatible with callers that are not multi-instance aware).
|
||||||
|
"""
|
||||||
self.init_schema()
|
self.init_schema()
|
||||||
where = []
|
where = []
|
||||||
params: list[Any] = []
|
params: list[Any] = []
|
||||||
@@ -309,6 +337,9 @@ class MediaIndex:
|
|||||||
where.append("hdr = 1")
|
where.append("hdr = 1")
|
||||||
elif hdr_filter == "SDR/unknown only":
|
elif hdr_filter == "SDR/unknown only":
|
||||||
where.append("(hdr IS NULL OR hdr = 0)")
|
where.append("(hdr IS NULL OR hdr = 0)")
|
||||||
|
if service_id:
|
||||||
|
where.append("service_id = ?")
|
||||||
|
params.append(service_id)
|
||||||
|
|
||||||
where_sql = " WHERE " + " AND ".join(where) if where else ""
|
where_sql = " WHERE " + " AND ".join(where) if where else ""
|
||||||
sort_sql = SORT_COLUMNS.get(sort_key, SORT_COLUMNS["title"])
|
sort_sql = SORT_COLUMNS.get(sort_key, SORT_COLUMNS["title"])
|
||||||
@@ -338,6 +369,7 @@ def build_media_index(
|
|||||||
fallback_prefix: str = "",
|
fallback_prefix: str = "",
|
||||||
progress_callback: Callable[[dict[str, Any]], None] | None = None,
|
progress_callback: Callable[[dict[str, Any]], None] | None = None,
|
||||||
should_cancel: Callable[[], bool] | None = None,
|
should_cancel: Callable[[], bool] | None = None,
|
||||||
|
service_id: str = "",
|
||||||
) -> int:
|
) -> int:
|
||||||
"""Fetch Jellyfin pages for all selected libraries and rebuild the index."""
|
"""Fetch Jellyfin pages for all selected libraries and rebuild the index."""
|
||||||
index = index or MediaIndex()
|
index = index or MediaIndex()
|
||||||
@@ -453,7 +485,7 @@ def build_media_index(
|
|||||||
logger.info("Media index finalizing rows=%s", len(normalized_rows))
|
logger.info("Media index finalizing rows=%s", len(normalized_rows))
|
||||||
emit("finalizing", "Writing index to disk")
|
emit("finalizing", "Writing index to disk")
|
||||||
ensure_not_cancelled()
|
ensure_not_cancelled()
|
||||||
count = index.replace_items(normalized_rows)
|
count = index.replace_items(normalized_rows, service_id=service_id)
|
||||||
duration = time.perf_counter() - started_at
|
duration = time.perf_counter() - started_at
|
||||||
index.set_metadata("build_duration_seconds", f"{duration:.3f}")
|
index.set_metadata("build_duration_seconds", f"{duration:.3f}")
|
||||||
processed_total = count
|
processed_total = count
|
||||||
|
|||||||
@@ -143,6 +143,10 @@ def get_service_data_harness() -> ServiceDataHarness:
|
|||||||
from media_library_viewer_api.services.qbittorrent_store import QBITTORRENT_CONCERN
|
from media_library_viewer_api.services.qbittorrent_store import QBITTORRENT_CONCERN
|
||||||
|
|
||||||
_HARNESS.register(QBITTORRENT_CONCERN)
|
_HARNESS.register(QBITTORRENT_CONCERN)
|
||||||
|
|
||||||
|
from media_library_viewer_api.services.media_index_impl import MEDIA_INDEX_CONCERN
|
||||||
|
|
||||||
|
_HARNESS.register(MEDIA_INDEX_CONCERN)
|
||||||
_HARNESS.run_migrations()
|
_HARNESS.run_migrations()
|
||||||
return _HARNESS
|
return _HARNESS
|
||||||
|
|
||||||
|
|||||||
@@ -174,6 +174,7 @@ def run_build(final_index_path: str | Path, staging_index_path: str | Path, serv
|
|||||||
fallback_prefix=settings.path_prefix,
|
fallback_prefix=settings.path_prefix,
|
||||||
progress_callback=lambda state: _progress_callback(final_index, pid, state),
|
progress_callback=lambda state: _progress_callback(final_index, pid, state),
|
||||||
should_cancel=lambda: _cancel_requested(final_index),
|
should_cancel=lambda: _cancel_requested(final_index),
|
||||||
|
service_id=service_id,
|
||||||
)
|
)
|
||||||
# Swap the staging database into place atomically.
|
# Swap the staging database into place atomically.
|
||||||
os.replace(staging_index.db_path, final_index.db_path)
|
os.replace(staging_index.db_path, final_index.db_path)
|
||||||
|
|||||||
@@ -241,6 +241,53 @@ class TestMediaIndexReplace:
|
|||||||
status = index.status()
|
status = index.status()
|
||||||
assert status.item_count == 2
|
assert status.item_count == 2
|
||||||
|
|
||||||
|
def test_replace_scoped_by_service_id_preserves_other_services(self, index):
|
||||||
|
"""Regression test: building for one Jellyfin must not wipe another's rows.
|
||||||
|
|
||||||
|
Before the migration, ``replace_items`` did ``DELETE FROM media_items``
|
||||||
|
(global clear). This test locks in the fix: a scoped replace preserves
|
||||||
|
rows belonging to a different service_id.
|
||||||
|
"""
|
||||||
|
rows_a = [
|
||||||
|
{"id": "a1", "title": "Alpha Movie", "type": "Movie", "library_id": "l1", "library_name": "L1"},
|
||||||
|
{"id": "a2", "title": "Alpha Show", "type": "Episode", "library_id": "l2", "library_name": "L2"},
|
||||||
|
]
|
||||||
|
rows_b = [
|
||||||
|
{"id": "b1", "title": "Beta Movie", "type": "Movie", "library_id": "l1", "library_name": "L1"},
|
||||||
|
]
|
||||||
|
|
||||||
|
index.replace_items(rows_a, service_id="svc-a")
|
||||||
|
assert index.status().item_count == 2
|
||||||
|
|
||||||
|
# Replacing for svc-b must NOT wipe svc-a's rows.
|
||||||
|
index.replace_items(rows_b, service_id="svc-b")
|
||||||
|
assert index.status().item_count == 3 # 2 from svc-a + 1 from svc-b
|
||||||
|
|
||||||
|
# Querying svc-a returns only its rows.
|
||||||
|
rows_a_result, total_a = index.query(
|
||||||
|
library_ids=["l1", "l2"],
|
||||||
|
media_types=["Movie", "Episode"],
|
||||||
|
service_id="svc-a",
|
||||||
|
)
|
||||||
|
assert total_a == 2
|
||||||
|
assert {r["id"] for r in rows_a_result} == {"a1", "a2"}
|
||||||
|
|
||||||
|
# Querying svc-b returns only its rows.
|
||||||
|
rows_b_result, total_b = index.query(
|
||||||
|
library_ids=["l1"],
|
||||||
|
media_types=["Movie"],
|
||||||
|
service_id="svc-b",
|
||||||
|
)
|
||||||
|
assert total_b == 1
|
||||||
|
assert rows_b_result[0]["id"] == "b1"
|
||||||
|
|
||||||
|
# Querying with no service_id returns all rows (backward-compat).
|
||||||
|
_, total_all = index.query(
|
||||||
|
library_ids=["l1", "l2"],
|
||||||
|
media_types=["Movie", "Episode"],
|
||||||
|
)
|
||||||
|
assert total_all == 3
|
||||||
|
|
||||||
|
|
||||||
class TestMediaIndexMetadata:
|
class TestMediaIndexMetadata:
|
||||||
def test_set_and_read_metadata(self, index):
|
def test_set_and_read_metadata(self, index):
|
||||||
|
|||||||
Reference in New Issue
Block a user