"""Live state and queued work following a renamed flow key (sections 1.2, 1.3
and 3 of docs/private/id_refac_rename.md). Memory always, Redis with
FLUKSIO_TEST_REDIS_HOST (and optional PORT)."""

import os
import threading
import time
import uuid

import pytest
import redis

from fluksio.flow import state as state_module
from fluksio.flow.namespace import HISTORY, NAMES, Rename
from fluksio.flow.queue import GROUP, MemoryWorkQueue, RedisWorkQueue, WorkItem
from fluksio.flow.state import MemoryState, RedisState

HEAT = Rename("heat", "warm")
BACK = Rename("warm", "heat")
NOTHING = dict.fromkeys(("queue_items", "delayed", "parked", "done_markers"), 0)


def _server() -> tuple[str, int]:
    host = os.environ.get("FLUKSIO_TEST_REDIS_HOST")
    if not host:
        pytest.skip("Set FLUKSIO_TEST_REDIS_HOST to run Redis integration checks")
    return host, int(os.environ.get("FLUKSIO_TEST_REDIS_PORT", "6379"))


def _drop(client, ns: str) -> None:
    keys = list(client.scan_iter(match=f"{ns}:*"))
    if keys:
        client.delete(*keys)


def _redis_queue(ns: str, consumer: str) -> RedisWorkQueue:
    host, port = _server()
    return RedisWorkQueue(host, port, namespace=ns, consumer=consumer)


@pytest.fixture
def redis_queue():
    queue = _redis_queue(f"test-rename-{uuid.uuid4().hex}", "engine-a")
    try:
        yield queue
    finally:
        _drop(queue._redis, queue._ns)
        queue.close()


@pytest.fixture(params=["memory", "redis"])
def queue(request):
    if request.param == "memory":
        yield MemoryWorkQueue()
    else:
        yield request.getfixturevalue("redis_queue")


@pytest.fixture(params=["memory", "redis"])
def state(request):
    if request.param == "memory":
        yield MemoryState()
        return
    host, port = _server()
    state = RedisState(host, port, namespace=f"test-rename-{uuid.uuid4().hex}", ttl=600)
    try:
        yield state
    finally:
        _drop(state._client, state._namespace)
        state.close()


def item(node: str, kind: str = "cascade", **fields) -> WorkItem:
    return WorkItem(kind=kind, node=node, flow=node.split(".")[0], **fields)


def _fill(queue) -> WorkItem:
    """Work in every place a queue keeps it, naming heat or not. Returns the
    item left claimed."""
    queue.add(item("heat.acked"))
    queue.ack(*queue.claim(1, 10))
    queue.add(item("heat.valve", outputs={"heat.temp": 21}))
    queue.add(item("hall.lamp", outputs={"heat.target": 21, "hall.on": True}))
    queue.add(item("heating.valve"))
    queue.add(WorkItem(kind="run", node="", flow="heat", run_id="r1"))
    queue.add_delayed(item("heat.pump", kind="flush"), 100.0)
    queue.move_due(150.0)
    (claimed,) = queue.claim(1, 10, due_only=True)
    queue.mark_done(claimed.entry_id, "heat.pump")
    queue.mark_done("other", "heating.valve")
    queue.add_delayed(item("heat.limit", guard_key="generation", guard_value="3"), 9e9)
    queue.add_delayed(item("heating.limit"), 9e9 + 1)
    queue.park("heat", item("heat.p"))
    queue.park("hall", item("hall.q", outputs={"heat.x": 1}))
    queue.dead_letter(item("heat.dead"), "3 deliveries")
    return claimed


def _dump(queue) -> object:
    """Everything a queue holds, apart from how long it has been idle."""
    if isinstance(queue, MemoryWorkQueue):
        return (
            list(queue._items),
            list(queue._due),
            list(queue._delayed),
            dict(queue._parked),
            list(queue._done),
            list(queue._held),
        )
    r = queue._redis
    dump = {}
    for key in sorted(r.scan_iter(match=f"{queue._ns}:*")):
        kind = r.type(key)
        if kind == "stream":
            info = r.xinfo_stream(key)
            groups = r.xinfo_groups(key)
            dump[key] = (
                r.xrange(key),
                [info[k] for k in ("last-generated-id", "entries-added")],
                [
                    {k: g[k] for k in ("last-delivered-id", "entries-read", "lag")}
                    for g in groups
                ],
                [
                    (p["message_id"], p["consumer"], p["times_delivered"])
                    for p in r.xpending_range(key, GROUP, "-", "+", 100)
                ]
                if groups
                else [],
            )
        elif kind == "zset":
            dump[key] = r.zrange(key, 0, -1, withscores=True)
        elif kind == "list":
            dump[key] = r.lrange(key, 0, -1)
        elif kind == "set":
            dump[key] = sorted(r.smembers(key))
        else:
            dump[key] = (r.get(key), r.ttl(key) > 3500)
    return dump


def test_queued_work_moves_to_the_new_names(queue):
    claimed = _fill(queue)
    before = queue.live_items()
    # Memory hands a claimed item out and forgets it.
    in_stream = 3 if isinstance(queue, RedisWorkQueue) else 2
    assert queue.count(HEAT) == {
        "queue_items": in_stream,
        "delayed": 1,
        "parked": 2,
        "done_markers": 1,
    }

    queue.rename(HEAT)

    def order(one):
        return one.entry_id, one.node, one.enqueued_at

    # Ids, deadlines, guards and everything else but the names stay.
    assert sorted(queue.live_items(), key=order) == sorted(
        [HEAT.item(one) or one for one in before], key=order
    )
    assert queue.count(HEAT) == NOTHING
    assert queue.was_done(claimed.entry_id, "warm.pump")
    assert not queue.was_done(claimed.entry_id, "heat.pump")
    assert queue.was_done("other", "heating.valve")
    assert queue.next_due() == 9e9
    assert queue.move_due(9e9) == 1
    assert [one.node for one in queue.claim(10, 10)] == [
        "warm.limit",
        "warm.valve",
        "hall.lamp",
        "heating.valve",
        "",
    ]
    assert queue.unpark("heat") == []
    assert [one.node for one in queue.unpark("warm")] == ["warm.p"]
    assert [one.outputs for one in queue.unpark("hall")] == [{"warm.x": 1}]
    if isinstance(queue, RedisWorkQueue):
        # History stays as it was.
        assert [d["node"] for d in queue.dead_letters()] == ["heat.dead"]
        assert queue._redis.xrange(queue._stream)[0][1]["node"] == "heat.acked"


def test_a_second_run_changes_nothing_and_renaming_back_restores_the_queue(queue):
    _fill(queue)
    queue.hold()
    queue.add(item("heat.held"))
    before = _dump(queue)

    queue.rename(HEAT)
    once = _dump(queue)
    queue.rename(HEAT)

    assert once != before
    assert _dump(queue) == once
    queue.rename(BACK)
    assert _dump(queue) == before


def test_held_work_is_released_in_order_behind_older_work(queue):
    queue.add(item("hall.a"))
    queue.hold()
    queue.add(item("heat.b"))
    queue.add(item("hall.c"))

    assert queue.held
    assert queue.count(HEAT)["queue_items"] == 1
    assert {one.node for one in queue.live_items()} == {"hall.a", "heat.b", "hall.c"}
    queue.rename(HEAT)
    queue.release()
    queue.add(item("hall.d"))

    assert not queue.held
    assert [one.node for one in queue.claim(10, 10)] == [
        "hall.a",
        "warm.b",
        "hall.c",
        "hall.d",
    ]


def test_a_hold_waits_for_an_add_already_writing_to_the_stream(
    redis_queue, monkeypatch
):
    """Adds do not wait for each other, but a hold waits for them, so no
    entry lands in a stream after the rebuild has read it."""
    writing, finish = threading.Event(), threading.Event()
    xadd = redis_queue._redis.xadd

    def slow(*args, **kwargs):
        writing.set()
        finish.wait(5)
        return xadd(*args, **kwargs)

    monkeypatch.setattr(redis_queue._redis, "xadd", slow)
    adding = threading.Thread(target=redis_queue.add, args=(item("heat.a"),))
    adding.start()
    assert writing.wait(5)
    holding = threading.Thread(target=redis_queue.hold)
    holding.start()
    holding.join(0.2)
    assert holding.is_alive()

    finish.set()
    holding.join(5)
    adding.join(5)
    assert redis_queue.count(HEAT)["queue_items"] == 1
    assert redis_queue.held
    redis_queue.release()


def test_a_guarded_timer_still_finds_its_guard(queue):
    state = MemoryState()
    state["__node__:heat.limit:generation"] = 3
    queue.add_delayed(item("heat.limit", guard_key="generation", guard_value="3"), 1.0)

    queue.rename(HEAT)
    state.rename_keys(HEAT)
    queue.move_due(1.0)

    (due,) = queue.claim(1, 10)
    assert (due.node, due.guard_key, due.guard_value) == (
        "warm.limit",
        "generation",
        "3",
    )
    # What the execution service compares it with (`Node.recall`).
    assert str(state.get(f"__node__:{due.node}:{due.guard_key}")) == due.guard_value


def test_a_pending_item_keeps_its_id_owner_and_count_and_is_not_repeated(
    redis_queue,
):
    redis_queue.add(item("heat.valve"))
    (claimed,) = redis_queue.claim(1, 10)
    # A second delivery, as a reclaim after a crash would make it.
    redis_queue._redis.xclaim(
        redis_queue._stream, GROUP, "engine-a", 0, [claimed.entry_id]
    )
    redis_queue.mark_done(claimed.entry_id, "heat.valve")

    redis_queue.rename(HEAT)

    (row,) = redis_queue._redis.xpending_range(redis_queue._stream, GROUP, "-", "+", 10)
    assert (row["message_id"], row["consumer"], row["times_delivered"]) == (
        claimed.entry_id,
        "engine-a",
        2,
    )
    other = _redis_queue(redis_queue._ns, "engine-b")
    try:
        (back,) = other.reclaim_stale(0)
        # What the pipeline asks before it runs a side effect again.
        assert other.was_done(back.entry_id, back.node)
    finally:
        other.close()
    assert (back.entry_id, back.node, back.deliveries) == (
        claimed.entry_id,
        "warm.valve",
        3,
    )
    marker = redis_queue._done_key(claimed.entry_id, "warm.valve")
    assert 3500 < redis_queue._redis.ttl(marker) <= 3600


def test_a_crash_before_a_stream_is_swapped_in_and_a_rerun_converge(
    redis_queue, monkeypatch
):
    _fill(redis_queue)
    before = _dump(redis_queue)

    def crash(*args):
        raise redis.ConnectionError

    with monkeypatch.context() as patch:
        patch.setattr(redis_queue._redis, "rename", crash)
        with pytest.raises(redis.ConnectionError):
            redis_queue.rename(HEAT)
    after = _dump(redis_queue)
    assert after.pop(f"{redis_queue._due_stream}:rename")
    assert after == before

    redis_queue.rename(HEAT)
    assert redis_queue.count(HEAT) == NOTHING
    redis_queue.rename(BACK)
    assert _dump(redis_queue) == before


def test_work_held_while_the_held_list_is_rewritten_survives(redis_queue, monkeypatch):
    redis_queue.hold()
    redis_queue.add(item("heat.a"))
    redis_queue.add(item("hall.b"))
    lrange = redis_queue._redis.lrange

    def racing(key, *args):
        raw = lrange(key, *args)
        if key == redis_queue._held_key and len(raw) == 2:
            redis_queue.add(item("heat.c"))
        return raw

    with monkeypatch.context() as patch:
        patch.setattr(redis_queue._redis, "lrange", racing)
        redis_queue.rename(HEAT)
    redis_queue.release()

    assert [one.node for one in redis_queue.claim(10, 10)] == [
        "warm.a",
        "hall.b",
        "heat.c",
    ]


def test_a_held_list_left_by_a_stopped_process_is_released_at_startup(
    redis_queue,
):
    redis_queue.hold()
    for i in range(150):
        redis_queue.add(item(f"hall.n{i}"))

    fresh = _redis_queue(redis_queue._ns, "engine-b")
    try:
        assert not fresh.held
        assert fresh.release_held() == 150
        assert fresh.release_held() == 0
        claimed = fresh.claim(200, 10)
    finally:
        fresh.close()
    assert [one.node for one in claimed] == [f"hall.n{i}" for i in range(150)]


def _fill_state(state) -> None:
    state.record(
        {"heat.temp": 21.5, "hall.temp": 19.0, "heating.temp": 30.0},
        {"__ts__:heat.temp": 1.0, "__ts__:hall.temp": 1.0},
        ["__version__:heat.temp", "__version__:hall.temp"],
        ts=1.0,
        limits={"heat.temp": 5},
    )
    for i in range(2, 8):
        state.append_history({"heat.temp": float(i)}, float(i), {"heat.temp": 5})
    state.update(
        {
            "__last_seen__:heat.valve:hall.temp": 1,
            "__last_seen__:hall.lamp:heat.temp": 2,
            "__in_ts__:heat.valve:heat.temp": 3.0,
            "__held__:heat.out": {"heat.temp": 1},
            "__flush__:heat.valve": 4.0,
            # A node's own state may mention any name: only its key moves.
            "__node__:heat.limit:wait:heat.temp": {"last": "heat.temp"},
            "__node__:hall.limit:wait:heat.temp": 5,
        }
    )


def _state_dump(state) -> object:
    keys = sorted(state.keys())
    names = sorted(state.message_names())
    return (
        keys,
        {k: state.get(k) for k in keys if not k.startswith((HISTORY, NAMES))},
        names,
        {name: state.history(name) for name in names},
    )


def test_state_moves_to_the_new_names_with_its_type_ttl_and_capacity(state):
    _fill_state(state)

    assert state.count(HEAT) == 10
    assert state.rename_keys(HEAT) == 10

    assert state.count(HEAT) == 0
    assert sorted(state.message_names()) == ["hall.temp", "heating.temp", "warm.temp"]
    assert not state.exists("heat.temp")
    assert state.get_multi(
        [
            "warm.temp",
            "__version__:warm.temp",
            "__last_seen__:warm.valve:hall.temp",
            "__last_seen__:hall.lamp:warm.temp",
            "__in_ts__:warm.valve:warm.temp",
            "__held__:warm.out",
            "__flush__:warm.valve",
            "__node__:warm.limit:wait:heat.temp",
            "__node__:hall.limit:wait:heat.temp",
        ]
    ) == {
        "warm.temp": 21.5,
        "__version__:warm.temp": 1,
        "__last_seen__:warm.valve:hall.temp": 1,
        "__last_seen__:hall.lamp:warm.temp": 2,
        "__in_ts__:warm.valve:warm.temp": 3.0,
        "__held__:warm.out": {"heat.temp": 1},
        "__flush__:warm.valve": 4.0,
        "__node__:warm.limit:wait:heat.temp": {"last": "heat.temp"},
        "__node__:hall.limit:wait:heat.temp": 5,
    }
    assert state.history("warm.temp") == [(float(i), float(i)) for i in range(3, 8)]
    if isinstance(state, MemoryState):
        assert state._history["warm.temp"].maxlen == 5
    else:
        history = state._key(f"{HISTORY}:warm.temp")
        assert state._client.type(history) == b"list"
        for key in (
            history,
            state._key("warm.temp"),
            state._key("__flush__:warm.valve"),
        ):
            assert 590 < state._client.ttl(key) <= 600


def test_renaming_state_twice_changes_nothing_and_back_restores_it(state):
    _fill_state(state)
    before = _state_dump(state)

    state.rename_keys(HEAT)
    once = _state_dump(state)

    assert state.rename_keys(HEAT) == 0
    assert _state_dump(state) == once
    state.rename_keys(BACK)
    assert _state_dump(state) == before


def test_a_state_rename_stopped_after_a_batch_completes_on_a_rerun(state, monkeypatch):
    monkeypatch.setattr(state_module, "RENAME_BATCH", 3)
    _fill_state(state)
    before = _state_dump(state)

    def stop(_step):
        raise ConnectionError

    with pytest.raises(ConnectionError):
        state.rename_keys(HEAT, progress=stop)
    state.rename_keys(HEAT)

    assert state.count(HEAT) == 0
    assert sorted(state.message_names()) == ["hall.temp", "heating.temp", "warm.temp"]
    state.rename_keys(BACK)
    assert _state_dump(state) == before


def test_state_is_never_moved_onto_a_key_that_exists(state):
    state["heat.temp"] = 1
    state["warm.temp"] = 2

    with pytest.raises(RuntimeError, match="exists"):
        state.rename_keys(HEAT)
    assert state.get("warm.temp") == 2


def test_another_engine_reading_the_queue_is_seen(redis_queue):
    assert redis_queue.server_version() >= (7, 0)
    redis_queue.claim(1, 10)
    assert redis_queue.other_consumers(10_000) == []

    other = _redis_queue(redis_queue._ns, "engine-b")
    try:
        other.claim(1, 10)
        assert redis_queue.other_consumers(10_000) == ["engine-b"]
        # Gone quiet for longer than asked about: a stopped engine.
        time.sleep(0.05)
        assert redis_queue.other_consumers(10) == []
    finally:
        other.close()