from fastapi import FastAPI, Request from fastapi.responses import JSONResponse from pydantic import BaseModel, field_validator from opentelemetry import trace from opentelemetry.trace import StatusCode from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor from common.logging_config import configure_logging from common.telemetry import setup_telemetry from . import chaos from .db import normal_engine, session_factory, starved_engine from . import repository SERVICE_NAME = "ledger" logger = configure_logging(SERVICE_NAME) tracer = trace.get_tracer(__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] ) class CreateTransactionRequest(BaseModel): account_id: str kind: str amount_minor: int currency: str merchant_name: str category_id: int | None = None @field_validator("amount_minor") @classmethod def deposit_amount_must_be_non_negative(cls, v, info): kind = info.data.get("kind") if kind == "deposit" and v < 0: raise ValueError("amount_minor must be non-negative for a deposit") return v class SettleRequest(BaseModel): risk_score: float @app.exception_handler(Exception) async def unhandled_exception_handler(request: Request, exc: Exception): logger.exception(f"unhandled exception on {request.url.path}: {exc}") return JSONResponse(status_code=500, content={"error": "internal error"}) @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 normal_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(f"readyz check failed: {exc}") return JSONResponse(status_code=503, content={"status": "unavailable"}) return {"status": "ok"} @app.get("/internal/accounts") async def list_accounts(): async with session_factory(starved=False) as session: rows = await repository.list_accounts(session) return [dict(row) for row in rows] @app.post("/internal/transactions") async def create_transaction(body: CreateTransactionRequest): logger.info(f"processing tx for account {body.account_id}, amount {body.amount_minor}") starved = await chaos.pool_starvation_enabled() poison_enabled = await chaos.poison_data_enabled() poison_tx_id = await chaos.find_poison_transaction_id(body.account_id) async with session_factory(starved=starved) as session: account = await repository.get_account(session, body.account_id) if account is None: return JSONResponse(status_code=404, content={"error": "account not found"}) if poison_tx_id is not None and poison_enabled: poison_row = await repository.get_transaction_currency(session, poison_tx_id) # poison_data: skip input validation on the seeded malformed rows # (ยง8) -- this naive access raises on the NULL currency on purpose. _ = poison_row.currency.upper() tx_id = await repository.create_transaction( session, account_id=body.account_id, kind=body.kind, amount_minor=body.amount_minor, currency=body.currency, merchant_name=body.merchant_name, category_id=body.category_id, ) return JSONResponse( status_code=201, content={ "transaction_id": tx_id, "account_id": body.account_id, "holder_name": account.holder_name, "kind": body.kind, "amount_minor": body.amount_minor, "category_id": body.category_id, "status": "pending", }, ) @app.post("/internal/transactions/{tx_id}/settle") async def settle(tx_id: str, body: SettleRequest): starved = await chaos.pool_starvation_enabled() row_lock = await chaos.row_lock_contention_enabled() with tracer.start_as_current_span( "ledger.settle", attributes={"ledgerline.tx.id": tx_id, "ledgerline.risk.score": body.risk_score}, ) as span: async with session_factory(starved=starved) as session: tx = await repository.get_transaction(session, tx_id) if tx is None: return JSONResponse(status_code=404, content={"error": "transaction not found"}) span.set_attribute("ledgerline.account.id", str(tx.account_id)) span.set_attribute("ledgerline.tx.amount_minor", tx.amount_minor) account = await repository.get_account(session, tx.account_id) use_lock = row_lock and bool(account.is_hot) balance = await repository.settle_transaction(session, tx=tx, risk_score=body.risk_score, use_row_lock=use_lock) logger.info(f"settled transaction {tx_id}") return {"status": "settled", "balance_minor": balance} @app.post("/internal/transactions/{tx_id}/reject") async def reject(tx_id: str, body: SettleRequest): starved = await chaos.pool_starvation_enabled() async with session_factory(starved=starved) as session: tx = await repository.get_transaction(session, tx_id) if tx is None: return JSONResponse(status_code=404, content={"error": "transaction not found"}) balance = await repository.reject_transaction(session, tx=tx, risk_score=body.risk_score) logger.info(f"rejected transaction {tx_id}") return {"status": "rejected", "balance_minor": balance}