from __future__ import annotations

"""
phone/ws_scene_bus.py - WebSocket "scene bus" for FRIDAY AR + ecosystem sync.

This is a lightweight broadcast server:
- Clients connect and receive the current AR scene state immediately.
- Clients can send {"type":"patch","patch":{...},"ts":<ms>} to apply a patch.
- Last-write-wins: we trust the patch ts if provided, else server time.

Security:
- For Phase AR-1 we allow LAN/Tailscale access and rely on the existing passphrase
  trust model for command channels. For production hardening, we can require an
  encrypted hello using the phone link key.
"""

import asyncio
import json
import time
from dataclasses import dataclass
from typing import Any, Set

import websockets
from websockets.server import WebSocketServerProtocol

from modules.ar_scene import load_state, apply_patch


def _now_ms() -> int:
    return int(time.time() * 1000)


@dataclass
class Client:
    ws: WebSocketServerProtocol


class SceneBus:
    def __init__(self) -> None:
        self._clients: Set[WebSocketServerProtocol] = set()
        self._lock = asyncio.Lock()

    async def broadcast(self, obj: dict[str, Any]) -> None:
        msg = json.dumps(obj)
        dead: list[WebSocketServerProtocol] = []
        async with self._lock:
            for ws in list(self._clients):
                try:
                    await ws.send(msg)
                except Exception:
                    dead.append(ws)
            for ws in dead:
                try:
                    self._clients.discard(ws)
                except Exception:
                    pass

    async def handler(self, ws: WebSocketServerProtocol) -> None:
        async with self._lock:
            self._clients.add(ws)
        # Send hello + current state
        try:
            await ws.send(json.dumps({"type": "hello", "ts": _now_ms(), "state": load_state()}))
        except Exception:
            return

        try:
            async for raw in ws:
                try:
                    msg = json.loads(raw)
                except Exception:
                    continue
                if not isinstance(msg, dict):
                    continue
                mtype = str(msg.get("type") or "")
                if mtype == "ping":
                    await ws.send(json.dumps({"type": "pong", "ts": _now_ms()}))
                    continue
                # Achievement broadcast from other device
                if mtype == "achievement":
                    ach = msg.get("achievement", {})
                    await self.broadcast({"type": "achievement", "ach": ach, "ts": _now_ms()})
                    continue
                if mtype != "patch":
                    continue
                patch = msg.get("patch")
                if not isinstance(patch, dict):
                    continue
                # Apply patch and broadcast authoritative new state
                r = apply_patch(patch)
                await self.broadcast({"type": "state", "ts": _now_ms(), "state": r.state})
        finally:
            async with self._lock:
                self._clients.discard(ws)


async def run(host: str = "0.0.0.0", port: int = 5050) -> None:
    bus = SceneBus()
    async with websockets.serve(bus.handler, host, int(port), max_size=512_000):
        await asyncio.Future()


def run_in_thread(host: str = "0.0.0.0", port: int = 5050) -> None:
    def _t():
        asyncio.run(run(host=host, port=port))

    import threading

    threading.Thread(target=_t, daemon=True).start()


# ── External Broadcast ──────────────────────────────────────────────────

async def broadcast_to_clients_async(obj: dict) -> None:
    """Broadcast a message through the bus from an async context."""
    try:
        msg = json.dumps(obj)
        async with websockets.connect("ws://127.0.0.1:5050") as ws:
            await ws.send(msg)
    except Exception:
        pass


def broadcast_to_clients(obj: dict) -> None:
    """Broadcast message to all connected WebSocket clients (thread-safe).

    S4: the old body always called asyncio.run(), which raises RuntimeError when a
    loop is already running — every broadcast from FastAPI handlers silently died.
    Detect the running loop and schedule instead.
    """
    try:
        try:
            loop = asyncio.get_running_loop()
        except RuntimeError:
            loop = None
        if loop is not None:
            loop.create_task(broadcast_to_clients_async(obj))
        else:
            asyncio.run(broadcast_to_clients_async(obj))
    except Exception:
        pass