"""Direct and portal browsers recover state without widening camera access."""

import asyncio
import json
from contextlib import suppress
from types import SimpleNamespace

import pytest

from fluksio.api import deps
from fluksio.api.routes import flows
from fluksio.cloud.connector import CloudConnector
from fluksio.flow import events
from fluksio.flow.events import QUEUE_SIZE, EventBus


@pytest.mark.parametrize("tunnel", [False, True])
@pytest.mark.parametrize("burst", [0, QUEUE_SIZE + 1])
def test_a_browser_catches_changes_during_its_snapshot(monkeypatch, tunnel, burst):
    async def run():
        bus = EventBus()
        bus.bind(asyncio.get_running_loop())
        monkeypatch.setattr(events, "event_bus", bus)
        monkeypatch.setattr(flows, "event_bus", bus)
        user = SimpleNamespace(id="viewer", is_superuser=False)
        for module in (flows, deps):
            monkeypatch.setattr(module, "user_from_token", lambda *a, **kw: user)
            monkeypatch.setattr(
                module, "hidden_from", lambda *a, **kw: frozenset({"f.camera"})
            )
        monkeypatch.setattr(flows, "panel_scope", lambda *a: None)
        monkeypatch.setattr(flows, "socket_token", lambda *a: "test-token")

        async def immediate(function, *args):
            return function(*args)

        monkeypatch.setattr(flows, "run_in_threadpool", immediate)
        current = {"f.flag": {"value": 0}, "f.camera": {"value": "private"}}
        controller = SimpleNamespace(
            values=lambda: dict(current),
            node_statuses=lambda: [],
            issues=[],
            paused_flows=lambda: [],
            placer=None,
        )
        app = SimpleNamespace(state=SimpleNamespace(flow_controller=controller))
        received = []
        client_values = {}
        caught_up = asyncio.Event()
        drained = asyncio.Event()

        class Socket:
            headers = {}

            async def accept(self):
                pass

            async def receive(self):
                await asyncio.Future()

            async def send(self, text):
                frame = json.loads(text)
                if frame["op"] == "ws_msg":
                    await self.send_text(frame["text"])

            async def send_text(self, text):
                payload = json.loads(text)
                received.append(payload)
                if payload["type"] == "snapshot":
                    assert "f.camera" not in payload["values"]
                    client_values.update(payload["values"])
                    if payload["values"]["f.flag"]["value"] == 1:
                        caught_up.set()
                    elif len(received) == 1:
                        # The state was read, but its frame is still sending.
                        # This value changes only once, then a burst loses its
                        # event and leaves only a snapshot able to recover it.
                        current["f.flag"] = {"value": 1}
                        bus.publish(
                            {"type": "message_value", "name": "f.flag", "value": 1}
                        )
                        for i in range(burst):
                            current["f.noise"] = {"value": i}
                            bus.publish(
                                {"type": "message_value", "name": "f.noise", "value": i}
                            )
                        # Recovery must forward retained critical events too.
                        bus.publish({"type": "node_error", "node": "f.example"})
                        await asyncio.sleep(0)
                for event in payload.get("events", [payload]):
                    if event["type"] == "message_value":
                        client_values[event["name"]] = {"value": event["value"]}
                    if event.get("name") == "f.flag" and event.get("value") == 1:
                        caught_up.set()
                    if event["type"] == "node_error":
                        drained.set()

        socket = Socket()
        socket.app = app
        if tunnel:
            action = CloudConnector(app)._stream(
                socket,
                {"id": "review", "path": "/api/v1/flows/ws", "query": "token=test"},
            )
        else:
            action = flows.flow_events(socket)
        task = asyncio.create_task(action)
        try:
            await asyncio.wait_for(caught_up.wait(), timeout=2)
            await asyncio.wait_for(drained.wait(), timeout=2)
        finally:
            task.cancel()
            with suppress(asyncio.CancelledError):
                await task
        assert sum(item["type"] == "snapshot" for item in received) == (
            2 if burst else 1
        )
        flattened = [
            event for payload in received for event in payload.get("events", [payload])
        ]
        assert sum(event["type"] == "node_error" for event in flattened) == 1
        if burst:
            # The client replaces values in arrival order. A snapshot ahead
            # of the retained backlog temporarily rolls values backwards.
            assert flattened[-1]["type"] == "snapshot"
        assert client_values == {k: v for k, v in current.items() if k != "f.camera"}
        assert not bus._subscribers

    asyncio.run(run())