trex-core/trex_core/application/ledger_outbox.py

"""Ledger outbox + writer (REQ-019, dcy-004 FR-3/FR-4).

The execution fill path does NOT call portfolio-hub directly. Instead it
enqueues a durable `LedgerWorkItem` keyed by `correlation_id`
(`{asset_id}:{order_id}:{execution_id}`). A separate `LedgerWriter` drains
the outbox into the `LedgerPort` adapter and classifies each attempt as one
of:

  completed           -- adapter accepted (new ledger entry).
  duplicate_completed -- the correlation_id was already done in this outbox
                         (local dedup, no adapter call).
  retryable_failure   -- transient adapter failure; backoff + retry (next_retry_at).
  conflict_blocked    -- portfolio-hub 409; operator must resolve, no retry.
  permanent_failure   -- 404/400/422 client error; terminal immediately (dcy-015).
  dead_lettered       -- retryable that exhausted max_attempts; terminal (dcy-015).

`permanent_failure` and `dead_lettered` both land in the single terminal
`delivery_failed` status (distinguished by last_outcome/last_error), mirroring
the `conflict_blocked` single-status pattern (dcy-015 OQ-2).

This separation (orbit `LedgerWriter` pattern) keeps slot transitions on the
critical path while making ledger writes eventually-consistent and idempotent.
"""
from __future__ import annotations

import logging
import random
from dataclasses import dataclass, replace
from datetime import datetime, timedelta, timezone
from enum import Enum
from typing import Callable, Optional, Protocol

from trex_core.application.alarm_emitter import AlarmEmitter
from trex_core.application.ledger_port import (
    LedgerConflictError,
    LedgerPermanentError,
    LedgerPort,
    LedgerRetryableError,
    LedgerTrade,
    LedgerWriteResult,
)

_log = logging.getLogger(__name__)

# dcy-015 / REQ-019: exponential backoff defaults (overridable via config/env).
_DEFAULT_RETRY_BASE_SEC = 1.0
_DEFAULT_RETRY_CAP_SEC = 300.0
_DEFAULT_MAX_ATTEMPTS = 8


class OutboxOutcome(str, Enum):
    COMPLETED = "completed"
    DUPLICATE_COMPLETED = "duplicate_completed"
    RETRYABLE_FAILURE = "retryable_failure"
    CONFLICT_BLOCKED = "conflict_blocked"
    PERMANENT_FAILURE = "permanent_failure"   # dcy-015: 404/400/422 — terminal now
    DEAD_LETTERED = "dead_lettered"           # dcy-015: retryable exhausted max_attempts


class WorkItemStatus(str, Enum):
    PENDING = "pending"          # awaiting next writer tick
    COMPLETED = "completed"      # adapter accepted (terminal success)
    CONFLICT_BLOCKED = "conflict_blocked"  # 409 — operator resolves (terminal failure)
    DELIVERY_FAILED = "delivery_failed"    # dcy-015 — permanent or dead-lettered (terminal)


# Outcomes that move a row to the terminal delivery_failed state.
_DELIVERY_FAILED_OUTCOMES = frozenset(
    {OutboxOutcome.PERMANENT_FAILURE, OutboxOutcome.DEAD_LETTERED}
)


def trade_cash_delta(trade: LedgerTrade) -> float:
    """Estimated cash impact of a trade (dcy-018 / REQ-029 OQ-4 estimate).

    buy → -(qty*price + fee + tax); sell → +(qty*price - fee - tax). The fee is
    the core's approximation, so this is an estimate that converges to the
    portfolio-hub authoritative value on option-D write-back."""
    notional = float(trade.qty) * float(trade.price)
    fees = float(trade.fee) + float(trade.tax)
    if trade.side == "sell":
        return notional - fees
    return -(notional + fees)


@dataclass(frozen=True)
class LedgerWorkItem:
    """One outbox row.

    Identity is `correlation_id`; the outbox MUST upsert/dedupe on it so the
    same fill enqueued twice produces a single row.
    """

    correlation_id: str
    trade: LedgerTrade
    status: WorkItemStatus = WorkItemStatus.PENDING
    attempts: int = 0
    last_outcome: Optional[OutboxOutcome] = None
    last_error: Optional[str] = None
    # dcy-015 / REQ-019: earliest time a retryable row may be re-attempted.
    # list_pending skips rows whose next_retry_at is still in the future.
    next_retry_at: Optional[datetime] = None
    created_at: Optional[datetime] = None
    updated_at: Optional[datetime] = None


class LedgerOutbox(Protocol):
    """Durable work-item store backing the ledger writer."""

    def enqueue(self, trade: LedgerTrade) -> LedgerWorkItem:
        """Enqueue a trade. Idempotent on correlation_id (returns the
        existing row unchanged if it is already present)."""
        ...

    def list_pending(self, limit: int = 32) -> list[LedgerWorkItem]:
        """Return pending work items in FIFO order, up to `limit`."""
        ...

    def pending_cash_delta(self) -> dict[str, float]:
        """Sum the estimated cash delta of ALL not-yet-settled (pending) rows,
        grouped by asset_id (dcy-018 / REQ-029, OQ-4 estimate overlay).

        buy → -(qty*price + fee + tax); sell → +(qty*price - fee - tax).
        Used to show an estimated cash before option-D write-back lands."""
        ...

    def get(self, correlation_id: str) -> Optional[LedgerWorkItem]:
        ...

    def mark_outcome(
        self,
        correlation_id: str,
        outcome: OutboxOutcome,
        *,
        error: Optional[str] = None,
        next_retry_at: Optional[datetime] = None,
    ) -> LedgerWorkItem:
        """Record the outcome of a writer attempt and advance status.

        For a retryable outcome `next_retry_at` schedules the backoff; terminal
        outcomes ignore it (dcy-015 / REQ-019)."""
        ...


@dataclass
class WriterStats:
    completed: int = 0
    duplicates: int = 0
    retryable: int = 0
    conflicts: int = 0
    delivery_failed: int = 0   # dcy-015: permanent + dead-lettered terminals


class LedgerWriter:
    """Drains the outbox into a `LedgerPort`. One process-wide writer.

    Local dedup is by correlation_id: if a work item is already marked
    `completed`, the writer reports `duplicate_completed` without touching
    the adapter (REQ-019 AC-4).
    """

    def __init__(
        self,
        outbox: LedgerOutbox,
        ledger: LedgerPort,
        *,
        snapshot_store=None,
        retry_base_sec: float = _DEFAULT_RETRY_BASE_SEC,
        retry_cap_sec: float = _DEFAULT_RETRY_CAP_SEC,
        max_attempts: int = _DEFAULT_MAX_ATTEMPTS,
        clock: Optional[Callable[[], datetime]] = None,
        rng: Optional[Callable[[], float]] = None,
        alarm_emitter: Optional[AlarmEmitter] = None,
    ) -> None:
        self._outbox = outbox
        self._ledger = ledger
        # dcy-018 / REQ-019/030 (option D): SnapshotStore for cash/quantity
        # write-back. None => write-back disabled (e.g. slice-1 / tests).
        self._snapshot_store = snapshot_store
        # dcy-015 / REQ-019: exponential backoff + jitter + max-attempts cap.
        self._retry_base = retry_base_sec
        self._retry_cap = retry_cap_sec
        self._max_attempts = max_attempts
        self._clock = clock or (lambda: datetime.now(tz=timezone.utc))
        self._rng = rng or random.random  # jitter source; inject for determinism
        # dcy-033 / REQ-100: best-effort alarm emitter.
        self._alarm_emitter = alarm_emitter

    def enqueue_fill(self, trade: LedgerTrade) -> LedgerWorkItem:
        return self._outbox.enqueue(trade)

    def run_once(self, limit: int = 32) -> WriterStats:
        stats = WriterStats()
        for item in self._outbox.list_pending(limit):
            outcome = self._process(item)
            if outcome is OutboxOutcome.COMPLETED:
                stats.completed += 1
            elif outcome is OutboxOutcome.DUPLICATE_COMPLETED:
                stats.duplicates += 1
            elif outcome is OutboxOutcome.RETRYABLE_FAILURE:
                stats.retryable += 1
            elif outcome is OutboxOutcome.CONFLICT_BLOCKED:
                stats.conflicts += 1
            elif outcome in _DELIVERY_FAILED_OUTCOMES:
                stats.delivery_failed += 1
        return stats

    def _backoff_delay(self, attempts: int) -> float:
        """Exponential backoff with jitter: min(base*2^(n-1), cap) + up to 25%."""
        base_delay = min(self._retry_base * (2 ** (attempts - 1)), self._retry_cap)
        return base_delay * (1.0 + 0.25 * self._rng())

    def _process(self, item: LedgerWorkItem) -> OutboxOutcome:
        # Local dedup: another writer pass (or restart) already completed this
        # outbox row. Skip the adapter call.
        existing = self._outbox.get(item.correlation_id)
        if existing is not None and existing.status is WorkItemStatus.COMPLETED:
            self._outbox.mark_outcome(item.correlation_id, OutboxOutcome.DUPLICATE_COMPLETED)
            return OutboxOutcome.DUPLICATE_COMPLETED
        try:
            outcome = self._ledger.record_trade(item.trade)
        except LedgerConflictError as exc:
            self._outbox.mark_outcome(
                item.correlation_id, OutboxOutcome.CONFLICT_BLOCKED, error=str(exc)
            )
            # dcy-033 / REQ-100: CRITICAL alarm — 409 conflict, operator resolution needed.
            if self._alarm_emitter is not None:
                self._alarm_emitter.publish_alarm(
                    "ledger_delivery_failed",
                    f"원장 409 충돌 (conflict_blocked): {item.correlation_id}",
                    fields={"correlation_id": item.correlation_id,
                            "asset_id": item.trade.asset_id,
                            "reason": "conflict_blocked"},
                )
            return OutboxOutcome.CONFLICT_BLOCKED
        except LedgerPermanentError as exc:
            # dcy-015 / REQ-017/018 (issue #13): non-retryable client error →
            # terminal delivery_failed immediately (no retry spam). Surface for ops.
            self._outbox.mark_outcome(
                item.correlation_id, OutboxOutcome.PERMANENT_FAILURE, error=str(exc)
            )
            _log.warning(
                "ledger delivery_failed (permanent): correlation_id=%s reason=%s",
                item.correlation_id, exc,
            )
            # dcy-033 / REQ-100: CRITICAL alarm — permanent failure, no retry.
            if self._alarm_emitter is not None:
                self._alarm_emitter.publish_alarm(
                    "ledger_delivery_failed",
                    f"원장 영구 실패 (permanent): {item.correlation_id}",
                    fields={"correlation_id": item.correlation_id,
                            "asset_id": item.trade.asset_id,
                            "reason": "permanent_failure"},
                )
            return OutboxOutcome.PERMANENT_FAILURE
        except LedgerRetryableError as exc:
            # dcy-015 / REQ-019: backoff, or dead-letter once max_attempts hit.
            new_attempts = item.attempts + 1
            if new_attempts >= self._max_attempts:
                self._outbox.mark_outcome(
                    item.correlation_id, OutboxOutcome.DEAD_LETTERED, error=str(exc)
                )
                _log.warning(
                    "ledger delivery_failed (dead-letter, attempts=%d): "
                    "correlation_id=%s reason=%s",
                    new_attempts, item.correlation_id, exc,
                )
                # dcy-033 / REQ-100: CRITICAL alarm — dead-lettered after max retries.
                if self._alarm_emitter is not None:
                    self._alarm_emitter.publish_alarm(
                        "ledger_delivery_failed",
                        f"원장 데드레터 (attempts={new_attempts}): {item.correlation_id}",
                        fields={"correlation_id": item.correlation_id,
                                "asset_id": item.trade.asset_id,
                                "attempts": new_attempts,
                                "reason": "dead_lettered"},
                    )
                return OutboxOutcome.DEAD_LETTERED
            next_retry_at = self._clock() + timedelta(
                seconds=self._backoff_delay(new_attempts)
            )
            self._outbox.mark_outcome(
                item.correlation_id, OutboxOutcome.RETRYABLE_FAILURE,
                error=str(exc), next_retry_at=next_retry_at,
            )
            return OutboxOutcome.RETRYABLE_FAILURE
        # dcy-018 / REQ-019/030 (option D): write back portfolio-hub authoritative
        # cash/quantity to the snapshot store on new+duplicate (OQ-2 b). *_after
        # is absolute, so applying it on duplicate is idempotent. best-effort —
        # never let a write-back failure block the ledger outcome (NFR-1).
        self._writeback(item.trade, outcome)
        # Adapter-level dedup (server says "already recorded"): outcome is
        # duplicate_completed; the ledger has exactly one entry for this
        # execution_id either way (REQ-019 AC-1, AC-4).
        if outcome.result is LedgerWriteResult.DUPLICATE_OK:
            self._outbox.mark_outcome(item.correlation_id, OutboxOutcome.DUPLICATE_COMPLETED)
            return OutboxOutcome.DUPLICATE_COMPLETED
        self._outbox.mark_outcome(item.correlation_id, OutboxOutcome.COMPLETED)
        return OutboxOutcome.COMPLETED

    def _writeback(self, trade: LedgerTrade, outcome) -> None:
        """Option D cash/quantity write-back (dcy-018). best-effort; skips when
        no SnapshotStore is wired or the response omits authoritative values."""
        if self._snapshot_store is None:
            return
        if outcome.cash_balance_after is None or outcome.asset_quantity_after is None:
            return  # older hub / missing fields → leave snapshot as-is
        try:
            self._snapshot_store.update_cash_position(
                trade.asset_id,
                outcome.cash_balance_after,
                outcome.asset_quantity_after,
            )
        except Exception:  # noqa: BLE001 - write-back must not block the ledger path
            _log.warning(
                "snapshot cash/position write-back failed: asset_id=%s",
                trade.asset_id, exc_info=True,
            )


# --- in-memory outbox (tests + slice-1 runtime) ------------------------------

class InMemoryLedgerOutbox:
    """Process-local outbox. Thread-safe; idempotent on correlation_id."""

    def __init__(self, *, clock: Optional[Callable[[], datetime]] = None) -> None:
        import threading

        self._rows: dict[str, LedgerWorkItem] = {}
        self._order: list[str] = []
        self._lock = threading.Lock()
        # dcy-015: shared clock lets tests align backoff (next_retry_at) with
        # list_pending's "now" deterministically.
        self._clock = clock or (lambda: datetime.now(tz=timezone.utc))

    def _now(self) -> datetime:
        return self._clock()

    def enqueue(self, trade: LedgerTrade) -> LedgerWorkItem:
        with self._lock:
            cid = trade.correlation_id
            existing = self._rows.get(cid)
            if existing is not None:
                return existing
            row = LedgerWorkItem(
                correlation_id=cid,
                trade=trade,
                status=WorkItemStatus.PENDING,
                created_at=self._now(),
                updated_at=self._now(),
            )
            self._rows[cid] = row
            self._order.append(cid)
            return row

    def list_pending(self, limit: int = 32) -> list[LedgerWorkItem]:
        now = self._now()
        with self._lock:
            out: list[LedgerWorkItem] = []
            for cid in self._order:
                row = self._rows[cid]
                # dcy-015 / REQ-019: skip rows still in backoff so they don't
                # occupy the FIFO batch (head-of-line protection, NFR-3).
                if row.status is WorkItemStatus.PENDING and (
                    row.next_retry_at is None or row.next_retry_at <= now
                ):
                    out.append(row)
                    if len(out) >= limit:
                        break
            return out

    def pending_cash_delta(self) -> dict[str, float]:
        with self._lock:
            out: dict[str, float] = {}
            for row in self._rows.values():
                if row.status is WorkItemStatus.PENDING:
                    aid = row.trade.asset_id
                    out[aid] = out.get(aid, 0.0) + trade_cash_delta(row.trade)
            return out

    def get(self, correlation_id: str) -> Optional[LedgerWorkItem]:
        with self._lock:
            return self._rows.get(correlation_id)

    def mark_outcome(
        self,
        correlation_id: str,
        outcome: OutboxOutcome,
        *,
        error: Optional[str] = None,
        next_retry_at: Optional[datetime] = None,
    ) -> LedgerWorkItem:
        with self._lock:
            row = self._rows[correlation_id]
            if outcome is OutboxOutcome.COMPLETED:
                new_status = WorkItemStatus.COMPLETED
            elif outcome is OutboxOutcome.CONFLICT_BLOCKED:
                new_status = WorkItemStatus.CONFLICT_BLOCKED
            elif outcome in _DELIVERY_FAILED_OUTCOMES:
                new_status = WorkItemStatus.DELIVERY_FAILED  # dcy-015 terminal
            elif outcome is OutboxOutcome.DUPLICATE_COMPLETED:
                # already terminal; keep COMPLETED status
                new_status = WorkItemStatus.COMPLETED
            else:  # retryable
                new_status = WorkItemStatus.PENDING
            updated = replace(
                row,
                status=new_status,
                attempts=row.attempts + (0 if outcome is OutboxOutcome.DUPLICATE_COMPLETED else 1),
                last_outcome=outcome,
                last_error=error,
                next_retry_at=(
                    next_retry_at if outcome is OutboxOutcome.RETRYABLE_FAILURE else None
                ),
                updated_at=self._now(),
            )
            self._rows[correlation_id] = updated
            return updated

    # --- introspection ---
    def all_items(self) -> list[LedgerWorkItem]:
        with self._lock:
            return [self._rows[cid] for cid in self._order]