Thundering Herd.

# Locking Strategy for Thundering Herd Prevention

import asyncio
from typing import Optional, Any, Callable
import json


class LockingCache:
    """
    Cache with mutex-based thundering herd protection.
    
    Only one request fetches from database on cache miss;
    others wait for the result.
    """
    
    def __init__(self, redis_client, default_ttl: int = 300):
        self.redis = redis_client
        self.default_ttl = default_ttl
        self.lock_ttl = 10  # Lock expires after 10 seconds (safety)
    
    async def get_or_fetch(
        self,
        key: str,
        fetch_func: Callable,
        ttl: int = None
    ) -> Any:
        """
        Get from cache, or fetch with lock protection.
        
        Args:
            key: Cache key
            fetch_func: Async function to fetch data if cache miss
            ttl: Cache TTL in seconds
        """
        ttl = ttl or self.default_ttl
        
        # Step 1: Try cache
        cached = await self.redis.get(key)
        if cached:
            return json.loads(cached)
        
        # Step 2: Cache miss - try to acquire lock
        lock_key = f"lock:{key}"
        lock_acquired = await self.redis.set(
            lock_key,
            "1",
            nx=True,  # Only set if not exists
            ex=self.lock_ttl  # Expire lock after 10 seconds
        )
        
        if lock_acquired:
            # We got the lock - fetch and cache
            try:
                value = await fetch_func()
                
                # Cache the result
                await self.redis.setex(key, ttl, json.dumps(value))
                
                return value
            finally:
                # Release lock. (If the fetch outlived lock_ttl, this could delete
                # another holder's lock; production code stores a unique token and
                # deletes only if it still matches, via a small Lua script.)
                await self.redis.delete(lock_key)
        else:
            # Lock held by another request - wait and retry
            return await self._wait_for_cache(key, fetch_func, ttl)
    
    async def _wait_for_cache(
        self,
        key: str,
        fetch_func: Callable,
        ttl: int,
        max_wait: float = 5.0,
        poll_interval: float = 0.05
    ) -> Any:
        """
        Wait for another request to populate cache.
        
        Falls back to fetching if wait times out.
        """
        waited = 0
        
        while waited < max_wait:
            # Check if cache is now populated
            cached = await self.redis.get(key)
            if cached:
                return json.loads(cached)
            
            # Wait a bit and retry
            await asyncio.sleep(poll_interval)
            waited += poll_interval
        
        # Timeout - fetch ourselves (lock holder may have failed)
        value = await fetch_func()
        await self.redis.setex(key, ttl, json.dumps(value))
        return value


# Usage
cache = LockingCache(redis_client)

async def get_product(product_id: str) -> dict:
    return await cache.get_or_fetch(
        key=f"product:{product_id}",
        fetch_func=lambda: db.fetch_one(
            "SELECT * FROM products WHERE id = $1",
            product_id
        ),
        ttl=300
    )

# Request Coalescing (Single Flight Pattern)

import asyncio
from typing import Dict, Any, Callable, Optional
from dataclasses import dataclass
import json


@dataclass
class InflightRequest:
    """Represents an in-flight fetch operation."""
    future: asyncio.Future
    created_at: float


class CoalescingCache:
    """
    Cache with request coalescing for thundering herd prevention.
    
    Multiple requests for the same key share a single database fetch.
    This is more efficient than locking because there's no polling.
    """
    
    def __init__(self, redis_client, default_ttl: int = 300):
        self.redis = redis_client
        self.default_ttl = default_ttl
        
        # Track in-flight requests: key -> Future
        self._inflight: Dict[str, InflightRequest] = {}
    
    async def get_or_fetch(
        self,
        key: str,
        fetch_func: Callable,
        ttl: int = None
    ) -> Any:
        """
        Get from cache, or fetch with request coalescing.
        
        If another request is already fetching this key,
        we join that request instead of making a duplicate fetch.
        """
        ttl = ttl or self.default_ttl
        
        # Step 1: Try cache
        cached = await self.redis.get(key)
        if cached:
            return json.loads(cached)
        
        # Step 2: Check for in-flight request. There is no await between the
        # check and the registration, so in single-threaded asyncio no lock is
        # needed (and never await while holding a lock shared by all keys).
        inflight = self._inflight.get(key)
        if inflight:
            # Another request is fetching - join it
            return await inflight.future
        
        # No in-flight request - we'll be the fetcher
        future = asyncio.get_running_loop().create_future()
        self._inflight[key] = InflightRequest(
            future=future,
            created_at=asyncio.get_running_loop().time()
        )
        
        # Step 3: Fetch data (we're the designated fetcher)
        try:
            value = await fetch_func()
            
            # Cache the result
            await self.redis.setex(key, ttl, json.dumps(value))
            
            # Notify all waiting requests
            future.set_result(value)
            
            return value
            
        except Exception as e:
            # Notify waiters of failure
            future.set_exception(e)
            raise
            
        finally:
            # Clean up in-flight tracking
            self._inflight.pop(key, None)


# More robust implementation with timeout
class RobustCoalescingCache:
    """
    Production-ready coalescing cache with timeouts and cleanup.
    """
    
    def __init__(
        self,
        redis_client,
        default_ttl: int = 300,
        fetch_timeout: float = 5.0,
        cleanup_interval: float = 60.0
    ):
        self.redis = redis_client
        self.default_ttl = default_ttl
        self.fetch_timeout = fetch_timeout
        
        self._inflight: Dict[str, InflightRequest] = {}
        
        # Start cleanup task
        asyncio.create_task(self._cleanup_loop(cleanup_interval))
    
    async def get_or_fetch(
        self,
        key: str,
        fetch_func: Callable,
        ttl: int = None
    ) -> Any:
        """Get from cache with coalescing and timeout."""
        ttl = ttl or self.default_ttl
        
        # Try cache first
        cached = await self.redis.get(key)
        if cached:
            return json.loads(cached)
        
        # Join an in-flight fetch, waiting outside any lock so other keys
        # are never blocked behind this one
        inflight = self._inflight.get(key)
        if inflight:
            try:
                return await asyncio.wait_for(
                    asyncio.shield(inflight.future),
                    timeout=self.fetch_timeout
                )
            except asyncio.TimeoutError:
                # Fetcher is taking too long, fetch ourselves
                pass
        
        # Register as fetcher (no await between check and set: no lock needed)
        future = asyncio.get_running_loop().create_future()
        self._inflight[key] = InflightRequest(
            future=future,
            created_at=asyncio.get_running_loop().time()
        )
        
        # Fetch with timeout
        try:
            value = await asyncio.wait_for(
                fetch_func(),
                timeout=self.fetch_timeout
            )
            
            await self.redis.setex(key, ttl, json.dumps(value))
            
            if not future.done():
                future.set_result(value)
            
            return value
            
        except Exception as e:
            if not future.done():
                future.set_exception(e)
            raise
            
        finally:
            # Only remove our own entry (a later fetcher may have replaced it)
            if self._inflight.get(key) and self._inflight[key].future is future:
                self._inflight.pop(key, None)
    
    async def _cleanup_loop(self, interval: float):
        """Clean up stale in-flight entries."""
        while True:
            await asyncio.sleep(interval)
            
            now = asyncio.get_running_loop().time()
            stale_keys = [
                key for key, inflight in self._inflight.items()
                if now - inflight.created_at > self.fetch_timeout * 2
            ]
            for key in stale_keys:
                self._inflight.pop(key, None)

beta = 1.0     # Tuning parameter (higher = earlier refresh)
delta = 0.2    # Seconds the last recompute took
ttl_remaining = expiry_time - current_time

# XFetch (Vattani et al., VLDB 2015): -log(random()) is an exponential
# random variable, so the chance of refreshing rises sharply near expiry,
# and slow-to-recompute values (large delta) refresh earlier
should_refresh = delta * beta * -math.log(random()) >= ttl_remaining

# Or simpler version:
should_refresh = random() < (1.0 - ttl_remaining / original_ttl) * 0.1

# Probabilistic Early Expiration

import random
import math
import time
from typing import Any, Callable, Optional
from dataclasses import dataclass
import json


@dataclass
class CachedValue:
    """Cached value with expiration metadata."""
    value: Any
    created_at: float
    ttl: float
    
    @property
    def expires_at(self) -> float:
        return self.created_at + self.ttl
    
    @property
    def remaining_ttl(self) -> float:
        return max(0, self.expires_at - time.time())
    
    @property
    def ttl_fraction_remaining(self) -> float:
        """Fraction of TTL remaining (0.0 to 1.0)."""
        return self.remaining_ttl / self.ttl if self.ttl > 0 else 0


class ProbabilisticCache:
    """
    Cache with probabilistic early expiration.
    
    Randomly refreshes entries before they expire,
    preventing synchronized mass expiration.
    """
    
    def __init__(
        self,
        redis_client,
        default_ttl: int = 300,
        beta: float = 1.0,  # Higher = more aggressive refresh
        delta: float = 1.0  # Expected recompute time in seconds
    ):
        self.redis = redis_client
        self.default_ttl = default_ttl
        self.beta = beta
        self.delta = delta
    
    async def get_or_fetch(
        self,
        key: str,
        fetch_func: Callable,
        ttl: int = None
    ) -> Any:
        """
        Get from cache with probabilistic early refresh.
        """
        ttl = ttl or self.default_ttl
        
        # Get cached value with metadata
        cached_data = await self._get_with_metadata(key)
        
        if cached_data:
            # Check if we should refresh early
            if self._should_refresh_early(cached_data):
                # Refresh in background (don't block request)
                asyncio.create_task(
                    self._refresh_cache(key, fetch_func, ttl)
                )
            
            return cached_data.value
        
        # Cache miss - fetch and cache
        value = await fetch_func()
        await self._set_with_metadata(key, value, ttl)
        return value
    
    def _should_refresh_early(self, cached: CachedValue) -> bool:
        """
        Probabilistically decide if we should refresh early.
        
        Probability increases as expiration approaches.
        Uses the XFetch algorithm (Vattani et al., "Optimal Probabilistic
        Cache Stampede Prevention", VLDB 2015).
        """
        remaining = cached.remaining_ttl
        
        if remaining <= 0:
            return True  # Already expired
        
        # XFetch: refresh if delta * beta * -ln(rand) >= remaining.
        # -ln(rand) is exponentially distributed, so refreshes cluster
        # just before expiry, earlier for slow recomputes (large delta)
        
        try:
            random_factor = -1 * math.log(random.random())
            threshold = self.delta * self.beta * random_factor
            return threshold >= remaining
        except ValueError:
            return False
    
    async def _get_with_metadata(self, key: str) -> Optional[CachedValue]:
        """Get cached value with expiration metadata."""
        # Store metadata alongside value
        pipe = self.redis.pipeline()
        pipe.get(f"{key}:data")
        pipe.get(f"{key}:meta")
        
        data, meta = await pipe.execute()
        
        if not data or not meta:
            return None
        
        meta = json.loads(meta)
        return CachedValue(
            value=json.loads(data),
            created_at=meta['created_at'],
            ttl=meta['ttl']
        )
    
    async def _set_with_metadata(self, key: str, value: Any, ttl: int):
        """Set cached value with metadata."""
        now = time.time()
        
        pipe = self.redis.pipeline()
        pipe.setex(f"{key}:data", ttl, json.dumps(value))
        pipe.setex(f"{key}:meta", ttl, json.dumps({
            'created_at': now,
            'ttl': ttl
        }))
        await pipe.execute()
    
    async def _refresh_cache(self, key: str, fetch_func: Callable, ttl: int):
        """Background refresh of cache entry."""
        try:
            value = await fetch_func()
            await self._set_with_metadata(key, value, ttl)
        except Exception as e:
            # Log but don't fail - old value still valid
            logger.warning(f"Background refresh failed for {key}: {e}")


# Simpler implementation using TTL with jitter
class JitteredTTLCache:
    """
    Simpler probabilistic expiration using TTL jitter.
    
    Instead of calculating probability, just randomize the TTL.
    """
    
    def __init__(
        self,
        redis_client,
        default_ttl: int = 300,
        jitter_percent: float = 0.1  # 10% jitter
    ):
        self.redis = redis_client
        self.default_ttl = default_ttl
        self.jitter_percent = jitter_percent
    
    def _jittered_ttl(self, base_ttl: int) -> int:
        """Add random jitter to TTL."""
        jitter_range = base_ttl * self.jitter_percent
        jitter = random.uniform(-jitter_range, jitter_range)
        return max(1, int(base_ttl + jitter))
    
    async def set(self, key: str, value: Any, ttl: int = None):
        """Set with jittered TTL."""
        ttl = ttl or self.default_ttl
        actual_ttl = self._jittered_ttl(ttl)
        
        await self.redis.setex(key, actual_ttl, json.dumps(value))
    
    async def get_or_fetch(
        self,
        key: str,
        fetch_func: Callable,
        ttl: int = None
    ) -> Any:
        """Get with jittered TTL on cache population."""
        cached = await self.redis.get(key)
        if cached:
            return json.loads(cached)
        
        value = await fetch_func()
        await self.set(key, value, ttl)
        return value

# Background Refresh Strategy

import asyncio
from typing import Dict, Any, Callable, List
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
import json


@dataclass
class RefreshConfig:
    """Configuration for a cached key that needs background refresh."""
    key: str
    fetch_func: Callable
    ttl: int
    refresh_interval: int  # Refresh this many seconds before expiry
    priority: int = 1  # Higher = more important


@dataclass
class CacheEntry:
    """Track cache entry for refresh scheduling."""
    key: str
    expires_at: datetime
    config: RefreshConfig


class BackgroundRefreshCache:
    """
    Cache with background refresh.
    
    A background task proactively refreshes entries before they expire.
    This ensures cache is always warm and no thundering herd occurs.
    """
    
    def __init__(
        self,
        redis_client,
        refresh_workers: int = 4
    ):
        self.redis = redis_client
        self.refresh_workers = refresh_workers
        
        # Registry of keys to refresh
        self._registered: Dict[str, RefreshConfig] = {}
        
        # Queue of pending refreshes
        self._refresh_queue: asyncio.PriorityQueue = asyncio.PriorityQueue()
        
        self._running = False
    
    async def start(self):
        """Start background refresh workers."""
        self._running = True
        
        # Start scheduler
        asyncio.create_task(self._scheduler_loop())
        
        # Start workers
        for i in range(self.refresh_workers):
            asyncio.create_task(self._worker_loop(i))
    
    async def stop(self):
        """Stop background refresh."""
        self._running = False
    
    def register(
        self,
        key: str,
        fetch_func: Callable,
        ttl: int = 300,
        refresh_before: int = 30  # Refresh 30s before expiry
    ):
        """Register a key for background refresh."""
        self._registered[key] = RefreshConfig(
            key=key,
            fetch_func=fetch_func,
            ttl=ttl,
            refresh_interval=ttl - refresh_before
        )
    
    async def get(self, key: str) -> Any:
        """Get from cache (should always hit if registered)."""
        cached = await self.redis.get(key)
        if cached:
            return json.loads(cached)
        
        # Miss - key might not be registered or first access
        if key in self._registered:
            config = self._registered[key]
            value = await config.fetch_func()
            await self._cache_and_schedule(key, value, config)
            return value
        
        return None
    
    async def _cache_and_schedule(
        self,
        key: str,
        value: Any,
        config: RefreshConfig
    ):
        """Cache value and schedule next refresh."""
        await self.redis.setex(key, config.ttl, json.dumps(value))
        
        # Schedule refresh before expiry
        refresh_at = datetime.now(timezone.utc) + timedelta(seconds=config.refresh_interval)
        
        await self._refresh_queue.put((
            refresh_at.timestamp(),  # Priority (earlier = higher priority)
            CacheEntry(key=key, expires_at=refresh_at, config=config)
        ))
    
    async def _scheduler_loop(self):
        """Schedule refreshes at the right time."""
        while self._running:
            try:
                # Peek at next item
                if self._refresh_queue.empty():
                    await asyncio.sleep(1)
                    continue
                
                priority, entry = await self._refresh_queue.get()
                
                # Wait until refresh time
                now = datetime.now(timezone.utc).timestamp()
                if priority > now:
                    # Put back and wait
                    await self._refresh_queue.put((priority, entry))
                    await asyncio.sleep(min(1, priority - now))
                    continue
                
                # Time to refresh - requeue for worker
                await self._do_refresh(entry)
                
            except Exception as e:
                logger.error(f"Scheduler error: {e}")
                await asyncio.sleep(1)
    
    async def _do_refresh(self, entry: CacheEntry):
        """Execute a cache refresh."""
        try:
            config = entry.config
            value = await config.fetch_func()
            await self._cache_and_schedule(entry.key, value, config)
            
            logger.debug(f"Refreshed cache: {entry.key}")
            
        except Exception as e:
            logger.error(f"Failed to refresh {entry.key}: {e}")
            # Reschedule with backoff
            await asyncio.sleep(5)
            await self._refresh_queue.put((
                datetime.now(timezone.utc).timestamp() + 5,
                entry
            ))
    
    async def _worker_loop(self, worker_id: int):
        """Worker that processes refresh queue."""
        # In this implementation, scheduler does the work
        # Workers could be used for parallel refreshes
        pass


# Usage
cache = BackgroundRefreshCache(redis_client)

# Register keys for background refresh
cache.register(
    key="homepage:featured",
    fetch_func=lambda: db.fetch("SELECT * FROM featured_products"),
    ttl=60,
    refresh_before=10  # Refresh 10s before expiry
)

cache.register(
    key="deals:page",
    fetch_func=lambda: db.fetch("SELECT * FROM active_deals"),
    ttl=300,
    refresh_before=30
)

await cache.start()

# Now gets always hit (assuming registration happened before first request)
featured = await cache.get("homepage:featured")

# Stale-While-Revalidate Pattern

import asyncio
import time
from typing import Any, Callable, Optional
from dataclasses import dataclass
import json


@dataclass
class StaleableValue:
    """Value that can be served stale."""
    data: Any
    cached_at: float
    fresh_ttl: float    # Serve fresh until this
    stale_ttl: float    # Allow stale until this
    
    @property
    def is_fresh(self) -> bool:
        return time.time() < self.cached_at + self.fresh_ttl
    
    @property
    def is_stale_but_usable(self) -> bool:
        now = time.time()
        return (self.cached_at + self.fresh_ttl <= now < 
                self.cached_at + self.stale_ttl)
    
    @property
    def is_expired(self) -> bool:
        return time.time() >= self.cached_at + self.stale_ttl


class StaleWhileRevalidateCache:
    """
    Cache that serves stale content while revalidating.
    
    Provides fast responses even on cache "expiry" by serving
    slightly stale data while refreshing in background.
    """
    
    def __init__(
        self,
        redis_client,
        fresh_ttl: int = 60,     # Serve fresh for 60s
        stale_ttl: int = 300     # Allow stale for 5min total
    ):
        self.redis = redis_client
        self.fresh_ttl = fresh_ttl
        self.stale_ttl = stale_ttl
        
        # Track in-flight revalidations
        self._revalidating: set = set()
    
    async def get_or_fetch(
        self,
        key: str,
        fetch_func: Callable
    ) -> Any:
        """
        Get from cache with stale-while-revalidate.
        
        Returns:
            - Fresh data if available
            - Stale data (with background refresh) if within stale window
            - Fresh fetch if expired
        """
        cached = await self._get_with_metadata(key)
        
        if cached:
            if cached.is_fresh:
                # Fresh - return immediately
                return cached.data
            
            elif cached.is_stale_but_usable:
                # Stale but usable - return and refresh
                self._trigger_revalidation(key, fetch_func)
                return cached.data
        
        # Expired or not cached - must fetch
        value = await fetch_func()
        await self._set_with_metadata(key, value)
        return value
    
    def _trigger_revalidation(self, key: str, fetch_func: Callable):
        """Trigger background revalidation if not already running."""
        if key in self._revalidating:
            return  # Already revalidating
        
        self._revalidating.add(key)
        asyncio.create_task(self._revalidate(key, fetch_func))
    
    async def _revalidate(self, key: str, fetch_func: Callable):
        """Background revalidation."""
        try:
            value = await fetch_func()
            await self._set_with_metadata(key, value)
            logger.debug(f"Revalidated: {key}")
        except Exception as e:
            logger.warning(f"Revalidation failed for {key}: {e}")
        finally:
            self._revalidating.discard(key)
    
    async def _get_with_metadata(self, key: str) -> Optional[StaleableValue]:
        """Get cached value with staleness metadata."""
        raw = await self.redis.get(key)
        if not raw:
            return None
        
        data = json.loads(raw)
        return StaleableValue(
            data=data['value'],
            cached_at=data['cached_at'],
            fresh_ttl=data['fresh_ttl'],
            stale_ttl=data['stale_ttl']
        )
    
    async def _set_with_metadata(self, key: str, value: Any):
        """Set value with staleness metadata."""
        data = {
            'value': value,
            'cached_at': time.time(),
            'fresh_ttl': self.fresh_ttl,
            'stale_ttl': self.stale_ttl
        }
        
        # Set TTL to stale_ttl (maximum lifetime)
        await self.redis.setex(
            key,
            self.stale_ttl,
            json.dumps(data)
        )


# Usage
cache = StaleWhileRevalidateCache(
    redis_client,
    fresh_ttl=60,    # Fresh for 1 minute
    stale_ttl=300    # Stale-but-usable for 5 minutes total
)

async def get_product(product_id: str) -> dict:
    return await cache.get_or_fetch(
        key=f"product:{product_id}",
        fetch_func=lambda: db.fetch_product(product_id)
    )

# Combined Thundering Herd Protection

class ProductionCache:
    """
    Production cache combining multiple strategies.
    
    - Request coalescing for all keys
    - Probabilistic expiration for distributed cache
    - Stale-while-revalidate for user experience
    - Background refresh for known hot keys
    """
    
    def __init__(self, redis_client):
        self.redis = redis_client
        
        # In-flight tracking for coalescing
        self._inflight: Dict[str, asyncio.Future] = {}
        
        # Hot keys for background refresh
        self._hot_keys: Dict[str, RefreshConfig] = {}
        
        # Configuration
        self.fresh_ttl = 60
        self.stale_ttl = 300
        self.jitter_percent = 0.1
    
    async def get_or_fetch(
        self,
        key: str,
        fetch_func: Callable,
        allow_stale: bool = True
    ) -> Any:
        """
        Get with combined protection strategies.
        """
        # Step 1: Try cache with stale-while-revalidate
        cached = await self._get_with_staleness(key)
        
        if cached:
            if cached.is_fresh:
                return cached.data
            
            if cached.is_stale_but_usable and allow_stale:
                # Serve stale, refresh in background
                asyncio.create_task(
                    self._coalesced_refresh(key, fetch_func)
                )
                return cached.data
        
        # Step 2: Need fresh data - use coalescing
        return await self._coalesced_fetch(key, fetch_func)
    
    async def _coalesced_fetch(
        self,
        key: str,
        fetch_func: Callable
    ) -> Any:
        """Fetch with request coalescing."""
        # Join an in-flight fetch (await outside any lock, so other keys
        # never queue behind this one)
        if key in self._inflight:
            return await self._inflight[key]
        
        future = asyncio.get_running_loop().create_future()
        self._inflight[key] = future
        
        try:
            value = await fetch_func()
            await self._set_with_jitter(key, value)
            future.set_result(value)
            return value
        except Exception as e:
            future.set_exception(e)
            raise
        finally:
            self._inflight.pop(key, None)
    
    async def _set_with_jitter(self, key: str, value: Any):
        """Set with probabilistic TTL jitter."""
        jitter = random.uniform(
            -self.fresh_ttl * self.jitter_percent,
            self.fresh_ttl * self.jitter_percent
        )
        actual_ttl = int(self.fresh_ttl + jitter)
        
        data = {
            'value': value,
            'cached_at': time.time(),
            'fresh_ttl': actual_ttl,
            'stale_ttl': self.stale_ttl
        }
        
        await self.redis.setex(key, self.stale_ttl, json.dumps(data))

# Production Thundering Herd Protection

import asyncio
import json
import logging
import math
import random
import time
from dataclasses import dataclass, field
from typing import Dict, Any, Callable, Optional, Set
from enum import Enum

logger = logging.getLogger(__name__)


# =============================================================================
# Configuration
# =============================================================================

@dataclass
class ThunderingHerdConfig:
    """Configuration for thundering herd protection."""
    
    # Coalescing
    enable_coalescing: bool = True
    coalesce_timeout: float = 5.0
    
    # Probabilistic refresh
    enable_probabilistic: bool = True
    jitter_percent: float = 0.1
    early_refresh_beta: float = 1.0
    
    # Stale-while-revalidate
    enable_stale: bool = True
    fresh_ttl: int = 60
    stale_ttl: int = 300
    
    # Locking fallback
    enable_locking: bool = True
    lock_ttl: int = 10
    
    # Metrics
    enable_metrics: bool = True


# =============================================================================
# Cache Value Wrapper
# =============================================================================

@dataclass
class CacheValue:
    """Wrapped cache value with metadata."""
    data: Any
    cached_at: float
    fresh_until: float
    stale_until: float
    
    @property
    def is_fresh(self) -> bool:
        return time.time() < self.fresh_until
    
    @property
    def is_stale_usable(self) -> bool:
        now = time.time()
        return self.fresh_until <= now < self.stale_until
    
    @property
    def is_expired(self) -> bool:
        return time.time() >= self.stale_until
    
    @property
    def remaining_fresh_ttl(self) -> float:
        return max(0, self.fresh_until - time.time())
    
    def to_json(self) -> str:
        return json.dumps({
            'data': self.data,
            'cached_at': self.cached_at,
            'fresh_until': self.fresh_until,
            'stale_until': self.stale_until
        })
    
    @classmethod
    def from_json(cls, raw: str) -> 'CacheValue':
        d = json.loads(raw)
        return cls(
            data=d['data'],
            cached_at=d['cached_at'],
            fresh_until=d['fresh_until'],
            stale_until=d['stale_until']
        )


# =============================================================================
# Metrics
# =============================================================================

@dataclass
class ThunderingHerdMetrics:
    """Metrics for thundering herd protection."""
    cache_hits: int = 0
    cache_misses: int = 0
    stale_serves: int = 0
    coalesced_requests: int = 0
    locks_acquired: int = 0
    locks_waited: int = 0
    early_refreshes: int = 0
    background_refreshes: int = 0
    errors: int = 0
    
    def to_dict(self) -> dict:
        return {
            'hits': self.cache_hits,
            'misses': self.cache_misses,
            'stale_serves': self.stale_serves,
            'coalesced': self.coalesced_requests,
            'locks_acquired': self.locks_acquired,
            'locks_waited': self.locks_waited,
            'early_refreshes': self.early_refreshes,
            'errors': self.errors,
            'hit_ratio': self.cache_hits / max(1, self.cache_hits + self.cache_misses)
        }


# =============================================================================
# Main Implementation
# =============================================================================

class ThunderingHerdCache:
    """
    Production-ready cache with comprehensive thundering herd protection.
    
    Combines:
    - Request coalescing (primary)
    - Stale-while-revalidate (user experience)
    - Probabilistic early refresh (distributed cache)
    - Locking (fallback)
    """
    
    def __init__(
        self,
        redis_client,
        config: ThunderingHerdConfig = None
    ):
        self.redis = redis_client
        self.config = config or ThunderingHerdConfig()
        self.metrics = ThunderingHerdMetrics()
        
        # Coalescing state (single-threaded asyncio: no lock needed as long as
        # there is no await between checking and registering a key)
        self._inflight: Dict[str, asyncio.Future] = {}
        
        # Revalidation state
        self._revalidating: Set[str] = set()
    
    async def get_or_fetch(
        self,
        key: str,
        fetch_func: Callable[[], Any],
        fresh_ttl: int = None,
        stale_ttl: int = None,
        allow_stale: bool = None
    ) -> Any:
        """
        Get from cache with thundering herd protection.
        
        Args:
            key: Cache key
            fetch_func: Async function to fetch data on miss
            fresh_ttl: Override fresh TTL
            stale_ttl: Override stale TTL
            allow_stale: Allow serving stale data
        
        Returns:
            Cached or fetched value
        """
        fresh_ttl = fresh_ttl or self.config.fresh_ttl
        stale_ttl = stale_ttl or self.config.stale_ttl
        allow_stale = allow_stale if allow_stale is not None else self.config.enable_stale
        
        # Step 1: Try cache
        cached = await self._get_cached(key)
        
        if cached:
            # Check freshness
            if cached.is_fresh:
                self.metrics.cache_hits += 1
                
                # Probabilistic early refresh
                if self.config.enable_probabilistic:
                    if self._should_refresh_early(cached):
                        self.metrics.early_refreshes += 1
                        asyncio.create_task(
                            self._background_refresh(key, fetch_func, fresh_ttl, stale_ttl)
                        )
                
                return cached.data
            
            # Stale but usable
            if cached.is_stale_usable and allow_stale:
                self.metrics.stale_serves += 1
                
                # Trigger background refresh
                asyncio.create_task(
                    self._background_refresh(key, fetch_func, fresh_ttl, stale_ttl)
                )
                
                return cached.data
        
        # Step 2: Cache miss - need fresh data
        self.metrics.cache_misses += 1
        
        if self.config.enable_coalescing:
            return await self._coalesced_fetch(key, fetch_func, fresh_ttl, stale_ttl)
        elif self.config.enable_locking:
            return await self._locked_fetch(key, fetch_func, fresh_ttl, stale_ttl)
        else:
            return await self._simple_fetch(key, fetch_func, fresh_ttl, stale_ttl)
    
    async def _get_cached(self, key: str) -> Optional[CacheValue]:
        """Get cached value with metadata."""
        try:
            raw = await self.redis.get(key)
            if raw:
                return CacheValue.from_json(raw)
        except Exception as e:
            logger.warning(f"Cache get error for {key}: {e}")
            self.metrics.errors += 1
        return None
    
    async def _set_cached(
        self,
        key: str,
        value: Any,
        fresh_ttl: int,
        stale_ttl: int
    ):
        """Set cached value with metadata."""
        # Add jitter if enabled
        if self.config.enable_probabilistic:
            jitter = fresh_ttl * self.config.jitter_percent
            fresh_ttl = int(fresh_ttl + random.uniform(-jitter, jitter))
        
        now = time.time()
        cached = CacheValue(
            data=value,
            cached_at=now,
            fresh_until=now + fresh_ttl,
            stale_until=now + stale_ttl
        )
        
        try:
            await self.redis.setex(key, stale_ttl, cached.to_json())
        except Exception as e:
            logger.warning(f"Cache set error for {key}: {e}")
            self.metrics.errors += 1
    
    def _should_refresh_early(self, cached: CacheValue) -> bool:
        """Probabilistically decide on early refresh."""
        remaining = cached.remaining_fresh_ttl
        if remaining <= 0:
            return True
        
        # XFetch with an assumed 1-second recompute time (delta = 1)
        try:
            beta = self.config.early_refresh_beta
            threshold = beta * (-1 * math.log(random.random()))
            return threshold >= remaining
        except ValueError:
            return False
    
    async def _coalesced_fetch(
        self,
        key: str,
        fetch_func: Callable,
        fresh_ttl: int,
        stale_ttl: int
    ) -> Any:
        """Fetch with request coalescing."""
        existing = self._inflight.get(key)
        if existing:
            # Join existing fetch (waiting outside any lock)
            self.metrics.coalesced_requests += 1
            try:
                return await asyncio.wait_for(
                    asyncio.shield(existing),
                    timeout=self.config.coalesce_timeout
                )
            except asyncio.TimeoutError:
                # Timeout - fetch ourselves
                pass
        
        # We're the fetcher
        future = asyncio.get_running_loop().create_future()
        self._inflight[key] = future
        
        try:
            value = await asyncio.wait_for(
                fetch_func(),
                timeout=self.config.coalesce_timeout
            )
            
            await self._set_cached(key, value, fresh_ttl, stale_ttl)
            
            if not future.done():
                future.set_result(value)
            
            return value
            
        except Exception as e:
            if not future.done():
                future.set_exception(e)
            raise
            
        finally:
            # Only remove our own entry (a later fetcher may have replaced it)
            if self._inflight.get(key) is future:
                self._inflight.pop(key, None)
    
    async def _locked_fetch(
        self,
        key: str,
        fetch_func: Callable,
        fresh_ttl: int,
        stale_ttl: int
    ) -> Any:
        """Fetch with lock protection."""
        lock_key = f"lock:{key}"
        
        # Try to acquire lock
        acquired = await self.redis.set(
            lock_key, "1",
            nx=True,
            ex=self.config.lock_ttl
        )
        
        if acquired:
            self.metrics.locks_acquired += 1
            try:
                value = await fetch_func()
                await self._set_cached(key, value, fresh_ttl, stale_ttl)
                return value
            finally:
                await self.redis.delete(lock_key)
        else:
            # Wait for lock holder
            self.metrics.locks_waited += 1
            return await self._wait_for_cache(key, fetch_func, fresh_ttl, stale_ttl)
    
    async def _wait_for_cache(
        self,
        key: str,
        fetch_func: Callable,
        fresh_ttl: int,
        stale_ttl: int,
        max_wait: float = 5.0
    ) -> Any:
        """Wait for another request to populate cache."""
        waited = 0
        poll_interval = 0.05
        
        while waited < max_wait:
            cached = await self._get_cached(key)
            if cached and not cached.is_expired:
                return cached.data
            
            await asyncio.sleep(poll_interval)
            waited += poll_interval
        
        # Timeout - fetch ourselves
        return await self._simple_fetch(key, fetch_func, fresh_ttl, stale_ttl)
    
    async def _simple_fetch(
        self,
        key: str,
        fetch_func: Callable,
        fresh_ttl: int,
        stale_ttl: int
    ) -> Any:
        """Simple fetch without protection (fallback)."""
        value = await fetch_func()
        await self._set_cached(key, value, fresh_ttl, stale_ttl)
        return value
    
    async def _background_refresh(
        self,
        key: str,
        fetch_func: Callable,
        fresh_ttl: int,
        stale_ttl: int
    ):
        """Background cache refresh."""
        if key in self._revalidating:
            return  # Already refreshing
        
        self._revalidating.add(key)
        self.metrics.background_refreshes += 1
        
        try:
            value = await fetch_func()
            await self._set_cached(key, value, fresh_ttl, stale_ttl)
            logger.debug(f"Background refresh completed: {key}")
        except Exception as e:
            logger.warning(f"Background refresh failed for {key}: {e}")
        finally:
            self._revalidating.discard(key)
    
    def get_metrics(self) -> dict:
        """Get thundering herd protection metrics."""
        return self.metrics.to_dict()


# =============================================================================
# Helper Functions
# =============================================================================

import math  # Need for probabilistic calculation


# =============================================================================
# Usage Example
# =============================================================================

async def example_usage():
    """Example usage of ThunderingHerdCache."""
    
    # Create cache with default config
    cache = ThunderingHerdCache(
        redis_client=redis,
        config=ThunderingHerdConfig(
            fresh_ttl=60,
            stale_ttl=300,
            enable_coalescing=True,
            enable_stale=True,
            enable_probabilistic=True
        )
    )
    
    # Get product (all protections active)
    product = await cache.get_or_fetch(
        key=f"product:{product_id}",
        fetch_func=lambda: db.fetch_product(product_id)
    )
    
    # Get critical data (no stale)
    inventory = await cache.get_or_fetch(
        key=f"inventory:{product_id}",
        fetch_func=lambda: db.fetch_inventory(product_id),
        allow_stale=False,
        fresh_ttl=30
    )
    
    # Check metrics
    print(cache.get_metrics())