from __future__ import annotations

import asyncio
from dataclasses import dataclass
from datetime import datetime
from typing import Any, Dict, Set, Tuple

from app.streaming.live_store import live_store
from app.services.price_simulator import gold_simulator


@dataclass(frozen=True)
class FeedKey:
    symbol: str
    timeframe: str


class LocalFeedProvider:
    """In-memory price feed backed by the gold price simulator."""

    def __init__(self) -> None:
        self._tasks: Dict[FeedKey, asyncio.Task] = {}
        self._pinned: Set[FeedKey] = set()
        self._lock = asyncio.Lock()
        self._active_counts: Dict[FeedKey, int] = {}

    def get_status(self) -> list[dict]:
        out: list[dict] = []
        for key, task in self._tasks.items():
            hist = live_store.get_history(key.symbol, key.timeframe)
            last_ts = hist[-1]["time"] if hist else None
            last_iso = None
            if isinstance(last_ts, (int, float)):
                try:
                    last_iso = datetime.utcfromtimestamp(int(last_ts)).isoformat() + "Z"
                except Exception:
                    last_iso = None
            out.append(
                {
                    "symbol": key.symbol,
                    "timeframe": key.timeframe,
                    "subscribers": self._active_counts.get(key, 0),
                    "last_event_time": last_iso,
                }
            )
        return out

    async def subscribe(self, symbol: str, timeframe: str = "1m") -> Tuple[asyncio.Queue, Any]:
        if timeframe != "1m":
            raise ValueError("LocalFeedProvider currently supports timeframe '1m' only")
        key = FeedKey(symbol=symbol.upper().replace("/", ""), timeframe=timeframe)
        queue: asyncio.Queue = asyncio.Queue(maxsize=100)
        live_store.subscribe(key.symbol, key.timeframe, queue)
        async with self._lock:
            self._active_counts[key] = self._active_counts.get(key, 0) + 1
            if key not in self._tasks:
                self._tasks[key] = asyncio.create_task(self._run_simulator_poller(key))

        async def _unsubscribe() -> None:
            live_store.unsubscribe(key.symbol, key.timeframe, queue)
            async with self._lock:
                self._active_counts[key] = max(0, self._active_counts.get(key, 0) - 1)
                if self._active_counts.get(key, 0) == 0 and key not in self._pinned:
                    task = self._tasks.pop(key, None)
                    if task:
                        task.cancel()
                    self._active_counts.pop(key, None)

        return queue, _unsubscribe

    async def ensure_stream(self, symbol: str, timeframe: str = "1m") -> None:
        if timeframe != "1m":
            raise ValueError("LocalFeedProvider currently supports timeframe '1m' only")
        key = FeedKey(symbol=symbol.upper().replace("/", ""), timeframe=timeframe)
        async with self._lock:
            self._pinned.add(key)
            if key not in self._tasks:
                self._tasks[key] = asyncio.create_task(self._run_simulator_poller(key))

    async def release_stream(self, symbol: str, timeframe: str = "1m") -> None:
        key = FeedKey(symbol=symbol.upper().replace("/", ""), timeframe=timeframe)
        async with self._lock:
            self._pinned.discard(key)
            if self._active_counts.get(key, 0) == 0:
                task = self._tasks.pop(key, None)
                if task:
                    task.cancel()

    async def _run_simulator_poller(self, key: FeedKey) -> None:
        symbol = key.symbol
        simulator = gold_simulator
        if simulator.current_price < 2400 or simulator.current_price > 2900:
            simulator.current_price = 2650.0
            simulator.base_price = 2650.0

        last_ts: int | None = None
        poll_interval = 3

        while True:
            try:
                candle = simulator.get_live_candle(interval="1min")
                tsec = candle.time

                if last_ts is None or tsec > last_ts:
                    dt = datetime.utcfromtimestamp(tsec)
                    evt = {
                        "symbol": symbol,
                        "timeframe": key.timeframe,
                        "open_time": dt.isoformat(),
                        "close_time": dt.isoformat(),
                        "open": candle.open,
                        "high": candle.high,
                        "low": candle.low,
                        "close": candle.close,
                        "volume": candle.volume or 0.0,
                        "is_closed": True,
                        "source": "local_simulator",
                    }
                    try:
                        live_store.ingest_bar(
                            symbol=symbol,
                            timeframe="1m",
                            bar={
                                "time": tsec,
                                "open": evt["open"],
                                "high": evt["high"],
                                "low": evt["low"],
                                "close": evt["close"],
                                "volume": evt["volume"],
                            },
                        )
                    except Exception:
                        pass

                    last_ts = tsec

                await asyncio.sleep(poll_interval)
            except asyncio.CancelledError:
                break
            except Exception:
                await asyncio.sleep(poll_interval)


local_feed = LocalFeedProvider()