from __future__ import annotations

import json
from dataclasses import dataclass
from typing import Any, Protocol
from uuid import uuid4

from .errors import QingflowApiError
from .json_types import JSONObject


class CloudAuthLock(Protocol):
    acquired: bool

    def release(self) -> None:
        ...


class CloudAuthCache(Protocol):
    def get_json(self, key: str) -> JSONObject | None:
        ...

    def set_json(self, key: str, value: JSONObject, ttl_seconds: int) -> None:
        ...

    def exists(self, key: str) -> bool:
        ...

    def delete(self, key: str) -> None:
        ...

    def acquire_lock(self, key: str, ttl_seconds: int) -> CloudAuthLock:
        ...


@dataclass(slots=True)
class RedisCloudAuthLock:
    cache: "RedisCloudAuthCache"
    key: str
    token: str
    acquired: bool

    def release(self) -> None:
        if not self.acquired:
            return
        self.cache.release_lock(self.key, self.token)


class RedisCloudAuthCache:
    def __init__(self, client: Any) -> None:
        self._client = client

    @classmethod
    def from_url(cls, redis_url: str) -> "RedisCloudAuthCache":
        try:
            import redis
        except ImportError as exc:
            raise QingflowApiError.config_error(
                "Cloud MCP mode requires the 'redis' Python package. Install qingflow-mcp with cloud dependencies."
            ) from exc
        return cls(redis.Redis.from_url(redis_url, decode_responses=False))

    def get_json(self, key: str) -> JSONObject | None:
        raw_value = self._client.get(key)
        if raw_value is None:
            return None
        if isinstance(raw_value, bytes):
            raw_value = raw_value.decode("utf-8")
        try:
            value = json.loads(str(raw_value))
        except json.JSONDecodeError as exc:
            raise QingflowApiError(category="cache", message=f"invalid Redis JSON for key '{key}': {exc}") from exc
        return value if isinstance(value, dict) else None

    def set_json(self, key: str, value: JSONObject, ttl_seconds: int) -> None:
        self._client.set(key, json.dumps(value, ensure_ascii=False), ex=max(1, int(ttl_seconds)))

    def exists(self, key: str) -> bool:
        return bool(self._client.exists(key))

    def delete(self, key: str) -> None:
        self._client.delete(key)

    def acquire_lock(self, key: str, ttl_seconds: int) -> RedisCloudAuthLock:
        token = uuid4().hex
        acquired = bool(self._client.set(key, token, nx=True, ex=max(1, int(ttl_seconds))))
        return RedisCloudAuthLock(cache=self, key=key, token=token, acquired=acquired)

    def release_lock(self, key: str, token: str) -> None:
        script = (
            "if redis.call('get', KEYS[1]) == ARGV[1] then "
            "return redis.call('del', KEYS[1]) "
            "else return 0 end"
        )
        self._client.eval(script, 1, key, token)