from future import annotations

import threading
import unittest
from collections import deque
from dataclasses import dataclass, field
from typing import Deque, Dict, Hashable, Optional

@dataclass(frozen=True)
class TenantConfig:
"""Rate limits for one tenant.

client_limit applies independently to each client, while tenant_limit applies
to all admitted requests from the tenant. Both limits use exact rolling
windows. active_timeout_ms bounds how long an absent fairness waiter can
retain its turn. eviction_ms controls client-state retention.
"""

client_limit: int
client_window_ms: int
tenant_limit: int
tenant_window_ms: int
active_timeout_ms: int = 60_000
eviction_ms: int = 120_000

def validate(self) -> None:
    if self.client_limit <= 0 or self.tenant_limit <= 0:
        raise ValueError("limits must be positive")
    if self.client_window_ms <= 0 or self.tenant_window_ms <= 0:
        raise ValueError("window lengths must be positive")
    if self.active_timeout_ms <= 0:
        raise ValueError("active_timeout_ms must be positive")
    if self.eviction_ms < max(self.client_window_ms, self.tenant_window_ms):
        raise ValueError("eviction_ms must be at least as large as both windows")
    if self.eviction_ms < self.active_timeout_ms:
        raise ValueError("eviction_ms must be at least active_timeout_ms")

@dataclass(frozen=True)
class AllowResult:
allowed: bool
retry_after_ms: int = 0

@dataclass
class _ClientState:
admitted: Deque[int] = field(default_factory=deque)
last_seen_ms: int = 0

@dataclass
class _TenantState:
admitted: Deque[int] = field(default_factory=deque)
clients: Dict[Hashable, _ClientState] = field(default_factory=dict)
waiters: Deque[Hashable] = field(default_factory=deque)
waiter_set: set[Hashable] = field(default_factory=set)
last_now_ms: Optional[int] = None
last_seen_ms: int = 0

class SlidingWindowRateLimiter:
"""Thread-safe, single-process, exact sliding-window rate limiter.

Sliding-window algorithm:
  Every admitted request timestamp is stored in a deque. Before evaluating a
  request, timestamps t satisfying t <= now - window are removed. Thus a
  request exactly one full window old no longer consumes capacity. This is a
  true log-based sliding window, so it has no calendar-bucket boundary burst.

  The result is exact, with O(1) amortized deque operations and memory
  proportional to admitted requests still inside configured windows. This is
  more accurate than weighted buckets but uses more memory. Limits provide a
  hard bound on live timestamp entries, and stale client objects are evicted.

Fair sharing:
  Uncontended tenant capacity is work-conserving and can be used by any
  client. Once the tenant ceiling is reached, denied clients enter a FIFO
  queue with at most one entry per client. As capacity expires, the head gets
  the next admission, then leaves the queue. A continuously busy client must
  rejoin behind other contenders, implementing round-robin redistribution
  without permanently reserving unused per-client shares. Inactive queue
  entries expire after active_timeout_ms so an abandoned turn cannot block
  the tenant forever.

Clock behavior:
  now_ms should come from a monotonic clock. If a caller supplies a duplicate
  or decreasing timestamp, time is clamped to the tenant's greatest observed
  timestamp. This prevents expired history from becoming live again.

Concurrency and deployment assumptions:
  A single re-entrant lock makes allow, configuration, and sweeping atomic
  across threads and async tasks sharing this object. For distributed use,
  the same state transition can be implemented atomically in Redis with a
  Lua script and sorted sets, or in a transactional strongly consistent
  store. FIFO waiter metadata must be updated in the same transaction as the
  request logs; local locks alone do not coordinate multiple processes.
"""

def __init__(self, default_config: Optional[TenantConfig] = None) -> None:
    if default_config is not None:
        default_config.validate()
    self._default_config = default_config
    self._configs: Dict[Hashable, TenantConfig] = {}
    self._tenants: Dict[Hashable, _TenantState] = {}
    self._lock = threading.RLock()

def configure_tenant(self, tenant_id: Hashable, config: TenantConfig) -> None:
    """Create or replace a tenant's configuration atomically."""
    config.validate()
    with self._lock:
        self._configs[tenant_id] = config

def allow(self, tenant_id: Hashable, client_id: Hashable, now_ms: int) -> AllowResult:
    """Atomically decide whether to admit one request.

    retry_after_ms is exact for a direct client/window or tenant/window
    denial. For a FIFO fairness denial it is the earliest temporal retry;
    actual admission also depends on earlier active waiters taking or
    abandoning their turns.
    """
    if not isinstance(now_ms, int):
        raise TypeError("now_ms must be an integer")

    with self._lock:
        cfg = self._config_for(tenant_id)
        tenant = self._tenants.get(tenant_id)
        if tenant is None:
            tenant = _TenantState(last_now_ms=now_ms, last_seen_ms=now_ms)
            self._tenants[tenant_id] = tenant

        now = self._normalize_time(tenant, now_ms)
        tenant.last_seen_ms = now
        self._prune(tenant.admitted, now, cfg.tenant_window_ms)

        client = tenant.clients.get(client_id)
        if client is None:
            client = _ClientState(last_seen_ms=now)
            tenant.clients[client_id] = client
        client.last_seen_ms = now
        self._prune(client.admitted, now, cfg.client_window_ms)

        self._purge_waiters(tenant, cfg, now)
        self._evict_clients(tenant, cfg, now, preserve=client_id)

        if len(client.admitted) >= cfg.client_limit:
            self._remove_waiter(tenant, client_id)
            retry = client.admitted[0] + cfg.client_window_ms - now
            return AllowResult(False, max(0, retry))

        if len(tenant.admitted) >= cfg.tenant_limit:
            self._enqueue_waiter(tenant, client_id)
            retry = tenant.admitted[0] + cfg.tenant_window_ms - now
            return AllowResult(False, max(0, retry))

        # Capacity exists. During contention, only the FIFO head may take it.
        if tenant.waiters and tenant.waiters[0] != client_id:
            self._enqueue_waiter(tenant, client_id)
            return AllowResult(False, 0)

        if tenant.waiters and tenant.waiters[0] == client_id:
            tenant.waiters.popleft()
            tenant.waiter_set.remove(client_id)

        client.admitted.append(now)
        tenant.admitted.append(now)
        return AllowResult(True, 0)

def sweep(self, now_ms: int) -> int:
    """Evict stale clients and empty tenant runtime states.

    Applications can call this periodically when traffic is low. Normal
    allow calls also clean the tenant being accessed. Returns the number of
    client and tenant runtime objects removed. Configurations are retained.
    """
    if not isinstance(now_ms, int):
        raise TypeError("now_ms must be an integer")

    removed = 0
    with self._lock:
        for tenant_id in list(self._tenants):
            cfg = self._config_for(tenant_id)
            tenant = self._tenants[tenant_id]
            now = self._normalize_time(tenant, now_ms)
            self._prune(tenant.admitted, now, cfg.tenant_window_ms)
            self._purge_waiters(tenant, cfg, now)
            before = len(tenant.clients)
            self._evict_clients(tenant, cfg, now, preserve=None)
            removed += before - len(tenant.clients)

            if (not tenant.clients and not tenant.admitted and not tenant.waiters
                    and now - tenant.last_seen_ms >= cfg.eviction_ms):
                del self._tenants[tenant_id]
                removed += 1
    return removed

def debug_state(self, tenant_id: Hashable) -> dict[str, int]:
    """Return aggregate state sizes, useful for monitoring and tests."""
    with self._lock:
        tenant = self._tenants.get(tenant_id)
        if tenant is None:
            return {"clients": 0, "tenant_events": 0, "waiters": 0,
                    "client_events": 0}
        return {
            "clients": len(tenant.clients),
            "tenant_events": len(tenant.admitted),
            "waiters": len(tenant.waiters),
            "client_events": sum(len(c.admitted) for c in tenant.clients.values()),
        }

def _config_for(self, tenant_id: Hashable) -> TenantConfig:
    cfg = self._configs.get(tenant_id, self._default_config)
    if cfg is None:
        raise KeyError(f"tenant {tenant_id!r} has no configuration")
    return cfg

@staticmethod
def _normalize_time(tenant: _TenantState, supplied: int) -> int:
    if tenant.last_now_ms is None or supplied >= tenant.last_now_ms:
        tenant.last_now_ms = supplied
        return supplied
    return tenant.last_now_ms

@staticmethod
def _prune(events: Deque[int], now: int, window_ms: int) -> None:
    boundary = now - window_ms
    while events and events[0] <= boundary:
        events.popleft()

@staticmethod
def _enqueue_waiter(tenant: _TenantState, client_id: Hashable) -> None:
    if client_id not in tenant.waiter_set:
        tenant.waiters.append(client_id)
        tenant.waiter_set.add(client_id)

@staticmethod
def _remove_waiter(tenant: _TenantState, client_id: Hashable) -> None:
    if client_id not in tenant.waiter_set:
        return
    tenant.waiters = deque(x for x in tenant.waiters if x != client_id)
    tenant.waiter_set.remove(client_id)

def _purge_waiters(self, tenant: _TenantState, cfg: TenantConfig, now: int) -> None:
    kept: Deque[Hashable] = deque()
    kept_set: set[Hashable] = set()
    for client_id in tenant.waiters:
        client = tenant.clients.get(client_id)
        if client is None or now - client.last_seen_ms >= cfg.active_timeout_ms:
            continue
        self._prune(client.admitted, now, cfg.client_window_ms)
        # A client blocked by its own quota does not reserve a tenant turn.
        if len(client.admitted) >= cfg.client_limit:
            continue
        kept.append(client_id)
        kept_set.add(client_id)
    tenant.waiters = kept
    tenant.waiter_set = kept_set

@staticmethod
def _evict_clients(tenant: _TenantState, cfg: TenantConfig, now: int,
                   preserve: Optional[Hashable]) -> None:
    for client_id in list(tenant.clients):
        if preserve is not None and client_id == preserve:
            continue
        client = tenant.clients[client_id]
        boundary = now - cfg.client_window_ms
        while client.admitted and client.admitted[0] <= boundary:
            client.admitted.popleft()
        if (not client.admitted and client_id not in tenant.waiter_set
                and now - client.last_seen_ms >= cfg.eviction_ms):
            del tenant.clients[client_id]

class SlidingWindowRateLimiterTests(unittest.TestCase):
def make_limiter(self, **overrides: int) -> SlidingWindowRateLimiter:
values = dict(client_limit=3, client_window_ms=1000,
tenant_limit=20, tenant_window_ms=1000,
active_timeout_ms=1000, eviction_ms=2000)
values.update(overrides)
return SlidingWindowRateLimiter(TenantConfig(**values))

def test_exact_window_boundary_is_expired(self) -> None:
    limiter = self.make_limiter(client_limit=1)
    self.assertTrue(limiter.allow("t", "c", 0).allowed)
    denied = limiter.allow("t", "c", 999)
    self.assertFalse(denied.allowed)
    self.assertEqual(denied.retry_after_ms, 1)
    self.assertTrue(limiter.allow("t", "c", 1000).allowed)

def test_idle_client_returns_after_full_window(self) -> None:
    limiter = self.make_limiter(client_limit=2)
    self.assertTrue(limiter.allow("t", "c", 10).allowed)
    self.assertTrue(limiter.allow("t", "c", 11).allowed)
    self.assertFalse(limiter.allow("t", "c", 12).allowed)
    self.assertTrue(limiter.allow("t", "c", 1011).allowed)

def test_duplicate_and_backwards_clock_are_clamped(self) -> None:
    limiter = self.make_limiter(client_limit=2)
    self.assertTrue(limiter.allow("t", "c", 100).allowed)
    self.assertTrue(limiter.allow("t", "c", 100).allowed)
    denied = limiter.allow("t", "c", 50)
    self.assertFalse(denied.allowed)
    self.assertEqual(denied.retry_after_ms, 1000)
    self.assertTrue(limiter.allow("t", "c", 1100).allowed)

def test_concurrent_requests_are_atomic(self) -> None:
    limiter = self.make_limiter(client_limit=10, tenant_limit=100)
    barrier = threading.Barrier(40)
    results: list[bool] = []
    results_lock = threading.Lock()

    def worker() -> None:
        barrier.wait()
        value = limiter.allow("t", "same-client", 500).allowed
        with results_lock:
            results.append(value)

    threads = [threading.Thread(target=worker) for _ in range(40)]
    for thread in threads:
        thread.start()
    for thread in threads:
        thread.join()

    self.assertEqual(sum(results), 10)
    self.assertEqual(len(results), 40)

def test_tenant_cap_combines_all_clients(self) -> None:
    limiter = self.make_limiter(client_limit=10, tenant_limit=2)
    self.assertTrue(limiter.allow("t", "a", 0).allowed)
    self.assertTrue(limiter.allow("t", "b", 1).allowed)
    denied = limiter.allow("t", "c", 2)
    self.assertFalse(denied.allowed)
    self.assertEqual(denied.retry_after_ms, 998)

def test_saturated_tenant_uses_fifo_fair_redistribution(self) -> None:
    limiter = self.make_limiter(client_limit=10, tenant_limit=2)
    self.assertTrue(limiter.allow("t", "a", 0).allowed)
    self.assertTrue(limiter.allow("t", "a", 1).allowed)

    self.assertFalse(limiter.allow("t", "b", 2).allowed)
    self.assertFalse(limiter.allow("t", "c", 3).allowed)

    # Both old slots expire, but a cannot monopolize them: b and c queued.
    self.assertFalse(limiter.allow("t", "a", 1001).allowed)
    self.assertTrue(limiter.allow("t", "b", 1001).allowed)
    self.assertFalse(limiter.allow("t", "b", 1001).allowed)
    self.assertTrue(limiter.allow("t", "c", 1001).allowed)

def test_abandoned_fairness_turn_eventually_expires(self) -> None:
    limiter = self.make_limiter(client_limit=10, tenant_limit=1,
                                active_timeout_ms=1000, eviction_ms=2000)
    self.assertTrue(limiter.allow("t", "a", 0).allowed)
    self.assertFalse(limiter.allow("t", "gone", 1).allowed)
    # At 1001 the waiter is inactive, so another client may use capacity.
    self.assertTrue(limiter.allow("t", "b", 1001).allowed)

def test_eviction_removes_stale_but_not_active_clients(self) -> None:
    limiter = self.make_limiter(client_window_ms=100, tenant_window_ms=100,
                                active_timeout_ms=100, eviction_ms=200)
    self.assertTrue(limiter.allow("t", "stale", 0).allowed)
    self.assertTrue(limiter.allow("t", "active", 150).allowed)

    limiter.sweep(201)
    state = limiter.debug_state("t")
    self.assertEqual(state["clients"], 1)
    self.assertEqual(state["client_events"], 1)

    # The active request is not dropped before its own window expires.
    self.assertFalse(limiter.allow("t", "active", 201).allowed
                     if limiter._default_config.client_limit == 1 else False)
    self.assertEqual(limiter.debug_state("t")["clients"], 1)
    limiter.sweep(351)
    self.assertEqual(limiter.debug_state("t")["clients"], 0)

def test_configuration_is_per_tenant(self) -> None:
    limiter = SlidingWindowRateLimiter()
    limiter.configure_tenant("small", TenantConfig(1, 100, 1, 100, 100, 100))
    limiter.configure_tenant("large", TenantConfig(2, 100, 3, 100, 100, 100))
    self.assertTrue(limiter.allow("small", "c", 0).allowed)
    self.assertFalse(limiter.allow("small", "c", 1).allowed)
    self.assertTrue(limiter.allow("large", "c", 0).allowed)
    self.assertTrue(limiter.allow("large", "c", 1).allowed)

def test_memory_is_bounded_by_live_limits_and_sweep(self) -> None:
    limiter = self.make_limiter(client_limit=5, tenant_limit=7,
                                client_window_ms=100, tenant_window_ms=100,
                                active_timeout_ms=100, eviction_ms=100)
    for index in range(100):
        limiter.allow("t", f"c{index}", 0)
    state = limiter.debug_state("t")
    self.assertLessEqual(state["tenant_events"], 7)
    self.assertLessEqual(state["client_events"], 7)
    limiter.sweep(100)
    self.assertEqual(limiter.debug_state("t")["clients"], 0)

if name == "main":
unittest.main()

import time
import threading
from collections import deque
import unittest
from typing import Dict, Tuple, Optional

Classe de dados auxiliar para a configuração do tenant

class TenantConfig:
def init(self, client_rate: int, client_window_ms: int, tenant_rate: int, tenant_window_ms: int):
self.client_rate = client_rate
self.client_window_ms = client_window_ms
self.tenant_rate = tenant_rate
self.tenant_window_ms = tenant_window_ms

class SlidingWindowRateLimiter:
"""
Um limitador de taxa multi-tenant, seguro para threads, usando um algoritmo de registo de janela deslizante.

Esta implementação fornece limitação de taxa por cliente sob um limite global por tenant.
Usa uma política de partilha justa em que nenhum cliente individual pode esgotar a quota do tenant
devido ao seu próprio limite individual.

Algoritmo: Registo de Janela Deslizante
- Para cada cliente e tenant, mantemos uma deque (um contentor semelhante a uma lista com inserções
  e remoções rápidas em ambas as extremidades) de timestamps dos pedidos recebidos dentro da janela.
- Quando chega um novo pedido, primeiro descartamos todos os timestamps mais antigos do que o tempo
  atual menos o tamanho da janela.
- Depois, verificamos se o número de timestamps restantes é inferior ao limite de taxa configurado.
- Se o pedido for permitido, o timestamp atual é adicionado ao registo.

Compromissos:
- Precisão: Este algoritmo é perfeitamente preciso. Acompanha corretamente a contagem de pedidos
  dentro da janela deslizante exata, evitando problemas com contadores de janela fixa em que
  picos no limite das janelas podem exceder a taxa.
- Utilização de Memória: A pegada de memória é proporcional ao limite de taxa, pois armazena um
  timestamp por pedido permitido dentro da janela. Para um cliente com uma taxa de 1000, irá
  armazenar até 1000 timestamps (aprox. 8 KB por cliente). Isto pode ser significativo para
  sistemas com um número muito grande de clientes ativos ou limites de taxa muito elevados.

Pressupostos:
- Implementação em Processo Único: O estado é armazenado em memória dentro da instância da classe. É
  seguro para threads para utilização dentro de um único processo de aplicação.
- Relógio Monotonicamente Crescente: Espera-se que o parâmetro `now_ms` seja um timestamp em milissegundos
  monotonicamente crescente. O limitador protege-se contra o tempo a andar
  para trás ignorando pedidos com timestamps mais antigos do que o mais recentemente processado.

Extensão Distribuída:
Para suportar um ambiente distribuído (por exemplo, vários servidores de API), o estado deve ser
externalizado para um armazenamento central de baixa latência, como Redis.
- Armazenamento de Estado: Um Redis Sorted Set poderia ser usado para cada registo de cliente/tenant, com tanto a
  pontuação como o valor a serem o timestamp do pedido. Isto permite consultas por intervalo eficientes.
- Atomicidade: Transações Redis (MULTI/EXEC) ou scripts Lua seriam essenciais para executar
  a operação de verificação e atualização (remover timestamps antigos, verificar contagem, adicionar novo timestamp)
  atomicamente, prevenindo condições de corrida entre diferentes servidores.
- Remoção: A expiração de chaves incorporada do Redis (`EXPIRE`) trataria naturalmente da remoção
  de dados de clientes obsoletos.
"""

def __init__(self):
    self._tenants: Dict[str, Dict] = {}
    self._lock = threading.Lock()  # Bloqueio global para adicionar/remover tenants

def add_tenant(self, tenant_id: str, config: TenantConfig):
    """Adiciona ou atualiza a configuração de um tenant."""
    with self._lock:
        if tenant_id not in self._tenants:
            self._tenants[tenant_id] = {
                "config": config,
                "clients": {},
                "tenant_log": deque(),
                "lock": threading.Lock() # Bloqueio por tenant para segurança em threads
            }
        else:
            # Atualiza a configuração em tempo real
            self._tenants[tenant_id]['config'] = config

def allow(self, tenant_id: str, client_id: str, now_ms: Optional[int] = None) -> Tuple[bool, int]:
    """
    Determina se um pedido é permitido para um determinado tenant e cliente.

    Args:
        tenant_id: O identificador do tenant.
        client_id: O identificador do cliente.
        now_ms: O timestamp atual em milissegundos. Se for None, é usado o tempo do sistema.

    Returns:
        Um tuplo (is_allowed, retry_after_ms).
        - is_allowed (bool): True se o pedido for permitido, False caso contrário.
        - retry_after_ms (int): Se negado, o número de milissegundos a esperar antes de o
          próximo pedido poder ser permitido. 0 se o pedido for permitido.
    """
    if now_ms is None:
        now_ms = int(time.time() * 1000)

    with self._lock:
        if tenant_id not in self._tenants:
            raise ValueError(f"Tenant '{tenant_id}' não configurado.")
        tenant_state = self._tenants[tenant_id]

    with tenant_state['lock']:
        config = tenant_state['config']

        # 1. Obter ou criar o estado do cliente
        if client_id not in tenant_state['clients']:
            tenant_state['clients'][client_id] = {"log": deque(), "last_seen_ms": 0}
        client_state = tenant_state['clients'][client_id]

        # Defender contra o relógio a andar para trás
        if now_ms < client_state['last_seen_ms']:
            return (False, 1) # Negar e sugerir um pequeno atraso para nova tentativa

        # 2. Remover entradas antigas e verificar o limite por cliente
        client_allowed, client_retry_ms = self._check_limit(
            client_state['log'], config.client_rate, config.client_window_ms, now_ms
        )
        if not client_allowed:
            return (False, client_retry_ms)

        # 3. Remover entradas antigas e verificar o limite por tenant
        tenant_allowed, tenant_retry_ms = self._check_limit(
            tenant_state['tenant_log'], config.tenant_rate, config.tenant_window_ms, now_ms
        )
        if not tenant_allowed:
            return (False, tenant_retry_ms)

        # 4. Se ambos forem permitidos, registar o pedido e permiti-lo
        client_state['log'].append(now_ms)
        client_state['last_seen_ms'] = now_ms
        tenant_state['tenant_log'].append(now_ms)

        # 5. Limpeza oportunista de clientes obsoletos para este tenant
        self._cleanup_stale_clients(tenant_state, now_ms)

        return (True, 0)

def _check_limit(self, log: deque, rate: int, window_ms: int, now_ms: int) -> Tuple[bool, int]:
    """Auxiliar para remover entradas antigas de um registo de timestamps e verificá-lo em relação a um limite de taxa."""
    # Remover timestamps antigos
    window_start_ms = now_ms - window_ms
    while log and log[0] <= window_start_ms:
        log.popleft()

    # Verificar se o limite é excedido
    if len(log) >= rate:
        oldest_timestamp = log[0]
        retry_after_ms = (oldest_timestamp + window_ms) - now_ms
        return (False, max(1, retry_after_ms)) # Garantir que a nova tentativa é de pelo menos 1 ms

    return (True, 0)

def _cleanup_stale_clients(self, tenant_state: Dict, now_ms: int):
    """
    Remove clientes que estiveram inativos durante muito tempo.
    Um cliente inativo é aquele cujo último pedido é mais antigo do que o dobro do tamanho da janela.
    Esta é uma limpeza oportunista, executada em linha com um pedido.
    """
    config = tenant_state['config']
    stale_threshold_ms = config.client_window_ms * 2
    stale_clients = [
        cid for cid, cstate in tenant_state['clients'].items()
        if now_ms - cstate['last_seen_ms'] > stale_threshold_ms
    ]
    for client_id in stale_clients:
        del tenant_state['clients'][client_id]

--- Conjunto de Testes ---

class TestSlidingWindowRateLimiter(unittest.TestCase):

def setUp(self):
    self.limiter = SlidingWindowRateLimiter()
    self.tenant_config = TenantConfig(
        client_rate=5,
        client_window_ms=1000,
        tenant_rate=15,
        tenant_window_ms=1000
    )
    self.limiter.add_tenant('test_tenant', self.tenant_config)
    self.now = int(time.time() * 1000)

def test_basic_client_limit(self):
    """Um único cliente deve ser limitado após exceder a sua taxa."""
    for i in range(5):
        allowed, _ = self.limiter.allow('test_tenant', 'client1', self.now + i)
        self.assertTrue(allowed)

    allowed, retry_after = self.limiter.allow('test_tenant', 'client1', self.now + 5)
    self.assertFalse(allowed)
    self.assertGreater(retry_after, 990) # Deve ser cerca de 1000 ms - 5 ms

def test_window_sliding(self):
    """Os pedidos devem voltar a ser permitidos após a janela ter passado."""
    for i in range(5):
        self.limiter.allow('test_tenant', 'client1', self.now + i)

    # Este pedido deve ser negado
    allowed, _ = self.limiter.allow('test_tenant', 'client1', self.now + 6)
    self.assertFalse(allowed)

    # Depois de a janela do primeiro pedido passar, um novo é permitido
    # O primeiro pedido foi em self.now, a janela é de 1000 ms.
    # Em self.now + 1001, o primeiro pedido expira.
    allowed, _ = self.limiter.allow('test_tenant', 'client1', self.now + 1001)
    self.assertTrue(allowed, "O pedido deve ser permitido depois de a janela deslizar")

def test_idle_client_returns(self):
    """A quota de um cliente inativo deve estar renovada quando ele regressar."""
    for i in range(5):
        self.limiter.allow('test_tenant', 'client1', self.now + i)
    
    # O cliente fica inativo durante mais do que uma janela
    new_time = self.now + 2000
    for i in range(5):
        allowed, _ = self.limiter.allow('test_tenant', 'client1', new_time + i)
        self.assertTrue(allowed, "Os pedidos devem ser permitidos para um cliente que regressa")
    
    allowed, _ = self.limiter.allow('test_tenant', 'client1', new_time + 5)
    self.assertFalse(allowed, "O cliente deve voltar a ser limitado após usar a nova quota")

def test_tenant_limit_and_fairness(self):
    """Vários clientes devem atingir coletivamente o limite do tenant, demonstrando justiça."""
    # 3 clientes, cada um faz 5 pedidos. Total de 15 pedidos, que é o limite do tenant.
    for i in range(5):
        # Cliente 1
        self.assertTrue(self.limiter.allow('test_tenant', 'c1', self.now + i)[0])
        # Cliente 2
        self.assertTrue(self.limiter.allow('test_tenant', 'c2', self.now + i)[0])
        # Cliente 3
        self.assertTrue(self.limiter.allow('test_tenant', 'c3', self.now + i)[0])

    # Neste ponto, o registo do tenant tem 15 pedidos. O próximo pedido de qualquer cliente deve falhar.
    self.assertFalse(self.limiter.allow('test_tenant', 'c4', self.now + 6)[0])
    self.assertFalse(self.limiter.allow('test_tenant', 'c1', self.now + 6)[0])

    # Testar a justiça: um cliente atinge o seu limite, mas outro pode prosseguir.
    limiter = SlidingWindowRateLimiter()
    limiter.add_tenant('fair_tenant', TenantConfig(client_rate=2, client_window_ms=1000, tenant_rate=10, tenant_window_ms=1000))
    # O cliente abusivo usa toda a sua quota
    self.assertTrue(limiter.allow('fair_tenant', 'hog', self.now)[0])
    self.assertTrue(limiter.allow('fair_tenant', 'hog', self.now + 1)[0])
    self.assertFalse(limiter.allow('fair_tenant', 'hog', self.now + 2)[0]) # O cliente abusivo é bloqueado

    # Outro cliente ainda deve conseguir fazer pedidos
    self.assertTrue(limiter.allow('fair_tenant', 'fair_client', self.now + 3)[0])

def test_stale_client_eviction(self):
    """Clientes obsoletos devem ser removidos da memória."""
    self.limiter.allow('test_tenant', 'stale_client', self.now)
    self.assertIn('stale_client', self.limiter._tenants['test_tenant']['clients'])

    # Fazer um novo pedido muito no futuro, acionando a limpeza
    future_time = self.now + self.tenant_config.client_window_ms * 3
    self.limiter.allow('test_tenant', 'active_client', future_time)

    self.assertNotIn('stale_client', self.limiter._tenants['test_tenant']['clients'])

def test_concurrency(self):
    """Deve lidar corretamente com pedidos concorrentes para o mesmo cliente."""
    limiter = SlidingWindowRateLimiter()
    limiter.add_tenant('concurrent_tenant', TenantConfig(client_rate=100, client_window_ms=1000, tenant_rate=1000, tenant_window_ms=1000))
    start_time = int(time.time() * 1000)
    num_threads = 10
    reqs_per_thread = 15
    allowed_count = [0]
    lock = threading.Lock()

    def worker():
        for _ in range(reqs_per_thread):
            now = int(time.time() * 1000)
            allowed, _ = limiter.allow('concurrent_tenant', 'shared_client', now)
            if allowed:
                with lock:
                    allowed_count[0] += 1
            time.sleep(0.001) # pequena pausa para aumentar a probabilidade de contenção

    threads = [threading.Thread(target=worker) for _ in range(num_threads)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

    # O número de pedidos permitidos deve ser exatamente o limite de taxa do cliente.
    self.assertEqual(allowed_count[0], 100)

def test_clock_regression(self):
    """Deve negar pedidos se o relógio andar para trás."""
    self.assertTrue(self.limiter.allow('test_tenant', 'client1', self.now + 100)[0])
    # Um pedido com um timestamp anterior deve ser negado
    allowed, retry_after = self.limiter.allow('test_tenant', 'client1', self.now + 99)
    self.assertFalse(allowed)
    self.assertEqual(retry_after, 1)

if name == 'main':
unittest.main()