From 1df61040c6fd454025f861d0108468a4f6650100 Mon Sep 17 00:00:00 2001 From: curo1305 Date: Tue, 23 Jun 2026 15:14:19 +0200 Subject: [PATCH] feat(14-02): add version-key, settings helpers and Phase 14 cache service functions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - cloud_analysis_versioning.py: compute_version_key with version>etag>fingerprint precedence (D-19); compute_fingerprint helper; content_hash is optional post- hydration only (D-20); raises ValueError only, no HTTPException - cloud_analysis_settings.py: get_or_create_user_analysis_settings (upsert on first access), update_user_analysis_settings with enum validation and tier-capped cache limit, build_default_settings for API responses; raises ValueError only - cloud_cache.py: Phase 14 durable cache functions added alongside existing Phase 12 in-memory TTL cache — create_cache_entry, list_cache_entries (owner-scoped), evict_lru_entries (LRU, skips pin_count>0 and active_job_count>0), increment/ decrement_quota_for_cache (atomic UPDATE pattern); re-exports compute_version_key from cloud_analysis_versioning so tests have one import site --- backend/services/cloud_analysis_settings.py | 163 +++++++++++ backend/services/cloud_analysis_versioning.py | 132 +++++++++ backend/services/cloud_cache.py | 253 +++++++++++++++++- 3 files changed, 546 insertions(+), 2 deletions(-) create mode 100644 backend/services/cloud_analysis_settings.py create mode 100644 backend/services/cloud_analysis_versioning.py diff --git a/backend/services/cloud_analysis_settings.py b/backend/services/cloud_analysis_settings.py new file mode 100644 index 0000000..5f12d87 --- /dev/null +++ b/backend/services/cloud_analysis_settings.py @@ -0,0 +1,163 @@ +""" +Cloud analysis user settings helpers — Phase 14. + +Provides get_or_create_user_analysis_settings and update_user_analysis_settings: +the single authority for reading and mutating per-user analysis preferences. + +Rules: + - Service raises ValueError only — never HTTPException (CLAUDE.md). + - Default values are defined here; the router layer never hard-codes them. + - Tier/admin cache limit maximums are enforced as a seam: pass the admin + maximum via max_cache_limit_bytes to update_user_analysis_settings. + - Invalid enum values are rejected before touching the database. +""" +from __future__ import annotations + +import uuid +from typing import Optional + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from db.models import UserAnalysisSettings + + +# ── Valid enum values ───────────────────────────────────────────────────────── + +VALID_PROGRESS_DETAIL = {"simple", "detailed"} +VALID_FAILURE_BEHAVIOR = {"pause_batch", "continue_item"} + +# Absolute floor for cache limit — prevents pathologically small values +MIN_CACHE_LIMIT_BYTES = 1 * 1024 * 1024 # 1 MB + +# Default tier maximum — overridable by caller (admin/tier seam) +DEFAULT_MAX_CACHE_LIMIT_BYTES = 5 * 1024 * 1024 * 1024 # 5 GB + +# Default per-user cache limit when row is created +DEFAULT_CACHE_LIMIT_BYTES = 512 * 1024 * 1024 # 512 MB + +# Default enum values (mirrors migration server_defaults) +DEFAULT_PROGRESS_DETAIL = "simple" +DEFAULT_FAILURE_BEHAVIOR = "pause_batch" + + +# ── Public API ───────────────────────────────────────────────────────────────── + +async def get_or_create_user_analysis_settings( + session: AsyncSession, + *, + user_id: uuid.UUID, +) -> UserAnalysisSettings: + """Return the user's analysis settings row, creating defaults if absent. + + Never raises — creates the row on first access so callers always receive + a valid settings object. + + Args: + session: Active async SQLAlchemy session. + user_id: Authenticated user UUID. + + Returns: + UserAnalysisSettings with current preferences or newly created defaults. + """ + uid = user_id if isinstance(user_id, uuid.UUID) else uuid.UUID(str(user_id)) + + result = await session.execute( + select(UserAnalysisSettings).where(UserAnalysisSettings.user_id == uid) + ) + existing = result.scalars().first() + if existing is not None: + return existing + + # Create defaults on first access + settings_row = UserAnalysisSettings( + user_id=uid, + analysis_progress_detail=DEFAULT_PROGRESS_DETAIL, + analysis_failure_behavior=DEFAULT_FAILURE_BEHAVIOR, + cloud_cache_limit_bytes=DEFAULT_CACHE_LIMIT_BYTES, + ) + session.add(settings_row) + await session.flush() + return settings_row + + +async def update_user_analysis_settings( + session: AsyncSession, + *, + user_id: uuid.UUID, + analysis_progress_detail: Optional[str] = None, + analysis_failure_behavior: Optional[str] = None, + cloud_cache_limit_bytes: Optional[int] = None, + max_cache_limit_bytes: int = DEFAULT_MAX_CACHE_LIMIT_BYTES, +) -> UserAnalysisSettings: + """Update analysis preferences for the user. + + Only provided (non-None) fields are updated. Enum values and cache limit + bounds are validated before the database is touched. + + Args: + session: Active async SQLAlchemy session. + user_id: Authenticated user UUID. + analysis_progress_detail: "simple" | "detailed" (or None to leave unchanged). + analysis_failure_behavior: "pause_batch" | "continue_item" (or None). + cloud_cache_limit_bytes: User-preferred cache byte limit (or None). + max_cache_limit_bytes: Tier/admin maximum (default 5 GB). Caller + supplies this so the seam is explicit. + + Returns: + Updated UserAnalysisSettings row. + + Raises: + ValueError: Invalid enum value or cache limit out of bounds. + """ + if analysis_progress_detail is not None: + if analysis_progress_detail not in VALID_PROGRESS_DETAIL: + raise ValueError( + f"Invalid analysis_progress_detail {analysis_progress_detail!r}. " + f"Valid values: {VALID_PROGRESS_DETAIL}" + ) + + if analysis_failure_behavior is not None: + if analysis_failure_behavior not in VALID_FAILURE_BEHAVIOR: + raise ValueError( + f"Invalid analysis_failure_behavior {analysis_failure_behavior!r}. " + f"Valid values: {VALID_FAILURE_BEHAVIOR}" + ) + + if cloud_cache_limit_bytes is not None: + if cloud_cache_limit_bytes < MIN_CACHE_LIMIT_BYTES: + raise ValueError( + f"cloud_cache_limit_bytes must be at least {MIN_CACHE_LIMIT_BYTES} bytes, " + f"got {cloud_cache_limit_bytes}" + ) + if cloud_cache_limit_bytes > max_cache_limit_bytes: + raise ValueError( + f"cloud_cache_limit_bytes {cloud_cache_limit_bytes} exceeds tier maximum " + f"{max_cache_limit_bytes}" + ) + + # Fetch or create the row (never fails) + row = await get_or_create_user_analysis_settings(session, user_id=user_id) + + if analysis_progress_detail is not None: + row.analysis_progress_detail = analysis_progress_detail + if analysis_failure_behavior is not None: + row.analysis_failure_behavior = analysis_failure_behavior + if cloud_cache_limit_bytes is not None: + row.cloud_cache_limit_bytes = cloud_cache_limit_bytes + + await session.flush() + return row + + +def build_default_settings() -> dict: + """Return a dict of default settings values for use in API responses. + + Used when no row exists yet and the caller needs a response without + committing a new row (e.g. GET /cache before any settings are saved). + """ + return { + "analysis_progress_detail": DEFAULT_PROGRESS_DETAIL, + "analysis_failure_behavior": DEFAULT_FAILURE_BEHAVIOR, + "cloud_cache_limit_bytes": DEFAULT_CACHE_LIMIT_BYTES, + } diff --git a/backend/services/cloud_analysis_versioning.py b/backend/services/cloud_analysis_versioning.py new file mode 100644 index 0000000..66267a4 --- /dev/null +++ b/backend/services/cloud_analysis_versioning.py @@ -0,0 +1,132 @@ +""" +Cloud analysis version-key helper — Phase 14. + +Provides compute_version_key: the single canonical helper for deriving the +idempotency key that determines whether a cloud item has changed since its +last analysis. + +Version key precedence (PATTERNS.md, D-19): + 1. provider `version` field (most reliable — survives filename/path changes) + 2. provider `etag` (HTTP-standard opaque identifier from HEAD/metadata) + 3. metadata fingerprint — deterministic hash of (provider_item_id, size, + modified_at, content_type) — used when provider supplies neither version + nor etag + 4. content_hash — accepted as an ADDITIONAL post-hydration fact, never + required before bytes are downloaded (D-20) + +Rules: + - Service raises ValueError only — never HTTPException (CLAUDE.md). + - Content hash is NEVER used as a fallback that would require byte download. + - The same inputs must always produce the same key (deterministic). + - Different inputs must reliably produce different keys. +""" +from __future__ import annotations + +import hashlib +from typing import Optional + + +# ── Public constant ─────────────────────────────────────────────────────────── + +#: Prefix tags that identify which precedence level produced the key. +#: These are load-bearing — callers can strip the prefix to get the raw value, +#: or compare keys with mismatched prefixes without hashing collisions. +_PREFIX_VERSION = "v:" +_PREFIX_ETAG = "e:" +_PREFIX_FINGERPRINT = "fp:" +_PREFIX_CONTENT_HASH = "ch:" + + +def compute_version_key( + *, + provider_item_id: str, + version: Optional[str], + etag: Optional[str], + size: Optional[int], + modified_at: Optional[str], + content_type: Optional[str], + content_hash: Optional[str] = None, +) -> str: + """Return the canonical version key for a cloud item. + + The key uniquely identifies the content state of a provider item. + Passing content_hash is optional and only enhances the fingerprint + after bytes have already been hydrated — it is never a precondition + for calling this function. + + Args: + provider_item_id: Opaque provider-assigned item identifier. + version: Provider version string (e.g. Google Drive `version`). + Takes precedence over etag. + etag: Provider etag (e.g. OneDrive/WebDAV ETag header). + Used when version is None. + size: Provider-reported byte size. Used in the fingerprint + fallback. + modified_at: Provider-reported last-modified timestamp string. + Used in the fingerprint fallback. + content_type: MIME content type. Used in the fingerprint fallback. + content_hash: Optional SHA-256 or similar hash computed after bytes + are already hydrated. When provided alongside a + version or etag, it is appended to the key to allow + post-analysis re-keying without triggering a new + download. When provided without version or etag, it + is used as a supplementary fingerprint component. + + Returns: + A non-empty string that uniquely represents the item's current content + state at the given precedence level. + + Raises: + ValueError: provider_item_id is empty or None. + """ + if not provider_item_id: + raise ValueError("provider_item_id must not be empty") + + if version: + base = f"{_PREFIX_VERSION}{version}" + elif etag: + base = f"{_PREFIX_ETAG}{etag}" + else: + base = _compute_fingerprint(provider_item_id, size, modified_at, content_type) + + # Post-hydration content hash enhances the key without requiring a new download. + if content_hash: + base = f"{base}|{_PREFIX_CONTENT_HASH}{content_hash}" + + return base + + +def compute_fingerprint( + provider_item_id: str, + size: Optional[int], + modified_at: Optional[str], + content_type: Optional[str], +) -> str: + """Return the stable metadata fingerprint for an item lacking version/etag. + + Public entry point for callers that need only the fingerprint portion + (e.g. when pre-computing the cloud_analysis_job_items.fingerprint field). + + The fingerprint is a hex-encoded SHA-256 of the canonical field string. + """ + return _compute_fingerprint(provider_item_id, size, modified_at, content_type) + + +# ── Internal helpers ────────────────────────────────────────────────────────── + +def _compute_fingerprint( + provider_item_id: str, + size: Optional[int], + modified_at: Optional[str], + content_type: Optional[str], +) -> str: + """Return a prefixed SHA-256 fingerprint of item metadata.""" + # Canonical form: fields joined with ':' using empty string for None + canonical = ":".join([ + provider_item_id, + str(size) if size is not None else "", + modified_at or "", + content_type or "", + ]) + digest = hashlib.sha256(canonical.encode()).hexdigest() + return f"{_PREFIX_FINGERPRINT}{digest}" diff --git a/backend/services/cloud_cache.py b/backend/services/cloud_cache.py index 19a1f7a..ca2b197 100644 --- a/backend/services/cloud_cache.py +++ b/backend/services/cloud_cache.py @@ -1,6 +1,7 @@ """ -Cloud folder listing cache for DocuVault. +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). @@ -17,13 +18,39 @@ Cache key scheme: "{user_id}:{provider}:{folder_id}" 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 -from typing import Any, Callable, Awaitable +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), @@ -77,6 +104,228 @@ async def get_cloud_folders_cached( 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.