# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/through-what-singleflight-coalescing-and-jittered-ttl-gates-must-an-instance-guard-cache-fills-to-prevent-thundering-herd-stampedes-and-negative-cache-penetration.md
# Remnant Warrant:    https://corpusfetch.com/c/through-what-singleflight-coalescing-and-jittered-ttl-gates-must-an-instance-guard-cache-fills-to-prevent-thundering-herd-stampedes-and-negative-cache-penetration.warrant.json
# Markdown SHA-256:   cea18c7c3dc3ce9c3bab6ba9e5e7b9dbab608b008da58e0071adfc87b43df2b2
# Code AST SHA-256:   18fafe81f81f48848c5d659cb8ddbdf8e1ae67912fb6bfb2ce1760ce562d22ae
# Verified Classes:   _CacheRecord, StampedeSafeCache
# Defensive Guards:   2 fail-closed invariants (zero external deps)

import asyncio
import random
import time
from dataclasses import dataclass
from typing import Awaitable, Callable, Generic, Hashable, TypeVar

K = TypeVar("K", bound=Hashable)
V = TypeVar("V")

# Distinct internal markers so `None` (valid database not-found result) is never
# confused with a cache miss!
_CACHE_MISS = object()
_NEGATIVE_HIT = object()


@dataclass(frozen=True)
class _CacheRecord:
    stored_value: object  # Either V or _NEGATIVE_HIT
    soft_expire_monotonic: float
    hard_expire_monotonic: float


class StampedeSafeCache(Generic[K, V]):
    '''
    Protects origin databases and APIs against cache stampedes, synchronized
    TTL avalanches, and missing-key penetration via:
    1. Per-key Singleflight `asyncio.Future` request coalescing.
    2. Explicit negative caching (`_NEGATIVE_HIT`) for missing keys (`None`).
    3. Randomized TTL jitter (`±15%`) and soft-TTL stale-while-revalidate.
    '''

    def __init__(
        self,
        *,
        base_ttl_sec: float = 300.0,
        negative_ttl_sec: float = 30.0,
        jitter_ratio: float = 0.15,
    ) -> None:
        if base_ttl_sec <= 0 or negative_ttl_sec <= 0 or not (0.0 <= jitter_ratio < 0.5):
            raise ValueError("Invalid TTL or jitter configuration")
        self.base_ttl_sec = base_ttl_sec
        self.negative_ttl_sec = negative_ttl_sec
        self.jitter_ratio = jitter_ratio
        self._store: dict[K, _CacheRecord] = {}
        self._inflight: dict[K, asyncio.Future[V | None]] = {}

    def _jittered_ttl(self, ttl: float) -> float:
        low = 1.0 - self.jitter_ratio
        high = 1.0 + self.jitter_ratio
        return ttl * random.uniform(low, high)

    def _lookup_raw(self, key: K, now: float) -> tuple[object, bool]:
        '''Returns `(value_or_miss, is_stale_but_usable)`.'''
        rec = self._store.get(key)
        if rec is None:
            return _CACHE_MISS, False
        if now >= rec.hard_expire_monotonic:
            self._store.pop(key, None)
            return _CACHE_MISS, False
        is_stale = now >= rec.soft_expire_monotonic
        return rec.stored_value, is_stale

    async def _populate_singleflight(
        self,
        key: K,
        origin_loader: Callable[[K], Awaitable[V | None]],
    ) -> V | None:
        '''
        Coalesces all concurrent callers for `key` onto a single in-flight Future.
        '''
        existing_future = self._inflight.get(key)
        if existing_future is not None:
            # Gate 1: Another worker is already loading `key`; await its exact result!
            return await asyncio.shield(existing_future)

        loop = asyncio.get_running_loop()
        leader_future: asyncio.Future[V | None] = loop.create_future()
        self._inflight[key] = leader_future

        try:
            loaded_val = await origin_loader(key)
            now = time.monotonic()

            if loaded_val is None:
                # Gate 2: Cache explicit negative sentinel to block missing-key penetration
                neg_ttl = self._jittered_ttl(self.negative_ttl_sec)
                self._store[key] = _CacheRecord(
                    stored_value=_NEGATIVE_HIT,
                    soft_expire_monotonic=now + neg_ttl,
                    hard_expire_monotonic=now + neg_ttl,
                )
            else:
                # Gate 3: Apply jittered soft + hard TTLs
                hard_ttl = self._jittered_ttl(self.base_ttl_sec)
                soft_ttl = hard_ttl * 0.8
                self._store[key] = _CacheRecord(
                    stored_value=loaded_val,
                    soft_expire_monotonic=now + soft_ttl,
                    hard_expire_monotonic=now + hard_ttl,
                )

            leader_future.set_result(loaded_val)
            return loaded_val

        except BaseException as exc:
            if not leader_future.done():
                leader_future.set_exception(exc)
                # Retrieve exception on leader_future so asyncio never logs "Future exception was never retrieved"
                leader_future.exception()
            raise
        finally:
            # Gate 4: Always clear in-flight entry so future calls never hang
            self._inflight.pop(key, None)

    async def get_or_load(
        self,
        key: K,
        origin_loader: Callable[[K], Awaitable[V | None]],
    ) -> V | None:
        now = time.monotonic()
        raw_val, is_stale = self._lookup_raw(key, now)

        if raw_val is not _CACHE_MISS:
            if is_stale and key not in self._inflight:
                # Soft-TTL expired: trigger non-blocking background refresh & return stale immediately
                asyncio.create_task(self._populate_singleflight(key, origin_loader))
            return None if raw_val is _NEGATIVE_HIT else raw_val  # type: ignore[return-value]

        return await self._populate_singleflight(key, origin_loader)