import asyncio import json import time from fastapi import FastAPI, Request from fastapi.responses import JSONResponse from common.chaos_flags import flags_fingerprint from common.logging_config import configure_logging from common.redis_client import redis_client from common.settings import settings import sqlalchemy as sa from . import chaos from .analytics import account_exists, compute_summary from .db import NormalSession, engine, session_factory SERVICE_NAME = "insights" logger = configure_logging(SERVICE_NAME) app = FastAPI(title=SERVICE_NAME) # 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("received chatty-health healthz probe from gateway") 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: logger.exception(f"readyz check failed: {exc}") 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