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..b6fec122 100755 --- a/backend/api/modules.py +++ b/backend/api/modules.py @@ -1289,10 +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.execute_query( - "UPDATE jobs SET status='cancelled', completed_at=? WHERE id=? AND status='running'", - (now, job_id), - ) + # 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 72ece827..191354bc 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 @@ -175,9 +174,14 @@ 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.worker.execute_query("SELECT 1") + db.maintenance.ping() checks["database"] = "ok" except Exception: checks["database"] = "error" @@ -185,15 +189,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") != "ok": + return error( + "Database unavailable", + code="DATABASE_UNAVAILABLE", + data=payload, + status_code=503, + ) + return ok("Healthy" if status == "ok" else "Degraded", payload) @router.get( @@ -755,22 +766,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 +792,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 +822,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 +853,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 +872,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 +882,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 +921,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 +929,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 +973,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 +981,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 +1013,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 +1047,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 +1088,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 9a273d2f..80ffcc64 100644 --- a/backend/util/database/media_stats.py +++ b/backend/util/database/media_stats.py @@ -72,6 +72,26 @@ 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).""" + # 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 datetime(created_at) >= datetime(?)", + (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 cc773099..f47cbfdc 100755 --- a/backend/util/database/poster_cache.py +++ b/backend/util/database/poster_cache.py @@ -326,9 +326,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, @@ -419,9 +422,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..99ccee97 --- /dev/null +++ b/backend/util/database/system_health.py @@ -0,0 +1,54 @@ +# 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, service) pair.""" + rows = self.execute_query( + """ + 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, + ) + return [dict(r) for r in rows or []] diff --git a/backend/util/database/worker.py b/backend/util/database/worker.py index 9e690892..6903ebe7 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,75 @@ 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.""" + # 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 datetime(received_at) >= datetime(?) 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 datetime(received_at) >= datetime(?) " + "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 datetime(received_at) >= datetime(?) " + "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_posters_queries.py b/tests/test_posters_queries.py index 888ec8c1..18152339 100644 --- a/tests/test_posters_queries.py +++ b/tests/test_posters_queries.py @@ -216,6 +216,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 new file mode 100644 index 00000000..1f7904c6 --- /dev/null +++ b/tests/test_system_queries.py @@ -0,0 +1,604 @@ +"""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_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) + _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"] + + +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" + + +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