Trixy7/trixy_core/utils/ratelimit/sliding_window.py

# -*- coding: utf-8 -*-
"""
Sliding-Window-Algorithmus für Rate-Limiting.

Implementiert das Sliding-Window-Verfahren für präzise
zeitbasierte Durchsatzbegrenzung.
"""

from __future__ import annotations

import threading
import time
from collections import deque
from dataclasses import dataclass, field
from datetime import datetime
from typing import Any


@dataclass
class SlidingWindowStats:
    """
    Statistiken für ein Sliding-Window.

    Bietet Einblick in die Nutzung des Zeitfensters.
    """

    requests_allowed: int = 0
    """Anzahl erlaubter Requests."""

    requests_denied: int = 0
    """Anzahl abgelehnter Requests."""

    last_request: datetime | None = None
    """Zeitpunkt des letzten Requests."""

    peak_requests_in_window: int = 0
    """Maximale Requests in einem Fenster."""

    @property
    def total_requests(self) -> int:
        """Gesamtzahl der Requests."""
        return self.requests_allowed + self.requests_denied

    @property
    def allow_rate(self) -> float:
        """Erlaubnisrate in Prozent."""
        if self.total_requests == 0:
            return 100.0
        return (self.requests_allowed / self.total_requests) * 100

    def record_allow(self, current_count: int) -> None:
        """Zeichnet einen erlaubten Request auf."""
        self.requests_allowed += 1
        self.last_request = datetime.now()
        self.peak_requests_in_window = max(
            self.peak_requests_in_window, current_count
        )

    def record_deny(self) -> None:
        """Zeichnet einen abgelehnten Request auf."""
        self.requests_denied += 1
        self.last_request = datetime.now()

    def reset(self) -> None:
        """Setzt alle Statistiken zurück."""
        self.requests_allowed = 0
        self.requests_denied = 0
        self.last_request = None
        self.peak_requests_in_window = 0

    def to_dict(self) -> dict[str, Any]:
        """Konvertiert die Statistiken in ein Dictionary."""
        return {
            "requests_allowed": self.requests_allowed,
            "requests_denied": self.requests_denied,
            "total_requests": self.total_requests,
            "allow_rate": self.allow_rate,
            "peak_requests_in_window": self.peak_requests_in_window,
            "last_request": (
                self.last_request.isoformat()
                if self.last_request else None
            ),
        }


class SlidingWindow:
    """
    Sliding-Window-Implementierung für Rate-Limiting.

    Zählt Requests innerhalb eines gleitenden Zeitfensters.
    Im Gegensatz zum Token-Bucket bietet dies eine striktere
    Kontrolle ohne Burst-Möglichkeit.

    Attributes:
        window_size: Größe des Zeitfensters in Sekunden.
        max_requests: Maximale Requests im Fenster.

    Example:
        window = SlidingWindow(window_size=60.0, max_requests=100)

        if window.is_allowed():
            # Request im Fenster erlaubt
            process_request()
        else:
            # Rate limit erreicht
            raise TooManyRequests()
    """

    def __init__(
        self,
        window_size: float,
        max_requests: int,
    ) -> None:
        """
        Initialisiert das Sliding-Window.

        Args:
            window_size: Fenstergröße in Sekunden.
            max_requests: Maximale Requests pro Fenster.
        """
        self._window_size = window_size
        self._max_requests = max_requests
        self._timestamps: deque[float] = deque()
        self._lock = threading.Lock()
        self._stats = SlidingWindowStats()

    @property
    def window_size(self) -> float:
        """Gibt die Fenstergröße zurück."""
        return self._window_size

    @property
    def max_requests(self) -> int:
        """Gibt das Maximum zurück."""
        return self._max_requests

    @property
    def stats(self) -> SlidingWindowStats:
        """Gibt die Statistiken zurück."""
        return self._stats

    def _cleanup(self, now: float) -> None:
        """Entfernt abgelaufene Einträge."""
        cutoff = now - self._window_size
        while self._timestamps and self._timestamps[0] < cutoff:
            self._timestamps.popleft()

    @property
    def current_count(self) -> int:
        """Gibt die aktuelle Anzahl im Fenster zurück."""
        with self._lock:
            self._cleanup(time.monotonic())
            return len(self._timestamps)

    def is_allowed(self) -> bool:
        """
        Prüft, ob ein Request erlaubt ist und zeichnet ihn auf.

        Returns:
            True wenn erlaubt.
        """
        now = time.monotonic()

        with self._lock:
            self._cleanup(now)

            if len(self._timestamps) < self._max_requests:
                self._timestamps.append(now)
                self._stats.record_allow(len(self._timestamps))
                return True

            self._stats.record_deny()
            return False

    def try_acquire(self, count: int = 1) -> tuple[bool, float]:
        """
        Versucht, mehrere Slots zu reservieren.

        Args:
            count: Anzahl zu reservierender Slots.

        Returns:
            Tuple (erfolgreich, Wartezeit in Sekunden).
        """
        now = time.monotonic()

        with self._lock:
            self._cleanup(now)

            available = self._max_requests - len(self._timestamps)

            if available >= count:
                for _ in range(count):
                    self._timestamps.append(now)
                self._stats.record_allow(len(self._timestamps))
                return True, 0.0

            # Berechne Wartezeit bis genug Slots frei werden
            if self._timestamps:
                needed = count - available
                # Wann werden die nächsten 'needed' Slots frei?
                if len(self._timestamps) >= needed:
                    oldest_needed = list(self._timestamps)[needed - 1]
                    wait_time = (oldest_needed + self._window_size) - now
                    wait_time = max(0, wait_time)
                else:
                    wait_time = self._window_size
            else:
                wait_time = 0

            self._stats.record_deny()
            return False, wait_time

    def wait_and_proceed(
        self,
        max_wait: float | None = None,
    ) -> bool:
        """
        Wartet und reserviert dann einen Slot.

        Args:
            max_wait: Maximale Wartezeit in Sekunden.

        Returns:
            True wenn erfolgreich.
        """
        start = time.monotonic()

        while True:
            success, wait_time = self.try_acquire()

            if success:
                return True

            if max_wait is not None:
                elapsed = time.monotonic() - start
                remaining = max_wait - elapsed

                if remaining <= 0:
                    return False

                wait_time = min(wait_time, remaining)

            if wait_time > 0:
                time.sleep(wait_time)

    def remaining(self) -> int:
        """
        Gibt die Anzahl verbleibender Slots im Fenster zurück.

        Returns:
            Anzahl verfügbarer Slots.
        """
        with self._lock:
            self._cleanup(time.monotonic())
            return max(0, self._max_requests - len(self._timestamps))

    def time_until_available(self) -> float:
        """
        Gibt die Zeit bis zum nächsten freien Slot zurück.

        Returns:
            Wartezeit in Sekunden (0 wenn sofort verfügbar).
        """
        now = time.monotonic()

        with self._lock:
            self._cleanup(now)

            if len(self._timestamps) < self._max_requests:
                return 0.0

            if self._timestamps:
                oldest = self._timestamps[0]
                return max(0, (oldest + self._window_size) - now)

            return 0.0

    def reset(self) -> None:
        """Leert das Fenster."""
        with self._lock:
            self._timestamps.clear()

    def __repr__(self) -> str:
        """String-Repräsentation."""
        return (
            f"SlidingWindow(window_size={self._window_size}, "
            f"max_requests={self._max_requests}, "
            f"current={self.current_count})"
        )


class SlidingWindowLimiter:
    """
    Rate-Limiter basierend auf Sliding-Windows.

    Verwaltet mehrere Fenster für verschiedene Schlüssel.
    """

    def __init__(
        self,
        window_size: float,
        max_requests: int,
        key_ttl: float = 3600.0,
    ) -> None:
        """
        Initialisiert den Limiter.

        Args:
            window_size: Fenstergröße in Sekunden.
            max_requests: Maximum pro Fenster pro Schlüssel.
            key_ttl: Zeit bis ungenutzte Fenster gelöscht werden.
        """
        self._window_size = window_size
        self._max_requests = max_requests
        self._key_ttl = key_ttl
        self._windows: dict[str, tuple[SlidingWindow, float]] = {}
        self._lock = threading.Lock()

    def _get_window(self, key: str) -> SlidingWindow:
        """Holt oder erstellt ein Fenster für einen Schlüssel."""
        now = time.monotonic()

        with self._lock:
            if key in self._windows:
                window, _ = self._windows[key]
                self._windows[key] = (window, now)
                return window

            window = SlidingWindow(
                window_size=self._window_size,
                max_requests=self._max_requests,
            )
            self._windows[key] = (window, now)
            return window

    def is_allowed(self, key: str) -> bool:
        """
        Prüft, ob ein Request erlaubt ist.

        Args:
            key: Identifikator für den Rate-Limit-Kontext.

        Returns:
            True wenn erlaubt.
        """
        window = self._get_window(key)
        return window.is_allowed()

    def get_remaining(self, key: str) -> int:
        """
        Gibt die verbleibenden Requests für einen Schlüssel zurück.

        Args:
            key: Identifikator.

        Returns:
            Anzahl verbleibender Requests.
        """
        window = self._get_window(key)
        return window.remaining()

    def get_retry_after(self, key: str) -> float:
        """
        Gibt die Wartezeit bis zum nächsten freien Slot zurück.

        Args:
            key: Identifikator.

        Returns:
            Wartezeit in Sekunden.
        """
        window = self._get_window(key)
        return window.time_until_available()

    def wait_and_proceed(
        self,
        key: str,
        max_wait: float | None = None,
    ) -> bool:
        """
        Wartet und reserviert dann einen Slot.

        Args:
            key: Identifikator.
            max_wait: Maximale Wartezeit.

        Returns:
            True wenn erfolgreich.
        """
        window = self._get_window(key)
        return window.wait_and_proceed(max_wait)

    def cleanup(self) -> int:
        """
        Entfernt abgelaufene Fenster.

        Returns:
            Anzahl der entfernten Fenster.
        """
        now = time.monotonic()
        expired_keys = []

        with self._lock:
            for key, (_, last_access) in self._windows.items():
                if now - last_access > self._key_ttl:
                    expired_keys.append(key)

            for key in expired_keys:
                del self._windows[key]

        return len(expired_keys)

    def reset(self, key: str | None = None) -> None:
        """
        Setzt ein oder alle Fenster zurück.

        Args:
            key: Optionaler spezifischer Schlüssel.
        """
        with self._lock:
            if key:
                if key in self._windows:
                    window, last_access = self._windows[key]
                    window.reset()
            else:
                for window, _ in self._windows.values():
                    window.reset()

    def get_stats(self, key: str) -> SlidingWindowStats | None:
        """
        Gibt Statistiken für einen Schlüssel zurück.

        Args:
            key: Identifikator.

        Returns:
            Statistiken oder None.
        """
        with self._lock:
            if key in self._windows:
                window, _ = self._windows[key]
                return window.stats
            return None

    @property
    def window_count(self) -> int:
        """Gibt die Anzahl aktiver Fenster zurück."""
        with self._lock:
            return len(self._windows)


class SlidingWindowLog(SlidingWindow):
    """
    Erweiterte Sliding-Window-Implementierung mit Logging.

    Speichert zusätzlich Metadaten zu jedem Request für
    Debugging und Analyse.
    """

    def __init__(
        self,
        window_size: float,
        max_requests: int,
        log_size: int = 1000,
    ) -> None:
        """
        Initialisiert das Sliding-Window mit Logging.

        Args:
            window_size: Fenstergröße in Sekunden.
            max_requests: Maximale Requests pro Fenster.
            log_size: Maximale Anzahl gespeicherter Log-Einträge.
        """
        super().__init__(window_size, max_requests)
        self._log: deque[dict[str, Any]] = deque(maxlen=log_size)

    def is_allowed(self, metadata: dict[str, Any] | None = None) -> bool:
        """
        Prüft, ob ein Request erlaubt ist und loggt ihn.

        Args:
            metadata: Optionale Metadaten zum Request.

        Returns:
            True wenn erlaubt.
        """
        result = super().is_allowed()

        entry = {
            "timestamp": datetime.now().isoformat(),
            "allowed": result,
            "current_count": self.current_count,
            "remaining": self.remaining(),
        }

        if metadata:
            entry["metadata"] = metadata

        self._log.append(entry)
        return result

    def get_log(
        self,
        limit: int | None = None,
        allowed_only: bool = False,
        denied_only: bool = False,
    ) -> list[dict[str, Any]]:
        """
        Gibt das Log zurück.

        Args:
            limit: Optionale Begrenzung der Einträge.
            allowed_only: Nur erlaubte Requests.
            denied_only: Nur abgelehnte Requests.

        Returns:
            Liste von Log-Einträgen.
        """
        entries = list(self._log)

        if allowed_only:
            entries = [e for e in entries if e["allowed"]]
        elif denied_only:
            entries = [e for e in entries if not e["allowed"]]

        if limit:
            entries = entries[-limit:]

        return entries

    def clear_log(self) -> None:
        """Löscht das Log."""
        self._log.clear()