"""Portfolio snapshot read + selection (REQ-027/028/029/030/031, dcy-005).

Application-layer port for reading portfolio-hub universe + in-process
snapshot store + asset-selection service. Domain stays free of HTTP and
RabbitMQ deps (NFR-1).

The snapshot is split between:
  - in-memory `SnapshotStore`   universe + cash/position + warnings (process-local;
                                warning is automatic/temporary, reset on restart)
  - durable `PauseStore`        portfolio_id -> paused (Postgres; manual/persistent,
                                survives restart and daily refresh — REQ-032)
"""
from __future__ import annotations

import threading
from dataclasses import dataclass, field, replace
from datetime import datetime, timezone
from enum import Enum
from typing import Optional, Protocol


# ---------------------------------------------------------------- portfolio-hub

@dataclass(frozen=True)
class PortfolioItem:
    """One asset in the daemon's portfolio universe (REQ-027 AC-1).

    Mirrors portfolio-hub's `list_portfolios_v1` PortfolioItem with **no price
    field** — pricing is the strategy module's concern, not core's.
    """
    portfolio_id: str               # == asset_id, dcy-004 OQ-1
    symbol: str
    cash_balance: float
    asset_quantity: float           # integer shares semantically; float for tolerance
    active_strategy_id: Optional[str] = None
    strategy_params: dict = field(default_factory=dict)  # opaque, pass-through


class PortfolioHubReadError(Exception):
    """Base class for portfolio-hub read failures."""


class PortfolioHubUnavailableError(PortfolioHubReadError):
    """portfolio-hub unreachable / 5xx / timeout — degraded signal upstream."""


class PortfolioHubReadPort(Protocol):
    """Read-side contract for portfolio-hub universe (REQ-027)."""

    def get_portfolios(self, daemon_id: str) -> list[PortfolioItem]:
        """Return all portfolios where `managed_by_daemon_id == daemon_id`."""
        ...

    def whoami(self) -> str:
        """Resolve daemon_id from the credentials/token (REQ-027 AC-4).

        Called once at startup so `daemon_id` lookup never depends on
        `TREX_DAEMON_ID` env (single source of truth = auth token).
        """
        ...


# ---------------------------------------------------------------- selection enums

class View(str, Enum):
    """Snapshot view modes (REQ-029, dcy-005 OQ-11/OQ-12)."""
    ALL = "all"             # universe ∪ holding slots
    BUYABLE = "buyable"     # universe − occupied slots
    HELD = "held"           # holding slots only


class DegradedReason(str, Enum):
    """Machine-readable degraded reason codes (REQ-028 AC-2, dcy-005 OQ-5)."""
    PORTFOLIO_HUB_UNAVAILABLE = "portfolio_hub_unavailable"
    SNAPSHOT_FETCH_FAILED = "snapshot_fetch_failed"
    CASH_POSITION_STALE = "cash_position_stale"


# ---------------------------------------------------------------- snapshot store

@dataclass(frozen=True)
class SnapshotAsset:
    """A portfolio item augmented with overlays (warning, sequence)."""
    item: PortfolioItem
    warning: bool = False
    updated_at: Optional[datetime] = None


class SnapshotStore:
    """In-memory snapshot — universe baseline + warning overlay.

    Two-phase: callers `swap(new_universe)` to atomically replace the universe
    after a fetch, and `set_warning(portfolio_id, on)` to toggle the automatic
    warning overlay (REQ-031). pause overlay lives in PauseStore (REQ-032).

    Thread-safe; the snapshot RPC handler can read concurrently with the
    scheduler refresh.
    """

    def __init__(self) -> None:
        self._lock = threading.Lock()
        self._assets: dict[str, SnapshotAsset] = {}
        self._degraded: Optional[DegradedReason] = None
        self._degraded_message: Optional[str] = None
        self._last_refresh_at: Optional[datetime] = None

    # --- universe ---
    def swap(self, items: list[PortfolioItem]) -> None:
        """Replace the entire universe atomically (immutable swap, Decision 6)."""
        now = datetime.now(tz=timezone.utc)
        with self._lock:
            prior_warnings = {pid for pid, a in self._assets.items() if a.warning}
            new_assets: dict[str, SnapshotAsset] = {}
            for item in items:
                # Carry forward warning if the same portfolio was warned before;
                # reconciliation will clear it on the next successful check.
                warn = item.portfolio_id in prior_warnings
                new_assets[item.portfolio_id] = SnapshotAsset(
                    item=item, warning=warn, updated_at=now,
                )
            self._assets = new_assets
            self._degraded = None
            self._degraded_message = None
            self._last_refresh_at = now

    def list_assets(self) -> list[SnapshotAsset]:
        with self._lock:
            return list(self._assets.values())

    def get(self, portfolio_id: str) -> Optional[SnapshotAsset]:
        with self._lock:
            return self._assets.get(portfolio_id)

    # --- warning overlay (automatic, in-memory only) ---
    def set_warning(self, portfolio_id: str, on: bool) -> None:
        with self._lock:
            a = self._assets.get(portfolio_id)
            if a is None or a.warning is on:
                return
            self._assets[portfolio_id] = replace(a, warning=on)

    def warnings(self) -> set[str]:
        with self._lock:
            return {pid for pid, a in self._assets.items() if a.warning}

    # --- option D cash/position (REQ-030 AC-2/AC-3 — wired but no-op until
    #     portfolio-hub#8 ships) ---
    def update_cash_position(
        self,
        portfolio_id: str,
        cash_balance: float,
        asset_quantity: float,
    ) -> None:
        with self._lock:
            a = self._assets.get(portfolio_id)
            if a is None:
                return
            self._assets[portfolio_id] = replace(
                a, item=replace(a.item, cash_balance=cash_balance,
                                 asset_quantity=asset_quantity),
                updated_at=datetime.now(tz=timezone.utc),
            )

    # --- degraded signal (REQ-028 AC-2) ---
    def set_degraded(self, reason: DegradedReason, message: str) -> None:
        with self._lock:
            self._degraded = reason
            self._degraded_message = _sanitize(message)

    def clear_degraded(self) -> None:
        with self._lock:
            self._degraded = None
            self._degraded_message = None

    def degraded_state(self) -> tuple[Optional[DegradedReason], Optional[str]]:
        with self._lock:
            return self._degraded, self._degraded_message

    def last_refresh_at(self) -> Optional[datetime]:
        with self._lock:
            return self._last_refresh_at


# ---------------------------------------------------------------- pause store

class PauseStore(Protocol):
    """Durable pause overlay — `portfolio_id` PK (REQ-032)."""

    def list_paused(self) -> set[str]:
        ...

    def pause(self, portfolio_id: str, *, actor: str = "operator") -> None:
        ...

    def resume(self, portfolio_id: str) -> None:
        ...


class InMemoryPauseStore:
    """Process-local pause store for tests + CI."""

    def __init__(self, initial: Optional[set[str]] = None) -> None:
        self._lock = threading.Lock()
        self._paused: set[str] = set(initial or set())

    def list_paused(self) -> set[str]:
        with self._lock:
            return set(self._paused)

    def pause(self, portfolio_id: str, *, actor: str = "operator") -> None:
        with self._lock:
            self._paused.add(portfolio_id)

    def resume(self, portfolio_id: str) -> None:
        with self._lock:
            self._paused.discard(portfolio_id)


# ---------------------------------------------------------------- kill-switch
# Day-scoped exclusion overlay (REQ-044/046 / BAT-012). Distinct from pause:
# keyed by (portfolio_id, trading_date) so it auto-clears next trading day.
GLOBAL_KILLSWITCH = "*"  # sentinel: whole-universe exclusion for the day


class KillSwitchStore(Protocol):
    """Day-scoped kill-switch exclusion overlay (REQ-044)."""

    def list_excluded(self, trading_date) -> set[str]:
        ...

    def exclude(self, portfolio_id: str, trading_date, *, actor: str = "kill-switch") -> None:
        ...

    def clear(self, portfolio_id: str, trading_date) -> None:
        ...


class InMemoryKillSwitchStore:
    """Process-local day-scoped kill-switch store for tests + CI."""

    def __init__(self) -> None:
        self._lock = threading.Lock()
        self._excluded: set[tuple[str, object]] = set()  # (portfolio_id, trading_date)

    def list_excluded(self, trading_date) -> set[str]:
        with self._lock:
            return {pid for pid, d in self._excluded if d == trading_date}

    def exclude(self, portfolio_id: str, trading_date, *, actor: str = "kill-switch") -> None:
        with self._lock:
            self._excluded.add((portfolio_id, trading_date))

    def clear(self, portfolio_id: str, trading_date) -> None:
        with self._lock:
            self._excluded.discard((portfolio_id, trading_date))


# ---------------------------------------------------------------- selection

@dataclass(frozen=True)
class SnapshotEnvelope:
    """REQ-028 response envelope (transport-agnostic)."""
    assets: list[dict]
    view: str
    available_slots: int
    degraded: bool
    reason: Optional[str]
    message: Optional[str] = None
    # REQ-028 CHG-20260608-01 (BAT-022): operator-facing per-slot state list,
    # additive and independent of strategy-facing `assets[]` (universe). Lets
    # `trex-core slots` show the daemon's real slot table (idle/holding/pending)
    # instead of universe portfolios. Always a list ([] when no slots).
    slots: list[dict] = field(default_factory=list)

    def to_dict(self) -> dict:
        return {
            "assets": list(self.assets),
            "view": self.view,
            "available_slots": self.available_slots,
            "degraded": self.degraded,
            "reason": self.reason,
            "message": self.message,
            "slots": list(self.slots),
        }


def _asset_to_dict(item: PortfolioItem) -> dict:
    """REQ-027 AC-1 / REQ-028 AC-3 — strict no-price field shape."""
    return {
        "symbol": item.symbol,
        "asset_id": item.portfolio_id,
        "cash_balance": item.cash_balance,
        "asset_quantity": item.asset_quantity,
        "active_strategy_id": item.active_strategy_id,
        "strategy_params": dict(item.strategy_params or {}),
    }


def slot_to_dict(s) -> dict:
    """Canonical operator-facing per-slot state shape (REQ-028 AC-4, BAT-022).

    Shared by the BAT-008 envelope (`build_snapshot_envelope`) and the BAT-007
    stub-shape handler so `trex-core slots` sees an identical slot dict on either
    path. Unlike `_asset_to_dict` (strategy universe, no-price), this is
    operator-only and exposes prices. External read field name `entry_price`
    is kept for contract compatibility (dcy-012 / NFR-3 — internally backed by
    the renamed `limit_price`, the pending order limit cleared once holding).
    """
    return {
        "slot_id": str(s.slot_id),
        "status": s.status.value,
        "symbol": s.symbol,
        "qty": float(s.qty) if s.qty is not None else None,
        "avg_price": float(s.avg_price) if s.avg_price is not None else None,
        "entry_price": float(s.limit_price) if s.limit_price is not None else None,
        "allocated_capital": float(s.allocated_capital) if s.allocated_capital is not None else None,
        "strategy_id": s.strategy_id,
        "sell_strategy_id": s.sell_strategy_id,
        "portfolio_id": s.portfolio_id,
        "forced_close_at": s.forced_close_at.isoformat() if s.forced_close_at else None,
        # dcy-020 / REQ-063 (BAT-029): asset-centric single view — universe-detached
        # (soft-retired/tombstone) marker so the operator sees retired assets.
        "universe_detached": getattr(s, "retired_at", None) is not None,
    }


def _sanitize(message: str) -> str:
    """Strip credentials/URLs/tokens — REQ-028 AC-2 sanitization."""
    if not message:
        return ""
    out = message
    for needle in ("authorization:", "Bearer ", "token=", "appsecret="):
        idx = out.lower().find(needle.lower())
        if idx >= 0:
            out = out[:idx] + "<redacted>"
            break
    # Truncate to keep response payload bounded.
    return out[:200]


def build_snapshot_envelope(
    *,
    store: SnapshotStore,
    pause_store: PauseStore,
    slot_repo,
    instance_key: str = "default",
    available_slots: int,
    view: View = View.ALL,
    strategy_key: Optional[str] = None,
    trading_date=None,
    killswitch_store: "Optional[KillSwitchStore]" = None,
) -> SnapshotEnvelope:
    """Compose base(view) − mask(pending·warning·pause·killswitch) − strategy_key.

    `slot_repo` is the SlotRepository used to read holding/pending state —
    instance-scoped (dcy-020/REQ-056). `trading_date` (default UTC today) is
    used only for the day-scoped kill-switch exclusion mask.
    `killswitch_store` (REQ-046) adds the day-scoped kill-switch exclusion to
    the mask: a `GLOBAL_KILLSWITCH` ("*") entry masks the whole universe.
    """
    # Read core authoritative slot state once.
    day = trading_date or datetime.now(tz=timezone.utc).date()
    slots = slot_repo.list_slots(instance_key)
    holding_pids: set[str] = set()
    pending_pids: set[str] = set()
    holding_qty: dict[str, float] = {}
    for s in slots:
        if s.portfolio_id is None:
            continue
        if s.status.value == "holding":
            holding_pids.add(s.portfolio_id)
            holding_qty[s.portfolio_id] = float(s.qty or 0.0)
        elif s.status.value in ("pending_buy", "pending_sell"):
            pending_pids.add(s.portfolio_id)

    assets = store.list_assets()
    paused = pause_store.list_paused()
    warned = {a.item.portfolio_id for a in assets if a.warning}
    # REQ-046: day-scoped kill-switch exclusion. '*' sentinel = whole universe.
    excluded = killswitch_store.list_excluded(day) if killswitch_store is not None else set()
    killswitch_global = GLOBAL_KILLSWITCH in excluded

    # Build the base set (view).
    by_pid = {a.item.portfolio_id: a for a in assets}
    base_pids: list[str]
    if view is View.HELD:
        base_pids = sorted(holding_pids)
    elif view is View.BUYABLE:
        if available_slots <= 0:
            base_pids = []
        else:
            # universe − occupied slots (pending OR holding occupy)
            occupied = holding_pids | pending_pids
            base_pids = [a.item.portfolio_id for a in assets
                         if a.item.portfolio_id not in occupied]
    else:  # ALL
        base_pids = list({a.item.portfolio_id for a in assets} | holding_pids)
        base_pids.sort()

    # Apply masks: pending OR warning OR pause OR kill-switch day-exclusion.
    masked = pending_pids | warned | paused | excluded

    # Filter and shape.
    out: list[dict] = []
    for pid in base_pids:
        if killswitch_global or pid in masked:
            continue
        if pid in by_pid:
            item = by_pid[pid].item
        elif pid in holding_pids:
            # holding slot for an asset NOT in the current universe (e.g. fetch
            # gap) — synthesize a minimal item using slot data so the response
            # still carries authoritative position info (REQ-029 slot overlay
            # priority over portfolio-hub presence).
            slot = next((s for s in slots if s.portfolio_id == pid), None)
            item = PortfolioItem(
                portfolio_id=pid,
                symbol=(slot.symbol if slot else "") or "",
                cash_balance=0.0,
                asset_quantity=float((slot.qty if slot else 0.0) or 0.0),
                active_strategy_id=(slot.strategy_id if slot else None),
                strategy_params={},
            )
        else:
            continue
        if strategy_key is not None and item.active_strategy_id != strategy_key:
            continue
        d = _asset_to_dict(item)
        # holding slot quantity wins for held assets (slot-overlay priority).
        if pid in holding_qty:
            d["asset_quantity"] = holding_qty[pid]
        out.append(d)

    # REQ-028 CHG-20260608-01 (BAT-022): operator-facing per-slot state — the
    # full daemon slot table for `day`, independent of the asset view filter.
    slot_states = [slot_to_dict(s) for s in slots]

    reason_code, message = store.degraded_state()
    return SnapshotEnvelope(
        assets=out,
        view=view.value,
        available_slots=max(0, int(available_slots)),
        degraded=reason_code is not None,
        reason=reason_code.value if reason_code is not None else None,
        message=message,
        slots=slot_states,
    )