"""HistoricalEodIngest — resumable survivorship-free historical EOD ingest into the warehouse.
NO live network: the Tiingo bar client + universe source are FAKES; the DuckDB store is real on a
tmp path.

Invariants proven here:
  (1) adjclose/close/volume are written as point-in-time features and round-trip via read_features
      AND read_panel (the cross-sectional path a walk-forward backtest reads).
  (2) ticker membership (symbol, exchange, start_date, end_date) is stored for EVERY name in the
      survivorship-free set INCLUDING delisted ones (a delisted ticker is RETAINED, not dropped).
  (3) re-running ingest is idempotent + cheap: no duplicate rows, and an already-present ticker is
      SKIPPED (its bar client is not called again).
  (4) a subset ingest(["AAA"]) touches ONLY AAA — no other ticker is fetched or written.
"""
from __future__ import annotations

import datetime as dt
import threading
import time

from fxhnt.adapters.warehouse.duckdb_feature_store import DuckDbFeatureStore
from fxhnt.application.historical_eod_ingest import HistoricalEodIngest

_EPOCH = dt.date(1970, 1, 1)


def _ts(iso: str) -> int:
    """ISO date → epoch-second at 00:00:00 UTC (the warehouse bar ts convention)."""
    return (dt.date.fromisoformat(iso) - _EPOCH).days * 86_400


# --- fakes -------------------------------------------------------------------------------------

# Three names: two active, one DELISTED (old endDate). The delisted one must survive into membership.
_MEMBERS = [
    ("AAA", "NYSE", "2000-01-01", "2026-06-01"),
    ("BBB", "NASDAQ", "2010-01-01", "2026-06-01"),
    ("DEAD", "AMEX", "1998-01-01", "2005-03-15"),  # delisted (survivorship-free retention target)
]

_BARS = {
    "AAA": [
        {"date": "2026-01-02", "adjClose": 90.0, "close": 100.0, "volume": 1_000.0},
        {"date": "2026-01-05", "adjClose": 91.5, "close": 101.0, "volume": 1_200.0},
    ],
    "BBB": [
        {"date": "2026-01-02", "adjClose": 40.0, "close": 50.0, "volume": 2_000.0},
    ],
    "DEAD": [
        {"date": "2005-03-14", "adjClose": 5.0, "close": 6.0, "volume": 300.0},
        {"date": "2005-03-15", "adjClose": 4.5, "close": 5.5, "volume": 280.0},
    ],
}


class FakeUniverseSource:
    """Enumerates the survivorship-free candidate set WITH dates (incl. delisted)."""

    def __init__(self) -> None:
        self.calls = 0

    def survivorship_free(self) -> list[tuple[str, str, str, str]]:
        self.calls += 1
        return list(_MEMBERS)


class FakeBarClient:
    """Returns fixed full-history bars per ticker; records which tickers were fetched (cost proxy)."""

    def __init__(self) -> None:
        self.fetched: list[str] = []

    def bars(self, symbol: str, start_date: str | None = None) -> list[dict]:
        self.fetched.append(symbol)
        return list(_BARS.get(symbol, []))


# --- tests -------------------------------------------------------------------------------------


def test_features_written_and_round_trip(tmp_path) -> None:
    store = DuckDbFeatureStore(str(tmp_path / "wh.duckdb"))
    svc = HistoricalEodIngest(FakeBarClient(), store, FakeUniverseSource())

    result = svc.ingest()

    assert result["tickers"] == 3
    assert result["rows"] == 5  # AAA 2 + BBB 1 + DEAD 2

    # round-trip via read_features: all three features present at the right ts
    aaa = store.read_features("AAA")
    assert aaa[0] == (_ts("2026-01-02"), {"adjclose": 90.0, "close": 100.0, "volume": 1_000.0})
    assert aaa[1] == (_ts("2026-01-05"), {"adjclose": 91.5, "close": 101.0, "volume": 1_200.0})

    # round-trip via read_panel (the backtest cross-sectional read), per feature
    adj = store.read_panel(["AAA", "BBB", "DEAD"], "adjclose")
    assert adj["AAA"][_ts("2026-01-02") // 86_400] == 90.0
    assert adj["BBB"][_ts("2026-01-02") // 86_400] == 40.0
    assert adj["DEAD"][_ts("2005-03-15") // 86_400] == 4.5
    vol = store.read_panel(["DEAD"], "volume")
    assert vol["DEAD"][_ts("2005-03-14") // 86_400] == 300.0
    store.close()


def test_delisted_membership_retained(tmp_path) -> None:
    store = DuckDbFeatureStore(str(tmp_path / "wh.duckdb"))
    HistoricalEodIngest(FakeBarClient(), store, FakeUniverseSource()).ingest()

    members = {r[0]: r for r in store.read_membership()}
    assert set(members) == {"AAA", "BBB", "DEAD"}
    # the delisted name is RETAINED with its real (old) end_date — not dropped, not "active-ified"
    assert members["DEAD"] == ("DEAD", "AMEX", "1998-01-01", "2005-03-15")
    assert members["AAA"] == ("AAA", "NYSE", "2000-01-01", "2026-06-01")
    store.close()


def test_rerun_is_idempotent_and_skips_present(tmp_path) -> None:
    store = DuckDbFeatureStore(str(tmp_path / "wh.duckdb"))
    bar_client = FakeBarClient()
    svc = HistoricalEodIngest(bar_client, store, FakeUniverseSource())

    svc.ingest()
    assert sorted(bar_client.fetched) == ["AAA", "BBB", "DEAD"]
    rows_after_first = len(store.read_features("AAA"))

    # re-run: already-present tickers are SKIPPED (bar client not called again)
    bar_client.fetched.clear()
    result2 = svc.ingest()
    assert bar_client.fetched == []  # nothing refetched
    assert result2["skipped"] == 3

    # no duplicate rows
    assert len(store.read_features("AAA")) == rows_after_first
    panel = store.read_panel(["AAA"], "close")
    assert len(panel["AAA"]) == 2  # still exactly the two original days
    store.close()


def test_empty_bars_ticker_not_refetched(tmp_path) -> None:
    """A name whose bars() returns [] (delisted, no price history) is recorded as PROCESSED via its
    membership row on the first run, so the next incremental run does NOT re-fetch it (no wasted call)."""
    store = DuckDbFeatureStore(str(tmp_path / "wh.duckdb"))
    bar_client = FakeBarClient()

    class EmptyBarUniverse:
        def __init__(self) -> None:
            self.calls = 0

        def survivorship_free(self) -> list[tuple[str, str, str, str]]:
            self.calls += 1
            # GONE has membership metadata but FakeBarClient has no bars for it → bars() returns []
            return [
                ("AAA", "NYSE", "2000-01-01", "2026-06-01"),
                ("GONE", "AMEX", "1990-01-01", "1995-12-31"),
            ]

    svc = HistoricalEodIngest(bar_client, store, EmptyBarUniverse())

    result = svc.ingest()
    # GONE was fetched once (its empty result recorded), AAA fetched + written
    assert sorted(bar_client.fetched) == ["AAA", "GONE"]
    assert result["empty"] == 1  # GONE
    assert result["tickers"] == 2  # both attempted this run
    # GONE wrote no price rows but DID get a membership row (the "processed" marker)
    assert "GONE" not in {c[0] for c in store.catalog()}
    assert "GONE" in {r[0] for r in store.read_membership()}

    # re-run: GONE is now marked processed (membership) → NOT refetched; AAA present in catalog → skipped
    bar_client.fetched.clear()
    result2 = svc.ingest()
    assert bar_client.fetched == []  # zero calls, GONE included
    assert result2["skipped"] == 2
    assert result2["tickers"] == 0
    assert result2["empty"] == 0
    store.close()


def test_concurrent_ingest_writes_all_and_is_thread_safe(tmp_path) -> None:
    """With max_workers>1 the FETCH runs on a thread pool but every DuckDB WRITE must happen on the
    MAIN thread (DuckDB is single-writer; concurrent writes to one connection/file corrupt it). Drive
    30 tickers (1 empty + 1 raising) and assert: all non-empty rows written, empty marked, raiser
    skipped, counts exact, AND the store was only ever touched from the calling (main) thread."""
    main_ident = threading.get_ident()
    n = 30
    raiser = "T07"  # one ticker whose fetch raises
    empty_ticker = "T11"  # one ticker whose bars come back empty
    syms = [f"T{i:02d}" for i in range(n)]
    members = [(s, "NYSE", "2000-01-01", "2026-06-01") for s in syms]

    bars_by_sym: dict[str, list[dict]] = {}
    for i, s in enumerate(syms):
        if s in (raiser, empty_ticker):
            continue
        # i+1 bars per ticker so we can verify exact row totals
        bars_by_sym[s] = [
            {"date": f"2026-01-{d + 1:02d}", "adjClose": 1.0 + d, "close": 2.0 + d, "volume": 100.0 + d}
            for d in range(i + 1)
        ]
    expected_rows = sum(len(b) for b in bars_by_sym.values())
    nonempty_syms = set(bars_by_sym)

    class ManyUniverse:
        def survivorship_free(self) -> list[tuple[str, str, str, str]]:
            return list(members)

    class ConcurrentBarClient:
        """Fetched on worker threads. Returns fixed bars; one ticker raises."""

        def __init__(self) -> None:
            self.fetched: list[str] = []
            self._lock = threading.Lock()

        def bars(self, symbol: str, start_date: str | None = None) -> list[dict]:
            with self._lock:
                self.fetched.append(symbol)
            if symbol == raiser:
                raise RuntimeError("boom")
            return list(bars_by_sym.get(symbol, []))

    class ThreadAssertingStore:
        """Wraps the real DuckDB store; records the calling thread of every mutating call and asserts
        it is the MAIN thread (worker threads must NEVER touch the store)."""

        def __init__(self, inner: DuckDbFeatureStore) -> None:
            self._inner = inner
            self.write_threads: set[int] = set()

        def catalog(self):
            return self._inner.catalog()

        def read_membership(self):
            return self._inner.read_membership()

        def write_features_bulk(self, items):
            self.write_threads.add(threading.get_ident())
            return self._inner.write_features_bulk(items)

        def upsert_membership(self, rows):
            self.write_threads.add(threading.get_ident())
            return self._inner.upsert_membership(rows)

        def read_features(self, symbol):
            return self._inner.read_features(symbol)

        def read_panel(self, symbols, feature):
            return self._inner.read_panel(symbols, feature)

        def close(self):
            return self._inner.close()

    inner = DuckDbFeatureStore(str(tmp_path / "wh.duckdb"))
    store = ThreadAssertingStore(inner)
    bar_client = ConcurrentBarClient()
    svc = HistoricalEodIngest(bar_client, store, ManyUniverse())

    result = svc.ingest(max_workers=8)

    # every ticker was attempted exactly once (raiser included, empty included)
    assert sorted(bar_client.fetched) == sorted(syms)
    # counts exact under concurrency
    assert result["tickers"] == n  # all attempted (incl. raiser + empty)
    assert result["rows"] == expected_rows
    # both the empty-bars ticker and the raiser produced no price rows → counted as "empty"
    # (no-rows-written), and both still get a membership "attempted" marker.
    assert result["empty"] == 2  # empty_ticker + raiser (both produced zero price rows)
    assert result["members"] == n  # every attempted ticker gets a membership "processed" marker
    assert result["skipped"] == 0

    # THREAD SAFETY: every store write happened on the MAIN thread only
    assert store.write_threads == {main_ident}

    # all non-empty tickers' rows are present and correct
    for s in nonempty_syms:
        feats = store.read_features(s)
        assert len(feats) == len(bars_by_sym[s])
    # empty ticker wrote no price rows but IS in membership (processed marker)
    assert empty_ticker not in {c[0] for c in store.catalog()}
    assert empty_ticker in {r[0] for r in store.read_membership()}
    # raiser wrote no price rows but IS recorded as attempted (membership) so it is not refetched
    assert raiser not in {c[0] for c in store.catalog()}
    assert raiser in {r[0] for r in store.read_membership()}

    # re-run: nothing refetched (all now present/attempted)
    bar_client.fetched.clear()
    result2 = svc.ingest(max_workers=8)
    assert bar_client.fetched == []
    assert result2["skipped"] == n
    assert result2["tickers"] == 0
    store.close()


def test_bounded_window_caps_in_flight(tmp_path) -> None:
    """The ingest must NOT submit all to-do tickers at once (that lets completed ticker-bar results
    accumulate in memory unboundedly → RSS ~2.3GB → OOM eviction in prod). Instead it keeps a bounded
    sliding window of at most `2 * max_workers` futures in flight / pending-write.

    Two proxies, both bounded by the window:
      (1) PEAK concurrently-in-flight `bars()` calls (the fetch fan-out).
      (2) PEAK RESULT BACKLOG = fetches that have COMPLETED (result in memory) but not yet been
          WRITTEN. THIS is the memory bug: with submit-all + a serial DuckDB writer, fast fetchers
          race ahead of the single writer and completed ticker-bar results pile up unbounded toward
          the universe size. A bounded window can never have more than `window` results outstanding
          because it refuses to submit ticker k+window until result k has been written + dropped.
    To surface (2) we make the WRITER slow relative to fetches so a backlog would form if unbounded.
    All tickers still ingest with correct counts — the window changes peak memory, not the result."""
    n = 60
    max_workers = 5
    window = 2 * max_workers  # 10
    syms = [f"W{i:02d}" for i in range(n)]
    members = [(s, "NYSE", "2000-01-01", "2026-06-01") for s in syms]

    base = dt.date(2020, 1, 1)
    bars_by_sym: dict[str, list[dict]] = {
        s: [
            {
                "date": (base + dt.timedelta(days=d)).isoformat(),
                "adjClose": 1.0 + d,
                "close": 2.0 + d,
                "volume": 100.0 + d,
            }
            for d in range(i + 1)
        ]
        for i, s in enumerate(syms)
    }
    expected_rows = sum(len(b) for b in bars_by_sym.values())

    class ManyUniverse:
        def survivorship_free(self) -> list[tuple[str, str, str, str]]:
            return list(members)

    # Shared backlog accounting: bar client marks a result "produced" (completed fetch held in
    # memory); the store marks it "consumed" (written + dropped). peak(produced - consumed) is the
    # number of completed-but-unwritten ticker results — the in-memory backlog the bug grows.
    backlog_lock = threading.Lock()
    state = {"outstanding": 0, "peak_backlog": 0}

    def _produced() -> None:
        with backlog_lock:
            state["outstanding"] += 1
            if state["outstanding"] > state["peak_backlog"]:
                state["peak_backlog"] = state["outstanding"]

    def _consumed() -> None:
        with backlog_lock:
            state["outstanding"] -= 1

    class InFlightTrackingBarClient:
        """Records PEAK concurrently-in-flight bars() calls AND marks each completed result as
        produced (held in memory until the main thread writes it)."""

        def __init__(self) -> None:
            self.fetched: list[str] = []
            self._lock = threading.Lock()
            self._in_flight = 0
            self.peak_in_flight = 0

        def bars(self, symbol: str, start_date: str | None = None) -> list[dict]:
            with self._lock:
                self.fetched.append(symbol)
                self._in_flight += 1
                if self._in_flight > self.peak_in_flight:
                    self.peak_in_flight = self._in_flight
            try:
                return list(bars_by_sym.get(symbol, []))
            finally:
                with self._lock:
                    self._in_flight -= 1
                _produced()  # result now sits in memory, awaiting a (slow) write

    class SlowWriterStore:
        """Wraps the real DuckDB store; the write is SLOW so completed fetches would back up in memory
        if the ingest submitted everything at once. Each write consumes one backlog item."""

        def __init__(self, inner: DuckDbFeatureStore) -> None:
            self._inner = inner

        def catalog(self):
            return self._inner.catalog()

        def read_membership(self):
            return self._inner.read_membership()

        def write_features_bulk(self, items):
            time.sleep(0.005)  # serial writer is the bottleneck (single DuckDB connection)
            return self._inner.write_features_bulk(items)

        def upsert_membership(self, rows):
            _consumed()  # one ticker fully handled (written + dropped from memory)
            return self._inner.upsert_membership(rows)

        def read_features(self, symbol):
            return self._inner.read_features(symbol)

        def read_panel(self, symbols, feature):
            return self._inner.read_panel(symbols, feature)

        def close(self):
            return self._inner.close()

    inner = DuckDbFeatureStore(str(tmp_path / "wh.duckdb"))
    store = SlowWriterStore(inner)
    bar_client = InFlightTrackingBarClient()
    svc = HistoricalEodIngest(bar_client, store, ManyUniverse())

    result = svc.ingest(max_workers=max_workers)

    # (1) PEAK in-flight fetches bounded by the window — NOT by the (much larger) universe size.
    assert bar_client.peak_in_flight <= window, (
        f"peak in-flight {bar_client.peak_in_flight} exceeded window {window} "
        f"(unbounded submit would approach {n})"
    )
    # (2) THE MEMORY BUG: peak result backlog bounded by the window. Submit-all lets this grow toward
    # n (60) because fast fetchers outrun the serial writer; the bounded window caps it at `window`.
    assert state["peak_backlog"] <= window, (
        f"peak result backlog {state['peak_backlog']} exceeded window {window} — completed "
        f"ticker-bar results accumulated unbounded (the OOM bug); should be ≤ {window}"
    )
    # All tickers still ingested with correct counts (window changes memory, not the result).
    assert sorted(bar_client.fetched) == sorted(syms)
    assert result["tickers"] == n
    assert result["rows"] == expected_rows
    assert result["empty"] == 0
    assert result["members"] == n
    assert result["skipped"] == 0
    for s in syms:
        assert len(store.read_features(s)) == len(bars_by_sym[s])
    store.close()


def test_subset_ingest_touches_only_named_ticker(tmp_path) -> None:
    store = DuckDbFeatureStore(str(tmp_path / "wh.duckdb"))
    bar_client = FakeBarClient()
    svc = HistoricalEodIngest(bar_client, store, FakeUniverseSource())

    result = svc.ingest(tickers=["AAA"])

    assert bar_client.fetched == ["AAA"]
    assert result["tickers"] == 1
    # only AAA in the warehouse
    assert [c[0] for c in store.catalog()] == ["AAA"]
    # membership still recorded for AAA (from the universe metadata), none for the others
    members = {r[0] for r in store.read_membership()}
    assert members == {"AAA"}
    store.close()
