LedgerLine/services/gateway/orchestrator.py

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,
    }