"""Running a key rename against a live engine (sections 2.3 and 2.4 of
docs/private/id_refac_rename.md): the barrier, calling it off, the splice and
the runtime controls it keeps. Memory always, the end-to-end queue check with
FLUKSIO_TEST_REDIS_HOST (and optional PORT)."""

import asyncio
import json
import os
import threading
import time
import uuid
from collections.abc import Iterator
from pathlib import Path
from types import SimpleNamespace
from typing import Any

import pytest
from sqlmodel import Session, col, delete, select

from fluksio.core.config import settings
from fluksio.core.db import engine
from fluksio.flow import executor
from fluksio.flow import rename as renaming
from fluksio.flow.alerts import AlertManager, AlertsConfig, Channel
from fluksio.flow.controller import NODE_TYPES, FlowController, NodeType
from fluksio.flow.dashboards import DashboardDef, DashboardStore, WidgetDef
from fluksio.flow.events import EventBus
from fluksio.flow.executor import ExecutionService
from fluksio.flow.messages import DType, MessageSpec
from fluksio.flow.nodes import Node
from fluksio.flow.pipeline import EngineHeld
from fluksio.flow.queue import MemoryWorkQueue, RedisWorkQueue, WorkItem
from fluksio.flow.rename import (
    RenameConflict,
    RenameGate,
    Renamer,
    RenameRequest,
    plan_rename,
    read_journal,
)
from fluksio.flow.schemas import FlowDef, FlowOrigin, NodeDef
from fluksio.flow.state import MemoryState, RedisState
from fluksio.flow.store import FlowStore
from fluksio.models import EngineEvent
from tests.utils.rename import leftovers, stability

#: What the probes did, by node id, in order.
EFFECTS: list[str] = []
#: A probe set to wait says so here, then waits for GO.
RUNNING = threading.Event()
GO = threading.Event()


class Probe(Node):
    """A node reaching outside, so doing it twice would be a second command.
    What it did is written down, and with ``wait`` it takes as long as the
    test says."""

    idempotent = False

    def __init__(self, requires=(), provides=(), params=None, name=None) -> None:
        super().__init__(
            f=self._act, requires=requires, provides=provides, params=params, name=name
        )

    def _act(self, params: dict[str, Any], **inputs: Any) -> None:
        if params.get("wait"):
            RUNNING.set()
            assert GO.wait(10)
        EFFECTS.append(self.id)


def _probe(node_id: str, requires=(), provides=(), **params: Any) -> NodeDef:
    def specs(names) -> list[MessageSpec]:
        return [MessageSpec(name=name, dtype=DType.FLOAT) for name in names]

    return NodeDef(
        id=node_id,
        type="probe",
        requires=specs(requires),
        provides=specs(provides),
        params=params,
    )


@pytest.fixture
def world(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Iterator[SimpleNamespace]:
    """`heating`, a sensor and a slow actor on it; `lights` reading it;
    `heating_extra`, which only looks alike; `home`, a dashboard charting
    `heating.temp`. Ids are keys, as for a flow from before ids."""
    monkeypatch.setattr(settings, "RENAME_JOURNAL_FILE", tmp_path / "rename.json")
    monkeypatch.setattr(settings, "ALERTS_FILE", tmp_path / "alerts.json")
    monkeypatch.setitem(
        NODE_TYPES, "probe", NodeType(title="Probe", description="", cls=Probe)
    )
    EFFECTS.clear()
    RUNNING.clear()
    GO.clear()
    store = FlowStore(tmp_path / "flows")
    store.write_flow(
        FlowDef(
            name="heating",
            nodes=[
                _probe("sensor", provides=["temp"]),
                _probe("act", requires=["temp"], wait=True),
            ],
        )
    )
    store.write_flow(
        FlowDef(name="lights", nodes=[_probe("lamp", requires=["heating.temp"])])
    )
    store.write_flow(FlowDef(name="heating_extra", nodes=[_probe("x", provides=["x"])]))
    dashboards = DashboardStore(store)
    dashboards.write(
        DashboardDef(
            name="home",
            widgets=[
                WidgetDef(
                    id="c",
                    type="chart",
                    config={
                        "series": [{"message": "heating.temp"}],
                        "history": {"points": 500},
                    },
                )
            ],
        )
    )
    yield SimpleNamespace(root=store.root, store=store, dashboards=dashboards)
    # Let a probe still waiting go, so its thread ends with the test.
    GO.set()
    with Session(engine) as session:
        session.execute(delete(EngineEvent))
        session.commit()


async def _engine(world, state=None, queue=None, consume: bool = True):
    """The engine over *world*'s store, built and, with *consume*, claiming."""
    events = EventBus()
    events.bind(asyncio.get_running_loop())
    queue = queue if queue is not None else MemoryWorkQueue()
    execution = ExecutionService(queue, max_cascades=2)
    alerts = AlertManager(
        events,
        config=AlertsConfig(
            channels=[
                Channel(
                    name="board", kind="dashboard", config={"message": "heating.alert"}
                )
            ]
        ),
    )
    controller = FlowController(
        world.store,
        state=state if state is not None else MemoryState(),
        events=events,
        execution=execution,
        alerts=alerts,
    )
    controller.dashboards = world.dashboards
    controller.set_history_limits(world.dashboards.history_requirements())
    if consume:
        await controller.start()
    else:
        await controller.reload(wait=None)
    renamer = Renamer(controller, world.dashboards, RenameGate(), events, alerts=alerts)
    # What the build said is not for whoever listens next.
    await events.settle()
    return SimpleNamespace(
        controller=controller,
        renamer=renamer,
        events=events,
        queue=queue,
        alerts=alerts,
        state=controller.state,
    )


def _request(eng, new: str = "warmth", obj: str = "heating", op: str = "op-1"):
    kind = "flow" if obj in eng.controller.store.list_flows() else "dashboard"
    plan = plan_rename(
        eng.controller, eng.renamer.dashboards, kind, obj, new, AlertsConfig()
    )
    return RenameRequest(
        request_id=op,
        expected_name=plan.name,
        expected_version=plan.version,
        new_name=new,
        fingerprint=plan.fingerprint,
    )


async def _until(check, timeout: float = 10.0) -> None:
    deadline = time.monotonic() + timeout
    while not check():
        assert time.monotonic() < deadline, "timed out"
        await asyncio.sleep(0.02)


async def _heard(events: EventBus, queue: asyncio.Queue) -> list[dict]:
    await events.settle()
    found = []
    while not queue.empty():
        found.append(queue.get_nowait())
    return found


def _files(root: Path) -> dict[str, bytes]:
    return {
        str(path.relative_to(root)): path.read_bytes()
        for path in sorted(root.rglob("*"))
        if path.is_file() and ".git" not in path.parts
    }


def test_a_rename_moves_the_engine_and_keeps_its_controls(world):
    """Stopped and paused are kept everywhere, quarantine only off the flows
    it rebuilds (decision 6), and everything the engine keeps by name
    follows."""
    world.store.write_enabled("lights", False)

    async def scenario() -> None:
        eng = await _engine(world)
        controller, pipeline = eng.controller, eng.controller.pipeline
        try:
            eng.state.set("heating.temp", 20.5)
            controller.pause_flow("heating")
            controller.pause_flow("heating_extra")
            eng.queue.park("heating", WorkItem("cascade", "heating.sensor", "heating"))
            for key in ("heating.sensor:loop", "heating_extra.x:loop"):
                controller.supervisor._flows[key] = key.split(".")[0]
                controller.supervisor._resting[key] = key.split(".")[0]
            controller.loaded["heating.act"].last_error = "boom"
            eng.events.emits["heating.act"] = 3
            eng.events.digests["heating.cam"] = ["d1"]
            eng.alerts._last_sent["node_error:heating.act"] = 7.0
            request = _request(eng)

            async with eng.events.subscribe() as heard:
                result = await eng.renamer.rename("flow", "heating", request, "me")
                told = await _heard(eng.events, heard)

            assert (result.old_name, result.new_name) == ("heating", "warmth")
            assert sorted(controller.loaded) == [
                "heating_extra.x",
                "lights.lamp",
                "warmth.act",
                "warmth.sensor",
            ]
            assert controller.loaded["warmth.act"].last_error == "boom"
            assert controller.disabled == {"lights"}
            assert controller.paused_flows() == ["heating_extra", "warmth"]
            assert controller.quarantined == {"heating_extra"}
            assert controller.pipeline is pipeline
            lamp = pipeline.get_node_by_id("lights.lamp")
            assert [spec.name for spec in lamp.input_ports] == ["warmth.temp"]
            assert eng.state.get("warmth.temp") == 20.5
            assert "heating.temp" not in eng.state
            assert controller.history_limits == {"warmth.temp": 500}
            assert [item.node for item in eng.queue.unpark("warmth")] == [
                "warmth.sensor"
            ]
            assert eng.events.emits == {"warmth.act": 3}
            assert eng.events.digests == {"warmth.cam": ["d1"]}
            assert eng.alerts._last_sent == {"node_error:warmth.act": 7.0}
            assert eng.alerts.config.channels[0].config == {"message": "warmth.alert"}

            renamed = [e for e in told if e["type"] == "renamed"]
            assert [
                {k: e[k] for k in ("kind", "id", "old", "new", "op")} for e in renamed
            ] == [
                {
                    "kind": "flow",
                    "id": "heating",
                    "old": "heating",
                    "new": "warmth",
                    "op": "op-1",
                }
            ]
            changed = [e["dashboard"] for e in told if e["type"] == "dashboard_changed"]
            assert changed == ["home", ""]
            (rebuilt,) = [e for e in told if e["type"] == "pipeline_rebuilt"]
            assert rebuilt["paused"] == ["heating_extra", "warmth"]

            assert not eng.renamer.gate.closed and eng.renamer.pending is None
            assert not controller._lock.locked()
            assert read_journal().phase == "done"
            # Nothing is held: a value goes in again.
            controller.publish_message("heating_extra.x", 1.0)
        finally:
            await eng.controller.stop()

    asyncio.run(scenario())


@pytest.mark.parametrize("how", ["cancel", "disconnect"])
def test_a_rename_called_off_while_draining_changes_nothing(world, how):
    """It waits for the cascade running as long as that takes, says so, and
    once called off leaves everything as it was, the engine running."""
    files = _files(world.root)
    gone = threading.Event()

    async def disconnected() -> bool:
        return gone.is_set()

    async def scenario() -> None:
        eng = await _engine(world)
        controller = eng.controller
        try:
            controller.get_node("heating.sensor").inject({"temp": 21.0})
            assert await asyncio.to_thread(RUNNING.wait, 5)
            async with eng.events.subscribe() as heard:
                renaming_ = asyncio.create_task(
                    eng.renamer.rename(
                        "flow", "heating", _request(eng), "me", disconnected
                    )
                )
                await _until(
                    lambda: eng.renamer.pending is not None
                    and eng.renamer.pending.busy == ["heating.sensor"]
                )
                waiting = [
                    e["busy"]
                    for e in await _heard(eng.events, heard)
                    if e["type"] == "rename_waiting"
                ]
            assert waiting[-1] == ["heating.sensor"]
            assert eng.renamer.gate.closed
            with pytest.raises(EngineHeld):
                controller.publish_message("heating_extra.x", 1.0)

            if how == "cancel":
                with pytest.raises(KeyError):
                    eng.renamer.cancel("heating", "op-2")
                eng.renamer.cancel("heating", "op-1")
            else:
                gone.set()
            with pytest.raises(RenameConflict) as refused:
                await renaming_
            assert (refused.value.status, refused.value.code) == (409, "aborted")

            assert _files(world.root) == files
            assert read_journal() is None
            assert not eng.renamer.gate.closed and eng.renamer.pending is None
            assert not controller._lock.locked()
            assert sorted(controller.loaded) == [
                "heating.act",
                "heating.sensor",
                "heating_extra.x",
                "lights.lamp",
            ]
            # The cascade it waited for finishes, and the next one runs.
            GO.set()
            await _until(lambda: sorted(EFFECTS) == ["heating.act", "lights.lamp"])
            controller.get_node("heating.sensor").inject({"temp": 22.0})
            await _until(lambda: len(EFFECTS) == 4)
        finally:
            GO.set()
            await controller.stop()

    asyncio.run(scenario())


def test_a_plan_that_moved_while_draining_is_refused_under_the_barrier(world):
    """A code-defined dependent that appeared while the rename waited."""

    async def scenario() -> None:
        eng = await _engine(world)
        try:
            eng.controller.get_node("heating.sensor").inject({"temp": 21.0})
            assert await asyncio.to_thread(RUNNING.wait, 5)
            renaming_ = asyncio.create_task(
                eng.renamer.rename("flow", "heating", _request(eng), "me")
            )
            await _until(lambda: eng.renamer.gate.closed)
            lights = world.store.read_flow("lights")
            world.store.write_flow(lights.model_copy(update={"origin": FlowOrigin()}))
            GO.set()
            with pytest.raises(RenameConflict) as refused:
                await renaming_
            assert (refused.value.status, refused.value.code) == (
                409,
                "code_defined_dependent",
            )
            assert refused.value.extra["plan"]["refusals"][0]["ids"] == ["lights"]
            assert world.store.read_flow("heating").name == "heating"
            assert not eng.renamer.gate.closed
            assert not eng.controller._lock.locked()
        finally:
            await eng.controller.stop()

    asyncio.run(scenario())


def test_a_failing_step_is_retried_until_the_rename_finishes(world, monkeypatch):
    """Decision 5: held, said, and finished in the background."""
    monkeypatch.setattr(renaming, "RETRY_S", (0.05, 0.1))
    failures = []

    def fault(point: str) -> None:
        if point == "state" and len(failures) < 2:
            failures.append(point)
            raise ConnectionError("state unreachable")

    monkeypatch.setattr(renaming, "fault", fault)

    async def scenario() -> None:
        eng = await _engine(world)
        try:
            request = _request(eng)
            async with eng.events.subscribe() as heard:
                with pytest.raises(RenameConflict) as busy:
                    await eng.renamer.rename("flow", "heating", request, "me")
                assert (busy.value.status, busy.value.code) == (503, "completing")
                pending = eng.renamer.pending
                assert pending.phase == "forward"
                assert "state unreachable" in pending.error
                assert eng.renamer.gate.closed
                for op, code in (
                    ("op-1", "completing"),
                    ("op-2", "rename_in_progress"),
                ):
                    other = request.model_copy(update={"request_id": op})
                    with pytest.raises(RenameConflict) as again:
                        await eng.renamer.rename("flow", "heating", other, "me")
                    assert (again.value.status, again.value.code) == (503, code)

                await asyncio.wait_for(eng.renamer._resume, 10)
                told = [e["type"] for e in await _heard(eng.events, heard)]
            assert told.count("engine_degraded") == 2
            assert "renamed" in told
            assert eng.renamer.pending is None and not eng.renamer.gate.closed
            assert read_journal().phase == "done"
            assert sorted(eng.controller.loaded)[-2:] == ["warmth.act", "warmth.sensor"]
            # Asked again, the same answer.
            again = await eng.renamer.rename("flow", "heating", request, "me")
            assert again == read_journal().result
        finally:
            await eng.controller.stop()

    asyncio.run(scenario())


def test_renaming_there_and_back_restores_the_namespace(world):
    """With the producers stopped, `heating` to `warmth` and back leaves
    documents, state and queued work as they were, but for versions."""

    def documents() -> dict[str, Any]:
        found = {}
        for path in sorted(world.root.rglob("*.json")):
            if ".git" not in path.parts:
                data = json.loads(path.read_text())
                if isinstance(data, dict):
                    data.pop("version", None)
                found[str(path.relative_to(world.root))] = data
        return found

    async def scenario() -> None:
        eng = await _engine(world, consume=False)
        try:
            eng.state.set("heating.temp", 20.5)
            eng.state.set("__node__:heating.act:last:x", 1)
            eng.queue.add(
                WorkItem("cascade", "heating.sensor", "heating", {"heating.temp": 1})
            )
            eng.queue.add_delayed(WorkItem("flush", "lights.lamp", "lights"), 9e9)
            before = documents(), stability(world.root)
            state = {key: eng.state.get(key) for key in eng.state.keys()}
            queued = [(i.node, i.outputs) for i in eng.queue.live_items()]

            await eng.renamer.rename("flow", "heating", _request(eng), "me")
            assert (
                leftovers(
                    "heating", world.root, ["heating", "lights"], eng.state, eng.queue
                )
                == []
            )
            await eng.renamer.rename(
                "flow", "heating", _request(eng, "heating", op="op-2"), "me"
            )

            assert (documents(), stability(world.root)) == before
            assert {key: eng.state.get(key) for key in eng.state.keys()} == state
            assert [(i.node, i.outputs) for i in eng.queue.live_items()] == queued
            assert sorted(eng.controller.loaded)[0] == "heating.act"
        finally:
            await eng.controller.stop()

    asyncio.run(scenario())


def test_a_dashboard_rename_is_its_documents_and_the_audit_alone(world):
    async def scenario() -> None:
        eng = await _engine(world)
        try:
            closings = []
            eng.renamer.gate.close = lambda *args: closings.append(args)
            request = _request(eng, "house", obj="home")
            async with eng.events.subscribe() as heard:
                result = await eng.renamer.rename("dashboard", "home", request, "me")
                told = await _heard(eng.events, heard)

            assert [(d.kind, d.id) for d in result.documents] == [("dashboard", "home")]
            assert world.dashboards.read("home").name == "house"
            assert closings == []
            assert [(e["type"], e.get("dashboard", e.get("new"))) for e in told] == [
                ("renamed", "house"),
                ("dashboard_changed", "home"),
            ]
            with Session(engine) as session:
                audit = session.exec(
                    select(EngineEvent.flow_id, EngineEvent.detail).where(
                        col(EngineEvent.type) == "audit"
                    )
                ).all()
            assert audit == [("", "renamed dashboard 'home' to 'house' [op op-1]")]
            again = await eng.renamer.rename("dashboard", "home", request, "me")
            assert again == result
        finally:
            await eng.controller.stop()

    asyncio.run(scenario())


def test_a_pending_side_effect_runs_exactly_once_across_a_redis_rename(
    world, monkeypatch
):
    """Claimed by an engine that went away: one item whose side effect was
    recorded as done, which must not run again, and one whose was not, which
    runs once, under the new name."""
    host = os.environ.get("FLUKSIO_TEST_REDIS_HOST")
    if not host:
        pytest.skip("Set FLUKSIO_TEST_REDIS_HOST to run Redis integration checks")
    port = int(os.environ.get("FLUKSIO_TEST_REDIS_PORT", "6379"))
    monkeypatch.setattr(executor, "RECLAIM_IDLE_MS", 50)
    monkeypatch.setattr(executor, "RECLAIM_INTERVAL_S", 0.05)
    namespace = f"test-rename-run-{uuid.uuid4().hex}"
    state = RedisState(host, port, namespace=namespace)
    queue = RedisWorkQueue(
        host, port, namespace=f"{namespace}-queue", consumer="engine-a"
    )
    # The act waits for nobody here.
    heating = world.store.read_flow("heating")
    heating.nodes[1].params = {}
    world.store.write_flow(heating)

    async def scenario() -> None:
        eng = await _engine(world, state=state, queue=queue, consume=False)
        try:
            for value in (1.0, 2.0):
                queue.add(
                    WorkItem(
                        "cascade", "heating.sensor", "heating", {"heating.temp": value}
                    )
                )
            done, pending = queue.claim(2, 10)
            queue.mark_done(done.entry_id, "heating.act")
            await eng.renamer.rename("flow", "heating", _request(eng), "me")
            assert [i.node for i in queue.live_items()] == ["warmth.sensor"] * 2

            eng.controller.execution.start()
            await _until(lambda: EFFECTS.count("lights.lamp") == 2)
            await asyncio.sleep(0.5)
            assert sorted(EFFECTS) == ["lights.lamp", "lights.lamp", "warmth.act"]
            assert queue.live_items() == []
            assert "heating.temp" not in state and "warmth.temp" in state
        finally:
            await eng.controller.stop()

    try:
        asyncio.run(scenario())
    finally:
        client = state._client
        stale = list(client.scan_iter(match=f"{namespace}*"))
        if stale:
            client.delete(*stale)
        state.close()