brain-of-reese/app/api/git_sources.py

"""Admin-managed sources API (phase 35, task 02; local kind, phase 38).

Admin-only CRUD under ``/api/git-sources`` (phase 16 pattern, A10
extended — the public API surface stays stateless and the signed cookie
remains the only session state, same as ``/api/steering`` and
``/api/sync``): the ``git_sources`` table holds the sources the Sync
button (phase 32) and ``import_docs`` (phase 28) import — ``kind='git'``
rows carry the repo URL to clone/pull, ``kind='local'`` rows (phase 38)
carry an existing directory on the server to walk directly. DB rows win
over ``BOR_GIT_SOURCES``, which is a git-only fallback while the table is
empty (the phase's locked decision — ``from_env`` tells the UI which list
it is looking at, so the page can show the env note only while the
fallback is active).

Routes: ``GET`` (DB rows oldest-first, or the env list with
``from_env: true`` while the table is empty; rows carry ``kind`` +
``path``, git rows — and env rows — report ``path: null``), ``POST``
(201, validated create; ``kind`` selects the validation: git → exactly
the phase-35 URL contract, local → an existing absolute directory, else
422 naming the path), ``POST /upload`` (phase 49 — admin archive upload:
``.tar``/``.tar.gz``/``.tgz``/``.zip`` streamed with a size cap, safely
unpacked, atomically swapped in over an existing folder of the same
name, row upserted, then the synchronous single-source scan — see
:func:`upload_archive`), ``DELETE /{source_id}`` (204). The whole
router sits behind :func:`app.core.auth.require_admin` — anonymous
callers get 403 on every route.

No credential-echo path: git URLs may embed ``user:pass@`` (phase 32's
masking discipline), so every git 409/422 detail is a fixed generic
string that never repeats the submitted URL. Local paths are not
secrets — the local 422/409 details name the (expanded) path so the
owner sees exactly which directory failed.

Scope boundary (phase locked decisions): the CRUD routes do NOT
clone, import, or prune anything — the existing Sync button performs
that (a removal prunes on the next sync, ``prune=True``). The phase-49
upload route is the exception: it unpacks the archive and then scans
the single source synchronously in the request (``import_sources``
with ``prune=True`` + the change-gated overview refresh) and answers
with the sync-style counts.
"""
from __future__ import annotations

import logging
import re
import shutil
import time
import uuid
from pathlib import Path
from typing import Literal, cast

from fastapi import APIRouter, Depends, File, HTTPException, Response, UploadFile
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session

from app.api.sync import _sanitize_error
from app.config import get_settings
from app.core.auth import require_admin
from app.db import get_db
from app.models import GitSource
from app.rag.archive_upload import (
    ARCHIVE_SUFFIXES,
    ArchiveUploadError,
    archive_source_name,
    swap_in,
    unpack_archive,
)
from app.rag.importer import import_sources
from app.rag.llm import LLMClient, ModelUnavailableError, check_models
from app.rag.overview import regenerate_overview
from app.schemas import GitSourceIn, GitSourceList, GitSourceOut, GitSourceRow, UploadOut

logger = logging.getLogger("app.api.git_sources")

router = APIRouter(
    prefix="/git-sources",
    tags=["git-sources"],
    dependencies=[Depends(require_admin)],  # phase 16 pattern: admin-only surface
)

#: One upload at a time (phase 49, task 02 — the phase-32 ``_task``
#: spirit): the flag is held from the name gate through the scan
#: response. A plain bool, not an ``asyncio.Lock`` — it is checked and
#: set with no await in between (a single app loop can never enter
#: twice), and it stays correct across requests that run on separate
#: event loops (the TestClient convention).
_upload_in_progress = False

#: Streaming read size while counting compressed upload bytes (1 MiB
#: chunks — the task-02 cap check granularity).
_STREAM_CHUNK = 1 << 20

#: Accepted git URL shapes — the trimmed URL must *start* with one of them.
#: Covers the phase-28 real URLs (HTTPS + ``git@`` SSH); scp-style
#: ``host:repo`` is deliberately rejected (422). ASSUMPTION (task 02): the
#: accepted shapes are exactly these four prefixes.
URL_RE = re.compile(r"^(https?://|ssh://|git@)")


@router.get("", response_model=GitSourceList)
def list_git_sources(
    db: Session = Depends(get_db),  # noqa: B008
) -> GitSourceList:
    """The effective source list (git + local rows, phase 38).

    DB rows ordered by ``(added_at, id)`` (oldest first, id tie-break for
    same-timestamp inserts) with ``from_env: false`` — each row carries
    its ``kind`` and, for local rows, the stored ``path`` (git rows and
    env rows report ``path: null``); while the table is empty, the
    ``BOR_GIT_SOURCES`` env URLs as git rows (the env fallback is
    git-only) with null ``id``/``added_at`` and ``from_env: true``.
    """
    rows = db.scalars(
        select(GitSource).order_by(GitSource.added_at.asc(), GitSource.id.asc())
    ).all()
    if rows:
        return GitSourceList(
            sources=[
                # ``ck_git_sources_kind`` (migration 0007) guarantees the
                # value is 'git' or 'local' — the cast documents that.
                GitSourceRow(
                    id=row.id,
                    kind=cast(Literal["git", "local"], row.kind),
                    url=row.url,
                    path=row.path,
                    added_at=row.added_at,
                )
                for row in rows
            ],
            from_env=False,
        )
    return GitSourceList(
        sources=[
            GitSourceRow(id=None, kind="git", url=url, path=None, added_at=None)
            for url in get_settings().git_source_list
        ],
        from_env=True,
    )


@router.post("", response_model=GitSourceOut, status_code=201)
def create_git_source(
    payload: GitSourceIn,
    db: Session = Depends(get_db),  # noqa: B008
) -> GitSourceOut:
    """Store one source (fields already trimmed by the schema).

    ``kind="git"`` (default) — exactly the phase-35 contract: 422 when
    the URL shape is not one of the accepted prefixes (generic detail —
    the input is never echoed), 409 when the trimmed URL is already
    stored (the unique index is the backstop against a concurrent insert
    the pre-check missed), 201 + the created row otherwise.

    ``kind="local"`` — ``path`` must expand (``~``) to an absolute,
    existing directory on the server: 422 naming the path otherwise
    (fail loud at add-time — the owner sees it immediately), 409 when
    the path is already stored (detail names the path), 201 + the stored
    row otherwise (``url`` holds the expanded path — the table's
    NOT-NULL location column).

    Wrong field combinations (git without url, local without path, both
    kinds' fields) are 422 with fixed, input-free details.
    """
    row = _create_git_row(payload, db) if payload.kind == "git" else _create_local_row(payload, db)
    return GitSourceOut(id=row.id, url=row.url, added_at=row.added_at)


def _commit_new(row: GitSource, duplicate_detail: str, db: Session) -> GitSource:
    """Insert ``row``; the unique index is the backstop — a concurrent
    insert the pre-check missed still yields the generic 409, never a
    500 (phase-35 convention, now shared by both kinds)."""
    db.add(row)
    try:
        db.commit()
    except IntegrityError:
        db.rollback()
        raise HTTPException(status_code=409, detail=duplicate_detail) from None
    db.refresh(row)
    return row


def _create_git_row(payload: GitSourceIn, db: Session) -> GitSource:
    """``kind=git`` — the phase-35 URL contract, unchanged (A10: no
    credential echo, so every detail is a fixed string)."""
    if payload.path is not None:
        raise HTTPException(status_code=422, detail="a git source takes a url, not a path")
    if payload.url is None:
        raise HTTPException(status_code=422, detail="a git source requires a url")
    url = payload.url
    if not URL_RE.match(url):
        raise HTTPException(
            status_code=422, detail="not a valid git URL (expected https://, ssh:// or git@…)"
        )
    if db.scalar(select(GitSource).where(GitSource.url == url)) is not None:
        raise HTTPException(status_code=409, detail="a git source with this URL already exists")
    return _commit_new(
        GitSource(url=url, kind="git"), "a git source with this URL already exists", db
    )


def _create_local_row(payload: GitSourceIn, db: Session) -> GitSource:
    """``kind=local`` — fail-loud add-time validation (phase 38):
    trimmed → ``expanduser()`` → absolute + existing directory, else 422
    naming the path (not a secret, unlike a git URL)."""
    if payload.url is not None:
        raise HTTPException(status_code=422, detail="a local source takes a path, not a url")
    if payload.path is None:
        raise HTTPException(status_code=422, detail="a local source requires a path")
    expanded = Path(payload.path).expanduser()
    if not expanded.is_absolute() or not expanded.is_dir():
        raise HTTPException(
            status_code=422, detail=f"local source path is not a directory: {expanded}"
        )
    path = str(expanded)
    if db.scalar(select(GitSource).where(GitSource.path == path)) is not None:
        raise HTTPException(
            status_code=409, detail=f"a local source with this path already exists: {path}"
        )
    # ``url`` is the table's NOT-NULL location column (phase 38: local
    # rows carry the expanded path there too — git URL shapes and absolute
    # paths cannot collide).
    return _commit_new(
        GitSource(url=path, kind="local", path=path),
        f"a local source with this path already exists: {path}",
        db,
    )


@router.post("/upload", response_model=UploadOut)
async def upload_archive(
    file: UploadFile = File(...),  # noqa: B008
    db: Session = Depends(get_db),  # noqa: B008
) -> UploadOut:
    """Upload a source archive and scan it (phase 49, task 02).

    The scan is **synchronous in the request** (phase locked decisions,
    owner-confirmed) and mirrors the admin sync pipeline:

    1. name/format gate — only ``.tar``/``.tar.gz``/``.tgz``/``.zip``
       (422 naming the accepted set) and a safe source name
       (``archive_source_name`` — its message is the 422 detail);
    2. one at a time — 409 ``an upload is already in progress``;
    3. stream the upload in 1 MiB chunks into a dotfile temp with the
       ``upload_max_mb`` cap — 413 naming the cap, temp deleted;
    4. unpack to a temp sibling (traversal/symlink/device/corrupt/
       over-cap all 422 with the task-01 user-safe message, temps
       deleted); a zero-entry archive is 422 ``the archive contains no
       files`` — an archive with only non-A9 files is a VALID
       replacement (the scan indexes nothing, prune removes the
       source's docs);
    5. atomic swap-in — a same-name re-upload replaces the previous
       folder in place; a failure leaves the previous folder/row/KB
       untouched (422);
    6. upsert the row by ``path`` (``kind='local'``; an existing row is
       left as-is — ``added_at`` preserved — and the unique index is
       the 409 backstop);
    7. fail-fast ``check_models`` — 503 with the sanitized
       model-unavailable message; the folder/row are already committed,
       so the next sync/re-upload retries idempotently;
    8. ``import_sources([folder], llm, prune=True)`` + the change-gated
       ``regenerate_overview``;
    9. one INFO log line (PLAN §9 / AGENTS.md rule 10);
    10. 200 with the sync-detail count keys (``UploadOut``).
    """
    started = time.monotonic()
    settings = get_settings()
    total = 0

    # 1. Name/format gate — the accepted formats first (the 422 names
    #    them), then the task-01 safe-name derivation. A BARE suffix
    #    ("tar.gz") is an accepted format with no usable stem — it
    #    passes here and gets task-01's "no usable source name" 422.
    #    No upload dir is created for a rejected name.
    filename = file.filename or ""
    lowered = filename.lower()
    if not any(
        lowered.endswith(suffix) or lowered == suffix.lstrip(".")
        for suffix in ARCHIVE_SUFFIXES
    ):
        raise HTTPException(
            status_code=422,
            detail="only .tar, .tar.gz, .tgz or .zip archives are accepted",
        )
    try:
        name = archive_source_name(filename)
    except ArchiveUploadError as e:
        raise HTTPException(status_code=422, detail=str(e)) from None

    # 2. One at a time — the flag is checked and set with no await
    #    between, so the single app loop can never enter twice.
    global _upload_in_progress
    if _upload_in_progress:
        raise HTTPException(status_code=409, detail="an upload is already in progress")
    _upload_in_progress = True

    upload_root = Path(settings.upload_dir).expanduser()
    upload_root.mkdir(parents=True, exist_ok=True)
    max_bytes = settings.upload_max_mb * 1024 * 1024
    temp_upload = upload_root / f".{name}.{uuid.uuid4().hex}.upload"
    temp_unpack = upload_root / f".{name}.{uuid.uuid4().hex}.unpack"
    try:
        # 3. Stream with the compressed-size cap — dotfile temps are
        #    hidden from the upload dir's listing.
        try:
            with open(temp_upload, "wb") as out:
                while chunk := await file.read(_STREAM_CHUNK):
                    total += len(chunk)
                    if total > max_bytes:
                        raise HTTPException(
                            status_code=413,
                            detail=f"the upload exceeds the {settings.upload_max_mb} MiB limit",
                        )
                    out.write(chunk)
        except HTTPException:
            temp_upload.unlink(missing_ok=True)
            raise
        # 4. Unpack to a temp sibling; the compressed bytes are no
        #    longer needed once unpacked (phase locked decision: only
        #    the unpacked content is kept).
        try:
            unpack_archive(temp_upload, temp_unpack, max_bytes)
        except ArchiveUploadError as e:
            temp_upload.unlink(missing_ok=True)
            shutil.rmtree(temp_unpack, ignore_errors=True)
            raise HTTPException(status_code=422, detail=str(e)) from None
        temp_upload.unlink(missing_ok=True)
        if not any(temp_unpack.iterdir()):
            # Zero entries = a user error. (Only non-A9 files is NOT an
            # error — it still has entries and is a valid replacement.)
            shutil.rmtree(temp_unpack, ignore_errors=True)
            raise HTTPException(status_code=422, detail="the archive contains no files")
        # 5. Swap in — a same-name re-upload replaces the previous
        #    folder atomically; a failure leaves it, the row, and the
        #    KB untouched.
        final_dir = upload_root / name
        try:
            swap_in(temp_unpack, final_dir)
        except ArchiveUploadError as e:
            shutil.rmtree(temp_unpack, ignore_errors=True)
            raise HTTPException(status_code=422, detail=str(e)) from None
        # 6. Upsert the row by path — no duplicates: an existing row is
        #    left exactly as it is (``added_at`` preserved); the unique
        #    index is the 409 backstop for a concurrent insert the
        #    pre-check missed.
        path = str(final_dir)
        if db.scalar(select(GitSource).where(GitSource.path == path)) is None:
            _commit_new(
                GitSource(url=path, kind="local", path=path),
                f"a local source with this path already exists: {path}",
                db,
            )
        # Release the request session NOW — the handler never touches
        # ``db`` again (the scan below uses its own sessions). If the
        # session stayed open, its uncommitted transaction (the
        # ``_commit_new`` refresh SELECT) would hold ``git_sources``
        # locks for the whole scan, and any concurrent TRUNCATE of the
        # KB tables (the E2E isolation fixtures) would deadlock against
        # the scan's own document locks — a cycle Postgres cannot see.
        # ``get_db``'s teardown close() is idempotent.
        db.close()
        # 7. Fail-fast models (phase 41) — 503 with the sanitized
        #    message; nothing else is rolled back (the folder/row are
        #    committed and the next sync/re-upload retries idempotently).
        llm = LLMClient()
        try:
            await check_models(llm)
        except ModelUnavailableError as e:
            raise HTTPException(status_code=503, detail=_sanitize_error(str(e))) from None
        # 8. Scan — single source, prune (dropped files leave the KB),
        #    then the change-gated overview refresh (phases 31/32).
        summary = await import_sources([final_dir], llm, prune=True)
        overview = False
        if summary.added + summary.updated > 0:
            overview = await regenerate_overview(llm)
    finally:
        _upload_in_progress = False
        # No temp may survive any failure path (defensive — each step
        # already cleans its own; on success both are already gone).
        temp_upload.unlink(missing_ok=True)
        shutil.rmtree(temp_unpack, ignore_errors=True)

    # 9. Per-upload log line (PLAN §9 / AGENTS.md rule 10).
    logger.info(
        "upload: name=%s file=%s bytes_in=%d files=%d added=%d updated=%d "
        "unchanged=%d pruned=%d errors=%d overview=%s total_ms=%d",
        name,
        filename,
        total,
        summary.files,
        summary.added,
        summary.updated,
        summary.unchanged,
        summary.pruned,
        summary.errors,
        overview,
        round((time.monotonic() - started) * 1000),
    )
    # 10. Respond 200 with the sync-style counts.
    return UploadOut(
        source=name,
        files=summary.files,
        added=summary.added,
        updated=summary.updated,
        unchanged=summary.unchanged,
        pruned=summary.pruned,
        errors=summary.errors,
        chunks=summary.chunks,
        overview=overview,
    )


@router.delete("/{source_id}", status_code=204)
def delete_git_source(
    source_id: uuid.UUID,
    db: Session = Depends(get_db),  # noqa: B008
) -> Response:
    """Remove a stored row; 404 when the id is unknown.

    Removing does not touch the clones or the index — the next Sync
    (``prune=True``) prunes the dropped repo (phase scope boundary).
    """
    row = db.get(GitSource, source_id)
    if row is None:
        raise HTTPException(status_code=404, detail="git source not found")
    db.delete(row)
    db.commit()
    return Response(status_code=204)