import time import uuid import httpx import structlog from opentelemetry import propagate, trace from opentelemetry.trace import SpanKind from common.http import downstream_client from common.redis_client import redis_client from common.settings import settings from . import chaos logger = structlog.get_logger("gateway") tracer = trace.get_tracer(__name__) REJECT_THRESHOLD = 0.8 class UpstreamFailure(Exception): def __init__(self, status_code: int, message: str): self.status_code = status_code self.message = message async def _post(client: httpx.AsyncClient, url: str, json: dict) -> httpx.Response: try: return await client.post(url, json=json) except httpx.TimeoutException as exc: raise UpstreamFailure(504, f"{url} timed out") from exc except httpx.RequestError as exc: raise UpstreamFailure(502, f"{url} unreachable") from exc async def _score_once(client: httpx.AsyncClient, payload: dict) -> dict: resp = await _post(client, f"{settings.fraud_url}/internal/score", payload) if resp.status_code >= 300: status = resp.status_code if resp.status_code in (502, 504) else 502 raise UpstreamFailure(status, "fraud scoring failed") return resp.json() async def _score_with_retry(client: httpx.AsyncClient, payload: dict, retry_storm: bool) -> dict: """retry_storm: on any non-2xx from fraud, retry 3 times immediately -- no backoff, no jitter, no circuit breaker (phase-0-base-project-spec.md ยง7). "No circuit breaker" is taken literally here: once triggered, it fires all 3 retries unconditionally rather than stopping early on the first success -- it never checks whether it still needs to. The last attempt's outcome wins.""" attempts = 4 if retry_storm else 1 result = None last_exc = None for _ in range(attempts): try: result = await _score_once(client, payload) last_exc = None except UpstreamFailure as exc: last_exc = exc if last_exc is not None: raise last_exc return result async def _run_chatty_health_probe(client: httpx.AsyncClient) -> None: """chatty_health: GET /healthz on all five downstreams, sequentially, before doing any real work.""" headers = {"X-Ledgerline-Probe": "chatty-health"} for name in chaos.DOWNSTREAMS: url = getattr(settings, f"{name}_url") try: await client.get(f"{url}/healthz", headers=headers, timeout=2.0) except httpx.RequestError: pass async def create_transaction(payload: dict) -> dict: start = time.perf_counter() logger.info( "transaction_started", account_id=payload["account_id"], kind=payload["kind"], amount_minor=payload["amount_minor"], ) retry_storm = await chaos.retry_storm_enabled() no_batching = await chaos.no_batching_enabled() chatty_health = await chaos.chatty_health_enabled() async with downstream_client() as client: if chatty_health: await _run_chatty_health_probe(client) create_resp = await _post( client, f"{settings.ledger_url}/internal/transactions", payload, ) if create_resp.status_code == 404: raise UpstreamFailure(404, "account not found") if create_resp.status_code >= 500: raise UpstreamFailure(create_resp.status_code, "ledger failed unexpectedly") if create_resp.status_code >= 300: raise UpstreamFailure(502, "ledger rejected the transaction") created = create_resp.json() tx_id = created["transaction_id"] score_payload = { "account_id": created["account_id"], "holder_name": created["holder_name"], "category_id": created["category_id"], } if no_batching: item_count = chaos.synthetic_line_item_count() scores = [] for _ in range(item_count): item_payload = {**score_payload, "amount_minor": max(1, created["amount_minor"] // item_count)} scores.append(await _score_with_retry(client, item_payload, retry_storm)) risk_score = max(s["risk_score"] for s in scores) else: score_payload["amount_minor"] = created["amount_minor"] scored = await _score_with_retry(client, score_payload, retry_storm) risk_score = scored["risk_score"] decision = "rejected" if risk_score > REJECT_THRESHOLD else "settled" action = "reject" if decision == "rejected" else "settle" settle_resp = await _post( client, f"{settings.ledger_url}/internal/transactions/{tx_id}/{action}", {"risk_score": risk_score}, ) if settle_resp.status_code >= 300: raise UpstreamFailure(502, "ledger failed to finalise the transaction") settled = settle_resp.json() try: fields = { "transaction_id": tx_id, "account_id": created["account_id"], "kind": created["kind"], "amount_minor": str(created["amount_minor"]), "status": settled["status"], "risk_score": str(risk_score), } with tracer.start_as_current_span( f"send {settings.stream_name}", kind=SpanKind.PRODUCER, attributes={ "messaging.system": "redis", "messaging.destination.name": settings.stream_name, "messaging.operation.type": "send", }, ) as span: # Redis Streams carry no trace context, so it travels in the # message itself. Injected inside the PRODUCER span, so the # consumers' parent is this span. propagate.inject(fields) msg_id = await redis_client.xadd(settings.stream_name, fields) span.set_attribute("messaging.message.id", msg_id) except Exception: logger.exception("stream_publish_failed", transaction_id=tx_id) summary = {} try: summary_resp = await client.get( f"{settings.insights_url}/internal/accounts/{created['account_id']}/quick_summary" ) if summary_resp.status_code == 200: summary = summary_resp.json() except httpx.RequestError: logger.exception("insights_summary_unavailable", transaction_id=tx_id) elapsed_ms = (time.perf_counter() - start) * 1000 logger.info( "transaction_finished", transaction_id=tx_id, status=settled["status"], risk_score=risk_score, duration_ms=round(elapsed_ms, 1), ) return { "transaction_id": tx_id, "status": settled["status"], "risk_score": risk_score, "balance_minor": settled["balance_minor"], "summary": summary, "elapsed_ms": elapsed_ms, }