kite/backend/services/cloud_items.py

"""
Owner-scoped cloud metadata reconciliation service — Phase 12.

All functions operate within strict (user_id, connection_id) ownership boundaries.
No function here raises FastAPI HTTPException — domain exceptions only.
No function calls the quota service or alters quotas.used_bytes.

Domain exceptions:
  ConnectionNotFound — connection does not exist or belongs to a different user.
  CloudItemNotFound  — cloud item does not exist for the given owner/connection.
"""
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
from sqlalchemy.ext.asyncio import AsyncSession

from db.models import CloudConnection, CloudFolderState, CloudItem
from storage.cloud_base import CloudListing, CloudResource


# ── Domain exceptions ─────────────────────────────────────────────────────────

class ConnectionNotFound(ValueError):
    """Connection does not exist or belongs to a different user."""


class CloudItemNotFound(ValueError):
    """Cloud item does not exist for the given owner/connection."""


# ── Cloud item detail dataclass ───────────────────────────────────────────────

@dataclass
class CloudItemDetail:
    """Credential-free detail payload for a single cloud item.

    Built by resolve_owned_cloud_item_detail from DB metadata only — no byte
    hydration is performed (T-14.1-06 / CACHE-03).

    Forbidden fields: object_key, credentials_enc, version_key, raw provider URLs.
    """

    # Core item identity
    id: str
    provider_item_id: str
    name: str
    kind: str
    parent_ref: Optional[str]

    # Provider metadata
    content_type: Optional[str]
    size: Optional[int]
    modified_at: Optional[datetime]
    etag: Optional[str]

    # Connection source metadata (D-04) — no raw provider URLs or credentials
    provider: str
    display_name: str
    location: Optional[str]  # path_snapshot or parent_ref; never a provider URL

    # Analysis fields (D-05, D-07, D-08)
    analysis_status: str
    semantic_index_status: str
    extracted_text: Optional[str]
    topics: List[str]

    # Capability / unsupported-analysis convenience fields (D-16)
    is_stale: bool
    unsupported_analysis_reason: Optional[str]


# ── Connection resolution ─────────────────────────────────────────────────────

async def resolve_owned_connection(
    session: AsyncSession,
    *,
    connection_id,
    user_id,
) -> CloudConnection:
    """Return the CloudConnection owned by user_id or raise ConnectionNotFound.

    Accepts UUID objects or string UUIDs for both parameters.
    Never returns a connection belonging to another user.
    """
    conn_uuid = connection_id if isinstance(connection_id, uuid.UUID) else uuid.UUID(str(connection_id))
    user_uuid = user_id if isinstance(user_id, uuid.UUID) else uuid.UUID(str(user_id))
    result = await session.execute(
        select(CloudConnection).where(
            CloudConnection.id == conn_uuid,
            CloudConnection.user_id == user_uuid,
        )
    )
    conn = result.scalars().first()
    if conn is None:
        raise ConnectionNotFound(
            f"Connection {connection_id!r} not found for user {user_id!r}"
        )
    return conn


# ── Cloud item detail resolution ─────────────────────────────────────────────

async def resolve_owned_cloud_item_detail(
    session: AsyncSession,
    *,
    user_id,
    connection_id,
    provider_item_id: str,
) -> CloudItemDetail:
    """Return owner-scoped cloud item detail from DB metadata only.

    Performs three DB reads:
      1. resolve_owned_connection — verifies the connection is owned by user_id.
      2. Select CloudItem by (connection_id, provider_item_id, user_id).
      3. Select CloudItemTopic→Topic join for topic names.

    No provider bytes are downloaded. No get_object or hydrate_and_cache_bytes
    is called (T-14.1-06, CACHE-03). This function is metadata-only.

    Args:
        session:          Active async SQLAlchemy session.
        user_id:          Authenticated user UUID.
        connection_id:    Cloud connection UUID.
        provider_item_id: Provider-side opaque item identifier.

    Returns:
        CloudItemDetail dataclass populated from DB rows.

    Raises:
        ConnectionNotFound: Connection does not exist or belongs to another user.
        CloudItemNotFound:  Cloud item does not exist for the given connection/owner.
    """
    from db.models import CloudItemTopic, Topic  # local import avoids circular

    # 1. Ownership gate (T-14.1-04, T-14.1-06)
    conn = await resolve_owned_connection(
        session, connection_id=connection_id, user_id=user_id
    )

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

    # 2. Resolve the cloud item (must belong to same user)
    result = await session.execute(
        select(CloudItem).where(
            CloudItem.connection_id == cid,
            CloudItem.provider_item_id == provider_item_id,
            CloudItem.user_id == uid,
            CloudItem.deleted_at.is_(None),
        )
    )
    item = result.scalars().first()
    if item is None:
        raise CloudItemNotFound(
            f"Cloud item {provider_item_id!r} not found in connection {connection_id!r}"
        )

    # 3. Resolve topic names via CloudItemTopic→Topic join (metadata-only)
    topics_result = await session.execute(
        select(Topic.name).join(
            CloudItemTopic, CloudItemTopic.topic_id == Topic.id
        ).where(
            CloudItemTopic.cloud_item_id == item.id
        )
    )
    topic_names: List[str] = list(topics_result.scalars().all())

    # 4. Derive convenience / source metadata fields

    # location: prefer path_snapshot (human path), fall back to parent_ref (opaque)
    # — never expose raw provider URLs in this field (T-14.1-03)
    location: Optional[str] = getattr(item, "path_snapshot", None) or item.parent_ref

    # unsupported_analysis_reason: populated when the item cannot be analyzed
    # Uses the same logic as cloud_analysis._is_supported without importing that module
    unsupported_reason: Optional[str] = None
    if item.kind == "folder":
        unsupported_reason = "Folders cannot be analyzed"
    else:
        # Simple extension/MIME check duplicated here to avoid a cross-service import.
        # The authoritative supported-type check lives in services.cloud_analysis._is_supported.
        from services.cloud_analysis import _is_supported as _ca_is_supported
        if not _ca_is_supported(item):
            unsupported_reason = "File type is not supported for analysis"

    is_stale = (item.analysis_status == "stale")

    return CloudItemDetail(
        id=str(item.id),
        provider_item_id=item.provider_item_id,
        name=item.name,
        kind=item.kind,
        parent_ref=item.parent_ref,
        content_type=item.content_type,
        size=item.provider_size,
        modified_at=item.modified_at,
        etag=item.etag,
        provider=conn.provider,
        display_name=conn.display_name_override or conn.display_name,
        location=location,
        analysis_status=item.analysis_status,
        semantic_index_status=item.semantic_index_status,
        extracted_text=item.extracted_text,
        topics=topic_names,
        is_stale=is_stale,
        unsupported_analysis_reason=unsupported_reason,
    )


# ── Item listing ──────────────────────────────────────────────────────────────

async def list_cloud_children(
    session: AsyncSession,
    *,
    user_id: str,
    connection_id: str,
    parent_ref: Optional[str],
) -> Sequence[CloudItem]:
    """Return non-deleted cloud items matching (user_id, connection_id, parent_ref).

    parent_ref=None matches items whose parent_ref is NULL (root children where
    parent is not tracked as a ref string).
    """
    uid_v = user_id if isinstance(user_id, uuid.UUID) else uuid.UUID(str(user_id))
    cid_v = connection_id if isinstance(connection_id, uuid.UUID) else uuid.UUID(str(connection_id))
    stmt = select(CloudItem).where(
        CloudItem.user_id == uid_v,
        CloudItem.connection_id == cid_v,
        CloudItem.deleted_at.is_(None),
    )
    if parent_ref is None:
        stmt = stmt.where(CloudItem.parent_ref.is_(None))
    else:
        stmt = stmt.where(CloudItem.parent_ref == parent_ref)

    result = await session.execute(stmt)
    return result.scalars().all()


# ── Item upsert ───────────────────────────────────────────────────────────────

async def upsert_cloud_item(
    session: AsyncSession,
    *,
    user_id: str,
    resource: CloudResource,
) -> CloudItem:
    """Insert or update a CloudItem for the given normalized resource.

    The DocuVault CloudItem UUID is preserved across provider rename/move:
    if a row already exists for (connection_id, provider_item_id), its id
    is retained and metadata fields are updated in place.

    user_id is always set from the caller — never from the resource alone —
    to enforce the owner boundary.
    """
    conn_uuid2 = resource.connection_id if isinstance(resource.connection_id, uuid.UUID) else uuid.UUID(str(resource.connection_id))
    result = await session.execute(
        select(CloudItem).where(
            CloudItem.connection_id == conn_uuid2,
            CloudItem.provider_item_id == resource.provider_item_id,
        )
    )
    existing = result.scalars().first()

    now = datetime.now(timezone.utc)

    if existing is not None:
        # Update metadata, preserve UUID (stable identity across rename/move)
        existing.name = resource.name
        existing.kind = resource.kind
        existing.parent_ref = resource.parent_ref
        existing.content_type = resource.content_type
        existing.provider_size = resource.size
        existing.etag = resource.etag
        existing.modified_at = resource.modified_at
        existing.last_seen_at = now
        existing.deleted_at = None  # un-delete if previously soft-deleted
        existing.updated_at = now
        await session.flush()
        return existing
    else:
        user_uuid_ins = user_id if isinstance(user_id, uuid.UUID) else uuid.UUID(str(user_id))
        conn_uuid_ins = resource.connection_id if isinstance(resource.connection_id, uuid.UUID) else uuid.UUID(str(resource.connection_id))
        item = CloudItem(
            id=uuid.uuid4(),
            user_id=user_uuid_ins,
            connection_id=conn_uuid_ins,
            provider_item_id=resource.provider_item_id,
            parent_ref=resource.parent_ref,
            name=resource.name,
            kind=resource.kind,
            content_type=resource.content_type,
            provider_size=resource.size,
            etag=resource.etag,
            modified_at=resource.modified_at,
            last_seen_at=now,
            analysis_status="pending",
            semantic_index_status="none",
        )
        session.add(item)
        await session.flush()
        return item


# ── Listing reconciliation ────────────────────────────────────────────────────

async def reconcile_cloud_listing(
    session: AsyncSession,
    *,
    user_id: str,
    connection_id: str,
    parent_ref: Optional[str],
    listing: CloudListing,
) -> None:
    """Reconcile a provider listing against durable cloud_items rows.

    For each item in listing.items: upsert metadata.

    Soft-deletion of missing children is ONLY performed when:
      - listing.complete is True (authoritative, full listing)

    Incomplete or failed listings (complete=False) must NEVER mark retained
    rows deleted — provider bytes remain the source of truth.

    This function never calls the quota service (D-18).
    """
    seen_provider_ids: set[str] = set()

    for resource in listing.items:
        await upsert_cloud_item(session, user_id=user_id, resource=resource)
        seen_provider_ids.add(resource.provider_item_id)

    if listing.complete:
        # Soft-delete items not present in the complete listing
        uid_v2 = user_id if isinstance(user_id, uuid.UUID) else uuid.UUID(str(user_id))
        cid_v2 = connection_id if isinstance(connection_id, uuid.UUID) else uuid.UUID(str(connection_id))
        stmt = select(CloudItem).where(
            CloudItem.user_id == uid_v2,
            CloudItem.connection_id == cid_v2,
            CloudItem.deleted_at.is_(None),
        )
        if parent_ref is None:
            stmt = stmt.where(CloudItem.parent_ref.is_(None))
        else:
            stmt = stmt.where(CloudItem.parent_ref == parent_ref)

        result = await session.execute(stmt)
        existing_items = result.scalars().all()

        now = datetime.now(timezone.utc)
        for item in existing_items:
            if item.provider_item_id not in seen_provider_ids:
                item.deleted_at = now
                item.updated_at = now

        await session.flush()


# ── Folder state helpers ──────────────────────────────────────────────────────

async def get_or_create_folder_state(
    session: AsyncSession,
    *,
    user_id: str,
    connection_id: str,
    parent_ref: str,
) -> CloudFolderState:
    """Return the CloudFolderState for (connection_id, parent_ref), creating if absent.

    Idempotent: repeated calls with the same arguments return the same row.
    parent_ref='' represents the connection root.
    """
    cid_fs = connection_id if isinstance(connection_id, uuid.UUID) else uuid.UUID(str(connection_id))
    uid_fs = user_id if isinstance(user_id, uuid.UUID) else uuid.UUID(str(user_id))
    result = await session.execute(
        select(CloudFolderState).where(
            CloudFolderState.user_id == uid_fs,
            CloudFolderState.connection_id == cid_fs,
            CloudFolderState.parent_ref == parent_ref,
        )
    )
    existing = result.scalars().first()
    if existing is not None:
        return existing

    fs = CloudFolderState(
        id=uuid.uuid4(),
        user_id=uid_fs,
        connection_id=cid_fs,
        parent_ref=parent_ref,
        refresh_state="fresh",
    )
    session.add(fs)
    await session.flush()
    return fs


# ── Listing application result ────────────────────────────────────────────────

@dataclass
class ListingResult:
    """Structured result from apply_listing_and_finalize.

    is_fresh: True only when the provider supplied a complete authoritative listing.
    warning_code: stable controlled code when is_fresh is False (never raw exception text).
    warning_message: stable controlled message for API consumers (no credentials/URLs).
    """
    is_fresh: bool
    warning_code: Optional[str] = None
    warning_message: Optional[str] = None


# ── Single listing application gate ──────────────────────────────────────────

async def apply_listing_and_finalize(
    session: AsyncSession,
    *,
    user_id: str,
    connection_id: str,
    parent_ref: Optional[str],
    listing: CloudListing,
) -> ListingResult:
    """Apply a provider listing and finalize CloudFolderState atomically.

    This is the SINGLE shared gate used by both the synchronous browse endpoint
    and the Celery background worker.  Neither caller may independently decide
    freshness — they MUST use this function.

    Semantics:
    - listing.complete=True (authoritative):
        Upserts all items, soft-deletes unseen children, marks folder FRESH,
        advances last_refreshed_at.  Returns ListingResult(is_fresh=True).

    - listing.complete=False (partial/unknown):
        Upserts only seen items (never deletes unseen children), does NOT
        advance last_refreshed_at, sets folder state to WARNING with stable
        code 'incomplete_listing'.  Returns ListingResult(is_fresh=False, …).

    No exception is raised on incomplete results — callers receive a structured
    result and must propagate the warning state to the API response or task result.
    This function never raises HTTPException — domain exceptions only.
    This function never calls the quota service.
    """
    # Reconcile items from the provider listing
    await reconcile_cloud_listing(
        session,
        user_id=user_id,
        connection_id=connection_id,
        parent_ref=parent_ref,
        listing=listing,
    )

    if listing.complete:
        # Authoritative: mark folder fresh and advance last_refreshed_at
        await update_folder_state(
            session,
            user_id=user_id,
            connection_id=connection_id,
            parent_ref=parent_ref or "",
            refresh_state="fresh",
        )
        return ListingResult(is_fresh=True)
    else:
        # Incomplete/partial: retain cached state; do NOT advance last_refreshed_at
        # We only update the warning fields, preserving last_refreshed_at from prior success.
        fs = await get_or_create_folder_state(
            session,
            user_id=user_id,
            connection_id=connection_id,
            parent_ref=parent_ref or "",
        )
        now = datetime.now(timezone.utc)
        # Set warning state without touching last_refreshed_at
        fs.refresh_state = "warning"
        fs.error_code = "incomplete_listing"
        fs.error_message = (
            "Provider returned an incomplete listing. Showing cached results."
        )
        fs.updated_at = now
        # Deliberately: do NOT update fs.last_refreshed_at
        await session.flush()
        return ListingResult(
            is_fresh=False,
            warning_code="incomplete_listing",
            warning_message="Provider returned an incomplete listing. Showing cached results.",
        )


async def update_folder_state(
    session: AsyncSession,
    *,
    user_id: str,
    connection_id: str,
    parent_ref: str,
    refresh_state: str,
    last_refreshed_at: Optional[datetime] = None,
    error_code: Optional[str] = None,
    error_message: Optional[str] = None,
) -> CloudFolderState:
    """Update the refresh state for a folder.

    Transitions: refreshing → fresh (success) or warning (failure).
    On success: set last_refreshed_at.
    On failure: retain last_refreshed_at, set error_code/message with controlled values.

    error_code and error_message must be controlled service values — never raw
    provider exception text.
    """
    fs = await get_or_create_folder_state(
        session, user_id=user_id, connection_id=connection_id, parent_ref=parent_ref
    )
    now = datetime.now(timezone.utc)
    fs.refresh_state = refresh_state
    fs.updated_at = now

    if last_refreshed_at is not None:
        fs.last_refreshed_at = last_refreshed_at
    elif refresh_state == "fresh":
        fs.last_refreshed_at = now

    if refresh_state == "fresh":
        fs.error_code = None
        fs.error_message = None
    elif error_code is not None:
        fs.error_code = error_code
        fs.error_message = error_message

    await session.flush()
    return fs