Design a URL Shortener at Scale.

# Short Code Generation with Pre-allocated Ranges
# Applies: Week 1, Day 1 - Partitioning concepts

import threading
from dataclasses import dataclass
from typing import Optional
import redis


CHARS = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"
SPACE = 62 ** 7              # ~3.5 trillion 7-character codes
MULTIPLIER = 1_580_030_173   # odd and not divisible by 31 -> coprime with 62^7


@dataclass
class KeyRange:
    """A range of pre-allocated counter values."""
    current: int          # Next counter value to hand out
    end: int              # Last counter value in range (inclusive)
    
    def next_value(self) -> Optional[int]:
        """Get next available counter value from this range."""
        if self.current > self.end:
            return None  # Range exhausted
        value = self.current
        self.current += 1
        return value
    
    def remaining(self) -> int:
        """How many values left in this range."""
        return max(0, self.end - self.current + 1)


def encode(counter: int) -> str:
    """
    Map a counter value to a 7-char code.
    Multiplying by a number coprime with 62^7 is a bijection on [0, 62^7),
    so codes stay unique but don't look sequential. (This only obscures
    order; use a keyed cipher such as a Feistel network if codes must be
    unguessable.)
    """
    n = (counter * MULTIPLIER) % SPACE
    code = ""
    for _ in range(7):
        code = CHARS[n % 62] + code
        n //= 62
    return code


class ShortCodeGenerator:
    """
    Generates unique short codes using pre-allocated ranges.
    
    Key concepts applied:
    - Partitioning (Day 1): Each server gets its own range
    - No coordination needed for individual code generation
    - Range allocation is the only synchronized operation
    """
    
    def __init__(self, redis_client: redis.Redis, server_id: str):
        self.redis = redis_client
        self.server_id = server_id
        self.current_range: Optional[KeyRange] = None
        self.lock = threading.Lock()
        
        # Configuration
        self.range_size = 10_000  # Codes per range
        self.refill_threshold = 1_000  # Get new range when below this
    
    def generate(self) -> str:
        """
        Generate a unique short code.
        
        This is O(1) in the common case - just increment a counter.
        Only contacts central service when range is exhausted.
        """
        with self.lock:
            # Check if we need a new range
            if self._needs_new_range():
                self._allocate_new_range()
            
            # Take next value from current range
            value = self.current_range.next_value()
            
            if value is None:
                # Range exhausted mid-operation (shouldn't happen often)
                self._allocate_new_range()
                value = self.current_range.next_value()
            
            return encode(value)
    
    def _needs_new_range(self) -> bool:
        """Check if we need to allocate a new range."""
        if self.current_range is None:
            return True
        return self.current_range.remaining() < self.refill_threshold
    
    def _allocate_new_range(self) -> None:
        """
        Allocate a new range from the central service.
        
        Uses Redis INCR for atomic range allocation.
        This is the only point of coordination.
        """
        # Atomically get next range number (1, 2, 3, ...)
        range_num = self.redis.incr("shortcode:range_counter")
        
        # Range k covers counter values [(k-1) * size, k * size - 1].
        # 62^7 / 10,000 = ~352 million ranges before the space runs out
        # (~96 years at 100M codes/day).
        start = (range_num - 1) * self.range_size
        if start + self.range_size > SPACE:
            raise RuntimeError("Short code space exhausted")
        
        self.current_range = KeyRange(
            current=start,
            end=start + self.range_size - 1
        )
        
        # Log allocation for debugging
        print(f"Server {self.server_id} allocated range starting at {start}")


# Usage Example
async def create_short_url(long_url: str, generator: ShortCodeGenerator) -> str:
    """Create a new short URL."""
    
    # Generate unique code (O(1), no DB lookup needed)
    short_code = generator.generate()
    
    # Store mapping in database
    await db.execute(
        """
        INSERT INTO urls (short_code, long_url, created_at, expires_at)
        VALUES ($1, $2, NOW(), NOW() + INTERVAL '5 years')
        """,
        short_code, long_url
    )
    
    # Cache for fast redirects
    await redis.setex(f"url:{short_code}", 86400, long_url)
    
    return f"https://tiny.link/{short_code}"

# Hot Key Handling for URL Redirects
# Applies: Week 1, Day 4 - Hot Keys and Skew

import random
import time
from typing import Optional, Tuple
import redis
from cachetools import TTLCache


class HotKeyAwareCache:
    """
    Multi-layer cache with hot key detection and mitigation.
    
    Applies Day 4 concepts:
    - Detection: Track access frequency
    - Mitigation: Multiple layers of caching
    - Distribution: Spread hot keys across replicas
    """
    
    def __init__(self, redis_cluster: redis.RedisCluster):
        self.redis = redis_cluster
        self.hot_key_threshold = 1000  # requests/minute
        self.replica_count = 3  # Replicas for hot keys
        
        # Local tracking of access patterns
        self.access_counts: dict[str, int] = {}
        self.hot_keys: set[str] = set()
        self.last_reset = time.time()
        
        # Layer 2: Local server cache (bounded LRU-style, short TTL so edits/deletes propagate)
        self.local_cache: TTLCache = TTLCache(maxsize=10_000, ttl=5)
    
    def get(self, short_code: str) -> Optional[str]:
        """
        Get long URL for short code with hot key handling.
        """
        key = f"url:{short_code}"
        
        # Track access for hot key detection
        self._track_access(key)
        
        # Layer 2: Check local cache first
        value = self.local_cache.get(key)
        if value is not None:
            return value
        
        # Layer 3 & 4: Redis lookup with hot key awareness
        if key in self.hot_keys:
            # Hot key: use replicated keys
            value = self._get_hot_key(key)
        else:
            # Normal key: direct lookup
            value = self._get_normal_key(key)
        
        if value is not None:
            self.local_cache[key] = value
        return value
    
    def _get_normal_key(self, key: str) -> Optional[str]:
        """Normal Redis lookup."""
        return self.redis.get(key)
    
    def _get_hot_key(self, key: str) -> Optional[str]:
        """
        Hot key lookup with distribution.
        
        Hot keys are replicated with suffixes to spread load.
        Client randomly selects a replica.
        """
        # Randomly select replica
        replica_suffix = random.randint(0, self.replica_count - 1)
        replicated_key = f"{key}:{replica_suffix}"
        
        value = self.redis.get(replicated_key)
        
        if value is None:
            # Fallback to original key (replication might be in progress)
            value = self.redis.get(key)
        
        return value
    
    def set(self, short_code: str, long_url: str, ttl: int = 86400) -> None:
        """
        Set URL mapping with hot key awareness.
        """
        key = f"url:{short_code}"
        
        # Always set the primary key
        self.redis.setex(key, ttl, long_url)
        
        # If this is a hot key, also set replicas
        if key in self.hot_keys:
            self._replicate_hot_key(key, long_url, ttl)
    
    def _replicate_hot_key(self, key: str, value: str, ttl: int) -> None:
        """Create replicas of a hot key."""
        pipeline = self.redis.pipeline()
        for i in range(self.replica_count):
            pipeline.setex(f"{key}:{i}", ttl, value)
        pipeline.execute()
    
    def _track_access(self, key: str) -> None:
        """
        Track access patterns to detect hot keys.
        
        Simple fixed-window counter (reset every minute).
        In production, use a Count-Min Sketch or a top-k streaming algorithm.
        """
        current_time = time.time()
        
        # Reset counts every minute
        if current_time - self.last_reset > 60:
            self._analyze_and_reset()
        
        # Increment access count
        self.access_counts[key] = self.access_counts.get(key, 0) + 1
    
    def _analyze_and_reset(self) -> None:
        """Analyze access patterns and identify hot keys."""
        new_hot_keys = set()
        
        for key, count in self.access_counts.items():
            if count > self.hot_key_threshold:
                new_hot_keys.add(key)
                
                # If newly hot, create replicas
                if key not in self.hot_keys:
                    self._promote_to_hot(key)
        
        # Keys that cooled down
        for key in self.hot_keys - new_hot_keys:
            self._demote_from_hot(key)
        
        self.hot_keys = new_hot_keys
        self.access_counts = {}
        self.last_reset = time.time()
    
    def _promote_to_hot(self, key: str) -> None:
        """Promote a key to hot status by creating replicas."""
        value = self.redis.get(key)
        if value:
            ttl = self.redis.ttl(key)
            self._replicate_hot_key(key, value, max(ttl, 3600))
            print(f"Promoted hot key: {key}")
    
    def _demote_from_hot(self, key: str) -> None:
        """Remove replicas for a key that's no longer hot."""
        pipeline = self.redis.pipeline()
        for i in range(self.replica_count):
            pipeline.delete(f"{key}:{i}")
        pipeline.execute()
        print(f"Demoted key: {key}")


# Redirect service using hot key aware cache
async def redirect(short_code: str, cache: HotKeyAwareCache) -> str:
    """
    Handle URL redirect with hot key protection.
    """
    # Try cache first (handles hot keys automatically)
    long_url = cache.get(short_code)
    
    if long_url is None:
        # Cache miss - query database
        long_url = await db.fetchval(
            "SELECT long_url FROM urls WHERE short_code = $1",
            short_code
        )
        
        if long_url is None:
            raise NotFoundError(f"URL not found: {short_code}")
        
        # Populate cache
        cache.set(short_code, long_url)
    
    # Fire-and-forget analytics event
    await analytics.record_click(short_code)
    
    return long_url

# Database Sharding with Consistent Hashing
# Applies: Week 1, Day 1 - Partitioning strategies

import hashlib
from bisect import bisect_left
from dataclasses import dataclass
from typing import List, Dict, Any, Optional
import asyncpg


@dataclass
class ShardConfig:
    """Configuration for a database shard."""
    shard_id: int
    host: str
    port: int
    database: str
    
    @property
    def dsn(self) -> str:
        return f"postgresql://{self.host}:{self.port}/{self.database}"


class ConsistentHashRing:
    """
    Consistent hash ring for shard selection.
    
    Applies Day 1 concepts:
    - Consistent hashing minimizes data movement on rebalance
    - Virtual nodes ensure even distribution
    - Deterministic: same key always maps to same shard
    """
    
    def __init__(self, shards: List[ShardConfig], virtual_nodes: int = 150):
        self.shards = {s.shard_id: s for s in shards}
        self.virtual_nodes = virtual_nodes
        self.ring: List[tuple[int, int]] = []  # (hash, shard_id)
        
        self._build_ring()
    
    def _build_ring(self) -> None:
        """Build the consistent hash ring with virtual nodes."""
        ring = []
        
        for shard in self.shards.values():
            # Create virtual nodes for each shard
            for i in range(self.virtual_nodes):
                # Hash the shard_id + virtual node index
                key = f"{shard.shard_id}:{i}"
                hash_val = self._hash(key)
                ring.append((hash_val, shard.shard_id))
        
        # Sort by hash value
        ring.sort(key=lambda x: x[0])
        self.ring = ring
        
        # Extract just the hash values for binary search
        self.ring_hashes = [h for h, _ in ring]
    
    def _hash(self, key: str) -> int:
        """Compute hash value for a key."""
        return int(hashlib.md5(key.encode()).hexdigest(), 16)
    
    def get_shard(self, key: str) -> ShardConfig:
        """
        Get the shard responsible for a key.
        
        Uses binary search for O(log n) lookup.
        """
        if not self.ring:
            raise ValueError("No shards configured")
        
        hash_val = self._hash(key)
        
        # Find the first node with hash >= key's hash
        idx = bisect_left(self.ring_hashes, hash_val)
        
        # Wrap around if we're past the last node
        if idx >= len(self.ring):
            idx = 0
        
        shard_id = self.ring[idx][1]
        return self.shards[shard_id]
    
    def add_shard(self, shard: ShardConfig) -> Dict[int, List[str]]:
        """
        Add a new shard to the ring.
        
        Returns mapping of shard_id -> keys that need to move.
        With consistent hashing, only ~1/n keys need to move.
        """
        self.shards[shard.shard_id] = shard
        self._build_ring()
        # In practice, you'd track which keys moved
        return {}
    
    def remove_shard(self, shard_id: int) -> Dict[int, List[str]]:
        """
        Remove a shard from the ring.
        
        Returns mapping of destination shard_id -> keys to receive.
        """
        if shard_id in self.shards:
            del self.shards[shard_id]
            self._build_ring()
        return {}


class ShardedURLDatabase:
    """
    Sharded database client for URL storage.
    
    Handles:
    - Shard selection via consistent hashing
    - Connection pooling per shard
    - Cross-shard queries when needed
    """
    
    def __init__(self, shard_configs: List[ShardConfig]):
        self.hash_ring = ConsistentHashRing(shard_configs)
        self.pools: Dict[int, asyncpg.Pool] = {}
    
    async def initialize(self) -> None:
        """Create connection pools for all shards."""
        for shard in self.hash_ring.shards.values():
            self.pools[shard.shard_id] = await asyncpg.create_pool(
                shard.dsn,
                min_size=5,
                max_size=20
            )
    
    async def create_url(
        self, 
        short_code: str, 
        long_url: str,
        user_id: Optional[str] = None
    ) -> None:
        """
        Create a new URL mapping.
        
        Writes to:
        1. URL shard (keyed by short_code) - for redirects
        2. User shard (keyed by user_id) - for listing
        """
        # Get shard for this short_code
        shard = self.hash_ring.get_shard(short_code)
        pool = self.pools[shard.shard_id]
        
        async with pool.acquire() as conn:
            await conn.execute(
                """
                INSERT INTO urls (short_code, long_url, user_id, created_at, expires_at)
                VALUES ($1, $2, $3, NOW(), NOW() + INTERVAL '5 years')
                """,
                short_code, long_url, user_id
            )
        
        # Also update user index if user is known
        if user_id:
            await self._update_user_index(user_id, short_code)
    
    async def get_url(self, short_code: str) -> Optional[str]:
        """
        Get long URL by short code.
        
        Single-shard operation - very fast.
        """
        shard = self.hash_ring.get_shard(short_code)
        pool = self.pools[shard.shard_id]
        
        async with pool.acquire() as conn:
            row = await conn.fetchrow(
                """
                SELECT long_url FROM urls 
                WHERE short_code = $1 
                  AND expires_at > NOW()
                """,
                short_code
            )
            return row['long_url'] if row else None
    
    async def get_user_urls(
        self, 
        user_id: str, 
        limit: int = 100
    ) -> List[Dict[str, Any]]:
        """
        Get all URLs for a user.
        
        Queries user index shard, then fetches details from URL shards.
        """
        # Get user's shard
        user_shard = self.hash_ring.get_shard(f"user:{user_id}")
        pool = self.pools[user_shard.shard_id]
        
        async with pool.acquire() as conn:
            rows = await conn.fetch(
                """
                SELECT short_code, created_at
                FROM user_urls
                WHERE user_id = $1
                ORDER BY created_at DESC
                LIMIT $2
                """,
                user_id, limit
            )
        
        # Fetch full details from URL shards (sequential here; asyncio.gather would parallelize)
        urls = []
        for row in rows:
            url_data = await self.get_url(row['short_code'])
            if url_data:
                urls.append({
                    'short_code': row['short_code'],
                    'long_url': url_data,
                    'created_at': row['created_at']
                })
        
        return urls
    
    async def _update_user_index(self, user_id: str, short_code: str) -> None:
        """Update the user's URL index."""
        user_shard = self.hash_ring.get_shard(f"user:{user_id}")
        pool = self.pools[user_shard.shard_id]
        
        async with pool.acquire() as conn:
            await conn.execute(
                """
                INSERT INTO user_urls (user_id, short_code, created_at)
                VALUES ($1, $2, NOW())
                """,
                user_id, short_code
            )

# Distributed Rate Limiting
# Applies: Week 1, Day 3 - Rate Limiting at Scale

import time
from dataclasses import dataclass
from enum import Enum
from typing import Tuple, Optional
import redis


class RateLimitTier(Enum):
    FREE = 100
    STARTER = 1_000
    BUSINESS = 10_000
    ENTERPRISE = 100_000


@dataclass
class RateLimitResult:
    """Result of a rate limit check."""
    allowed: bool
    limit: int
    remaining: int
    reset_at: int  # Unix timestamp when limit resets
    retry_after: Optional[int] = None  # Seconds to wait if blocked


class SlidingWindowRateLimiter:
    """
    Sliding window rate limiter using Redis.
    
    Applies Day 3 concepts:
    - Sliding window for smooth limiting (no burst at window start)
    - Redis for distributed state
    - Lua script for atomic operations
    """
    
    # Lua script for atomic rate limit check and update
    RATE_LIMIT_SCRIPT = """
    local key = KEYS[1]
    local now = tonumber(ARGV[1])
    local window = tonumber(ARGV[2])
    local limit = tonumber(ARGV[3])
    
    -- Remove old entries outside the window
    redis.call('ZREMRANGEBYSCORE', key, 0, now - window)
    
    -- Count current requests in window
    local count = redis.call('ZCARD', key)
    
    if count < limit then
        -- Under limit: add this request
        redis.call('ZADD', key, now, now .. ':' .. math.random())
        redis.call('EXPIRE', key, window)
        return {1, limit - count - 1}  -- allowed, remaining
    else
        -- Over limit: reject
        -- Get oldest entry to calculate retry time
        local oldest = redis.call('ZRANGE', key, 0, 0, 'WITHSCORES')
        local retry_after = 0
        if #oldest > 0 then
            retry_after = math.ceil(oldest[2] + window - now)
        end
        return {0, 0, retry_after}  -- blocked, remaining, retry_after
    end
    """
    
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client
        self.window_seconds = 60  # 1-minute window
        
        # Register the Lua script
        self.rate_limit_sha = self.redis.script_load(self.RATE_LIMIT_SCRIPT)
    
    def check(
        self, 
        api_key: str, 
        tier: RateLimitTier
    ) -> RateLimitResult:
        """
        Check if request is allowed under rate limit.
        
        Uses sliding window algorithm:
        - Smoother than fixed windows (no thundering herd at window start)
        - More accurate than token bucket for per-minute limits
        - Single Redis round-trip with Lua script
        """
        key = f"ratelimit:{api_key}"
        now = time.time()
        limit = tier.value
        
        try:
            result = self.redis.evalsha(
                self.rate_limit_sha,
                1,  # Number of keys
                key,
                now,
                self.window_seconds,
                limit
            )
            
            allowed = result[0] == 1
            remaining = result[1]
            
            if allowed:
                return RateLimitResult(
                    allowed=True,
                    limit=limit,
                    remaining=remaining,
                    reset_at=int(now) + self.window_seconds
                )
            else:
                retry_after = result[2] if len(result) > 2 else self.window_seconds
                return RateLimitResult(
                    allowed=False,
                    limit=limit,
                    remaining=0,
                    reset_at=int(now) + retry_after,
                    retry_after=retry_after
                )
                
        except redis.RedisError as e:
            # Fail open: if Redis is down, allow the request
            # Log and alert, but don't block users
            print(f"Rate limiter error: {e}")
            return RateLimitResult(
                allowed=True,
                limit=limit,
                remaining=limit,
                reset_at=int(now) + self.window_seconds
            )
    
    def get_headers(self, result: RateLimitResult) -> dict:
        """Generate conventional (X-RateLimit-*) response headers."""
        headers = {
            'X-RateLimit-Limit': str(result.limit),
            'X-RateLimit-Remaining': str(result.remaining),
            'X-RateLimit-Reset': str(result.reset_at)
        }
        
        if not result.allowed:
            headers['Retry-After'] = str(result.retry_after)
        
        return headers


# FastAPI middleware integration
from fastapi import Request, HTTPException
from fastapi.responses import JSONResponse

class RateLimitMiddleware:
    """
    FastAPI middleware for rate limiting.
    """
    
    def __init__(self, rate_limiter: SlidingWindowRateLimiter):
        self.rate_limiter = rate_limiter
    
    async def __call__(self, request: Request, call_next):
        # Extract API key from header
        api_key = request.headers.get('X-API-Key')
        
        if not api_key:
            # No API key = free tier with IP-based limiting
            api_key = f"ip:{request.client.host}"
            tier = RateLimitTier.FREE
        else:
            # Look up tier for this API key (cached)
            tier = await self._get_tier(api_key)
        
        # Check rate limit
        result = self.rate_limiter.check(api_key, tier)
        
        if not result.allowed:
            return JSONResponse(
                status_code=429,
                content={
                    'error': 'Rate limit exceeded',
                    'retry_after': result.retry_after
                },
                headers=self.rate_limiter.get_headers(result)
            )
        
        # Process request
        response = await call_next(request)
        
        # Add rate limit headers to response
        for key, value in self.rate_limiter.get_headers(result).items():
            response.headers[key] = value
        
        return response
    
    async def _get_tier(self, api_key: str) -> RateLimitTier:
        """Look up the rate limit tier for an API key."""
        # In production, this would query a cache/database
        # For now, return a default
        return RateLimitTier.STARTER

# Replication and Consistency Configuration
# Applies: Week 1, Day 2 - Replication Trade-offs

from enum import Enum
from typing import Optional
import asyncpg


class ConsistencyLevel(Enum):
    """
    Consistency levels for different operations.
    
    Applies Day 2 concepts:
    - Strong consistency when correctness matters
    - Eventual consistency when speed matters
    """
    STRONG = "strong"       # Read from primary
    READ_YOUR_WRITES = "ryw"  # Read from the synchronous replica
    EVENTUAL = "eventual"   # Any replica is fine


class ReplicatedDatabase:
    """
    Database client with configurable consistency levels.
    """
    
    def __init__(
        self,
        primary_dsn: str,
        replica_dsns: list[str],
        sync_replica_dsn: str
    ):
        self.primary_dsn = primary_dsn
        self.replica_dsns = replica_dsns
        self.sync_replica_dsn = sync_replica_dsn
        
        self.primary_pool: Optional[asyncpg.Pool] = None
        self.replica_pools: list[asyncpg.Pool] = []
        self.sync_replica_pool: Optional[asyncpg.Pool] = None
    
    async def initialize(self) -> None:
        """Create connection pools."""
        self.primary_pool = await asyncpg.create_pool(
            self.primary_dsn,
            min_size=10,
            max_size=50
        )
        
        for dsn in self.replica_dsns:
            pool = await asyncpg.create_pool(dsn, min_size=5, max_size=20)
            self.replica_pools.append(pool)
        
        self.sync_replica_pool = await asyncpg.create_pool(
            self.sync_replica_dsn,
            min_size=5,
            max_size=20
        )
    
    async def write(self, query: str, *args) -> None:
        """
        Execute a write query on the primary.
        
        Writes always go to primary for consistency.
        """
        async with self.primary_pool.acquire() as conn:
            await conn.execute(query, *args)
    
    async def read(
        self,
        query: str,
        *args,
        consistency: ConsistencyLevel = ConsistencyLevel.EVENTUAL
    ):
        """
        Execute a read query with specified consistency level.
        
        Consistency levels:
        - STRONG: Read from primary (freshest, but adds load)
        - READ_YOUR_WRITES: Read from sync replica (needs synchronous_commit =
          remote_apply; with 'on' the replica may not have replayed the write yet)
        - EVENTUAL: Read from any replica (fastest, slight delay)
        """
        pool = self._select_pool(consistency)
        
        async with pool.acquire() as conn:
            return await conn.fetch(query, *args)
    
    def _select_pool(self, consistency: ConsistencyLevel) -> asyncpg.Pool:
        """Select appropriate pool based on consistency requirement."""
        if consistency == ConsistencyLevel.STRONG:
            return self.primary_pool
        elif consistency == ConsistencyLevel.READ_YOUR_WRITES:
            return self.sync_replica_pool
        else:  # EVENTUAL
            # Pick a random async replica
            import random
            return random.choice(self.replica_pools)


# URL Service with appropriate consistency levels
class URLService:
    """
    URL service demonstrating consistency trade-offs.
    """
    
    def __init__(self, db: ReplicatedDatabase, cache: HotKeyAwareCache):
        self.db = db
        self.cache = cache
    
    async def create_url(self, short_code: str, long_url: str) -> dict:
        """
        Create a new URL.
        
        Consistency: Strong write (must not lose data)
        """
        await self.db.write(
            """
            INSERT INTO urls (short_code, long_url, created_at)
            VALUES ($1, $2, NOW())
            """,
            short_code, long_url
        )
        
        # Also cache it (eventual consistency OK here)
        self.cache.set(short_code, long_url)
        
        return {'short_code': short_code, 'long_url': long_url}
    
    async def get_url_for_redirect(self, short_code: str) -> Optional[str]:
        """
        Get URL for redirect (critical path).
        
        Consistency: Eventual (speed matters most)
        
        Stale read is OK: worst case, user sees old destination
        for a few hundred milliseconds after an update.
        """
        # Try cache first
        long_url = self.cache.get(short_code)
        if long_url:
            return long_url
        
        # Cache miss: query replica
        rows = await self.db.read(
            "SELECT long_url FROM urls WHERE short_code = $1",
            short_code,
            consistency=ConsistencyLevel.EVENTUAL
        )
        
        if rows:
            long_url = rows[0]['long_url']
            self.cache.set(short_code, long_url)
            return long_url
        
        return None
    
    async def get_url_after_create(self, short_code: str) -> Optional[dict]:
        """
        Get URL immediately after creation.
        
        Consistency: Read-your-writes (user expects to see their URL)
        
        After creating a URL, user immediately views it.
        Must see the URL they just created.
        """
        rows = await self.db.read(
            """
            SELECT short_code, long_url, created_at, click_count
            FROM urls WHERE short_code = $1
            """,
            short_code,
            consistency=ConsistencyLevel.READ_YOUR_WRITES
        )
        
        if rows:
            return dict(rows[0])
        return None
    
    async def get_analytics(self, short_code: str) -> Optional[dict]:
        """
        Get analytics for a URL.
        
        Consistency: Eventual (analytics can be slightly delayed)
        
        Analytics are inherently delayed anyway, so eventual
        consistency is perfect here.
        """
        rows = await self.db.read(
            """
            SELECT DATE(clicked_at) as date,
                   COUNT(*) as clicks
            FROM url_clicks
            WHERE short_code = $1
            GROUP BY DATE(clicked_at)
            ORDER BY date DESC
            LIMIT 30
            """,
            short_code,
            consistency=ConsistencyLevel.EVENTUAL
        )
        
        return {'daily_clicks': [dict(r) for r in rows]}