02. Scaling Reads — Caching, Replicas & CDN

import redis, json
cache = redis.Redis(host='cache.internal', port=6379)

def get_user_profile(user_id: int) -> dict:
    cache_key = f"user:{user_id}:profile"

    # Step 1: Check cache
    cached = cache.get(cache_key)
    if cached:
        return json.loads(cached)  # Cache HIT

    # Step 2: Cache MISS — read from database
    profile = db.execute("SELECT * FROM users WHERE id = %s", (user_id,))

    # Step 3: Populate cache for next time
    cache.setex(cache_key, 300, json.dumps(profile))  # TTL = 5 min
    return profile

def update_user_profile(user_id: int, updates: dict) -> dict:
    # Write to database
    db.execute("UPDATE users SET name=%s WHERE id=%s", (updates['name'], user_id))
    # Write to cache in the same operation
    updated = db.execute("SELECT * FROM users WHERE id=%s", (user_id,)).fetchone()
    cache.setex(f"user:{user_id}:profile", 300, json.dumps(updated))
    return updated

def update_user_async(user_id: int, updates: dict):
    # Write to cache (fast — user gets immediate response)
    cache.setex(f"user:{user_id}:profile", 300, json.dumps(updates))
    # Queue the database write
    write_queue.put({"user_id": user_id, "data": updates})

# Background worker drains the queue and writes to DB in batches

class ReadThroughCache:
    def __init__(self, cache_client, ttl=300):
        self.cache = cache_client
        self.ttl = ttl

    def get(self, key: str, loader_fn):
        """Get from cache. On miss, loader_fn fetches from source."""
        value = self.cache.get(key)
        if value:
            return json.loads(value)

        # Cache loads from DB on miss -- transparent to caller
        value = loader_fn()
        if value:
            self.cache.setex(key, self.ttl, json.dumps(value))
        return value

# Usage -- caching logic is hidden from application code
user_cache = ReadThroughCache(cache, ttl=300)
profile = user_cache.get(
    f"user:{user_id}",
    lambda: db.execute("SELECT * FROM users WHERE id=%s", (user_id,)).fetchone()
)

1. WHAT to cache?     Hot data, expensive computations, static content
2. WHERE to cache?    Browser, CDN, app server, distributed cache, DB cache
3. HOW to populate?   Cache-aside, read-through, write-through, write-behind
4. WHEN to invalidate? TTL, event-based, version-based
5. WHAT on miss?      Fetch from source, queue and wait, return stale

Content Type          Suggested TTL   Reasoning
──────────────────────────────────────────────────
Static assets         24 hours+       Rarely change
Product catalog       1 hour          Changes daily
User profile          5 minutes       Changes occasionally
Account balance       0 (no cache)    Must be real-time
Trending topics       30 seconds      Changes constantly

def update_product(product_id: int, updates: dict):
    db.execute("UPDATE products SET ... WHERE id = %s", (product_id,))
    # Invalidate product cache and all related caches
    cache.delete(f"product:{product_id}")
    cache.delete(f"category:{updates['category_id']}:products")
    # Notify other services via event bus
    event_bus.publish("product.updated", {"product_id": product_id})

def get_product_versioned(product_id: int) -> dict:
    version = cache.get(f"product:{product_id}:version") or "0"
    cache_key = f"product:{product_id}:v{version}"
    cached = cache.get(cache_key)
    if cached:
        return json.loads(cached)
    product = db.get_product(product_id)
    cache.setex(cache_key, 3600, json.dumps(product))
    return product

def update_product(product_id: int, updates: dict):
    db.update_product(product_id, updates)
    cache.incr(f"product:{product_id}:version")  # Old key becomes unreachable

def get_with_lock(key: str, loader_fn, ttl: int = 300):
    """Only one request fetches from DB on cache miss."""
    value = cache.get(key)
    if value:
        return json.loads(value)

    lock_key = f"lock:{key}"
    if cache.set(lock_key, "1", nx=True, ex=10):  # Acquire lock
        try:
            value = loader_fn()
            cache.setex(key, ttl, json.dumps(value))
            return value
        finally:
            cache.delete(lock_key)
    else:
        time.sleep(0.05)  # Losers wait, then retry from cache
        return json.loads(cache.get(key)) or loader_fn()

class DatabaseRouter:
    def __init__(self, primary_dsn, replica_dsns):
        self.primary = connect(primary_dsn)
        self.replicas = [connect(dsn) for dsn in replica_dsns]

    def get_read_connection(self):
        return random.choice(self.replicas)

    def get_write_connection(self):
        return self.primary

    def get_read_after_write_connection(self):
        """For reads that MUST see the latest write."""
        return self.primary

router = DatabaseRouter(
    primary_dsn="postgresql://primary:5432/app",
    replica_dsns=["postgresql://replica1:5432/app", "postgresql://replica2:5432/app"]
)

SYNCHRONOUS:  0ms lag, slower writes, replica failure blocks all writes
ASYNCHRONOUS: 10-1000ms lag, fast writes, stale reads possible
SEMI-SYNC:    At least 1 replica confirms. Balance of both.

def handle_read_your_writes(user_id: int, last_write_ts: float):
    """Route to primary if the user's last write was recent."""
    if time.time() - last_write_ts < 2.0:
        return router.get_write_connection()  # Read from primary

    replica = router.get_read_connection()
    lag = get_replication_lag(replica)
    if lag > 1.0:
        return router.get_write_connection()  # Fallback to primary
    return replica

STATIC (always CDN): Images, CSS, JS, fonts. Cache forever, bust with filename hash.
SEMI-STATIC (short TTL): Product pages, blog posts. Cache 1-60 min, invalidate on update.
DYNAMIC (usually not CDN): User-specific data, authenticated responses.

@app.route("/api/products/<int:product_id>")
def get_product(product_id):
    product = db.get_product(product_id)
    response = make_response(json.dumps(product))
    # CDN caches 5 min, browser caches 1 min
    response.headers['Cache-Control'] = 'public, max-age=60, s-maxage=300'
    response.headers['ETag'] = f'"{hash(json.dumps(product))}"'
    return response

@app.route("/api/users/<int:user_id>/profile")
def get_profile(user_id):
    response = make_response(json.dumps(db.get_user(user_id)))
    response.headers['Cache-Control'] = 'private, max-age=60'  # No CDN
    return response

PULL: CDN fetches from origin on first request. Simple, cold start on first hit.
PUSH: You upload to CDN proactively. No cold start, more operational complexity.

CHOOSE REDIS WHEN:
- You need data structures (sorted sets, lists, hashes, HyperLogLog)
- You need persistence (RDB snapshots or AOF)
- You need pub/sub for real-time features
- You need Lua scripting for atomic operations
- You need built-in replication and clustering
- Use cases: sessions, rate limiting, leaderboards, queues, pub/sub

CHOOSE MEMCACHED WHEN:
- You need simple key-value caching with the largest possible cache
- You need multi-threaded performance on multi-core machines
- You do not need persistence or data structures
- You want a simpler operational model
- Use cases: HTML fragment caching, DB query result caching

from redis.cluster import RedisCluster

rc = RedisCluster(
    startup_nodes=[
        {"host": "redis-1", "port": 6379},
        {"host": "redis-2", "port": 6379},
        {"host": "redis-3", "port": 6379},
    ],
    decode_responses=True
)

# Cluster shards data using CRC16(key) % 16384 hash slots
rc.set("user:1001:session", json.dumps(session_data), ex=1800)

# Pipeline for batch operations (reduces round trips)
pipe = rc.pipeline()
for user_id in user_ids:
    pipe.get(f"user:{user_id}:profile")
results = pipe.execute()

LAYER 1: BROWSER CACHE     → ~40% of requests never leave browser
LAYER 2: CDN               → ~30% served from edge (5ms latency)
LAYER 3: REDIS              → ~20% served from cache (1ms latency)
LAYER 4: READ REPLICAS      → ~8% served from replicas (10ms latency)
LAYER 5: PRIMARY DATABASE   → ~2% reach the primary (20ms latency)

Of 100,000 req/sec:
  40,000 → browser    |  18,000 → CDN    |  8,400 → Redis
  2,800  → replicas   |  800    → primary DB