Hot Keys and Skew.

from collections import defaultdict
import time
import threading

class HotKeyDetector:
    def __init__(
        self,
        window_seconds: int = 60,
        hot_threshold: int = 10000,
        check_interval: float = 1.0
    ):
        self.window_seconds = window_seconds
        self.hot_threshold = hot_threshold
        self.check_interval = check_interval
        
        # key -> list of timestamps
        self.request_times: dict[str, list[float]] = defaultdict(list)
        self.lock = threading.Lock()
        self.hot_keys: set[str] = set()
        
        # Start background checker
        self._start_checker()
    
    def record(self, key: str) -> bool:
        """Record a request. Returns True if key is hot."""
        now = time.time()
        
        with self.lock:
            self.request_times[key].append(now)
            return key in self.hot_keys
    
    def _check_hot_keys(self):
        """Periodic check for hot keys."""
        now = time.time()
        cutoff = now - self.window_seconds
        
        with self.lock:
            new_hot_keys = set()
            keys_to_clean = []
            
            for key, times in self.request_times.items():
                # Filter to window
                recent = [t for t in times if t > cutoff]
                
                if len(recent) == 0:
                    keys_to_clean.append(key)
                else:
                    self.request_times[key] = recent
                    if len(recent) >= self.hot_threshold:
                        new_hot_keys.add(key)
            
            # Clean up cold keys
            for key in keys_to_clean:
                del self.request_times[key]
            
            # Detect newly hot keys
            newly_hot = new_hot_keys - self.hot_keys
            no_longer_hot = self.hot_keys - new_hot_keys
            
            if newly_hot:
                self._alert_new_hot_keys(newly_hot)
            if no_longer_hot:
                self._alert_cooled_keys(no_longer_hot)
            
            self.hot_keys = new_hot_keys
    
    def _alert_new_hot_keys(self, keys: set[str]):
        print(f"🔥 NEW HOT KEYS DETECTED: {keys}")
        # Send to monitoring system, trigger mitigation
    
    def _alert_cooled_keys(self, keys: set[str]):
        print(f"❄️ Keys cooled down: {keys}")
    
    def _start_checker(self):
        def run():
            while True:
                time.sleep(self.check_interval)
                self._check_hot_keys()
        
        thread = threading.Thread(target=run, daemon=True)
        thread.start()

import mmh3  # MurmurHash
import numpy as np

class CountMinSketch:
    """
    Probabilistic data structure for frequency estimation.
    Uses fixed memory regardless of number of unique keys.
    """
    
    def __init__(self, width: int = 10000, depth: int = 7):
        self.width = width
        self.depth = depth
        self.table = np.zeros((depth, width), dtype=np.int64)
    
    def _hash(self, key: str, seed: int) -> int:
        return mmh3.hash(key, seed) % self.width
    
    def increment(self, key: str, count: int = 1):
        for i in range(self.depth):
            idx = self._hash(key, i)
            self.table[i][idx] += count
    
    def estimate(self, key: str) -> int:
        """Returns estimated count (may overestimate, never underestimates)."""
        min_count = float('inf')
        for i in range(self.depth):
            idx = self._hash(key, i)
            min_count = min(min_count, self.table[i][idx])
        return int(min_count)
    
    def reset(self):
        """Reset all counters (call periodically for time-windowed counting)."""
        self.table.fill(0)


class CMSHotKeyDetector:
    """Hot key detection using Count-Min Sketch."""
    
    def __init__(self, hot_threshold: int = 10000, window_seconds: int = 60):
        self.cms = CountMinSketch()
        self.hot_threshold = hot_threshold
        self.window_seconds = window_seconds
        
        # Track potential hot keys for exact counting
        self.potential_hot_keys: dict[str, int] = {}
        self.confirmed_hot_keys: set[str] = set()
        
        # Periodic reset
        self._start_reset_timer()
    
    def record(self, key: str) -> bool:
        """Record request. Returns True if key is confirmed hot."""
        self.cms.increment(key)
        estimate = self.cms.estimate(key)
        
        # If estimate exceeds threshold, do exact counting
        if estimate >= self.hot_threshold * 0.8:  # 80% threshold for promotion
            self.potential_hot_keys[key] = self.potential_hot_keys.get(key, 0) + 1
            
            if self.potential_hot_keys[key] >= self.hot_threshold:
                if key not in self.confirmed_hot_keys:
                    self.confirmed_hot_keys.add(key)
                    self._on_hot_key_detected(key)
                return True
        
        return key in self.confirmed_hot_keys
    
    def _on_hot_key_detected(self, key: str):
        print(f"🔥 HOT KEY: {key}")
    
    def _start_reset_timer(self):
        def reset_loop():
            while True:
                time.sleep(self.window_seconds)
                self.cms.reset()
                self.potential_hot_keys.clear()
                # Keep confirmed_hot_keys until explicitly cleared
        
        threading.Thread(target=reset_loop, daemon=True).start()

# Prometheus metrics for hot key monitoring

from prometheus_client import Counter, Histogram, Gauge

# Request count per key (use with care - high cardinality)
key_requests = Counter(
    'requests_by_key_total',
    'Total requests per key',
    ['key']
)

# Hot key gauge
hot_key_count = Gauge(
    'hot_keys_count',
    'Number of currently hot keys'
)

# Request latency by partition
partition_latency = Histogram(
    'partition_request_latency_seconds',
    'Request latency by partition',
    ['partition_id'],
    buckets=[.001, .005, .01, .025, .05, .1, .25, .5, 1, 2.5, 5, 10]
)

# Partition load imbalance
partition_load = Gauge(
    'partition_requests_per_second',
    'Requests per second per partition',
    ['partition_id']
)

# Prometheus alerting rules
groups:
  - name: hot_keys
    rules:
      - alert: HotKeyDetected
        expr: rate(requests_by_key_total[1m]) > 1000
        for: 30s
        labels:
          severity: warning
        annotations:
          summary: "Hot key detected: {{ $labels.key }}"
      
      - alert: PartitionImbalance
        expr: |
          max(partition_requests_per_second) / 
          avg(partition_requests_per_second) > 5
        for: 1m
        labels:
          severity: critical
        annotations:
          summary: "Partition load imbalance detected"

class PredictiveHotKeyDetector:
    """
    Predict hot keys based on access pattern acceleration.
    A key becoming hot shows rapid growth before hitting threshold.
    """
    
    def __init__(self, growth_threshold: float = 10.0, window_seconds: int = 60):
        self.growth_threshold = growth_threshold
        self.window_seconds = window_seconds
        
        # Track request rates over time
        self.rate_history: dict[str, list[tuple[float, int]]] = defaultdict(list)
    
    def record_rate(self, key: str, current_rate: int):
        """Record current rate for a key."""
        now = time.time()
        self.rate_history[key].append((now, current_rate))
        
        # Keep only recent history
        cutoff = now - self.window_seconds
        self.rate_history[key] = [
            (t, r) for t, r in self.rate_history[key] if t > cutoff
        ]
        
        # Check for acceleration
        if self._is_accelerating(key):
            self._alert_predicted_hot_key(key, current_rate)
    
    def _is_accelerating(self, key: str) -> bool:
        history = self.rate_history[key]
        if len(history) < 3:
            return False
        
        # Compare recent rate to older rate
        recent = history[-1][1]
        older = history[0][1]
        
        if older == 0:
            return recent > 100  # Sudden spike from zero
        
        growth = recent / older
        return growth >= self.growth_threshold
    
    def _alert_predicted_hot_key(self, key: str, rate: int):
        print(f"⚠️ PREDICTED HOT KEY: {key} (rate: {rate}, accelerating)")

class ExternalSignalMonitor:
    """Monitor external signals that predict hot keys."""
    
    def __init__(self):
        self.known_upcoming_hot_keys: dict[str, dict] = {}
    
    def register_marketing_campaign(self, key: str, start_time: datetime, expected_traffic: int):
        """Marketing knows they're about to drive traffic to a URL."""
        self.known_upcoming_hot_keys[key] = {
            'type': 'marketing_campaign',
            'start_time': start_time,
            'expected_traffic': expected_traffic
        }
        self._pre_warm_caches(key)
        self._pre_scale_partition(key)
    
    def check_social_media_trends(self):
        """Poll social media for URLs being shared."""
        # Check Twitter, Reddit, etc. for your domain
        # If a URL is trending, pre-warm it
        pass
    
    def _pre_warm_caches(self, key: str):
        """Load key into cache before traffic arrives."""
        print(f"Pre-warming caches for {key}")
    
    def _pre_scale_partition(self, key: str):
        """Add capacity to the partition owning this key."""
        print(f"Pre-scaling partition for {key}")

class HotKeyCachingStrategy:
    def __init__(self, redis_client, db_client, hot_key_detector):
        self.redis = redis_client
        self.db = db_client
        self.detector = hot_key_detector
        
        # Different TTLs for hot vs normal keys
        self.normal_ttl = 300  # 5 minutes
        self.hot_key_ttl = 3600  # 1 hour (keep hot keys cached longer)
    
    def get(self, key: str) -> Optional[str]:
        # Try cache
        value = self.redis.get(key)
        if value:
            return value
        
        # Cache miss - get from DB
        value = self.db.get(key)
        if value:
            # Use longer TTL for hot keys
            ttl = self.hot_key_ttl if self.detector.is_hot(key) else self.normal_ttl
            self.redis.setex(key, ttl, value)
        
        return value

import random
import math

class StampedeProtectedCache:
    """
    Implements probabilistic early expiration to prevent cache stampede.
    Based on the XFetch algorithm.
    """
    
    def __init__(self, redis_client, db_client, beta: float = 1.0):
        self.redis = redis_client
        self.db = db_client
        self.beta = beta  # Controls early recomputation probability
    
    def get(self, key: str, ttl: int = 300) -> Optional[str]:
        # Get value with metadata
        data = self.redis.hgetall(f"cache:{key}")
        
        if not data:
            return self._recompute_and_cache(key, ttl)
        
        value = data.get('value')
        delta = float(data.get('delta', 1))  # Time to recompute
        expiry = float(data.get('expiry', 0))
        
        now = time.time()
        
        # Probabilistic early expiration
        # Higher probability of recompute as we approach expiry
        time_until_expiry = expiry - now
        
        if time_until_expiry <= 0:
            # Actually expired
            return self._recompute_and_cache(key, ttl)
        
        # XFetch: recompute early if time_until_expiry < delta * beta * -ln(random())
        if time_until_expiry < delta * self.beta * (-math.log(random.random())):
            # Early recomputation (async, don't block)
            threading.Thread(
                target=self._recompute_and_cache,
                args=(key, ttl),
                daemon=True
            ).start()
        
        return value
    
    def _recompute_and_cache(self, key: str, ttl: int) -> Optional[str]:
        start = time.time()
        value = self.db.get(key)
        delta = time.time() - start  # How long recomputation took
        
        if value:
            expiry = time.time() + ttl
            self.redis.hset(f"cache:{key}", mapping={
                'value': value,
                'delta': delta,
                'expiry': expiry
            })
            self.redis.expire(f"cache:{key}", ttl + 60)  # Extra buffer
        
        return value

class HotKeyReplicaRouter:
    def __init__(self, partition_map, hot_key_replicas: dict[str, list]):
        self.partition_map = partition_map
        self.hot_key_replicas = hot_key_replicas  # key -> list of replica connections
    
    def get_connection(self, key: str):
        if key in self.hot_key_replicas:
            # Spread randomly across hot key replicas
            replicas = self.hot_key_replicas[key]
            return random.choice(replicas)
        else:
            # Normal routing
            partition = self.partition_map.get_partition(key)
            return partition.get_replica()

class SplitKeyStrategy:
    def __init__(self, split_factor: int = 10):
        self.split_factor = split_factor
        self.hot_keys: set[str] = set()
    
    def mark_hot(self, key: str):
        """Mark a key as hot, enabling splitting."""
        self.hot_keys.add(key)
    
    def mark_cold(self, key: str):
        """Mark a key as no longer hot."""
        self.hot_keys.discard(key)
    
    def get_read_key(self, key: str) -> str:
        """Get the physical key for reading."""
        if key in self.hot_keys:
            suffix = random.randint(0, self.split_factor - 1)
            return f"{key}:{suffix}"
        return key
    
    def get_write_keys(self, key: str) -> list[str]:
        """Get all physical keys for writing."""
        if key in self.hot_keys:
            return [f"{key}:{i}" for i in range(self.split_factor)]
        return [key]
    
    def read(self, key: str, cache) -> Optional[str]:
        """Read from a random split."""
        physical_key = self.get_read_key(key)
        return cache.get(physical_key)
    
    def write(self, key: str, value: str, cache):
        """Write to all splits."""
        physical_keys = self.get_write_keys(key)
        for pk in physical_keys:
            cache.set(pk, value)

from functools import lru_cache
from cachetools import TTLCache
import threading

class TwoLevelCache:
    def __init__(self, redis_client, db_client, local_max_size: int = 1000, local_ttl: int = 10):
        self.redis = redis_client
        self.db = db_client
        
        # Local in-memory cache with TTL
        self.local_cache = TTLCache(maxsize=local_max_size, ttl=local_ttl)
        self.local_lock = threading.Lock()
        
        # Track hot keys for local caching
        self.hot_keys: set[str] = set()
    
    def get(self, key: str) -> Optional[str]:
        # Level 1: Local cache (only for hot keys)
        if key in self.hot_keys:
            with self.local_lock:
                if key in self.local_cache:
                    return self.local_cache[key]
        
        # Level 2: Redis
        value = self.redis.get(key)
        if value:
            if key in self.hot_keys:
                with self.local_lock:
                    self.local_cache[key] = value
            return value
        
        # Level 3: Database
        value = self.db.get(key)
        if value:
            self.redis.setex(key, 300, value)
            if key in self.hot_keys:
                with self.local_lock:
                    self.local_cache[key] = value
        
        return value
    
    def mark_hot(self, key: str):
        self.hot_keys.add(key)
    
    def invalidate(self, key: str):
        """Invalidate across all levels."""
        with self.local_lock:
            self.local_cache.pop(key, None)
        self.redis.delete(key)

class DistributedLocalCache:
    def __init__(self, redis_client, channel: str = "cache_invalidation"):
        self.redis = redis_client
        self.channel = channel
        self.local_cache = TTLCache(maxsize=1000, ttl=10)
        
        # Subscribe to invalidation messages
        self._start_subscriber()
    
    def _start_subscriber(self):
        def listen():
            pubsub = self.redis.pubsub()
            pubsub.subscribe(self.channel)
            
            for message in pubsub.listen():
                if message['type'] == 'message':
                    key = message['data']
                    self.local_cache.pop(key, None)
        
        threading.Thread(target=listen, daemon=True).start()
    
    def invalidate(self, key: str):
        """Invalidate locally and broadcast to other servers."""
        self.local_cache.pop(key, None)
        self.redis.publish(self.channel, key)

import asyncio
from typing import Dict, Optional
from collections import defaultdict

class RequestCoalescer:
    """
    Coalesce multiple concurrent requests for the same key
    into a single database query.
    """
    
    def __init__(self, db_client):
        self.db = db_client
        self.pending: Dict[str, asyncio.Future] = {}
        self.lock = asyncio.Lock()
    
    async def get(self, key: str) -> Optional[str]:
        async with self.lock:
            if key in self.pending:
                # Another request is already fetching this key
                # Wait for its result
                return await self.pending[key]
            
            # We're the first - create a future for others to wait on
            future = asyncio.Future()
            self.pending[key] = future
        
        try:
            # Fetch from database
            value = await self.db.get(key)
            future.set_result(value)
            return value
        except Exception as e:
            future.set_exception(e)
            raise
        finally:
            async with self.lock:
                del self.pending[key]


# Synchronous version using threading
class SyncRequestCoalescer:
    def __init__(self, db_client):
        self.db = db_client
        self.pending: Dict[str, threading.Event] = {}
        self.results: Dict[str, any] = {}
        self.lock = threading.Lock()
    
    def get(self, key: str) -> Optional[str]:
        with self.lock:
            if key in self.pending:
                event = self.pending[key]
            else:
                event = threading.Event()
                self.pending[key] = event
                
                # We're the leader - start the fetch
                threading.Thread(
                    target=self._fetch,
                    args=(key, event),
                    daemon=True
                ).start()
        
        # Wait for result
        event.wait(timeout=5.0)
        
        with self.lock:
            return self.results.get(key)
    
    def _fetch(self, key: str, event: threading.Event):
        try:
            value = self.db.get(key)
            with self.lock:
                self.results[key] = value
        finally:
            event.set()
            # Cleanup after a delay to handle late arrivals
            time.sleep(0.1)
            with self.lock:
                self.pending.pop(key, None)
                self.results.pop(key, None)

from dataclasses import dataclass
from typing import Optional, Set
import random
import time
import threading
import redis
import psycopg2

@dataclass
class URLMapping:
    short_code: str
    original_url: str
    created_at: float
    is_hot: bool = False

class SkewResistantURLShortener:
    """
    URL shortener designed to handle highly skewed traffic.
    0.01% of URLs can receive 90% of traffic without degradation.
    """
    
    def __init__(
        self,
        redis_cluster: redis.RedisCluster,
        db_pool: psycopg2.pool.ThreadedConnectionPool,
        hot_key_split_factor: int = 10
    ):
        self.redis = redis_cluster
        self.db_pool = db_pool
        self.split_factor = hot_key_split_factor
        
        # Layer 2: Local cache for hot keys
        self.local_cache: dict[str, tuple[str, float]] = {}  # key -> (value, expiry)
        self.local_cache_lock = threading.Lock()
        self.local_cache_ttl = 5  # seconds
        
        # Hot key tracking
        self.hot_keys: Set[str] = set()
        self.hot_key_detector = CMSHotKeyDetector(
            hot_threshold=1000,  # 1000 req/min = hot
            window_seconds=60
        )
        
        # Request coalescing
        self.coalescer = SyncRequestCoalescer(self)
        
        # Metrics
        self.stats = {
            'local_hits': 0,
            'redis_hits': 0,
            'db_hits': 0,
            'total': 0
        }
    
    def redirect(self, short_code: str) -> Optional[str]:
        """
        Get the original URL for a short code.
        Multi-layer caching with hot key optimization.
        """
        self.stats['total'] += 1
        
        # Track for hot key detection
        is_hot = self.hot_key_detector.record(short_code)
        if is_hot and short_code not in self.hot_keys:
            self._promote_to_hot(short_code)
        
        # Layer 2: Check local cache (hot keys only)
        if short_code in self.hot_keys:
            url = self._check_local_cache(short_code)
            if url:
                self.stats['local_hits'] += 1
                return url
        
        # Layer 3: Check Redis (with split keys for hot keys)
        url = self._check_redis(short_code)
        if url:
            self.stats['redis_hits'] += 1
            self._update_local_cache(short_code, url)
            return url
        
        # Layer 4: Database (with coalescing)
        url = self.coalescer.get(short_code)
        if url:
            self.stats['db_hits'] += 1
            self._update_redis(short_code, url)
            self._update_local_cache(short_code, url)
        
        return url
    
    def _check_local_cache(self, key: str) -> Optional[str]:
        """Check local in-memory cache."""
        with self.local_cache_lock:
            if key in self.local_cache:
                value, expiry = self.local_cache[key]
                if time.time() < expiry:
                    return value
                else:
                    del self.local_cache[key]
        return None
    
    def _update_local_cache(self, key: str, value: str):
        """Update local cache for hot keys."""
        if key in self.hot_keys:
            with self.local_cache_lock:
                self.local_cache[key] = (value, time.time() + self.local_cache_ttl)
    
    def _check_redis(self, key: str) -> Optional[str]:
        """Check Redis, using split keys for hot keys."""
        if key in self.hot_keys:
            # Read from random split
            suffix = random.randint(0, self.split_factor - 1)
            redis_key = f"url:{key}:{suffix}"
        else:
            redis_key = f"url:{key}"
        
        return self.redis.get(redis_key)
    
    def _update_redis(self, key: str, value: str):
        """Update Redis, writing to all splits for hot keys."""
        if key in self.hot_keys:
            # Write to all splits
            pipe = self.redis.pipeline()
            for i in range(self.split_factor):
                pipe.setex(f"url:{key}:{i}", 3600, value)  # 1 hour TTL
            pipe.execute()
        else:
            self.redis.setex(f"url:{key}", 300, value)  # 5 min TTL
    
    def _db_get(self, key: str) -> Optional[str]:
        """Fetch from database."""
        conn = self.db_pool.getconn()
        try:
            with conn.cursor() as cur:
                cur.execute(
                    "SELECT original_url FROM urls WHERE short_code = %s",
                    (key,)
                )
                result = cur.fetchone()
                return result[0] if result else None
        finally:
            self.db_pool.putconn(conn)
    
    get = _db_get  # SyncRequestCoalescer(self) calls self.get(key)
    
    def _promote_to_hot(self, key: str):
        """Promote a key to hot status."""
        print(f"🔥 Promoting {key} to hot key")
        self.hot_keys.add(key)
        
        # Pre-populate all Redis splits
        url = self.redis.get(f"url:{key}") or self._db_get(key)
        if url:
            self._update_redis(key, url)
    
    def _demote_from_hot(self, key: str):
        """Demote a key from hot status."""
        print(f"❄️ Demoting {key} from hot key")
        self.hot_keys.discard(key)
        
        # Clean up local cache
        with self.local_cache_lock:
            self.local_cache.pop(key, None)
        
        # Clean up split keys (keep one)
        for i in range(1, self.split_factor):
            self.redis.delete(f"url:{key}:{i}")
    
    def get_stats(self) -> dict:
        """Return cache hit statistics."""
        total = self.stats['total'] or 1
        return {
            'total_requests': total,
            'local_hit_rate': self.stats['local_hits'] / total,
            'redis_hit_rate': self.stats['redis_hits'] / total,
            'db_hit_rate': self.stats['db_hits'] / total,
            'hot_key_count': len(self.hot_keys)
        }

# Cloudflare Worker for URL shortener caching
WORKER_CODE = """
addEventListener('fetch', event => {
  event.respondWith(handleRequest(event))
})

async function handleRequest(event) {
  const request = event.request
  const url = new URL(request.url)
  const shortCode = url.pathname.slice(1)  // Remove leading /
  
  // Check edge cache
  const cache = caches.default
  const cacheKey = new Request(url.toString(), request)
  let response = await cache.match(cacheKey)
  
  if (response) {
    // Cache hit - add header for debugging
    response = new Response(response.body, response)
    response.headers.set('X-Cache', 'HIT')
    return response
  }
  
  // Cache miss - fetch from origin
  response = await fetch(request)
  
  // Cache successful redirects
  if (response.status === 301 || response.status === 302) {
    const cacheResponse = response.clone()
    
    // Cache for 1 hour at edge
    const headers = new Headers(cacheResponse.headers)
    headers.set('Cache-Control', 'public, max-age=3600')
    
    const cachedResponse = new Response(cacheResponse.body, {
      status: cacheResponse.status,
      statusText: cacheResponse.statusText,
      headers: headers
    })
    
    event.waitUntil(cache.put(cacheKey, cachedResponse))
  }
  
  response = new Response(response.body, response)  // fetched headers are immutable
  response.headers.set('X-Cache', 'MISS')
  return response
}
"""

class InstagramFeedStrategy:
    CELEBRITY_THRESHOLD = 10000
    
    def __init__(self, feed_cache, post_store):
        self.feed_cache = feed_cache
        self.post_store = post_store
    
    def on_new_post(self, author_id: str, post_id: str, follower_count: int):
        if follower_count < self.CELEBRITY_THRESHOLD:
            # Push to all followers' feeds
            self._push_to_followers(author_id, post_id)
        else:
            # Just store the post - will be pulled on read
            self.post_store.add_celebrity_post(author_id, post_id)
    
    def get_feed(self, user_id: str, followed_celebrities: list[str]) -> list[str]:
        # Get pre-computed feed
        feed = self.feed_cache.get_feed(user_id)
        
        # Merge celebrity posts
        for celeb_id in followed_celebrities:
            celeb_posts = self.post_store.get_recent_posts(celeb_id)
            feed = self._merge_by_time(feed, celeb_posts)
        
        return feed[:100]  # Top 100 posts

class ApproximateLikeCounter:
    """
    Approximate counting for high-volume counters.
    Shows "1.2M likes" instead of "1,234,567 likes".
    """
    
    def __init__(self, redis_client):
        self.redis = redis_client
        
        # Exact counts for small numbers
        self.exact_threshold = 10000
        
        # Sampling rate for large counts
        self.sample_rate = 0.01  # Count 1 in 100
    
    def increment(self, post_id: str):
        current = self.get_count(post_id)
        
        if current < self.exact_threshold:
            # Exact counting for small numbers
            self.redis.incr(f"likes:{post_id}")
        else:
            # Probabilistic increment for large numbers
            if random.random() < self.sample_rate:
                self.redis.incrby(f"likes:{post_id}", int(1 / self.sample_rate))
    
    def get_count(self, post_id: str) -> int:
        return int(self.redis.get(f"likes:{post_id}") or 0)
    
    def get_display_count(self, post_id: str) -> str:
        count = self.get_count(post_id)
        
        if count < 1000:
            return str(count)
        elif count < 1000000:
            return f"{count // 1000}K"
        else:
            return f"{count / 1000000:.1f}M"

import zlib
from collections import defaultdict

class PostShardManager:
    """
    Manages post data across shards with hot shard mitigation.
    """
    
    def __init__(self, num_shards: int = 100):
        self.num_shards = num_shards
        self.shard_load: dict[int, int] = defaultdict(int)
        self.hot_posts: set[str] = set()
    
    def get_shard(self, post_id: str) -> int:
        # crc32, not built-in hash(): str hash() is salted per process
        return zlib.crc32(post_id.encode()) % self.num_shards
    
    def record_access(self, post_id: str):
        shard = self.get_shard(post_id)
        self.shard_load[shard] += 1
        
        # Detect hot shards
        avg_load = sum(self.shard_load.values()) / self.num_shards
        if self.shard_load[shard] > avg_load * 5:
            self._handle_hot_shard(shard)
    
    def _handle_hot_shard(self, shard: int):
        # Options:
        # 1. Add read replicas for this shard
        # 2. Migrate some data to less loaded shards
        # 3. Cache aggressively for this shard
        print(f"⚠️ Hot shard detected: {shard}")

class BufferedCounter:
    """
    Buffer increments locally, flush to central store periodically.
    Trades some accuracy and latency for throughput.
    """
    
    def __init__(self, redis_client, flush_interval: float = 1.0):
        self.redis = redis_client
        self.flush_interval = flush_interval
        
        self.local_counts: dict[str, int] = defaultdict(int)
        self.lock = threading.Lock()
        
        self._start_flusher()
    
    def increment(self, key: str, amount: int = 1):
        with self.lock:
            self.local_counts[key] += amount
    
    def _flush(self):
        with self.lock:
            counts_to_flush = dict(self.local_counts)
            self.local_counts.clear()
        
        if counts_to_flush:
            pipe = self.redis.pipeline()
            for key, count in counts_to_flush.items():
                pipe.incrby(f"counter:{key}", count)
            pipe.execute()
    
    def _start_flusher(self):
        def flush_loop():
            while True:
                time.sleep(self.flush_interval)
                self._flush()
        
        threading.Thread(target=flush_loop, daemon=True).start()

class GCounter:
    """
    CRDT G-Counter: Each node maintains its own count.
    Total = sum of all node counts.
    """
    
    def __init__(self, node_id: str, redis_client):
        self.node_id = node_id
        self.redis = redis_client
    
    def increment(self, key: str, amount: int = 1):
        # Increment only this node's counter
        self.redis.hincrby(f"gcounter:{key}", self.node_id, amount)
    
    def get_total(self, key: str) -> int:
        # Sum all nodes' counters
        counts = self.redis.hgetall(f"gcounter:{key}")
        return sum(int(v) for v in counts.values())
    
    def merge(self, key: str, other_counts: dict[str, int]):
        # Merge by taking max of each node's count
        pipe = self.redis.pipeline()
        for node_id, count in other_counts.items():
            # HSET with max
            current = int(self.redis.hget(f"gcounter:{key}", node_id) or 0)
            if count > current:
                pipe.hset(f"gcounter:{key}", node_id, count)
        pipe.execute()

class ShardedCounter:
    """
    Shard a counter across multiple keys to reduce contention.
    """
    
    def __init__(self, redis_client, num_shards: int = 100):
        self.redis = redis_client
        self.num_shards = num_shards
    
    def increment(self, key: str, amount: int = 1):
        # Pick a random shard
        shard = random.randint(0, self.num_shards - 1)
        self.redis.incrby(f"counter:{key}:{shard}", amount)
    
    def get_count(self, key: str) -> int:
        # Sum all shards
        pipe = self.redis.pipeline()
        for i in range(self.num_shards):
            pipe.get(f"counter:{key}:{i}")
        
        results = pipe.execute()
        return sum(int(r or 0) for r in results)

# Problem: All messages for user "viral_celebrity" go to one partition
producer.send('events', key='viral_celebrity', value=event)

# Solution 1: Add random suffix for known hot keys
if user_id in known_hot_users:
    partition_key = f"{user_id}:{random.randint(0, 9)}"
else:
    partition_key = user_id

producer.send('events', key=partition_key, value=event)

# Solution 2: Use null key for hot events (spread across partitions; sticky partitioner since Kafka 2.4)
if is_hot_event(event):
    producer.send('events', key=None, value=event)
else:
    producer.send('events', key=event.user_id, value=event)

# Problem: All items for "global_config" on one partition
table.get_item(Key={'pk': 'global_config', 'sk': 'settings'})

# Solution 1: Copy the item to N shard keys (writes update all 10), read a random one
def get_global_config():
    shards = [f"global_config#{i}" for i in range(10)]
    shard = random.choice(shards)
    return table.get_item(Key={'pk': shard, 'sk': 'settings'})

# Solution 2: DAX (DynamoDB Accelerator) for read caching
dax_client = amazondax.AmazonDaxClient.resource(endpoint_url='...')
dax_table = dax_client.Table('my-table')
# Hot reads served from cache

# Solution: Route hot documents to dedicated indices
def get_index_for_document(doc_id: str, is_hot: bool) -> str:
    if is_hot:
        return 'hot_documents'  # Index with more shards/replicas
    else:
        return 'documents'

# Hot index configuration
{
    "settings": {
        "number_of_shards": 1,  # A single hot doc lives on one shard anyway
        "number_of_replicas": 5  # More replicas = handle more reads
    }
}

import threading
import time
import random
from collections import defaultdict
from dataclasses import dataclass, field
from typing import Dict, Set, Optional, Callable
from enum import Enum
import mmh3
import numpy as np

class HotKeyType(Enum):
    READ_HEAVY = "read_heavy"
    WRITE_HEAVY = "write_heavy"
    MIXED = "mixed"

@dataclass
class HotKeyInfo:
    key: str
    detected_at: float
    request_count: int
    key_type: HotKeyType
    mitigation_active: bool = False

class ProductionHotKeyHandler:
    """
    Production-ready hot key detection and mitigation system.
    
    Features:
    - Count-Min Sketch for memory-efficient detection
    - Automatic classification (read/write heavy)
    - Pluggable mitigation strategies
    - Graceful promotion/demotion
    - Comprehensive metrics
    """
    
    def __init__(
        self,
        hot_threshold: int = 1000,  # requests per window
        window_seconds: int = 60,
        cms_width: int = 10000,
        cms_depth: int = 7,
        on_hot_key: Optional[Callable[[HotKeyInfo], None]] = None,
        on_cool_key: Optional[Callable[[str], None]] = None
    ):
        self.hot_threshold = hot_threshold
        self.window_seconds = window_seconds
        self.on_hot_key = on_hot_key
        self.on_cool_key = on_cool_key
        
        # Count-Min Sketch for approximate counting
        self.cms = np.zeros((cms_depth, cms_width), dtype=np.int64)
        self.cms_depth = cms_depth
        self.cms_width = cms_width
        
        # Exact counting for suspected hot keys
        self.suspected_keys: Dict[str, list] = defaultdict(list)
        
        # Confirmed hot keys
        self.hot_keys: Dict[str, HotKeyInfo] = {}
        
        # Read/write tracking for classification
        self.read_counts: Dict[str, int] = defaultdict(int)
        self.write_counts: Dict[str, int] = defaultdict(int)
        
        # Thread safety
        self.lock = threading.RLock()
        
        # Background tasks
        self._start_background_tasks()
        
        # Metrics
        self.metrics = {
            'total_requests': 0,
            'hot_key_detections': 0,
            'false_positives': 0,
            'mitigations_activated': 0
        }
    
    def record_read(self, key: str) -> bool:
        """Record a read operation. Returns True if key is hot."""
        return self._record(key, is_write=False)
    
    def record_write(self, key: str) -> bool:
        """Record a write operation. Returns True if key is hot."""
        return self._record(key, is_write=True)
    
    def _record(self, key: str, is_write: bool) -> bool:
        with self.lock:
            self.metrics['total_requests'] += 1
            
            # Track read/write ratio
            if is_write:
                self.write_counts[key] += 1
            else:
                self.read_counts[key] += 1
            
            # Update Count-Min Sketch
            self._cms_increment(key)
            estimate = self._cms_estimate(key)
            
            # Check if already confirmed hot (keep its timestamps current for cool-down checks)
            if key in self.hot_keys:
                self.suspected_keys[key].append(time.time())
                return True
            
            # Check if exceeds threshold for suspicion
            if estimate >= self.hot_threshold * 0.8:
                self._track_suspected(key)
            
            return False
    
    def _cms_increment(self, key: str):
        for i in range(self.cms_depth):
            idx = mmh3.hash(key, i) % self.cms_width
            self.cms[i][idx] += 1
    
    def _cms_estimate(self, key: str) -> int:
        min_count = float('inf')
        for i in range(self.cms_depth):
            idx = mmh3.hash(key, i) % self.cms_width
            min_count = min(min_count, self.cms[i][idx])
        return int(min_count)
    
    def _track_suspected(self, key: str):
        """Track suspected hot key with exact timestamps."""
        now = time.time()
        self.suspected_keys[key].append(now)
        
        # Check if confirmed hot
        cutoff = now - self.window_seconds
        recent = [t for t in self.suspected_keys[key] if t > cutoff]
        self.suspected_keys[key] = recent
        
        if len(recent) >= self.hot_threshold:
            self._confirm_hot(key, len(recent))
    
    def _confirm_hot(self, key: str, count: int):
        """Confirm a key as hot and trigger mitigation."""
        if key in self.hot_keys:
            return
        
        key_type = self._classify_key(key)
        info = HotKeyInfo(
            key=key,
            detected_at=time.time(),
            request_count=count,
            key_type=key_type
        )
        
        self.hot_keys[key] = info
        self.metrics['hot_key_detections'] += 1
        
        print(f"🔥 HOT KEY DETECTED: {key} (type: {key_type.value}, count: {count})")
        
        if self.on_hot_key:
            self.on_hot_key(info)
    
    def _classify_key(self, key: str) -> HotKeyType:
        """Classify hot key as read-heavy, write-heavy, or mixed."""
        reads = self.read_counts.get(key, 0)
        writes = self.write_counts.get(key, 0)
        total = reads + writes
        
        if total == 0:
            return HotKeyType.MIXED
        
        read_ratio = reads / total
        
        if read_ratio >= 0.9:
            return HotKeyType.READ_HEAVY
        elif read_ratio <= 0.1:
            return HotKeyType.WRITE_HEAVY
        else:
            return HotKeyType.MIXED
    
    def _check_cooled_keys(self):
        """Check if any hot keys have cooled down."""
        now = time.time()
        cooled = []
        
        with self.lock:
            for key, info in list(self.hot_keys.items()):
                # Check current rate
                recent = [t for t in self.suspected_keys.get(key, []) 
                         if t > now - self.window_seconds]
                self.suspected_keys[key] = recent
                
                if len(recent) < self.hot_threshold * 0.5:  # 50% of threshold
                    cooled.append(key)
        
        for key in cooled:
            self._cool_key(key)
    
    def _cool_key(self, key: str):
        """Handle a key cooling down."""
        with self.lock:
            if key in self.hot_keys:
                del self.hot_keys[key]
                print(f"❄️ Key cooled: {key}")
                
                if self.on_cool_key:
                    self.on_cool_key(key)
    
    def _reset_cms(self):
        """Periodically reset CMS to avoid count accumulation."""
        with self.lock:
            self.cms.fill(0)
            self.read_counts.clear()
            self.write_counts.clear()
    
    def _start_background_tasks(self):
        """Start background monitoring tasks."""
        def check_loop():
            while True:
                time.sleep(5)  # Check every 5 seconds
                self._check_cooled_keys()
        
        def reset_loop():
            while True:
                time.sleep(self.window_seconds)
                self._reset_cms()
        
        threading.Thread(target=check_loop, daemon=True).start()
        threading.Thread(target=reset_loop, daemon=True).start()
    
    def is_hot(self, key: str) -> bool:
        """Check if a key is currently hot."""
        return key in self.hot_keys
    
    def get_hot_keys(self) -> Dict[str, HotKeyInfo]:
        """Get all currently hot keys."""
        return dict(self.hot_keys)
    
    def get_metrics(self) -> dict:
        """Get handler metrics."""
        return {
            **self.metrics,
            'current_hot_keys': len(self.hot_keys),
            'suspected_keys': len(self.suspected_keys)
        }


# Example mitigation strategies
class HotKeyMitigationStrategies:
    def __init__(self, cache, db):
        self.cache = cache
        self.db = db
        self.split_factor = 10
        self.split_keys: Set[str] = set()
    
    def handle_hot_key(self, info: HotKeyInfo):
        """Apply appropriate mitigation based on key type."""
        if info.key_type == HotKeyType.READ_HEAVY:
            self._mitigate_read_heavy(info.key)
        elif info.key_type == HotKeyType.WRITE_HEAVY:
            self._mitigate_write_heavy(info.key)
        else:
            self._mitigate_mixed(info.key)
    
    def _mitigate_read_heavy(self, key: str):
        """Mitigation for read-heavy hot keys."""
        print(f"Applying read-heavy mitigation for {key}")
        
        # 1. Extend cache TTL
        value = self.cache.get(key) or self.db.get(key)
        if value:
            self.cache.setex(key, 3600, value)  # 1 hour TTL
        
        # 2. Enable key splitting for cache reads
        self._enable_split_reads(key, value)
    
    def _mitigate_write_heavy(self, key: str):
        """Mitigation for write-heavy hot keys."""
        print(f"Applying write-heavy mitigation for {key}")
        
        # Enable sharded writes
        self.split_keys.add(key)
    
    def _mitigate_mixed(self, key: str):
        """Mitigation for mixed hot keys."""
        print(f"Applying mixed mitigation for {key}")
        
        # Apply both strategies
        value = self.cache.get(key) or self.db.get(key)
        self._enable_split_reads(key, value)
        self.split_keys.add(key)
    
    def _enable_split_reads(self, key: str, value):
        """Distribute value across split keys."""
        if value:
            for i in range(self.split_factor):
                self.cache.setex(f"{key}:{i}", 3600, value)
            self.split_keys.add(key)
    
    def handle_cool_key(self, key: str):
        """Clean up when key cools down."""
        print(f"Cleaning up mitigation for {key}")
        
        # Remove split keys
        for i in range(self.split_factor):
            self.cache.delete(f"{key}:{i}")
        
        self.split_keys.discard(key)
    
    def get(self, key: str):
        """Get with hot key awareness."""
        if key in self.split_keys:
            # Read from random split
            split_key = f"{key}:{random.randint(0, self.split_factor - 1)}"
            return self.cache.get(split_key)
        return self.cache.get(key)
    
    def set(self, key: str, value: str):
        """Set with hot key awareness."""
        if key in self.split_keys:
            # Write to all splits
            for i in range(self.split_factor):
                self.cache.setex(f"{key}:{i}", 3600, value)
        else:
            self.cache.setex(key, 300, value)