"""The work queue, and what the execution service does with it."""

import threading
import time

from fluksio.flow import executor
from fluksio.flow import queue as queue_module
from fluksio.flow.executor import ExecutionService
from fluksio.flow.messages import DType, MessageSpec
from fluksio.flow.nodes import Node
from fluksio.flow.pipeline import Pipeline
from fluksio.flow.queue import MemoryWorkQueue, WorkItem
from fluksio.flow.state import MemoryState


def test_items_come_back_in_the_order_they_went_in():
    queue = MemoryWorkQueue()
    for i in range(3):
        queue.add(WorkItem(kind="cascade", node=f"f.n{i}", flow="f"))

    claimed = queue.claim(10, 10)

    assert [item.node for item in claimed] == ["f.n0", "f.n1", "f.n2"]
    # Every item gets an id, which is what idempotency markers hang off.
    assert all(item.entry_id for item in claimed)


def test_claiming_an_empty_queue_waits_and_gives_up():
    queue = MemoryWorkQueue()
    started = time.monotonic()

    assert queue.claim(1, 50) == []
    assert time.monotonic() - started >= 0.04


def test_a_delayed_item_stays_put_until_it_is_due():
    queue = MemoryWorkQueue()
    queue.add_delayed(WorkItem(kind="cascade", node="f.n", flow="f"), time.time() + 60)

    assert queue.claim(1, 10) == []
    assert queue.move_due(time.time()) == 0

    assert queue.move_due(time.time() + 61) == 1
    assert [i.node for i in queue.claim(1, 10)] == ["f.n"]


def test_the_queue_says_when_the_soonest_delayed_item_is_due():
    """What lets the engine sleep to a deadline rather than poll for it."""
    queue = MemoryWorkQueue()
    assert queue.next_due() is None

    queue.add_delayed(WorkItem(kind="cascade", node="f.late", flow="f"), 500.0)
    queue.add_delayed(WorkItem(kind="cascade", node="f.soon", flow="f"), 100.0)
    assert queue.next_due() == 100.0

    queue.move_due(200.0)
    assert queue.next_due() == 500.0


def test_scheduling_says_a_deadline_moved():
    """A delay scheduled for 200ms must cut short a sleep to the next second."""
    queue = MemoryWorkQueue()
    woken = threading.Event()
    queue.on_delayed = woken.set

    queue.add_delayed(WorkItem(kind="cascade", node="f.n", flow="f"), time.time() + 0.2)

    assert woken.is_set()


def test_a_due_timer_is_claimed_before_the_backlog():
    """Lateness is what a timer is measured by; queued work waits on no clock."""
    queue = MemoryWorkQueue()
    for i in range(3):
        queue.add(WorkItem(kind="cascade", node=f"f.queued{i}", flow="f"))
    queue.add_delayed(WorkItem(kind="cascade", node="f.timer", flow="f"), 100.0)

    queue.move_due(200.0)

    assert [i.node for i in queue.claim(10, 10)][0] == "f.timer"


def test_a_delay_fires_without_waiting_out_the_poll():
    """The whole point: a 200ms delay is 200ms late, not up to a second."""
    queue = MemoryWorkQueue()
    service = ExecutionService(queue)
    dispatched: list[float] = []
    service._dispatch = lambda item: dispatched.append(time.monotonic())  # type: ignore[method-assign]
    service.start()
    try:
        started = time.monotonic()
        queue.add_delayed(
            WorkItem(kind="cascade", node="f.n", flow="f"), time.time() + 0.2
        )
        # Claimed by the consumer thread, which is what calls _dispatch.
        deadline = time.monotonic() + 2.0
        while not dispatched and time.monotonic() < deadline:
            time.sleep(0.01)
    finally:
        service.stop()

    assert dispatched, "the deferred item never ran"
    late = dispatched[0] - started - 0.2
    # A fixed one-second poll made this up to 1.0s; the budget is generous
    # because CI is not a real-time machine.
    assert late < 0.3, f"fired {late:.3f}s late"


def test_parked_work_comes_back_oldest_first():
    queue = MemoryWorkQueue()
    for i in range(3):
        queue.park("heating", WorkItem(kind="cascade", node=f"f.n{i}", flow="heating"))

    assert [i.node for i in queue.unpark("heating")] == ["f.n0", "f.n1", "f.n2"]
    # Unparking empties it, so a second resume does not replay the same work.
    assert queue.unpark("heating") == []


def test_a_deleted_flow_leaves_nothing_parked():
    queue = MemoryWorkQueue()
    queue.park("gone", WorkItem(kind="cascade", node="gone.n", flow="gone"))

    queue.clear_flow("gone")

    assert queue.unpark("gone") == []


def test_claimed_work_counts_as_in_flight_until_it_is_acknowledged():
    """The health tile's "in flight" reads zero without this."""
    queue = MemoryWorkQueue()
    queue.add(WorkItem(kind="cascade", node="f.n", flow="f"))

    assert queue.stats()["pending"] == 0
    # The stream length was never a backlog, so the key is gone from both queues.
    assert "depth" not in queue.stats()

    (item,) = queue.claim(1, 10)
    assert queue.stats()["pending"] == 1

    queue.ack(item)
    assert queue.stats()["pending"] == 0


def test_work_waiting_to_be_claimed_is_the_backlog():
    """`pending` is what is running; an engine hours behind reports it as idle."""
    queue = MemoryWorkQueue()
    queue.add(WorkItem(kind="cascade", node="f.n", flow="f"))

    assert queue.stats()["backlog"] == 1
    assert queue.stats()["pending"] == 0

    (item,) = queue.claim(1, 10)
    assert queue.stats()["backlog"] == 0
    assert queue.stats()["pending"] == 1

    queue.ack(item)
    assert queue.stats()["backlog"] == 0


def test_a_sustained_backlog_says_the_engine_is_behind():
    """A flow enqueuing faster than the pool drains produced no signal at all."""
    events: list[dict] = []
    queue = MemoryWorkQueue()
    service = ExecutionService(queue)
    service._publish = events.append  # type: ignore[method-assign]
    for _ in range(executor.BACKLOG_DEGRADED):
        queue.add(WorkItem(kind="cascade", node="f.n", flow="f"))

    for _ in range(executor.BACKLOG_STRIKES - 1):
        service._check_backlog()
    assert events == []

    service._check_backlog()
    assert [e["type"] for e in events] == ["engine_degraded"]
    assert service.stats()["behind"] is True

    # Said once, not once every five seconds for as long as it lasts.
    service._check_backlog()
    assert len(events) == 1

    # And a drained queue clears it, so the next backlog is announced again.
    queue.claim(executor.BACKLOG_DEGRADED, 10)
    service._check_backlog()
    assert service.stats()["behind"] is False


def _pipeline_with_a_consumer() -> tuple[Pipeline, Node, MemoryState, list]:
    """A source whose message a consumer records."""
    seen: list[float] = []

    def consume(reading, params):
        seen.append(reading)
        return {"doubled": reading * 2}

    source = Node(
        f=lambda params: None,
        provides=[MessageSpec(name="reading", port="reading", dtype=DType.FLOAT)],
        name="source",
    )
    consumer = Node(
        f=consume,
        requires=[MessageSpec(name="reading", port="reading", dtype=DType.FLOAT)],
        provides=[MessageSpec(name="doubled", port="doubled", dtype=DType.FLOAT)],
        name="consumer",
    )
    source.assign_flow("f", "source")
    consumer.assign_flow("f", "consumer")

    state = MemoryState()
    queue = MemoryWorkQueue()
    pipeline = Pipeline(nodes=[source, consumer], state=state, work_queue=queue)
    return pipeline, source, state, seen


def test_a_trigger_is_journaled_rather_than_run_on_the_spot():
    pipeline, source, state, seen = _pipeline_with_a_consumer()

    source.inject({"reading": 3.0})

    # Nothing ran yet: the value is in the queue, not in state.
    assert seen == []
    assert "f.reading" not in state

    service = ExecutionService(pipeline._queue)
    service.bind(pipeline)
    for item in pipeline._queue.claim(10, 10):
        service._run_item(item)

    assert seen == [3.0]
    assert state["f.doubled"] == 6.0


def test_work_for_a_paused_flow_is_held_and_released_on_resume():
    pipeline, source, state, seen = _pipeline_with_a_consumer()
    service = ExecutionService(pipeline._queue)
    service.bind(pipeline)

    pipeline.pause("f")
    source.inject({"reading": 1.0})
    for item in pipeline._queue.claim(10, 10):
        service._run_item(item)

    assert seen == []

    pipeline.resume("f")
    for item in pipeline._queue.unpark("f"):
        pipeline._queue.add(item)
    for item in pipeline._queue.claim(10, 10):
        service._run_item(item)

    assert seen == [1.0]


def test_a_step_runs_one_held_item_and_leaves_the_flow_paused():
    pipeline, source, _state, seen = _pipeline_with_a_consumer()
    service = ExecutionService(pipeline._queue)
    service.bind(pipeline)

    pipeline.pause("f")
    source.inject({"reading": 1.0})
    source.inject({"reading": 2.0})
    for item in pipeline._queue.claim(10, 10):
        service._run_item(item)
    assert seen == []

    assert service.step("f") == "f.source"

    assert seen == [1.0]
    # Still paused, and the second value is still waiting for the next step.
    assert pipeline.is_paused("f")
    assert [i.outputs for i in pipeline._queue.unpark("f")] == [{"f.reading": 2.0}]


def test_stepping_a_flow_with_nothing_held_says_so_rather_than_failing():
    pipeline, _source, _state, _seen = _pipeline_with_a_consumer()
    service = ExecutionService(pipeline._queue)
    service.bind(pipeline)

    pipeline.pause("f")

    assert service.step("f") is None


def test_work_for_a_stopped_flow_is_dropped():
    pipeline, source, state, seen = _pipeline_with_a_consumer()
    stopped = Pipeline(
        nodes=pipeline.nodes,
        state=pipeline.state,
        work_queue=pipeline._queue,
        disabled_flows={"f"},
    )
    service = ExecutionService(pipeline._queue)
    service.bind(stopped)

    # Reaching the queue at all takes a direct add: trigger drops it earlier.
    stopped._queue.add(
        WorkItem(kind="cascade", node="f.source", flow="f", outputs={"f.reading": 1.0})
    )
    for item in stopped._queue.claim(10, 10):
        service._run_item(item)

    assert seen == []


def test_an_item_that_keeps_coming_back_is_dead_lettered():
    pipeline, _source, _state, seen = _pipeline_with_a_consumer()
    service = ExecutionService(pipeline._queue)
    service.bind(pipeline)

    item = WorkItem(
        kind="cascade",
        node="f.source",
        flow="f",
        outputs={"f.reading": 1.0},
        deliveries=4,
    )
    service._run_item(item)

    # Given up on rather than run again, so a poison item cannot loop forever.
    assert seen == []


def test_an_item_for_a_node_that_no_longer_exists_is_dropped():
    pipeline, _source, _state, seen = _pipeline_with_a_consumer()
    service = ExecutionService(pipeline._queue)
    service.bind(pipeline)

    service._run_item(WorkItem(kind="cascade", node="f.removed", flow="f"))

    assert seen == []


def test_a_replayed_item_does_not_repeat_a_side_effect():
    """At-least-once delivery must not mean two of the same outgoing request."""
    calls: list[float] = []

    def send(reading, params):
        calls.append(reading)
        return None

    source = Node(
        f=lambda params: None,
        provides=[MessageSpec(name="reading", port="reading", dtype=DType.FLOAT)],
        name="source",
    )

    class SendingNode(Node):
        """Stands in for the built-ins that reach outside."""

        idempotent = False

    sender = SendingNode(
        f=send,
        requires=[MessageSpec(name="reading", port="reading", dtype=DType.FLOAT)],
        name="sender",
    )
    source.assign_flow("f", "source")
    sender.assign_flow("f", "sender")

    queue = MemoryWorkQueue()
    pipeline = Pipeline(nodes=[source, sender], state=MemoryState(), work_queue=queue)
    service = ExecutionService(queue)
    service.bind(pipeline)

    source.inject({"reading": 5.0})
    (item,) = queue.claim(10, 10)
    service._run_item(item)
    assert calls == [5.0]

    # The same item again, as a reaper would hand it back after a crash.
    item.deliveries = 2
    service._run_item(item)

    assert calls == [5.0]


# -----------------------------------------------------------------------------
# Telling the queue that a long node is working, not lost
#
# What marks an item abandoned is nobody touching it. A node with no timeout
# may run for hours, so the engine holding it says so on a timer — and an
# engine that died says nothing, which is the distinction the reaper needs.
# -----------------------------------------------------------------------------


class RecordingQueue(MemoryWorkQueue):
    """A memory queue that writes down what it was asked to hold on to."""

    def __init__(self) -> None:
        super().__init__()
        self.touched: list[list[str]] = []

    def touch(self, entry_ids: list[str]) -> None:
        self.touched.append(list(entry_ids))


def test_work_in_flight_is_touched_until_it_finishes(monkeypatch):
    monkeypatch.setattr(executor, "TOUCH_INTERVAL_S", 0.0)
    monkeypatch.setattr(executor, "DELAYED_INTERVAL_S", 0.05)
    running = threading.Event()
    release = threading.Event()

    def slow(reading, params):
        running.set()
        release.wait(5)
        return {"doubled": reading * 2}

    source = Node(
        f=lambda params: None,
        provides=[MessageSpec(name="reading", dtype=DType.FLOAT)],
        name="source",
    )
    consumer = Node(
        f=slow,
        requires=[MessageSpec(name="reading", dtype=DType.FLOAT)],
        provides=[MessageSpec(name="doubled", dtype=DType.FLOAT)],
        name="consumer",
    )
    source.assign_flow("f", "source")
    consumer.assign_flow("f", "consumer")

    queue = RecordingQueue()
    pipeline = Pipeline(nodes=[source, consumer], state=MemoryState(), work_queue=queue)
    service = ExecutionService(queue)
    service.bind(pipeline)
    service.start()
    try:
        source.inject({"reading": 3.0})
        assert running.wait(5)
        # Give the timer a couple of passes while the node is still in there.
        time.sleep(0.2)
        held = [ids for ids in queue.touched if ids]
        assert held, "a running item was never touched"

        release.set()
        time.sleep(0.3)
        # Once it is done it is acknowledged, so there is nothing to hold.
        assert queue.touched[-1] == []
    finally:
        release.set()
        service.stop()


def test_no_more_is_claimed_than_the_pool_can_run():
    """A backlog belongs in the queue, not inside the process.

    Claiming ahead of the pool used to leave every waiting item counted as a
    busy cascade and holding its journal entry open, so four cascade threads
    reported hundreds in flight on an engine that was merely behind.
    """
    release = threading.Event()

    def slow(reading, params):
        release.wait(5)
        return {"doubled": reading * 2}

    source = Node(
        f=lambda params: None,
        provides=[MessageSpec(name="reading", dtype=DType.FLOAT)],
        name="source",
    )
    consumer = Node(
        f=slow,
        requires=[MessageSpec(name="reading", dtype=DType.FLOAT)],
        provides=[MessageSpec(name="doubled", dtype=DType.FLOAT)],
        name="consumer",
    )
    source.assign_flow("f", "source")
    consumer.assign_flow("f", "consumer")

    queue = MemoryWorkQueue()
    pipeline = Pipeline(nodes=[source, consumer], state=MemoryState(), work_queue=queue)
    service = ExecutionService(queue)
    service.bind(pipeline)
    service.start()
    try:
        for i in range(40):
            queue.add(
                WorkItem(
                    kind="cascade",
                    node="f.source",
                    flow="f",
                    outputs={"f.reading": float(i)},
                )
            )
        time.sleep(0.5)

        stats = service.stats()
        assert stats["cascades_busy"] <= service.max_cascades
        # And the journal entries of what is only waiting are still free.
        assert stats["pending"] <= service.max_cascades
        # Waiting is not idle: the rest of the forty is the backlog.
        assert stats["backlog"] >= 30
    finally:
        release.set()
        service.stop()


# -----------------------------------------------------------------------------
# Emissions: values a node publishes while it is still running
# -----------------------------------------------------------------------------


def _emitting_pipeline() -> tuple[Pipeline, Node, MemoryState, list]:
    """A source with a streaming port, and a consumer that records every value."""
    seen: list[float] = []

    def consume(chunk, params):
        seen.append(chunk)
        return None

    source = Node(
        f=lambda params: None,
        provides=[
            MessageSpec(name="chunk", port="chunk", dtype=DType.FLOAT, stream=True)
        ],
        name="source",
    )
    consumer = Node(
        f=consume,
        requires=[MessageSpec(name="chunk", port="chunk", dtype=DType.FLOAT)],
        name="consumer",
    )
    source.assign_flow("f", "source")
    consumer.assign_flow("f", "consumer")

    state = MemoryState()
    queue = MemoryWorkQueue()
    pipeline = Pipeline(nodes=[source, consumer], state=state, work_queue=queue)
    return pipeline, source, state, seen


def test_every_emitted_chunk_reaches_the_consumer():
    """A consumer slower than its producer must not skip what it missed.

    Reading state instead would give whichever chunk is newest by the time the
    item runs — fine for a temperature, lossy for a second of speech.
    """
    pipeline, source, state, seen = _emitting_pipeline()
    service = ExecutionService(pipeline._queue)
    service.bind(pipeline)

    # Both emitted before either item is claimed, so state has moved on.
    pipeline.publish_emission(source, {"f.chunk": 1.0})
    pipeline.publish_emission(source, {"f.chunk": 2.0})
    assert state["f.chunk"] == 2.0

    for item in pipeline._queue.claim(10, 10):
        service._run_item(item)

    assert seen == [1.0, 2.0]


def test_a_delivered_emission_does_not_write_state_a_second_time():
    """The value in state stays the newest one, whenever an item is claimed."""
    pipeline, source, state, seen = _emitting_pipeline()
    service = ExecutionService(pipeline._queue)
    service.bind(pipeline)

    pipeline.publish_emission(source, {"f.chunk": 1.0})
    # What the node returned at the end, after the emission it made on the way.
    pipeline.apply_outputs(source, {"f.chunk": 9.0})

    for item in pipeline._queue.claim(10, 10):
        service._run_item(item)

    assert seen[0] == 1.0
    # Re-applying the carried chunk here is what would undo the final value.
    assert state["f.chunk"] == 9.0


def test_a_throttled_emission_wakes_nothing():
    pipeline, source, _state, seen = _emitting_pipeline()
    source.provides["f.chunk"] = MessageSpec(
        name="f.chunk", port="chunk", dtype=DType.FLOAT, stream=True, interval=60
    )
    service = ExecutionService(pipeline._queue)
    service.bind(pipeline)

    pipeline.publish_emission(source, {"f.chunk": 1.0})
    pipeline.publish_emission(source, {"f.chunk": 2.0})
    for item in pipeline._queue.claim(10, 10):
        service._run_item(item)

    # The first is let through; the second is held by the interval, and a value
    # nothing published is nothing to wake on.
    assert seen == [1.0]


def test_a_due_timer_is_claimable_while_the_backlog_is_not():
    """What a saturated engine uses to get a motor's stop out.

    Every cascade slot held by a long node used to mean the due lane was not
    merely behind the backlog but unread, because the capacity gate sits in
    front of the claim.
    """
    queue = MemoryWorkQueue()
    queue.add(WorkItem(kind="cascade", node="f.backlog", flow="f"))
    queue.add_delayed(
        WorkItem(kind="cascade", node="f.timer", flow="f"), time.time() - 1
    )
    assert queue.move_due(time.time()) == 1

    claimed = queue.claim(10, 10, due_only=True)
    assert [item.node for item in claimed] == ["f.timer"]

    # And the backlog is still there for whoever comes next.
    assert [item.node for item in queue.claim(10, 10)] == ["f.backlog"]


def test_a_due_only_claim_of_nothing_does_not_take_the_backlog():
    queue = MemoryWorkQueue()
    queue.add(WorkItem(kind="cascade", node="f.backlog", flow="f"))

    assert queue.claim(10, 10, due_only=True) == []
    assert len(queue.claim(10, 10)) == 1


def test_done_markers_do_not_grow_without_bound():
    """The Redis side expires them after an hour; this one is capped by count."""
    queue = MemoryWorkQueue()
    for i in range(queue_module.DONE_MARKERS + 500):
        queue.mark_done(f"entry-{i}", "f.node")

    assert len(queue._done) == queue_module.DONE_MARKERS
    # The newest are what redelivery would ask about.
    assert queue.was_done(f"entry-{queue_module.DONE_MARKERS + 499}", "f.node")
    assert not queue.was_done("entry-0", "f.node")