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 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."""


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


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