"""Per-workspace rate limiting (build plan Section 8).

A simple in-process fixed-window limiter. It is a no-op unless
``rate_limit_enabled`` is set. A Redis-backed limiter can drop in behind the same
``enforce_run_rate`` seam for multi-replica deployments.
"""

from __future__ import annotations

import time
import uuid

from legate.config import get_settings


class RateLimitError(Exception):
    """Raised when a caller exceeds their rate limit."""


class FixedWindowLimiter:
    def __init__(self) -> None:
        self._buckets: dict[tuple[str, int], int] = {}

    def allow(self, key: str, limit: int, window_sec: int) -> bool:
        bucket = int(time.time() // window_sec)
        composite = (key, bucket)
        count = self._buckets.get(composite, 0) + 1
        self._buckets[composite] = count
        if len(self._buckets) > 10_000:  # opportunistic cleanup of stale windows
            self._buckets = {k: v for k, v in self._buckets.items() if k[1] >= bucket}
        return count <= limit

    def reset(self) -> None:
        self._buckets.clear()


_limiter = FixedWindowLimiter()


def reset_limiter() -> None:
    _limiter.reset()


def enforce_run_rate(workspace_id: uuid.UUID) -> None:
    settings = get_settings()
    if not settings.rate_limit_enabled:
        return
    if not _limiter.allow(f"runs:{workspace_id}", settings.rate_limit_runs_per_minute, 60):
        raise RateLimitError(
            f"rate limit exceeded: at most {settings.rate_limit_runs_per_minute} runs per minute"
        )
