""" Cloud cache service for DocuVault — Phase 12 in-memory layer + Phase 14 durable cache. ── Phase 12: in-memory folder listing cache ───────────────────────────────────── Provides a module-level TTLCache singleton for caching cloud provider folder listings with a 60-second TTL (D-16: live API calls with 60-second in-memory TTL). Thread-safety: cachetools.TTLCache is NOT thread-safe by itself. A threading.Lock is required for all reads and writes (RESEARCH.md Pattern 8). The fetch function is called OUTSIDE the lock to prevent blocking the asyncio event loop while an outbound cloud API call is in flight. Cache key scheme: "{user_id}:{provider}:{folder_id}" - user_id ensures strict per-user isolation - provider namespace-separates entries from different cloud backends - folder_id identifies the specific folder whose listing is cached References: RESEARCH.md Pattern 8 — TTLCache thread safety + asyncio integration D-16 — 60-second TTL, in-memory cache (not Redis) ── Phase 14: durable byte cache (cloud_byte_cache_entries) ───────────────────── Exports create_cache_entry, evict_lru_entries, list_cache_entries, increment_quota_for_cache, decrement_quota_for_cache for use by Celery tasks and the analysis API. Also re-exports compute_version_key from services.cloud_analysis_versioning so tests and service callers have a single import location. Rules: - Raises ValueError or domain exceptions only — never HTTPException. - Eviction never removes entries with pin_count > 0 or active_job_count > 0. - Eviction is per-user LRU (oldest last_accessed_at first). - quota.used_bytes is adjusted only via increment_quota_for_cache / decrement_quota_for_cache which use the project's atomic UPDATE pattern. - object_key is never returned in API responses (T-14-02). """ from __future__ import annotations import threading import uuid from datetime import datetime, timezone from typing import Any, Callable, Awaitable, List, Optional, Sequence from cachetools import TTLCache from sqlalchemy import select, update, text from sqlalchemy.ext.asyncio import AsyncSession from db.models import CloudByteCacheEntry from services.cloud_analysis_versioning import compute_version_key as _compute_version_key # noqa: F401 # Re-export so callers can do: from services.cloud_cache import compute_version_key compute_version_key = _compute_version_key # Module-level singleton: maxsize=1000 (sufficient for ~50 users × 20 folders), # ttl=60 seconds per D-16. _folder_cache: TTLCache = TTLCache(maxsize=1000, ttl=60) # Lock required for all _folder_cache access (cachetools.TTLCache is not thread-safe) _folder_cache_lock = threading.Lock() async def get_cloud_folders_cached( user_id: str, provider: str, folder_id: str, fetch_fn: Callable[[], Awaitable[list]], ) -> list: """Return cached folder listing, or call fetch_fn and cache the result. The cache key is "{user_id}:{provider}:{folder_id}". The fetch_fn coroutine is awaited OUTSIDE the lock so that a slow cloud API call does not block other asyncio tasks from acquiring the lock. A race condition where two concurrent callers both miss the cache and both call fetch_fn is acceptable — the second result overwrites the first, and both callers receive consistent data. Args: user_id: The authenticated user's UUID string. provider: The cloud provider identifier (e.g. "google_drive"). folder_id: The provider-native folder/directory identifier. fetch_fn: An async callable (no arguments) that returns the folder listing list when called. Only invoked on cache miss. Returns: The folder listing list (from cache or fresh from fetch_fn). """ cache_key = f"{user_id}:{provider}:{folder_id}" # Check cache under lock with _folder_cache_lock: if cache_key in _folder_cache: return _folder_cache[cache_key] # Cache miss — call fetch_fn outside the lock to not block the event loop result = await fetch_fn() # Store result in cache under lock with _folder_cache_lock: _folder_cache[cache_key] = result return result # ══ Phase 14: durable byte cache operations ══════════════════════════════════ class CacheQuotaExceeded(ValueError): """User's quota cannot accommodate the requested cache entry size.""" async def create_cache_entry( session: AsyncSession, *, user_id: uuid.UUID, connection_id: uuid.UUID, cloud_item_id: uuid.UUID, provider_item_id: str, version_key: str, object_key: str, content_type: Optional[str] = None, size_bytes: int, content_hash: Optional[str] = None, pin_count: int = 0, active_job_count: int = 0, last_accessed_at: Optional[datetime] = None, ) -> CloudByteCacheEntry: """Create and flush a new cloud_byte_cache_entries row. Does NOT update quota — call increment_quota_for_cache separately. Does NOT enforce uniqueness beyond the DB constraint; callers must check for an existing entry before calling this function. Args: session: Active async SQLAlchemy session. user_id: Owner UUID. connection_id: Cloud connection UUID. cloud_item_id: CloudItem row UUID. provider_item_id: Snapshot of the provider item id for audit/debug. version_key: Idempotency key (from compute_version_key). object_key: Private MinIO object key — UUID-based, never exposed. content_type: Optional MIME type. size_bytes: Actual byte count stored in MinIO. content_hash: Optional post-hydration content hash. pin_count: Active preview/download leases (default 0). active_job_count: Active Celery jobs using these bytes (default 0). last_accessed_at: Optional LRU timestamp; defaults to now(). Returns: Flushed (but not committed) CloudByteCacheEntry row. """ now = datetime.now(timezone.utc) entry = CloudByteCacheEntry( user_id=user_id if isinstance(user_id, uuid.UUID) else uuid.UUID(str(user_id)), connection_id=connection_id if isinstance(connection_id, uuid.UUID) else uuid.UUID(str(connection_id)), cloud_item_id=cloud_item_id if isinstance(cloud_item_id, uuid.UUID) else uuid.UUID(str(cloud_item_id)), provider_item_id=provider_item_id, version_key=version_key, object_key=object_key, content_type=content_type, size_bytes=size_bytes, content_hash=content_hash, pin_count=pin_count, active_job_count=active_job_count, last_accessed_at=last_accessed_at or now, ) session.add(entry) await session.flush() return entry async def list_cache_entries( session: AsyncSession, *, user_id: uuid.UUID, include_evicted: bool = False, ) -> Sequence[CloudByteCacheEntry]: """Return cache entries owned by user_id. Ownership is strictly enforced — only rows where user_id matches are returned. By default, evicted entries (evicted_at IS NOT NULL) are excluded. Args: session: Active async SQLAlchemy session. user_id: Owner UUID. include_evicted: If True, include entries with evicted_at set. Returns: Sequence of CloudByteCacheEntry rows (may be empty). """ uid = user_id if isinstance(user_id, uuid.UUID) else uuid.UUID(str(user_id)) stmt = select(CloudByteCacheEntry).where(CloudByteCacheEntry.user_id == uid) if not include_evicted: stmt = stmt.where(CloudByteCacheEntry.evicted_at.is_(None)) result = await session.execute(stmt) return result.scalars().all() async def evict_lru_entries( session: AsyncSession, *, user_id: uuid.UUID, max_cache_bytes: int, ) -> List[uuid.UUID]: """Mark least-recently-used unpinned entries as evicted until total is within limit. Eviction rules (D-14, D-16): - Only entries with pin_count == 0 and active_job_count == 0 are eligible. - Entries are evicted in ascending last_accessed_at order (oldest first). - Eviction stops when the sum of retained entry size_bytes <= max_cache_bytes. - evicted_at is stamped to now(); the caller is responsible for deleting the MinIO objects and calling decrement_quota_for_cache per evicted entry. Args: session: Active async SQLAlchemy session. user_id: Owner UUID. max_cache_bytes: Target byte ceiling (entries evicted until we fit within). Returns: List of evicted CloudByteCacheEntry UUIDs (empty if nothing needed eviction). """ uid = user_id if isinstance(user_id, uuid.UUID) else uuid.UUID(str(user_id)) # Fetch all non-evicted entries for this user ordered by LRU ascending stmt = ( select(CloudByteCacheEntry) .where( CloudByteCacheEntry.user_id == uid, CloudByteCacheEntry.evicted_at.is_(None), ) .order_by(CloudByteCacheEntry.last_accessed_at.asc().nulls_first()) ) result = await session.execute(stmt) entries = result.scalars().all() # Sum total retained bytes total_bytes = sum(e.size_bytes for e in entries) now = datetime.now(timezone.utc) evicted_ids: List[uuid.UUID] = [] for entry in entries: if total_bytes <= max_cache_bytes: break # Skip pinned or actively-used entries if entry.pin_count > 0 or entry.active_job_count > 0: continue # Evict this entry entry.evicted_at = now entry.updated_at = now total_bytes -= entry.size_bytes evicted_ids.append(entry.id) if evicted_ids: await session.flush() return evicted_ids async def increment_quota_for_cache( session: AsyncSession, *, user_id: uuid.UUID, size_bytes: int, ) -> None: """Atomically increment quota.used_bytes for a cached byte entry. Uses the project's atomic UPDATE pattern to prevent TOCTOU race conditions: UPDATE quotas SET used_bytes = used_bytes + $delta WHERE user_id = $user_id AND (used_bytes + $delta) <= limit_bytes Raises: CacheQuotaExceeded: The increment would exceed the user's quota limit. Note: Against SQLite (test environments) this runs as a non-atomic update; the atomic guarantee is provided only by PostgreSQL in production. """ uid = user_id if isinstance(user_id, uuid.UUID) else uuid.UUID(str(user_id)) # Atomic UPDATE … RETURNING pattern (same as quota enforcement in upload flow) result = await session.execute( text( "UPDATE quotas SET used_bytes = used_bytes + :delta " "WHERE user_id = :uid AND (used_bytes + :delta) <= limit_bytes " "RETURNING used_bytes" ), {"delta": size_bytes, "uid": str(uid)}, ) row = result.fetchone() if row is None: raise CacheQuotaExceeded( f"User {uid} quota exhausted — cannot cache {size_bytes} bytes" ) async def decrement_quota_for_cache( session: AsyncSession, *, user_id: uuid.UUID, size_bytes: int, ) -> None: """Atomically decrement quota.used_bytes when a cached entry is evicted. Clamps to 0 to prevent negative quota values (defensive — should not occur in practice if increment/decrement are always called in pairs). Args: session: Active async SQLAlchemy session. user_id: Owner UUID. size_bytes: Byte count to release. """ uid = user_id if isinstance(user_id, uuid.UUID) else uuid.UUID(str(user_id)) await session.execute( text( "UPDATE quotas SET used_bytes = GREATEST(0, used_bytes - :delta) " "WHERE user_id = :uid" ), {"delta": size_bytes, "uid": str(uid)}, ) # ── Phase 12: in-memory folder listing cache ────────────────────────────────── def invalidate_provider_cache(user_id: str, provider: str) -> None: """Invalidate all cached folder listings for a specific user + provider. Called when a cloud connection is disconnected, credentials are updated, or any event that makes the cached listings stale. Args: user_id: The authenticated user's UUID string. provider: The cloud provider identifier to invalidate. """ prefix = f"{user_id}:{provider}:" with _folder_cache_lock: keys_to_delete = [k for k in list(_folder_cache.keys()) if k.startswith(prefix)] for key in keys_to_delete: del _folder_cache[key]