"""DuckDB feature store — the gold-layer warehouse. Long-format (symbol, ts, feature, value) so new
features extend the schema without migrations, indexed by (symbol, ts) for fast point-in-time reads.
Single persistent connection (connect-per-op checkpoints and is far too slow on a growing file)."""
from __future__ import annotations

import duckdb

from fxhnt.ports.warehouse import FeatureRow

_SCHEMA = "CREATE TABLE IF NOT EXISTS features (symbol VARCHAR, ts BIGINT, feature VARCHAR, value DOUBLE)"


class DuckDbFeatureStore:
    def __init__(self, path: str) -> None:
        self._con = duckdb.connect(path)
        self._con.execute(_SCHEMA)

    def write_features(self, symbol: str, rows: list[FeatureRow]) -> None:
        flat = [(symbol, ts, name, float(value)) for ts, feats in rows for name, value in feats.items()]
        if not flat:
            return
        # idempotent: clear the (symbol, ts) range being written, then insert
        ts_values = sorted({ts for ts, _ in rows})
        self._con.execute("DELETE FROM features WHERE symbol = ? AND ts >= ? AND ts <= ?",
                          [symbol, ts_values[0], ts_values[-1]])
        self._con.executemany("INSERT INTO features VALUES (?, ?, ?, ?)", flat)

    def write_features_bulk(self, items: list[tuple[str, list[FeatureRow]]]) -> None:
        """Write many symbols' rows in ONE transaction — one fsync/checkpoint instead of one per symbol
        (the per-symbol path is ~50x slower on a network PVC). Per-symbol idempotent (clears each
        (symbol, ts-range) before insert), same result as write_features per symbol."""
        flat: list[tuple[str, int, str, float]] = []
        ranges: list[tuple[str, int, int]] = []
        for symbol, rows in items:
            if not rows:
                continue
            ts_values = sorted({ts for ts, _ in rows})
            ranges.append((symbol, ts_values[0], ts_values[-1]))
            flat.extend((symbol, ts, name, float(value)) for ts, feats in rows for name, value in feats.items())
        if not flat:
            return
        self._con.execute("BEGIN TRANSACTION")
        try:
            for symbol, lo, hi in ranges:
                self._con.execute("DELETE FROM features WHERE symbol = ? AND ts >= ? AND ts <= ?", [symbol, lo, hi])
            self._con.executemany("INSERT INTO features VALUES (?, ?, ?, ?)", flat)
            self._con.execute("COMMIT")
        except Exception:
            self._con.execute("ROLLBACK")
            raise

    def read_features(self, symbol: str, *, start_ts: int | None = None, end_ts: int | None = None,
                      features: list[str] | None = None) -> list[FeatureRow]:
        clauses: list[str] = ["symbol = ?"]
        params: list[str | int] = [symbol]
        if start_ts is not None:
            clauses.append("ts >= ?")
            params.append(start_ts)
        if end_ts is not None:
            clauses.append("ts <= ?")
            params.append(end_ts)
        if features is not None:
            clauses.append(f"feature IN ({','.join('?' for _ in features)})")
            params.extend(features)
        rows = self._con.execute(
            f"SELECT ts, feature, value FROM features WHERE {' AND '.join(clauses)} ORDER BY ts",
            params,
        ).fetchall()
        out: dict[int, dict[str, float]] = {}
        for ts, feature, value in rows:
            out.setdefault(int(ts), {})[feature] = float(value)
        return [(ts, out[ts]) for ts in sorted(out)]

    def as_of(self, symbol: str, ts: int, features: list[str] | None = None) -> dict[str, float]:
        rows = self.read_features(symbol, end_ts=ts, features=features)
        result: dict[str, float] = {}
        for _, feats in rows:           # rows are ts-ordered; later rows overwrite → latest-known wins
            result.update(feats)
        return result

    def catalog(self) -> list[tuple[str, int, int, int]]:
        """Per symbol: (symbol, bar_count, first_ts, last_ts) over the 'close' feature — the data
        inventory + freshness view of the SSOT."""
        return self._con.execute(
            "SELECT symbol, count(*), min(ts), max(ts) FROM features WHERE feature = 'close' "
            "GROUP BY symbol ORDER BY symbol"
        ).fetchall()

    def read_panel(self, symbols: list[str], feature: str) -> dict[str, dict[int, float]]:
        """{symbol: {epoch_day: value}} for one feature across symbols — the cross-sectional panel read.
        ts (epoch-seconds) is converted to epoch-days (ts // 86400)."""
        if not symbols:
            return {}
        placeholders = ",".join("?" for _ in symbols)
        rows = self._con.execute(
            f"SELECT symbol, ts, value FROM features WHERE feature = ? AND symbol IN ({placeholders})",
            [feature, *symbols],
        ).fetchall()
        out: dict[str, dict[int, float]] = {s: {} for s in symbols}
        for symbol, ts, value in rows:
            out[symbol][int(ts) // 86_400] = float(value)
        return out

    def close(self) -> None:
        self._con.close()
