video: real TTL+LRU cache (thread-safe) + cache search — the kettui way

The inline dict cache was a latent bug: no lock (the engine singleton is hit by
concurrent Flask + worker threads) and a wholesale-clear cliff at 256 (nuked hot
entries). Extracted a thread-safe TTL+LRU TTLCache into an importable core/ module
with seam tests (expiry via injected clock, LRU-not-wholesale eviction, a
concurrency stress test). Engine now uses it; search is cached too (60s, ownership
re-stamped fresh). Deliberately NOT persisted to disk — durable data already lives
in video.db; that tier would be over-engineering for a self-hosted app.

core/video/enrichment/cache.py

@@ -0,0 +1,50 @@
+"""A small thread-safe TTL + LRU cache for the video enrichment engine.
+
+The engine is a process-wide singleton hit concurrently by Flask request threads
+AND the worker threads, so the cache must be locked. Entries expire after a TTL;
+when the cache is full the LEAST-RECENTLY-USED entry is evicted (not the whole
+cache wholesale). Isolated: imports only the stdlib; no music, no DB.
+"""
+
+from __future__ import annotations
+
+import threading
+import time
+from collections import OrderedDict
+
+
+class TTLCache:
+    def __init__(self, maxsize: int = 256, ttl: float = 1800.0, clock=time.monotonic):
+        self._max = max(1, int(maxsize))
+        self._ttl = float(ttl)
+        self._clock = clock                      # injectable for tests
+        self._lock = threading.Lock()
+        self._data: OrderedDict = OrderedDict()  # key -> (expires_at, value)
+
+    def get(self, key):
+        now = self._clock()
+        with self._lock:
+            hit = self._data.get(key)
+            if hit is None:
+                return None
+            if hit[0] <= now:                    # expired
+                del self._data[key]
+                return None
+            self._data.move_to_end(key)          # mark recently used
+            return hit[1]
+
+    def put(self, key, value, ttl: float | None = None) -> None:
+        expires_at = self._clock() + (self._ttl if ttl is None else float(ttl))
+        with self._lock:
+            self._data[key] = (expires_at, value)
+            self._data.move_to_end(key)
+            while len(self._data) > self._max:   # evict LRU (oldest), not everything
+                self._data.popitem(last=False)
+
+    def clear(self) -> None:
+        with self._lock:
+            self._data.clear()
+
+    def __len__(self) -> int:
+        with self._lock:
+            return len(self._data)

core/video/enrichment/engine.py

@@ -11,6 +11,7 @@ import threading
 
 from utils.logging_config import get_logger
 
+from .cache import TTLCache
 from .worker import VideoEnrichmentWorker
 
 logger = get_logger("video_enrichment.engine")
@@ -28,25 +29,20 @@ class VideoEnrichmentEngine:
         # OMDb ratings (IMDb/RT/Metacritic) — not a matcher, so not a worker;
         # backfilled on the lazy detail refresh.
         self.ratings_client = ratings_client
-        # Short-TTL cache for live TMDB detail extras / preview payloads, so
-        # re-opening a title is instant instead of re-hitting TMDB.
-        self._tmdb_cache = {}
+        # Thread-safe TTL+LRU cache for live TMDB detail extras / preview payloads /
+        # person pages / seasons / trending, so re-opening a title is instant
+        # instead of re-hitting TMDB. (Volatile by design — durable art/episodes/
+        # ratings live in video.db; we don't persist this tier.)
+        self._cache = TTLCache(maxsize=256, ttl=1800)
         # Restore each worker's persisted pause state (survives restart).
         for w in self.workers.values():
             w.restore_paused()
 
     def _cache_get(self, key):
-        import time
-        hit = self._tmdb_cache.get(key)
-        if hit and hit[0] > time.monotonic():
-            return hit[1]
-        return None
+        return self._cache.get(key)
 
     def _cache_put(self, key, data, ttl=1800):
-        import time
-        if len(self._tmdb_cache) > 256:        # cheap bound — clear wholesale
-            self._tmdb_cache.clear()
-        self._tmdb_cache[key] = (time.monotonic() + ttl, data)
+        self._cache.put(key, data, ttl=ttl)
 
     def _backfill_ratings(self, kind, item_id):
         # The OMDb worker owns the ratings client (fallback to an injected one
@@ -263,11 +259,17 @@ class VideoEnrichmentEngine:
         w = self.workers.get("tmdb")
         if not w or not w.enabled:
             return []
-        try:
-            results = w.client.search(query) or []
-        except Exception:
-            logger.exception("video search failed for %r", query)
-            return []
+        # Short TTL — identical queries within ~a minute reuse the result; ownership
+        # is re-stamped fresh below so 'In Library' badges stay current.
+        key = ("search", (query or "").strip().lower())
+        results = self._cache_get(key)
+        if results is None:
+            try:
+                results = w.client.search(query) or []
+                self._cache_put(key, results, ttl=60)
+            except Exception:
+                logger.exception("video search failed for %r", query)
+                return []
         for r in results:
             if r.get("kind") in ("movie", "show") and r.get("tmdb_id"):
                 r["library_id"] = self.db.library_id_for_tmdb(r["kind"], r["tmdb_id"])

tests/test_video_cache.py

@@ -0,0 +1,61 @@
+"""Seam tests for the video enrichment TTL+LRU cache (importable core/ module)."""
+
+from __future__ import annotations
+
+import threading
+
+from core.video.enrichment.cache import TTLCache
+
+
+def test_get_miss_and_hit():
+    c = TTLCache(maxsize=4, ttl=100)
+    assert c.get("a") is None
+    c.put("a", 1)
+    assert c.get("a") == 1
+
+
+def test_ttl_expiry_with_injected_clock():
+    t = {"now": 0.0}
+    c = TTLCache(maxsize=4, ttl=10, clock=lambda: t["now"])
+    c.put("a", 1)
+    t["now"] = 9.9
+    assert c.get("a") == 1            # still fresh
+    t["now"] = 10.1
+    assert c.get("a") is None         # expired → evicted
+    assert len(c) == 0
+
+
+def test_per_put_ttl_override():
+    t = {"now": 0.0}
+    c = TTLCache(ttl=1000, clock=lambda: t["now"])
+    c.put("a", 1, ttl=5)
+    t["now"] = 6
+    assert c.get("a") is None
+
+
+def test_lru_eviction_not_wholesale():
+    c = TTLCache(maxsize=2, ttl=100)
+    c.put("a", 1)
+    c.put("b", 2)
+    assert c.get("a") == 1            # touch 'a' → most-recently-used
+    c.put("c", 3)                     # over capacity → evict LRU ('b'), NOT everything
+    assert c.get("a") == 1            # survived
+    assert c.get("c") == 3            # survived
+    assert c.get("b") is None         # the only one evicted
+    assert len(c) == 2
+
+
+def test_thread_safe_under_concurrency():
+    c = TTLCache(maxsize=64, ttl=100)
+
+    def worker(n):
+        for i in range(500):
+            c.put(("k", (n * 500 + i) % 80), i)
+            c.get(("k", i % 80))
+
+    threads = [threading.Thread(target=worker, args=(n,)) for n in range(8)]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join()
+    assert len(c) <= 64               # bound held, no crash/corruption