"""Recently-revoked token denylist for refresh-token rotation (U2 / U3).

When a refresh token is rotated, its hash is added to this denylist for a
short window (default 30 seconds) so a concurrent retry using the old
token can be detected.  If the old token is reused within that window,
the :class:`SessionService` treats it as **token reuse** → revokes
all sessions for the user.

This is the industry-standard refresh-token rotation pattern (Auth0,
Okta, AWS) that closes the window where an attacker who captured the
old token can still use it after the legitimate user has refreshed.

Two backends are provided:

- :class:`InMemoryRecentlyRevoked` — single-process, default for Tauri
  sidecar / dev mode.  Bounded by ``max_entries``; oldest entries are
  evicted on overflow.
- :class:`RedisRecentlyRevoked` — multi-process via Redis ``SET`` +
  ``EXPIRE``; suitable for server deployments with multiple uvicorn
  workers.

The default in-process backend is the one used by the sidecar.  Tests
that want full control over the window can pass a custom
:class:`InMemoryRecentlyRevoked` instance to :class:`SessionService`.
"""

from __future__ import annotations

import hashlib
import logging
import time
from collections import OrderedDict
from typing import Protocol

logger = logging.getLogger(__name__)

DEFAULT_TTL_SECONDS = 30
DEFAULT_MAX_ENTRIES = 10_000


def hash_token(token: str) -> str:
    """Return the canonical SHA-256 hex digest used as the denylist key."""
    return hashlib.sha256(token.encode("utf-8")).hexdigest()


class RecentlyRevoked(Protocol):
    """Storage interface for recently-revoked refresh-token hashes."""

    def add(self, token_hash: str, ttl_seconds: int = DEFAULT_TTL_SECONDS) -> None:
        """Add ``token_hash`` to the denylist for ``ttl_seconds``."""
        ...

    def contains(self, token_hash: str) -> bool:
        """Return ``True`` if the hash is still in the denylist window."""
        ...

    def clear(self) -> None:
        """Empty the denylist (used by tests)."""
        ...


class InMemoryRecentlyRevoked:
    """Bounded in-memory LRU denylist for the Tauri sidecar / dev mode.

    Entries auto-expire after ``ttl_seconds``.  When the cache exceeds
    ``max_entries`` the oldest entries are evicted.  All operations are
    O(1).
    """

    def __init__(self, max_entries: int = DEFAULT_MAX_ENTRIES) -> None:
        self._max = max_entries
        self._entries: "OrderedDict[str, float]" = OrderedDict()

    def add(self, token_hash: str, ttl_seconds: int = DEFAULT_TTL_SECONDS) -> None:
        expires_at = time.monotonic() + ttl_seconds
        # Move-to-end semantics: re-adding an existing key resets the TTL
        # and refreshes its position (LRU).
        self._entries[token_hash] = expires_at
        self._entries.move_to_end(token_hash)
        while len(self._entries) > self._max:
            self._entries.popitem(last=False)

    def contains(self, token_hash: str) -> bool:
        expires_at = self._entries.get(token_hash)
        if expires_at is None:
            return False
        if expires_at <= time.monotonic():
            # Expired — drop it and report False.
            self._entries.pop(token_hash, None)
            return False
        return True

    def clear(self) -> None:
        self._entries.clear()

    def __len__(self) -> int:  # pragma: no cover — debug helper
        return len(self._entries)


class RedisRecentlyRevoked:
    """Redis-backed denylist for multi-process server deployments.

    Keys are stored with ``SET key value EX <ttl>`` so they auto-expire.
    The ``value`` is irrelevant; only the key's existence matters.

    Falls back to a no-op (returns ``False`` from ``contains``) when the
    ``redis`` package is unavailable or the connection fails — this
    matches the policy "never block login on infrastructure that's
    transiently down".
    """

    KEY_PREFIX = "auth:revoked:"

    def __init__(self, client: object | None = None) -> None:
        self._client = client

    def add(self, token_hash: str, ttl_seconds: int = DEFAULT_TTL_SECONDS) -> None:
        if self._client is None:
            return
        try:
            self._client.set(self.KEY_PREFIX + token_hash, "1", ex=ttl_seconds)  # type: ignore[attr-defined]
        except Exception as exc:  # noqa: BLE001
            logger.warning("RedisRecentlyRevoked.add failed: %s", exc)

    def contains(self, token_hash: str) -> bool:
        if self._client is None:
            return False
        try:
            return bool(self._client.exists(self.KEY_PREFIX + token_hash))  # type: ignore[attr-defined]
        except Exception as exc:  # noqa: BLE001
            logger.warning("RedisRecentlyRevoked.contains failed: %s", exc)
            return False

    def clear(self) -> None:
        # No global flush; entries auto-expire.  This method exists for
        # test symmetry with the in-memory backend.
        return


def build_default() -> RecentlyRevoked:
    """Build the default denylist for the current process.

    Returns the in-memory implementation.  Callers (e.g.
    :class:`SessionService`) can pass a custom instance for testing
    or to use the Redis backend in production.
    """
    return InMemoryRecentlyRevoked()
