LedgerLine/services/insights/app.py

import asyncio
import json
import time

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from opentelemetry import trace
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor
from opentelemetry.trace import StatusCode

from common.chaos_flags import flags_fingerprint
from common.logging_config import configure_logging
from common.metrics import observe_db_pools
from common.redis_client import redis_client
from common.settings import settings
from common.telemetry import setup_telemetry

import sqlalchemy as sa

from . import chaos
from .analytics import account_exists, compute_summary
from .db import NormalSession, engine, normal_engine, session_factory, starved_engine

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

FastAPIInstrumentor.instrument_app(
    app,
    excluded_urls="/healthz",
    exclude_spans=["receive", "send"],
)
SQLAlchemyInstrumentor().instrument(
    engines=[normal_engine.sync_engine, starved_engine.sync_engine]
)
observe_db_pools({"normal": normal_engine, "starved": starved_engine})

# Cache-stampede protection for the Redis-cached branch below: a burst of
# concurrent requests for the same not-yet-cached (account, flags) key would
# otherwise all pay the full computation cost at once instead of one paying
# it and the rest waiting on the result. Not needed for unbounded_cache,
# which never caches by design.
_inflight_locks: dict[str, asyncio.Lock] = {}


def _lock_for(key: str) -> asyncio.Lock:
    lock = _inflight_locks.get(key)
    if lock is None:
        lock = asyncio.Lock()
        _inflight_locks[key] = lock
    return lock


@app.get("/")
async def root():
    return {"service": SERVICE_NAME, "version": "0.1.0"}


@app.get("/healthz")
async def healthz(request: Request):
    if request.headers.get("X-Ledgerline-Probe") == "chatty-health":
        logger.info("healthz_probe_received", probe="chatty-health")
    return {"status": "ok", "service": SERVICE_NAME}


@app.get("/readyz")
async def readyz():
    try:
        async with engine.connect() as conn:
            await conn.exec_driver_sql("SELECT 1")
    except Exception as exc:
        span = trace.get_current_span()
        span.record_exception(exc)
        span.set_status(StatusCode.ERROR, str(exc))
        logger.exception("readyz_check_failed")
        return JSONResponse(status_code=503, content={"status": "unavailable"})
    return {"status": "ok"}


@app.get("/internal/accounts/{account_id}/quick_summary")
async def quick_summary(account_id: str):
    """A cheap, account-primary-key-only lookup embedded in every transaction
    response (phase-0-base-project-spec.md §6.1's "summary" field) -- deliberately
    separate from the 120-day aggregation below (§6.2), which is the one where
    n_plus_one and the missing index are meant to bite. That aggregation scans
    `transactions` by account_id with no supporting index on purpose; embedding
    it in every single transaction's response would make the missing-index
    pathology fire on every request instead of only on the analytics endpoint."""
    async with NormalSession() as session:
        result = await session.execute(
            sa.text("SELECT currency, balance_minor FROM accounts WHERE id = :id"),
            {"id": account_id},
        )
        row = result.first()
    if row is None:
        return JSONResponse(status_code=404, content={"error": "account not found"})
    return {"account_id": account_id, "currency": row.currency, "balance_minor": row.balance_minor}


@app.get("/internal/accounts/{account_id}/summary")
async def summary(account_id: str):
    n_plus_one = await chaos.n_plus_one_enabled()
    cold_cache = await chaos.cold_cache_enabled()
    unbounded_cache = await chaos.unbounded_cache_enabled()
    fingerprint = await flags_fingerprint()

    # unbounded_cache means every single request is a guaranteed fresh miss
    # needing a real connection -- under concurrent load that alone is enough
    # demand to exhaust a small pool, without needing n_plus_one too (see
    # services/insights/db.py).
    starved = unbounded_cache

    async with session_factory(starved) as session:
        if not await account_exists(session, account_id):
            return JSONResponse(status_code=404, content={"error": "account not found"})

        if unbounded_cache:
            # Broken on purpose: keyed by a timestamp that's unique per call,
            # so this never hits -- it only ever grows. See chaos.py.
            key = (account_id, fingerprint, time.time_ns())
            result = await compute_summary(session, account_id, n_plus_one=n_plus_one)
            chaos.UNBOUNDED_CACHE[key] = result
            return result

        ttl = settings.summary_cache_ttl_seconds_cold if cold_cache else settings.summary_cache_ttl_seconds
        cache_key = f"summary:{account_id}:{fingerprint}"
        cached = await redis_client.get(cache_key)
        if cached is not None:
            return json.loads(cached)

    async with _lock_for(cache_key):
        cached = await redis_client.get(cache_key)
        if cached is not None:
            return json.loads(cached)
        async with session_factory(starved) as session:
            result = await compute_summary(session, account_id, n_plus_one=n_plus_one)
        await redis_client.setex(cache_key, ttl, json.dumps(result))
        return result