rfc-app/backend/app/docs_specs.py

"""v0.20.0 — on-site framework-specs surface source.

Sibling of `docs_sessions.py` (v0.19.0 / roadmap item #30): the
framework mediates a gitea fetch on behalf of the browser so the
rendered `/docs/specs/*` surface inherits the same chrome as
`/docs/user-guide` and `/docs/sessions/*` and stays free of any
cross-origin gestures from the frontend.

Two read endpoints, both anonymous-reachable:

  GET /api/docs/specs/manifest   — the configured spec list
  GET /api/docs/specs/<name>     — a single spec body (markdown)

The framework-default manifest is OHM-flavored (rfc-app's own SPEC.md
+ flotilla's SPEC.md on `git.wiggleverse.org`) for the same reason
`docs_sessions.py`'s defaults are: OHM is the only deployment to
date. A deployment running its own spec set overrides the manifest
via the `OHM_DOCS_SPECS` env var (set through flotilla's overlay).
History is intentionally not surfaced here — the operator-stated
intent is "current version only; git is the history surface".

Per-spec entries carry three fields:

  name   — URL-safe slug (`[a-z0-9-]+`) — the path segment
  title  — human-readable label shown in the nav and the page header
  url    — the upstream raw URL the framework fetches

Validation:

  - The configured list must be a JSON array of `{name, title, url}`
    objects. A malformed `OHM_DOCS_SPECS` value (bad JSON, wrong
    shape, invalid slug) logs a warning and falls back to the default
    so a typo in the overlay doesn't crash startup.
  - Each `name` is checked against `^[a-z0-9-]+$` before the manifest
    is accepted. The route layer also validates the path-bound `name`
    parameter before any network call, so a malformed URL never
    reaches the cache or the upstream.

Cache shape mirrors `docs_sessions.py`: in-process `dict` + monotonic
TTL check, negative results (404) cached, no external dep. The
manifest is cheap (parsed from an env var, no network), so it has no
TTL — every request re-derives it. Per-spec content has a 5-minute
default TTL (env-tunable via `OHM_DOCS_SPECS_CONTENT_TTL_SEC`).

§3 invariant 1 is preserved: the framework holds no secret bytes;
the upstream specs are public-repo raw URLs, the fetch carries no
auth header.
"""
from __future__ import annotations

import json
import logging
import os
import re
import threading
import time
from typing import Any

import httpx

log = logging.getLogger(__name__)


# The framework-default spec set. OHM-flavored per the same precedent
# `docs_sessions.py` set: the only live deployment is OHM, so the
# default points there. A deployment running its own specs overrides
# `OHM_DOCS_SPECS` via the overlay.
_DEFAULT_SPECS: list[dict[str, str]] = [
    {
        "name": "rfc-app",
        "title": "rfc-app SPEC",
        "url": (
            "https://git.wiggleverse.org/ben.stull/rfc-app/"
            "raw/branch/main/SPEC.md"
        ),
    },
    {
        "name": "flotilla",
        "title": "flotilla SPEC",
        "url": (
            "https://git.wiggleverse.org/wiggleverse/ohm-rfc-app-flotilla/"
            "raw/branch/main/SPEC.md"
        ),
    },
]

_DEFAULT_CONTENT_TTL_SEC = 300.0

# URL-safe slug. Matches `docs_sessions.py`'s `_SESSION_DIR_RE` spirit
# (rejecting anything that could resolve outside the intended layout)
# but with the lowercase-alphanumeric-plus-dash shape the manifest
# enforces. Path traversal (`..`), separators (`/`), tilde, uppercase,
# and whitespace all fail this regex; the route layer rejects 400
# before any cache or network call.
_NAME_RE = re.compile(r"^[a-z0-9-]+$")

_HTTP_TIMEOUT_SEC = 5.0


def _env_float(name: str, default: float) -> float:
    raw = os.environ.get(name, "").strip()
    if not raw:
        return default
    try:
        return float(raw)
    except ValueError:
        log.warning("invalid %s=%r — falling back to %s", name, raw, default)
        return default


def _content_ttl() -> float:
    return _env_float("OHM_DOCS_SPECS_CONTENT_TTL_SEC", _DEFAULT_CONTENT_TTL_SEC)


def _is_valid_name(name: str) -> bool:
    """Slug guard for path-bound `name` parameters.

    Mirrors `docs_sessions._is_valid_session_dir`'s contract: the
    route layer calls this before any network or cache work, so a
    malformed name never escapes the FastAPI surface.
    """
    return bool(isinstance(name, str) and _NAME_RE.match(name))


def _parse_spec_entry(entry: Any) -> dict[str, str] | None:
    """Validate a single manifest entry; return None if invalid.

    Required fields: `name`, `title`, `url`. All three must be
    non-empty strings; `name` must match `_NAME_RE`. The validator is
    strict: an entry that fails any check is dropped from the manifest
    (and the caller logs at warning level).
    """
    if not isinstance(entry, dict):
        return None
    name = entry.get("name")
    title = entry.get("title")
    url = entry.get("url")
    if not isinstance(name, str) or not _is_valid_name(name):
        return None
    if not isinstance(title, str) or not title.strip():
        return None
    if not isinstance(url, str) or not url.strip():
        return None
    return {"name": name, "title": title.strip(), "url": url.strip()}


def _load_configured_specs() -> list[dict[str, str]]:
    """Parse `OHM_DOCS_SPECS` (if set) or return the default list.

    Malformed JSON or wrong-shape values log a warning and fall back
    to the default — the deployment continues to render the spec
    surface rather than crashing startup. The strict validation (each
    entry's name slug, presence of all three fields) drops bad entries
    one-by-one; if every entry is dropped, the default applies.
    """
    raw = os.environ.get("OHM_DOCS_SPECS", "").strip()
    if not raw:
        return list(_DEFAULT_SPECS)
    try:
        parsed = json.loads(raw)
    except (json.JSONDecodeError, ValueError) as e:
        log.warning(
            "OHM_DOCS_SPECS is not valid JSON (%s) — falling back to default", e
        )
        return list(_DEFAULT_SPECS)
    if not isinstance(parsed, list):
        log.warning(
            "OHM_DOCS_SPECS must be a JSON array — falling back to default"
        )
        return list(_DEFAULT_SPECS)
    out: list[dict[str, str]] = []
    seen: set[str] = set()
    for entry in parsed:
        validated = _parse_spec_entry(entry)
        if validated is None:
            log.warning(
                "OHM_DOCS_SPECS entry %r failed validation — dropped", entry
            )
            continue
        if validated["name"] in seen:
            log.warning(
                "OHM_DOCS_SPECS has duplicate name %r — dropped", validated["name"]
            )
            continue
        seen.add(validated["name"])
        out.append(validated)
    if not out:
        log.warning(
            "OHM_DOCS_SPECS yielded no valid entries — falling back to default"
        )
        return list(_DEFAULT_SPECS)
    return out


# ---------------------------------------------------------------------------
# In-process TTL cache
# ---------------------------------------------------------------------------
#
# Same shape as `docs_sessions.py`: plain dict + `time.monotonic()` check,
# no external dep. The cache value is a `(stored_at, payload)` tuple;
# `payload` may carry an error-shape sentinel for negative caching (404s).
# Lock guards read-modify-write across worker tasks; entries are immutable
# once stored so reads under the lock are fast.

_lock = threading.Lock()
_cache: dict[str, tuple[float, dict[str, Any]]] = {}


def _cache_get(key: str, ttl_sec: float) -> dict[str, Any] | None:
    with _lock:
        entry = _cache.get(key)
        if entry is None:
            return None
        stored_at, payload = entry
        if time.monotonic() - stored_at > ttl_sec:
            return None
        return payload


def _cache_put(key: str, payload: dict[str, Any]) -> None:
    with _lock:
        _cache[key] = (time.monotonic(), payload)


def reset_cache() -> None:
    """Drop every cached entry. Test seam — not called in production."""
    with _lock:
        _cache.clear()


# ---------------------------------------------------------------------------
# Public fetch surface
# ---------------------------------------------------------------------------
#
# Each fetcher returns a `{status, ...}` dict, same convention as
# `docs_sessions.py`:
#   "ok"     — payload field carries the body / manifest
#   "404"    — gitea returned 404 (or the configured name doesn't exist)
#   "error"  — gitea returned 5xx, timed out, or returned malformed data
#
# The route layer maps these onto HTTP responses; keeping the mapping
# out of this module makes the cache transparent to the test harness.


async def _http_get(url: str) -> tuple[int, str]:
    """Perform a single GET against `url`; return (status_code, body).

    On timeout or network error, returns (599, error_message). The 599
    pseudo-status maps to a 502 at the route layer the same way an
    upstream 5xx does — the caller doesn't care which leg of the
    network broke.
    """
    try:
        async with httpx.AsyncClient(timeout=_HTTP_TIMEOUT_SEC) as client:
            r = await client.get(url)
            return r.status_code, r.text
    except httpx.HTTPError as e:
        log.warning("specs fetch failed for %s: %s", url, e)
        return 599, f"fetch error: {e}"


def fetch_specs_manifest() -> dict[str, Any]:
    """Return the configured spec manifest.

    The manifest is derived from the `OHM_DOCS_SPECS` env var (or the
    framework default if unset / malformed) and carries no network
    work — it's safe to call on every request. The return shape mirrors
    the docs_sessions manifest endpoint for frontend consistency:

      {"status": "ok", "specs": [{"name", "title", "url"}, ...]}

    The "url" field is exposed in the manifest so the frontend can
    offer a "view source on gitea" affordance alongside each rendered
    spec (operator-stated intent: "include the history so you can see
    it in git" — that gesture lives in the source link, not on the
    rendered page).
    """
    specs = _load_configured_specs()
    return {"status": "ok", "specs": specs}


async def fetch_spec(name: str) -> dict[str, Any]:
    """Fetch a single spec body by its manifest `name`.

    The caller is expected to have validated `name` against
    `_is_valid_name` before calling this — an invalid name shouldn't
    reach the network. We re-check inside as defense-in-depth: a
    bogus name here returns the same `{status: "404"}` shape so the
    route layer's `404 → HTTP 404` mapping handles it uniformly.

    Returns one of:
      {"status": "ok", "body": "..."}
      {"status": "404"}                      — no such spec OR upstream 404
      {"status": "error", "detail": "..."}   — upstream 5xx / timeout
    """
    if not _is_valid_name(name):
        return {"status": "404"}

    cache_key = f"spec:{name}"
    cached = _cache_get(cache_key, _content_ttl())
    if cached is not None:
        return cached

    specs = _load_configured_specs()
    match = next((s for s in specs if s["name"] == name), None)
    if match is None:
        # Cache the negative — a deployment with an unstable manifest
        # would still benefit from the TTL window, and the cached 404
        # is automatically displaced when the next request happens
        # after TTL expiry.
        payload: dict[str, Any] = {"status": "404"}
        _cache_put(cache_key, payload)
        return payload

    url = match["url"]
    status, body = await _http_get(url)

    if status == 200:
        payload = {"status": "ok", "body": body}
        _cache_put(cache_key, payload)
        return payload
    if status == 404:
        payload = {"status": "404"}
        _cache_put(cache_key, payload)
        return payload
    return {"status": "error", "detail": f"upstream returned {status}"}