kite/backend/services/cloud_cache.py

"""
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, retain_or_reuse_cache_entry,
pin_cache_entry, release_cache_entry, update_cache_access 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.

Exports hydrate_and_cache_bytes for the open/preview/download routes (Plan 06):
  hydrate_and_cache_bytes provides the full cache-or-fetch lifecycle for content
  routes: cache hit avoids a provider byte fetch; cache miss fetches, stores,
  and pins the entry so eviction cannot occur mid-response.

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)},
    )


async def retain_or_reuse_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,
) -> tuple[CloudByteCacheEntry, bool]:
    """Look up a non-evicted cache entry matching the version_key; create one if absent.

    Implements the Phase 14 cache lifecycle step 3 (PATTERNS.md): reuse matching
    non-evicted entry before hydrating from provider.

    Ownership is strictly enforced — queries filter by (user_id, connection_id,
    cloud_item_id, version_key) so a foreign user can never reuse another user's entry.

    When creating a new entry this function does NOT call increment_quota_for_cache.
    The caller is responsible for reserving quota atomically before retaining bytes
    and calling increment_quota_for_cache before 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: Provider item id snapshot for audit.
        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:      Byte count stored in MinIO.
        content_hash:    Optional post-hydration content hash.

    Returns:
        Tuple (entry, created_new) where created_new=True if a new row was inserted,
        False if an existing non-evicted entry was reused.
    """
    uid = user_id if isinstance(user_id, uuid.UUID) else uuid.UUID(str(user_id))
    cid = connection_id if isinstance(connection_id, uuid.UUID) else uuid.UUID(str(connection_id))
    iid = cloud_item_id if isinstance(cloud_item_id, uuid.UUID) else uuid.UUID(str(cloud_item_id))

    now = datetime.now(timezone.utc)

    # Step 3: check for existing entry (evicted or not) for this version key.
    # The unique constraint on (user_id, connection_id, cloud_item_id, version_key) means
    # at most one row exists per version key.  We handle both states:
    #   - non-evicted → reuse (touch last_accessed_at)
    #   - evicted → reactivate (clear evicted_at, update object_key + size)
    stmt = (
        select(CloudByteCacheEntry)
        .where(
            CloudByteCacheEntry.user_id == uid,
            CloudByteCacheEntry.connection_id == cid,
            CloudByteCacheEntry.cloud_item_id == iid,
            CloudByteCacheEntry.version_key == version_key,
        )
        .limit(1)
    )
    result = await session.execute(stmt)
    existing = result.scalars().first()

    if existing is not None:
        if existing.evicted_at is None:
            # Active entry — reuse it
            existing.last_accessed_at = now
            existing.updated_at = now
            await session.flush()
            return existing, False
        else:
            # Evicted entry — reactivate with fresh bytes and updated object key
            existing.evicted_at = None
            existing.object_key = object_key
            existing.size_bytes = size_bytes
            existing.content_type = content_type
            existing.content_hash = content_hash
            existing.last_accessed_at = now
            existing.updated_at = now
            await session.flush()
            return existing, True

    # No existing entry — create a new one
    # Quota must be reserved by caller before bytes are stored in MinIO
    new_entry = await create_cache_entry(
        session,
        user_id=uid,
        connection_id=cid,
        cloud_item_id=iid,
        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,
    )
    return new_entry, True


async def pin_cache_entry(
    session: AsyncSession,
    *,
    entry_id: uuid.UUID,
    user_id: uuid.UUID,
) -> CloudByteCacheEntry:
    """Increment pin_count to mark an active preview or download lease.

    Pinned entries are protected from LRU eviction (PATTERNS.md step 7).
    Ownership is asserted — user_id must match the entry's user_id.

    Args:
        session:  Active async SQLAlchemy session.
        entry_id: CloudByteCacheEntry UUID.
        user_id:  Authenticated user UUID — must own the entry.

    Returns:
        Updated CloudByteCacheEntry.

    Raises:
        ValueError: Entry not found or does not belong to user_id.
    """
    uid = user_id if isinstance(user_id, uuid.UUID) else uuid.UUID(str(user_id))
    eid = entry_id if isinstance(entry_id, uuid.UUID) else uuid.UUID(str(entry_id))

    result = await session.execute(
        select(CloudByteCacheEntry).where(
            CloudByteCacheEntry.id == eid,
            CloudByteCacheEntry.user_id == uid,
        )
    )
    entry = result.scalars().first()
    if entry is None:
        raise ValueError(f"Cache entry {eid} not found for user {uid}")

    now = datetime.now(timezone.utc)
    entry.pin_count += 1
    entry.last_accessed_at = now
    entry.updated_at = now
    await session.flush()
    return entry


async def release_cache_entry(
    session: AsyncSession,
    *,
    entry_id: uuid.UUID,
    user_id: uuid.UUID,
) -> CloudByteCacheEntry:
    """Decrement pin_count when a preview or download lease ends.

    Clamped to 0 — defensive against double-release (PATTERNS.md step 8).

    Args:
        session:  Active async SQLAlchemy session.
        entry_id: CloudByteCacheEntry UUID.
        user_id:  Authenticated user UUID — must own the entry.

    Returns:
        Updated CloudByteCacheEntry.

    Raises:
        ValueError: Entry not found or does not belong to user_id.
    """
    uid = user_id if isinstance(user_id, uuid.UUID) else uuid.UUID(str(user_id))
    eid = entry_id if isinstance(entry_id, uuid.UUID) else uuid.UUID(str(entry_id))

    result = await session.execute(
        select(CloudByteCacheEntry).where(
            CloudByteCacheEntry.id == eid,
            CloudByteCacheEntry.user_id == uid,
        )
    )
    entry = result.scalars().first()
    if entry is None:
        raise ValueError(f"Cache entry {eid} not found for user {uid}")

    now = datetime.now(timezone.utc)
    entry.pin_count = max(0, entry.pin_count - 1)
    entry.updated_at = now
    await session.flush()
    return entry


async def update_cache_access(
    session: AsyncSession,
    *,
    entry_id: uuid.UUID,
    user_id: uuid.UUID,
) -> CloudByteCacheEntry:
    """Touch last_accessed_at to update LRU position.

    Called after a successful cache hit to maintain accurate LRU ordering.

    Args:
        session:  Active async SQLAlchemy session.
        entry_id: CloudByteCacheEntry UUID.
        user_id:  Authenticated user UUID — must own the entry.

    Returns:
        Updated CloudByteCacheEntry.

    Raises:
        ValueError: Entry not found or does not belong to user_id.
    """
    uid = user_id if isinstance(user_id, uuid.UUID) else uuid.UUID(str(user_id))
    eid = entry_id if isinstance(entry_id, uuid.UUID) else uuid.UUID(str(entry_id))

    result = await session.execute(
        select(CloudByteCacheEntry).where(
            CloudByteCacheEntry.id == eid,
            CloudByteCacheEntry.user_id == uid,
        )
    )
    entry = result.scalars().first()
    if entry is None:
        raise ValueError(f"Cache entry {eid} not found for user {uid}")

    now = datetime.now(timezone.utc)
    entry.last_accessed_at = now
    entry.updated_at = now
    await session.flush()
    return entry


# ── Phase 14: open/preview/download cache hydration ──────────────────────────


async def hydrate_and_cache_bytes(
    session: AsyncSession,
    *,
    user_id: uuid.UUID,
    connection_id: uuid.UUID,
    cloud_item_id: uuid.UUID,
    provider_item_id: str,
    version_key: str,
    content_type: Optional[str],
    fetch_fn: "Callable[[], Awaitable[bytes]]",
    minio_client: "Any",
    cache_limit_bytes: int,
) -> tuple[bytes, Optional[uuid.UUID]]:
    """Fetch content bytes for a cloud open/preview/download request, routing through the cache.

    Cache-hit path (PATTERNS.md step 3):
      - Locate a non-evicted CloudByteCacheEntry matching (user_id, connection_id,
        cloud_item_id, version_key).
      - Pin the entry before reading from MinIO so eviction cannot occur mid-response
        (PATTERNS.md step 7).
      - Touch last_accessed_at to update LRU position.
      - Read bytes from MinIO via object_key.
      - Release pin in a finally block (PATTERNS.md step 8).
      - Return (bytes, cache_entry_id).

    Cache-miss path:
      - Call fetch_fn() to download bytes from the provider.
      - Attempt to store bytes in MinIO under a UUID key (non-fatal on failure).
      - If MinIO write succeeds and quota allows: create/reactivate a cache entry
        and pin it (protect from eviction during response streaming).
      - Release pin in a finally block.
      - Return (bytes, cache_entry_id) — cache_entry_id is None if caching failed.

    Rules:
      - Provider bytes are NEVER fetched for the cache-hit path.
      - Pin is always released in a finally block, even if the caller raises.
      - MinIO write failure is non-fatal: bytes are returned from the provider
        fetch path regardless.
      - Quota failure is non-fatal: bytes are returned but the cache entry is not
        created (caller gets bytes without MinIO caching).
      - object_key is never returned in any form — it stays internal.
      - Raises ValueError only — never HTTPException.

    Args:
        session:           Active async SQLAlchemy session.
        user_id:           Authenticated user UUID.
        connection_id:     Cloud connection UUID.
        cloud_item_id:     CloudItem row UUID.
        provider_item_id:  Provider item ID (for audit/snapshot on cache entry).
        version_key:       Computed version key from compute_version_key.
        content_type:      MIME type of the file.
        fetch_fn:          Async callable (no args) that returns provider bytes.
        minio_client:      Object with async put_object(key, data, content_type)
                           and get_object(key) methods.
        cache_limit_bytes: User's configured byte cache limit; used for eviction
                           after releasing the pin.

    Returns:
        Tuple (bytes, cache_entry_id) where cache_entry_id is the UUID of the
        pinned cache entry, or None if no cache entry was created/used.
    """
    uid = user_id if isinstance(user_id, uuid.UUID) else uuid.UUID(str(user_id))
    cid = connection_id if isinstance(connection_id, uuid.UUID) else uuid.UUID(str(connection_id))
    iid = cloud_item_id if isinstance(cloud_item_id, uuid.UUID) else uuid.UUID(str(cloud_item_id))

    # ── Cache-hit path ──────────────────────────────────────────────────────
    # Look for a non-evicted entry matching the version key.
    stmt = (
        select(CloudByteCacheEntry)
        .where(
            CloudByteCacheEntry.user_id == uid,
            CloudByteCacheEntry.connection_id == cid,
            CloudByteCacheEntry.cloud_item_id == iid,
            CloudByteCacheEntry.version_key == version_key,
            CloudByteCacheEntry.evicted_at.is_(None),
        )
        .limit(1)
    )
    result = await session.execute(stmt)
    existing_entry = result.scalars().first()

    if existing_entry is not None:
        # Pin the entry before reading so eviction cannot occur mid-response.
        await pin_cache_entry(session, entry_id=existing_entry.id, user_id=uid)
        await session.commit()
        try:
            # Touch LRU position
            await update_cache_access(session, entry_id=existing_entry.id, user_id=uid)
            await session.commit()
            # Read bytes from MinIO (non-fatal if MinIO is unavailable — fall through)
            try:
                file_bytes = await minio_client.get_object(existing_entry.object_key)
                return file_bytes, existing_entry.id
            except Exception:
                # Cache hit but MinIO read failed — fall through to provider fetch below.
                # The entry will be released in the finally block; we re-fetch from provider.
                pass
        finally:
            # Always release pin — clamp protects against double-release.
            try:
                await release_cache_entry(session, entry_id=existing_entry.id, user_id=uid)
                await session.commit()
            except Exception:
                pass  # Non-fatal; log would go here in production

    # ── Cache-miss (or MinIO read failure) path ─────────────────────────────
    # Download bytes from the provider.
    file_bytes = await fetch_fn()

    # Attempt to store in MinIO and create/reactivate a cache entry.
    # All steps are non-fatal — any failure returns the provider bytes without caching.
    cache_entry_id: Optional[uuid.UUID] = None
    pinned_entry_id: Optional[uuid.UUID] = None

    try:
        # Build a UUID-based private object key (T-14-02: never exposed in API responses).
        object_key = f"cache/{uid}/{uuid.uuid4()}"

        # Write bytes to MinIO
        await minio_client.put_object(
            object_key,
            file_bytes,
            content_type or "application/octet-stream",
        )

        size_bytes = len(file_bytes)

        # Attempt atomic quota increment before creating the cache entry.
        try:
            await increment_quota_for_cache(session, user_id=uid, size_bytes=size_bytes)
        except CacheQuotaExceeded:
            # Quota exhausted — skip cache entry creation; return provider bytes.
            await minio_client.delete_object(object_key)
            return file_bytes, None

        # Create or reactivate the cache entry.
        entry, _ = await retain_or_reuse_cache_entry(
            session,
            user_id=uid,
            connection_id=cid,
            cloud_item_id=iid,
            provider_item_id=provider_item_id,
            version_key=version_key,
            object_key=object_key,
            content_type=content_type,
            size_bytes=size_bytes,
        )
        await session.commit()

        # Pin for the duration of the response.
        await pin_cache_entry(session, entry_id=entry.id, user_id=uid)
        await session.commit()
        pinned_entry_id = entry.id
        cache_entry_id = entry.id

    except Exception:
        # Non-fatal: bytes already fetched from provider; return them without caching.
        return file_bytes, None

    # Release pin after bytes have been served.
    try:
        await release_cache_entry(session, entry_id=pinned_entry_id, user_id=uid)
        await session.commit()
    except Exception:
        pass  # Non-fatal

    # Run eviction if needed — non-fatal, fire-and-forget in the same session.
    try:
        await evict_lru_entries(session, user_id=uid, max_cache_bytes=cache_limit_bytes)
        await session.commit()
    except Exception:
        pass

    return file_bytes, cache_entry_id


# ── 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]