"""The engine runs a flow by its key and finds it by its id.

Every test runs twice, once with the id equal to the key the way a flow from
before ids has it, and once with them apart. A mix-up of the two is invisible
in the first run, which is why the second one exists.
"""

import asyncio
from pathlib import Path
from types import SimpleNamespace

import pytest

from fluksio.flow.controller import HOOK_PREFIX, FlowController
from fluksio.flow.messages import DType, MessageSpec
from fluksio.flow.queue import MemoryWorkQueue, WorkItem
from fluksio.flow.schemas import FlowDef, NodeDef
from fluksio.flow.store import FlowStore

SENSOR = "def process():\n    return {'reading': 21.5}\n"
DOUBLE = "def process(reading):\n    return {'doubled': reading * 2}\n"


@pytest.fixture(params=["house", "x_house"], ids=["id=key", "id!=key"])
def flow_id(request: pytest.FixtureRequest) -> str:
    return str(request.param)


@pytest.fixture
def store(tmp_path: Path, flow_id: str) -> FlowStore:
    store = FlowStore(tmp_path / "flows")
    store.write_flow(
        FlowDef(
            id=flow_id,
            name="house",
            nodes=[
                NodeDef(
                    id="sensor",
                    provides=[MessageSpec(name="reading", dtype=DType.FLOAT)],
                ),
                NodeDef(
                    id="double",
                    requires=[MessageSpec(name="reading", dtype=DType.FLOAT)],
                    provides=[MessageSpec(name="doubled", dtype=DType.FLOAT)],
                ),
                NodeDef(
                    id="hook",
                    type="http",
                    params={"url": "/tick", "method": "POST"},
                    provides=[MessageSpec(name="tick", dtype=DType.FLOAT)],
                ),
            ],
        )
    )
    store.write_node_source(flow_id, "sensor", SENSOR)
    store.write_node_source(flow_id, "double", DOUBLE)
    return store


def test_a_flow_runs_under_its_key_from_the_code_its_id_holds(
    store: FlowStore, flow_id: str
):
    engine = FlowController(store)
    asyncio.run(engine.reload())

    entry = engine.loaded["house.double"]
    assert (entry.flow, entry.flow_id) == ("house", flow_id)
    # Read by the id: by the key it would be the template, and missing.
    assert not entry.missing_source and entry.status.value == "active"
    assert engine.loaded["house.hook"].node.url == f"{HOOK_PREFIX}/{flow_id}/tick"

    engine.run_flow("house")

    assert engine.values()["house.doubled"]["value"] == 43.0
    assert {s.id for s in engine.node_statuses("house")} >= {"house.double"}


def test_stopping_writes_under_the_id_and_gates_the_key(store: FlowStore, flow_id: str):
    engine = FlowController(store)

    async def scenario() -> None:
        await engine.reload()
        await engine.set_enabled(flow_id, "house", False)
        assert engine.disabled == {"house"}
        assert not store.read_enabled(flow_id)
        # Read back by id when everything is built again.
        await engine.reload()
        assert engine.disabled == {"house"}

    asyncio.run(scenario())
    assert (store.root / flow_id / "runtime.json").exists()


def test_a_flow_rebuilt_by_id_is_taken_out_by_its_key_once_gone(
    store: FlowStore, flow_id: str
):
    engine = FlowController(store)

    async def scenario() -> None:
        await engine.reload()
        assert "house.sensor" in engine.loaded
        await engine.reload_flow(flow_id)
        assert "house.sensor" in engine.loaded

        store.delete_flow(flow_id)
        await engine.reload_flow(flow_id)

    asyncio.run(scenario())
    assert not [node for node in engine.loaded if node.startswith("house.")]
    assert engine.pipeline is not None and not engine.pipeline.flow_nodes("house")


def test_a_draft_previews_by_id_and_reports_by_key(store: FlowStore, flow_id: str):
    draft = store.read_flow(flow_id).model_copy(update={"title": "Edited"})
    store.write_draft(draft, draft.version)
    engine = FlowController(store)

    preview = engine.preview(flow_id)

    assert {node.id for node in preview.nodes} == {
        "house.sensor",
        "house.double",
        "house.hook",
    }
    assert all(node.status == "active" for node in preview.nodes)
    engine.run_preview(flow_id)
    assert engine.state.get("house.doubled") == 43.0


def test_what_a_key_leaves_behind_is_found_and_forgotten(store: FlowStore):
    engine = FlowController(store)
    queue = MemoryWorkQueue()
    engine.execution = SimpleNamespace(queue=queue)  # type: ignore[assignment]
    engine.state.set("house.reading", 1.0)
    engine.state.set("__node__:house.double:last", 2)
    engine.state.set("housework.reading", 3.0)
    queue.add_delayed(WorkItem(kind="flush", node="house.double", flow="house"), 9e9)
    queue.park("house", WorkItem(kind="cascade", node="house.sensor", flow="house"))
    # Another flow's work, carrying one of this key's messages.
    queue.add(
        WorkItem(
            kind="cascade", node="garden.pump", flow="garden", outputs={"house.x": 1}
        )
    )

    found = engine.key_residue("house")

    assert "house.reading" in found and "__node__:house.double:last" in found
    assert "housework.reading" not in found
    assert sum("queued work" in what for what in found) == 3
    assert engine.key_residue("garden") == ["queued work of 'garden.pump'"]

    engine.forget_flow("house")

    # What the other flow queued is its own, and runs as it would have.
    assert engine.key_residue("house") == ["queued work of 'garden.pump'"]
    assert engine.state.get("housework.reading") == 3.0