"""Repository for the cockpit's forward tables. Idempotent upserts (re-ingest is safe). `migrate()` creates
the tables and, on Postgres only, promotes forward_nav to a TimescaleDB hypertable (guarded; on SQLite it
stays a plain table so tests run without Postgres)."""
from __future__ import annotations

import datetime as dt
from typing import Any

from sqlalchemy import create_engine, select, text
from sqlalchemy.orm import Session
from sqlalchemy.pool import StaticPool

from fxhnt.adapters.persistence.cockpit_models import (
    BacktestSummaryRow,
    CockpitBase,
    ForwardBookStateRow,
    ForwardNavRow,
    ForwardRecordRow,
    ForwardSummaryRow,
    IbkrPitVolumeRow,
    StrategyAllocationRow,
    StrategyAnchorRow,
    StrategyFundingRow,
    StrategyHealthRow,
    StrategyPromotionRow,
    StrategyRegistryRow,
    XspOptionBarRow,
    YahooPitCloseRow,
    YahooPitVolumeRow,
)
from fxhnt.application.forward_models import BacktestSummary, ForwardSummary
from fxhnt.application.forward_models import ForwardNavRow as NavRowDTO
from fxhnt.registry import STRATEGY_REGISTRY


class ForwardNavRepo:
    def __init__(self, dsn: str) -> None:
        # StaticPool keeps a single in-memory SQLite connection alive across sessions; only applied for the
        # in-memory SQLite case (tests). StaticPool is NOT used for Postgres.
        kw: dict[str, Any] = {"future": True}
        if dsn == "sqlite://" or (dsn.startswith("sqlite://") and ":memory:" in dsn):
            kw |= {"connect_args": {"check_same_thread": False}, "poolclass": StaticPool}
        self._engine = create_engine(dsn, **kw)
        self._is_pg = self._engine.dialect.name == "postgresql"

    def migrate(self) -> None:
        CockpitBase.metadata.create_all(self._engine)
        if self._is_pg:
            with self._engine.begin() as c:
                c.execute(text("CREATE EXTENSION IF NOT EXISTS timescaledb"))
                c.execute(text(
                    "SELECT create_hypertable('forward_nav', by_range('date'), if_not_exists => TRUE)"
                ))
        # create_all does NOT add columns to an already-existing forward_summary table (prod Postgres has
        # rows; a persistent test sqlite file lags the ORM) — additively ensure the B1 policy-audit columns
        # on BOTH dialects so the ORM's SELECT never references a missing column.
        self._ensure_summary_policy_columns()
        self._seed_registry()

    def _ensure_summary_policy_columns(self) -> None:
        """Idempotently add forward_summary.policy_version / policy_hash to an EXISTING table. Postgres has
        `ADD COLUMN IF NOT EXISTS`; sqlite does not, so PRAGMA-check the columns first. A fresh table already
        has them from create_all, so this is a no-op there."""
        with self._engine.begin() as c:
            if self._is_pg:
                c.execute(text("ALTER TABLE forward_summary ADD COLUMN IF NOT EXISTS policy_version INTEGER"))
                c.execute(text("ALTER TABLE forward_summary ADD COLUMN IF NOT EXISTS policy_hash VARCHAR(64)"))
            else:
                cols = {row[1] for row in c.execute(text("PRAGMA table_info(forward_summary)"))}
                if "policy_version" not in cols:
                    c.execute(text("ALTER TABLE forward_summary ADD COLUMN policy_version INTEGER"))
                if "policy_hash" not in cols:
                    c.execute(text("ALTER TABLE forward_summary ADD COLUMN policy_hash VARCHAR(64)"))

    def _seed_registry(self) -> None:
        with Session(self._engine) as s:
            for sid, e in STRATEGY_REGISTRY.items():
                s.merge(StrategyRegistryRow(
                    strategy_id=sid, display_name=e["display_name"], sleeve=e["sleeve"],
                    venue=e.get("venue", ""), capital_base=float(e.get("capital_base", 0.0)),
                    gate_spec=e["gate_spec"], enabled=True))
            # PRUNE rows for strategies removed from the Python registry, so retirements stick — else a
            # merge-only seed leaves stale tracks on the dashboard forever (the lingering-`funding` bug).
            for row in s.query(StrategyRegistryRow).all():
                if row.strategy_id not in STRATEGY_REGISTRY:
                    s.delete(row)
            s.commit()

    def upsert_rows(self, rows: list[NavRowDTO], at: dt.datetime) -> None:
        with Session(self._engine) as s:
            for r in rows:
                s.merge(ForwardNavRow(strategy_id=r.strategy_id,
                                      date=dt.date.fromisoformat(r.date),  # DTO str → DB DATE
                                      ret=r.ret, nav=r.nav, src_updated_at=at))
            s.commit()

    def replace_nav(self, strategy_id: str, rows: list[NavRowDTO], at: dt.datetime) -> None:
        """Replace a strategy's ENTIRE forward_nav series with exactly `rows` — the derived cache must equal
        the current record, so a re-inception (new t0) leaves no stale pre-t0 rows behind. Delete-then-insert
        in one transaction (atomic)."""
        with Session(self._engine) as s:
            s.query(ForwardNavRow).filter(ForwardNavRow.strategy_id == strategy_id).delete()
            for r in rows:
                s.merge(ForwardNavRow(strategy_id=r.strategy_id, date=dt.date.fromisoformat(r.date),
                                      ret=r.ret, nav=r.nav, src_updated_at=at))
            s.commit()

    def upsert_summary(self, sm: ForwardSummary, gate_status: str, gate_reason: str, at: dt.datetime,
                       *, policy_version: int | None = None, policy_hash: str | None = None) -> None:
        with Session(self._engine) as s:
            s.merge(ForwardSummaryRow(
                strategy_id=sm.strategy_id, as_of=sm.as_of, days=sm.days, nav=sm.nav,
                total_return=sm.total_return, sharpe=sm.sharpe, maxdd=sm.maxdd,
                gate_status=gate_status, gate_reason=gate_reason, src_updated_at=at,
                policy_version=policy_version, policy_hash=policy_hash))
            s.commit()

    def upsert_backtest_summary(self, sm: BacktestSummary, at: dt.datetime) -> None:
        with Session(self._engine) as s:
            s.merge(BacktestSummaryRow(
                strategy_id=sm.strategy_id, as_of=sm.as_of, cagr=sm.cagr, ann_vol=sm.ann_vol,
                sharpe=sm.sharpe, max_drawdown=sm.max_drawdown, passed=sm.passed, dsr=sm.dsr,
                is_sharpe=sm.is_sharpe, oos_sharpe=sm.oos_sharpe, pvalue=sm.pvalue, src_updated_at=at))
            s.commit()

    def all_backtest_summaries(self) -> list[BacktestSummaryRow]:
        with Session(self._engine) as s:
            return list(s.scalars(select(BacktestSummaryRow).order_by(BacktestSummaryRow.strategy_id)))

    def replace_health(self, rows: list[StrategyHealthRow], at: dt.datetime) -> None:
        """Replace the ENTIRE strategy_health snapshot (delete-then-insert, atomic) so a track that RECOVERS
        clears its old STALE/NO_REF row and a retired track leaves no stale health row behind — the table
        reflects exactly this run's verdicts."""
        with Session(self._engine) as s:
            s.query(StrategyHealthRow).delete()
            for r in rows:
                r.src_updated_at = at
                s.add(r)
            s.commit()

    def all_health(self) -> list[StrategyHealthRow]:
        with Session(self._engine) as s:
            return list(s.scalars(select(StrategyHealthRow).order_by(StrategyHealthRow.strategy_id)))

    def xsp_freeze_max_date(self) -> str | None:
        """The latest date present in the opra-pit-anchored freeze table (`xsp_option_bars`), ISO yyyy-mm-dd,
        or None when the table is empty. The freeze-staleness axis: a frozen-input recompute is bounded by
        this max date, so if the freeze stops advancing the track is STALE even though its forward 'updates'."""
        from sqlalchemy import func
        with Session(self._engine) as s:
            d = s.scalar(select(func.max(XspOptionBarRow.date)))
            return d.isoformat() if d else None

    def promote(self, strategy_id: str, at: dt.datetime) -> None:
        """Record a ONE-WAY research→deploy promotion (idempotent: merge on the PK, keeps the first
        promoted_at on re-runs since callers only promote a track that is not already promoted)."""
        with Session(self._engine) as s:
            s.merge(StrategyPromotionRow(strategy_id=strategy_id, promoted_at=at))
            s.commit()

    def promoted_strategy_ids(self) -> set[str]:
        """The set of strategy_ids promoted to deploy (the fleet's effective tier reads this)."""
        with Session(self._engine) as s:
            return set(s.scalars(select(StrategyPromotionRow.strategy_id)))

    def all_nav_histories(self) -> dict[str, list[NavRowDTO]]:
        """One query for every strategy's nav series, grouped by strategy_id (avoids N+1 on the cockpit)."""
        stmt = select(ForwardNavRow).order_by(ForwardNavRow.strategy_id, ForwardNavRow.date)
        out: dict[str, list[NavRowDTO]] = {}
        with Session(self._engine) as s:
            for r in s.scalars(stmt):
                out.setdefault(r.strategy_id, []).append(
                    NavRowDTO(strategy_id=r.strategy_id, date=r.date.isoformat(), ret=r.ret, nav=r.nav))
        return out

    def nav_history(self, strategy_id: str) -> list[NavRowDTO]:
        stmt = (select(ForwardNavRow).where(ForwardNavRow.strategy_id == strategy_id)
                .order_by(ForwardNavRow.date))
        with Session(self._engine) as s:
            return [NavRowDTO(strategy_id=r.strategy_id,
                              date=r.date.isoformat(),  # DB DATE → DTO ISO str (datetime.date on both PG and SQLite)
                              ret=r.ret, nav=r.nav)
                    for r in s.scalars(stmt)]

    def all_summaries(self) -> list[ForwardSummaryRow]:
        with Session(self._engine) as s:
            return list(s.scalars(select(ForwardSummaryRow).order_by(ForwardSummaryRow.strategy_id)))

    def cum_return(self, strategy_id: str) -> float | None:
        """The persisted `forward_summary.total_return` for `strategy_id`, or None if no summary row exists
        yet (no recorded forward days). Used by `exec_reconcile.exec_vs_sim` (Task 3) to compare a strategy's
        executed vs modeled cumulative forward return without re-deriving it from the nav history."""
        with Session(self._engine) as s:
            row = s.get(ForwardSummaryRow, strategy_id)
            return row.total_return if row is not None else None

    def registry(self) -> list[StrategyRegistryRow]:
        """The cockpit-visible registry rows. SINGLE SOURCE for the hide: an ARCHIVED sleeve (its
        `STRATEGY_REGISTRY` entry carries `archived: True` — e.g. a shelved/FALSIFIED strategy) is
        excluded here, so EVERY caller (fleet overview, per-strategy gate specs, `detail()`'s reg lookup)
        drops it automatically. The DB row is still seeded (code + data kept) — it is only hidden from the UI."""
        with Session(self._engine) as s:
            rows = s.scalars(select(StrategyRegistryRow).order_by(StrategyRegistryRow.strategy_id))
            return [r for r in rows if not STRATEGY_REGISTRY.get(r.strategy_id, {}).get("archived")]

    def has_forward_history(self, strategy_id: str) -> bool:
        """True if this track has ANY persisted forward history — a forward_summary row OR any forward_anchor
        row (active or archived). Used by the engine to tell a genuine first run (no history) from a lost
        active anchor (history exists but no active anchor) so the latter fails loud instead of silently
        re-inceptioning."""
        with Session(self._engine) as s:
            if s.query(ForwardSummaryRow).filter(ForwardSummaryRow.strategy_id == strategy_id).first():
                return True
            return s.query(StrategyAnchorRow).filter(
                StrategyAnchorRow.strategy_id == strategy_id).first() is not None

    def active_anchor(self, strategy_id: str) -> StrategyAnchorRow | None:
        stmt = (select(StrategyAnchorRow)
                .where(StrategyAnchorRow.strategy_id == strategy_id,
                       StrategyAnchorRow.archived_at.is_(None))
                .order_by(StrategyAnchorRow.created_at.desc()))
        with Session(self._engine) as s:
            return s.scalars(stmt).first()

    def set_anchor(self, strategy_id: str, t0: str, definition_hash: str, definition_version: int,
                   params_json: dict[str, Any], record_mode: str, at: dt.datetime) -> None:
        """Self-guards the "at most one active anchor per strategy" invariant: any currently-active row
        for this strategy_id with a different created_at is archived before the new row is merged in, so
        callers no longer need to remember to call archive_anchor first. A same-`at` re-run (idempotent
        re-ingest of the same anchor) is left alone and just merges."""
        with Session(self._engine) as s:
            for row in s.query(StrategyAnchorRow).filter(
                    StrategyAnchorRow.strategy_id == strategy_id,
                    StrategyAnchorRow.archived_at.is_(None),
                    StrategyAnchorRow.created_at != at):
                row.archived_at = at
            s.merge(StrategyAnchorRow(
                strategy_id=strategy_id, created_at=at, t0=t0, definition_hash=definition_hash,
                definition_version=definition_version, params_json=params_json,
                record_mode=record_mode, archived_at=None))
            s.commit()

    def archive_anchor(self, strategy_id: str, at: dt.datetime) -> None:
        """Mark the currently-active anchor archived. No-op if none is active."""
        with Session(self._engine) as s:
            for row in s.query(StrategyAnchorRow).filter(
                    StrategyAnchorRow.strategy_id == strategy_id,
                    StrategyAnchorRow.archived_at.is_(None)):
                row.archived_at = at
            s.commit()

    def append_record(self, strategy_id: str, date: str, ret: float, at: dt.datetime) -> None:
        """`date` must be an ISO-8601 (YYYY-MM-DD) date string; it is normalized (zero-padded) and
        validated via `date.fromisoformat` before the merge, since the (strategy_id, date) PK and the
        lexicographic `since` filter in `record_series` both depend on a canonical, zero-padded form —
        a malformed or non-zero-padded date would silently break idempotency and filtering."""
        date = dt.date.fromisoformat(date).isoformat()
        with Session(self._engine) as s:
            s.merge(ForwardRecordRow(strategy_id=strategy_id, date=date, ret=ret, booked_at=at))
            s.commit()

    def record_series(self, strategy_id: str, since: str | None = None) -> list[tuple[str, float]]:
        """`since`, if given, must be an ISO-8601 (YYYY-MM-DD) date string; it is normalized the same way
        as `append_record` so the lexicographic `>=` comparison against the stored dates is correct."""
        if since is not None:
            since = dt.date.fromisoformat(since).isoformat()
        stmt = (select(ForwardRecordRow).where(ForwardRecordRow.strategy_id == strategy_id)
                .order_by(ForwardRecordRow.date))
        if since is not None:
            stmt = stmt.where(ForwardRecordRow.date >= since)
        with Session(self._engine) as s:
            return [(r.date, r.ret) for r in s.scalars(stmt)]

    def get_book_state(self, strategy_id: str) -> dict[str, Any]:
        with Session(self._engine) as s:
            row = s.get(ForwardBookStateRow, strategy_id)
            return dict(row.extra_json) if row is not None else {}

    def set_book_state(self, strategy_id: str, extra: dict[str, Any], at: dt.datetime) -> None:
        with Session(self._engine) as s:
            s.merge(ForwardBookStateRow(strategy_id=strategy_id, extra_json=extra, updated_at=at))
            s.commit()

    def snapshot_yahoo_closes(self, symbol: str, series: dict[str, float], at: dt.datetime) -> None:
        """Insert only (symbol, date) pairs not already stored — first-seen wins so the PIT value is frozen."""
        with Session(self._engine) as s:
            have = {d for (d,) in s.execute(
                select(YahooPitCloseRow.date).where(YahooPitCloseRow.symbol == symbol)).all()}
            for d, px in series.items():
                if d not in have:
                    s.add(YahooPitCloseRow(symbol=symbol, date=d, adj_close=float(px), first_seen_at=at))
            s.commit()

    def yahoo_pit_closes(self, symbol: str) -> dict[str, float]:
        stmt = (select(YahooPitCloseRow).where(YahooPitCloseRow.symbol == symbol)
                .order_by(YahooPitCloseRow.date))
        with Session(self._engine) as s:
            return {r.date: r.adj_close for r in s.scalars(stmt)}

    def snapshot_yahoo_volumes(self, symbol: str, series: dict[str, float], at: dt.datetime) -> None:
        """Insert only (symbol, date) pairs not already stored — first-seen wins so the PIT value is frozen."""
        with Session(self._engine) as s:
            have = {d for (d,) in s.execute(
                select(YahooPitVolumeRow.date).where(YahooPitVolumeRow.symbol == symbol)).all()}
            for d, v in series.items():
                if d not in have:
                    s.add(YahooPitVolumeRow(symbol=symbol, date=d, volume=float(v), first_seen_at=at))
            s.commit()

    def yahoo_pit_volumes(self, symbol: str) -> dict[str, float]:
        stmt = (select(YahooPitVolumeRow).where(YahooPitVolumeRow.symbol == symbol)
                .order_by(YahooPitVolumeRow.date))
        with Session(self._engine) as s:
            return {r.date: r.volume for r in s.scalars(stmt)}

    def snapshot_ibkr_volumes(self, symbol: str, series: dict[str, float], at: dt.datetime) -> None:
        """Insert only (symbol, date) pairs not already stored — first-seen wins so the PIT value is frozen."""
        with Session(self._engine) as s:
            have = {d for (d,) in s.execute(
                select(IbkrPitVolumeRow.date).where(IbkrPitVolumeRow.symbol == symbol)).all()}
            for d, v in series.items():
                if d not in have:
                    s.add(IbkrPitVolumeRow(symbol=symbol, date=d, volume=float(v), first_seen_at=at))
            s.commit()

    def ibkr_pit_volumes(self, symbol: str) -> dict[str, float]:
        stmt = (select(IbkrPitVolumeRow).where(IbkrPitVolumeRow.symbol == symbol)
                .order_by(IbkrPitVolumeRow.date))
        with Session(self._engine) as s:
            return {r.date: r.volume for r in s.scalars(stmt)}

    def replace_allocations(self, rows: list[StrategyAllocationRow], at: dt.datetime) -> None:
        """Replace the ENTIRE strategy_allocation snapshot with `rows` in one transaction (delete-then-insert),
        so a strategy that drops out of the allocation this run leaves no stale row behind — the table reflects
        exactly the current allocation."""
        with Session(self._engine) as s:
            s.query(StrategyAllocationRow).delete()
            for r in rows:
                r.computed_at = at
                s.add(r)
            s.commit()

    def current_allocation(self) -> dict[str, float]:
        with Session(self._engine) as s:
            return {r.strategy_id: r.target_dollars for r in s.scalars(select(StrategyAllocationRow))}

    def replace_funding(self, rows: list[StrategyFundingRow], at: dt.datetime) -> None:
        """Replace the ENTIRE strategy_funding snapshot (delete-then-insert) so an edge that leaves the
        effective-deploy set leaves no stale funding row behind."""
        with Session(self._engine) as s:
            s.query(StrategyFundingRow).delete()
            for r in rows:
                r.computed_at = at
                s.add(r)
            s.commit()

    def current_funding(self) -> dict[str, str]:
        with Session(self._engine) as s:
            return {r.strategy_id: r.state for r in s.scalars(select(StrategyFundingRow))}

    def all_allocations(self) -> list[StrategyAllocationRow]:
        with Session(self._engine) as s:
            return list(s.scalars(select(StrategyAllocationRow)))

    def all_funding(self) -> list[StrategyFundingRow]:
        with Session(self._engine) as s:
            return list(s.scalars(select(StrategyFundingRow)))
