Sage100-vps/services/redis_service.py

import redis.asyncio as redis
from typing import Optional
import logging
import json

from config.config import settings

logger = logging.getLogger(__name__)


class RedisService:
    _instance: Optional["RedisService"] = None
    _client: Optional[redis.Redis] = None

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance

    async def connect(self) -> None:
        if self._client is not None:
            return

        try:
            self._client = redis.from_url(
                settings.redis_url,
                password=settings.redis_password,
                encoding="utf-8",
                decode_responses=True,
                socket_timeout=5.0,
                socket_connect_timeout=5.0,
            )

            await self._client.ping()
            logger.info("Connexion Redis etablie")

        except Exception as e:
            logger.error(f"Erreur connexion Redis: {e}")
            self._client = None
            raise

    async def disconnect(self) -> None:
        if self._client:
            await self._client.close()
            self._client = None
            logger.info("Connexion Redis fermee")

    async def is_connected(self) -> bool:
        if not self._client:
            return False
        try:
            await self._client.ping()
            return True
        except Exception:
            return False

    @property
    def client(self) -> redis.Redis:
        if not self._client:
            raise RuntimeError("Redis non connecte. Appelez connect() d'abord.")
        return self._client

    async def blacklist_token(self, token_id: str, ttl_seconds: int) -> bool:
        try:
            key = f"{settings.token_blacklist_prefix}{token_id}"
            await self.client.setex(key, ttl_seconds, "1")
            logger.debug(f"Token {token_id[:8]}... ajoute a la blacklist")
            return True
        except Exception as e:
            logger.error(f"Erreur blacklist token: {e}")
            return False

    async def is_token_blacklisted(self, token_id: str) -> bool:
        try:
            key = f"{settings.token_blacklist_prefix}{token_id}"
            result = await self.client.exists(key)
            return result > 0
        except Exception as e:
            logger.error(f"Erreur verification blacklist: {e}")
            return False

    async def blacklist_user_tokens(
        self, user_id: str, ttl_seconds: int = 86400
    ) -> bool:
        try:
            key = f"{settings.token_blacklist_prefix}user:{user_id}"
            import time

            await self.client.setex(key, ttl_seconds, str(int(time.time())))
            logger.info(f"Tokens utilisateur {user_id} invalides")
            return True
        except Exception as e:
            logger.error(f"Erreur invalidation tokens utilisateur: {e}")
            return False

    async def get_user_token_invalidation_time(self, user_id: str) -> Optional[int]:
        try:
            key = f"{settings.token_blacklist_prefix}user:{user_id}"
            result = await self.client.get(key)
            return int(result) if result else None
        except Exception as e:
            logger.error(f"Erreur lecture invalidation: {e}")
            return None

    async def increment_rate_limit(self, key: str, window_seconds: int) -> int:
        try:
            full_key = f"{settings.rate_limit_prefix}{key}"

            pipe = self.client.pipeline()
            pipe.incr(full_key)
            pipe.expire(full_key, window_seconds)
            results = await pipe.execute()

            return results[0]
        except Exception as e:
            logger.error(f"Erreur increment rate limit: {e}")
            return 0

    async def get_rate_limit_count(self, key: str) -> int:
        try:
            full_key = f"{settings.rate_limit_prefix}{key}"
            result = await self.client.get(full_key)
            return int(result) if result else 0
        except Exception as e:
            logger.error(f"Erreur lecture rate limit: {e}")
            return 0

    async def reset_rate_limit(self, key: str) -> bool:
        try:
            full_key = f"{settings.rate_limit_prefix}{key}"
            await self.client.delete(full_key)
            return True
        except Exception as e:
            logger.error(f"Erreur reset rate limit: {e}")
            return False

    async def store_refresh_token_metadata(
        self, token_id: str, user_id: str, fingerprint_hash: str, ttl_seconds: int
    ) -> bool:
        try:
            key = f"refresh_token:{token_id}"
            data = json.dumps(
                {
                    "user_id": user_id,
                    "fingerprint_hash": fingerprint_hash,
                    "used": False,
                }
            )
            await self.client.setex(key, ttl_seconds, data)
            return True
        except Exception as e:
            logger.error(f"Erreur stockage metadata refresh token: {e}")
            return False

    async def get_refresh_token_metadata(self, token_id: str) -> Optional[dict]:
        try:
            key = f"refresh_token:{token_id}"
            data = await self.client.get(key)
            return json.loads(data) if data else None
        except Exception as e:
            logger.error(f"Erreur lecture metadata refresh token: {e}")
            return None

    async def mark_refresh_token_used(self, token_id: str) -> bool:
        try:
            key = f"refresh_token:{token_id}"
            data = await self.client.get(key)
            if not data:
                return False

            metadata = json.loads(data)
            metadata["used"] = True
            metadata["used_at"] = int(__import__("time").time())

            ttl = await self.client.ttl(key)
            if ttl > 0:
                await self.client.setex(key, ttl, json.dumps(metadata))

            return True
        except Exception as e:
            logger.error(f"Erreur marquage refresh token: {e}")
            return False

    async def delete_refresh_token(self, token_id: str) -> bool:
        try:
            key = f"refresh_token:{token_id}"
            result = await self.client.delete(key)
            return result > 0
        except Exception as e:
            logger.error(f"Erreur suppression refresh token: {e}")
            return False


redis_service = RedisService()


async def get_redis() -> RedisService:
    if not await redis_service.is_connected():
        await redis_service.connect()
    return redis_service