web/services/cache.py

"""Cache TTL+LRU thread-safe per i documenti elaborati.

Lo stato di sessione vive qui (nella cache), non nel server: ogni documento e'
identificato dal proprio hash, cosi' documenti identici non vengono rielaborati.
"""

from __future__ import annotations

import threading
import time
from collections import OrderedDict
from dataclasses import dataclass
from typing import Generic, TypeVar

T = TypeVar("T")


@dataclass(slots=True)
class _Entry(Generic[T]):
    value: T
    expires_at: float


class TTLCache(Generic[T]):
    def __init__(self, max_items: int = 256, ttl: int = 3600) -> None:
        self._max = max_items
        self._ttl = ttl
        self._data: OrderedDict[str, _Entry[T]] = OrderedDict()
        self._lock = threading.RLock()

    def get(self, key: str) -> T | None:
        with self._lock:
            entry = self._data.get(key)
            if entry is None:
                return None
            if entry.expires_at < time.time():
                del self._data[key]
                return None
            self._data.move_to_end(key)
            return entry.value

    def set(self, key: str, value: T) -> None:
        with self._lock:
            self._data[key] = _Entry(value=value, expires_at=time.time() + self._ttl)
            self._data.move_to_end(key)
            while len(self._data) > self._max:
                self._data.popitem(last=False)

    def delete_prefix(self, prefix: str) -> int:
        """Rimuove tutte le voci la cui chiave inizia con `prefix`.

        Usato per invalidare i render in cache di un documento (chiavi
        `{doc_id}|...`) quando il documento viene rianalizzato da zero.
        Ritorna il numero di voci rimosse.
        """
        with self._lock:
            keys = [k for k in self._data if k.startswith(prefix)]
            for k in keys:
                del self._data[k]
            return len(keys)

    def has(self, key: str) -> bool:
        """Verifica l'esistenza SENZA promuovere la voce nell'ordine LRU."""
        with self._lock:
            entry = self._data.get(key)
            if entry is None:
                return False
            if entry.expires_at < time.time():
                del self._data[key]
                return False
            return True

    def __len__(self) -> int:
        with self._lock:
            return len(self._data)