"""Shared low-level primitives for machine service credentials.

Storage-agnostic, feature-agnostic helpers for any surface that issues
opaque bearer tokens to non-human callers:

* token generation (160-bit base32 secret with a caller-chosen prefix)
  and domain-separated HMAC-SHA-256 fingerprinting (no offline-dictionary
  risk on a 160-bit random secret; O(1) lookup by hash on every call);
* Redis fixed-window rate limiting with a process-local fallback so a
  Redis outage degrades the cap (per-worker) instead of removing it.

The External API (`app.api_clients`) is built on these. The
identity-consumer pull surface predates this module and keeps its own
copy; consolidating it is deferred to avoid destabilising freshly
shipped code.
"""
from __future__ import annotations

import asyncio
import base64
import logging
import secrets
import time
from datetime import UTC, datetime

from cryptography.hazmat.primitives import hashes, hmac

logger = logging.getLogger(__name__)

#: 20 bytes = 160 bits, encoded as 32 base32 chars (no padding).
_SECRET_BYTES = 20


def hash_token(plaintext: str) -> str:
    """Keyed SHA-256 fingerprint (fits a ``String(64)`` column).

    Service tokens carry 160 bits of random entropy, but a keyed digest still
    limits damage if token hashes are exfiltrated independently of the
    application secret. Legacy unkeyed SHA-256 rows are upgraded lazily by the
    owning token service when the plaintext is next presented successfully.
    """
    from app.core.config import settings

    key_material = (
        settings.CREDENTIAL_ENCRYPTION_KEY or settings.JWT_SECRET_KEY
    ).encode("utf-8")
    signer = hmac.HMAC(key_material, hashes.SHA256())
    signer.update(b"helix-service-token\x00" + plaintext.encode("utf-8"))
    return signer.finalize().hex()


def generate_token(prefix: str) -> tuple[str, str]:
    """Return ``(plaintext, prefix_display)``.

    ``plaintext`` is ``prefix`` + 32 lowercase base32 chars (the only
    time the secret exists in the clear). ``prefix_display`` is the
    prefix plus the first 2 secret chars — safe to persist so an admin
    can identify a credential at a glance.
    """
    raw = secrets.token_bytes(_SECRET_BYTES)
    body = base64.b32encode(raw).decode("ascii").rstrip("=").lower()
    plaintext = f"{prefix}{body}"
    return plaintext, plaintext[: len(prefix) + 2]


def as_aware_utc(dt: datetime | None) -> datetime | None:
    """Coerce a possibly-naive (MySQL DATETIME) value to UTC-aware.

    Helix writes are always UTC, so naive reads coerce losslessly."""
    if dt is None:
        return None
    if dt.tzinfo is None:
        return dt.replace(tzinfo=UTC)
    return dt


# ── Redis fixed-window rate limiting (with process-local fallback) ─────

_LOCAL_WINDOWS: dict[str, tuple[float, int]] = {}
_LOCAL_WINDOWS_MAX = 4096


def _local_window_incr(key: str, window: int) -> int:
    now = time.monotonic()
    started, count = _LOCAL_WINDOWS.get(key, (now, 0))
    if now - started >= window:
        started, count = now, 0
    count += 1
    if len(_LOCAL_WINDOWS) >= _LOCAL_WINDOWS_MAX and key not in _LOCAL_WINDOWS:
        for k, (st, _c) in list(_LOCAL_WINDOWS.items()):
            if now - st >= window:
                _LOCAL_WINDOWS.pop(k, None)
        while len(_LOCAL_WINDOWS) >= _LOCAL_WINDOWS_MAX:
            _LOCAL_WINDOWS.pop(next(iter(_LOCAL_WINDOWS)), None)
    _LOCAL_WINDOWS[key] = (started, count)
    return count


async def fixed_window_incr(key: str, window: int) -> int:
    """Atomic fixed-window counter (SET NX + INCR). Degrades to a
    process-local window on Redis failure — never uncapped."""
    from app.core.redis_client import get_redis

    try:
        redis = get_redis()
        async with redis.pipeline(transaction=True) as pipe:
            await pipe.set(key, 0, ex=window, nx=True)
            await pipe.incr(key)
            results = await pipe.execute()
        return int(results[1])
    except Exception as exc:  # noqa: BLE001 — degrade to local window
        logger.warning(
            "service-token rate window Redis error (local fallback): %s",
            type(exc).__name__,
        )
        await asyncio.sleep(0.05)
        return _local_window_incr(key, window)


async def peek_local_window(key: str) -> int | None:
    """Read-only peek of the process-local window (Redis-down path)."""
    entry = _LOCAL_WINDOWS.get(key)
    return entry[1] if entry is not None else None


async def window_ttl(key: str, fallback: int) -> int:
    from app.core.redis_client import get_redis

    try:
        ttl = await get_redis().ttl(key)
        return int(ttl) if ttl and ttl > 0 else fallback
    except Exception:  # noqa: BLE001
        return fallback


__all__ = [
    "as_aware_utc",
    "fixed_window_incr",
    "generate_token",
    "hash_token",
    "peek_local_window",
    "window_ttl",
]