kite/backend/api/cloud/browse.py

"""
Canonical connection-ID browse endpoint for Phase 12.

GET /api/cloud/connections/{connection_id}/items
  - Owner-scoped: resolves connection by UUID, not provider name
  - Stale-while-revalidate: returns durable cached rows immediately; schedules
    background refresh if stale or first visit
  - T-12-01: IDOR-protected via resolve_owned_connection
  - T-12-03: response uses whitelisted schemas, no credentials in response
  - D-18/CACHE-01: never downloads file bytes, never mutates quota
"""
from __future__ import annotations

import asyncio
import uuid
from typing import Optional

import httpx
from fastapi import APIRouter, Depends, HTTPException, Request, status
from sqlalchemy.ext.asyncio import AsyncSession

from api.cloud.schemas import (
    CloudBrowseResponse,
    CloudCapabilityOut,
    CloudItemOut,
    FolderFreshnessOut,
)
from db.models import CloudConnection, CloudItem, CloudFolderState, CloudItemTopic, Topic, User
from sqlalchemy import select
from deps.auth import get_regular_user
from deps.db import get_db
from deps.utils import parse_uuid
from services.cloud_items import (
    ConnectionNotFound,
    apply_listing_and_finalize,
    list_cloud_children,
    get_or_create_folder_state,
    reconcile_cloud_listing,
    resolve_owned_connection,
    update_folder_state,
)
from services.rate_limiting import account_limiter
from storage.cloud_backend_factory import build_cloud_resource_adapter
from storage.cloud_utils import decrypt_credentials, encrypt_credentials
from config import settings

router = APIRouter()

_DISPLAY_NAMES = {
    "google_drive": "Google Drive",
    "onedrive": "OneDrive",
    "nextcloud": "Nextcloud",
    "webdav": "WebDAV server",
}


def _master_key() -> bytes:
    return settings.cloud_creds_key.encode()


def _decrypt_connection(conn: CloudConnection, user_id: uuid.UUID) -> dict:
    return decrypt_credentials(_master_key(), str(user_id), conn.credentials_enc)


def _capability_out(caps: dict) -> dict[str, CloudCapabilityOut]:
    return {
        action: CloudCapabilityOut(
            action=cap.action,
            state=cap.state,
            reason=cap.reason,
            message=cap.message,
        )
        for action, cap in caps.items()
    }


def _item_out(
    item: CloudItem,
    topic_names: list[str] | None = None,
) -> CloudItemOut:
    """Build a CloudItemOut from a CloudItem ORM row.

    Phase 14.1: topic_names and analysis_status/is_stale are populated from
    a caller-supplied batched topic lookup so we avoid N+1 queries.
    Folder rows always receive topic_names=[] and analysis_status=None.
    extracted_text is deliberately excluded (T-14.1-10 — row payload must stay lightweight).
    """
    is_file = item.kind == "file"
    status = item.analysis_status if is_file else None
    return CloudItemOut(
        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,
        capabilities={},  # Phase 13 will add per-item capabilities
        # Phase 14.1 row-parity fields (T-14.1-10 allowlist)
        topics=topic_names or [],
        analysis_status=status,
        is_stale=(status == "stale"),
    )


async def _batch_load_topics(
    session,
    file_ids: list,
) -> dict:
    """Batch-load topic names for a list of CloudItem UUIDs.

    Returns a dict mapping cloud_item_id (UUID) → list[str] of topic names.
    Uses a single JOIN query to avoid N+1 when building browse rows (T-14.1-10).
    """
    if not file_ids:
        return {}

    stmt = (
        select(CloudItemTopic.cloud_item_id, Topic.name)
        .join(Topic, Topic.id == CloudItemTopic.topic_id)
        .where(CloudItemTopic.cloud_item_id.in_(file_ids))
        .order_by(CloudItemTopic.cloud_item_id, Topic.name)
    )
    result = await session.execute(stmt)
    rows = result.fetchall()

    topics_by_item: dict = {}
    for cloud_item_id, topic_name in rows:
        topics_by_item.setdefault(cloud_item_id, []).append(topic_name)
    return topics_by_item


def _freshness_out(fs: CloudFolderState) -> FolderFreshnessOut:
    return FolderFreshnessOut(
        refresh_state=fs.refresh_state,
        last_refreshed_at=fs.last_refreshed_at,
        error_code=fs.error_code,
        error_message=fs.error_message,
    )


# ── Legacy helpers used by compatibility route in connections.py ─────────────

async def _fetch_google_drive_folders(credentials: dict, folder_id: str) -> list:
    """Legacy listing helper preserved for compatibility route."""
    import datetime as dt
    from google.oauth2.credentials import Credentials as GoogleCreds
    from googleapiclient.discovery import build

    expiry = None
    if expiry_str := credentials.get("expiry"):
        try:
            expiry = dt.datetime.fromisoformat(expiry_str)
        except ValueError:
            pass

    creds = GoogleCreds(
        token=credentials.get("access_token"),
        refresh_token=credentials.get("refresh_token"),
        token_uri=credentials.get("token_uri", "https://oauth2.googleapis.com/token"),
        client_id=credentials.get("client_id"),
        client_secret=credentials.get("client_secret"),
        expiry=expiry,
    )

    def _list_files() -> list:
        service = build("drive", "v3", credentials=creds, cache_discovery=False)
        response = service.files().list(
            q=f"'{folder_id}' in parents and trashed=false",
            fields="files(id,name,mimeType,size)",
            pageSize=200,
        ).execute()
        items = []
        for item in response.get("files", []):
            is_dir = item.get("mimeType") == "application/vnd.google-apps.folder"
            items.append({
                "id": item["id"],
                "name": item["name"],
                "is_dir": is_dir,
                "size": int(item.get("size", 0)) if not is_dir else 0,
            })
        return items

    return await asyncio.to_thread(_list_files)


async def _fetch_onedrive_folders(credentials: dict, folder_id: str) -> list:
    """Legacy listing helper preserved for compatibility route."""
    if folder_id in ("root", ""):
        url = "https://graph.microsoft.com/v1.0/me/drive/root/children"
    else:
        url = f"https://graph.microsoft.com/v1.0/me/drive/items/{folder_id}/children"

    async with httpx.AsyncClient() as client:
        resp = await client.get(
            url,
            headers={"Authorization": f"Bearer {credentials.get('access_token', '')}"},
            timeout=30,
        )
        resp.raise_for_status()
        data = resp.json()

    items = []
    for item in data.get("value", []):
        is_dir = "folder" in item
        items.append({
            "id": item["id"],
            "name": item["name"],
            "is_dir": is_dir,
            "size": item.get("size", 0) if not is_dir else 0,
        })
    return items


async def _fetch_webdav_folders(provider: str, credentials: dict, folder_id: str) -> list:
    """Legacy listing helper preserved for compatibility route."""
    from storage.cloud_backend_factory import build_cloud_backend
    webdav_path = "" if folder_id == "root" else folder_id
    backend = build_cloud_backend(provider, credentials)
    return await backend.list_folder(webdav_path)


# ── Canonical connection-ID browse endpoint ──────────────────────────────────

@router.get("/connections/{connection_id}/items", response_model=CloudBrowseResponse)
@account_limiter.limit("100/minute")
async def browse_connection_items(
    connection_id: uuid.UUID,
    request: Request,
    parent_ref: Optional[str] = None,
    session: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_regular_user),
) -> CloudBrowseResponse:
    """Browse the items of a connection by UUID.

    T-12-01: IDOR protection — connection resolved by owner UUID, not provider alone.
    T-12-03: response schema excludes credentials_enc and all credential fields.
    D-18/CACHE-01: returns durable cached rows; never downloads file bytes.
    D-11/D-12: cached rows returned immediately; background refresh scheduled if stale.
    D-13/D-14: on refresh failure, cached rows retained with warning state.
    """
    request.state.current_user = current_user

    uid_str = str(current_user.id)
    cid_str = str(connection_id)

    # Resolve connection ownership (T-12-01)
    try:
        conn = await resolve_owned_connection(
            session,
            connection_id=cid_str,
            user_id=uid_str,
        )
    except ConnectionNotFound:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Connection not found")

    # Ensure folder state row exists
    folder_state = await get_or_create_folder_state(
        session,
        user_id=uid_str,
        connection_id=cid_str,
        parent_ref=parent_ref or "",
    )
    await session.commit()

    # Return durable cached rows
    cached_items = await list_cloud_children(
        session,
        user_id=uid_str,
        connection_id=cid_str,
        parent_ref=parent_ref,
    )

    # Get connection-level capabilities (from a fresh call to avoid stale state)
    try:
        credentials = _decrypt_connection(conn, current_user.id)
        adapter = build_cloud_resource_adapter(conn.provider, credentials)
        conn_caps = await adapter.get_capabilities(connection_id, current_user.id)
    except Exception:
        from storage.cloud_base import ACTIONS, CloudCapability, STATE_TEMPORARILY_UNAVAILABLE, REASON_OFFLINE
        conn_caps = {
            action: CloudCapability(
                action=action,
                state=STATE_TEMPORARILY_UNAVAILABLE,
                reason=REASON_OFFLINE,
                message="Provider temporarily unreachable.",
            )
            for action in ACTIONS
        }

    # Refresh in background if first visit (no items), currently refreshing,
    # never refreshed, or stale (last refresh > 5 minutes ago)
    import datetime as _dt
    _stale_threshold = _dt.timedelta(minutes=5)
    _is_stale = (
        folder_state.last_refreshed_at is None
        or (_dt.datetime.now(_dt.timezone.utc) - folder_state.last_refreshed_at) > _stale_threshold
    )
    should_refresh = (
        not cached_items
        or folder_state.refresh_state in ("refreshing",)
        or _is_stale
    )
    if should_refresh and not cached_items:
        # First visit: do a synchronous bounded fetch so user sees content.
        # apply_listing_and_finalize is the single shared gate — it handles
        # complete vs. incomplete semantics without unconditional fresh state.
        try:
            credentials = _decrypt_connection(conn, current_user.id)
            adapter = build_cloud_resource_adapter(conn.provider, credentials)
            listing = await adapter.list_folder(
                connection_id,
                current_user.id,
                parent_ref=parent_ref,
            )
            await apply_listing_and_finalize(
                session,
                user_id=uid_str,
                connection_id=cid_str,
                parent_ref=parent_ref,
                listing=listing,
            )
            await session.commit()
            # Re-read items after reconciliation
            cached_items = await list_cloud_children(
                session,
                user_id=uid_str,
                connection_id=cid_str,
                parent_ref=parent_ref,
            )
            # Re-fetch folder state
            folder_state = await get_or_create_folder_state(
                session,
                user_id=uid_str,
                connection_id=cid_str,
                parent_ref=parent_ref or "",
            )
        except Exception as exc:
            # Retain cached rows (empty on first visit), set warning state
            await update_folder_state(
                session,
                user_id=uid_str,
                connection_id=cid_str,
                parent_ref=parent_ref or "",
                refresh_state="warning",
                error_code="provider_error",
                error_message="Provider temporarily unavailable. Retrying in background.",
            )
            await session.commit()
            folder_state = await get_or_create_folder_state(
                session,
                user_id=uid_str,
                connection_id=cid_str,
                parent_ref=parent_ref or "",
            )
    elif should_refresh and cached_items:
        # Has cached items — schedule background refresh via Celery
        try:
            from tasks.cloud_tasks import refresh_cloud_folder
            refresh_cloud_folder.delay(
                str(current_user.id),
                str(connection_id),
                parent_ref,
            )
        except Exception:
            pass  # Celery unavailable — serve stale cache without failing browse

    display_name = conn.display_name_override or conn.display_name

    # Phase 14.1: Batch-load topic names for all file rows in one query (T-14.1-10).
    # Folder rows are excluded from the topic lookup (folders never carry topics).
    file_ids = [item.id for item in cached_items if item.kind == "file"]
    topics_by_item = await _batch_load_topics(session, file_ids)

    return CloudBrowseResponse(
        connection_id=str(connection_id),
        provider=conn.provider,
        display_name=display_name,
        parent_ref=parent_ref,
        items=[
            _item_out(item, topic_names=topics_by_item.get(item.id))
            for item in cached_items
        ],
        capabilities=_capability_out(conn_caps),
        freshness=_freshness_out(folder_state),
    )