api_server.py

"""REST API for document upload and download.

Separate from MCP server — mounted alongside it in server.py.
Requires the same API_KEY auth when set.
"""

import logging
from pathlib import Path

from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import FileResponse, JSONResponse
from starlette.routing import Route

logger = logging.getLogger(__name__)

# Max upload size: 100 MB
_MAX_UPLOAD_BYTES = 100 * 1024 * 1024

_ALLOWED_EXTENSIONS = {
    ".md", ".pdf", ".png", ".jpg", ".jpeg", ".gif", ".webp",
    ".docx", ".doc", ".pptx", ".rtf", ".epub",
    ".html", ".htm", ".csv", ".txt", ".xlsx", ".xls",
}

_DEFAULT_LIST_LIMIT = 200


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def _api_error(code: str, message: str, status_code: int = 400) -> JSONResponse:
    """Build a structured JSON error response."""
    return JSONResponse({"error": True, "code": code, "message": message}, status_code=status_code)


def _safe_subpath(base: Path, user_input: str) -> Path | None:
    """Resolve a user-provided relative path safely under base.

    Returns the resolved Path if safe, or None if it escapes base.
    """
    candidate = (base / user_input).resolve()
    try:
        candidate.relative_to(base.resolve())
        return candidate
    except ValueError:
        return None


# ---------------------------------------------------------------------------
# Endpoints
# ---------------------------------------------------------------------------


async def upload(request: Request) -> JSONResponse:
    """Upload a file to documents_root.

    Multipart form data:
        file: The file to upload (required).
        directory: Subdirectory within documents_root (optional, default: root).
    """
    docs_root: Path = request.app.state.documents_root

    # Early size rejection via Content-Length header
    content_length = request.headers.get("content-length")
    if content_length and int(content_length) > _MAX_UPLOAD_BYTES:
        return _api_error("file_too_large", f"File exceeds {_MAX_UPLOAD_BYTES // (1024*1024)} MB limit", 413)

    content_type = request.headers.get("content-type", "")
    if "multipart/form-data" not in content_type:
        return _api_error("invalid_request", "Expected multipart/form-data")

    form = await request.form()
    file = form.get("file")
    if file is None:
        return _api_error("missing_file", "No file field in upload")

    filename = file.filename
    if not filename:
        return _api_error("missing_filename", "File has no filename")

    # Validate extension
    ext = Path(filename).suffix.lower()
    if ext not in _ALLOWED_EXTENSIONS:
        return _api_error("invalid_file_type", f"File type '{ext}' not allowed. Allowed: {sorted(_ALLOWED_EXTENSIONS)}")

    # Sanitize filename — prevent path traversal
    safe_name = Path(filename).name
    if not safe_name or safe_name.startswith("."):
        return _api_error("invalid_filename", "Invalid filename")

    # Resolve target directory safely
    directory = form.get("directory", "")
    if isinstance(directory, str) and directory.strip():
        target_dir = _safe_subpath(docs_root, directory.strip().strip("/"))
        if target_dir is None:
            return _api_error("invalid_directory", "Directory escapes documents root")
    else:
        target_dir = docs_root

    # Read file with size enforcement
    content = await file.read()
    if len(content) > _MAX_UPLOAD_BYTES:
        return _api_error("file_too_large", f"File exceeds {_MAX_UPLOAD_BYTES // (1024*1024)} MB limit", 413)
    if len(content) == 0:
        return _api_error("empty_file", "Uploaded file is empty")

    target_dir.mkdir(parents=True, exist_ok=True)
    target_path = target_dir / safe_name

    target_path.write_bytes(content)

    rel_path = str(target_path.relative_to(docs_root)).replace("\\", "/")
    logger.info("Uploaded: %s (%d bytes)", rel_path, len(content))

    # "doc_id" is a deprecated alias of "rel_path": this value has always been a
    # documents_root-relative path, never an index doc id (DocIDStore mints those
    # and flow_index_vault namespaces them per source, e.g. "documents::00001").
    # Kept so existing clients keep working; scheduled for removal in #1232.
    return JSONResponse(
        {"uploaded": True, "rel_path": rel_path, "doc_id": rel_path, "size": len(content)},
        status_code=201,
    )


async def download(request: Request) -> FileResponse | JSONResponse:
    """Download a file by its documents_root-relative path.

    GET /api/documents/{rel_path:path}

    The path parameter is a filesystem path under documents_root — the index's
    ``rel_path`` field — not an index ``doc_id``.
    """
    docs_root: Path = request.app.state.documents_root

    rel_path = request.path_params.get("rel_path", "")
    if not rel_path:
        return _api_error("missing_rel_path", "rel_path path parameter required")

    file_path = _safe_subpath(docs_root, rel_path)
    if file_path is None:
        return _api_error("invalid_path", "Path escapes documents root")

    if not file_path.is_file():
        return _api_error("not_found", f"File not found: {rel_path}", 404)

    return FileResponse(file_path, filename=file_path.name)


async def list_documents(request: Request) -> JSONResponse:
    """List files in a directory within documents_root.

    GET /api/documents/?directory=optional/subdir&limit=200&offset=0
    """
    docs_root: Path = request.app.state.documents_root
    directory = request.query_params.get("directory", "")
    limit = min(int(request.query_params.get("limit", _DEFAULT_LIST_LIMIT)), _DEFAULT_LIST_LIMIT)
    offset = int(request.query_params.get("offset", 0))

    if directory:
        target_dir = _safe_subpath(docs_root, directory.strip().strip("/"))
        if target_dir is None:
            return _api_error("invalid_directory", "Directory escapes documents root")
    else:
        target_dir = docs_root

    if not target_dir.is_dir():
        return _api_error("not_found", f"Directory not found: {directory}", 404)

    files = []
    for entry in sorted(target_dir.iterdir(), key=lambda p: p.name):
        rel = str(entry.relative_to(docs_root)).replace("\\", "/")
        if entry.is_dir():
            files.append({"name": entry.name, "type": "directory", "path": rel})
        elif entry.suffix.lower() in _ALLOWED_EXTENSIONS:
            files.append({
                "name": entry.name,
                "type": "file",
                "path": rel,
                "size": entry.stat().st_size,
            })

    total = len(files)
    files = files[offset:offset + limit]

    return JSONResponse({
        "directory": directory or ".",
        "files": files,
        "total": total,
        "offset": offset,
        "limit": limit,
    })


def _bool_param(value: object, default: bool = False) -> bool:
    if value is None:
        return default
    if isinstance(value, bool):
        return value
    if isinstance(value, str):
        return value.strip().lower() in {"1", "true", "yes", "on"}
    return bool(value)


def _int_param(value: object, default: int) -> int:
    try:
        return int(value)  # type: ignore[arg-type]
    except (TypeError, ValueError):
        return default


def _search_kwargs(payload: dict) -> dict:
    return {
        "query": payload.get("query", ""),
        "top_k": _int_param(payload.get("top_k"), 10),
        "doc_id_prefix": payload.get("doc_id_prefix"),
        "source_type": payload.get("source_type"),
        "source_name": payload.get("source_name"),
        "tags": payload.get("tags"),
        "status": payload.get("status"),
        "folder": payload.get("folder"),
        "prefer_recent": _bool_param(payload.get("prefer_recent"), False),
        "metadata_filters": payload.get("metadata_filters"),
        "enr_doc_type": payload.get("enr_doc_type"),
        "enr_topics": payload.get("enr_topics"),
        "filter": payload.get("filter"),
    }


async def search(request: Request) -> JSONResponse:
    """Search the existing index via the MCP file_search implementation.

    POST JSON body mirrors file_search parameters. This endpoint exists for
    HTTP-only clients such as external LangChain retriever adapters.
    """
    if "application/json" not in request.headers.get("content-type", ""):
        return _api_error("invalid_request", "Expected application/json")

    try:
        payload = await request.json()
    except Exception:
        return _api_error("invalid_json", "Request body must be valid JSON")
    if not isinstance(payload, dict):
        return _api_error("invalid_request", "JSON body must be an object")

    from starlette.concurrency import run_in_threadpool

    from mcp_server import _file_search_impl

    # Blocking work (embed → LanceDB → rerank) must not run on the serving
    # event loop: it starves every other route, including the unauthenticated
    # /health probe the container healthcheck polls with a 5s timeout (#1086).
    result = await run_in_threadpool(_file_search_impl, **_search_kwargs(payload))
    status_code = 400 if isinstance(result, dict) and result.get("error") else 200
    return JSONResponse(result, status_code=status_code)


async def index_document(request: Request) -> JSONResponse:
    """Targeted single-document index (TICKET-6).

    POST JSON: {"rel_path"|"abs_path"|"doc_id": ..., "source_name"?: "documents",
    "force"?: false}. Called by comm-data-store-hooks per newly-deposited
    attachment so it becomes OCR'd/described/searchable within seconds without a
    full-source scan. Inherits the server's Bearer API_KEY auth like the rest of
    /api/*. Idempotent: an unchanged file is skipped without re-OCR. Known gap:
    keyword (FTS) visibility awaits the next full sweep; vector search sees the
    doc immediately.
    """
    if "application/json" not in request.headers.get("content-type", ""):
        return _api_error("invalid_request", "Expected application/json")
    try:
        payload = await request.json()
    except Exception:
        return _api_error("invalid_json", "Request body must be valid JSON")
    if not isinstance(payload, dict):
        return _api_error("invalid_request", "JSON body must be an object")

    target = payload.get("rel_path") or payload.get("abs_path") or payload.get("doc_id")
    if not target:
        return _api_error("missing_target", "One of rel_path, abs_path, or doc_id is required")
    source_name = payload.get("source_name") or "documents"
    force = bool(payload.get("force", False))

    from starlette.concurrency import run_in_threadpool

    import flow_index_vault

    try:
        result = await run_in_threadpool(
            flow_index_vault.index_document_flow,
            target=str(target),
            source_name=str(source_name),
            force=force,
        )
    except ValueError as exc:  # unknown source, etc.
        return _api_error("invalid_request", str(exc))
    except Exception as exc:  # noqa: BLE001 — surface as 500 with reason
        logger.exception("index_document failed for %s", target)
        return JSONResponse(
            {"status": "error", "reason": "index_failed", "detail": str(exc)},
            status_code=500,
        )

    if result.get("status") == "queued":
        status_code = 202
    else:
        status_code = 404 if result.get("status") == "error" else 200
    return JSONResponse(result, status_code=status_code)


def build_api_app(documents_root: Path) -> Starlette:
    """Build the REST API Starlette app.

    Args:
        documents_root: Path to the documents directory (injected, not re-loaded per request).
    """
    # Starlette dispatches on the first matching route, and a ``:path``
    # convertor compiles to ``(?P<rel_path>.*)`` — which also matches the empty
    # remainder. So every exact route under /documents/ must be registered
    # before the catch-all download route, or it is shadowed by it.
    routes = [
        Route("/upload", upload, methods=["POST"]),
        Route("/search", search, methods=["POST"]),
        Route("/index/document", index_document, methods=["POST"]),
        Route("/documents", list_documents, methods=["GET"]),
        Route("/documents/", list_documents, methods=["GET"]),
        Route("/documents/{rel_path:path}", download, methods=["GET"]),
    ]
    app = Starlette(routes=routes)
    app.state.documents_root = Path(documents_root)
    return app