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 collections.abc import Callable
from dataclasses import dataclass, replace
from datetime import datetime, timedelta, timezone
from enum import Enum
from typing import Optional, Protocol

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

_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,
        ledger_skip_store=None,
        instance_key: str | None = 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
        self._ledger_skip_store = ledger_skip_store
        self._instance_key = instance_key

    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 process_correlation_id(self, correlation_id: str) -> OutboxOutcome | None:
        """Process one specific outbox row immediately."""
        item = self._outbox.get(correlation_id)
        if item is None:
            return None
        return self._process(item)

    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)
            self._resolve_ledger_skip(item.trade)
            return OutboxOutcome.DUPLICATE_COMPLETED
        self._outbox.mark_outcome(item.correlation_id, OutboxOutcome.COMPLETED)
        self._resolve_ledger_skip(item.trade)
        return OutboxOutcome.COMPLETED

    def _resolve_ledger_skip(self, trade: LedgerTrade) -> None:
        if (
            self._ledger_skip_store is None
            or self._instance_key is None
            or not trade.asset_id
            or not trade.order_id
            or not trade.execution_id
        ):
            return
        records = self._ledger_skip_store.resolve_for_delivery(
            instance_key=self._instance_key,
            portfolio_id=trade.asset_id,
            side=trade.side,
            order_id=trade.order_id,
            execution_id=trade.execution_id,
            resolution_reason="ledger_recorded",
        )
        for record in records:
            publish_ledger_skip_alarm(self._alarm_emitter, record, resolved=True)

    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]