Caching Patterns.

# Cache-Aside Implementation

class CacheAsideRepository:
    """
    Cache-aside pattern: Application manages cache explicitly.
    
    Also known as "lazy loading" because cache is only populated
    on read, not proactively.
    """
    
    def __init__(self, cache_client, db_client, default_ttl: int = 300):
        self.cache = cache_client
        self.db = db_client
        self.default_ttl = default_ttl
    
    async def get_product(self, product_id: str) -> dict:
        """
        Get product with cache-aside pattern.
        
        1. Check cache
        2. If miss, load from DB
        3. Populate cache
        4. Return
        """
        cache_key = f"product:{product_id}"
        
        # Step 1: Check cache
        cached = await self.cache.get(cache_key)
        if cached:
            return json.loads(cached)
        
        # Step 2: Cache miss - load from database
        product = await self.db.fetch_one(
            "SELECT * FROM products WHERE id = $1",
            product_id
        )
        
        if not product:
            return None
        
        # Step 3: Populate cache for next time
        await self.cache.setex(
            cache_key,
            self.default_ttl,
            json.dumps(dict(product))
        )
        
        # Step 4: Return
        return dict(product)
    
    async def update_product(self, product_id: str, data: dict) -> dict:
        """
        Update product - write to DB, invalidate cache.
        
        We invalidate rather than update to avoid race conditions.
        """
        # Write to database
        product = await self.db.fetch_one(
            """
            UPDATE products 
            SET name = $2, price = $3, updated_at = NOW()
            WHERE id = $1
            RETURNING *
            """,
            product_id, data['name'], data['price']
        )
        
        # Invalidate cache (not update!)
        cache_key = f"product:{product_id}"
        await self.cache.delete(cache_key)
        
        return dict(product)

# Read-Through Implementation (Cache-side)

class ReadThroughCache:
    """
    Read-through cache: Cache is responsible for loading data.
    
    The application only interacts with the cache, not the database.
    Cache handles misses by loading from the database.
    """
    
    def __init__(self, db_client, default_ttl: int = 300):
        self.db = db_client
        self.default_ttl = default_ttl
        self._cache = {}  # In real impl, this is Redis with loader
    
    async def get(self, key: str, loader_query: str = None) -> any:
        """
        Get value, loading from database if not cached.
        
        In production, this is typically configured at the cache level,
        not per-request. Libraries like Caffeine (Java) or dogpile (Python)
        support this pattern natively.
        """
        # Check cache
        if key in self._cache:
            entry = self._cache[key]
            if not self._is_expired(entry):
                return entry['value']
        
        # Cache miss - load from database
        if loader_query:
            value = await self.db.fetch_one(loader_query)
        else:
            value = await self._default_loader(key)
        
        # Store in cache
        self._cache[key] = {
            'value': value,
            'expires_at': time.time() + self.default_ttl
        }
        
        return value
    
    async def _default_loader(self, key: str) -> any:
        """Default loader parses key to determine query."""
        # key format: "entity:id" e.g., "product:123"
        entity, id = key.split(':')
        query = f"SELECT * FROM {entity}s WHERE id = $1"
        return await self.db.fetch_one(query, id)
    
    def _is_expired(self, entry: dict) -> bool:
        return time.time() > entry['expires_at']


# Application code is simpler
class ProductService:
    def __init__(self, cache: ReadThroughCache):
        self.cache = cache
    
    async def get_product(self, product_id: str) -> dict:
        # Application doesn't know about database!
        return await self.cache.get(f"product:{product_id}")

# Write-Through Implementation

class WriteThroughCache:
    """
    Write-through cache: Writes go through cache to database.
    
    Keeps the cache in step with the database on every write
    (concurrent writers or a failed cache write can still leave it stale).
    Trade-off: Higher write latency.
    """
    
    def __init__(self, cache_client, db_client, default_ttl: int = 300):
        self.cache = cache_client
        self.db = db_client
        self.default_ttl = default_ttl
    
    async def set(self, key: str, value: any, db_query: str, db_params: tuple) -> bool:
        """
        Write-through: Update database, then cache.
        
        Both must succeed for the write to be considered successful.
        """
        try:
            # Step 1: Write to database FIRST
            await self.db.execute(db_query, *db_params)
            
            # Step 2: Write to cache
            await self.cache.setex(
                key,
                self.default_ttl,
                json.dumps(value)
            )
            
            return True
            
        except Exception as e:
            # If DB write fails, don't update cache
            # If cache write fails after DB write, we have inconsistency!
            # In production, may need to invalidate cache here
            logger.error(f"Write-through failed: {e}")
            raise
    
    async def get(self, key: str) -> any:
        """Read from cache (with read-through on miss)."""
        cached = await self.cache.get(key)
        if cached:
            return json.loads(cached)
        
        # Miss - need to load from database
        # (combine with read-through for complete solution)
        return None


# Usage
class ProductService:
    def __init__(self, write_through_cache: WriteThroughCache):
        self.cache = write_through_cache
    
    async def update_product(self, product_id: str, data: dict) -> bool:
        return await self.cache.set(
            key=f"product:{product_id}",
            value=data,
            db_query="UPDATE products SET name=$2, price=$3 WHERE id=$1",
            db_params=(product_id, data['name'], data['price'])
        )

# Write-Behind Implementation

import asyncio
from collections import defaultdict
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Dict, List


@dataclass
class PendingWrite:
    key: str
    value: any
    db_query: str
    db_params: tuple
    queued_at: datetime


class WriteBehindCache:
    """
    Write-behind cache: Writes are cached immediately,
    database is updated asynchronously.
    
    WARNING: Risk of data loss if cache fails before DB write!
    """
    
    def __init__(
        self,
        cache_client,
        db_client,
        batch_size: int = 100,
        flush_interval: float = 1.0
    ):
        self.cache = cache_client
        self.db = db_client
        self.batch_size = batch_size
        self.flush_interval = flush_interval
        
        # Pending writes queue
        self._pending: Dict[str, PendingWrite] = {}
        self._running = False
    
    async def start(self):
        """Start the background writer."""
        self._running = True
        # keep a reference: the event loop only holds weak references to tasks
        self._flush_task = asyncio.create_task(self._flush_loop())
    
    async def stop(self):
        """Stop and flush remaining writes."""
        self._running = False
        await self._flush_to_database()
    
    async def set(self, key: str, value: any, db_query: str, db_params: tuple) -> bool:
        """
        Write-behind: Update cache immediately, queue DB write.
        
        Returns immediately after cache update - very fast!
        """
        # Step 1: Write to cache immediately
        await self.cache.set(key, json.dumps(value))
        
        # Step 2: Queue for database write
        # If same key written multiple times, only latest is kept
        self._pending[key] = PendingWrite(
            key=key,
            value=value,
            db_query=db_query,
            db_params=db_params,
            queued_at=datetime.now(timezone.utc)
        )
        
        # Step 3: Return immediately (don't wait for DB)
        return True
    
    async def _flush_loop(self):
        """Background loop that flushes writes to database."""
        while self._running:
            await asyncio.sleep(self.flush_interval)
            
            if self._pending:
                await self._flush_to_database()
    
    async def _flush_to_database(self):
        """Flush pending writes to database in batch."""
        if not self._pending:
            return
        
        # Take a snapshot of pending writes
        writes = list(self._pending.values())[:self.batch_size]
        keys_to_remove = [w.key for w in writes]
        
        try:
            # Batch write to database
            async with self.db.transaction():
                for write in writes:
                    await self.db.execute(write.db_query, *write.db_params)
            
            # Remove from pending queue
            for key in keys_to_remove:
                self._pending.pop(key, None)
            
            logger.info(f"Flushed {len(writes)} writes to database")
            
        except Exception as e:
            # DB write failed - writes stay in pending queue
            # They'll be retried on next flush
            logger.error(f"Database flush failed: {e}")
            # In production: alert, retry with backoff, eventual DLQ


# Usage - note how fast writes are
class AnalyticsService:
    """Write-behind is great for high-volume, loss-tolerant writes."""
    
    def __init__(self, write_behind_cache: WriteBehindCache):
        self.cache = write_behind_cache
    
    async def record_view_count(self, video_id: str, view_count: int):
        """
        Persist a video's latest view count (e.g. from a Redis INCR).
        Updated many times per second; we can tolerate some loss,
        but need low latency.
        """
        await self.cache.set(
            key=f"views:{video_id}",  # one key per video: repeated writes coalesce
            value={"video_id": video_id, "views": view_count},
            db_query="UPDATE videos SET view_count = $2 WHERE id = $1",
            db_params=(video_id, view_count)
        )
        # Returns immediately; only the latest count per video reaches the DB

# Basic Cache-Aside Implementation
# WARNING: Not production-ready - for learning only

import json
from typing import Optional, Any


class SimpleCache:
    """
    Minimal cache-aside implementation.
    
    Good for understanding the pattern, not for production.
    """
    
    def __init__(self, redis_client, default_ttl: int = 300):
        self.redis = redis_client
        self.default_ttl = default_ttl
    
    async def get(self, key: str) -> Optional[Any]:
        """Get from cache, returns None on miss."""
        value = await self.redis.get(key)
        if value:
            return json.loads(value)
        return None
    
    async def set(self, key: str, value: Any, ttl: int = None) -> None:
        """Set in cache with TTL."""
        ttl = ttl or self.default_ttl
        await self.redis.setex(key, ttl, json.dumps(value))
    
    async def delete(self, key: str) -> None:
        """Delete from cache (invalidation)."""
        await self.redis.delete(key)


# Usage example
async def get_product(product_id: str, cache: SimpleCache, db) -> dict:
    cache_key = f"product:{product_id}"
    
    # Check cache
    product = await cache.get(cache_key)
    if product:
        return product  # Cache hit!
    
    # Cache miss - load from database
    product = await db.fetch_one(
        "SELECT * FROM products WHERE id = $1",
        product_id
    )
    
    if product:
        # Store in cache for next time
        await cache.set(cache_key, dict(product))
    
    return dict(product) if product else None

# Production-Ready Cache-Aside Implementation

import asyncio
import json
import hashlib
import logging
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from typing import Optional, Dict, Any, List, Callable, TypeVar, Generic
from enum import Enum
import time

logger = logging.getLogger(__name__)

T = TypeVar('T')


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

@dataclass
class CacheConfig:
    """Configuration for the cache layer."""
    default_ttl: int = 300  # 5 minutes
    max_ttl: int = 86400    # 24 hours
    key_prefix: str = "app"
    key_version: str = "v1"  # Bump to invalidate all keys on deploy
    
    # Circuit breaker settings
    failure_threshold: int = 5
    recovery_timeout: int = 30
    
    # Timeouts
    get_timeout: float = 0.1   # 100ms
    set_timeout: float = 0.2   # 200ms


class CacheStatus(Enum):
    HIT = "hit"
    MISS = "miss"
    ERROR = "error"
    SKIP = "skip"  # Circuit breaker open


@dataclass
class CacheResult(Generic[T]):
    """Result of a cache operation."""
    status: CacheStatus
    value: Optional[T] = None
    latency_ms: float = 0
    error: Optional[str] = None


# =============================================================================
# Serialization
# =============================================================================

class CacheSerializer:
    """Handles serialization of cache values."""
    
    @staticmethod
    def serialize(value: Any) -> str:
        """Serialize value to JSON string."""
        return json.dumps(value, default=str, ensure_ascii=False)
    
    @staticmethod
    def deserialize(data: str) -> Any:
        """Deserialize JSON string to value."""
        return json.loads(data)


# =============================================================================
# Circuit Breaker
# =============================================================================

class CircuitBreaker:
    """
    Circuit breaker for cache operations.
    
    Prevents cascading failures when cache is unhealthy.
    """
    
    def __init__(self, failure_threshold: int = 5, recovery_timeout: int = 30):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        
        self._failures = 0
        self._last_failure_time: Optional[float] = None
        self._state = "closed"  # closed, open, half-open
    
    def record_success(self):
        """Record a successful operation."""
        self._failures = 0
        self._state = "closed"
    
    def record_failure(self):
        """Record a failed operation."""
        self._failures += 1
        self._last_failure_time = time.time()
        
        if self._failures >= self.failure_threshold:
            self._state = "open"
            logger.warning(f"Circuit breaker opened after {self._failures} failures")
    
    def is_open(self) -> bool:
        """Check if circuit is open (should skip cache)."""
        if self._state == "closed":
            return False
        
        if self._state == "open":
            # Check if recovery timeout has passed
            if time.time() - self._last_failure_time > self.recovery_timeout:
                self._state = "half-open"
                return False
            return True
        
        return False  # half-open: let requests through as a trial (this sketch doesn't limit it to one probe)


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

@dataclass
class CacheMetrics:
    """Tracks cache performance metrics."""
    hits: int = 0
    misses: int = 0
    errors: int = 0
    skips: int = 0
    total_latency_ms: float = 0
    
    @property
    def total_requests(self) -> int:
        return self.hits + self.misses + self.errors + self.skips
    
    @property
    def hit_ratio(self) -> float:
        if self.total_requests == 0:
            return 0.0
        return self.hits / self.total_requests
    
    @property
    def avg_latency_ms(self) -> float:
        if self.total_requests == 0:
            return 0.0
        return self.total_latency_ms / self.total_requests
    
    def record(self, result: CacheResult):
        """Record a cache operation result."""
        self.total_latency_ms += result.latency_ms
        
        if result.status == CacheStatus.HIT:
            self.hits += 1
        elif result.status == CacheStatus.MISS:
            self.misses += 1
        elif result.status == CacheStatus.ERROR:
            self.errors += 1
        elif result.status == CacheStatus.SKIP:
            self.skips += 1


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

class ProductionCache:
    """
    Production-ready cache-aside implementation.
    
    Features:
    - Automatic serialization
    - Circuit breaker for resilience
    - Metrics tracking
    - Versioned keys
    - Bulk operations
    - Graceful degradation
    """
    
    def __init__(
        self,
        redis_client,
        config: CacheConfig = None,
        metrics_client = None
    ):
        self.redis = redis_client
        self.config = config or CacheConfig()
        self.external_metrics = metrics_client
        
        self.serializer = CacheSerializer()
        self.circuit_breaker = CircuitBreaker(
            failure_threshold=self.config.failure_threshold,
            recovery_timeout=self.config.recovery_timeout
        )
        self.metrics = CacheMetrics()
    
    def _build_key(self, key: str) -> str:
        """Build full cache key with prefix and version."""
        return f"{self.config.key_prefix}:{self.config.key_version}:{key}"
    
    async def get(self, key: str) -> CacheResult[Any]:
        """
        Get value from cache.
        
        Returns CacheResult with status, value, and metrics.
        """
        start_time = time.time()
        full_key = self._build_key(key)
        
        # Check circuit breaker
        if self.circuit_breaker.is_open():
            result = CacheResult(status=CacheStatus.SKIP)
            self.metrics.record(result)
            return result
        
        try:
            # Get from Redis with timeout
            raw_value = await asyncio.wait_for(
                self.redis.get(full_key),
                timeout=self.config.get_timeout
            )
            
            latency_ms = (time.time() - start_time) * 1000
            
            if raw_value is None:
                result = CacheResult(
                    status=CacheStatus.MISS,
                    latency_ms=latency_ms
                )
            else:
                value = self.serializer.deserialize(raw_value)
                result = CacheResult(
                    status=CacheStatus.HIT,
                    value=value,
                    latency_ms=latency_ms
                )
            
            self.circuit_breaker.record_success()
            self.metrics.record(result)
            return result
            
        except asyncio.TimeoutError:
            logger.warning(f"Cache get timeout for key: {key}")
            self.circuit_breaker.record_failure()
            result = CacheResult(
                status=CacheStatus.ERROR,
                error="timeout",
                latency_ms=(time.time() - start_time) * 1000
            )
            self.metrics.record(result)
            return result
            
        except Exception as e:
            logger.error(f"Cache get error for key {key}: {e}")
            self.circuit_breaker.record_failure()
            result = CacheResult(
                status=CacheStatus.ERROR,
                error=str(e),
                latency_ms=(time.time() - start_time) * 1000
            )
            self.metrics.record(result)
            return result
    
    async def set(
        self,
        key: str,
        value: Any,
        ttl: int = None
    ) -> bool:
        """
        Set value in cache.
        
        Returns True on success, False on failure.
        Failures are logged but don't raise exceptions.
        """
        if self.circuit_breaker.is_open():
            return False
        
        full_key = self._build_key(key)
        ttl = min(ttl or self.config.default_ttl, self.config.max_ttl)
        
        try:
            serialized = self.serializer.serialize(value)
            
            await asyncio.wait_for(
                self.redis.setex(full_key, ttl, serialized),
                timeout=self.config.set_timeout
            )
            
            self.circuit_breaker.record_success()
            return True
            
        except asyncio.TimeoutError:
            logger.warning(f"Cache set timeout for key: {key}")
            self.circuit_breaker.record_failure()
            return False
            
        except Exception as e:
            logger.error(f"Cache set error for key {key}: {e}")
            self.circuit_breaker.record_failure()
            return False
    
    async def delete(self, key: str) -> bool:
        """Delete a key from cache (invalidation)."""
        if self.circuit_breaker.is_open():
            return False
        
        full_key = self._build_key(key)
        
        try:
            await self.redis.delete(full_key)
            self.circuit_breaker.record_success()
            return True
        except Exception as e:
            logger.error(f"Cache delete error for key {key}: {e}")
            self.circuit_breaker.record_failure()
            return False
    
    async def get_many(self, keys: List[str]) -> Dict[str, CacheResult]:
        """
        Get multiple values in a single round-trip.
        
        Much more efficient than multiple individual gets.
        """
        if self.circuit_breaker.is_open():
            return {key: CacheResult(status=CacheStatus.SKIP) for key in keys}
        
        start_time = time.time()
        full_keys = [self._build_key(k) for k in keys]
        
        try:
            # MGET for all keys at once
            raw_values = await asyncio.wait_for(
                self.redis.mget(full_keys),
                timeout=self.config.get_timeout * 2  # Allow more time for bulk
            )
            
            latency_ms = (time.time() - start_time) * 1000
            self.circuit_breaker.record_success()
            
            results = {}
            for key, raw_value in zip(keys, raw_values):
                if raw_value is None:
                    results[key] = CacheResult(
                        status=CacheStatus.MISS,
                        latency_ms=latency_ms / len(keys)
                    )
                else:
                    value = self.serializer.deserialize(raw_value)
                    results[key] = CacheResult(
                        status=CacheStatus.HIT,
                        value=value,
                        latency_ms=latency_ms / len(keys)
                    )
                self.metrics.record(results[key])
            
            return results
            
        except Exception as e:
            logger.error(f"Cache get_many error: {e}")
            self.circuit_breaker.record_failure()
            return {
                key: CacheResult(status=CacheStatus.ERROR, error=str(e))
                for key in keys
            }
    
    async def set_many(self, items: Dict[str, Any], ttl: int = None) -> bool:
        """Set multiple values in a single round-trip."""
        if self.circuit_breaker.is_open():
            return False
        
        ttl = min(ttl or self.config.default_ttl, self.config.max_ttl)
        
        try:
            pipe = self.redis.pipeline()
            
            for key, value in items.items():
                full_key = self._build_key(key)
                serialized = self.serializer.serialize(value)
                pipe.setex(full_key, ttl, serialized)
            
            await asyncio.wait_for(
                pipe.execute(),
                timeout=self.config.set_timeout * 2
            )
            
            self.circuit_breaker.record_success()
            return True
            
        except Exception as e:
            logger.error(f"Cache set_many error: {e}")
            self.circuit_breaker.record_failure()
            return False
    
    def get_stats(self) -> Dict[str, Any]:
        """Get cache statistics."""
        return {
            "hits": self.metrics.hits,
            "misses": self.metrics.misses,
            "errors": self.metrics.errors,
            "skips": self.metrics.skips,
            "hit_ratio": round(self.metrics.hit_ratio, 4),
            "avg_latency_ms": round(self.metrics.avg_latency_ms, 2),
            "circuit_breaker_state": self.circuit_breaker._state,
        }


# =============================================================================
# Repository Pattern with Cache
# =============================================================================

class CachedProductRepository:
    """
    Product repository with caching.
    
    Demonstrates the cache-aside pattern in a repository layer.
    """
    
    def __init__(
        self,
        db_pool,
        cache: ProductionCache,
        ttl_seconds: int = 300
    ):
        self.db = db_pool
        self.cache = cache
        self.ttl = ttl_seconds
    
    async def get_by_id(self, product_id: str) -> Optional[Dict]:
        """Get product by ID with caching."""
        cache_key = f"product:{product_id}"
        
        # Try cache first
        result = await self.cache.get(cache_key)
        
        if result.status == CacheStatus.HIT:
            logger.debug(f"Cache hit for product {product_id}")
            return result.value
        
        # Cache miss or error - load from database
        product = await self.db.fetch_one(
            "SELECT * FROM products WHERE id = $1",
            product_id
        )
        
        if product:
            product_dict = dict(product)
            # Populate cache (best effort)
            await self.cache.set(cache_key, product_dict, self.ttl)
            return product_dict
        
        return None
    
    async def get_by_ids(self, product_ids: List[str]) -> Dict[str, Dict]:
        """Get multiple products efficiently."""
        cache_keys = [f"product:{pid}" for pid in product_ids]
        
        # Try cache for all
        cache_results = await self.cache.get_many(cache_keys)
        
        # Separate hits and misses
        products = {}
        missing_ids = []
        
        for pid, key in zip(product_ids, cache_keys):
            result = cache_results[key]
            if result.status == CacheStatus.HIT:
                products[pid] = result.value
            else:
                missing_ids.append(pid)
        
        # Load missing from database
        if missing_ids:
            rows = await self.db.fetch(
                "SELECT * FROM products WHERE id = ANY($1)",
                missing_ids
            )
            
            to_cache = {}
            for row in rows:
                product = dict(row)
                pid = product['id']
                products[pid] = product
                to_cache[f"product:{pid}"] = product
            
            # Cache the missing ones (best effort)
            if to_cache:
                await self.cache.set_many(to_cache, self.ttl)
        
        return products
    
    async def update(self, product_id: str, data: Dict) -> Optional[Dict]:
        """Update product and invalidate cache."""
        # Update database
        product = await self.db.fetch_one(
            """
            UPDATE products 
            SET name = $2, price = $3, updated_at = NOW()
            WHERE id = $1
            RETURNING *
            """,
            product_id, data['name'], data['price']
        )
        
        if product:
            # Invalidate cache
            await self.cache.delete(f"product:{product_id}")
            return dict(product)
        
        return None
    
    async def delete(self, product_id: str) -> bool:
        """Delete product and invalidate cache."""
        result = await self.db.execute(
            "DELETE FROM products WHERE id = $1",
            product_id
        )
        
        # Always try to invalidate (even if delete found nothing)
        await self.cache.delete(f"product:{product_id}")
        
        return result == "DELETE 1"

# Graceful degradation on cache failure
async def get_product_safe(product_id: str) -> Optional[dict]:
    cache_key = f"product:{product_id}"
    
    # Cache.get() returns CacheResult, never throws
    result = await cache.get(cache_key)
    
    if result.status == CacheStatus.HIT:
        return result.value
    
    # Miss, error, or skip - all go to database
    # The circuit breaker ensures we don't keep hammering a dead cache
    product = await db.fetch_one(
        "SELECT * FROM products WHERE id = $1",
        product_id
    )
    
    if product:
        # Only try to cache if circuit is closed
        if result.status != CacheStatus.SKIP:
            await cache.set(cache_key, dict(product))
        return dict(product)
    
    return None