Api-Gateway/app/discovery.py

"""Keep service endpoint catalogs in sync with the upstream's OpenAPI document."""
import time

import httpx
from sqlalchemy.orm import Session

from app.models import Endpoint, Service

METHODS = {"GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"}

SPEC_PATHS = ("/openapi.json", "/swagger.json", "/api-docs")

# Syncing hits the upstream over HTTP, so results are cached per service and
# only refreshed after the TTL (or on demand with force=True).
SYNC_TTL_SECONDS = 300
_last_sync: dict[int, float] = {}      # service id -> monotonic time of last attempt
_spec_found: dict[int, bool] = {}      # service id -> did the last attempt find a spec
_spec_cache: dict[int, dict] = {}      # service id -> raw OpenAPI document of last fetch


def fetch_spec(base_url: str, verify_tls: bool = True) -> dict | None:
    for candidate in SPEC_PATHS:
        try:
            resp = httpx.get(base_url.rstrip("/") + candidate, timeout=2,
                             follow_redirects=True, verify=verify_tls)
            if resp.status_code == 200:
                data = resp.json()
                if isinstance(data, dict) and "paths" in data:
                    return data
        except (httpx.HTTPError, ValueError):
            continue
    return None


def spec_status(service_id: int) -> bool | None:
    """Did the last sync attempt find an OpenAPI document? None = never tried."""
    return _spec_found.get(service_id)


def sync_service(db: Session, service: Service, force: bool = False) -> tuple[bool, bool]:
    """Mirror the upstream OpenAPI paths into the endpoint catalog: add new
    operations, refresh descriptions, delete operations that disappeared
    (their key grants go with them). Services without a reachable spec are
    left untouched. Returns (spec_found, catalog_changed); results are cached
    between syncs."""
    now = time.monotonic()
    if not force and now - _last_sync.get(service.id, float("-inf")) < SYNC_TTL_SECONDS:
        return _spec_found.get(service.id, False), False
    _last_sync[service.id] = now

    spec = fetch_spec(service.base_url, service.verify_tls)
    _spec_found[service.id] = spec is not None
    if spec is None:
        return False, False
    _spec_cache[service.id] = spec
    return True, apply_spec(db, service, spec)


def get_spec(db: Session, service: Service) -> dict | None:
    """Raw upstream OpenAPI document, refreshed through the same TTL cache as
    the endpoint sync. Falls back to the last known document if the upstream
    is temporarily unreachable."""
    sync_service(db, service)
    return _spec_cache.get(service.id)


def validate_service(db: Session, service: Service) -> dict:
    """Connectivity probe for the admin UI. Prefers the OpenAPI document —
    finding one both proves reachability and refreshes the endpoint cache —
    and falls back to a plain GET on the base URL otherwise."""
    start = time.perf_counter()
    spec = fetch_spec(service.base_url, service.verify_tls)
    if spec is not None:
        _last_sync[service.id] = time.monotonic()
        _spec_found[service.id] = True
        _spec_cache[service.id] = spec
        apply_spec(db, service, spec)
        return {"ok": True, "spec_found": True,
                "endpoints": len(service.endpoints),
                "latency_ms": round((time.perf_counter() - start) * 1000, 1)}
    try:
        resp = httpx.get(service.base_url, timeout=5, follow_redirects=True,
                         verify=service.verify_tls)
        return {"ok": True, "spec_found": False, "status_code": resp.status_code,
                "endpoints": len(service.endpoints),
                "latency_ms": round((time.perf_counter() - start) * 1000, 1)}
    except httpx.HTTPError as exc:
        return {"ok": False, "spec_found": False,
                "error": str(exc) or type(exc).__name__,
                "latency_ms": round((time.perf_counter() - start) * 1000, 1)}


def apply_spec(db: Session, service: Service, spec: dict) -> bool:
    """Write an OpenAPI document's paths into the endpoint catalog.
    Returns whether anything changed."""
    wanted: dict[tuple[str, str], str] = {}
    for path, operations in spec["paths"].items():
        if not isinstance(operations, dict):
            continue
        for method, op in operations.items():
            method = method.upper()
            if method not in METHODS:
                continue
            summary = op.get("summary", "") if isinstance(op, dict) else ""
            wanted[(method, path)] = summary

    existing = {(e.method, e.path): e for e in service.endpoints}
    changed = False
    for (method, path), summary in wanted.items():
        endpoint = existing.get((method, path))
        if endpoint is None:
            db.add(Endpoint(service_id=service.id, method=method, path=path,
                            description=summary))
            changed = True
        elif endpoint.description != summary:
            endpoint.description = summary
            changed = True
    for key, endpoint in existing.items():
        if key not in wanted:
            db.delete(endpoint)
            changed = True
    db.commit()
    db.expire(service, ["endpoints"])
    return changed