"""Engine holds for a rename (section 2.2 of docs/private/id_refac_rename.md).

The pipeline refuses what it would run outside the queue, and the execution
service stops claiming and its timers once claimed work has finished. Both
wait without a limit and can be aborted.
"""

import asyncio
import threading
import time

import pytest

from fluksio.flow import executor
from fluksio.flow.controller import FlowController, LoadedNode, RebuildBusy
from fluksio.flow.executor import ExecutionService
from fluksio.flow.messages import DType, MessageSpec
from fluksio.flow.namespace import Rename
from fluksio.flow.nodes import Node
from fluksio.flow.pipeline import EngineHeld, Pipeline
from fluksio.flow.queue import MemoryWorkQueue, WorkItem
from fluksio.flow.state import MemoryState
from fluksio.flow.store import FlowStore


def _engine(consume, queue=None) -> tuple[Pipeline, ExecutionService, Node]:
    """A source whose readings ``consume`` gets, and a service draining the
    queue one cascade at a time, so order shows."""

    def consumer_body(reading, params):
        consume(reading)

    source = Node(
        f=lambda params: None,
        provides=[MessageSpec(name="reading", dtype=DType.FLOAT)],
        name="source",
    )
    consumer = Node(
        f=consumer_body,
        requires=[MessageSpec(name="reading", dtype=DType.FLOAT)],
        name="consumer",
    )
    source.assign_flow("f", "source")
    consumer.assign_flow("f", "consumer")
    queue = queue or MemoryWorkQueue()
    service = ExecutionService(queue, max_cascades=1)
    pipeline = Pipeline(
        nodes=[source, consumer],
        state=MemoryState(),
        work_queue=queue,
        node_pool=service.node_pool,
    )
    service.bind(pipeline)
    return pipeline, service, source


def _in_thread(fn, *args) -> tuple[threading.Thread, list]:
    result: list = []
    thread = threading.Thread(target=lambda: result.append(fn(*args)), daemon=True)
    thread.start()
    return thread, result


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


def _blocking(seen: list) -> tuple[threading.Event, threading.Event, object]:
    """A consumer that records and then waits until told to finish."""
    running, finish = threading.Event(), threading.Event()

    def consume(reading):
        seen.append(reading)
        running.set()
        finish.wait(5)

    return running, finish, consume


def test_a_hold_waits_out_a_running_cascade_without_a_limit(monkeypatch):
    # What a rebuild's drain gives up after, which a hold does not.
    monkeypatch.setattr(executor, "DRAIN_TIMEOUT_S", 0.05)
    seen: list = []
    running, finish, consume = _blocking(seen)
    _, service, source = _engine(consume)
    service.start()
    try:
        source.inject({"reading": 1.0})
        assert running.wait(5)
        busy: list = []
        thread, held = _in_thread(service.hold, threading.Event(), busy.append)
        thread.join(0.3)
        assert thread.is_alive()
        assert busy == [["f.source"]]

        source.inject({"reading": 2.0})
        finish.set()
        thread.join(5)
        assert held == [True]
        time.sleep(0.2)
        assert seen == [1.0]

        service.release()
        _until(lambda: seen == [1.0, 2.0])
    finally:
        finish.set()
        service.stop()


class SlowClaim(MemoryWorkQueue):
    """A queue whose claim is still under way when a hold begins."""

    def __init__(self) -> None:
        super().__init__()
        self.claiming = threading.Event()
        self.go = threading.Event()

    def claim(self, count, block_ms, due_only=False):
        items = super().claim(count, block_ms, due_only)
        if items:
            self.claiming.set()
            self.go.wait(5)
        return items


def test_pause_intake_waits_out_a_claim_already_under_way(monkeypatch):
    seen: list = []
    queue = SlowClaim()
    _, service, source = _engine(seen.append, queue)
    wake = service._capacity_changed

    def closing() -> None:
        wake()
        # The claim lands only once the intake has closed.
        queue.go.set()

    monkeypatch.setattr(service, "_capacity_changed", closing)
    service.start()
    try:
        source.inject({"reading": 1.0})
        assert queue.claiming.wait(5)
        service.pause_intake()
        assert seen == [1.0]
        assert service.inflight == 0
    finally:
        queue.go.set()
        service.stop()


def test_a_reload_that_cannot_drain_leaves_the_pipeline_as_it_was(
    tmp_path, monkeypatch
):
    monkeypatch.setattr(executor, "DRAIN_TIMEOUT_S", 0.05)
    seen: list = []
    running, finish, consume = _blocking(seen)
    pipeline, service, source = _engine(consume)
    controller = FlowController(FlowStore(tmp_path / "flows"), execution=service)
    controller.pipeline = pipeline
    service.start()
    try:
        source.inject({"reading": 1.0})
        assert running.wait(5)
        with pytest.raises(RebuildBusy):
            asyncio.run(controller.reload())
        assert controller.pipeline is pipeline
        assert service._pipeline is pipeline
        assert service._intake.is_set()
    finally:
        finish.set()
        service.stop()


def test_a_hold_waits_out_a_claim_already_under_way():
    seen: list = []
    queue = SlowClaim()
    _, service, source = _engine(seen.append, queue)
    service.start()
    try:
        source.inject({"reading": 1.0})
        assert queue.claiming.wait(5)
        thread, held = _in_thread(service.hold, threading.Event())
        thread.join(0.2)
        assert thread.is_alive()

        queue.go.set()
        thread.join(5)
        assert held == [True]
        assert seen == [1.0]
        assert service.inflight == 0
        service.release()
    finally:
        queue.go.set()
        service.stop()


def test_an_aborted_hold_leaves_the_engine_as_it_was():
    seen: list = []
    running, finish, consume = _blocking(seen)
    pipeline, service, source = _engine(consume)
    service.start()
    try:
        source.inject({"reading": 1.0})
        assert running.wait(5)
        abort = threading.Event()
        thread, held = _in_thread(service.hold, abort)
        thread.join(0.2)
        assert thread.is_alive()

        abort.set()
        thread.join(5)
        assert held == [False]
        finish.set()

        # Claiming and promotion carry on.
        assert pipeline.defer(source, {"f.reading": 2.0}, 0.05)
        source.inject({"reading": 3.0})
        _until(lambda: sorted(seen) == [1.0, 2.0, 3.0])
        assert not service.queue.held
    finally:
        finish.set()
        service.stop()


def test_an_aborted_pipeline_hold_lets_publishes_through_again():
    seen: list = []
    running, finish, consume = _blocking(seen)
    pipeline, _, _ = _engine(consume)
    publishing, _ = _in_thread(pipeline.publish, {"f.reading": 1.0})
    try:
        assert running.wait(5)
        busy: list = []
        abort = threading.Event()
        thread, held = _in_thread(pipeline.hold, abort, busy.append)
        thread.join(0.2)
        assert thread.is_alive()
        assert busy == [["f.reading"]]

        abort.set()
        thread.join(5)
        assert held == [False]
        finish.set()
        publishing.join(5)
        pipeline.publish({"f.reading": 2.0})
        assert seen == [1.0, 2.0]
    finally:
        finish.set()


def test_publish_is_refused_while_held():
    seen: list = []
    pipeline, _, _ = _engine(seen.append)
    assert pipeline.hold(threading.Event())

    with pytest.raises(EngineHeld):
        pipeline.publish({"f.reading": 1.0})
    assert "f.reading" not in pipeline.state

    pipeline.release()
    pipeline.publish({"f.reading": 2.0})
    assert seen == [2.0]


def test_a_trigger_by_hand_while_held_is_refused_not_failed(tmp_path, monkeypatch):
    seen: list = []
    pipeline, _, source = _engine(seen.append)
    controller = FlowController(FlowStore(tmp_path / "flows"))
    controller.pipeline = pipeline
    controller.loaded[source.id] = LoadedNode(id=source.id, flow="f", node=source)
    failures: list = []
    monkeypatch.setattr(Pipeline, "publish_error", lambda *args: failures.append(args))
    assert pipeline.hold(threading.Event())

    with pytest.raises(EngineHeld):
        controller.trigger_node(source.id, {"reading": 1.0})
    assert failures == [] and seen == []


def test_work_arriving_during_a_hold_runs_after_it_in_order():
    seen: list = []
    pipeline, service, source = _engine(seen.append)
    service.start()
    try:
        abort = threading.Event()
        assert pipeline.hold(abort)
        assert service.hold(abort)
        source.inject({"reading": 1.0})
        service.queue.hold()
        source.inject({"reading": 2.0})
        source.inject({"reading": 3.0})
        time.sleep(0.2)
        assert seen == []

        service.queue.release()
        service.release()
        pipeline.release()
        _until(lambda: seen == [1.0, 2.0, 3.0])
    finally:
        service.stop()


def test_a_queue_failure_while_held_drops_the_value_rather_than_run_it(
    monkeypatch,
):
    seen: list = []
    pipeline, service, source = _engine(seen.append)

    def down(item):
        raise ConnectionError("queue down")

    monkeypatch.setattr(service.queue, "add", down)
    assert pipeline.hold(threading.Event())
    source.inject({"reading": 1.0})
    assert seen == []

    pipeline.release()
    source.inject({"reading": 2.0})
    assert seen == [2.0]


def test_reclaimed_work_waiting_to_run_follows_a_rename():
    service = ExecutionService(MemoryWorkQueue())
    service._reclaimed.extend(
        [
            WorkItem(kind="cascade", node="heat.valve", flow="heat", entry_id="1-0"),
            WorkItem(kind="cascade", node="heating.pump", flow="heating"),
        ]
    )
    service.rename_reclaimed(Rename("heat", "warm"))
    assert [(i.node, i.flow) for i in service._reclaimed] == [
        ("warm.valve", "warm"),
        ("heating.pump", "heating"),
    ]
    assert service._reclaimed[0].entry_id == "1-0"