shelfmark/shelfmark/core/cache.py

"""Thread-safe in-memory cache with TTL support."""

import threading
import time
from dataclasses import dataclass
from functools import wraps
from typing import TYPE_CHECKING, ParamSpec, TypeVar, cast

from shelfmark.core.logger import setup_logger

if TYPE_CHECKING:
    from collections.abc import Callable

logger = setup_logger(__name__)

P = ParamSpec("P")
R = TypeVar("R")


@dataclass
class CacheEntry:
    """A cached value with expiration time."""

    value: object
    expires_at: float


class CacheService:
    """Thread-safe in-memory cache with TTL support."""

    def __init__(self, max_size: int = 1000) -> None:
        """Initialize cache with max_size entries before eviction."""
        self._cache: dict[str, CacheEntry] = {}
        self._lock = threading.Lock()
        self._max_size = max_size

    def get(self, key: str) -> object | None:
        """Get cached value if not expired."""
        with self._lock:
            entry = self._cache.get(key)
            if entry is None:
                return None

            if time.time() > entry.expires_at:
                del self._cache[key]
                return None

            return entry.value

    def set(self, key: str, value: object, ttl: int) -> None:
        """Cache value with TTL in seconds."""
        with self._lock:
            # Evict oldest entries if at capacity
            if len(self._cache) >= self._max_size:
                self._evict_oldest()

            self._cache[key] = CacheEntry(value=value, expires_at=time.time() + ttl)

    def invalidate(self, key: str) -> bool:
        """Remove specific cache entry. Returns True if found."""
        with self._lock:
            if key in self._cache:
                del self._cache[key]
                return True
            return False

    def invalidate_prefix(self, prefix: str) -> int:
        """Remove all cache entries whose keys start with prefix."""
        with self._lock:
            matching_keys = [key for key in self._cache if key.startswith(prefix)]
            for key in matching_keys:
                del self._cache[key]
            return len(matching_keys)

    def clear(self) -> None:
        """Clear all cache entries."""
        with self._lock:
            self._cache.clear()

    def cleanup_expired(self) -> int:
        """Remove all expired entries. Returns count removed."""
        with self._lock:
            now = time.time()
            expired_keys = [key for key, entry in self._cache.items() if entry.expires_at < now]
            for key in expired_keys:
                del self._cache[key]
            return len(expired_keys)

    def _evict_oldest(self) -> None:
        """Evict ~10% of oldest entries. Called with lock held."""
        if not self._cache:
            return

        # Remove ~10% of entries, oldest first
        entries_to_remove = max(1, len(self._cache) // 10)
        sorted_entries = sorted(self._cache.items(), key=lambda x: x[1].expires_at)

        for key, _ in sorted_entries[:entries_to_remove]:
            del self._cache[key]

    def stats(self) -> dict[str, int]:
        """Get cache statistics (size, max_size)."""
        with self._lock:
            return {"size": len(self._cache), "max_size": self._max_size}


# Global cache instance for metadata providers
_metadata_cache = CacheService(max_size=1000)


def get_metadata_cache() -> CacheService:
    """Get the global metadata cache instance."""
    return _metadata_cache


def cache_key(*args: object, **kwargs: object) -> str:
    """Generate cache key from arguments."""
    parts = [str(arg) for arg in args]
    parts.extend(f"{k}={v}" for k, v in sorted(kwargs.items()))
    return ":".join(parts)


def _coerce_ttl_seconds(value: object, *, default: int) -> int:
    """Normalize cache TTL values read from config or decorator arguments."""
    if isinstance(value, bool):
        return default
    if isinstance(value, int):
        return value if value > 0 else default
    if isinstance(value, str):
        stripped = value.strip()
        if stripped.isdigit():
            parsed = int(stripped)
            return parsed if parsed > 0 else default
    return default


def cacheable(
    ttl: int | None = None,
    ttl_key: str | None = None,
    ttl_default: int = 300,
    key_prefix: str = "",
) -> Callable[[Callable[P, R]], Callable[P, R]]:
    """Cache function results with a static or config-backed TTL."""

    def decorator(func: Callable[P, R]) -> Callable[P, R]:
        @wraps(func)
        def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            # Check if metadata caching is enabled
            from shelfmark.core.config import config

            if not config.get("METADATA_CACHE_ENABLED", True):
                # Caching disabled, execute function directly
                return func(*args, **kwargs)

            # Determine TTL: static or from config
            if ttl is not None:
                effective_ttl = ttl
            elif ttl_key:
                effective_ttl = _coerce_ttl_seconds(
                    config.get(ttl_key, ttl_default),
                    default=ttl_default,
                )
            else:
                effective_ttl = ttl_default

            # Generate cache key from function name and arguments
            # Skip 'self' argument if present (first arg of method)
            cache_args = args[1:] if args and hasattr(args[0], func.__name__) else args

            key = cache_key(key_prefix or func.__name__, *cache_args, **kwargs)

            # Check cache
            cached = _metadata_cache.get(key)
            if cached is not None:
                return cast("R", cached)

            # Execute function and cache result
            result = func(*args, **kwargs)

            # Only cache non-None results
            if result is not None:
                _metadata_cache.set(key, result, effective_ttl)

            return result

        return wrapper

    return decorator