buff/backend/app/services/anti_repeat_service.py

"""
Anti-repeat call attack service
"""
import time
from typing import Tuple
from app.core.redis_cache import redis_cache
from app.core.logger import logger


class AntiRepeatService:
    """Anti-repeat call attack service"""

    @staticmethod
    async def check_and_set_repeat_lock(action: str, identifier: str, lock_duration: int = 5) -> Tuple[bool, str]:
        """
        Check and set repeat call lock

        Args:
            action: Action type (login, register, unbind, recharge)
            identifier: Identifier (usually username)
            lock_duration: Lock duration in seconds, default 5

        Returns:
            tuple: (allow_call, error_message)
        """
        try:
            # Generate lock key
            lock_key = f"repeat_lock:{action}:{identifier}"

            # Check if lock exists
            exists = await redis_cache.exists(lock_key)
            if exists:
                return False, f"Operation too frequent, please try again after {lock_duration} seconds"

            # Set lock
            success = await redis_cache.set(lock_key, {"timestamp": int(time.time())}, lock_duration)
            if not success:
                logger.warning(f"Failed to set repeat call lock: {action}:{identifier}")
                # Allow call even if setting fails (graceful degradation)
                return True, ""

            logger.debug(f"Successfully set repeat call lock: {action}:{identifier}, duration {lock_duration} seconds")
            return True, ""

        except Exception as e:
            logger.exception(f"Failed to check repeat call lock: {action}:{identifier}")
            # Allow call when exception occurs (graceful degradation)
            return True, ""


# Global instance
anti_repeat_service = AntiRepeatService()