From fd31e9e79fe1e290534e2c439dc1b2dc8a498799 Mon Sep 17 00:00:00 2001 From: chodeus Date: Fri, 14 Aug 2026 07:53:49 +0800 Subject: [PATCH 1/5] refactor(db): media_cache splits into domain mixins MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Edit-history, metadata-completeness and library-stats move to sibling modules as mixins; MediaCache keeps its exact class surface (dir() is byte-identical against main) and every import path re-exports unchanged. media_cache.py: 1791 -> 1254 lines. Move-only — each block byte-identical to its origin. --- backend/util/database/__init__.py | 4 +- backend/util/database/media_cache.py | 547 +------------------- backend/util/database/media_edit_history.py | 45 ++ backend/util/database/media_metadata.py | 109 ++++ backend/util/database/media_stats.py | 414 +++++++++++++++ 5 files changed, 575 insertions(+), 544 deletions(-) create mode 100644 backend/util/database/media_edit_history.py create mode 100644 backend/util/database/media_metadata.py create mode 100644 backend/util/database/media_stats.py diff --git a/backend/util/database/__init__.py b/backend/util/database/__init__.py index ed3279d3..ad29b948 100644 --- a/backend/util/database/__init__.py +++ b/backend/util/database/__init__.py @@ -12,11 +12,11 @@ from .db_base import DatabaseBase, escape_like from .holiday import HolidayStatus from .media_asset_matches import MediaAssetMatches -from .media_cache import ( +from .media_cache import MediaCache +from .media_metadata import ( INCOMPLETE_METADATA_FIELDS, INCOMPLETE_METADATA_INT_FIELDS, NEVER_POPULATED_FIELDS, - MediaCache, is_missing_value, ) from .plex_cache import PlexCache diff --git a/backend/util/database/media_cache.py b/backend/util/database/media_cache.py index 41953746..f2bbc953 100755 --- a/backend/util/database/media_cache.py +++ b/backend/util/database/media_cache.py @@ -5,119 +5,12 @@ from backend.util.normalization import normalize_titles from .db_base import DatabaseBase, escape_like +from .media_edit_history import EditHistoryMixin +from .media_metadata import MetadataCompletenessMixin +from .media_stats import StatsMixin -# Library-health SQL fragments, shared across the stats queries so by_type, -# totals and by_instance can't drift. Everything is counted in CONTENT UNITS — -# a movie, an album, or a single TV episode — to match each *arr's native unit: -# units = total content units -# in_library = units whose file is present -# missing = monitored units that are released/aired but have no file -# upcoming = monitored units not yet released/aired -# A Sonarr SEASON row expands to its episode counts; a movie/album is 1 unit; -# shows, seasons-as-rows... no — shows and artists are CONTAINERS (0 units; the -# episodes/albums carry the counts), so they fall through to 0 automatically. -# "released" gates simple (movie/album) units: Lidarr albums use their own -# release_date (absent = released), movies/shows use the *arr `status` (mirrors -# release_readiness.UNRELEASED_STATUSES). For seasons, "aired" IS the gate: -# missing = aired-on-disk shortfall, upcoming = not-yet-aired. -_RELEASED_SQL = ( - "CASE WHEN asset_type='album' " - "THEN (release_date IS NULL OR release_date <= date('now')) " - "ELSE (status IS NULL OR status NOT IN " - "('announced','deleted','tba','upcoming')) END" -) -_IS_SEASON = "(asset_type='show' AND season_number IS NOT NULL)" -_IS_SIMPLE = "asset_type IN ('movie','album')" # 1-unit content types - -_UNITS_TOTAL_SQL = ( - f"SUM(CASE WHEN {_IS_SEASON} THEN COALESCE(total_episodes,0) " - f"WHEN {_IS_SIMPLE} THEN 1 ELSE 0 END)" -) -# An album under an UNMONITORED artist isn't "wanted" — Lidarr's Wanted page -# excludes it, so missing/upcoming do too. NULL (pre-column rows) = monitored. -_ALBUM_ARTIST_OK = "(asset_type != 'album' OR COALESCE(artist_monitored, 1) = 1)" -_IN_LIBRARY_SQL = ( - f"SUM(CASE WHEN {_IS_SEASON} THEN COALESCE(episode_files,0) " - f"WHEN {_IS_SIMPLE} AND has_content=1 THEN 1 ELSE 0 END)" -) -_MISSING_SQL = ( - f"SUM(CASE WHEN {_IS_SEASON} AND monitored=1 " - "THEN MAX(0, COALESCE(aired_episodes,0) - COALESCE(episode_files,0)) " - f"WHEN {_IS_SIMPLE} AND monitored=1 AND COALESCE(has_content,0)=0 " - f"AND {_RELEASED_SQL} AND {_ALBUM_ARTIST_OK} THEN 1 ELSE 0 END)" -) -_UPCOMING_SQL = ( - f"SUM(CASE WHEN {_IS_SEASON} AND monitored=1 " - "THEN MAX(0, COALESCE(total_episodes,0) - COALESCE(aired_episodes,0)) " - f"WHEN {_IS_SIMPLE} AND monitored=1 AND COALESCE(has_content,0)=0 " - f"AND NOT ({_RELEASED_SQL}) AND {_ALBUM_ARTIST_OK} THEN 1 ELSE 0 END)" -) -# Row-count fragments (containers): for the artist/show context cards. -_MONITORED_SQL = "SUM(CASE WHEN monitored=1 THEN 1 ELSE 0 END)" -_SHOW_COUNT_SQL = ( - "SUM(CASE WHEN asset_type='show' AND season_number IS NULL THEN 1 ELSE 0 END)" -) -_MONITORED_SHOWS_SQL = ( - "SUM(CASE WHEN asset_type='show' AND season_number IS NULL AND monitored=1 " - "THEN 1 ELSE 0 END)" -) -_SEASON_COUNT_SQL = ( - "SUM(CASE WHEN asset_type='show' AND season_number IS NOT NULL THEN 1 ELSE 0 END)" -) -_ARTIST_COUNT_SQL = "SUM(CASE WHEN asset_type='artist' THEN 1 ELSE 0 END)" -_MONITORED_ARTISTS_SQL = ( - "SUM(CASE WHEN asset_type='artist' AND monitored=1 THEN 1 ELSE 0 END)" -) - -# Columns find_incomplete_metadata may test, and the subset stored as INTEGER -# (they can't hold '', so they compare against NULL/0 instead). -INCOMPLETE_METADATA_FIELDS = frozenset( - { - "rating", - "studio", - "language", - "genre", - "runtime", - "edition", - "tmdb_id", - "tvdb_id", - "imdb_id", - "year", - } -) -INCOMPLETE_METADATA_INT_FIELDS = frozenset({"tmdb_id", "tvdb_id", "runtime"}) - - -def is_missing_value(field: str, value) -> bool: - """True when `value` is missing for `field` (INT fields: None/0; else None/'').""" - if field in INCOMPLETE_METADATA_INT_FIELDS: - return value is None or value == 0 - return value is None or value == "" - - -# Fields the ARR normalize layer never populates for a given asset_type, so -# flagging them as "missing" is a false positive. Radarr has no tvdbId, -# Sonarr has no tmdbId, Lidarr (artist) uses MusicBrainz IDs and leaves -# tmdb/tvdb/imdb + rating/runtime/language/edition as None by design. -NEVER_POPULATED_FIELDS = { - "movie": {"tvdb_id"}, - "show": {"tmdb_id"}, - "artist": { - "tmdb_id", - "tvdb_id", - "imdb_id", - "rating", - "runtime", - "language", - "edition", - }, -} -# asset_types with their own expected-field rules; anything else matches -# find_incomplete_metadata's catch-all clause. -_KNOWN_ASSET_TYPES = ("movie", "show", "artist") - - -class MediaCache(DatabaseBase): + +class MediaCache(EditHistoryMixin, MetadataCompletenessMixin, StatsMixin, DatabaseBase): """ Interface for the media_cache table. Provides CRUD and sync operations for tracked media assets. @@ -1041,347 +934,6 @@ def search( return {"items": items, "total": total, "limit": limit, "offset": offset} - def get_stats( - self, asset_type: Optional[str] = None, period_days: int = None - ) -> dict: - """Aggregate statistics from media_cache.""" - conditions = [] - params: list = [] - if asset_type and asset_type != "all": - conditions.append("asset_type = ?") - params.append(asset_type) - if period_days: - conditions.append("created_at >= datetime('now', ?)") - params.append(f"-{period_days} days") - - where = ("WHERE " + " AND ".join(conditions)) if conditions else "" - params = tuple(params) - - # Library-health metrics — see the module-level _*_SQL fragments. - rows = ( - self.execute_query( - f""" - SELECT asset_type, - COUNT(*) as total, - {_UNITS_TOTAL_SQL} as units, - {_IN_LIBRARY_SQL} as in_library, - {_MISSING_SQL} as missing, - {_UPCOMING_SQL} as upcoming, - {_MONITORED_SQL} as monitored, - {_SHOW_COUNT_SQL} as show_count, - {_MONITORED_SHOWS_SQL} as monitored_shows, - {_SEASON_COUNT_SQL} as season_count, - COUNT(DISTINCT instance_name) as instances - FROM media_cache {where} - GROUP BY asset_type - """, - params, - fetch_all=True, - ) - or [] - ) - - totals = self.execute_query( - f""" - SELECT {_UNITS_TOTAL_SQL} as total, - {_IN_LIBRARY_SQL} as in_library, - {_MISSING_SQL} as missing, - {_UPCOMING_SQL} as upcoming, - {_MONITORED_SQL} as monitored - FROM media_cache {where} - """, - params, - fetch_one=True, - ) - - return { - "by_type": rows, - # Headline numbers are in content units (movies + episodes + albums). - "total": (totals["total"] or 0) if totals else 0, - "in_library": (totals["in_library"] or 0) if totals else 0, - "missing": (totals["missing"] or 0) if totals else 0, - "upcoming": (totals["upcoming"] or 0) if totals else 0, - "monitored": (totals["monitored"] or 0) if totals else 0, - } - - def get_detailed_stats( - self, asset_type: Optional[str] = None, period_days: int = None - ) -> dict: - """Extended statistics with breakdowns by multiple dimensions.""" - conditions = [] - params_list: list = [] - if asset_type and asset_type != "all": - conditions.append("asset_type = ?") - params_list.append(asset_type) - if period_days: - conditions.append("created_at >= datetime('now', ?)") - params_list.append(f"-{period_days} days") - - where = ("WHERE " + " AND ".join(conditions)) if conditions else "" - params = tuple(params_list) - - # Base stats (same as get_stats) - base = self.get_stats(asset_type=asset_type, period_days=period_days) - - # By instance (source = service type: radarr/sonarr/lidarr/plex) - by_instance = ( - self.execute_query( - f"""SELECT instance_name, source, COUNT(*) as total, - {_UNITS_TOTAL_SQL} as units, - {_IN_LIBRARY_SQL} as in_library, - {_MISSING_SQL} as missing, - {_UPCOMING_SQL} as upcoming, - {_MONITORED_SQL} as monitored, - {_SHOW_COUNT_SQL} as show_count, - {_MONITORED_SHOWS_SQL} as monitored_shows, - {_SEASON_COUNT_SQL} as season_count, - {_ARTIST_COUNT_SQL} as artist_count, - {_MONITORED_ARTISTS_SQL} as monitored_artists - FROM media_cache {where} - GROUP BY instance_name, source ORDER BY total DESC""", - params, - fetch_all=True, - ) - or [] - ) - - # By status - by_status = ( - self.execute_query( - f"""SELECT status, COUNT(*) as count - FROM media_cache {where + (" AND" if where else "WHERE")} status IS NOT NULL AND status != '' - GROUP BY status ORDER BY count DESC""", - params, - fetch_all=True, - ) - or [] - ) - - # By language - by_language = ( - self.execute_query( - f"""SELECT language, COUNT(*) as count - FROM media_cache {where + (" AND" if where else "WHERE")} language IS NOT NULL AND language != '' - GROUP BY language ORDER BY count DESC""", - params, - fetch_all=True, - ) - or [] - ) - - # By rating (content ratings like PG, R, TV-MA) - by_rating = ( - self.execute_query( - f"""SELECT rating, COUNT(*) as count - FROM media_cache {where + (" AND" if where else "WHERE")} rating IS NOT NULL AND rating != '' - GROUP BY rating ORDER BY count DESC""", - params, - fetch_all=True, - ) - or [] - ) - - # By studio (top 50) - by_studio = ( - self.execute_query( - f"""SELECT studio, COUNT(*) as count - FROM media_cache {where + (" AND" if where else "WHERE")} studio IS NOT NULL AND studio != '' - GROUP BY studio ORDER BY count DESC LIMIT 50""", - params, - fetch_all=True, - ) - or [] - ) - - # By year/decade - by_decade = ( - self.execute_query( - f"""SELECT (CAST(year AS INTEGER) / 10) * 10 as decade, COUNT(*) as count - FROM media_cache {where + (" AND" if where else "WHERE")} year IS NOT NULL AND year != '' - GROUP BY decade ORDER BY decade DESC""", - params, - fetch_all=True, - ) - or [] - ) - by_decade = [ - {"decade": f"{r['decade']}s", "count": r["count"]} - for r in by_decade - if r.get("decade") - ] - - # By runtime buckets - by_runtime = ( - self.execute_query( - f"""SELECT - CASE - WHEN CAST(runtime AS INTEGER) < 30 THEN 'Under 30m' - WHEN CAST(runtime AS INTEGER) < 60 THEN '30-60m' - WHEN CAST(runtime AS INTEGER) < 90 THEN '60-90m' - WHEN CAST(runtime AS INTEGER) < 120 THEN '90-120m' - WHEN CAST(runtime AS INTEGER) < 150 THEN '120-150m' - ELSE '150m+' - END as bucket, - COUNT(*) as count - FROM media_cache {where + (" AND" if where else "WHERE")} runtime IS NOT NULL AND runtime != '' AND CAST(runtime AS INTEGER) > 0 - GROUP BY bucket ORDER BY MIN(CAST(runtime AS INTEGER))""", - params, - fetch_all=True, - ) - or [] - ) - - # Monitored counts - mon_row = self.execute_query( - f"""SELECT - SUM(CASE WHEN monitored = 1 THEN 1 ELSE 0 END) as monitored, - SUM(CASE WHEN monitored = 0 THEN 1 ELSE 0 END) as unmonitored - FROM media_cache {where}""", - params, - fetch_one=True, - ) - monitored = { - "monitored": mon_row["monitored"] or 0 if mon_row else 0, - "unmonitored": mon_row["unmonitored"] or 0 if mon_row else 0, - } - - # By genre (Python-side aggregation since genre is JSON array) - genre_rows = ( - self.execute_query( - f"SELECT genre FROM media_cache {where + (' AND' if where else 'WHERE')} genre IS NOT NULL AND genre != ''", - params, - fetch_all=True, - ) - or [] - ) - genre_counts: dict = {} - for row in genre_rows: - raw = row.get("genre", "") - if not raw: - continue - parsed_genres = [] - try: - parsed = json.loads(raw) - if isinstance(parsed, list): - parsed_genres = [str(g).strip() for g in parsed if g] - except (json.JSONDecodeError, TypeError): - parsed_genres = [g.strip() for g in raw.split(",") if g.strip()] - for g in parsed_genres: - genre_counts[g] = genre_counts.get(g, 0) + 1 - by_genre = sorted( - [{"genre": k, "count": v} for k, v in genre_counts.items()], - key=lambda x: x["count"], - reverse=True, - ) - - # By root folder (where media lives on disk — *arr only; Plex has none) - by_root_folder = ( - self.execute_query( - f"""SELECT root_folder, COUNT(*) as count - FROM media_cache {where + (" AND" if where else "WHERE")} root_folder IS NOT NULL AND root_folder != '' - GROUP BY root_folder ORDER BY count DESC""", - params, - fetch_all=True, - ) - or [] - ) - - # By tag (Python-side aggregation since tags is a JSON array of names) - tag_rows = ( - self.execute_query( - f"SELECT tags FROM media_cache {where + (' AND' if where else 'WHERE')} tags IS NOT NULL AND tags != '' AND tags != '[]'", - params, - fetch_all=True, - ) - or [] - ) - tag_counts: dict = {} - for row in tag_rows: - raw = row.get("tags", "") - if not raw: - continue - try: - parsed = json.loads(raw) - tags = ( - [str(t).strip() for t in parsed if t] - if isinstance(parsed, list) - else [] - ) - except (json.JSONDecodeError, TypeError): - tags = [t.strip() for t in raw.split(",") if t.strip()] - for t in tags: - tag_counts[t] = tag_counts.get(t, 0) + 1 - by_tags = sorted( - [{"tag": k, "count": v} for k, v in tag_counts.items()], - key=lambda x: x["count"], - reverse=True, - ) - - recently_added = self.get_recently_added(asset_type=asset_type) - - return { - **base, - "by_instance": by_instance, - "recently_added": recently_added, - "by_root_folder": by_root_folder, - "by_tags": by_tags, - "by_status": by_status, - "by_language": by_language, - "by_rating": by_rating, - "by_studio": by_studio, - "by_decade": by_decade, - "by_runtime": by_runtime, - "by_genre": by_genre, - "monitored": monitored, - } - - def get_recently_added( - self, asset_type: Optional[str] = None, limit: int = 12 - ) -> dict: - """Most recently added library items, keyed on ``created_at`` (stamped - once on first insert = first-seen time). - - Rows that predate created_at stamping carry NULL and never appear here, - so this reflects genuinely new additions going forward — it is not a - backfill of the existing library. Per-item ``added_age_seconds`` uses - SQLite's clock (matching the snapshot-age fields) so the frontend never - has to parse a bare timestamp. - """ - conditions = ["created_at IS NOT NULL"] - params: list = [] - if asset_type and asset_type != "all": - conditions.append("asset_type = ?") - params.append(asset_type) - where = "WHERE " + " AND ".join(conditions) - - def _count(days: int) -> int: - row = self.execute_query( - f"SELECT COUNT(*) AS n FROM media_cache {where} " - "AND created_at >= datetime('now', ?)", - tuple(params + [f"-{days} days"]), - fetch_one=True, - ) - return (row["n"] or 0) if row else 0 - - items = ( - self.execute_query( - f"""SELECT title, asset_type, instance_name, source, year, - CAST(strftime('%s','now') - strftime('%s', created_at) AS REAL) - AS added_age_seconds - FROM media_cache {where} - ORDER BY created_at DESC LIMIT ?""", - tuple(params + [limit]), - fetch_all=True, - ) - or [] - ) - - return { - "last_7d": _count(7), - "last_30d": _count(30), - "items": items, - } - def get_distinct_genres(self, asset_type: Optional[str] = None) -> List[str]: """Extract unique genres from all media_cache entries.""" where = "" @@ -1608,59 +1160,6 @@ def find_low_rated( or [] ) - @staticmethod - def _empty_field_clauses(fields: List[str]) -> str: - """OR-joined "is null or blank" test for each field, INTEGER-aware.""" - # SQL mirror of is_missing_value — keep in sync. - return " OR ".join( - f"({f} IS NULL OR {f} = 0)" - if f in INCOMPLETE_METADATA_INT_FIELDS - else f"({f} IS NULL OR {f} = '')" - for f in fields - ) - - def find_incomplete_metadata( - self, fields: List[str], limit: int = 200, offset: int = 0 - ) -> list: - """Return rows missing any of `fields` that their asset_type can populate.""" - # Field names are interpolated into SQL — drop anything off the - # column allow-list before building the clause. - requested = [f for f in fields if f in INCOMPLETE_METADATA_FIELDS] - if not requested: - return [] - - # One clause per known asset_type, using only the fields that type can - # populate; a catch-all keeps unrecognised types matching on the lot. - subclauses = [] - params: list = [] - for atype in _KNOWN_ASSET_TYPES: - never = NEVER_POPULATED_FIELDS.get(atype, set()) - effective = [f for f in requested if f not in never] - if not effective: - continue - subclauses.append( - f"(asset_type = ? AND ({self._empty_field_clauses(effective)}))" - ) - params.append(atype) - - unknown_placeholders = ",".join(["?"] * len(_KNOWN_ASSET_TYPES)) - subclauses.append( - f"(asset_type NOT IN ({unknown_placeholders}) " - f"AND ({self._empty_field_clauses(requested)}))" - ) - params.extend(_KNOWN_ASSET_TYPES) - - where = " OR ".join(subclauses) - return ( - self.execute_query( - f"SELECT * FROM media_cache WHERE {where} " - "ORDER BY title ASC, id ASC LIMIT ? OFFSET ?", - (*params, limit, offset), - fetch_all=True, - ) - or [] - ) - def get_arr_linked(self) -> list: """Return the identity/location fields of every row sourced from an *arr.""" return ( @@ -1701,42 +1200,6 @@ def find_by_original_file_basename(self, basename: str) -> list: or [] ) - def record_edit( - self, - media_id: int, - edited_at: str, - edited_by: str, - field: str, - old_value: Any, - new_value: Any, - ) -> None: - """Append one field's old→new change to the media edit audit trail.""" - self.execute_query( - "INSERT INTO media_edit_history " - "(media_id, edited_at, edited_by, field, old_value, new_value) " - "VALUES (?, ?, ?, ?, ?, ?)", - ( - media_id, - edited_at, - edited_by, - field, - None if old_value is None else str(old_value), - None if new_value is None else str(new_value), - ), - ) - - def get_edit_history(self, media_id: int, limit: int = 100) -> list: - """Return one media item's edit audit trail, newest first.""" - return ( - self.execute_query( - "SELECT * FROM media_edit_history WHERE media_id=? " - "ORDER BY edited_at DESC, id DESC LIMIT ?", - (media_id, limit), - fetch_all=True, - ) - or [] - ) - def sync_for_instance( self, instance_name: str, diff --git a/backend/util/database/media_edit_history.py b/backend/util/database/media_edit_history.py new file mode 100644 index 00000000..307b0131 --- /dev/null +++ b/backend/util/database/media_edit_history.py @@ -0,0 +1,45 @@ +"""media_edit_history table access, mixed into MediaCache.""" + +from typing import Any + +from .db_base import DatabaseBase + + +class EditHistoryMixin(DatabaseBase): + """Append-and-read access to the per-media edit audit trail.""" + + def record_edit( + self, + media_id: int, + edited_at: str, + edited_by: str, + field: str, + old_value: Any, + new_value: Any, + ) -> None: + """Append one field's old→new change to the media edit audit trail.""" + self.execute_query( + "INSERT INTO media_edit_history " + "(media_id, edited_at, edited_by, field, old_value, new_value) " + "VALUES (?, ?, ?, ?, ?, ?)", + ( + media_id, + edited_at, + edited_by, + field, + None if old_value is None else str(old_value), + None if new_value is None else str(new_value), + ), + ) + + def get_edit_history(self, media_id: int, limit: int = 100) -> list: + """Return one media item's edit audit trail, newest first.""" + return ( + self.execute_query( + "SELECT * FROM media_edit_history WHERE media_id=? " + "ORDER BY edited_at DESC, id DESC LIMIT ?", + (media_id, limit), + fetch_all=True, + ) + or [] + ) diff --git a/backend/util/database/media_metadata.py b/backend/util/database/media_metadata.py new file mode 100644 index 00000000..df0aff0a --- /dev/null +++ b/backend/util/database/media_metadata.py @@ -0,0 +1,109 @@ +"""Metadata-completeness rules for media_cache, mixed into MediaCache.""" + +from typing import List + +from .db_base import DatabaseBase + +# Columns find_incomplete_metadata may test, and the subset stored as INTEGER +# (they can't hold '', so they compare against NULL/0 instead). +INCOMPLETE_METADATA_FIELDS = frozenset( + { + "rating", + "studio", + "language", + "genre", + "runtime", + "edition", + "tmdb_id", + "tvdb_id", + "imdb_id", + "year", + } +) +INCOMPLETE_METADATA_INT_FIELDS = frozenset({"tmdb_id", "tvdb_id", "runtime"}) + + +def is_missing_value(field: str, value) -> bool: + """True when `value` is missing for `field` (INT fields: None/0; else None/'').""" + if field in INCOMPLETE_METADATA_INT_FIELDS: + return value is None or value == 0 + return value is None or value == "" + + +# Fields the ARR normalize layer never populates for a given asset_type, so +# flagging them as "missing" is a false positive. Radarr has no tvdbId, +# Sonarr has no tmdbId, Lidarr (artist) uses MusicBrainz IDs and leaves +# tmdb/tvdb/imdb + rating/runtime/language/edition as None by design. +NEVER_POPULATED_FIELDS = { + "movie": {"tvdb_id"}, + "show": {"tmdb_id"}, + "artist": { + "tmdb_id", + "tvdb_id", + "imdb_id", + "rating", + "runtime", + "language", + "edition", + }, +} +# asset_types with their own expected-field rules; anything else matches +# find_incomplete_metadata's catch-all clause. +_KNOWN_ASSET_TYPES = ("movie", "show", "artist") + + +class MetadataCompletenessMixin(DatabaseBase): + """Queries for media_cache rows missing metadata their asset_type can hold.""" + + @staticmethod + def _empty_field_clauses(fields: List[str]) -> str: + """OR-joined "is null or blank" test for each field, INTEGER-aware.""" + # SQL mirror of is_missing_value — keep in sync. + return " OR ".join( + f"({f} IS NULL OR {f} = 0)" + if f in INCOMPLETE_METADATA_INT_FIELDS + else f"({f} IS NULL OR {f} = '')" + for f in fields + ) + + def find_incomplete_metadata( + self, fields: List[str], limit: int = 200, offset: int = 0 + ) -> list: + """Return rows missing any of `fields` that their asset_type can populate.""" + # Field names are interpolated into SQL — drop anything off the + # column allow-list before building the clause. + requested = [f for f in fields if f in INCOMPLETE_METADATA_FIELDS] + if not requested: + return [] + + # One clause per known asset_type, using only the fields that type can + # populate; a catch-all keeps unrecognised types matching on the lot. + subclauses = [] + params: list = [] + for atype in _KNOWN_ASSET_TYPES: + never = NEVER_POPULATED_FIELDS.get(atype, set()) + effective = [f for f in requested if f not in never] + if not effective: + continue + subclauses.append( + f"(asset_type = ? AND ({self._empty_field_clauses(effective)}))" + ) + params.append(atype) + + unknown_placeholders = ",".join(["?"] * len(_KNOWN_ASSET_TYPES)) + subclauses.append( + f"(asset_type NOT IN ({unknown_placeholders}) " + f"AND ({self._empty_field_clauses(requested)}))" + ) + params.extend(_KNOWN_ASSET_TYPES) + + where = " OR ".join(subclauses) + return ( + self.execute_query( + f"SELECT * FROM media_cache WHERE {where} " + "ORDER BY title ASC, id ASC LIMIT ? OFFSET ?", + (*params, limit, offset), + fetch_all=True, + ) + or [] + ) diff --git a/backend/util/database/media_stats.py b/backend/util/database/media_stats.py new file mode 100644 index 00000000..f94b154a --- /dev/null +++ b/backend/util/database/media_stats.py @@ -0,0 +1,414 @@ +"""Library-health statistics over media_cache, mixed into MediaCache.""" + +import json +from typing import Optional + +from .db_base import DatabaseBase + +# Library-health SQL fragments, shared across the stats queries so by_type, +# totals and by_instance can't drift. Everything is counted in CONTENT UNITS — +# a movie, an album, or a single TV episode — to match each *arr's native unit: +# units = total content units +# in_library = units whose file is present +# missing = monitored units that are released/aired but have no file +# upcoming = monitored units not yet released/aired +# A Sonarr SEASON row expands to its episode counts; a movie/album is 1 unit; +# shows, seasons-as-rows... no — shows and artists are CONTAINERS (0 units; the +# episodes/albums carry the counts), so they fall through to 0 automatically. +# "released" gates simple (movie/album) units: Lidarr albums use their own +# release_date (absent = released), movies/shows use the *arr `status` (mirrors +# release_readiness.UNRELEASED_STATUSES). For seasons, "aired" IS the gate: +# missing = aired-on-disk shortfall, upcoming = not-yet-aired. +_RELEASED_SQL = ( + "CASE WHEN asset_type='album' " + "THEN (release_date IS NULL OR release_date <= date('now')) " + "ELSE (status IS NULL OR status NOT IN " + "('announced','deleted','tba','upcoming')) END" +) +_IS_SEASON = "(asset_type='show' AND season_number IS NOT NULL)" +_IS_SIMPLE = "asset_type IN ('movie','album')" # 1-unit content types + +_UNITS_TOTAL_SQL = ( + f"SUM(CASE WHEN {_IS_SEASON} THEN COALESCE(total_episodes,0) " + f"WHEN {_IS_SIMPLE} THEN 1 ELSE 0 END)" +) +# An album under an UNMONITORED artist isn't "wanted" — Lidarr's Wanted page +# excludes it, so missing/upcoming do too. NULL (pre-column rows) = monitored. +_ALBUM_ARTIST_OK = "(asset_type != 'album' OR COALESCE(artist_monitored, 1) = 1)" +_IN_LIBRARY_SQL = ( + f"SUM(CASE WHEN {_IS_SEASON} THEN COALESCE(episode_files,0) " + f"WHEN {_IS_SIMPLE} AND has_content=1 THEN 1 ELSE 0 END)" +) +_MISSING_SQL = ( + f"SUM(CASE WHEN {_IS_SEASON} AND monitored=1 " + "THEN MAX(0, COALESCE(aired_episodes,0) - COALESCE(episode_files,0)) " + f"WHEN {_IS_SIMPLE} AND monitored=1 AND COALESCE(has_content,0)=0 " + f"AND {_RELEASED_SQL} AND {_ALBUM_ARTIST_OK} THEN 1 ELSE 0 END)" +) +_UPCOMING_SQL = ( + f"SUM(CASE WHEN {_IS_SEASON} AND monitored=1 " + "THEN MAX(0, COALESCE(total_episodes,0) - COALESCE(aired_episodes,0)) " + f"WHEN {_IS_SIMPLE} AND monitored=1 AND COALESCE(has_content,0)=0 " + f"AND NOT ({_RELEASED_SQL}) AND {_ALBUM_ARTIST_OK} THEN 1 ELSE 0 END)" +) +# Row-count fragments (containers): for the artist/show context cards. +_MONITORED_SQL = "SUM(CASE WHEN monitored=1 THEN 1 ELSE 0 END)" +_SHOW_COUNT_SQL = ( + "SUM(CASE WHEN asset_type='show' AND season_number IS NULL THEN 1 ELSE 0 END)" +) +_MONITORED_SHOWS_SQL = ( + "SUM(CASE WHEN asset_type='show' AND season_number IS NULL AND monitored=1 " + "THEN 1 ELSE 0 END)" +) +_SEASON_COUNT_SQL = ( + "SUM(CASE WHEN asset_type='show' AND season_number IS NOT NULL THEN 1 ELSE 0 END)" +) +_ARTIST_COUNT_SQL = "SUM(CASE WHEN asset_type='artist' THEN 1 ELSE 0 END)" +_MONITORED_ARTISTS_SQL = ( + "SUM(CASE WHEN asset_type='artist' AND monitored=1 THEN 1 ELSE 0 END)" +) + + +class StatsMixin(DatabaseBase): + """Aggregate library-health statistics over the media_cache table.""" + + def get_stats( + self, asset_type: Optional[str] = None, period_days: int = None + ) -> dict: + """Aggregate statistics from media_cache.""" + conditions = [] + params: list = [] + if asset_type and asset_type != "all": + conditions.append("asset_type = ?") + params.append(asset_type) + if period_days: + conditions.append("created_at >= datetime('now', ?)") + params.append(f"-{period_days} days") + + where = ("WHERE " + " AND ".join(conditions)) if conditions else "" + params = tuple(params) + + # Library-health metrics — see the module-level _*_SQL fragments. + rows = ( + self.execute_query( + f""" + SELECT asset_type, + COUNT(*) as total, + {_UNITS_TOTAL_SQL} as units, + {_IN_LIBRARY_SQL} as in_library, + {_MISSING_SQL} as missing, + {_UPCOMING_SQL} as upcoming, + {_MONITORED_SQL} as monitored, + {_SHOW_COUNT_SQL} as show_count, + {_MONITORED_SHOWS_SQL} as monitored_shows, + {_SEASON_COUNT_SQL} as season_count, + COUNT(DISTINCT instance_name) as instances + FROM media_cache {where} + GROUP BY asset_type + """, + params, + fetch_all=True, + ) + or [] + ) + + totals = self.execute_query( + f""" + SELECT {_UNITS_TOTAL_SQL} as total, + {_IN_LIBRARY_SQL} as in_library, + {_MISSING_SQL} as missing, + {_UPCOMING_SQL} as upcoming, + {_MONITORED_SQL} as monitored + FROM media_cache {where} + """, + params, + fetch_one=True, + ) + + return { + "by_type": rows, + # Headline numbers are in content units (movies + episodes + albums). + "total": (totals["total"] or 0) if totals else 0, + "in_library": (totals["in_library"] or 0) if totals else 0, + "missing": (totals["missing"] or 0) if totals else 0, + "upcoming": (totals["upcoming"] or 0) if totals else 0, + "monitored": (totals["monitored"] or 0) if totals else 0, + } + + def get_detailed_stats( + self, asset_type: Optional[str] = None, period_days: int = None + ) -> dict: + """Extended statistics with breakdowns by multiple dimensions.""" + conditions = [] + params_list: list = [] + if asset_type and asset_type != "all": + conditions.append("asset_type = ?") + params_list.append(asset_type) + if period_days: + conditions.append("created_at >= datetime('now', ?)") + params_list.append(f"-{period_days} days") + + where = ("WHERE " + " AND ".join(conditions)) if conditions else "" + params = tuple(params_list) + + # Base stats (same as get_stats) + base = self.get_stats(asset_type=asset_type, period_days=period_days) + + # By instance (source = service type: radarr/sonarr/lidarr/plex) + by_instance = ( + self.execute_query( + f"""SELECT instance_name, source, COUNT(*) as total, + {_UNITS_TOTAL_SQL} as units, + {_IN_LIBRARY_SQL} as in_library, + {_MISSING_SQL} as missing, + {_UPCOMING_SQL} as upcoming, + {_MONITORED_SQL} as monitored, + {_SHOW_COUNT_SQL} as show_count, + {_MONITORED_SHOWS_SQL} as monitored_shows, + {_SEASON_COUNT_SQL} as season_count, + {_ARTIST_COUNT_SQL} as artist_count, + {_MONITORED_ARTISTS_SQL} as monitored_artists + FROM media_cache {where} + GROUP BY instance_name, source ORDER BY total DESC""", + params, + fetch_all=True, + ) + or [] + ) + + # By status + by_status = ( + self.execute_query( + f"""SELECT status, COUNT(*) as count + FROM media_cache {where + (" AND" if where else "WHERE")} status IS NOT NULL AND status != '' + GROUP BY status ORDER BY count DESC""", + params, + fetch_all=True, + ) + or [] + ) + + # By language + by_language = ( + self.execute_query( + f"""SELECT language, COUNT(*) as count + FROM media_cache {where + (" AND" if where else "WHERE")} language IS NOT NULL AND language != '' + GROUP BY language ORDER BY count DESC""", + params, + fetch_all=True, + ) + or [] + ) + + # By rating (content ratings like PG, R, TV-MA) + by_rating = ( + self.execute_query( + f"""SELECT rating, COUNT(*) as count + FROM media_cache {where + (" AND" if where else "WHERE")} rating IS NOT NULL AND rating != '' + GROUP BY rating ORDER BY count DESC""", + params, + fetch_all=True, + ) + or [] + ) + + # By studio (top 50) + by_studio = ( + self.execute_query( + f"""SELECT studio, COUNT(*) as count + FROM media_cache {where + (" AND" if where else "WHERE")} studio IS NOT NULL AND studio != '' + GROUP BY studio ORDER BY count DESC LIMIT 50""", + params, + fetch_all=True, + ) + or [] + ) + + # By year/decade + by_decade = ( + self.execute_query( + f"""SELECT (CAST(year AS INTEGER) / 10) * 10 as decade, COUNT(*) as count + FROM media_cache {where + (" AND" if where else "WHERE")} year IS NOT NULL AND year != '' + GROUP BY decade ORDER BY decade DESC""", + params, + fetch_all=True, + ) + or [] + ) + by_decade = [ + {"decade": f"{r['decade']}s", "count": r["count"]} + for r in by_decade + if r.get("decade") + ] + + # By runtime buckets + by_runtime = ( + self.execute_query( + f"""SELECT + CASE + WHEN CAST(runtime AS INTEGER) < 30 THEN 'Under 30m' + WHEN CAST(runtime AS INTEGER) < 60 THEN '30-60m' + WHEN CAST(runtime AS INTEGER) < 90 THEN '60-90m' + WHEN CAST(runtime AS INTEGER) < 120 THEN '90-120m' + WHEN CAST(runtime AS INTEGER) < 150 THEN '120-150m' + ELSE '150m+' + END as bucket, + COUNT(*) as count + FROM media_cache {where + (" AND" if where else "WHERE")} runtime IS NOT NULL AND runtime != '' AND CAST(runtime AS INTEGER) > 0 + GROUP BY bucket ORDER BY MIN(CAST(runtime AS INTEGER))""", + params, + fetch_all=True, + ) + or [] + ) + + # Monitored counts + mon_row = self.execute_query( + f"""SELECT + SUM(CASE WHEN monitored = 1 THEN 1 ELSE 0 END) as monitored, + SUM(CASE WHEN monitored = 0 THEN 1 ELSE 0 END) as unmonitored + FROM media_cache {where}""", + params, + fetch_one=True, + ) + monitored = { + "monitored": mon_row["monitored"] or 0 if mon_row else 0, + "unmonitored": mon_row["unmonitored"] or 0 if mon_row else 0, + } + + # By genre (Python-side aggregation since genre is JSON array) + genre_rows = ( + self.execute_query( + f"SELECT genre FROM media_cache {where + (' AND' if where else 'WHERE')} genre IS NOT NULL AND genre != ''", + params, + fetch_all=True, + ) + or [] + ) + genre_counts: dict = {} + for row in genre_rows: + raw = row.get("genre", "") + if not raw: + continue + parsed_genres = [] + try: + parsed = json.loads(raw) + if isinstance(parsed, list): + parsed_genres = [str(g).strip() for g in parsed if g] + except (json.JSONDecodeError, TypeError): + parsed_genres = [g.strip() for g in raw.split(",") if g.strip()] + for g in parsed_genres: + genre_counts[g] = genre_counts.get(g, 0) + 1 + by_genre = sorted( + [{"genre": k, "count": v} for k, v in genre_counts.items()], + key=lambda x: x["count"], + reverse=True, + ) + + # By root folder (where media lives on disk — *arr only; Plex has none) + by_root_folder = ( + self.execute_query( + f"""SELECT root_folder, COUNT(*) as count + FROM media_cache {where + (" AND" if where else "WHERE")} root_folder IS NOT NULL AND root_folder != '' + GROUP BY root_folder ORDER BY count DESC""", + params, + fetch_all=True, + ) + or [] + ) + + # By tag (Python-side aggregation since tags is a JSON array of names) + tag_rows = ( + self.execute_query( + f"SELECT tags FROM media_cache {where + (' AND' if where else 'WHERE')} tags IS NOT NULL AND tags != '' AND tags != '[]'", + params, + fetch_all=True, + ) + or [] + ) + tag_counts: dict = {} + for row in tag_rows: + raw = row.get("tags", "") + if not raw: + continue + try: + parsed = json.loads(raw) + tags = ( + [str(t).strip() for t in parsed if t] + if isinstance(parsed, list) + else [] + ) + except (json.JSONDecodeError, TypeError): + tags = [t.strip() for t in raw.split(",") if t.strip()] + for t in tags: + tag_counts[t] = tag_counts.get(t, 0) + 1 + by_tags = sorted( + [{"tag": k, "count": v} for k, v in tag_counts.items()], + key=lambda x: x["count"], + reverse=True, + ) + + recently_added = self.get_recently_added(asset_type=asset_type) + + return { + **base, + "by_instance": by_instance, + "recently_added": recently_added, + "by_root_folder": by_root_folder, + "by_tags": by_tags, + "by_status": by_status, + "by_language": by_language, + "by_rating": by_rating, + "by_studio": by_studio, + "by_decade": by_decade, + "by_runtime": by_runtime, + "by_genre": by_genre, + "monitored": monitored, + } + + def get_recently_added( + self, asset_type: Optional[str] = None, limit: int = 12 + ) -> dict: + """Most recently added library items, keyed on ``created_at`` (stamped + once on first insert = first-seen time). + + Rows that predate created_at stamping carry NULL and never appear here, + so this reflects genuinely new additions going forward — it is not a + backfill of the existing library. Per-item ``added_age_seconds`` uses + SQLite's clock (matching the snapshot-age fields) so the frontend never + has to parse a bare timestamp. + """ + conditions = ["created_at IS NOT NULL"] + params: list = [] + if asset_type and asset_type != "all": + conditions.append("asset_type = ?") + params.append(asset_type) + where = "WHERE " + " AND ".join(conditions) + + def _count(days: int) -> int: + row = self.execute_query( + f"SELECT COUNT(*) AS n FROM media_cache {where} " + "AND created_at >= datetime('now', ?)", + tuple(params + [f"-{days} days"]), + fetch_one=True, + ) + return (row["n"] or 0) if row else 0 + + items = ( + self.execute_query( + f"""SELECT title, asset_type, instance_name, source, year, + CAST(strftime('%s','now') - strftime('%s', created_at) AS REAL) + AS added_age_seconds + FROM media_cache {where} + ORDER BY created_at DESC LIMIT ?""", + tuple(params + [limit]), + fetch_all=True, + ) + or [] + ) + + return { + "last_7d": _count(7), + "last_30d": _count(30), + "items": items, + } From 0fccc970bccb4b60468203ec039bb65d90532e3d Mon Sep 17 00:00:00 2001 From: chodeus Date: Fri, 14 Aug 2026 09:03:23 +0800 Subject: [PATCH 2/5] refactor(api): system/jobs/modules/instances SQL moves behind interfaces MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit backend/api holds zero raw SQL now. 23 sites: system.py's maintenance and health queries get DbMaintenance and SystemHealth owners; job queries land on DBWorker (table-allowlist guarded); counts move to their cache owners. The artwork-matches reset now reports the DELETE's own rowcount — the old pre-count claimed user-locked rows as deleted while preserving them. datetime.utcnow() leaves the repo (4 sites, now(timezone.utc); cutoff comparisons proven byte-equivalent against both stored shapes). VACUUM and ping get honest owners; every ORDER BY under LIMIT carries an id tiebreaker. --- backend/api/instances.py | 7 +- backend/api/jobs.py | 14 +- backend/api/media_api.py | 8 +- backend/api/modules.py | 5 +- backend/api/system.py | 217 ++------- backend/util/database/__init__.py | 14 + backend/util/database/collection_cache.py | 8 + backend/util/database/maintenance.py | 104 ++++ backend/util/database/media_asset_matches.py | 19 +- backend/util/database/media_cache.py | 9 + backend/util/database/media_stats.py | 17 + backend/util/database/poster_cache.py | 6 +- backend/util/database/system_health.py | 52 ++ backend/util/database/worker.py | 68 ++- backend/util/notification.py | 4 +- tests/test_media_api_queries.py | 3 +- tests/test_system_queries.py | 485 +++++++++++++++++++ 17 files changed, 815 insertions(+), 225 deletions(-) create mode 100644 backend/util/database/maintenance.py create mode 100644 backend/util/database/system_health.py create mode 100644 tests/test_system_queries.py diff --git a/backend/api/instances.py b/backend/api/instances.py index 3f353c03..9126f5a1 100755 --- a/backend/api/instances.py +++ b/backend/api/instances.py @@ -1681,12 +1681,7 @@ async def get_instance_stats( else: # Count rows cheaply rather than loading the whole library to len() # it (a Lidarr instance is tens of thousands of rows). - row = db.media.execute_query( - "SELECT COUNT(*) AS n FROM media_cache WHERE instance_name=?", - (instance_id,), - fetch_one=True, - ) - total = row["n"] if row else 0 + total = db.media.count_by_instance(instance_id) # ARR freshness comes from sync_state (written when the ARR sync # completes), not media_cache.updated_at — the latter only moves on # changed rows, so it would read "stale" right after a fresh sync of diff --git a/backend/api/jobs.py b/backend/api/jobs.py index 853f0ac2..80fdee9f 100755 --- a/backend/api/jobs.py +++ b/backend/api/jobs.py @@ -142,20 +142,12 @@ async def list_webhook_origins( Summarize webhook jobs in the last N days by origin (client_host + endpoint) and status. Helpful for spotting a noisy Sonarr instance or a dead webhook. """ - from datetime import datetime, timedelta + from datetime import datetime, timedelta, timezone try: days = max(1, min(days, 90)) - cutoff = (datetime.utcnow() - timedelta(days=days)).isoformat() - rows = ( - db.worker.execute_query( - "SELECT id, payload, status, received_at FROM jobs " - "WHERE type='webhook' AND received_at >= ? ORDER BY received_at DESC", - (cutoff,), - fetch_all=True, - ) - or [] - ) + cutoff = (datetime.now(timezone.utc) - timedelta(days=days)).isoformat() + rows = db.worker.jobs_of_type_since("webhook", cutoff) from collections import Counter by_origin: Counter = Counter() diff --git a/backend/api/media_api.py b/backend/api/media_api.py index 2c40e41d..dd5e7125 100644 --- a/backend/api/media_api.py +++ b/backend/api/media_api.py @@ -1709,10 +1709,10 @@ async def update_media_metadata( # Audit trail: capture each field's old→new so reverts and diffs # are possible later without having to replay the whole edit stream. - from datetime import datetime as _dt + from datetime import datetime as _dt, timezone as _tz edited_by = getattr(request.state, "user", "") or "unknown" - now_iso = _dt.utcnow().isoformat() + now_iso = _dt.now(_tz.utc).isoformat() for field, new_value in update_kwargs.items(): old_value = item.get(field) if old_value == new_value: @@ -2000,9 +2000,9 @@ async def generate_collection_from_tag( {"created": False, "matched_media": len(media_rows)}, ) - from datetime import datetime as _dt + from datetime import datetime as _dt, timezone as _tz - created_at = _dt.utcnow().isoformat() + created_at = _dt.now(_tz.utc).isoformat() coll_id = db.poster.create_collection( name, f"Auto-generated from tag '{tag}'", created_at ) diff --git a/backend/api/modules.py b/backend/api/modules.py index 5ca8ac5c..a59a4d3a 100755 --- a/backend/api/modules.py +++ b/backend/api/modules.py @@ -1289,10 +1289,7 @@ async def cancel_module_execution( if request_cancellation(job_id): # Update job status to reflect cancellation is in progress now = datetime.now(timezone.utc).isoformat() - db.worker.execute_query( - "UPDATE jobs SET status='cancelled', completed_at=? WHERE id=? AND status='running'", - (now, job_id), - ) + db.worker.cancel_running_job(job_id, now) logger.info(f"Cancellation requested for module {name} job {job_id}") return ok( f"Cancellation requested for module {name} job {job_id}", diff --git a/backend/api/system.py b/backend/api/system.py index 72ece827..6196c928 100755 --- a/backend/api/system.py +++ b/backend/api/system.py @@ -9,10 +9,9 @@ import json import os import shutil -import sqlite3 import time import zipfile -from datetime import datetime +from datetime import datetime, timezone from pathlib import Path from typing import Any, List, Optional @@ -177,7 +176,7 @@ async def health_check(request: Request) -> JSONResponse: db = getattr(request.app.state, "db", None) if db: try: - db.worker.execute_query("SELECT 1") + db.maintenance.ping() checks["database"] = "ok" except Exception: checks["database"] = "error" @@ -755,22 +754,10 @@ async def get_health_snapshots( logger: Any = Depends(get_logger), db: ChubDB = Depends(get_database), ) -> JSONResponse: + """Return the newest scheduler health snapshots, optionally for one instance.""" try: limit = max(1, min(limit, 500)) - if instance: - rows = db.worker.execute_query( - "SELECT * FROM system_health_snapshots WHERE instance_name=? " - "ORDER BY snapshot_at DESC LIMIT ?", - (instance, limit), - fetch_all=True, - ) - else: - rows = db.worker.execute_query( - "SELECT * FROM system_health_snapshots ORDER BY snapshot_at DESC LIMIT ?", - (limit,), - fetch_all=True, - ) - snaps = [dict(r) for r in rows or []] + snaps = db.system_health.recent_snapshots(limit=limit, instance=instance) return ok(f"Retrieved {len(snaps)} snapshots", {"snapshots": snaps}) except Exception as e: logger.error(f"Error fetching health snapshots: {e}") @@ -793,40 +780,21 @@ async def get_system_digest( logger: Any = Depends(get_logger), db: ChubDB = Depends(get_database), ) -> JSONResponse: + """Aggregate media/job/health activity over the last N days.""" from datetime import timedelta try: days = max(1, min(days, 90)) - cutoff = (datetime.utcnow() - timedelta(days=days)).isoformat() + cutoff = (datetime.now(timezone.utc) - timedelta(days=days)).isoformat() - # Media added in window - media_row = db.worker.execute_query( - "SELECT COUNT(*) AS total FROM media_cache WHERE created_at >= ?", - (cutoff,), - fetch_one=True, - ) - media_added = media_row["total"] if media_row else 0 + media_added = db.media.count_added_since(cutoff) + job_counts = db.worker.count_by_status_since(cutoff) + failed_runs = db.worker.recent_failures(cutoff, limit=20) - # Job stats in window - job_rows = db.worker.execute_query( - "SELECT status, COUNT(*) AS total FROM jobs WHERE received_at >= ? GROUP BY status", - (cutoff,), - fetch_all=True, - ) - job_counts = {r["status"]: r["total"] for r in job_rows or []} - - # Failed module runs in window, with module names - failed_runs = db.worker.execute_query( - "SELECT id, type, payload, error, received_at FROM jobs " - "WHERE status='error' AND received_at >= ? " - "ORDER BY received_at DESC LIMIT 20", - (cutoff,), - fetch_all=True, - ) recent_failures = [] import json as _json - for r in failed_runs or []: + for r in failed_runs: payload = r["payload"] try: payload = _json.loads(payload) if isinstance(payload, str) else payload @@ -842,21 +810,7 @@ async def get_system_digest( } ) - # Latest health per instance - health_rows = db.worker.execute_query( - """ - SELECT s.instance_name, s.service, s.status, s.response_time_ms, - s.status_code, s.snapshot_at, s.error - FROM system_health_snapshots s - INNER JOIN ( - SELECT instance_name, MAX(snapshot_at) AS latest - FROM system_health_snapshots GROUP BY instance_name - ) latest ON s.instance_name = latest.instance_name - AND s.snapshot_at = latest.latest - """, - fetch_all=True, - ) - latest_health = [dict(r) for r in health_rows or []] + latest_health = db.system_health.latest_per_instance() return ok( f"Digest for last {days}d", @@ -887,29 +841,14 @@ async def get_cleanup_candidates( logger: Any = Depends(get_logger), db: ChubDB = Depends(get_database), ) -> JSONResponse: + """Report counts of items worth cleaning up; read-only, no mutations.""" try: - old_jobs_row = db.worker.execute_query( - "SELECT COUNT(*) AS total FROM jobs WHERE status='error'", - fetch_one=True, - ) - unmatched_media_row = db.worker.execute_query( - "SELECT COUNT(*) AS total FROM media_cache WHERE matched=0", - fetch_one=True, - ) - unmatched_coll_row = db.worker.execute_query( - "SELECT COUNT(*) AS total FROM collections_cache WHERE matched=0", - fetch_one=True, - ) return ok( "Cleanup candidates", { - "errored_jobs": old_jobs_row["total"] if old_jobs_row else 0, - "unmatched_media": unmatched_media_row["total"] - if unmatched_media_row - else 0, - "unmatched_collections": unmatched_coll_row["total"] - if unmatched_coll_row - else 0, + "errored_jobs": db.worker.count_by_status("error"), + "unmatched_media": db.media.count_unmatched(), + "unmatched_collections": db.collection.count_unmatched(), }, ) except Exception as e: @@ -921,29 +860,6 @@ async def get_cleanup_candidates( ) -# Tables listed in db-stats / vacuum responses. Kept explicit instead of -# enumerating sqlite_master so we don't surface internal SQLite tables -# (sqlite_sequence, sqlite_stat1) that users have no reason to see. -_DB_STATS_TABLES = ( - "media_cache", - "poster_cache", - "collections_cache", - "plex_media_cache", - "jobs", - "webhook_cache", - "gdrive_stats", - "scan_cache", - "system_health_snapshots", - "media_edit_history", - "upgradinatorr_progress", - "poster_collections", - "poster_collection_items", - "holiday_status", - "run_state", - "schema_migrations", -) - - @router.get( "/system/db-stats", summary="Database statistics", @@ -954,64 +870,23 @@ async def get_db_stats( logger: Any = Depends(get_logger), db: ChubDB = Depends(get_database), ) -> JSONResponse: + """Return per-table row counts, SQLite page stats and the migration log.""" try: logger.debug("Serving GET /api/system/db-stats") - tables = [] - existing = { - row["name"] - for row in ( - db.worker.execute_query( - "SELECT name FROM sqlite_master WHERE type='table'", - fetch_all=True, - ) - or [] - ) - } - for name in _DB_STATS_TABLES: - if name not in existing: - continue - count_row = db.worker.execute_query( - f"SELECT COUNT(*) AS total FROM {name}", fetch_one=True - ) - tables.append( - {"name": name, "rows": count_row["total"] if count_row else 0} - ) - - page_size_row = db.worker.execute_query("PRAGMA page_size", fetch_one=True) - page_count_row = db.worker.execute_query("PRAGMA page_count", fetch_one=True) - freelist_row = db.worker.execute_query("PRAGMA freelist_count", fetch_one=True) - page_size = int(page_size_row["page_size"]) if page_size_row else 0 - page_count = int(page_count_row["page_count"]) if page_count_row else 0 - freelist_count = int(freelist_row["freelist_count"]) if freelist_row else 0 - total_bytes = page_size * page_count - free_bytes = page_size * freelist_count + pages = db.maintenance.page_stats() try: file_bytes = os.path.getsize(db.db_path) except OSError: - file_bytes = total_bytes - - migrations = [] - if "schema_migrations" in existing: - migrations = ( - db.worker.execute_query( - "SELECT name, applied_at FROM schema_migrations ORDER BY applied_at DESC, name DESC", - fetch_all=True, - ) - or [] - ) + file_bytes = pages["total_bytes"] return ok( "Database statistics", { - "tables": tables, - "page_size": page_size, - "page_count": page_count, - "freelist_count": freelist_count, - "total_bytes": total_bytes, - "free_bytes": free_bytes, + "tables": db.maintenance.table_row_counts(), + **pages, "file_bytes": file_bytes, - "schema_migrations": migrations, + "schema_migrations": db.maintenance.list_migrations(), }, ) except Exception as e: @@ -1034,6 +909,7 @@ def vacuum_database( logger: Any = Depends(get_logger), db: ChubDB = Depends(get_database), ) -> JSONResponse: + """Run SQLite VACUUM and report the bytes it reclaimed.""" try: logger.debug("Serving POST /api/system/db/vacuum") try: @@ -1041,16 +917,8 @@ def vacuum_database( except OSError: bytes_before = 0 - # VACUUM cannot run inside an explicit transaction and ignores the - # connection's normal isolation; bypass the worker's helper and use - # a raw sqlite3 connection scoped to this call. start = time.time() - conn = sqlite3.connect(db.db_path, timeout=60) - try: - conn.isolation_level = None - conn.execute("VACUUM") - finally: - conn.close() + db.maintenance.vacuum() duration_ms = int((time.time() - start) * 1000) try: @@ -1093,6 +961,7 @@ async def clear_poster_cache( logger: Any = Depends(get_logger), db: ChubDB = Depends(get_database), ) -> JSONResponse: + """Delete every poster_cache row so the next poster_renamerr run rescans.""" try: logger.debug("Serving POST /api/system/db/poster-cache/clear") # Serialize against poster_renamerr's clear()+rebuild+match critical @@ -1100,19 +969,14 @@ async def clear_poster_cache( from backend.modules.poster_renamerr import _POSTER_CACHE_REBUILD_LOCK with _POSTER_CACHE_REBUILD_LOCK: - count_row = db.worker.execute_query( - "SELECT COUNT(*) AS total FROM poster_cache", fetch_one=True - ) - before = int(count_row["total"]) if count_row else 0 - - db.poster.clear() + deleted = db.poster.clear() logger.info( - f"Wiped poster_cache via /api/system/db/poster-cache/clear ({before} rows)" + f"Wiped poster_cache via /api/system/db/poster-cache/clear ({deleted} rows)" ) return ok( "Poster cache cleared", - {"deleted": before}, + {"deleted": deleted}, ) except Exception as e: logger.error(f"Error clearing poster_cache: {e}") @@ -1137,20 +1001,16 @@ async def clear_artwork_matches( logger: Any = Depends(get_logger), db: ChubDB = Depends(get_database), ) -> JSONResponse: + """Delete every media_asset_matches row, ignores and locks included.""" try: logger.debug("Serving POST /api/system/db/artwork-matches/clear") - count_row = db.worker.execute_query( - "SELECT COUNT(*) AS total FROM media_asset_matches", fetch_one=True - ) - before = int(count_row["total"]) if count_row else 0 - - db.media_asset_matches.clear() + deleted = db.media_asset_matches.clear() logger.info( f"Wiped media_asset_matches via /api/system/db/artwork-matches/clear " - f"({before} rows)" + f"({deleted} rows)" ) - return ok("Artwork match state cleared", {"deleted": before}) + return ok("Artwork match state cleared", {"deleted": deleted}) except Exception as e: logger.error(f"Error clearing media_asset_matches: {e}") return error( @@ -1175,6 +1035,7 @@ async def reset_poster_matches( logger: Any = Depends(get_logger), db: ChubDB = Depends(get_database), ) -> JSONResponse: + """Clear poster-match state for media and collections, keeping curated rows.""" try: logger.debug("Serving POST /api/system/db/poster-matches/reset") media_reset = db.media.reset_match_state() @@ -1215,22 +1076,16 @@ async def reset_artwork_matches( logger: Any = Depends(get_logger), db: ChubDB = Depends(get_database), ) -> JSONResponse: + """Drop auto-matched artwork rows, preserving the user's ignores and locks.""" try: logger.debug("Serving POST /api/system/db/artwork-matches/reset") - count_row = db.worker.execute_query( - "SELECT COUNT(*) AS total FROM media_asset_matches " - "WHERE ignored IS NULL OR ignored = 0", - fetch_one=True, - ) - before = int(count_row["total"]) if count_row else 0 - - db.media_asset_matches.clear(keep_ignored=True) + deleted = db.media_asset_matches.clear(keep_ignored=True) logger.info( "Reset additional-artwork coverage via " - f"/api/system/db/artwork-matches/reset ({before} rows; ignores kept)" + f"/api/system/db/artwork-matches/reset ({deleted} rows; ignores kept)" ) - return ok("Artwork match coverage reset", {"deleted": before}) + return ok("Artwork match coverage reset", {"deleted": deleted}) except Exception as e: logger.error(f"Error resetting artwork match state: {e}") return error( diff --git a/backend/util/database/__init__.py b/backend/util/database/__init__.py index ad29b948..6d3145d2 100644 --- a/backend/util/database/__init__.py +++ b/backend/util/database/__init__.py @@ -11,6 +11,7 @@ from .collection_cache import CollectionCache from .db_base import DatabaseBase, escape_like from .holiday import HolidayStatus +from .maintenance import DbMaintenance from .media_asset_matches import MediaAssetMatches from .media_cache import MediaCache from .media_metadata import ( @@ -25,6 +26,7 @@ from .schema import SchemaManager from .stats import Stats from .sync_state import SyncState +from .system_health import SystemHealth from .tmdb_id_cache import TmdbIdCache from .tmdb_details_cache import TmdbDetailsCache from .tmdb_images_cache import TmdbImagesCache @@ -221,6 +223,16 @@ def sync_state(self) -> SyncState: """Access to per-instance last-completed-sync timestamps.""" return self._get_interface("sync_state", SyncState) + @property + def system_health(self) -> SystemHealth: + """Access to the scheduler's periodic instance-health snapshots.""" + return self._get_interface("system_health", SystemHealth) + + @property + def maintenance(self) -> DbMaintenance: + """Access to whole-file database operations (row counts, pages, VACUUM).""" + return self._get_interface("maintenance", DbMaintenance) + @property def holiday(self) -> HolidayStatus: """Access to holiday status operations.""" @@ -407,6 +419,7 @@ def my_db_operation(db): __all__ = [ "DatabaseBase", + "DbMaintenance", "SchemaManager", "PlexCache", "CollectionCache", @@ -414,6 +427,7 @@ def my_db_operation(db): "RunState", "Stats", "SyncState", + "SystemHealth", "ChubDB", "DBWorker", "HolidayStatus", diff --git a/backend/util/database/collection_cache.py b/backend/util/database/collection_cache.py index b4f800d0..154317fd 100755 --- a/backend/util/database/collection_cache.py +++ b/backend/util/database/collection_cache.py @@ -262,6 +262,14 @@ def set_match_provenance( (matched_at, matched_poster_file, id), ) + def count_unmatched(self) -> int: + """Rows with no poster match — the Unmatched page's collections figure.""" + row = self.execute_query( + "SELECT COUNT(*) AS total FROM collections_cache WHERE matched=0", + fetch_one=True, + ) + return int(row["total"]) if row else 0 + def clear(self) -> None: """Delete all rows from collections_cache.""" self.execute_query("DELETE FROM collections_cache") diff --git a/backend/util/database/maintenance.py b/backend/util/database/maintenance.py new file mode 100644 index 00000000..8e611a0a --- /dev/null +++ b/backend/util/database/maintenance.py @@ -0,0 +1,104 @@ +# util/database/maintenance.py + +import sqlite3 +from typing import Any, Dict, List, Set + +from .db_base import DatabaseBase + + +class DbMaintenance(DatabaseBase): + """Whole-file database operations: liveness, row counts, page stats, VACUUM. + + Everything here is about the SQLite file itself rather than any one table, + so it has no natural home on a per-table cache interface. + """ + + # Explicit allowlist: the interpolated COUNT(*) may only name a fixed + # literal, and internal SQLite tables stay hidden. + STATS_TABLES = ( + "media_cache", + "poster_cache", + "collections_cache", + "plex_media_cache", + "jobs", + "webhook_cache", + "gdrive_stats", + "scan_cache", + "system_health_snapshots", + "media_edit_history", + "upgradinatorr_progress", + "poster_collections", + "poster_collection_items", + "holiday_status", + "run_state", + "schema_migrations", + ) + + def ping(self) -> bool: + """Round-trip the simplest possible query; raises if the DB is unusable.""" + self.execute_query("SELECT 1", fetch_one=True) + return True + + def existing_tables(self) -> Set[str]: + """Names of every table currently present in the schema.""" + rows = ( + self.execute_query( + "SELECT name FROM sqlite_master WHERE type='table'", fetch_all=True + ) + or [] + ) + return {row["name"] for row in rows} + + def table_row_counts(self) -> List[Dict[str, Any]]: + """Row count per STATS_TABLES entry, skipping tables not in this schema.""" + existing = self.existing_tables() + counts = [] + for name in self.STATS_TABLES: + if name not in existing: + continue + # name comes from the STATS_TABLES literal tuple, never from a caller. + row = self.execute_query( + f"SELECT COUNT(*) AS total FROM {name}", # noqa: S608 + fetch_one=True, + ) + counts.append({"name": name, "rows": row["total"] if row else 0}) + return counts + + def page_stats(self) -> Dict[str, int]: + """SQLite page/freelist counters plus the byte totals derived from them.""" + page_size_row = self.execute_query("PRAGMA page_size", fetch_one=True) + page_count_row = self.execute_query("PRAGMA page_count", fetch_one=True) + freelist_row = self.execute_query("PRAGMA freelist_count", fetch_one=True) + page_size = int(page_size_row["page_size"]) if page_size_row else 0 + page_count = int(page_count_row["page_count"]) if page_count_row else 0 + freelist_count = int(freelist_row["freelist_count"]) if freelist_row else 0 + return { + "page_size": page_size, + "page_count": page_count, + "freelist_count": freelist_count, + "total_bytes": page_size * page_count, + "free_bytes": page_size * freelist_count, + } + + def list_migrations(self) -> List[Dict[str, Any]]: + """Applied schema migrations, newest first; empty when the table is absent.""" + if "schema_migrations" not in self.existing_tables(): + return [] + return ( + self.execute_query( + "SELECT name, applied_at FROM schema_migrations " + "ORDER BY applied_at DESC, name DESC", + fetch_all=True, + ) + or [] + ) + + def vacuum(self) -> None: + """Compact the database file (SQLite VACUUM).""" + # Bypasses get_connection: VACUUM can't run inside a transaction. + conn = sqlite3.connect(self.db_path, timeout=60) + try: + conn.isolation_level = None + conn.execute("VACUUM") + finally: + conn.close() diff --git a/backend/util/database/media_asset_matches.py b/backend/util/database/media_asset_matches.py index ead5b486..8f9b6bc4 100644 --- a/backend/util/database/media_asset_matches.py +++ b/backend/util/database/media_asset_matches.py @@ -186,21 +186,16 @@ def purge_season_logos(self) -> int: ) return int(deleted or 0) - def clear(self, keep_ignored: bool = False) -> None: - """Delete artwork-match rows. - - With ``keep_ignored`` the user's intentional rows — the per-type "not - needed" flags (ignored=1) AND manual-pick locks (user_confirmed=1) — are - preserved and only auto-matched applied/failed provenance is dropped. - Used by the Unmatched page's artwork reset, which must honour both the - user's ignores and their locked picks (so a re-run reuses the chosen - file instead of re-resolving it). - """ + def clear(self, keep_ignored: bool = False) -> int: + """Delete artwork-match rows; returns rows deleted.""" + # keep_ignored preserves the user's intent rows: ignored=1 AND + # user_confirmed=1 (the reset must honour locked picks). if keep_ignored: - self.execute_query( + deleted = self.execute_query( "DELETE FROM media_asset_matches " "WHERE (ignored IS NULL OR ignored = 0) " "AND (user_confirmed IS NULL OR user_confirmed = 0)" ) else: - self.execute_query("DELETE FROM media_asset_matches") + deleted = self.execute_query("DELETE FROM media_asset_matches") + return int(deleted or 0) diff --git a/backend/util/database/media_cache.py b/backend/util/database/media_cache.py index f2bbc953..fbd6bb97 100755 --- a/backend/util/database/media_cache.py +++ b/backend/util/database/media_cache.py @@ -315,6 +315,15 @@ def get_by_instance(self, instance_name: str) -> list: or [] ) + def count_by_instance(self, instance_name: str) -> int: + """Count rows for one instance without loading them (libraries are large).""" + row = self.execute_query( + "SELECT COUNT(*) AS total FROM media_cache WHERE instance_name=?", + (instance_name,), + fetch_one=True, + ) + return int(row["total"]) if row else 0 + def get_by_id(self, id: int) -> Optional[dict]: """Return a single media_cache row by its unique integer ID.""" return self.execute_query( diff --git a/backend/util/database/media_stats.py b/backend/util/database/media_stats.py index f94b154a..2badd8a0 100644 --- a/backend/util/database/media_stats.py +++ b/backend/util/database/media_stats.py @@ -72,6 +72,23 @@ class StatsMixin(DatabaseBase): """Aggregate library-health statistics over the media_cache table.""" + def count_added_since(self, cutoff: str) -> int: + """Rows first seen at or after an ISO cutoff (created_at is first-insert).""" + row = self.execute_query( + "SELECT COUNT(*) AS total FROM media_cache WHERE created_at >= ?", + (cutoff,), + fetch_one=True, + ) + return int(row["total"]) if row else 0 + + def count_unmatched(self) -> int: + """Rows with no poster match — the Unmatched page's media figure.""" + row = self.execute_query( + "SELECT COUNT(*) AS total FROM media_cache WHERE matched=0", + fetch_one=True, + ) + return int(row["total"]) if row else 0 + def get_stats( self, asset_type: Optional[str] = None, period_days: int = None ) -> dict: diff --git a/backend/util/database/poster_cache.py b/backend/util/database/poster_cache.py index de7d0955..5e458284 100755 --- a/backend/util/database/poster_cache.py +++ b/backend/util/database/poster_cache.py @@ -406,9 +406,9 @@ def get_by_normalized_title( sql += " ORDER BY priority DESC, id DESC LIMIT 1" return self.execute_query(sql, params, fetch_one=True) - def clear(self) -> None: - """Delete all rows from poster_cache.""" - self.execute_query("DELETE FROM poster_cache") + def clear(self) -> int: + """Delete all rows from poster_cache; returns rows deleted.""" + return int(self.execute_query("DELETE FROM poster_cache") or 0) def analyze(self) -> None: """Refresh planner stats; without them the match queries mis-pick indexes.""" diff --git a/backend/util/database/system_health.py b/backend/util/database/system_health.py new file mode 100644 index 00000000..c7b07c2f --- /dev/null +++ b/backend/util/database/system_health.py @@ -0,0 +1,52 @@ +# util/database/system_health.py + +from typing import Any, Dict, List, Optional + +from .db_base import DatabaseBase + + +class SystemHealth(DatabaseBase): + """Read access to the periodic instance-health probes the scheduler records. + + Rows are append-only snapshots of one (instance, service) probe; the + scheduler owns writing and pruning them. + """ + + def recent_snapshots( + self, limit: int = 50, instance: Optional[str] = None + ) -> List[Dict[str, Any]]: + """Newest snapshots, optionally for one instance.""" + # `, id DESC` breaks ties: probes in one scheduler pass share snapshot_at, + # so without it the LIMIT window would be arbitrary. + if instance: + rows = self.execute_query( + "SELECT * FROM system_health_snapshots WHERE instance_name=? " + "ORDER BY snapshot_at DESC, id DESC LIMIT ?", + (instance, limit), + fetch_all=True, + ) + else: + rows = self.execute_query( + "SELECT * FROM system_health_snapshots " + "ORDER BY snapshot_at DESC, id DESC LIMIT ?", + (limit,), + fetch_all=True, + ) + return [dict(r) for r in rows or []] + + def latest_per_instance(self) -> List[Dict[str, Any]]: + """The most recent snapshot row for each instance.""" + rows = self.execute_query( + """ + SELECT s.instance_name, s.service, s.status, s.response_time_ms, + s.status_code, s.snapshot_at, s.error + FROM system_health_snapshots s + INNER JOIN ( + SELECT instance_name, MAX(snapshot_at) AS latest + FROM system_health_snapshots GROUP BY instance_name + ) latest ON s.instance_name = latest.instance_name + AND s.snapshot_at = latest.latest + """, + fetch_all=True, + ) + return [dict(r) for r in rows or []] diff --git a/backend/util/database/worker.py b/backend/util/database/worker.py index 9e690892..a1d48547 100755 --- a/backend/util/database/worker.py +++ b/backend/util/database/worker.py @@ -5,7 +5,7 @@ import time from dataclasses import dataclass from enum import Enum -from typing import Callable, Dict, Optional +from typing import Any, Callable, Dict, List, Optional from .db_base import DatabaseBase @@ -488,6 +488,72 @@ def job_stats(self, table_name: str = "jobs", error_limit: int = 10): "error_code": "DB_JOB_STATS_ERROR", } + def count_by_status(self, status: str, table_name: str = "jobs") -> int: + """Number of jobs currently in one status.""" + self._check_table(table_name) + row = self.execute_query( + f"SELECT COUNT(*) AS total FROM {table_name} WHERE status=?", # noqa: S608 + (status,), + fetch_one=True, + ) + return int(row["total"]) if row else 0 + + def count_by_status_since( + self, cutoff: str, table_name: str = "jobs" + ) -> Dict[str, int]: + """{status: count} for jobs received at or after an ISO cutoff.""" + self._check_table(table_name) + rows = self.execute_query( + f"SELECT status, COUNT(*) AS total FROM {table_name} " # noqa: S608 + "WHERE received_at >= ? GROUP BY status", + (cutoff,), + fetch_all=True, + ) + return {row["status"]: row["total"] for row in rows or []} + + def recent_failures( + self, cutoff: str, limit: int = 20, table_name: str = "jobs" + ) -> List[Dict[str, Any]]: + """Errored jobs since an ISO cutoff, newest first.""" + self._check_table(table_name) + # `, id DESC` breaks ties: received_at has second-level collisions under + # a webhook burst, so without it the LIMIT window would be arbitrary. + rows = self.execute_query( + f"SELECT id, type, payload, error, received_at FROM {table_name} " # noqa: S608 + "WHERE status='error' AND received_at >= ? " + "ORDER BY received_at DESC, id DESC LIMIT ?", + (cutoff, limit), + fetch_all=True, + ) + return [dict(r) for r in rows or []] + + def jobs_of_type_since( + self, job_type: str, cutoff: str, table_name: str = "jobs" + ) -> List[Dict[str, Any]]: + """Jobs of one type received at or after an ISO cutoff, newest first.""" + self._check_table(table_name) + rows = self.execute_query( + f"SELECT id, payload, status, received_at FROM {table_name} " # noqa: S608 + "WHERE type=? AND received_at >= ? ORDER BY received_at DESC, id DESC", + (job_type, cutoff), + fetch_all=True, + ) + return [dict(r) for r in rows or []] + + def cancel_running_job( + self, job_id: int, completed_at: str, table_name: str = "jobs" + ) -> int: + """Mark a still-running job cancelled; returns rows changed (0 if it finished).""" + self._check_table(table_name) + return ( + self.execute_query( + f"UPDATE {table_name} SET status='cancelled', completed_at=? " # noqa: S608 + "WHERE id=? AND status='running'", + (completed_at, job_id), + ) + or 0 + ) + def update_progress(self, table_name: str, job_id: int, progress: int): self._check_table(table_name) self.execute_query( diff --git a/backend/util/notification.py b/backend/util/notification.py index b2c1ade2..feb57888 100755 --- a/backend/util/notification.py +++ b/backend/util/notification.py @@ -416,13 +416,13 @@ def send_discord_notification( output: Any, test: bool = False, ) -> Tuple[bool, str]: - from datetime import datetime + from datetime import datetime, timezone from backend.util.notification_formatting import format_for_discord hook = auth_data.webhook.rstrip("/") bot_name = auth_data.bot_name - timestamp = datetime.utcnow().isoformat() + timestamp = datetime.now(timezone.utc).isoformat() color = self.resolve_color(output, auth_data.color, default=0x00FF00) if test: diff --git a/tests/test_media_api_queries.py b/tests/test_media_api_queries.py index 350b31d2..3a1de7da 100644 --- a/tests/test_media_api_queries.py +++ b/tests/test_media_api_queries.py @@ -380,7 +380,8 @@ def test_orphaned_purge_route_deletes_and_guards_empty_ids(db): doomed = _seed(db, "Doomed") keeper = _seed(db, "Keeper") - assert client.post("/api/media/orphaned/purge", json={"ids": []}).status_code == 400 + empty = client.post("/api/media/orphaned/purge", json={"ids": []}) + assert empty.status_code == 400 resp = client.post("/api/media/orphaned/purge", json={"ids": [doomed, 999999]}) assert resp.status_code == 200 # The absent id must not inflate the count — `purged` is what was deleted. diff --git a/tests/test_system_queries.py b/tests/test_system_queries.py new file mode 100644 index 00000000..1277acca --- /dev/null +++ b/tests/test_system_queries.py @@ -0,0 +1,485 @@ +"""Tests for the system.py / jobs.py query methods that moved into the DB interfaces. + +Two layers: the new DbMaintenance / SystemHealth / DBWorker / cache methods against +a real temp database, then the system handlers whose contract nothing else pinned. +""" + +import os +import sys + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) + +from backend.util.config import ConfigError # noqa: E402 +from backend.util.database import ChubDB # noqa: E402 + + +class _StubLog: + """Swallows every log call; get_adapter returns itself.""" + + def __getattr__(self, _): + """Any log method is a no-op.""" + return lambda *a, **k: None + + def get_adapter(self, *_a, **_kw): + """Adapters are the same sink.""" + return self + + +@pytest.fixture +def db(tmp_path): + """A ChubDB backed by a real (temporary) sqlite file.""" + with ChubDB(_StubLog(), db_path=str(tmp_path / "chub.db")) as database: + yield database + + +def _client(db): + """Mount the system router on a bare app carrying main.py's ConfigError handler.""" + import backend.api.main as apimain + import backend.api.system as system + + app = FastAPI() + app.state.logger = _StubLog() + app.state.db = db + app.add_exception_handler(ConfigError, apimain.handle_config_error) + app.include_router(system.router) + return TestClient(app, raise_server_exceptions=False) + + +def _seed_job(db, status="pending", received_at="2026-01-02T00:00:00+00:00", **fields): + """Insert one jobs row with an exact status/received_at and return its id.""" + row = { + "type": "module_run", + "status": status, + "received_at": received_at, + "payload": "{}", + **fields, + } + keys = ", ".join(row) + marks = ", ".join("?" for _ in row) + return db.worker.execute_query( + f"INSERT INTO jobs ({keys}) VALUES ({marks})", + tuple(row.values()), + last_row_id=True, + ) + + +def _seed_snapshot(db, instance, snapshot_at, status="healthy", service="radarr"): + """Insert one system_health_snapshots row and return its id.""" + return db.worker.execute_query( + "INSERT INTO system_health_snapshots " + "(snapshot_at, service, instance_name, status) VALUES (?, ?, ?, ?)", + (snapshot_at, service, instance, status), + last_row_id=True, + ) + + +def _seed_media(db, title, instance_name="radarr", matched=None): + """Upsert one media row, optionally flipping its matched flag, and return its id.""" + db.media.upsert( + {"title": title, "normalized_title": title.lower(), "year": 2021}, + "movie", + "radarr", + instance_name, + ) + row = db.media.execute_query( + "SELECT id FROM media_cache WHERE title=? AND instance_name=?", + (title, instance_name), + fetch_one=True, + ) + if matched is not None: + db.media.execute_query( + "UPDATE media_cache SET matched=? WHERE id=?", (matched, row["id"]) + ) + return row["id"] + + +# --- DbMaintenance.ping ----------------------------------------------------- + + +def test_ping_round_trips_a_live_database(db): + """A usable DB answers True rather than a falsy 'probably fine'.""" + assert db.maintenance.ping() is True + + +def test_ping_raises_on_an_unusable_database(tmp_path): + """/api/health reports 'error' only because ping propagates — never swallow it.""" + import sqlite3 + + from backend.util.database import DbMaintenance + + broken = tmp_path / "broken.db" + broken.write_bytes(b"this is not a sqlite file" * 100) + # Built directly: ChubDB's schema init would raise before ping ever ran. + maintenance = DbMaintenance(logger=_StubLog(), db_path=str(broken)) + + with pytest.raises(sqlite3.DatabaseError): + maintenance.ping() + + +# --- DbMaintenance.table_row_counts / page_stats / list_migrations ---------- + + +def test_table_row_counts_reports_seeded_rows(db): + """Counts come from the tables themselves, not from a cached figure.""" + for title in ("Dune", "Sicario"): + _seed_media(db, title) + + counts = {t["name"]: t["rows"] for t in db.maintenance.table_row_counts()} + assert counts["media_cache"] == 2 + assert counts["jobs"] == 0 + + +def test_table_row_counts_hides_sqlite_internal_tables(db): + """The allowlist is the filter — enumerating sqlite_master would leak these.""" + _seed_job(db) # jobs has an INTEGER PRIMARY KEY, so sqlite_sequence may exist + + names = {t["name"] for t in db.maintenance.table_row_counts()} + assert not any(n.startswith("sqlite_") for n in names) + assert names <= set(db.maintenance.STATS_TABLES) + + +def test_table_row_counts_skips_a_table_the_schema_lacks(db): + """An allowlisted-but-absent table is skipped, not counted into an error.""" + db.maintenance.execute_query("DROP TABLE gdrive_stats") + + names = {t["name"] for t in db.maintenance.table_row_counts()} + assert "gdrive_stats" not in names + assert "media_cache" in names + + +def test_page_stats_derives_its_byte_totals_from_the_page_size(db): + """total_bytes/free_bytes are page_size multiplied by the right counter.""" + stats = db.maintenance.page_stats() + + assert stats["page_size"] > 0 and stats["page_count"] > 0 + assert stats["total_bytes"] == stats["page_size"] * stats["page_count"] + assert stats["free_bytes"] == stats["page_size"] * stats["freelist_count"] + + +def test_list_migrations_is_newest_first(db): + """Ordering is applied_at DESC, so an older entry can't lead the list.""" + db.maintenance.execute_query( + "INSERT INTO schema_migrations (name, applied_at) VALUES (?, ?)", + ("20260101_old", "2026-01-01T00:00:00"), + ) + db.maintenance.execute_query( + "INSERT INTO schema_migrations (name, applied_at) VALUES (?, ?)", + ("20260202_new", "2026-02-02T00:00:00"), + ) + + # Schema init writes its own real migrations, so compare positions not slices. + names = [m["name"] for m in db.maintenance.list_migrations()] + assert names.index("20260202_new") < names.index("20260101_old") + + +def test_list_migrations_is_empty_when_the_table_is_absent(db): + """The existence guard keeps a pre-migrations database from erroring.""" + db.maintenance.execute_query("DROP TABLE schema_migrations") + + assert db.maintenance.list_migrations() == [] + + +# --- DbMaintenance.vacuum --------------------------------------------------- + + +def test_vacuum_releases_the_free_pages_a_bulk_delete_left(db): + """VACUUM is what empties the freelist; a no-op would leave it populated.""" + for i in range(500): + db.poster.execute_query( + "INSERT INTO poster_cache (file, title, year, asset_type, " + "normalized_title) VALUES (?, ?, 2020, 'movie', ?)", + (f"/src/p{i}.jpg", f"Movie {i}", f"movie{i}"), + ) + db.poster.execute_query("DELETE FROM poster_cache") + assert db.maintenance.page_stats()["freelist_count"] > 0 + + db.maintenance.vacuum() + + assert db.maintenance.page_stats()["freelist_count"] == 0 + + +# --- SystemHealth ----------------------------------------------------------- + + +def test_recent_snapshots_is_newest_first_under_the_limit(db): + """Newest-first ordering plus the LIMIT decide the window.""" + for at in ("2026-01-01T00:00:00", "2026-01-03T00:00:00", "2026-01-02T00:00:00"): + _seed_snapshot(db, "radarr-main", at) + + rows = db.system_health.recent_snapshots(limit=2) + assert [r["snapshot_at"] for r in rows] == [ + "2026-01-03T00:00:00", + "2026-01-02T00:00:00", + ] + + +def test_recent_snapshots_scopes_to_one_instance(db): + """The instance filter is a WHERE, not a post-filter on an unbounded read.""" + _seed_snapshot(db, "radarr-main", "2026-01-01T00:00:00") + _seed_snapshot(db, "sonarr-main", "2026-01-02T00:00:00") + + rows = db.system_health.recent_snapshots(instance="radarr-main") + assert [r["instance_name"] for r in rows] == ["radarr-main"] + + +def test_latest_per_instance_keeps_one_row_per_instance(db): + """The self-join collapses history to each instance's most recent probe.""" + _seed_snapshot(db, "radarr-main", "2026-01-01T00:00:00", status="unhealthy") + _seed_snapshot(db, "radarr-main", "2026-01-05T00:00:00", status="healthy") + _seed_snapshot(db, "sonarr-main", "2026-01-02T00:00:00", status="timeout") + + rows = db.system_health.latest_per_instance() + assert len(rows) == 2 # the superseded radarr probe must not survive the join + latest = {r["instance_name"]: r for r in rows} + assert latest["radarr-main"]["status"] == "healthy" + assert latest["sonarr-main"]["status"] == "timeout" + + +# --- DBWorker job reporting ------------------------------------------------- + + +def test_count_by_status_counts_only_that_status(db): + """The status is a bound parameter in the WHERE, not a filter on everything.""" + _seed_job(db, status="error") + _seed_job(db, status="error") + _seed_job(db, status="success") + + assert db.worker.count_by_status("error") == 2 + assert db.worker.count_by_status("cancelled") == 0 + + +def test_count_by_status_since_groups_inside_the_window(db): + """Only jobs at or after the cutoff are grouped, and each status keeps its own count.""" + _seed_job(db, status="error", received_at="2026-01-05T00:00:00+00:00") + _seed_job(db, status="success", received_at="2026-01-05T00:00:00+00:00") + _seed_job(db, status="success", received_at="2026-01-06T00:00:00+00:00") + _seed_job(db, status="success", received_at="2025-12-01T00:00:00+00:00") + + counts = db.worker.count_by_status_since("2026-01-01T00:00:00+00:00") + assert counts == {"error": 1, "success": 2} + + +def test_recent_failures_excludes_successes_and_pre_cutoff_rows(db): + """Both halves of the WHERE matter: status='error' AND inside the window.""" + doomed = _seed_job(db, status="error", received_at="2026-01-05T00:00:00+00:00") + _seed_job(db, status="success", received_at="2026-01-05T00:00:00+00:00") + _seed_job(db, status="error", received_at="2025-01-01T00:00:00+00:00") + + rows = db.worker.recent_failures("2026-01-01T00:00:00+00:00") + assert [r["id"] for r in rows] == [doomed] + assert set(rows[0]) == {"id", "type", "payload", "error", "received_at"} + + +def test_recent_failures_breaks_received_at_ties_by_id(db): + """A webhook burst shares received_at — without `, id DESC` the LIMIT is arbitrary.""" + ids = [ + _seed_job(db, status="error", received_at="2026-01-05T00:00:00+00:00") + for _ in range(3) + ] + + rows = db.worker.recent_failures("2026-01-01T00:00:00+00:00", limit=2) + assert [r["id"] for r in rows] == [ids[2], ids[1]] + + +def test_jobs_of_type_since_filters_on_both_type_and_cutoff(db): + """A module_run in the window and an old webhook are both excluded.""" + wanted = _seed_job(db, type="webhook", received_at="2026-01-05T00:00:00+00:00") + _seed_job(db, type="module_run", received_at="2026-01-05T00:00:00+00:00") + _seed_job(db, type="webhook", received_at="2025-01-01T00:00:00+00:00") + + rows = db.worker.jobs_of_type_since("webhook", "2026-01-01T00:00:00+00:00") + assert [r["id"] for r in rows] == [wanted] + + +def test_cancel_running_job_reports_the_row_it_changed(db): + """The count is the DB's rowcount, so a second cancel reports zero.""" + job = _seed_job(db, status="running") + + assert db.worker.cancel_running_job(job, "2026-01-05T00:00:00+00:00") == 1 + assert db.worker.cancel_running_job(job, "2026-01-05T00:00:00+00:00") == 0 + + +def test_cancel_running_job_leaves_a_pending_job_alone(db): + """`AND status='running'` is the guard — a queued job must not be cancelled.""" + pending = _seed_job(db, status="pending") + + assert db.worker.cancel_running_job(pending, "2026-01-05T00:00:00+00:00") == 0 + assert db.worker.get_job_by_id("jobs", pending)["status"] == "pending" + + +def test_job_reporting_refuses_a_table_outside_the_allowlist(db): + """Every new jobs method runs _check_table, so a stray table name can't reach SQL.""" + for call in ( + lambda: db.worker.count_by_status("error", table_name="media_cache"), + lambda: db.worker.count_by_status_since("2026-01-01", table_name="media_cache"), + lambda: db.worker.recent_failures("2026-01-01", table_name="media_cache"), + lambda: db.worker.jobs_of_type_since("x", "2026-01-01", table_name="media_cache"), + lambda: db.worker.cancel_running_job(1, "2026-01-01", table_name="media_cache"), + ): + with pytest.raises(ValueError): + call() + + +# --- MediaCache / CollectionCache counts ------------------------------------ + + +def test_count_added_since_honours_the_created_at_cutoff(db): + """created_at is stamped on first insert; the cutoff must actually filter.""" + _seed_media(db, "Dune") + db.media.execute_query( + "UPDATE media_cache SET created_at='2020-01-01 00:00:00' WHERE title='Dune'" + ) + _seed_media(db, "Sicario") + db.media.execute_query( + "UPDATE media_cache SET created_at='2026-01-05 00:00:00' WHERE title='Sicario'" + ) + + assert db.media.count_added_since("2026-01-01 00:00:00") == 1 + assert db.media.count_added_since("2019-01-01 00:00:00") == 2 + + +def test_count_unmatched_counts_only_unmatched_media(db): + """matched=0 is the filter; a matched row must not inflate the figure.""" + _seed_media(db, "Dune", matched=0) + _seed_media(db, "Sicario", matched=1) + + assert db.media.count_unmatched() == 1 + + +def test_collection_count_unmatched_counts_only_unmatched_collections(db): + """The collections figure reads collections_cache, never media_cache.""" + db.collection.upsert({"title": "Marvel", "library_name": "Movies"}, "plex1") + db.collection.upsert({"title": "DC", "library_name": "Movies"}, "plex1") + marvel = db.collection.get_by_title_and_instance("Marvel", "plex1", "Movies") + db.collection.execute_query( + "UPDATE collections_cache SET matched=1 WHERE id=?", (marvel["id"],) + ) + # Two unmatched media decoys, so reading the wrong table gives a wrong number. + _seed_media(db, "Decoy", matched=0) + _seed_media(db, "Decoy Two", matched=0) + + assert db.collection.count_unmatched() == 1 + + +def test_count_by_instance_scopes_to_that_instance(db): + """The instance name is in the WHERE — the count is not the whole table.""" + _seed_media(db, "Dune", instance_name="radarr-main") + _seed_media(db, "Sicario", instance_name="radarr-main") + _seed_media(db, "Arrival", instance_name="radarr-4k") + + assert db.media.count_by_instance("radarr-main") == 2 + assert db.media.count_by_instance("radarr-nope") == 0 + + +# --- clear() rowcounts ------------------------------------------------------ + + +def test_poster_clear_returns_the_rows_it_deleted(db): + """The wipe reports its own rowcount, so a repeat wipe reports zero.""" + db.poster.execute_query( + "INSERT INTO poster_cache (file, title, year, asset_type, normalized_title) " + "VALUES ('/src/a.jpg', 'A', 2020, 'movie', 'a')" + ) + + assert db.poster.clear() == 1 + assert db.poster.clear() == 0 + + +def test_artwork_clear_keep_ignored_counts_only_what_it_deleted(db): + """A user_confirmed row survives, so it must not be counted as deleted.""" + db.media_asset_matches.upsert( + target_kind="media", + target_id=1, + image_type="logo", + match_status="applied", + ) + db.media_asset_matches.set_ignored("media", 2, "background", True) + db.media_asset_matches.set_user_confirmed("media", 3, "logo", True) + + assert db.media_asset_matches.clear(keep_ignored=True) == 1 + assert db.media_asset_matches.get_one("media", 2, "background")["ignored"] == 1 + assert db.media_asset_matches.get_one("media", 3, "logo")["user_confirmed"] == 1 + + +# --- Route contracts not covered elsewhere --------------------------------- + + +def test_artwork_reset_route_reports_only_the_rows_it_deleted(db): + """The response count comes from the DELETE, not from a wider pre-count.""" + db.media_asset_matches.upsert( + target_kind="media", target_id=1, image_type="logo", match_status="applied" + ) + db.media_asset_matches.set_user_confirmed("media", 3, "logo", True) + + body = _client(db).post("/api/system/db/artwork-matches/reset").json() + assert body["data"]["deleted"] == 1 + + +def test_poster_cache_clear_route_reports_the_deleted_rows(db): + """The wipe count is the rowcount of the DELETE the route performed.""" + db.poster.execute_query( + "INSERT INTO poster_cache (file, title, year, asset_type, normalized_title) " + "VALUES ('/src/a.jpg', 'A', 2020, 'movie', 'a')" + ) + + body = _client(db).post("/api/system/db/poster-cache/clear").json() + assert body["data"]["deleted"] == 1 + assert db.stats.count_poster_cache() == 0 + + +def test_cleanup_candidates_route_reports_each_count_from_its_own_table(db): + """errored_jobs / unmatched_media / unmatched_collections must not cross-wire.""" + # Deliberately distinct counts — equal ones would let a swapped pair pass. + for _ in range(3): + _seed_job(db, status="error") + for title in ("Dune", "Sicario"): + _seed_media(db, title, matched=0) + _seed_media(db, "Arrival", matched=1) + db.collection.upsert({"title": "Marvel", "library_name": "Movies"}, "plex1") + + data = _client(db).get("/api/system/cleanup-candidates").json()["data"] + assert data == { + "errored_jobs": 3, + "unmatched_media": 2, + "unmatched_collections": 1, + } + + +def test_digest_route_reports_the_window_it_was_asked_for(db): + """Jobs outside the ?days window are excluded from the counts and failures.""" + _seed_job(db, status="error", received_at="2020-01-01T00:00:00+00:00") + _seed_job(db, status="success") # dated 2026-01-02, far outside a 1-day window + _seed_snapshot(db, "radarr-main", "2026-01-01T00:00:00") + + data = _client(db).get("/api/system/digest?days=1").json()["data"] + assert data["window_days"] == 1 + assert data["job_counts"] == {} + assert data["recent_failures"] == [] + assert [h["instance_name"] for h in data["latest_instance_health"]] == [ + "radarr-main" + ] + + +def test_health_snapshots_route_clamps_its_limit(db): + """limit is clamped into 1..500 before it reaches the query.""" + for i in range(3): + _seed_snapshot(db, "radarr-main", f"2026-01-0{i + 1}T00:00:00") + client = _client(db) + + zero = client.get("/api/system/health/snapshots?limit=0").json() + assert len(zero["data"]["snapshots"]) == 1 + huge = client.get("/api/system/health/snapshots?limit=99999").json() + assert len(huge["data"]["snapshots"]) == 3 + + +def test_health_snapshots_route_filters_by_instance(db): + """The ?instance query reaches the WHERE, not just the response message.""" + _seed_snapshot(db, "radarr-main", "2026-01-01T00:00:00") + _seed_snapshot(db, "sonarr-main", "2026-01-02T00:00:00") + + body = _client(db).get("/api/system/health/snapshots?instance=sonarr-main").json() + assert [s["instance_name"] for s in body["data"]["snapshots"]] == ["sonarr-main"] From ee6d5ed275d7894976efe65dbe7d07756b057edb Mon Sep 17 00:00:00 2001 From: chodeus Date: Sat, 15 Aug 2026 04:08:11 +0800 Subject: [PATCH 3/5] fix: compare created_at cutoffs as instants, not text created_at is CURRENT_TIMESTAMP ('YYYY-MM-DD HH:MM:SS') but the cutoff is Python isoformat, so ' ' < 'T' dropped every row added later in the day than the cutoff. datetime() on both sides also normalizes the offset. --- backend/util/database/media_stats.py | 5 ++++- backend/util/database/poster_cache.py | 5 ++++- tests/test_posters_queries.py | 13 +++++++++++++ tests/test_system_queries.py | 16 ++++++++++++++++ 4 files changed, 37 insertions(+), 2 deletions(-) diff --git a/backend/util/database/media_stats.py b/backend/util/database/media_stats.py index 7388eace..80ffcc64 100644 --- a/backend/util/database/media_stats.py +++ b/backend/util/database/media_stats.py @@ -74,8 +74,11 @@ class StatsMixin(DatabaseBase): def count_added_since(self, cutoff: str) -> int: """Rows first seen at or after an ISO cutoff (created_at is first-insert).""" + # created_at is CURRENT_TIMESTAMP ('YYYY-MM-DD HH:MM:SS'), the cutoff is + # Python isoformat ('...T...'): compare instants, not text. No index to lose. row = self.execute_query( - "SELECT COUNT(*) AS total FROM media_cache WHERE created_at >= ?", + "SELECT COUNT(*) AS total FROM media_cache " + "WHERE datetime(created_at) >= datetime(?)", (cutoff,), fetch_one=True, ) diff --git a/backend/util/database/poster_cache.py b/backend/util/database/poster_cache.py index 5e458284..6ff7b2f9 100755 --- a/backend/util/database/poster_cache.py +++ b/backend/util/database/poster_cache.py @@ -313,9 +313,12 @@ def added_since( ) -> list: """Return poster_cache rows added at or after the ISO-8601 cutoff.""" it_sql, it_params = self._image_type_clause(image_type) + # The cutoff is caller-supplied, so its separator/offset needn't match the + # stored form — compare instants. created_at is deliberately unindexed. rows = ( self.execute_query( - "SELECT * FROM poster_cache WHERE created_at >= ?" + it_sql + " " + "SELECT * FROM poster_cache " + "WHERE datetime(created_at) >= datetime(?)" + it_sql + " " "ORDER BY created_at DESC LIMIT ?", (iso_cutoff, *it_params, int(limit)), fetch_all=True, diff --git a/tests/test_posters_queries.py b/tests/test_posters_queries.py index 80b1d809..cf27b05f 100644 --- a/tests/test_posters_queries.py +++ b/tests/test_posters_queries.py @@ -215,6 +215,19 @@ def test_find_missing_dimensions_walks_ids_in_order_under_the_limit(db): assert set(rows[0]) == {"id", "file"} +# --- PosterCache.added_since ----------------------------------------------- + + +def test_added_since_reads_the_cutoff_as_an_instant_not_a_string(db): + """created_at is stored in UTC; a cutoff in another offset must be converted first.""" + _seed_poster(db, "Later", created_at="2026-01-05T10:00:00+00:00") + _seed_poster(db, "Earlier", created_at="2026-01-05T08:00:00+00:00") + + # 17:00+08:00 is 09:00 UTC, but a TEXT compare reads "17" as after both rows. + rows = db.poster.added_since("2026-01-05T17:00:00+08:00") + assert [r["title"] for r in rows] == ["Later"] + + # --- MediaCache / CollectionCache approve_match + reopen_for_review --------- diff --git a/tests/test_system_queries.py b/tests/test_system_queries.py index 1277acca..8104abef 100644 --- a/tests/test_system_queries.py +++ b/tests/test_system_queries.py @@ -342,6 +342,22 @@ def test_count_added_since_honours_the_created_at_cutoff(db): assert db.media.count_added_since("2019-01-01 00:00:00") == 2 +def test_count_added_since_counts_a_current_timestamp_row_against_an_iso_cutoff(db): + """created_at is CURRENT_TIMESTAMP-shaped; the digest's ISO cutoff must still match it.""" + _seed_media(db, "Dune") + db.media.execute_query( + "UPDATE media_cache SET created_at='2026-01-05 18:30:00' WHERE title='Dune'" + ) + _seed_media(db, "Sicario") + db.media.execute_query( + "UPDATE media_cache SET created_at='2026-01-05 03:00:00' WHERE title='Sicario'" + ) + + # ' ' sorts below 'T' at index 10, so a TEXT compare drops Dune despite it + # being the later row. This is the exact cutoff shape system.py builds. + assert db.media.count_added_since("2026-01-05T09:00:00.123456+00:00") == 1 + + def test_count_unmatched_counts_only_unmatched_media(db): """matched=0 is the filter; a matched row must not inflate the figure.""" _seed_media(db, "Dune", matched=0) From f09bb159d4556708c5f40c61e036e36a44fc02e2 Mon Sep 17 00:00:00 2001 From: chodeus Date: Sat, 15 Aug 2026 06:53:43 +0800 Subject: [PATCH 4/5] fix: 503 on an unusable database, one health row per service The health endpoint backs the Docker HEALTHCHECK, so a dead DB must not answer 200. latest_per_instance grouped by instance alone, dropping a service that probed earlier; cancel now reports 409 when the guarded update finds nothing; jobs cutoffs compare as instants. --- backend/api/modules.py | 8 ++- backend/api/system.py | 25 +++++--- backend/util/database/system_health.py | 20 +++--- backend/util/database/worker.py | 9 ++- tests/test_system_queries.py | 86 ++++++++++++++++++++++++++ 5 files changed, 126 insertions(+), 22 deletions(-) diff --git a/backend/api/modules.py b/backend/api/modules.py index a59a4d3a..b6fec122 100755 --- a/backend/api/modules.py +++ b/backend/api/modules.py @@ -1289,7 +1289,13 @@ async def cancel_module_execution( if request_cancellation(job_id): # Update job status to reflect cancellation is in progress now = datetime.now(timezone.utc).isoformat() - db.worker.cancel_running_job(job_id, now) + # 0 rows means it stopped running between the check above and here. + if not db.worker.cancel_running_job(job_id, now): + return error( + f"Job {job_id} finished before it could be cancelled", + code="JOB_NOT_RUNNING", + status_code=409, + ) logger.info(f"Cancellation requested for module {name} job {job_id}") return ok( f"Cancellation requested for module {name} job {job_id}", diff --git a/backend/api/system.py b/backend/api/system.py index 6196c928..43b0c700 100755 --- a/backend/api/system.py +++ b/backend/api/system.py @@ -184,15 +184,22 @@ async def health_check(request: Request) -> JSONResponse: version = get_version() - return ok( - "Healthy" if status == "ok" else "Degraded", - { - "status": status, - "version": version, - "uptime_seconds": uptime, - "checks": checks, - }, - ) + payload = { + "status": status, + "version": version, + "uptime_seconds": uptime, + "checks": checks, + } + # Docker HEALTHCHECK curls -f this, so an unusable DB must not answer 200. + # A stopped worker stays 200 — degraded, but still serving. + if checks.get("database") == "error": + return error( + "Database unavailable", + code="DATABASE_UNAVAILABLE", + data=payload, + status_code=503, + ) + return ok("Healthy" if status == "ok" else "Degraded", payload) @router.get( diff --git a/backend/util/database/system_health.py b/backend/util/database/system_health.py index c7b07c2f..99ccee97 100644 --- a/backend/util/database/system_health.py +++ b/backend/util/database/system_health.py @@ -35,17 +35,19 @@ def recent_snapshots( return [dict(r) for r in rows or []] def latest_per_instance(self) -> List[Dict[str, Any]]: - """The most recent snapshot row for each instance.""" + """The most recent snapshot row for each (instance, service) pair.""" rows = self.execute_query( """ - SELECT s.instance_name, s.service, s.status, s.response_time_ms, - s.status_code, s.snapshot_at, s.error - FROM system_health_snapshots s - INNER JOIN ( - SELECT instance_name, MAX(snapshot_at) AS latest - FROM system_health_snapshots GROUP BY instance_name - ) latest ON s.instance_name = latest.instance_name - AND s.snapshot_at = latest.latest + SELECT instance_name, service, status, response_time_ms, + status_code, snapshot_at, error + FROM ( + SELECT s.*, ROW_NUMBER() OVER ( + PARTITION BY instance_name, service + ORDER BY snapshot_at DESC, id DESC + ) AS rn + FROM system_health_snapshots s + ) + WHERE rn = 1 """, fetch_all=True, ) diff --git a/backend/util/database/worker.py b/backend/util/database/worker.py index a1d48547..6903ebe7 100755 --- a/backend/util/database/worker.py +++ b/backend/util/database/worker.py @@ -502,10 +502,12 @@ def count_by_status_since( self, cutoff: str, table_name: str = "jobs" ) -> Dict[str, int]: """{status: count} for jobs received at or after an ISO cutoff.""" + # datetime() on both sides: the cutoff is caller-supplied, so a differing + # separator or UTC offset would silently shift a TEXT compare's window. self._check_table(table_name) rows = self.execute_query( f"SELECT status, COUNT(*) AS total FROM {table_name} " # noqa: S608 - "WHERE received_at >= ? GROUP BY status", + "WHERE datetime(received_at) >= datetime(?) GROUP BY status", (cutoff,), fetch_all=True, ) @@ -520,7 +522,7 @@ def recent_failures( # a webhook burst, so without it the LIMIT window would be arbitrary. rows = self.execute_query( f"SELECT id, type, payload, error, received_at FROM {table_name} " # noqa: S608 - "WHERE status='error' AND received_at >= ? " + "WHERE status='error' AND datetime(received_at) >= datetime(?) " "ORDER BY received_at DESC, id DESC LIMIT ?", (cutoff, limit), fetch_all=True, @@ -534,7 +536,8 @@ def jobs_of_type_since( self._check_table(table_name) rows = self.execute_query( f"SELECT id, payload, status, received_at FROM {table_name} " # noqa: S608 - "WHERE type=? AND received_at >= ? ORDER BY received_at DESC, id DESC", + "WHERE type=? AND datetime(received_at) >= datetime(?) " + "ORDER BY received_at DESC, id DESC", (job_type, cutoff), fetch_all=True, ) diff --git a/tests/test_system_queries.py b/tests/test_system_queries.py index 8104abef..5d9eb1a0 100644 --- a/tests/test_system_queries.py +++ b/tests/test_system_queries.py @@ -499,3 +499,89 @@ def test_health_snapshots_route_filters_by_instance(db): body = _client(db).get("/api/system/health/snapshots?instance=sonarr-main").json() assert [s["instance_name"] for s in body["data"]["snapshots"]] == ["sonarr-main"] + + +def test_latest_per_instance_keeps_every_service_on_a_shared_instance(db): + """Grouping by instance alone dropped whichever service probed less recently.""" + _seed_snapshot(db, "shared", "2026-01-05 10:00:00", service="radarr") + _seed_snapshot(db, "shared", "2026-01-05 10:05:00", service="sonarr") + + rows = db.system_health.latest_per_instance() + + assert sorted(r["service"] for r in rows) == ["radarr", "sonarr"] + + +def test_latest_per_instance_returns_one_row_per_pair_on_a_timestamp_tie(db): + """A scheduler pass writes identical snapshot_at values; id breaks the tie.""" + _seed_snapshot(db, "radarr1", "2026-01-05 10:00:00", status="healthy") + newest = _seed_snapshot(db, "radarr1", "2026-01-05 10:00:00", status="error") + + rows = db.system_health.latest_per_instance() + + assert len(rows) == 1 + assert rows[0]["status"] == "error" # highest id wins, deterministically + assert newest # the tie-breaking row is the one that was seeded last + + +def test_health_route_answers_503_when_the_database_is_unusable(db, monkeypatch): + """Docker HEALTHCHECK curls -f this, so an unusable DB must not answer 200.""" + healthy = _client(db).get("/api/health") + assert healthy.status_code == 200 + + def _boom(): + raise RuntimeError("disk gone") + + monkeypatch.setattr(type(db.maintenance), "ping", lambda _self: _boom()) + resp = _client(db).get("/api/health") + + assert resp.status_code == 503 + body = resp.json() + assert body["error_code"] == "DATABASE_UNAVAILABLE" + assert body["data"]["checks"]["database"] == "error" # diagnostics survive + + +def test_count_by_status_since_reads_the_cutoff_as_an_instant(db): + """A caller-supplied offset cutoff must not shift the window via TEXT compare.""" + db.worker.execute_query( + "INSERT INTO jobs (type, payload, status, received_at) VALUES (?,?,?,?)", + ("webhook", "{}", "error", "2026-01-05 18:30:00"), + ) + + counts = db.worker.count_by_status_since("2026-01-05T09:00:00+00:00") + + assert counts.get("error") == 1 + + +def _modules_client(db): + """Mount the modules router the way _client mounts system's.""" + from backend.api import modules as modules_api + + app = FastAPI() + app.state.logger = _StubLog() + app.state.db = db + app.include_router(modules_api.router) + return TestClient(app, raise_server_exceptions=False) + + +def test_cancel_reports_conflict_when_the_job_stopped_first(db, monkeypatch): + """cancel_running_job returning 0 means it finished — don't answer 'cancelling'.""" + job_id = db.worker.execute_query( + "INSERT INTO jobs (type, payload, status, received_at) VALUES (?,?,?,?)", + ("poster_renamerr", "{}", "running", "2026-01-05T10:00:00+00:00"), + last_row_id=True, + ) + # The handler imports it inside the function, so patch it at the source. + monkeypatch.setattr( + "backend.util.job_processor.request_cancellation", lambda _id: True + ) + # The guarded UPDATE finds nothing: the job stopped between check and write. + monkeypatch.setattr( + type(db.worker), "cancel_running_job", lambda *_a, **_kw: 0 + ) + + resp = _modules_client(db).delete( + f"/api/modules/poster_renamerr/execution/{job_id}" + ) + + assert resp.status_code == 409, resp.text + assert resp.json()["error_code"] == "JOB_NOT_RUNNING" From 9f71e45f2f0a749f22873692598d157c2827d825 Mon Sep 17 00:00:00 2001 From: chodeus Date: Sat, 15 Aug 2026 08:01:00 +0800 Subject: [PATCH 5/5] fix: health fails closed when there is no database handle An absent state.db left checks['database'] unset, so the == 'error' test read it as healthy and answered 200. Mark it unavailable and gate on != 'ok'. --- backend/api/system.py | 9 +++++++-- tests/test_system_queries.py | 17 +++++++++++++++++ 2 files changed, 24 insertions(+), 2 deletions(-) diff --git a/backend/api/system.py b/backend/api/system.py index 43b0c700..191354bc 100755 --- a/backend/api/system.py +++ b/backend/api/system.py @@ -174,7 +174,12 @@ async def health_check(request: Request) -> JSONResponse: # Database health db = getattr(request.app.state, "db", None) - if db: + if db is None: + # No handle at all is as unserviceable as a failing ping — say so, so + # the check below can't read an absent key as healthy. + checks["database"] = "unavailable" + status = "degraded" + else: try: db.maintenance.ping() checks["database"] = "ok" @@ -192,7 +197,7 @@ async def health_check(request: Request) -> JSONResponse: } # Docker HEALTHCHECK curls -f this, so an unusable DB must not answer 200. # A stopped worker stays 200 — degraded, but still serving. - if checks.get("database") == "error": + if checks.get("database") != "ok": return error( "Database unavailable", code="DATABASE_UNAVAILABLE", diff --git a/tests/test_system_queries.py b/tests/test_system_queries.py index 5d9eb1a0..1f7904c6 100644 --- a/tests/test_system_queries.py +++ b/tests/test_system_queries.py @@ -585,3 +585,20 @@ def test_cancel_reports_conflict_when_the_job_stopped_first(db, monkeypatch): assert resp.status_code == 409, resp.text assert resp.json()["error_code"] == "JOB_NOT_RUNNING" + + +def test_health_route_answers_503_when_there_is_no_database_handle(db): + """No handle is as unserviceable as a failing ping — it must not pass the check.""" + from backend.api import system as system_api + + app = FastAPI() + app.state.logger = _StubLog() # state.db deliberately never set + app.include_router(system_api.router) + + resp = TestClient(app, raise_server_exceptions=False).get("/api/health") + + assert resp.status_code == 503 + body = resp.json() + assert body["error_code"] == "DATABASE_UNAVAILABLE" + assert body["data"]["checks"]["database"] == "unavailable" + assert body["data"]["status"] == "degraded" # payload isn't claiming health