"""实例 embed 壳:后台定时 tick + SSE 通知前端拉 JSON 快照(对齐中控 dashboard)."""
from __future__ import annotations

import json
import os
import queue
import threading
from collections.abc import Iterator
from typing import Any, Callable

from flask import Flask, Response, stream_with_context

INSTANCE_LIVE_TICK_SEC = float(os.getenv("INSTANCE_LIVE_TICK_SEC", "5"))
INSTANCE_SSE_HEARTBEAT_SEC = float(os.getenv("INSTANCE_SSE_HEARTBEAT_SEC", "25"))


class InstanceLivePush:
    def __init__(self) -> None:
        self._lock = threading.Lock()
        self.version = 0
        self._subscribers: list[queue.Queue[str | None]] = []
        self._stop = threading.Event()
        self._thread: threading.Thread | None = None

    def start(self) -> None:
        if self._thread and self._thread.is_alive():
            return
        self._stop.clear()
        self._thread = threading.Thread(target=self._loop, daemon=True, name="instance-live-push")
        self._thread.start()

    def stop(self) -> None:
        self._stop.set()
        self._broadcast(close=True)

    def tick(self, reason: str = "poll") -> int:
        with self._lock:
            self.version += 1
            ver = self.version
        payload = json.dumps({"live_version": ver, "reason": reason}, ensure_ascii=False)
        self._broadcast(payload)
        return ver

    def event_dict(self) -> dict[str, Any]:
        return {"live_version": self.version, "tick_sec": INSTANCE_LIVE_TICK_SEC}

    def _loop(self) -> None:
        while not self._stop.is_set():
            self.tick("poll")
            if self._stop.wait(INSTANCE_LIVE_TICK_SEC):
                break

    def _broadcast(self, event: str | None = None, *, close: bool = False) -> None:
        with self._lock:
            subs = list(self._subscribers)
        dead: list[queue.Queue[str | None]] = []
        for q in subs:
            try:
                q.put_nowait(None if close else event)
            except Exception:
                dead.append(q)
        if dead:
            with self._lock:
                for q in dead:
                    if q in self._subscribers:
                        self._subscribers.remove(q)

    def _subscribe(self) -> queue.Queue[str | None]:
        q: queue.Queue[str | None] = queue.Queue(maxsize=16)
        with self._lock:
            self._subscribers.append(q)
        return q

    def _unsubscribe(self, q: queue.Queue[str | None]) -> None:
        with self._lock:
            if q in self._subscribers:
                self._subscribers.remove(q)

    def iter_sse(self) -> Iterator[str]:
        q = self._subscribe()
        try:
            yield self._format_event(self.event_dict() | {"reason": "connect"})
            while True:
                try:
                    raw = q.get(timeout=INSTANCE_SSE_HEARTBEAT_SEC)
                except queue.Empty:
                    yield ": heartbeat\n\n"
                    continue
                if raw is None:
                    break
                yield f"event: live\ndata: {raw}\n\n"
        finally:
            self._unsubscribe(q)

    @staticmethod
    def _format_event(data: dict[str, Any]) -> str:
        return "event: live\ndata: " + json.dumps(data, ensure_ascii=False) + "\n\n"


instance_live_push = InstanceLivePush()


def notify_instance_balance_changed() -> int:
    """划转/兑换后通知 embed 壳拉最新资金快照."""
    return instance_live_push.tick("balance")


def register_instance_live_routes(app: Flask, login_required: Callable) -> None:
    instance_live_push.start()

    @login_required
    @app.route("/api/instance/live/stream")
    def api_instance_live_stream():
        return Response(
            stream_with_context(instance_live_push.iter_sse()),
            mimetype="text/event-stream",
            headers={
                "Cache-Control": "no-cache",
                "Connection": "keep-alive",
                "X-Accel-Buffering": "no",
            },
        )