reuse-surface/reuse_surface/hub/app.py

from __future__ import annotations

import asyncio
import json
import os
from pathlib import Path
from typing import Any

import yaml
from fastapi import Depends, FastAPI, Header, HTTPException, Query, Request
from fastapi.responses import JSONResponse, Response

from reuse_surface.hub.compose import compose_from_store, DEFAULT_DOMAIN
from reuse_surface.hub.store import HubStore
from reuse_surface.hub.webhooks import (
    SIGNATURE_HEADERS,
    push_touches_registry_index,
    verify_signature,
)

HUB_VERSION = "0.1.0"

# Registration fields that change what the federated index composes to. A
# PATCH touching any of them invalidates the composed index; a PATCH that only
# edits, say, the description does not.
COMPOSITION_FIELDS = frozenset({"enabled", "url", "index", "required", "domain"})


def _db_path() -> Path:
    return Path(os.environ.get("REUSE_SURFACE_DB", "/data/reuse.db"))


def _cache_dir() -> Path:
    return Path(os.environ.get("REUSE_SURFACE_CACHE_DIR", "/data/cache"))


def _write_token() -> str:
    return os.environ.get("REUSE_SURFACE_TOKEN", "")


def _store() -> HubStore:
    return HubStore(_db_path())


def _webhook_secret() -> str:
    return os.environ.get("REUSE_SURFACE_FORGEJO_WEBHOOK_SECRET", "")


def _http_error(status: int, error: str, message: str) -> HTTPException:
    return HTTPException(
        status_code=status,
        detail={"error": error, "message": message, "details": []},
    )


def _require_auth(authorization: str | None = Header(default=None)) -> None:
    write_token = _write_token()
    if not write_token:
        raise _http_error(
            503, "misconfigured", "REUSE_SURFACE_TOKEN is not configured"
        )
    if not authorization or not authorization.startswith("Bearer "):
        raise _http_error(401, "unauthorized", "Bearer token required")
    token = authorization.removeprefix("Bearer ").strip()
    if token != write_token:
        raise _http_error(401, "unauthorized", "Invalid token")


def create_app() -> FastAPI:
    app = FastAPI(title="reuse-surface federation hub", version=HUB_VERSION)
    store = _store()
    # Serializes concurrent recompose triggers (manual POST, webhook, future
    # scheduled fallback) so a burst of pushes coalesces into one compose
    # pass instead of overlapping ones (T02 design principle 2 debounce).
    compose_lock = asyncio.Lock()

    @app.get("/health")
    def health() -> dict[str, str]:
        return {"status": "ok", "service": "reuse-surface", "version": HUB_VERSION}

    @app.get("/v1/repos")
    def list_repos() -> dict[str, Any]:
        repos = store.list_repos()
        return {"count": len(repos), "repos": repos}

    @app.post("/v1/repos", status_code=201, dependencies=[Depends(_require_auth)])
    async def register_repo(request: Request) -> dict[str, Any]:
        payload = await request.json()
        try:
            registration = store.create_repo(payload)
        except FileExistsError as exc:
            raise _http_error(409, "conflict", str(exc)) from exc
        except ValueError as exc:
            raise _http_error(400, "validation_error", str(exc)) from exc
        store.mark_stale()
        return registration

    @app.get("/v1/repos/{repo}")
    def get_repo(repo: str) -> dict[str, Any]:
        registration = store.get_repo(repo)
        if registration is None:
            raise _http_error(404, "not_found", f"repo not found: {repo}")
        return registration

    @app.patch("/v1/repos/{repo}", dependencies=[Depends(_require_auth)])
    async def update_repo(repo: str, request: Request) -> dict[str, Any]:
        payload = await request.json()
        try:
            registration = store.update_repo(repo, payload)
        except KeyError as exc:
            raise _http_error(404, "not_found", str(exc)) from exc
        except ValueError as exc:
            raise _http_error(400, "validation_error", str(exc)) from exc
        if COMPOSITION_FIELDS & set(payload):
            store.mark_stale()
        return registration

    @app.delete("/v1/repos/{repo}", status_code=204, dependencies=[Depends(_require_auth)])
    def delete_repo(repo: str) -> Response:
        if not store.delete_repo(repo):
            raise _http_error(404, "not_found", f"repo not found: {repo}")
        store.mark_stale()
        return Response(status_code=204)

    async def _federated_response(
        refresh: bool,
        accept: str | None,
        format_param: str,
    ) -> Response:
        # composed_at/stale track recomposes that actually refetched, not
        # every plain GET -- a plain GET otherwise serves current best-effort
        # data (compose_from_store's own per-source cache_ttl_seconds still
        # applies) and must not clear a staleness signal nothing acted on.
        #
        # A GET *does* refresh when the index is marked stale. Registration
        # writes set that flag, and without this a newly registered or
        # re-enabled repo stays invisible for as long as its cached index
        # survives -- silently, since the response would keep reporting
        # stale: false. Clearing the flag here is honest because this pass
        # really did refetch.
        async with compose_lock:
            effective_refresh = refresh or store.get_compose_state()["stale"]
            try:
                federated, warnings = compose_from_store(
                    store, refresh=effective_refresh, cache_dir=_cache_dir(), domain=DEFAULT_DOMAIN
                )
            except FileNotFoundError as exc:
                raise _http_error(502, "compose_error", str(exc)) from exc
            if effective_refresh:
                store.record_compose()
            compose_state = store.get_compose_state()

        federated["composed_at"] = compose_state["composed_at"]
        federated["stale"] = compose_state["stale"]

        use_yaml = format_param == "yaml" or (accept and "yaml" in accept.lower())
        headers: dict[str, str] = {}
        if compose_state["composed_at"]:
            headers["X-Composed-At"] = compose_state["composed_at"]
        if warnings:
            headers["X-Federation-Warnings"] = "; ".join(warnings)
        if use_yaml:
            body = yaml.safe_dump(federated, sort_keys=False)
            return Response(content=body, media_type="application/yaml", headers=headers)
        return JSONResponse(content=federated, headers=headers)

    @app.get("/v1/federated", response_model=None)
    async def get_federated(
        request: Request,
        refresh: bool = Query(default=False),
        format: str = Query(default="json"),
    ) -> Response:
        return await _federated_response(refresh, request.headers.get("accept"), format)

    @app.post("/v1/federated/compose", response_model=None, dependencies=[Depends(_require_auth)])
    async def compose_federated(
        request: Request,
        format: str = Query(default="json"),
    ) -> Response:
        return await _federated_response(True, request.headers.get("accept"), format)

    @app.post("/v1/webhooks/forgejo", response_model=None)
    async def forgejo_webhook(request: Request) -> Response:
        body = await request.body()
        signature = None
        for header_name in SIGNATURE_HEADERS:
            signature = request.headers.get(header_name)
            if signature:
                break
        secret = _webhook_secret()
        if not secret:
            raise _http_error(
                503, "misconfigured", "REUSE_SURFACE_FORGEJO_WEBHOOK_SECRET is not configured"
            )
        if not verify_signature(secret, body, signature):
            raise _http_error(401, "unauthorized", "invalid or missing webhook signature")

        try:
            payload = json.loads(body)
        except json.JSONDecodeError as exc:
            raise _http_error(400, "validation_error", f"invalid JSON payload: {exc}") from exc

        if not push_touches_registry_index(payload):
            return JSONResponse(content={"accepted": False, "reason": "no registry/indexes/ change"})

        async with compose_lock:
            store.mark_stale()
            try:
                compose_from_store(
                    store, refresh=True, cache_dir=_cache_dir(), domain=DEFAULT_DOMAIN
                )
            except FileNotFoundError as exc:
                # Recompose failed -- leave stale=1 so the next GET/scheduled
                # trigger reports it honestly rather than silently swallowing
                # the failure.
                raise _http_error(502, "compose_error", str(exc)) from exc
            composed_at = store.record_compose()

        return JSONResponse(content={"accepted": True, "composed_at": composed_at})

    @app.post("/v1/reuse-events", status_code=201, dependencies=[Depends(_require_auth)])
    async def record_reuse_event(request: Request) -> dict[str, Any]:
        payload = await request.json()
        try:
            return store.record_reuse_event(payload)
        except ValueError as exc:
            raise _http_error(400, "validation_error", str(exc)) from exc

    @app.get("/v1/reuse-events")
    def list_reuse_events(
        capability_id: str | None = Query(default=None),
    ) -> dict[str, Any]:
        events = store.list_reuse_events(capability_id)
        return {"count": len(events), "events": events}

    return app


def main() -> None:
    import uvicorn

    host = os.environ.get("REUSE_SURFACE_LISTEN_HOST", "0.0.0.0")
    port = int(os.environ.get("REUSE_SURFACE_LISTEN_PORT", "8000"))
    uvicorn.run(create_app(), host=host, port=port, reload=False)