LedgerLine/services/gateway/app.py

import asyncio
import uuid
from pathlib import Path

import httpx
from fastapi import FastAPI, Request
from fastapi.responses import FileResponse, JSONResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel, field_validator
from opentelemetry import trace
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.instrumentation.utils import suppress_instrumentation

from common.chaos_flags import ALL_FLAGS, get_flags, set_flag, set_profile
from common.logging_config import configure_logging
from common.settings import settings
from common.telemetry import setup_telemetry

from . import orchestrator
from .orchestrator import UpstreamFailure

SERVICE_NAME = "gateway"
logger = configure_logging(SERVICE_NAME)
setup_telemetry(service_name=SERVICE_NAME)
app = FastAPI(title=SERVICE_NAME)

FastAPIInstrumentor.instrument_app(
    app,
    excluded_urls="/healthz,/api/health/",
    exclude_spans=["receive", "send"],
)

STATIC_DIR = Path(__file__).parent / "static"
app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")

# A shared, connection-pooling client for the simple read-only proxy routes
# below -- opening a fresh httpx.AsyncClient (and TCP connection) per request
# is real, needless overhead under concurrent load.
_proxy_client = httpx.AsyncClient(
    timeout=settings.downstream_timeout_seconds,
    limits=httpx.Limits(max_connections=100, max_keepalive_connections=50),
)

DOWNSTREAM_URLS = {
    "ledger": settings.ledger_url,
    "fraud": settings.fraud_url,
    "sanctions": settings.sanctions_url,
    "insights": settings.insights_url,
    "notifier": settings.notifier_url,
}


class TransactionRequest(BaseModel):
    account_id: str
    kind: str
    amount_minor: int
    currency: str
    merchant_name: str
    category_id: int | None = None

    @field_validator("kind")
    @classmethod
    def kind_must_be_known(cls, v):
        if v not in ("deposit", "withdrawal", "purchase"):
            raise ValueError("kind must be one of deposit, withdrawal, purchase")
        return v

    @field_validator("amount_minor")
    @classmethod
    def deposit_amount_must_be_non_negative(cls, v, info):
        if info.data.get("kind") == "deposit" and v < 0:
            raise ValueError("amount_minor must be non-negative for a deposit")
        return v


class ChaosFlagRequest(BaseModel):
    flag: str
    enabled: bool


class ChaosProfileRequest(BaseModel):
    profile: str


@app.get("/", include_in_schema=False)
async def index():
    """Serves the frontend (phase-0-base-project-spec.md §10). This overrides
    §4's generic `GET / -> {"service", "version"}` info route, which the
    other five HTTP services still expose as-is."""
    return FileResponse(STATIC_DIR / "index.html")


@app.get("/healthz")
async def healthz():
    return {"status": "ok", "service": SERVICE_NAME}


@app.get("/readyz")
async def readyz():
    return {"status": "ok"}


@app.get("/api/accounts")
async def list_accounts():
    resp = await _proxy_client.get(f"{settings.ledger_url}/internal/accounts")
    return JSONResponse(status_code=resp.status_code, content=resp.json())


@app.get("/api/accounts/{account_id}/insights")
async def account_insights(account_id: str):
    resp = await _proxy_client.get(f"{settings.insights_url}/internal/accounts/{account_id}/summary")
    return JSONResponse(status_code=resp.status_code, content=resp.json())


@app.post("/api/transactions", status_code=201)
async def submit_transaction(body: TransactionRequest):
    request_id = _request_id()
    try:
        result = await orchestrator.create_transaction(body.model_dump())
    except UpstreamFailure as exc:
        log = logger.error if exc.status_code >= 500 else logger.warning
        log(
            "transaction_failed",
            account_id=body.account_id,
            status_code=exc.status_code,
            reason=exc.message,
            cause=repr(exc.__cause__) if exc.__cause__ else None,
        )
        return JSONResponse(
            status_code=exc.status_code, content={"error": exc.message, "request_id": request_id}
        )
    return JSONResponse(status_code=201, content={**result, "request_id": request_id})


def _request_id() -> str:
    """The response's `request_id` is the trace ID: paste it into Tempo to get
    the whole request. 32 hex digits, which is still a valid UUID. Falls back to
    a random one if no span is recording (telemetry not set up)."""
    ctx = trace.get_current_span().get_span_context()
    return format(ctx.trace_id, "032x") if ctx.is_valid else uuid.uuid4().hex


@app.get("/api/health/all")
async def health_all():
    async def check(name, url):
        try:
            resp = await _proxy_client.get(f"{url}/healthz", timeout=2.0)
            return name, "ok" if resp.status_code == 200 else "down"
        except httpx.RequestError:
            return name, "down"

    # The route itself is in excluded_urls; without this its five downstream
    # /healthz calls would each still be a parentless root trace.
    with suppress_instrumentation():
        results = await asyncio.gather(*[check(name, url) for name, url in DOWNSTREAM_URLS.items()])
    services = {"gateway": "ok", **dict(results)}
    return {"services": services}


@app.get("/api/admin/chaos")
async def get_chaos():
    return {"flags": await get_flags()}


@app.post("/api/admin/chaos")
async def post_chaos(body: ChaosFlagRequest):
    if body.flag not in ALL_FLAGS:
        return JSONResponse(status_code=400, content={"error": f"unknown flag {body.flag!r}"})
    await set_flag(body.flag, body.enabled)
    return {"flags": await get_flags()}


@app.post("/api/admin/chaos/profile")
async def post_chaos_profile(body: ChaosProfileRequest):
    try:
        flags = await set_profile(body.profile)
    except ValueError as exc:
        return JSONResponse(status_code=400, content={"error": str(exc)})
    return {"flags": flags}