"""WebSocket → Redis pub/sub (fan-out to API/SSE) + Postgres (latest ticker/book, trades, 1-s book samples, WS candles)."""

from __future__ import annotations

import asyncio
import json
import time
from typing import Any

import asyncpg
import redis.asyncio as aioredis

from ktrade.ingestion.backfill import upsert_candles
from ktrade.kraken.models import PairMeta
from ktrade.kraken.ws import KrakenWS
from ktrade.logging import get_logger

log = get_logger("ingest.stream")
CHANNEL = "ktrade:md"


class StreamIngestor:
    def __init__(self, pool: asyncpg.Pool, redis: aioredis.Redis, ws_url: str, pairs: dict[str, PairMeta], book_depth: int, ohlc_intervals: list[int]) -> None:
        self.pool = pool
        self.redis = redis
        self.pairs = pairs
        precision = {s: (p.pair_decimals, p.lot_decimals) for s, p in pairs.items()}
        self.ws = KrakenWS(ws_url, list(pairs), self.on_event, book_depth=book_depth, ohlc_intervals=ohlc_intervals, precision=precision)
        self._last_book_sample: dict[str, int] = {}
        self._iso = None
        self.counts: dict[str, int] = {}

    async def run(self) -> None:
        await asyncio.gather(self.ws.run(), self._health_loop())

    async def _health_loop(self) -> None:
        while True:
            await self.redis.set("ktrade:stream:health", json.dumps({
                "connected": self.ws.connected, "last_msg_age": round(time.monotonic() - self.ws.last_msg_at, 1) if self.ws.last_msg_at else None,
                "reconnects": self.ws.reconnects, "checksum_failures": self.ws.checksum_failures, "messages": self.ws.messages, "counts": self.counts, "ts": time.time(),
            }), ex=30)
            await asyncio.sleep(5)

    async def on_event(self, channel: str, mtype: str, data: list[dict[str, Any]], received_at: float) -> None:
        self.counts[channel] = self.counts.get(channel, 0) + 1
        await self.redis.publish(CHANNEL, json.dumps({"channel": channel, "type": mtype, "data": data, "received_at": received_at}))
        try:
            if channel == "ticker":
                await self._store_ticker(data, received_at)
            elif channel == "book":
                await self._store_book(data, received_at)
            elif channel == "trade":
                await self._store_trades(data, received_at)
            elif channel == "ohlc":
                await self._store_ohlc(data)
        except Exception as e:  # noqa: BLE001
            log.warning("ingest.stream.store_error", channel=channel, error=str(e))

    async def _store_ticker(self, data: list[dict[str, Any]], received_at: float) -> None:
        async with self.pool.acquire() as conn:
            for d in data:
                await conn.execute(
                    "INSERT INTO ticker_latest (symbol, payload, exchange_ts, received_at) VALUES ($1,$2,$3,$4) ON CONFLICT (symbol) DO UPDATE SET payload=EXCLUDED.payload, exchange_ts=EXCLUDED.exchange_ts, received_at=EXCLUDED.received_at",
                    d["symbol"], d, received_at, received_at,
                )

    async def _store_book(self, data: list[dict[str, Any]], received_at: float) -> None:
        now = int(received_at)
        async with self.pool.acquire() as conn:
            for d in data:
                await conn.execute(
                    "INSERT INTO book_latest (symbol, payload, exchange_ts, received_at, valid) VALUES ($1,$2,$3,$4,$5) ON CONFLICT (symbol) DO UPDATE SET payload=EXCLUDED.payload, exchange_ts=EXCLUDED.exchange_ts, received_at=EXCLUDED.received_at, valid=EXCLUDED.valid",
                    d["symbol"], d, d.get("timestamp"), received_at, bool(d.get("valid", True)),
                )
                if self._last_book_sample.get(d["symbol"]) != now and d.get("best_bid"):
                    self._last_book_sample[d["symbol"]] = now
                    await conn.execute(
                        "INSERT INTO book_samples (symbol, ts, best_bid, best_ask, spread_bps, bid_depth, ask_depth, imbalance) VALUES ($1,$2,$3,$4,$5,$6,$7,$8) ON CONFLICT DO NOTHING",
                        d["symbol"], now, d["best_bid"], d["best_ask"], d.get("spread_bps"), sum(q for _, q in d["bids"]), sum(q for _, q in d["asks"]), d.get("imbalance"),
                    )

    async def _store_trades(self, data: list[dict[str, Any]], received_at: float) -> None:
        rows = []
        for d in data:
            ts = _parse_ts(d.get("timestamp"))
            rows.append((d["symbol"], int(d["trade_id"]), ts, float(d["price"]), float(d["qty"]), d["side"], d.get("ord_type", "limit"), received_at))
        if rows:
            async with self.pool.acquire() as conn:
                await conn.executemany("INSERT INTO trades (symbol, trade_id, ts, price, volume, side, order_type, received_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8) ON CONFLICT DO NOTHING", rows)

    async def _store_ohlc(self, data: list[dict[str, Any]]) -> None:
        now = time.time()
        async with self.pool.acquire() as conn:
            for d in data:
                iv = int(d["interval"])
                ts = int(_parse_ts(d["interval_begin"]))
                closed = ts + iv * 60 <= now
                await upsert_candles(conn, d["symbol"], iv, [(ts, float(d["open"]), float(d["high"]), float(d["low"]), float(d["close"]), float(d.get("vwap", d["close"])), float(d["volume"]), int(d.get("trades", 0)), closed)], "ws")


def _parse_ts(s: str | None) -> float:
    """RFC3339 with up to nanosecond precision → unix seconds (float)."""
    if not s:
        return time.time()
    import re
    from datetime import datetime

    m = re.match(r"^(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2})(?:\.(\d+))?(Z|[+-]\d{2}:\d{2})?$", s)
    if not m:
        return datetime.fromisoformat(s.replace("Z", "+00:00")).timestamp()
    head, frac, tz = m.group(1), m.group(2) or "", m.group(3) or "Z"
    tz = "+00:00" if tz == "Z" else tz
    return datetime.fromisoformat(f"{head}.{frac[:6].ljust(6, '0')}{tz}").timestamp()