"""Connectivity messages must distinguish cached command errors from fresh loss.

These contracts run real status/overlay handlers and the real WebSocket reader.
The slow-state barrier pauses the actual DB save before SQL, leaving the reader
free to fence a late Mapping ACK without waiting for a database worker.
"""
from __future__ import annotations

import asyncio
import threading

import pytest
from fastapi import WebSocketDisconnect
from sqlalchemy import select

from hockey_data.models import VmixDevice
from tests.test_build192_mapping_delivery import DEVICE, profile, rows


async def _warm(tmp_path, *, connected=True):
    database, hub, socket, user, catalog, assignment = await profile(tmp_path, rows(8))
    await hub.receive_status(DEVICE, {"type": "heartbeat", "vmix": {"connected": True}})
    applied = await hub.apply_mapping_to_device(DEVICE, source_codes={"sample"}, only_changed=True)
    assert applied["ok"] and applied["applied"] == 8 and len(hub._mapping_value_cache) == 8
    if not connected:
        await hub.receive_status(DEVICE, {"type": "heartbeat", "vmix": {
            "connected": False, "error": "dial tcp: connection refused"}})
        assert not hub._mapping_value_cache
    return database, hub, socket, user, catalog, assignment


async def _blocked_reader(hub, socket, monkeypatch):
    release = threading.Event()
    entered = asyncio.Event()
    loop = asyncio.get_running_loop()
    original_save = hub._save_device_status
    blocked = False

    def slow_save(*args):
        nonlocal blocked
        if not blocked:
            blocked = True
            loop.call_soon_threadsafe(entered.set)
            assert release.wait(5), "test state DB worker was not released"
        return original_save(*args)

    monkeypatch.setattr(hub, "_save_device_status", slow_save)
    hub._status_saved_at.pop(DEVICE, None)
    reader = asyncio.create_task(hub.read_agent_messages(DEVICE, socket))
    await socket.incoming.put({"type": "heartbeat", "vmix": {"connected": True}})
    try:
        await asyncio.wait_for(entered.wait(), 2)
    except BaseException:
        release.set()
        reader.cancel()
        await asyncio.gather(reader, return_exceptions=True)
        raise
    return reader, release


async def _stop_reader(hub, socket, reader, release):
    release.set()
    if not reader.done():
        await socket.incoming.put(WebSocketDisconnect())
    outcome = await asyncio.gather(reader, return_exceptions=True)
    assert isinstance(outcome[0], WebSocketDisconnect), outcome
    await hub.stop_auto_refresh()


async def _reader_barrier(hub, socket, message):
    # Reading this matching ACK proves the previous status frame has passed
    # through the fast reader, even while the state worker remains blocked.
    request_id = "qa-connectivity-reader-barrier"
    future = asyncio.get_running_loop().create_future()
    hub._pending_commands[request_id] = (DEVICE, future)
    try:
        await socket.incoming.put(message)
        await socket.incoming.put({"type": "command.ack", "request_id": request_id, "ok": True})
        assert (await asyncio.wait_for(future, 1))["ok"] is True
    finally:
        hub._pending_commands.pop(request_id, None)


@pytest.mark.parametrize("kind,error_location,error", [
    ("heartbeat", "nested", "HTTP 500"),
    ("vmix.status", "nested", "HTTP 404 Not Found"),
    ("heartbeat", "top", "HTTP 503"),
    ("vmix.status", "reason", "HTTP 400 Bad Request"),
])
def test_cached_command_http_failure_preserves_real_status_cache_and_next_mapping(tmp_path, kind, error_location, error):
    async def scenario():
        database, hub, socket, _, catalog, _ = await _warm(tmp_path)
        generation = hub._mapping_generations[DEVICE]
        cached = dict(hub._mapping_value_cache)
        message = {"type": kind, "vmix": {"connected": False}}
        if error_location == "nested":
            message["vmix"]["error"] = error
        else:
            message["error" if error_location == "top" else "reason"] = error
        try:
            await hub.receive_status(DEVICE, message)
            assert hub._device_status[DEVICE]["vmix_connected"] is True
            assert hub._mapping_generations[DEVICE] == generation
            assert hub._mapping_value_cache == cached
            with database.session() as session:
                device = session.scalar(select(VmixDevice).where(VmixDevice.device_uuid == DEVICE))
                assert device.vmix_connected is True
            before = len(socket.executed)
            unchanged = await hub.apply_mapping_to_device(DEVICE, source_codes={"sample"}, only_changed=True)
            assert unchanged["unchanged"] == 8 and len(socket.executed) == before
            catalog["sample.row.0"] = "new-after-command-error"
            changed = await hub.apply_mapping_to_device(DEVICE, source_codes={"sample"}, only_changed=True)
            assert changed["ok"] and changed["applied"] == 1 and changed["unchanged"] == 7
            assert socket.executed[-1]["Value"] == "new-after-command-error"
            assert hub._mapping_generations[DEVICE] == generation
        finally:
            await hub.stop_auto_refresh()
    asyncio.run(scenario())


@pytest.mark.parametrize("kind,error", [
    ("heartbeat", "dial tcp: connection refused"),
    ("vmix.status", "context deadline exceeded"),
    ("heartbeat", None),
    ("vmix.inventory", "HTTP 500"),
    ("overlay_status", "HTTP 500"),
])
def test_fresh_probe_transport_or_unknown_loss_invalidates_real_status_and_cache(tmp_path, kind, error):
    async def scenario():
        _, hub, _, _, _, _ = await _warm(tmp_path)
        generation = hub._mapping_generations[DEVICE]
        vmix = {"connected": False}
        if error is not None:
            vmix["error"] = error
        try:
            await hub.receive_status(DEVICE, {"type": kind, "vmix": vmix})
            assert hub._device_status[DEVICE]["vmix_connected"] is False
            assert hub._mapping_generations[DEVICE] > generation
            assert not hub._mapping_value_cache
        finally:
            await hub.stop_auto_refresh()
    asyncio.run(scenario())


@pytest.mark.parametrize("initial,value", [(True, "false"), (False, "true"), (False, "false"), (True, 0), (False, 1), (True, None)])
def test_status_connectivity_requires_actual_boolean_not_truthiness(tmp_path, initial, value):
    async def scenario():
        _, hub, _, _, _, _ = await _warm(tmp_path, connected=initial)
        generation, cached = hub._mapping_generations[DEVICE], dict(hub._mapping_value_cache)
        recoveries = []
        hub._request_mapping_recovery = recoveries.append
        try:
            await hub.receive_status(DEVICE, {"type": "vmix.status", "vmix": {"connected": value}})
            assert hub._device_status[DEVICE]["vmix_connected"] is initial
            assert hub._mapping_generations[DEVICE] == generation and hub._mapping_value_cache == cached
            assert recoveries == [], "A malformed connectivity value must not revive a disconnected Agent"
        finally:
            await hub.stop_auto_refresh()
    asyncio.run(scenario())


@pytest.mark.parametrize("initial,value", [(True, "false"), (False, "true"), (False, "false"), (True, 0), (False, 1), (True, None)])
def test_actual_overlay_handler_ignores_non_boolean_connectivity_without_fabricating_tally(tmp_path, initial, value):
    async def scenario():
        _, hub, _, _, _, _ = await _warm(tmp_path, connected=initial)
        hub._runtime_overlay_state[DEVICE] = {"1": {"input": "Previous input"}}
        hub._runtime_overlay_authoritative[DEVICE] = initial
        previous = dict(hub._runtime_overlay_state[DEVICE])
        generation, cached = hub._mapping_generations[DEVICE], dict(hub._mapping_value_cache)
        recoveries = []
        hub._request_mapping_recovery = recoveries.append
        try:
            await hub.receive_overlay_state(DEVICE, {"type": "vmix.overlay_state", "vmix_connected": value,
                "overlays": {"1": {"input": "Untrusted new input"}}})
            assert hub._device_status[DEVICE]["vmix_connected"] is initial
            assert hub._runtime_overlay_state[DEVICE] == previous
            assert hub._runtime_overlay_authoritative[DEVICE] is initial
            assert hub._mapping_generations[DEVICE] == generation and hub._mapping_value_cache == cached
            assert recoveries == []
        finally:
            await hub.stop_auto_refresh()
    asyncio.run(scenario())


def test_actual_overlay_false_is_fresh_loss_even_when_error_is_http500(tmp_path):
    async def scenario():
        _, hub, _, _, _, _ = await _warm(tmp_path)
        hub._runtime_overlay_state[DEVICE] = {"1": {"input": "Score"}}
        hub._runtime_overlay_authoritative[DEVICE] = True
        generation = hub._mapping_generations[DEVICE]
        try:
            await hub.receive_overlay_state(DEVICE, {"type": "vmix.overlay_state", "vmix_connected": False,
                "error": "HTTP 500", "overlays": {}})
            assert hub._device_status[DEVICE]["vmix_connected"] is False
            assert hub._mapping_generations[DEVICE] > generation and not hub._mapping_value_cache
            assert DEVICE not in hub._runtime_overlay_state and hub._runtime_overlay_authoritative[DEVICE] is False
        finally:
            await hub.stop_auto_refresh()
    asyncio.run(scenario())


@pytest.mark.parametrize("kind", ["heartbeat", "vmix.status"])
def test_reader_does_not_fence_cached_http_command_failure_while_state_db_is_blocked(tmp_path, monkeypatch, kind):
    async def scenario():
        _, hub, socket, _, catalog, _ = await _warm(tmp_path)
        generation, cached = hub._mapping_generations[DEVICE], dict(hub._mapping_value_cache)
        reader, release = await _blocked_reader(hub, socket, monkeypatch)
        try:
            await _reader_barrier(hub, socket, {"type": kind, "vmix": {"connected": False, "error": "HTTP 500"}})
            assert not release.is_set()
            assert hub._mapping_generations[DEVICE] == generation and hub._mapping_value_cache == cached
            assert hub._device_status[DEVICE]["vmix_connected"] is True
            catalog["sample.row.0"] = "still-delivered-while-state-worker-is-blocked"
            result = await asyncio.wait_for(hub.apply_mapping_to_device(DEVICE, source_codes={"sample"}, only_changed=True), 2)
            assert result["ok"] and result["applied"] == 1 and result["unchanged"] == 7
            assert socket.executed[-1]["Value"] == "still-delivered-while-state-worker-is-blocked"
            assert not release.is_set() and hub._mapping_generations[DEVICE] == generation
        finally:
            await _stop_reader(hub, socket, reader, release)
    asyncio.run(scenario())


@pytest.mark.parametrize("kind,value,after_loss", [
    ("vmix.status", 0, False),
    ("vmix.status", "true", True),
    ("vmix.status", 1, True),
    ("vmix.overlay_state", "false", True),
])
def test_reader_ignores_non_boolean_connectivity_without_loss_or_premature_recovery(tmp_path, monkeypatch, kind, value, after_loss):
    async def scenario():
        _, hub, socket, _, _, _ = await _warm(tmp_path)
        reader, release = await _blocked_reader(hub, socket, monkeypatch)
        recoveries = []
        hub._request_mapping_recovery = recoveries.append
        try:
            if after_loss:
                await _reader_barrier(hub, socket, {"type": "heartbeat", "vmix": {"connected": False}})
                assert not hub._mapping_value_cache
            generation, cached = hub._mapping_generations[DEVICE], dict(hub._mapping_value_cache)
            message = {"type": kind, "vmix_connected": value, "overlays": {}} if kind == "vmix.overlay_state" else {
                "type": kind, "vmix": {"connected": value}}
            await _reader_barrier(hub, socket, message)
            assert not release.is_set()
            assert hub._mapping_generations[DEVICE] == generation and hub._mapping_value_cache == cached
            assert recoveries == [], "Unknown connectivity must not start recovery before a fresh positive probe"
        finally:
            await _stop_reader(hub, socket, reader, release)
    asyncio.run(scenario())


@pytest.mark.parametrize("loss", [
    {"type": "vmix.inventory", "vmix": {"connected": False, "error": "HTTP 500"}, "inventory": {}},
    {"type": "vmix.overlay_state", "vmix_connected": False, "error": "HTTP 500", "overlays": {}},
    {"type": "heartbeat", "vmix": {"connected": False, "error": "dial tcp: connection refused"}},
    {"type": "vmix.status", "vmix": {"connected": False, "error": "context deadline exceeded"}},
    {"type": "heartbeat", "vmix": {"connected": False}},
], ids=["fresh-inventory-http500", "fresh-overlay-http500", "refused", "timeout", "legacy-unknown"])
def test_reader_fences_fresh_loss_before_state_db_and_late_ack_cannot_restore_mapping_cache(tmp_path, monkeypatch, loss):
    async def scenario():
        _, hub, socket, _, catalog, _ = await _warm(tmp_path)
        generation = hub._mapping_generations[DEVICE]
        reader, release = await _blocked_reader(hub, socket, monkeypatch)
        socket.auto_ack = False
        socket.command_arrived.clear()
        catalog["sample.row.0"] = "late-ack-must-not-confirm"
        task = asyncio.create_task(hub.apply_mapping_to_device(DEVICE, source_codes={"sample"}, only_changed=True))
        try:
            await asyncio.wait_for(socket.command_arrived.wait(), 2)
            packet = socket.mapping_packets[-1]
            await socket.incoming.put(loss)
            # The reader sees loss before this ACK. The actual status saver is
            # still blocked in a separate thread and cannot do the invalidation.
            await socket.incoming.put(socket.payload(packet))
            result = await asyncio.wait_for(asyncio.shield(task), 2)
            assert not release.is_set()
            assert hub._device_status[DEVICE]["vmix_connected"] is True
            assert hub._mapping_generations[DEVICE] > generation
            assert not result["ok"] and result["applied"] == 0 and not hub._mapping_value_cache
            assert not hub._pending_commands
        finally:
            release.set()
            if not task.done():
                task.cancel()
            await asyncio.gather(task, return_exceptions=True)
            await _stop_reader(hub, socket, reader, release)
    asyncio.run(scenario())