from __future__ import annotations

import logging
import time
from typing import Any

from backend.app.core.config import settings

logger = logging.getLogger(__name__)

_fallback_mode = False
_redis_singleton: Any = None


class _InMemoryRedis:
    """Minimal in-memory Redis stand-in for environments without a Redis server.

    Only implements the subset of the Redis API used by Orsync Scenarist so the app
    can start and serve read-only / stateless endpoints on Hugging Face Spaces
    or similar platforms.
    """

    def __init__(self) -> None:
        self._data: dict[str, Any] = {}
        self._hashes: dict[str, dict[str, Any]] = {}
        self._lists: dict[str, list[Any]] = {}
        self._sets: dict[str, set[str]] = {}
        self._zsets: dict[str, dict[str, float]] = {}

    # ── String commands ───────────────────────────────────────────────
    def get(self, key: str) -> Any:
        return self._data.get(key)

    def set(self, key: str, value: Any, **kwargs: Any) -> bool:
        self._data[key] = value
        return True

    def exists(self, *keys: str) -> int:
        return sum(1 for k in keys if k in self._data)

    def delete(self, *keys: str) -> int:
        removed = 0
        for k in keys:
            if k in self._data:
                del self._data[k]
                removed += 1
        return removed

    def setex(self, key: str, ttl: int, value: Any) -> bool:
        self._data[key] = value
        return True

    def setnx(self, key: str, value: Any) -> bool:
        if key not in self._data:
            self._data[key] = value
            return True
        return False

    def incr(self, key: str) -> int:
        val = int(self._data.get(key, 0)) + 1
        self._data[key] = str(val)
        return val

    def expire(self, key: str, ttl: int) -> bool:
        return key in self._data

    # ── Hash commands ─────────────────────────────────────────────────
    def hset(self, name: str, key: str, value: Any) -> int:
        self._hashes.setdefault(name, {})[key] = value
        return 1

    def hget(self, name: str, key: str) -> Any:
        return self._hashes.get(name, {}).get(key)

    def hgetall(self, name: str) -> dict[str, Any]:
        return dict(self._hashes.get(name, {}))

    def hdel(self, name: str, *keys: str) -> int:
        h = self._hashes.get(name, {})
        removed = 0
        for k in keys:
            if k in h:
                del h[k]
                removed += 1
        return removed

    def hvals(self, name: str) -> list[Any]:
        return list(self._hashes.get(name, {}).values())

    def hlen(self, name: str) -> int:
        return len(self._hashes.get(name, {}))

    # ── List commands ─────────────────────────────────────────────────
    def lpush(self, name: str, *values: Any) -> int:
        lst = self._lists.setdefault(name, [])
        for v in values:
            lst.insert(0, v)
        return len(lst)

    def rpush(self, name: str, *values: Any) -> int:
        lst = self._lists.setdefault(name, [])
        lst.extend(values)
        return len(lst)

    def lrange(self, name: str, start: int, end: int) -> list[Any]:
        lst = self._lists.get(name, [])
        if end == -1:
            return lst[start:]
        return lst[start : end + 1]

    def llen(self, name: str) -> int:
        return len(self._lists.get(name, []))

    def lpop(self, name: str) -> Any:
        lst = self._lists.get(name, [])
        return lst.pop(0) if lst else None

    # ── Set commands ──────────────────────────────────────────────────
    def sadd(self, name: str, *values: str) -> int:
        s = self._sets.setdefault(name, set())
        before = len(s)
        s.update(values)
        return len(s) - before

    def sismember(self, name: str, value: str) -> bool:
        return value in self._sets.get(name, set())

    def smembers(self, name: str) -> set[str]:
        return set(self._sets.get(name, set()))

    def scard(self, name: str) -> int:
        return len(self._sets.get(name, set()))

    # ── Sorted-set commands ───────────────────────────────────────────
    def zadd(self, name: str, mapping: dict[str, float], **kwargs: Any) -> int:
        zs = self._zsets.setdefault(name, {})
        added = 0
        for member, score in mapping.items():
            if member not in zs:
                added += 1
            zs[member] = score
        return added

    def zrevrange(self, name: str, start: int, end: int, withscores: bool = False) -> list:
        zs = self._zsets.get(name, {})
        items = sorted(zs.items(), key=lambda x: x[1], reverse=True)
        if end == -1:
            sliced = items[start:]
        else:
            sliced = items[start : end + 1]
        if withscores:
            return sliced
        return [m for m, _ in sliced]

    def zrem(self, name: str, *values: str) -> int:
        zs = self._zsets.get(name, {})
        removed = 0
        for value in values:
            if value in zs:
                del zs[value]
                removed += 1
        return removed

    # ── Stream commands (no-op stubs) ─────────────────────────────────
    def xadd(self, name: str, fields: dict, **kwargs: Any) -> str:
        return "0-0"

    def xreadgroup(self, group: str, consumer: str, streams: dict, **kwargs: Any) -> list:
        return []

    def xgroup_create(self, name: str, group: str, **kwargs: Any) -> bool:
        return True

    def xack(self, name: str, group: str, *ids: str) -> int:
        return 0

    # ── Scan (iterator) ───────────────────────────────────────────────
    def scan_iter(self, match: str = "*", count: int = 100) -> list:
        import fnmatch
        return [k for k in self._data if fnmatch.fnmatch(k, match)]

    # ── Pipeline stub ─────────────────────────────────────────────────
    def pipeline(self, **kwargs: Any) -> "_InMemoryRedis":
        return self

    def execute(self) -> list:
        return []

    # ── Misc ──────────────────────────────────────────────────────────
    def ping(self) -> bool:
        return True

    @classmethod
    def from_url(cls, *args: Any, **kwargs: Any) -> "_InMemoryRedis":
        return cls()


def get_redis_client():  # type: ignore[return]
    """Return a Redis client, retrying briefly for an embedded server to start.

    Falls back to an in-memory stub if Redis is unreachable after retries.
    Once connected, the client is cached as a singleton.
    """
    global _fallback_mode, _redis_singleton

    if _redis_singleton is not None:
        return _redis_singleton

    if _fallback_mode:
        if _redis_singleton is None:
            _redis_singleton = _InMemoryRedis()
        return _redis_singleton

    from redis import Redis

    # Retry a few times — the embedded redis-server may still be starting
    for attempt in range(5):
        try:
            client = Redis.from_url(settings.redis_url, decode_responses=True)
            client.ping()
            _redis_singleton = client
            if attempt > 0:
                logger.info("Redis connected after %d retries", attempt)
            return client
        except Exception:
            if attempt < 4:
                time.sleep(0.5)

    logger.warning(
        "Redis is not reachable at %s — falling back to in-memory stub. "
        "Auth, sessions, outbox, and caching will be ephemeral.",
        settings.redis_url,
    )
    _fallback_mode = True
    _redis_singleton = _InMemoryRedis()
    return _redis_singleton