"""Redis delivery checks, enabled by FLUKSIO_TEST_REDIS_HOST (and optional PORT)."""

import logging
import os
import uuid
from concurrent.futures import ThreadPoolExecutor

import orjson
import pytest
import redis

from fluksio.flow.namespace import touches
from fluksio.flow.queue import RedisWorkQueue, WorkItem


@pytest.fixture
def queue():
    host = os.environ.get("FLUKSIO_TEST_REDIS_HOST")
    if not host:
        pytest.skip("Set FLUKSIO_TEST_REDIS_HOST to run Redis integration checks")
    queue = RedisWorkQueue(
        host=host,
        port=int(os.environ.get("FLUKSIO_TEST_REDIS_PORT", "6379")),
        namespace=f"test-queue-{uuid.uuid4().hex}",
    )
    try:
        yield queue
    finally:
        keys = list(queue._redis.scan_iter(match=f"{queue._ns}:*"))
        if keys:
            queue._redis.delete(*keys)
        queue.close()


def test_concurrent_promotions_deliver_each_due_item_once(queue):
    for i in range(150):
        queue.add_delayed(
            WorkItem(kind="cascade", node=f"f.n{i}", flow="f", outputs={"f.x": i}),
            100.0,
        )
    queue.add_delayed(WorkItem(kind="cascade", node="f.later", flow="f"), 300.0)

    with ThreadPoolExecutor(max_workers=2) as pool:
        moved = list(pool.map(queue.move_due, [200.0, 200.0]))

    assert sorted(moved) == [50, 100]
    assert queue.move_due(200.0) == 0
    assert queue.next_due() == 300.0
    claimed = queue.claim(200, 10, due_only=True)
    assert len(claimed) == 150
    assert {item.outputs["f.x"] for item in claimed} == set(range(150))
    assert all(item.not_before == 100.0 for item in claimed)
    assert all(item.entry_id.startswith("due:") for item in claimed)


def test_a_refused_stream_write_keeps_the_timer_for_retry(queue):
    queue.add_delayed(WorkItem(kind="cascade", node="f.stop", flow="f"), 100.0)
    # A Redis command error does not roll a Lua script back. The stream write
    # must succeed before the delayed entry is removed.
    queue._redis.delete(queue._due_stream)
    queue._redis.set(queue._due_stream, "not a stream")

    with pytest.raises(redis.ResponseError, match="WRONGTYPE"):
        queue.move_due(200.0)

    assert queue.next_due() == 100.0
    queue._redis.delete(queue._due_stream)
    queue._ensure_group()
    assert queue.move_due(200.0) == 1
    assert queue.move_due(200.0) == 0
    assert [item.node for item in queue.claim(10, 10, due_only=True)] == ["f.stop"]


def test_clearing_a_flow_drops_its_queued_claimed_and_held_work(queue):
    queue.add(WorkItem(kind="cascade", node="gone.p", flow="gone"))
    (claimed,) = queue.claim(1, 10)
    assert claimed.node == "gone.p"
    queue.add(WorkItem(kind="cascade", node="gone.a", flow="gone"))
    queue.add(
        WorkItem(kind="cascade", node="kept.a", flow="kept", outputs={"gone.x": 1})
    )
    queue.add(WorkItem(kind="run", node="", flow="gone", run_id="r1"))
    queue.add_delayed(WorkItem(kind="flush", node="gone.b", flow="gone"), 9e9)
    queue.add_delayed(WorkItem(kind="flush", node="kept.b", flow="kept"), 9e9)
    queue.park("gone", WorkItem(kind="cascade", node="gone.c", flow="gone"))
    assert len(queue.live_items()) == 7

    queue.clear_flow("gone")

    assert sorted((i.kind, i.node) for i in queue.live_items()) == [
        ("cascade", "kept.a"),
        ("flush", "kept.b"),
        ("run", ""),
    ]
    # Nothing comes back to be run again: not redelivered, not reclaimed.
    assert queue.reclaim_stale(0) == []
    queue.ack(claimed)
    assert [i.node for i in queue.claim(10, 10)] == ["kept.a", ""]


def _old_layout(queue, root, *, acked):
    """What the queue wrote under the state backend's root before it moved."""
    r = queue._redis
    item = WorkItem(kind="cascade", node="home.tick", flow="home").to_fields()
    r.xgroup_create(f"{root}:__queue__", "engine", id="0", mkstream=True)
    r.xadd(f"{root}:__queue__", item)
    r.xadd(f"{root}:__queue__", item)
    # Both delivered and acknowledged, or one claimed and one never handed out.
    read = r.xreadgroup(
        "engine", "old", {f"{root}:__queue__": ">"}, None if acked else 1
    )
    if acked:
        r.xack(f"{root}:__queue__", "engine", *[i for i, _ in read[0][1]])
    flush = WorkItem(kind="flush", node="home.house", flow="home").to_fields()
    r.zadd(f"{root}:__delayed__", {orjson.dumps(flush): 1787297711.0})
    r.xadd(f"{root}:__dead__", {**item, "reason": "failed"})
    r.rpush(f"{root}:__parked__:home", orjson.dumps(item))
    r.set(f"{root}:__done__:1-0:home.tick", "1")


def test_the_old_layout_is_deleted_once_and_the_queue_kept(queue, caplog):
    root = f"{queue._ns}-old"
    queue.add(WorkItem(kind="cascade", node="f.a", flow="f"))
    queue.add_delayed(WorkItem(kind="flush", node="f.b", flow="f"), 9e9)
    queue.park("f", WorkItem(kind="cascade", node="f.c", flow="f"))
    current = set(queue._redis.scan_iter(match=f"{queue._ns}:*"))
    _old_layout(queue, root, acked=True)

    with caplog.at_level(logging.WARNING, logger="fluksio.flow.queue"):
        queue.drop_legacy(root)

    assert list(queue._redis.scan_iter(match=f"{root}:*")) == []
    assert set(queue._redis.scan_iter(match=f"{queue._ns}:*")) == current
    assert len(queue.live_items()) == 3
    logged = caplog.text
    assert "1 flush home.house" in logged and "__parked__:home" in logged
    assert "__dead__" in logged and "__done__:*' (1 keys)" in logged
    caplog.clear()
    queue.drop_legacy(root)
    assert caplog.records == []


def test_undelivered_old_work_is_kept_and_counted(queue, caplog):
    root = f"{queue._ns}-old"
    _old_layout(queue, root, acked=False)

    with caplog.at_level(logging.WARNING, logger="fluksio.flow.queue"):
        queue.drop_legacy(root)

    # Claimed and never acknowledged, and never handed out at all.
    assert set(queue._redis.scan_iter(match=f"{root}:*")) == {f"{root}:__queue__"}
    assert "2 cascade home.tick" in caplog.text
    assert [i.node for i in queue.live_items()] == ["home.tick", "home.tick"]
    assert not touches("home", "__queue__")
    queue._redis.delete(f"{root}:__queue__")