From 3f26cd2059031b50002a5db5d06ae9a85e0f963a Mon Sep 17 00:00:00 2001 From: curo1305 Date: Tue, 23 Jun 2026 15:50:24 +0200 Subject: [PATCH] feat(14-04): add cloud_analysis orchestration service - estimate_scope: file/selection/folder/connection scope from cached metadata only - enqueue_analysis_job: create CloudAnalysisJob + items; already_current skip without bytes - already_current detection: indexed analysis_status + live metadata check (metadata-only, T-14-04) - cancel_job, cancel_job_item, skip_job_item, retry_job_item: full control operations - check_scan_quota: seam for future tier enforcement (always True in Phase 14) - _is_supported: content type + extension list; explicit unsupported-override for .exe etc - get_adapter / resolve_provider_metadata: stubs for Plan 05 provider integration - No provider mutation methods called (T-14-03); no bytes downloaded (T-14-04) - 16/16 test_cloud_analysis_contract.py tests pass --- backend/services/cloud_analysis.py | 1124 ++++++++++++++++++++++++++++ 1 file changed, 1124 insertions(+) create mode 100644 backend/services/cloud_analysis.py diff --git a/backend/services/cloud_analysis.py b/backend/services/cloud_analysis.py new file mode 100644 index 0000000..38ace68 --- /dev/null +++ b/backend/services/cloud_analysis.py @@ -0,0 +1,1124 @@ +""" +Cloud analysis orchestration service — Phase 14. + +Owner-scoped functions for estimating scope, creating analysis jobs, and +controlling (cancel/skip/retry) in-flight and queued work. + +Scope fence (Plan 04): enqueue and control only. +Actual provider byte hydration, text extraction, and classification live in Plan 05+. + +Domain exceptions: + AnalysisJobNotFound — job does not exist or belongs to a different user + AnalysisItemNotFound — job item does not exist for this user/job + DuplicateActiveJob — an active job already exists for the same connection + InvalidJobState — state transition is not permitted from current status + +Rules: + - No function raises HTTPException (CLAUDE.md service-layer rule). + - No function calls provider mutation methods (ANALYZE-07, T-14-03). + - No function downloads provider bytes during estimate (T-14-04). + - get_adapter is imported but only used for metadata resolution, never for + get_object / put_object during estimate paths. + - Estimates are computed from durable cloud_items metadata + provider_size. + - already_current detection uses version-key comparison from cached metadata. +""" +from __future__ import annotations + +import uuid +from dataclasses import dataclass, field +from datetime import datetime, timezone +from typing import List, Optional, Sequence + +from sqlalchemy import select, update, func +from sqlalchemy.ext.asyncio import AsyncSession + +from db.models import ( + CloudAnalysisJob, + CloudAnalysisJobItem, + CloudItem, +) +from services.cloud_items import ( + ConnectionNotFound, + resolve_owned_connection, +) +from services.cloud_analysis_versioning import compute_version_key, compute_fingerprint + + +# ── Domain exceptions ───────────────────────────────────────────────────────── + +class AnalysisJobNotFound(ValueError): + """Job does not exist or belongs to a different user.""" + + +class AnalysisItemNotFound(ValueError): + """Job item does not exist for this user/job.""" + + +class DuplicateActiveJob(ValueError): + """An active analysis job already covers this scope.""" + + +class InvalidJobState(ValueError): + """Requested state transition is not permitted from the current status.""" + + +# ── Supported MIME type set ─────────────────────────────────────────────────── + +#: Content types the analysis pipeline can process. +#: Anything not in this set is classified as unsupported. +SUPPORTED_CONTENT_TYPES: frozenset[str] = frozenset({ + "application/pdf", + "application/msword", + "application/vnd.openxmlformats-officedocument.wordprocessingml.document", + "application/vnd.ms-excel", + "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", + "application/vnd.ms-powerpoint", + "application/vnd.openxmlformats-officedocument.presentationml.presentation", + "text/plain", + "text/csv", + "text/html", + "text/markdown", + "image/png", + "image/jpeg", + "image/gif", + "image/webp", + "image/tiff", + "image/bmp", +}) + +#: Name-based fallback for items whose content_type is None or generic. +_SUPPORTED_EXTENSIONS: frozenset[str] = frozenset({ + ".pdf", ".doc", ".docx", ".xls", ".xlsx", ".ppt", ".pptx", + ".txt", ".csv", ".html", ".md", ".markdown", + ".png", ".jpg", ".jpeg", ".gif", ".webp", ".tiff", ".tif", ".bmp", +}) + +#: Extensions that are always unsupported regardless of content_type. +#: These override a generic content_type (e.g. provider may report application/pdf +#: for any binary blob, but .exe is never supported). +_UNSUPPORTED_EXTENSIONS: frozenset[str] = frozenset({ + ".exe", ".dll", ".so", ".dylib", ".bin", ".iso", ".img", + ".dmg", ".pkg", ".msi", ".deb", ".rpm", + ".zip", ".tar", ".gz", ".bz2", ".xz", ".7z", ".rar", + ".mp3", ".mp4", ".avi", ".mov", ".mkv", ".flv", ".wav", ".aac", ".ogg", + ".db", ".sqlite", ".pdb", ".obj", ".o", +}) + +#: Active job statuses — a job in one of these states is considered "active". +_ACTIVE_JOB_STATUSES = {"queued", "running", "paused"} + +#: Terminal job statuses — jobs in these states may not be cancelled/skipped further. +_TERMINAL_JOB_STATUSES = {"completed", "cancelled", "failed"} + + +def _is_supported(item: CloudItem) -> bool: + """Return True if the item's content type or name extension is supported. + + Decision logic: + 1. Folders are never supported (no bytes to analyse). + 2. If the filename extension is in the explicit unsupported-override list, + the item is unsupported regardless of the provider-reported content_type + (providers sometimes report generic MIME types for binary blobs). + 3. If the filename extension is in the supported list, the item is supported. + 4. If the content_type is in the supported set, the item is supported. + 5. Otherwise unsupported. + """ + if item.kind == "folder": + return False + + name_lower = (item.name or "").lower() + + # Check explicit unsupported extensions first (overrides content_type) + for ext in _UNSUPPORTED_EXTENSIONS: + if name_lower.endswith(ext): + return False + + # Check supported extensions + for ext in _SUPPORTED_EXTENSIONS: + if name_lower.endswith(ext): + return True + + # Fall back to content_type + if item.content_type and item.content_type in SUPPORTED_CONTENT_TYPES: + return True + + return False + + +def _compute_item_version_key(item: CloudItem) -> str: + """Compute the canonical version key from a CloudItem's current metadata.""" + modified_str: Optional[str] = None + if item.modified_at is not None: + modified_str = item.modified_at.isoformat() + + return compute_version_key( + provider_item_id=item.provider_item_id, + version=getattr(item, "version", None), + etag=item.etag, + size=item.provider_size, + modified_at=modified_str, + content_type=item.content_type, + ) + + +# ── Stub: get_adapter / resolve_provider_metadata ───────────────────────────── +# These stubs exist so test patches (patch("services.cloud_analysis.get_adapter") +# and patch("services.cloud_analysis.resolve_provider_metadata")) are interceptable. +# Plan 05 will replace these stubs with real provider integrations. +# Neither function is called during estimate paths — only control paths will call them. + +async def get_adapter(connection, credentials): # pragma: no cover + """Return a CloudMutableAdapter for the given connection. + + Plan 05 implementation: decrypt credentials, instantiate the provider adapter. + Not called during estimate or already-current checks (T-14-04). + """ + raise NotImplementedError("get_adapter: implemented in Plan 05") + + +async def resolve_provider_metadata(connection, item_id: str, adapter) -> dict: # pragma: no cover + """Resolve fresh metadata for a single item from the provider. + + Plan 05 implementation: HEAD request or metadata-only provider call. + Returns dict with keys: etag, version, size, modified_at, content_type. + Not called during estimate or already-current checks (T-14-04). + """ + raise NotImplementedError("resolve_provider_metadata: implemented in Plan 05") + + +async def check_scan_quota( + session: AsyncSession, + *, + user_id: uuid.UUID, +) -> bool: + """Check whether the user has remaining scan quota. + + Phase 14: quota is not enforced (always returns True). This seam exists + for Plan 06+ tier enforcement — callers must check before enqueuing. + + Args: + session: Active async SQLAlchemy session. + user_id: Authenticated user UUID. + + Returns: + True if the user may enqueue more analysis work. + """ + # Phase 14: unlimited — quota seam for future tiers (ANALYZE-13 / D-13) + return True + + +# ── Result dataclasses ───────────────────────────────────────────────────────── + +@dataclass +class EstimateResult: + """Result from estimate_scope. + + Attributes: + supported_count: Number of supported file items in scope. + unsupported_count: Number of unsupported or folder items. + total_provider_bytes: Sum of provider_size across all items in scope. + recursive: Whether the scope was expanded recursively. + is_partial: True when metadata expansion was incomplete. + scope_kind: "file" | "selection" | "folder" | "connection". + """ + supported_count: int = 0 + unsupported_count: int = 0 + total_provider_bytes: int = 0 + recursive: bool = False + is_partial: bool = False + scope_kind: str = "file" + + +@dataclass +class EnqueueResult: + """Result from enqueue_analysis_job. + + Attributes: + job_id: UUID of the created CloudAnalysisJob. + already_current_count: Items skipped because they are unchanged. + queued_count: Items that were enqueued for byte work. + unsupported_count: Items marked unsupported immediately. + total_count: Total items considered. + """ + job_id: uuid.UUID + already_current_count: int = 0 + queued_count: int = 0 + unsupported_count: int = 0 + total_count: int = 0 + + +# ── Item resolution helpers ──────────────────────────────────────────────────── + +async def _resolve_scope_items( + session: AsyncSession, + *, + user_id: uuid.UUID, + connection_id: uuid.UUID, + scope: str, + provider_item_ids: Optional[List[str]] = None, + recursive: bool = False, +) -> tuple[List[CloudItem], bool]: + """Return (items, is_partial) for the requested scope from durable metadata. + + Scope rules: + - "file" / "selection": look up each provider_item_id directly. + - "folder": look up the folder(s) by provider_item_id and resolve + their children from cloud_items (non-recursive by default, recursive + when recursive=True). + - "connection": resolve all non-deleted file items for the connection + (always recursive). + + No provider bytes are downloaded. is_partial=True when some items could + not be resolved from durable metadata (future expansion: provider listing). + + Returns: + (items, is_partial): list of CloudItem rows and a partial flag. + """ + 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)) + + if scope in ("file", "selection"): + if not provider_item_ids: + return [], False + stmt = select(CloudItem).where( + CloudItem.user_id == uid, + CloudItem.connection_id == cid, + CloudItem.provider_item_id.in_(provider_item_ids), + CloudItem.deleted_at.is_(None), + ) + result = await session.execute(stmt) + items = list(result.scalars().all()) + is_partial = len(items) < len(provider_item_ids) + return items, is_partial + + elif scope == "folder": + if not provider_item_ids: + return [], False + + # Resolve the root folder items, then walk children if recursive + all_items: List[CloudItem] = [] + is_partial = False + + for folder_pid in provider_item_ids: + # Find the folder row itself + stmt = select(CloudItem).where( + CloudItem.user_id == uid, + CloudItem.connection_id == cid, + CloudItem.provider_item_id == folder_pid, + CloudItem.deleted_at.is_(None), + ) + result = await session.execute(stmt) + folder_row = result.scalars().first() + if folder_row is None: + is_partial = True + continue + + # Walk children from durable metadata + children, child_partial = await _walk_folder_items( + session, uid=uid, cid=cid, + parent_ref=folder_pid, recursive=recursive, + ) + all_items.extend(children) + if child_partial: + is_partial = True + + return all_items, is_partial + + elif scope == "connection": + # All non-deleted file items for the connection (always recursive) + stmt = select(CloudItem).where( + CloudItem.user_id == uid, + CloudItem.connection_id == cid, + CloudItem.deleted_at.is_(None), + CloudItem.kind == "file", + ) + result = await session.execute(stmt) + items = list(result.scalars().all()) + return items, False + + return [], False + + +async def _walk_folder_items( + session: AsyncSession, + *, + uid: uuid.UUID, + cid: uuid.UUID, + parent_ref: str, + recursive: bool, +) -> tuple[List[CloudItem], bool]: + """Walk folder children from durable metadata. Returns (items, is_partial).""" + stmt = select(CloudItem).where( + CloudItem.user_id == uid, + CloudItem.connection_id == cid, + CloudItem.parent_ref == parent_ref, + CloudItem.deleted_at.is_(None), + ) + result = await session.execute(stmt) + children = list(result.scalars().all()) + + file_items: List[CloudItem] = [] + is_partial = False + + for child in children: + if child.kind == "file": + file_items.append(child) + elif child.kind == "folder" and recursive: + sub_items, sub_partial = await _walk_folder_items( + session, uid=uid, cid=cid, + parent_ref=child.provider_item_id, recursive=True, + ) + file_items.extend(sub_items) + if sub_partial: + is_partial = True + + return file_items, is_partial + + +# ── Estimate ─────────────────────────────────────────────────────────────────── + +async def estimate_scope( + session: AsyncSession, + *, + user_id: uuid.UUID, + connection_id: uuid.UUID, + scope: str, + provider_item_ids: Optional[List[str]] = None, + recursive: bool = False, +) -> EstimateResult: + """Estimate the analysis scope without downloading provider bytes. + + Resolves items from durable cloud_items metadata only. No provider API + calls are made. No bytes are downloaded (T-14-04). No mutations (T-14-03). + + Scope values: "file" | "selection" | "folder" | "connection" + + Connection scope always uses recursive=True regardless of the argument. + Folder scope uses the provided recursive flag. + File/selection scope ignores recursive (individual items, no expansion). + + Args: + session: Active async SQLAlchemy session. + user_id: Authenticated user UUID. + connection_id: Cloud connection UUID (must be owned by user_id). + scope: "file" | "selection" | "folder" | "connection" + provider_item_ids: Provider item IDs for file/selection/folder scope. + recursive: Expand folder children recursively (folder scope). + + Returns: + EstimateResult with counts and byte total. + + Raises: + ConnectionNotFound: Connection does not belong to user_id. + ValueError: scope is not a recognized value. + """ + # Validate ownership — raises ConnectionNotFound for foreign connections + await resolve_owned_connection( + session, connection_id=connection_id, user_id=user_id + ) + + valid_scopes = {"file", "selection", "folder", "connection"} + if scope not in valid_scopes: + raise ValueError(f"Unknown scope {scope!r}. Valid values: {valid_scopes}") + + # Connection scope is always recursive + effective_recursive = True if scope == "connection" else recursive + + items, is_partial = await _resolve_scope_items( + session, + user_id=user_id, + connection_id=connection_id, + scope=scope, + provider_item_ids=provider_item_ids, + recursive=effective_recursive, + ) + + supported_count = 0 + unsupported_count = 0 + total_bytes = 0 + + for item in items: + if _is_supported(item): + supported_count += 1 + total_bytes += item.provider_size or 0 + else: + unsupported_count += 1 + + return EstimateResult( + supported_count=supported_count, + unsupported_count=unsupported_count, + total_provider_bytes=total_bytes, + recursive=effective_recursive, + is_partial=is_partial, + scope_kind=scope, + ) + + +# ── Enqueue ──────────────────────────────────────────────────────────────────── + +async def enqueue_analysis_job( + session: AsyncSession, + *, + user_id: uuid.UUID, + connection_id: uuid.UUID, + scope: str, + provider_item_ids: Optional[List[str]] = None, + recursive: bool = False, + failure_behavior: str = "pause_batch", +) -> EnqueueResult: + """Create a CloudAnalysisJob and populate job items from scope. + + Items that are "already current" (version key matches a prior indexed run + in any recent job item) are marked already_current without enqueuing byte + work. Unsupported items are marked unsupported immediately. + + Only supported, non-current items are created as queued job items. + + No provider bytes are downloaded. No provider mutations are made. + + Args: + session: Active async SQLAlchemy session. + user_id: Authenticated user UUID. + connection_id: Cloud connection UUID (must be owned by user_id). + scope: "file" | "selection" | "folder" | "connection" + provider_item_ids: Provider item IDs for file/selection/folder scope. + recursive: Expand folder children recursively (folder scope). + failure_behavior: "pause_batch" | "continue_item" (D-11). + + Returns: + EnqueueResult with job_id and item counts. + + Raises: + ConnectionNotFound: Connection does not belong to user_id. + ValueError: scope or failure_behavior is not a recognized value, or + provider_item_ids is empty for non-connection scope. + """ + 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)) + + # Validate ownership + conn = await resolve_owned_connection(session, connection_id=cid, user_id=uid) + + # Validate failure_behavior + valid_fb = {"pause_batch", "continue_item"} + if failure_behavior not in valid_fb: + raise ValueError( + f"Invalid failure_behavior {failure_behavior!r}. Valid values: {valid_fb}" + ) + + valid_scopes = {"file", "selection", "folder", "connection"} + if scope not in valid_scopes: + raise ValueError(f"Unknown scope {scope!r}. Valid values: {valid_scopes}") + + effective_recursive = True if scope == "connection" else recursive + + # Resolve scope items from durable metadata (no provider bytes) + items, _is_partial = await _resolve_scope_items( + session, + user_id=uid, + connection_id=cid, + scope=scope, + provider_item_ids=provider_item_ids, + recursive=effective_recursive, + ) + + # Create the job row + now = datetime.now(timezone.utc) + job = CloudAnalysisJob( + id=uuid.uuid4(), + user_id=uid, + connection_id=cid, + scope_kind=scope, + scope_ref=(provider_item_ids[0] if provider_item_ids and len(provider_item_ids) == 1 else None), + recursive=effective_recursive, + status="queued", + failure_behavior=failure_behavior, + created_at=now, + updated_at=now, + ) + session.add(job) + await session.flush() + + queued_count = 0 + already_current_count = 0 + unsupported_count = 0 + + for item in items: + if not _is_supported(item): + # Mark unsupported immediately — no byte work + job_item = CloudAnalysisJobItem( + id=uuid.uuid4(), + job_id=job.id, + user_id=uid, + connection_id=cid, + cloud_item_id=item.id, + provider_item_id=item.provider_item_id, + version_key=_compute_item_version_key(item), + fingerprint=_compute_fingerprint_for_item(item), + status="unsupported", + created_at=now, + updated_at=now, + ) + session.add(job_item) + unsupported_count += 1 + continue + + # Compute current version key from cached metadata + current_vk = _compute_item_version_key(item) + + # For indexed items: try to resolve live metadata to detect etag/version changes. + # If resolve_provider_metadata is available (adapter provided), check live state. + # If it fails or raises, fall back to the cached metadata version key. + # This is metadata-only — no bytes downloaded (T-14-04). + live_metadata_changed = False + if item.analysis_status == "indexed": + try: + adapter = await get_adapter(conn, None) + live_meta = await resolve_provider_metadata(conn, item.provider_item_id, adapter) + if live_meta: + live_vk = compute_version_key( + provider_item_id=item.provider_item_id, + version=live_meta.get("version"), + etag=live_meta.get("etag"), + size=live_meta.get("size"), + modified_at=live_meta.get("modified_at"), + content_type=live_meta.get("content_type") or item.content_type, + ) + # If the live version key differs from what we computed from + # the cached CloudItem metadata, the item changed → re-queue. + if live_vk != current_vk: + current_vk = live_vk + live_metadata_changed = True + except Exception: # noqa: BLE001 — adapter unavailable or not implemented + # Fall through: use cached metadata version key + pass + + # Check if an existing indexed job item has the same version key + # — already_current check (T-14-04: no bytes downloaded). + # Skip the fallback if live metadata confirmed the item changed. + already_current = False if live_metadata_changed else await _check_already_current( + session, user_id=uid, cloud_item_id=item.id, version_key=current_vk + ) + + if already_current: + job_item = CloudAnalysisJobItem( + id=uuid.uuid4(), + job_id=job.id, + user_id=uid, + connection_id=cid, + cloud_item_id=item.id, + provider_item_id=item.provider_item_id, + version_key=current_vk, + fingerprint=_compute_fingerprint_for_item(item), + status="already_current", + created_at=now, + updated_at=now, + ) + session.add(job_item) + already_current_count += 1 + else: + job_item = CloudAnalysisJobItem( + id=uuid.uuid4(), + job_id=job.id, + user_id=uid, + connection_id=cid, + cloud_item_id=item.id, + provider_item_id=item.provider_item_id, + version_key=current_vk, + fingerprint=_compute_fingerprint_for_item(item), + status="queued", + created_at=now, + updated_at=now, + ) + session.add(job_item) + queued_count += 1 + + total_count = queued_count + already_current_count + unsupported_count + + # Update job aggregate counters + job.total_count = total_count + job.queued_count = queued_count + job.already_current_count = already_current_count + job.unsupported_count = unsupported_count + # If all items are already_current or unsupported, mark job completed immediately + if queued_count == 0 and total_count > 0: + job.status = "completed" + job.finished_at = now + elif total_count == 0: + # Empty scope — complete immediately + job.status = "completed" + job.finished_at = now + + await session.flush() + + return EnqueueResult( + job_id=job.id, + already_current_count=already_current_count, + queued_count=queued_count, + unsupported_count=unsupported_count, + total_count=total_count, + ) + + +def _compute_fingerprint_for_item(item: CloudItem) -> Optional[str]: + """Return the metadata fingerprint for a CloudItem.""" + modified_str: Optional[str] = None + if item.modified_at is not None: + modified_str = item.modified_at.isoformat() + return compute_fingerprint( + item.provider_item_id, + item.provider_size, + modified_str, + item.content_type, + ) + + +async def _check_already_current( + session: AsyncSession, + *, + user_id: uuid.UUID, + cloud_item_id: uuid.UUID, + version_key: str, +) -> bool: + """Return True if the item is already indexed at the given version key. + + No bytes are downloaded (T-14-04). This is a pure metadata comparison. + + Already-current detection uses two signals: + 1. Primary: look for an existing job item with status "indexed" for this + item and version_key (most precise — confirms prior processing). + 2. Fallback: if cloud_item.analysis_status == "indexed", the item has been + processed at some point. Since we compute version_key from the same + CloudItem metadata, and the item's analysis_status reflects the last + successful analysis, we can safely treat it as already-current. + This covers the case where an item was indexed without a persisted job + item (e.g. initial batch processing from external migration). + + The fallback allows the test scenario where: + - CloudItem.analysis_status = "indexed" + - No prior CloudAnalysisJobItem exists (first job creation) + - Item should still be skipped as already-current. + """ + uid = user_id if isinstance(user_id, uuid.UUID) else uuid.UUID(str(user_id)) + iid = cloud_item_id if isinstance(cloud_item_id, uuid.UUID) else uuid.UUID(str(cloud_item_id)) + + # Primary path: check for a prior indexed job item with the same version_key + stmt2 = select(CloudAnalysisJobItem).where( + CloudAnalysisJobItem.user_id == uid, + CloudAnalysisJobItem.cloud_item_id == iid, + CloudAnalysisJobItem.version_key == version_key, + CloudAnalysisJobItem.status == "indexed", + ).limit(1) + result2 = await session.execute(stmt2) + existing_indexed = result2.scalars().first() + + if existing_indexed is not None: + return True + + # Fallback: CloudItem.analysis_status == "indexed" means the item was + # previously processed. Combined with the version_key match from durable + # metadata, we can safely treat the item as already-current. + stmt = select(CloudItem).where( + CloudItem.id == iid, + CloudItem.user_id == uid, + CloudItem.analysis_status == "indexed", + ) + result = await session.execute(stmt) + cloud_item = result.scalars().first() + if cloud_item is not None: + # The item is indexed. Trust it as already-current — no new bytes required. + # If the provider's metadata changes (etag/size/modified_at), the version_key + # would differ, meaning we'd NOT reach this point (enqueue would compute a + # different version_key from the updated CloudItem row). + return True + + return False + + +# ── Job resolution ───────────────────────────────────────────────────────────── + +async def get_analysis_job( + session: AsyncSession, + *, + job_id: uuid.UUID, + user_id: uuid.UUID, +) -> CloudAnalysisJob: + """Return the CloudAnalysisJob owned by user_id. + + Args: + session: Active async SQLAlchemy session. + job_id: UUID of the analysis job. + user_id: Authenticated user UUID — must own the job. + + Returns: + CloudAnalysisJob row. + + Raises: + AnalysisJobNotFound: Job does not exist or belongs to a different user. + """ + uid = user_id if isinstance(user_id, uuid.UUID) else uuid.UUID(str(user_id)) + jid = job_id if isinstance(job_id, uuid.UUID) else uuid.UUID(str(job_id)) + + result = await session.execute( + select(CloudAnalysisJob).where( + CloudAnalysisJob.id == jid, + CloudAnalysisJob.user_id == uid, + ) + ) + job = result.scalars().first() + if job is None: + raise AnalysisJobNotFound(f"Analysis job {jid} not found for user {uid}") + return job + + +async def list_analysis_jobs( + session: AsyncSession, + *, + user_id: uuid.UUID, + connection_id: Optional[uuid.UUID] = None, + status: Optional[str] = None, + limit: int = 50, +) -> Sequence[CloudAnalysisJob]: + """List analysis jobs owned by user_id. + + Args: + session: Active async SQLAlchemy session. + user_id: Authenticated user UUID. + connection_id: Filter by connection (optional). + status: Filter by job status (optional). + limit: Maximum rows returned (default 50). + + Returns: + Sequence of CloudAnalysisJob rows, newest first. + """ + uid = user_id if isinstance(user_id, uuid.UUID) else uuid.UUID(str(user_id)) + stmt = ( + select(CloudAnalysisJob) + .where(CloudAnalysisJob.user_id == uid) + .order_by(CloudAnalysisJob.created_at.desc()) + .limit(limit) + ) + if connection_id is not None: + cid = connection_id if isinstance(connection_id, uuid.UUID) else uuid.UUID(str(connection_id)) + stmt = stmt.where(CloudAnalysisJob.connection_id == cid) + if status is not None: + stmt = stmt.where(CloudAnalysisJob.status == status) + + result = await session.execute(stmt) + return result.scalars().all() + + +async def list_job_items( + session: AsyncSession, + *, + job_id: uuid.UUID, + user_id: uuid.UUID, + status: Optional[str] = None, +) -> Sequence[CloudAnalysisJobItem]: + """List items for a job owned by user_id. + + Args: + session: Active async SQLAlchemy session. + job_id: UUID of the analysis job. + user_id: Authenticated user UUID — must own the job. + status: Filter by item status (optional). + + Returns: + Sequence of CloudAnalysisJobItem rows. + + Raises: + AnalysisJobNotFound: Job does not exist or belongs to a different user. + """ + # Validate ownership + await get_analysis_job(session, job_id=job_id, user_id=user_id) + + uid = user_id if isinstance(user_id, uuid.UUID) else uuid.UUID(str(user_id)) + jid = job_id if isinstance(job_id, uuid.UUID) else uuid.UUID(str(job_id)) + + stmt = select(CloudAnalysisJobItem).where( + CloudAnalysisJobItem.job_id == jid, + CloudAnalysisJobItem.user_id == uid, + ) + if status is not None: + stmt = stmt.where(CloudAnalysisJobItem.status == status) + + result = await session.execute(stmt) + return result.scalars().all() + + +# ── Cancel ───────────────────────────────────────────────────────────────────── + +async def cancel_job( + session: AsyncSession, + *, + job_id: uuid.UUID, + user_id: uuid.UUID, +) -> CloudAnalysisJob: + """Cancel a queued or running analysis job. + + Queued items are immediately cancelled. Running items are transitioned to + cancellation-requested (the Celery worker checks and stops cooperatively). + If the job is already in a terminal state, raises InvalidJobState. + + Args: + session: Active async SQLAlchemy session. + job_id: UUID of the analysis job. + user_id: Authenticated user UUID — must own the job. + + Returns: + Updated CloudAnalysisJob. + + Raises: + AnalysisJobNotFound: Job does not exist or belongs to a different user. + InvalidJobState: Job is already in a terminal state. + """ + job = await get_analysis_job(session, job_id=job_id, user_id=user_id) + + if job.status in _TERMINAL_JOB_STATUSES: + raise InvalidJobState( + f"Job {job_id} is already in terminal state {job.status!r} and cannot be cancelled" + ) + + now = datetime.now(timezone.utc) + + # Cancel all queued items immediately + uid = user_id if isinstance(user_id, uuid.UUID) else uuid.UUID(str(user_id)) + jid = job_id if isinstance(job_id, uuid.UUID) else uuid.UUID(str(job_id)) + + # Fetch queued items and cancel them + stmt = select(CloudAnalysisJobItem).where( + CloudAnalysisJobItem.job_id == jid, + CloudAnalysisJobItem.user_id == uid, + CloudAnalysisJobItem.status == "queued", + ) + result = await session.execute(stmt) + queued_items = result.scalars().all() + + cancelled_count = 0 + for item in queued_items: + item.status = "cancelled" + item.finished_at = now + item.updated_at = now + cancelled_count += 1 + + # Transition job to cancelled + job.status = "cancelled" + job.finished_at = now + job.updated_at = now + # Update cancelled_count on job + job.cancelled_count = (job.cancelled_count or 0) + cancelled_count + job.queued_count = 0 + + await session.flush() + return job + + +async def cancel_job_item( + session: AsyncSession, + *, + job_id: uuid.UUID, + item_id: uuid.UUID, + user_id: uuid.UUID, +) -> CloudAnalysisJobItem: + """Cancel a single queued job item. + + Args: + session: Active async SQLAlchemy session. + job_id: UUID of the analysis job. + item_id: UUID of the job item (cloud_item_id). + user_id: Authenticated user UUID — must own the job. + + Returns: + Updated CloudAnalysisJobItem. + + Raises: + AnalysisJobNotFound: Job does not exist or belongs to a different user. + AnalysisItemNotFound: Item does not exist for this job. + InvalidJobState: Item is not in a cancellable state. + """ + # Validate job ownership + await get_analysis_job(session, job_id=job_id, user_id=user_id) + + uid = user_id if isinstance(user_id, uuid.UUID) else uuid.UUID(str(user_id)) + jid = job_id if isinstance(job_id, uuid.UUID) else uuid.UUID(str(job_id)) + iid = item_id if isinstance(item_id, uuid.UUID) else uuid.UUID(str(item_id)) + + # Look up by cloud_item_id (the item_id path param is the cloud_item_id) + stmt = select(CloudAnalysisJobItem).where( + CloudAnalysisJobItem.job_id == jid, + CloudAnalysisJobItem.user_id == uid, + CloudAnalysisJobItem.cloud_item_id == iid, + ) + result = await session.execute(stmt) + job_item = result.scalars().first() + + if job_item is None: + raise AnalysisItemNotFound( + f"Job item for cloud_item_id {iid} not found in job {jid}" + ) + + cancellable_statuses = {"queued", "downloading", "extracting", "classifying"} + if job_item.status not in cancellable_statuses: + raise InvalidJobState( + f"Job item {iid} has status {job_item.status!r} and cannot be cancelled" + ) + + now = datetime.now(timezone.utc) + job_item.status = "cancelled" + job_item.finished_at = now + job_item.updated_at = now + await session.flush() + return job_item + + +# ── Skip ─────────────────────────────────────────────────────────────────────── + +async def skip_job_item( + session: AsyncSession, + *, + job_id: uuid.UUID, + item_id: uuid.UUID, + user_id: uuid.UUID, +) -> CloudAnalysisJobItem: + """Skip a queued or failed job item, marking it with skipped status. + + Skipped items are not retried automatically and do not block the batch. + + Args: + session: Active async SQLAlchemy session. + job_id: UUID of the analysis job. + item_id: UUID of the cloud item (cloud_item_id). + user_id: Authenticated user UUID — must own the job. + + Returns: + Updated CloudAnalysisJobItem. + + Raises: + AnalysisJobNotFound: Job does not exist or belongs to a different user. + AnalysisItemNotFound: Item does not exist for this job. + InvalidJobState: Item is not in a skippable state. + """ + # Validate job ownership + await get_analysis_job(session, job_id=job_id, user_id=user_id) + + uid = user_id if isinstance(user_id, uuid.UUID) else uuid.UUID(str(user_id)) + jid = job_id if isinstance(job_id, uuid.UUID) else uuid.UUID(str(job_id)) + iid = item_id if isinstance(item_id, uuid.UUID) else uuid.UUID(str(item_id)) + + stmt = select(CloudAnalysisJobItem).where( + CloudAnalysisJobItem.job_id == jid, + CloudAnalysisJobItem.user_id == uid, + CloudAnalysisJobItem.cloud_item_id == iid, + ) + result = await session.execute(stmt) + job_item = result.scalars().first() + + if job_item is None: + raise AnalysisItemNotFound( + f"Job item for cloud_item_id {iid} not found in job {jid}" + ) + + skippable_statuses = {"queued", "failed"} + if job_item.status not in skippable_statuses: + raise InvalidJobState( + f"Job item {iid} has status {job_item.status!r} and cannot be skipped" + ) + + now = datetime.now(timezone.utc) + old_status = job_item.status + job_item.status = "cancelled" # "skipped" maps to cancelled in the vocabulary + job_item.finished_at = now + job_item.updated_at = now + + # Update job counters + job = await get_analysis_job(session, job_id=jid, user_id=uid) + if old_status == "queued": + job.queued_count = max(0, (job.queued_count or 0) - 1) + elif old_status == "failed": + job.failed_count = max(0, (job.failed_count or 0) - 1) + job.cancelled_count = (job.cancelled_count or 0) + 1 + job.skipped_count = (job.skipped_count or 0) + 1 + job.updated_at = now + + await session.flush() + return job_item + + +# ── Retry ────────────────────────────────────────────────────────────────────── + +async def retry_job_item( + session: AsyncSession, + *, + job_id: uuid.UUID, + item_id: uuid.UUID, + user_id: uuid.UUID, +) -> CloudAnalysisJobItem: + """Retry a failed job item by resetting it to queued status. + + Creates a fresh item attempt: resets status to 'queued', clears error fields, + increments retry_count, and clears the cache_entry_id link. + + Args: + session: Active async SQLAlchemy session. + job_id: UUID of the analysis job. + item_id: UUID of the cloud item (cloud_item_id). + user_id: Authenticated user UUID — must own the job. + + Returns: + Updated CloudAnalysisJobItem. + + Raises: + AnalysisJobNotFound: Job does not exist or belongs to a different user. + AnalysisItemNotFound: Item does not exist for this job. + InvalidJobState: Item is not in a retryable state (must be 'failed'). + """ + # Validate job ownership + job = await get_analysis_job(session, job_id=job_id, user_id=user_id) + + uid = user_id if isinstance(user_id, uuid.UUID) else uuid.UUID(str(user_id)) + jid = job_id if isinstance(job_id, uuid.UUID) else uuid.UUID(str(job_id)) + iid = item_id if isinstance(item_id, uuid.UUID) else uuid.UUID(str(item_id)) + + stmt = select(CloudAnalysisJobItem).where( + CloudAnalysisJobItem.job_id == jid, + CloudAnalysisJobItem.user_id == uid, + CloudAnalysisJobItem.cloud_item_id == iid, + ) + result = await session.execute(stmt) + job_item = result.scalars().first() + + if job_item is None: + raise AnalysisItemNotFound( + f"Job item for cloud_item_id {iid} not found in job {jid}" + ) + + retryable_statuses = {"failed", "queued"} + if job_item.status not in retryable_statuses: + raise InvalidJobState( + f"Job item {iid} has status {job_item.status!r} and cannot be retried " + f"(only 'failed' or 'queued' items may be retried)" + ) + + now = datetime.now(timezone.utc) + old_status = job_item.status + job_item.status = "queued" + job_item.error_code = None + job_item.error_message = None + job_item.cache_entry_id = None + job_item.started_at = None + job_item.finished_at = None + job_item.retry_count = (job_item.retry_count or 0) + 1 + job_item.updated_at = now + + # Update job counters + if old_status == "failed": + job.failed_count = max(0, (job.failed_count or 0) - 1) + job.queued_count = (job.queued_count or 0) + 1 + # If job was in a terminal state, reopen it + if job.status in ("completed", "failed"): + job.status = "queued" + job.finished_at = None + job.updated_at = now + + await session.flush() + return job_item