From d7e5c75cb18f3ef2ad3e68b610fd1b8637168bf4 Mon Sep 17 00:00:00 2001 From: Zak Bearman Date: Sun, 13 Sep 2026 16:18:30 +1200 Subject: [PATCH] Serve recent stages from SQLite with configurable background refresh --- backend/app/config.py | 1 + backend/app/db.py | 17 +++++++- backend/app/main.py | 2 + backend/app/routers/admin.py | 9 ++++ backend/app/routers/requests.py | 66 ++++++++++++++++++++--------- backend/app/runtime.py | 1 + backend/tests/test_recent_stages.py | 38 +++++++++++++++-- frontend/app/admin/SettingsPage.tsx | 5 ++- 8 files changed, 115 insertions(+), 24 deletions(-) diff --git a/backend/app/config.py b/backend/app/config.py index a5a2640..4cb4629 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -67,6 +67,7 @@ class Settings(BaseSettings): requests_sync_ttl_minutes: int = Field( default=1440, validation_alias=AliasChoices("REQUESTS_SYNC_TTL_MINUTES") ) + requests_stage_refresh_minutes: int = Field(default=15, ge=1, le=1440, validation_alias=AliasChoices("REQUESTS_STAGE_REFRESH_MINUTES")) requests_poll_interval_seconds: int = Field( default=300, validation_alias=AliasChoices("REQUESTS_POLL_INTERVAL_SECONDS") ) diff --git a/backend/app/db.py b/backend/app/db.py index 86cae3b..7983723 100644 --- a/backend/app/db.py +++ b/backend/app/db.py @@ -187,6 +187,7 @@ def _has_secure_bootstrap_admin_credentials() -> bool: def init_db() -> None: with _connect() as conn: + conn.execute("CREATE TABLE IF NOT EXISTS request_stage_cache (request_id INTEGER PRIMARY KEY, source_updated TEXT, ready INTEGER NOT NULL, checked_at REAL NOT NULL)") conn.execute("""CREATE TABLE IF NOT EXISTS user_duplicate_repairs ( id INTEGER PRIMARY KEY AUTOINCREMENT, kept_user_id INTEGER NOT NULL, archive_json TEXT NOT NULL, repaired_by TEXT NOT NULL, repaired_at TEXT NOT NULL)""") @@ -855,6 +856,7 @@ def save_action( ) -> None: created_at = datetime.now(timezone.utc).isoformat() with _connect() as conn: + conn.execute('UPDATE request_stage_cache SET checked_at = 0 WHERE request_id = ?', (request_id,)) conn.execute( """ INSERT INTO actions (request_id, action_id, label, status, message, created_at) @@ -2879,7 +2881,7 @@ def get_cached_requests_since(since_iso: str) -> list[Dict[str, Any]]: rows = conn.execute( """ SELECT request_id, media_id, media_type, status, title, year, requested_by, - requested_by_norm, requested_by_id, created_at + requested_by_norm, requested_by_id, created_at, updated_at FROM requests_cache WHERE created_at >= ? ORDER BY created_at DESC, request_id DESC @@ -2900,6 +2902,7 @@ def get_cached_requests_since(since_iso: str) -> list[Dict[str, Any]]: "requested_by_norm": row[7], "requested_by_id": row[8], "created_at": row[9], + "updated_at": row[10], } ) return results @@ -4027,3 +4030,15 @@ def cleanup_history(days: int) -> Dict[str, int]: (cutoff,), ).rowcount return {"actions": actions, "snapshots": snapshots} + + +def get_request_stage_cache(): + with _connect() as conn: + return {row[0]: {'source_updated': row[1], 'ready': bool(row[2]), 'checked_at': row[3]} + for row in conn.execute('SELECT request_id, source_updated, ready, checked_at FROM request_stage_cache')} + + +def save_request_stage_cache(rows): + with _connect() as conn: + conn.executemany('INSERT OR REPLACE INTO request_stage_cache (request_id, source_updated, ready, checked_at) VALUES (?, ?, ?, ?)', rows) + conn.execute('DELETE FROM request_stage_cache WHERE request_id NOT IN (SELECT request_id FROM requests_cache)') diff --git a/backend/app/main.py b/backend/app/main.py index 0459ceb..4b41b5e 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -13,6 +13,7 @@ from .db import has_admin_user, init_db from .routers.requests import ( router as requests_router, startup_warmup_requests_cache, + run_local_request_stage_loop, run_requests_delta_loop, run_daily_requests_full_sync, run_daily_db_cleanup, @@ -267,6 +268,7 @@ async def startup() -> None: return _launch_background_task("jellyfin-sync", run_daily_jellyfin_sync) _launch_background_task("requests-warmup", startup_warmup_requests_cache) + _launch_background_task("request-local-stages", run_local_request_stage_loop) _launch_background_task("requests-delta-loop", run_requests_delta_loop) _launch_background_task("requests-full-sync", run_daily_requests_full_sync) _launch_background_task("db-cleanup", run_daily_db_cleanup) diff --git a/backend/app/routers/admin.py b/backend/app/routers/admin.py index 550946e..b568927 100644 --- a/backend/app/routers/admin.py +++ b/backend/app/routers/admin.py @@ -248,6 +248,7 @@ SETTING_KEYS: List[str] = [ "log_background_sync_level", "requests_sync_ttl_minutes", "requests_poll_interval_seconds", + "requests_stage_refresh_minutes", "requests_delta_sync_interval_minutes", "requests_full_sync_time", "requests_cleanup_time", @@ -683,6 +684,14 @@ async def update_settings(payload: Dict[str, Any]) -> Dict[str, Any]: changed_keys.append(key) continue value_to_store = str(value).strip() if isinstance(value, str) else str(value) + if key == "requests_stage_refresh_minutes": + try: + interval = int(value_to_store) + except (TypeError, ValueError) as exc: + raise HTTPException(status_code=400, detail="Local stage refresh must be a whole number from 1 to 1440 minutes") from exc + if not 1 <= interval <= 1440: + raise HTTPException(status_code=400, detail="Local stage refresh must be from 1 to 1440 minutes") + value_to_store = str(interval) if key == "issue_confirmation_contact_attempts": try: attempts = int(value_to_store) diff --git a/backend/app/routers/requests.py b/backend/app/routers/requests.py index e7add34..2e5bebf 100644 --- a/backend/app/routers/requests.py +++ b/backend/app/routers/requests.py @@ -23,6 +23,8 @@ from ..auth import get_current_user from ..runtime import get_runtime_settings from .images import cache_tmdb_image, is_tmdb_cached from ..db import ( + get_request_stage_cache, + save_request_stage_cache, add_portal_item_activity, get_portal_item, save_action, @@ -157,6 +159,7 @@ async def _request_is_available_in_jellyfin( media_type: Optional[str], request_payload: Optional[Dict[str, Any]], availability_cache: Dict[str, bool], + raise_errors: bool = False, ) -> bool: if not jellyfin.configured() or not title: return False @@ -168,6 +171,8 @@ async def _request_is_available_in_jellyfin( try: search = await jellyfin.search_items(title, types, limit=50) except Exception: + if raise_errors: + raise availability_cache[cache_key] = False return False if isinstance(search, dict): @@ -2741,15 +2746,8 @@ async def recent_requests( runtime = get_runtime_settings() client = JellyseerrClient(runtime.jellyseerr_base_url, runtime.jellyseerr_api_key) mode = (runtime.requests_data_source or "prefer_cache").lower() - allow_remote = mode == "always_js" - if allow_remote: - if not client.configured(): - raise HTTPException(status_code=400, detail="Seerr not configured") - try: - await _ensure_requests_cache(client) - except httpx.HTTPStatusError as exc: - raise HTTPException(status_code=502, detail=str(exc)) from exc - + # Browsing is always local. Synchronization is owned by background workers. + allow_remote = False username_norm = _normalize_username(user.get("username", "")) requested_by_id = user.get("jellyseerr_user_id") requested_by = None if user.get("role") == "admin" else username_norm @@ -2773,7 +2771,8 @@ async def recent_requests( matched = 0 cache_mode = (runtime.artwork_cache_mode or "remote").lower() allow_title_hydrate = False - allow_artwork_hydrate = client.configured() + allow_artwork_hydrate = False + stage_cache = await asyncio.to_thread(get_request_stage_cache) jellyfin = JellyfinClient(runtime.jellyfin_base_url, runtime.jellyfin_api_key) jellyfin_cache: Dict[str, bool] = {} results = [] @@ -2787,7 +2786,7 @@ async def recent_requests( ) year = row.get("year") details = None - if row.get("request_id") and mode != "always_js": + if row.get("request_id"): cached_payload = get_request_cache_payload(int(row["request_id"])) if isinstance(cached_payload, dict): details = cached_payload @@ -2874,14 +2873,8 @@ async def recent_requests( status = 5 status_label = "Repair in progress" elif status_label in {"Working on it", "Ready to watch", "Partially ready"}: - is_available = await _request_is_available_in_jellyfin( - jellyfin, - title, - year, - row.get("media_type"), - details if isinstance(details, dict) else None, - jellyfin_cache, - ) + saved = stage_cache.get(row.get("request_id")) or {} + is_available = bool(saved.get("ready")) and saved.get("source_updated") == row.get("updated_at") status_label = _status_label_with_jellyfin(status, is_available) if status_label == STATUS_LABELS[4]: status = 4 @@ -3755,3 +3748,38 @@ async def action_grab( save_action, request_id, "grab", "Download selected release", "failed", failure_message ) raise HTTPException(status_code=502, detail=failure_message) + + +async def refresh_local_request_stages(): + runtime = get_runtime_settings() + jellyfin = JellyfinClient(runtime.jellyfin_base_url, runtime.jellyfin_api_key) + if not jellyfin.configured(): + return + rows = await asyncio.to_thread(get_cached_requests_since, (datetime.now(timezone.utc) - timedelta(days=RECENT_CACHE_MAX_DAYS)).isoformat()) + saved = await asyncio.to_thread(get_request_stage_cache) + interval = max(1, min(1440, int(runtime.requests_stage_refresh_minutes))) * 60 + now = time.time() + due = [row for row in rows if row.get('status') == 5 and + (now - (saved.get(row['request_id']) or {}).get('checked_at', 0) >= interval or + (saved.get(row['request_id']) or {}).get('source_updated') != row.get('updated_at'))] + semaphore = asyncio.Semaphore(4) + async def check(row): + async with semaphore: + payload = await asyncio.to_thread(get_request_cache_payload, row['request_id']) + try: + ready = await _request_is_available_in_jellyfin(jellyfin, row.get('title'), row.get('year'), row.get('media_type'), payload, {}, raise_errors=True) + return (row['request_id'], row.get('updated_at'), int(ready), time.time()) + except Exception: + logger.warning('Local request stage refresh failed request_id=%s', row['request_id']) + return None + checked = await asyncio.gather(*(check(row) for row in due)) + await asyncio.to_thread(save_request_stage_cache, [row for row in checked if row is not None]) + + +async def run_local_request_stage_loop(): + while True: + try: + await refresh_local_request_stages() + except Exception: + logger.exception('Local request stage refresh failed') + await asyncio.sleep(30) diff --git a/backend/app/runtime.py b/backend/app/runtime.py index 9f55e16..10eddcc 100644 --- a/backend/app/runtime.py +++ b/backend/app/runtime.py @@ -17,6 +17,7 @@ _INT_FIELDS = { "log_file_backup_count", "requests_sync_ttl_minutes", "requests_poll_interval_seconds", + "requests_stage_refresh_minutes", "requests_delta_sync_interval_minutes", "requests_cleanup_days", "issue_confirmation_contact_attempts", diff --git a/backend/tests/test_recent_stages.py b/backend/tests/test_recent_stages.py index 5a15799..87f3399 100644 --- a/backend/tests/test_recent_stages.py +++ b/backend/tests/test_recent_stages.py @@ -1,6 +1,6 @@ import unittest from types import SimpleNamespace -from unittest.mock import AsyncMock, patch +from unittest.mock import AsyncMock, AsyncMock, patch from contextlib import ExitStack from backend.app.routers import requests @@ -14,11 +14,11 @@ class RecentStageTests(unittest.IsolatedAsyncioTestCase): requested_by_id=10) for i, status in [(1,5),(2,5),(3,4),(4,6),(5,5),(6,2),(7,1),(8,3)]] async def available(client, title, *args): return title in {'1','3','4','5'} with ExitStack() as stack: - for name, value in [('get_runtime_settings', runtime), ('_recent_cache_stale', False), + for name, value in [('get_runtime_settings', runtime), ('get_request_stage_cache', {1:{'ready':True},3:{'ready':True},4:{'ready':True},5:{'ready':True}}), ('_recent_cache_stale', False), ('active_repair_request_ids', {'5'}), ('get_request_cache_payload', None)]: stack.enter_context(patch.object(requests, name, return_value=value)) stack.enter_context(patch.dict(requests._recent_cache, {'items':rows})) - stack.enter_context(patch.object(requests, '_request_is_available_in_jellyfin', new=available)) + stack.enter_context(patch.object(requests, '_request_is_available_in_jellyfin', new=AsyncMock(side_effect=AssertionError('Recent requests must not call Jellyfin')))) user={'role':'user','username':'viewer','jellyseerr_user_id':10} expected={'working':[2,5], 'ready':[1,3], 'partial':[4], 'approved':[6], 'pending':[7], 'declined':[8], 'in_progress':[2,4,5,6]} @@ -31,3 +31,35 @@ class RecentStageTests(unittest.IsolatedAsyncioTestCase): self.assertEqual(result['results'][0]['status'],4) result=await requests.recent_requests(take=20,skip=0,days=0,stage='all',user={**user,'jellyseerr_user_id':99}) self.assertEqual(result['results'],[]) + + async def test_background_refresh_skips_fresh_rows_and_preserves_failures(self): + import time + rows=[{'request_id':i,'status':5,'title':str(i),'updated_at':'v1'} for i in [1,2,3]] + runtime=SimpleNamespace(jellyfin_base_url='http://jellyfin',jellyfin_api_key='test',requests_stage_refresh_minutes=15) + with ExitStack() as stack: + stack.enter_context(patch.object(requests,'get_runtime_settings',return_value=runtime)) + stack.enter_context(patch.object(requests,'get_cached_requests_since',return_value=rows)) + stack.enter_context(patch.object(requests,'get_request_stage_cache',return_value={1:{'source_updated':'v1','checked_at':time.time()}})) + stack.enter_context(patch.object(requests,'get_request_cache_payload',return_value={})) + check=stack.enter_context(patch.object(requests,'_request_is_available_in_jellyfin',new=AsyncMock(side_effect=[True,RuntimeError('offline')])) ) + save=stack.enter_context(patch.object(requests,'save_request_stage_cache')) + await requests.refresh_local_request_stages() + self.assertEqual(check.await_count,2) + written=save.call_args.args[0] + self.assertEqual(len(written),1) + self.assertEqual(written[0][:3],(2,'v1',1)) + + +from backend.tests.test_backend_quality import TempDatabaseMixin +from backend.app import db + +class StagePersistenceTests(TempDatabaseMixin, unittest.TestCase): + def test_saved_stages_survive_initialization_and_actions_mark_due(self): + with db._connect() as conn: + conn.execute("INSERT INTO requests_cache (request_id, payload_json) VALUES (42, '{}')") + db.save_request_stage_cache([(42,'v1',1,12345)]) + db.init_db() + self.assertTrue(db.get_request_stage_cache()[42]['ready']) + db.save_action('42','search_releases','Search','ok') + self.assertEqual(db.get_request_stage_cache()[42]['checked_at'],0) + self.assertTrue(db.get_request_stage_cache()[42]['ready']) diff --git a/frontend/app/admin/SettingsPage.tsx b/frontend/app/admin/SettingsPage.tsx index a35ff58..b9ff984 100644 --- a/frontend/app/admin/SettingsPage.tsx +++ b/frontend/app/admin/SettingsPage.tsx @@ -82,6 +82,7 @@ const NUMBER_SETTINGS = new Set([ 'log_file_backup_count', 'requests_sync_ttl_minutes', 'requests_poll_interval_seconds', + 'requests_stage_refresh_minutes', 'requests_delta_sync_interval_minutes', 'requests_cleanup_days', 'issue_confirmation_contact_attempts', @@ -386,7 +387,7 @@ const STANDARD_SECTION_GROUPS: Record< key: 'requests-advanced', title: 'Advanced scheduling', description: 'How frequently the background worker checks whether a full refresh is due.', - keys: ['requests_poll_interval_seconds'], + keys: ['requests_poll_interval_seconds', 'requests_stage_refresh_minutes'], }, { key: 'requests-sync', @@ -505,6 +506,7 @@ const SETTING_LABEL_OVERRIDES: Record = { qbittorrent_username: 'Web UI username', qbittorrent_password: 'Web UI password', requests_sync_ttl_minutes: 'Request cache freshness (minutes)', + requests_stage_refresh_minutes: 'Local stage refresh (minutes)', requests_poll_interval_seconds: 'Full-sync eligibility check (seconds)', requests_delta_sync_interval_minutes: 'Recent-change sync interval (minutes)', requests_full_sync_time: 'Daily full-sync time', @@ -1027,6 +1029,7 @@ export default function SettingsPage({ section }: SettingsPageProps) { 'qBittorrent server URL for download status (FQDN or IP). Scheme is optional.', qbittorrent_username: 'qBittorrent login username.', qbittorrent_password: 'qBittorrent login password.', + requests_stage_refresh_minutes: 'Refresh saved request stages in the background. Default: 15 minutes; range: 1 to 1440. Shorter intervals increase Jellyfin traffic and server load. Recent requests always load from the local database.', requests_sync_ttl_minutes: 'How long saved requests stay fresh before a refresh is needed.', requests_poll_interval_seconds: 'How often Magent checks if a full refresh should run.',