import threading
import time
from typing import TypeVar, Generic, Optional, Any, List

from cacheout.cache import T_ON_DELETE_CALLBACK, Cache

T = TypeVar("T")


def key_of_list(items: list) -> tuple[Any]:
    """
        对列表存储时使用原表哈希
    :param items:
    :return:
    """
    return tuple(items)


class SafeTTLCache(Generic[T]):

    def __init__(
        self, maxsize=10000, ttl: int = 60, on_delete: T_ON_DELETE_CALLBACK = None
    ):
        """
            线程安全的 TTL 缓存
        :param maxsize:最大容量
        :param ttl: 过期时间
        :param on_delete: 删除后回调
        """
        self._cache = Cache(
            maxsize=maxsize, ttl=ttl, timer=time.monotonic, on_delete=on_delete
        )
        self.ttl = ttl
        self._lock = threading.RLock()
        self.access_count_limit = 1000
        self.access_count = 0

    def _auto_refresh(self):
        with self._lock:
            self.access_count += 1
            if self.access_count >= self.access_count_limit:
                self._cache.delete_expired()
                self.access_count = 0

    def get(self, key, default: T = None, ttl=None) -> Optional[T]:
        with self._lock:
            self._auto_refresh()
            result = self._cache.get(key, default)
            if result is not None:
                if ttl is not None:
                    self._cache.set(key, result, ttl)
                else:
                    self._cache.set(key, result, self.ttl)
            return result

    def refresh(self, key, default: T = None, ttl=None) -> Optional[T]:
        """
            刷新ttl 机制, 实现通过get刷新
        :param key:
        :param default:
        :param ttl:
        :return:
        """
        self.get(key, default=default, ttl=ttl)

    def put(self, key, value, ttl=None):
        with self._lock:
            self._auto_refresh()
            if ttl is not None:
                self._cache.set(key, value, ttl)
            else:
                self._cache.set(key, value, self.ttl)

    def get_all_values(self) -> Optional[List[T]]:
        with self._lock:
            self._auto_refresh()
            return list(self._cache.values())

    def get_all_items(self) -> list[tuple[Any, T]]:
        with self._lock:
            self._auto_refresh()
            return list(self._cache.items())

    def clear(self):
        with self._lock:
            self._cache.clear()

    def contains(self, key) -> bool:
        return self._cache.has(key)