src/api/cache.py

"""Tiny in-process TTL cache for read-path responses.

The cloud backend talks to Turso / Qdrant / R2 across regions; every call is
an HTTP round-trip. The read endpoints (book list, book detail, status
counts) are recomputed on every navigation even though the underlying data
barely changes between clicks — that's the bulk of the "moving between pages
is slow" latency the owner reported.

A short-TTL in-memory cache collapses repeat reads to a single process-local
lookup. The HF Space runs a single replica, so a process-local dict is
correct; user-facing writes (create / patch / delete book, label edits)
invalidate by key or key-prefix so a read never serves data the same user
just changed. The TTL is the backstop for changes that happen outside the
request path (e.g. a pipeline job flipping a book's status) — those become
visible within ``DEFAULT_TTL`` seconds.

Keep it dumb: no LRU eviction (the key space is tiny — a handful of books ×
filter combinations), no async, no cross-process coherence.
"""

from __future__ import annotations

import threading
import time
from typing import Callable, TypeVar

DEFAULT_TTL = 30.0  # seconds

_T = TypeVar("_T")


class _Entry:
    __slots__ = ("value", "expires")

    def __init__(self, value: object, expires: float) -> None:
        self.value = value
        self.expires = expires


_store: dict[str, _Entry] = {}
_lock = threading.Lock()


def get_or_set(key: str, producer: Callable[[], _T], *, ttl: float = DEFAULT_TTL) -> _T:
    """Return the cached value for ``key`` or compute, store, and return it.

    ``producer`` runs OUTSIDE the lock so concurrent misses don't serialize
    behind one another's DB calls (a brief duplicate-compute on a cold key is
    cheaper than holding the lock across a cross-region round-trip).
    """
    now = time.monotonic()
    with _lock:
        e = _store.get(key)
        if e is not None and e.expires > now:
            return e.value  # type: ignore[return-value]
    value = producer()
    with _lock:
        _store[key] = _Entry(value, time.monotonic() + ttl)
    return value


def invalidate(key: str) -> None:
    with _lock:
        _store.pop(key, None)


def invalidate_prefix(prefix: str) -> None:
    with _lock:
        for k in [k for k in _store if k.startswith(prefix)]:
            _store.pop(k, None)


def clear() -> None:
    """Drop everything — test hook + a hard reset for cache-coherence bugs."""
    with _lock:
        _store.clear()