import threading
import time
from collections.abc import Callable
from typing import Any

_REDIS_TEST_OBSERVED: list[int] = []
_REDIS_OBSERVED_LOCK = threading.Lock()


def redis_test_task(value: int) -> None:
    with _REDIS_OBSERVED_LOCK:
        _REDIS_TEST_OBSERVED.append(int(value))


def clear_redis_observed() -> None:
    with _REDIS_OBSERVED_LOCK:
        _REDIS_TEST_OBSERVED.clear()


def redis_observed_values() -> list[int]:
    with _REDIS_OBSERVED_LOCK:
        return list(_REDIS_TEST_OBSERVED)


class FakeRedisClient:
    def __init__(self) -> None:
        self._queues: dict[str, list[bytes]] = {}
        self._kv: dict[str, bytes] = {}
        self._lock = threading.Lock()

    def ping(self) -> bool:
        return True

    def llen(self, queue_name: str) -> int:
        with self._lock:
            return len(self._queues.get(queue_name, []))

    def rpush(self, queue_name: str, payload: str) -> int:
        with self._lock:
            queue = self._queues.setdefault(queue_name, [])
            queue.append(payload.encode("utf-8"))
            return len(queue)

    def setex(self, key: str, ttl_seconds: int, payload: str) -> bool:
        _ = int(ttl_seconds)
        with self._lock:
            self._kv[str(key)] = payload.encode("utf-8")
        return True

    def get(self, key: str):
        with self._lock:
            return self._kv.get(str(key))

    def blpop(self, queue_name: str, timeout: int = 1):
        deadline = time.time() + max(0, int(timeout))
        while True:
            with self._lock:
                queue = self._queues.setdefault(queue_name, [])
                if queue:
                    payload = queue.pop(0)
                    return (queue_name.encode("utf-8"), payload)
            if time.time() >= deadline:
                return None
            time.sleep(0.01)

    def lrange(self, queue_name: str, start: int, end: int):
        with self._lock:
            queue = list(self._queues.get(queue_name, []))
        safe_start = max(0, int(start))
        safe_end = int(end)
        if safe_end < 0:
            safe_end = len(queue) + safe_end
        safe_end = min(safe_end, len(queue) - 1)
        if safe_start > safe_end or safe_start >= len(queue):
            return []
        return queue[safe_start : safe_end + 1]

    def lpop(self, queue_name: str):
        with self._lock:
            queue = self._queues.setdefault(queue_name, [])
            if not queue:
                return None
            return queue.pop(0)

    def queue_snapshot(self, queue_name: str) -> list[str]:
        with self._lock:
            raw = list(self._queues.get(queue_name, []))
        return [item.decode("utf-8", errors="replace") for item in raw]


class AlwaysFullRedisClient(FakeRedisClient):
    def llen(self, queue_name: str) -> int:
        _ = queue_name
        return 999999


def wait_until(predicate: Callable[[], bool], timeout_s: float = 1.5) -> bool:
    deadline = time.time() + timeout_s
    while time.time() < deadline:
        if predicate():
            return True
        time.sleep(0.01)
    return False


def redis_client_factory(fake_client: FakeRedisClient) -> Callable[[str], Any]:
    return lambda _url: fake_client
