System Design Interview: How to Design Any System (Step-by-Step)

POST /tweets          { user_id, content, media_ids[] }  → tweet_id
GET  /feed/{user_id}  ?limit=20&cursor=<token>           → Tweet[]
GET  /tweet/{id}                                         → Tweet
DELETE /tweet/{id}                                       → 204

import hashlib
import bisect

class ConsistentHashRing:
    def __init__(self, nodes=None, replicas=150):
        self.replicas = replicas
        self.ring = {}
        self.sorted_keys = []
        for node in (nodes or []):
            self.add_node(node)

    def add_node(self, node):
        for i in range(self.replicas):
            key = self._hash(f"{node}:{i}")
            self.ring[key] = node
            bisect.insort(self.sorted_keys, key)

    def remove_node(self, node):
        for i in range(self.replicas):
            key = self._hash(f"{node}:{i}")
            self.ring.pop(key, None)
            idx = bisect.bisect_left(self.sorted_keys, key)
            if idx < len(self.sorted_keys) and self.sorted_keys[idx] == key:
                self.sorted_keys.pop(idx)

    def get_node(self, key):
        if not self.ring:
            return None
        h = self._hash(key)
        idx = bisect.bisect(self.sorted_keys, h) % len(self.sorted_keys)
        return self.ring[self.sorted_keys[idx]]

    def _hash(self, key):
        return int(hashlib.md5(key.encode()).hexdigest(), 16)

POST /shorten    { long_url, custom_alias?, ttl_days? } → { short_code }
GET  /{code}     → 301/302 redirect to long_url

import secrets
import string

ALPHABET = string.ascii_letters + string.digits  # 62 chars

def generate_code(length=7) -> str:
    # 62^7 = 3.5 trillion unique codes
    return ''.join(secrets.choice(ALPHABET) for _ in range(length))

def encode_base62(num: int) -> str:
    ALPHABET = "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ"
    if num == 0:
        return ALPHABET[0]
    result = []
    while num:
        result.append(ALPHABET[num % 62])
        num //= 62
    return ''.join(reversed(result))

CREATE TABLE urls (
    short_code   CHAR(7)      PRIMARY KEY,
    long_url     TEXT         NOT NULL,
    user_id      BIGINT,
    created_at   TIMESTAMPTZ  NOT NULL DEFAULT now(),
    expires_at   TIMESTAMPTZ,
    click_count  BIGINT       NOT NULL DEFAULT 0
);
CREATE INDEX ON urls (long_url);  -- for dedup check

# Strong consistency: R + W > N (e.g., N=3, W=2, R=2)
session.execute(
    SimpleStatement(query, consistency_level=ConsistencyLevel.QUORUM)
)

# Eventual consistency: faster, less durable
session.execute(
    SimpleStatement(query, consistency_level=ConsistencyLevel.ONE)
)

import threading
import json
from pathlib import Path

class KVStore:
    def __init__(self, wal_path="wal.log"):
        self._store: dict = {}
        self._lock = threading.RWLock() if hasattr(threading, 'RWLock') else threading.Lock()
        self._wal = open(wal_path, "a")
        self._replay_wal(wal_path)

    def get(self, key: str):
        with self._lock:
            return self._store.get(key)

    def put(self, key: str, value):
        with self._lock:
            self._append_wal("PUT", key, value)
            self._store[key] = value

    def delete(self, key: str):
        with self._lock:
            self._append_wal("DEL", key, None)
            self._store.pop(key, None)

    def _append_wal(self, op, key, value):
        entry = json.dumps({"op": op, "key": key, "value": value}) + "\n"
        self._wal.write(entry)
        self._wal.flush()  # fsync in production

    def _replay_wal(self, path):
        try:
            for line in Path(path).read_text().splitlines():
                entry = json.loads(line)
                if entry["op"] == "PUT":
                    self._store[entry["key"]] = entry["value"]
                elif entry["op"] == "DEL":
                    self._store.pop(entry["key"], None)
        except FileNotFoundError:
            pass

# Feed write path (simplified)
def handle_new_tweet(tweet: Tweet, follower_ids: list[int]):
    # Store tweet
    tweet_store.put(tweet.id, tweet)

    # Fan out to regular followers
    regular_followers = [
        uid for uid in follower_ids
        if follower_count[uid] < 10_000
    ]
    for follower_id in regular_followers:
        redis.lpush(f"feed:{follower_id}", tweet.id)
        redis.ltrim(f"feed:{follower_id}", 0, 800)  # keep last 800

# Feed read path
def get_feed(user_id: int, cursor: str, limit: int = 20) -> list[Tweet]:
    # Get pre-computed feed from cache
    cached_ids = redis.lrange(f"feed:{user_id}", 0, limit - 1)

    # Inject tweets from celebrities the user follows
    celebrity_ids = get_celebrity_followees(user_id)
    celebrity_tweets = fetch_recent_tweets(celebrity_ids, since=cursor)

    # Merge and sort
    all_ids = merge_sorted([cached_ids, [t.id for t in celebrity_tweets]])
    return [tweet_store.get(tid) for tid in all_ids[:limit]]

# From the ConsistentHashRing implementation above:
# Adding 1 node to a 3-node ring:
# Without vnodes: ~33% of keys remapped
# With 150 vnodes: still ~25% remapped (1/4 of ring), but distribution is uniform

ring = ConsistentHashRing(["cache-1", "cache-2", "cache-3"], replicas=150)

# Route request to correct cache node
node = ring.get_node("user:12345")
cache_client = cache_pool[node]
result = cache_client.get("user:12345")

import time
import redis

class TokenBucketRateLimiter:
    def __init__(self, redis_client, capacity: int, refill_rate: float):
        self.redis = redis_client
        self.capacity = capacity
        self.refill_rate = refill_rate  # tokens per second

    def allow(self, user_id: str) -> bool:
        key = f"rate_limit:{user_id}"
        now = time.time()

        pipe = self.redis.pipeline()
        pipe.hgetall(key)
        results = pipe.execute()
        data = results[0]

        if data:
            tokens = float(data[b"tokens"])
            last_refill = float(data[b"last_refill"])
            elapsed = now - last_refill
            tokens = min(self.capacity, tokens + elapsed * self.refill_rate)
        else:
            tokens = self.capacity
            last_refill = now

        if tokens < 1:
            return False

        pipe = self.redis.pipeline()
        pipe.hset(key, mapping={"tokens": tokens - 1, "last_refill": now})
        pipe.expire(key, int(self.capacity / self.refill_rate) + 10)
        pipe.execute()
        return True

def sliding_window_allow(redis_client, user_id: str, limit: int, window_sec: int) -> bool:
    now = int(time.time())
    current_window = now // window_sec
    prev_window = current_window - 1

    current_key = f"rl:{user_id}:{current_window}"
    prev_key = f"rl:{user_id}:{prev_window}"

    pipe = redis_client.pipeline()
    pipe.get(current_key)
    pipe.get(prev_key)
    current_count, prev_count = [int(x or 0) for x in pipe.execute()]

    # Weight prev window by how much it overlaps with current sliding window
    elapsed_in_window = now % window_sec
    weighted_prev = prev_count * (1 - elapsed_in_window / window_sec)
    estimated_count = weighted_prev + current_count

    if estimated_count >= limit:
        return False

    pipe = redis_client.pipeline()
    pipe.incr(current_key)
    pipe.expire(current_key, window_sec * 2)
    pipe.execute()
    return True

LRU  — evict least recently used (most common)
LFU  — evict least frequently used (better for hot-key patterns)
TTL  — expire by time (simplest, use for session data)
ARC  — Adaptive Replacement Cache, tracks both recency and frequency

import threading
import time

# Solution 1: Probabilistic early expiration
def get_with_early_expiry(cache, db, key, ttl, beta=1.0):
    cached = cache.get_with_ttl(key)  # returns (value, remaining_ttl)
    if cached:
        value, remaining_ttl = cached
        # Probabilistically recompute before expiry
        if -beta * math.log(random.random()) > remaining_ttl:
            value = db.get(key)
            cache.set(key, value, ttl)
        return value
    value = db.get(key)
    cache.set(key, value, ttl)
    return value

# Solution 2: Mutex lock on cache miss
_locks = {}
_lock_meta = threading.Lock()

def get_with_mutex(cache, db, key, ttl):
    value = cache.get(key)
    if value:
        return value

    with _lock_meta:
        if key not in _locks:
            _locks[key] = threading.Lock()
        lock = _locks[key]

    with lock:
        # Check again after acquiring lock
        value = cache.get(key)
        if value:
            return value
        value = db.get(key)
        cache.set(key, value, ttl)
        return value

from bitarray import bitarray
import mmh3  # MurmurHash3

class BloomFilter:
    def __init__(self, size: int, hash_count: int):
        self.size = size
        self.hash_count = hash_count
        self.bit_array = bitarray(size)
        self.bit_array.setall(0)

    def add(self, item: str):
        for seed in range(self.hash_count):
            idx = mmh3.hash(item, seed) % self.size
            self.bit_array[idx] = 1

    def __contains__(self, item: str) -> bool:
        return all(
            self.bit_array[mmh3.hash(item, seed) % self.size]
            for seed in range(self.hash_count)
        )

# 1 billion URLs, ~1% false positive rate: ~10 bits/element = 1.25 GB
url_filter = BloomFilter(size=10_000_000_000, hash_count=7)

Service → Notification Service → Queue → Workers → Third-party providers

# Notification schema
{
    "notification_id": "uuid",
    "user_id": 12345,
    "type": "push|email|sms",
    "template_id": "order_shipped",
    "template_vars": {"order_id": "ORD-999", "eta": "tomorrow"},
    "priority": "high|normal|low",
    "dedup_key": "order_shipped:ORD-999:user:12345"
}

def dispatch_notification(notification: dict):
    user_prefs = user_prefs_store.get(notification["user_id"])

    channels = []
    if user_prefs.push_enabled and notification["type"] in ("push", "all"):
        channels.append(PushWorker)
    if user_prefs.email_enabled and notification["type"] in ("email", "all"):
        channels.append(EmailWorker)
    if user_prefs.sms_enabled and notification["type"] in ("sms", "urgent"):
        channels.append(SMSWorker)

    for channel in channels:
        queue.enqueue(channel.QUEUE_NAME, notification)

def retry_delay(attempt: int, base_ms=1000) -> float:
    cap = 32_000  # max 32 seconds
    delay = min(cap, base_ms * (2 ** attempt))
    jitter = delay * 0.2 * random.random()
    return (delay + jitter) / 1000  # in seconds

-- Users
CREATE TABLE users (
    id          BIGSERIAL PRIMARY KEY,
    username    VARCHAR(50) UNIQUE NOT NULL,
    email       VARCHAR(255) UNIQUE NOT NULL,
    created_at  TIMESTAMPTZ DEFAULT now()
);

-- Posts
CREATE TABLE posts (
    id          BIGSERIAL PRIMARY KEY,
    user_id     BIGINT NOT NULL REFERENCES users(id),
    content     TEXT NOT NULL,
    media_urls  TEXT[],
    created_at  TIMESTAMPTZ DEFAULT now()
);
CREATE INDEX ON posts (user_id, created_at DESC);

-- Follow graph
CREATE TABLE follows (
    follower_id  BIGINT NOT NULL REFERENCES users(id),
    followee_id  BIGINT NOT NULL REFERENCES users(id),
    created_at   TIMESTAMPTZ DEFAULT now(),
    PRIMARY KEY (follower_id, followee_id)
);
CREATE INDEX ON follows (followee_id);  -- "who follows me" queries

-- Likes (high write volume — consider denormalized counter on posts)
CREATE TABLE likes (
    post_id    BIGINT NOT NULL REFERENCES posts(id),
    user_id    BIGINT NOT NULL REFERENCES users(id),
    created_at TIMESTAMPTZ DEFAULT now(),
    PRIMARY KEY (post_id, user_id)
);

-- Denormalized counter to avoid COUNT(*) on likes
ALTER TABLE posts ADD COLUMN like_count BIGINT DEFAULT 0;

Shard 0: user_id 0–10M
Shard 1: user_id 10M–20M
...

def get_shard(user_id: int, num_shards: int) -> int:
    return hash(user_id) % num_shards

41 bits timestamp | 10 bits machine ID | 12 bits sequence

import time

class SnowflakeIDGenerator:
    EPOCH = 1288834974657  # Twitter epoch (Nov 4, 2010)
    MACHINE_ID_BITS = 10
    SEQUENCE_BITS = 12
    MAX_MACHINE_ID = (1 << MACHINE_ID_BITS) - 1
    MAX_SEQUENCE = (1 << SEQUENCE_BITS) - 1

    def __init__(self, machine_id: int):
        assert 0 <= machine_id <= self.MAX_MACHINE_ID
        self.machine_id = machine_id
        self.sequence = 0
        self.last_timestamp = -1

    def generate(self) -> int:
        ts = int(time.time() * 1000) - self.EPOCH
        if ts == self.last_timestamp:
            self.sequence = (self.sequence + 1) & self.MAX_SEQUENCE
            if self.sequence == 0:
                while ts <= self.last_timestamp:
                    ts = int(time.time() * 1000) - self.EPOCH
        else:
            self.sequence = 0
        self.last_timestamp = ts
        return (ts << 22) | (self.machine_id << 12) | self.sequence

/data/kafka/
  topic-orders/
    partition-0/
      00000000000000000000.log   (segment 0)
      00000000000000012345.log   (segment 12345)
      00000000000000012345.index

from confluent_kafka import Producer

producer = Producer({
    "bootstrap.servers": "kafka-1:9092,kafka-2:9092",
    "linger.ms": 5,          # wait up to 5ms to batch messages
    "batch.size": 65536,     # 64KB batch
    "acks": "all",           # wait for all replicas
    "enable.idempotence": True,
})

def send_order_event(order: dict):
    producer.produce(
        topic="orders",
        key=str(order["user_id"]).encode(),  # same user → same partition → ordered
        value=json.dumps(order).encode(),
        callback=delivery_report,
    )
    producer.poll(0)  # trigger callbacks

def delivery_report(err, msg):
    if err:
        logger.error(f"Delivery failed: {err}")

from confluent_kafka import Consumer

consumer = Consumer({
    "bootstrap.servers": "kafka-1:9092",
    "group.id": "order-processor",
    "auto.offset.reset": "earliest",
    "enable.auto.commit": False,  # manual commit for exactly-once processing
})
consumer.subscribe(["orders"])

while True:
    msg = consumer.poll(timeout=1.0)
    if msg is None or msg.error():
        continue
    order = json.loads(msg.value())
    process_order(order)  # idempotent
    consumer.commit(message=msg)  # commit only after successful processing

class TrieNode:
    def __init__(self):
        self.children: dict[str, 'TrieNode'] = {}
        self.is_end = False
        self.frequency = 0
        self.top_suggestions: list[tuple[int, str]] = []  # (freq, word), max-heap

class AutocompleteTrie:
    def __init__(self, max_suggestions=10):
        self.root = TrieNode()
        self.max_suggestions = max_suggestions

    def insert(self, word: str, frequency: int):
        node = self.root
        for ch in word.lower():
            if ch not in node.children:
                node.children[ch] = TrieNode()
            node = node.children[ch]
        node.is_end = True
        node.frequency = frequency
        self._update_suggestions(word, frequency)

    def _update_suggestions(self, word: str, freq: int):
        node = self.root
        for ch in word.lower():
            node = node.children[ch]
            suggestions = node.top_suggestions
            # Keep top-K sorted by frequency (use heap for efficiency)
            if len(suggestions) < self.max_suggestions:
                suggestions.append((freq, word))
                suggestions.sort(reverse=True)
            elif freq > suggestions[-1][0]:
                suggestions[-1] = (freq, word)
                suggestions.sort(reverse=True)

    def search(self, prefix: str) -> list[str]:
        node = self.root
        for ch in prefix.lower():
            if ch not in node.children:
                return []
            node = node.children[ch]
        return [word for _, word in node.top_suggestions]

def charge_card(
    user_id: int,
    amount_cents: int,
    currency: str,
    idempotency_key: str,  # client-generated UUID, stored by caller
) -> PaymentResult:
    # Check if we already processed this request
    existing = db.get_payment_by_idempotency_key(idempotency_key)
    if existing:
        return existing  # return same result, no double charge

    # Create payment record in PENDING state FIRST
    payment_id = db.create_payment(
        user_id=user_id,
        amount=amount_cents,
        currency=currency,
        idempotency_key=idempotency_key,
        status="PENDING",
    )

    try:
        # Call payment provider (Stripe, Adyen, etc.)
        provider_result = stripe.charge(
            amount=amount_cents,
            currency=currency,
            customer_id=user_id,
            idempotency_key=idempotency_key,
        )
        db.update_payment(payment_id, status="SUCCESS", provider_id=provider_result.id)
        return PaymentResult(success=True, payment_id=payment_id)
    except stripe.CardError as e:
        db.update_payment(payment_id, status="FAILED", error=str(e))
        return PaymentResult(success=False, error=str(e))
    except Exception as e:
        # Network timeout — status is UNKNOWN, not FAILED
        db.update_payment(payment_id, status="UNKNOWN", error=str(e))
        # Reconciliation job will resolve this async
        raise

PENDING → SUCCESS → REFUNDED
        → FAILED
        → UNKNOWN → (reconciliation) → SUCCESS | FAILED

CREATE TABLE ledger (
    id           BIGSERIAL PRIMARY KEY,
    account_id   BIGINT NOT NULL,
    amount_cents BIGINT NOT NULL,  -- positive = credit, negative = debit
    type         VARCHAR(50) NOT NULL,  -- 'charge', 'refund', 'payout'
    reference_id VARCHAR(255) NOT NULL,
    created_at   TIMESTAMPTZ DEFAULT now()
);
-- Balance = SUM(amount_cents) WHERE account_id = X

User → Upload Service → Raw Storage (S3)
                      → Transcoding Job Queue (Kafka)
                      → Transcoding Workers (FFmpeg)
                      → Multiple resolution outputs (S3: 360p, 720p, 1080p, 4K)
                      → CDN edge nodes

#EXTM3U
#EXT-X-VERSION:3
#EXT-X-STREAM-INF:BANDWIDTH=800000,RESOLUTION=640x360
360p/playlist.m3u8
#EXT-X-STREAM-INF:BANDWIDTH=2500000,RESOLUTION=1280x720
720p/playlist.m3u8
#EXT-X-STREAM-INF:BANDWIDTH=8000000,RESOLUTION=1920x1080
1080p/playlist.m3u8

CREATE TABLE videos (
    id           VARCHAR(11) PRIMARY KEY,  -- YouTube-style ID
    user_id      BIGINT NOT NULL,
    title        TEXT NOT NULL,
    description  TEXT,
    duration_sec INT,
    status       VARCHAR(20) NOT NULL,  -- 'processing', 'ready', 'failed'
    view_count   BIGINT DEFAULT 0,
    created_at   TIMESTAMPTZ DEFAULT now()
);

import geohash  # python-geohash library

# Driver location update
def update_driver_location(driver_id: int, lat: float, lng: float):
    gh = geohash.encode(lat, lng, precision=6)  # ~1km accuracy
    # Store in Redis geospatial index
    redis.geoadd("active_drivers", lng, lat, driver_id)
    # Also store in geohash bucket for fast proximity queries
    redis.sadd(f"drivers:geohash:{gh}", driver_id)
    redis.expire(f"drivers:geohash:{gh}", 30)  # drivers expire if not updated

# Find nearby drivers
def find_nearby_drivers(lat: float, lng: float, radius_km: float = 5) -> list[int]:
    results = redis.georadius(
        "active_drivers", lng, lat,
        radius_km, unit="km",
        withdist=True, count=20, sort="ASC"
    )
    return [(int(driver_id), dist) for driver_id, dist in results]

REQUESTED → ACCEPTED → DRIVER_ARRIVING → IN_PROGRESS → COMPLETED
          → CANCELLED (by rider or driver)
          → NO_DRIVER_FOUND

import uuid
import time
import redis

class DistributedLock:
    def __init__(self, redis_client, key: str, ttl_ms: int = 10000):
        self.redis = redis_client
        self.key = f"lock:{key}"
        self.ttl_ms = ttl_ms
        self.token = str(uuid.uuid4())

    def acquire(self, retry_times=3, retry_delay_ms=200) -> bool:
        for _ in range(retry_times):
            # SET key token NX PX ttl — atomic, only sets if not exists
            result = self.redis.set(
                self.key, self.token,
                nx=True, px=self.ttl_ms
            )
            if result:
                return True
            time.sleep(retry_delay_ms / 1000)
        return False

    def release(self):
        # Lua script ensures we only release our own lock (atomic check-and-delete)
        script = """
        if redis.call("get", KEYS[1]) == ARGV[1] then
            return redis.call("del", KEYS[1])
        else
            return 0
        end
        """
        self.redis.eval(script, 1, self.key, self.token)

    def __enter__(self):
        if not self.acquire():
            raise RuntimeError(f"Could not acquire lock: {self.key}")
        return self

    def __exit__(self, *args):
        self.release()

# Usage
with DistributedLock(redis_client, "order:12345", ttl_ms=5000):
    process_order(12345)

from enum import Enum
import time

class CircuitState(Enum):
    CLOSED = "closed"      # normal operation
    OPEN = "open"          # failing, reject all calls
    HALF_OPEN = "half_open"  # test if service recovered

class CircuitBreaker:
    def __init__(self, failure_threshold=5, recovery_timeout=60):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.failure_count = 0
        self.last_failure_time = None
        self.state = CircuitState.CLOSED

    def call(self, func, *args, **kwargs):
        if self.state == CircuitState.OPEN:
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = CircuitState.HALF_OPEN
            else:
                raise Exception("Circuit breaker OPEN — service unavailable")

        try:
            result = func(*args, **kwargs)
            if self.state == CircuitState.HALF_OPEN:
                self.reset()
            return result
        except Exception:
            self.failure_count += 1
            self.last_failure_time = time.time()
            if self.failure_count >= self.failure_threshold:
                self.state = CircuitState.OPEN
            raise

    def reset(self):
        self.failure_count = 0
        self.state = CircuitState.CLOSED

import hashlib

def compute_chunk_hash(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()

def upload_file(file_path: str, user_id: int, chunk_size: int = 4 * 1024 * 1024):
    chunks = []
    with open(file_path, "rb") as f:
        while True:
            data = f.read(chunk_size)
            if not data:
                break
            chunk_hash = compute_chunk_hash(data)
            chunks.append(chunk_hash)

            # Only upload if not already stored (content-addressable)
            if not chunk_store.exists(chunk_hash):
                chunk_store.put(chunk_hash, data)

    # Store file metadata: list of chunk hashes in order
    file_id = metadata_store.create_file(
        user_id=user_id,
        filename=os.path.basename(file_path),
        chunk_hashes=chunks,
        size=os.path.getsize(file_path),
    )
    return file_id

CREATE TABLE files (
    id           UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    user_id      BIGINT NOT NULL,
    parent_id    UUID REFERENCES files(id),  -- folder hierarchy
    name         VARCHAR(255) NOT NULL,
    size_bytes   BIGINT NOT NULL,
    is_folder    BOOLEAN NOT NULL DEFAULT FALSE,
    version      INT NOT NULL DEFAULT 1,
    chunk_hashes TEXT[],  -- ordered list of content-addressable chunk hashes
    created_at   TIMESTAMPTZ DEFAULT now(),
    updated_at   TIMESTAMPTZ DEFAULT now()
);

import numpy as np
from sklearn.metrics.pairwise import cosine_similarity

class ItemBasedCF:
    def __init__(self):
        self.item_vectors = {}  # item_id → embedding vector
        self.similarity_matrix = None

    def train(self, interaction_matrix: np.ndarray, item_ids: list[int]):
        # interaction_matrix: users × items, values = implicit feedback (views, clicks)
        # Transpose: items × users
        item_user_matrix = interaction_matrix.T
        self.similarity_matrix = cosine_similarity(item_user_matrix)
        self.item_ids = item_ids

    def recommend(self, user_history: list[int], n: int = 10) -> list[int]:
        if not user_history:
            return self._popular_items(n)

        # Average similarity across all items user interacted with
        history_indices = [self.item_ids.index(i) for i in user_history if i in self.item_ids]
        scores = self.similarity_matrix[history_indices].mean(axis=0)

        # Exclude already-seen items
        for idx in history_indices:
            scores[idx] = -1

        top_indices = np.argsort(scores)[-n:][::-1]
        return [self.item_ids[i] for i in top_indices]

import redis

class Leaderboard:
    def __init__(self, redis_client, name: str):
        self.redis = redis_client
        self.key = f"leaderboard:{name}"

    def add_score(self, user_id: int, score: float):
        self.redis.zadd(self.key, {str(user_id): score})

    def increment_score(self, user_id: int, delta: float) -> float:
        return self.redis.zincrby(self.key, delta, str(user_id))

    def get_rank(self, user_id: int) -> int:
        # ZREVRANK: rank from highest (rank 0 = #1)
        rank = self.redis.zrevrank(self.key, str(user_id))
        return rank + 1 if rank is not None else None

    def get_top_k(self, k: int = 10) -> list[dict]:
        results = self.redis.zrevrange(self.key, 0, k - 1, withscores=True)
        return [
            {"rank": i + 1, "user_id": int(uid), "score": score}
            for i, (uid, score) in enumerate(results)
        ]

    def get_user_context(self, user_id: int, window: int = 5) -> list[dict]:
        # Return user's neighbors on the leaderboard
        rank = self.redis.zrevrank(self.key, str(user_id))
        if rank is None:
            return []
        start = max(0, rank - window)
        end = rank + window
        results = self.redis.zrevrange(self.key, start, end, withscores=True)
        return [
            {"rank": start + i + 1, "user_id": int(uid), "score": score}
            for i, (uid, score) in enumerate(results)
        ]

# O(log N) for add/update/rank-lookup
# O(log N + K) for top-K
lb = Leaderboard(redis_client, "weekly_game_2024_01")
lb.add_score(user_id=42, score=9850)
lb.get_top_k(10)

final_score = 0.7 × global_freq + 0.3 × personal_freq

OrderService         → publishes ORDER_CREATED
  InventoryService   → reserves stock → publishes STOCK_RESERVED
    PaymentService   → charges card → publishes PAYMENT_PROCESSED
      ShipmentService → creates shipment → publishes ORDER_FULFILLED

On failure at any step → publish failure event → compensating transactions run in reverse
  ShipmentService fails → publishes SHIPMENT_FAILED
    PaymentService (compensate) → refunds charge
      InventoryService (compensate) → releases reservation

class OrderSaga:
    def __init__(self, order_id: int):
        self.order_id = order_id
        self.steps = [
            Step(
                action=lambda: inventory_service.reserve(order_id),
                compensation=lambda: inventory_service.release(order_id),
            ),
            Step(
                action=lambda: payment_service.charge(order_id),
                compensation=lambda: payment_service.refund(order_id),
            ),
            Step(
                action=lambda: shipment_service.create(order_id),
                compensation=lambda: shipment_service.cancel(order_id),
            ),
        ]

    def execute(self):
        completed = []
        for step in self.steps:
            try:
                step.action()
                completed.append(step)
            except Exception as e:
                # Rollback in reverse order
                for done_step in reversed(completed):
                    done_step.compensation()
                raise SagaRollbackError(f"Saga failed at step: {e}")

# FastAPI WebSocket gateway (simplified)
from fastapi import WebSocket
import asyncio

connected_users: dict[int, WebSocket] = {}

async def websocket_endpoint(websocket: WebSocket, user_id: int):
    await websocket.accept()
    connected_users[user_id] = websocket

    try:
        while True:
            data = await websocket.receive_json()

            if data["type"] == "message":
                await handle_message(user_id, data)
            elif data["type"] == "typing":
                await broadcast_typing(user_id, data["channel_id"])
    finally:
        del connected_users[user_id]

async def handle_message(sender_id: int, data: dict):
    msg = await message_store.save({
        "channel_id": data["channel_id"],
        "sender_id": sender_id,
        "content": data["content"],
        "created_at": datetime.utcnow().isoformat(),
    })

    # Deliver to all online members of the channel
    members = await channel_service.get_members(data["channel_id"])
    for member_id in members:
        if member_id in connected_users:
            await connected_users[member_id].send_json({
                "type": "message",
                "message": msg,
            })
        else:
            # User is offline — send push notification
            await push_service.notify(member_id, msg)

-- Cassandra
CREATE TABLE messages (
    channel_id  UUID,
    message_id  BIGINT,  -- Snowflake ID (time-ordered)
    sender_id   UUID,
    content     TEXT,
    PRIMARY KEY (channel_id, message_id)
) WITH CLUSTERING ORDER BY (message_id DESC);

class APIGateway:
    def __init__(self):
        self.auth = JWTAuthMiddleware()
        self.rate_limiter = TokenBucketRateLimiter(redis_client, capacity=100, refill_rate=10)
        self.router = ServiceRouter()
        self.circuit_breakers = {}  # service_name → CircuitBreaker

    async def handle(self, request: Request) -> Response:
        # 1. Auth
        try:
            user = await self.auth.authenticate(request)
        except AuthError:
            return Response(status=401, body="Unauthorized")

        # 2. Rate limiting
        if not self.rate_limiter.allow(user.id):
            return Response(status=429, body="Too Many Requests",
                          headers={"Retry-After": "1"})

        # 3. Route to upstream service
        service, path = self.router.resolve(request.path)

        # 4. Circuit breaker
        cb = self.circuit_breakers.setdefault(service, CircuitBreaker())

        try:
            upstream_response = await cb.call(
                http_client.forward, request, service, path
            )
        except Exception:
            return Response(status=503, body="Service temporarily unavailable")

        # 5. Response transformation (e.g., strip internal headers)
        return self.transform_response(upstream_response)

# Prometheus client instrumentation
from prometheus_client import Counter, Histogram, Gauge, start_http_server
import time

requests_total = Counter("http_requests_total", "Total HTTP requests", ["method", "endpoint", "status"])
request_duration = Histogram("http_request_duration_seconds", "Request duration", ["endpoint"])
active_connections = Gauge("active_connections", "Active WebSocket connections")

def track_request(func):
    def wrapper(request):
        start = time.time()
        try:
            response = func(request)
            requests_total.labels(
                method=request.method,
                endpoint=request.path,
                status=response.status_code
            ).inc()
            return response
        finally:
            request_duration.labels(endpoint=request.path).observe(time.time() - start)
    return wrapper

-- Step 1: Expand — add new column (both versions work)
ALTER TABLE users ADD COLUMN full_name VARCHAR(255);

-- Step 2: Migrate data (run as a background job)
UPDATE users SET full_name = first_name || ' ' || last_name WHERE full_name IS NULL;

-- Step 3: Deploy new code that reads full_name (old code still reads first_name)

-- Step 4: Contract — drop old columns (only after old code is fully retired)
ALTER TABLE users DROP COLUMN first_name;
ALTER TABLE users DROP COLUMN last_name;

-- Standard: locks the table
CREATE INDEX ON users (email);

-- PostgreSQL: build index without locking (takes longer but safe in prod)
CREATE INDEX CONCURRENTLY ON users (email);

# WRONG: race condition
def purchase(product_id, quantity):
    stock = db.query("SELECT stock FROM inventory WHERE id = ?", product_id)
    if stock >= quantity:  # another request reads stock=5 here too
        db.execute("UPDATE inventory SET stock = stock - ? WHERE id = ?", quantity, product_id)
        # Both succeed → stock goes negative

UPDATE inventory
SET stock = stock - :quantity, version = version + 1
WHERE product_id = :product_id
  AND stock >= :quantity
  AND version = :expected_version;
-- Returns 0 rows updated → retry with fresh data

def reserve_stock(product_id: int, quantity: int) -> bool:
    key = f"stock:{product_id}"
    # Lua script: atomic check-and-decrement
    script = """
    local stock = tonumber(redis.call("get", KEYS[1]))
    if stock == nil or stock < tonumber(ARGV[1]) then
        return 0
    end
    redis.call("decrby", KEYS[1], ARGV[1])
    return 1
    """
    result = redis.eval(script, 1, key, quantity)
    if result:
        # Async: persist reservation to DB
        queue.enqueue("persist_reservation", product_id, quantity)
    return bool(result)

Browser cache (TTL: seconds–minutes)
  → CDN edge cache (TTL: minutes–hours)
    → CDN origin shield (aggregates misses, reduces origin load)
      → Your origin servers

Cache-Control: public, max-age=86400          # cache for 1 day
Cache-Control: public, max-age=31536000, immutable  # forever (hashed filenames)
Cache-Control: no-store                       # never cache (sensitive data)
Vary: Accept-Encoding                         # separate cache for gzip vs brotli

# G-Counter (grow-only counter): each node increments its own slot
class GCounter:
    def __init__(self, node_id: str, nodes: list[str]):
        self.node_id = node_id
        self.counts = {n: 0 for n in nodes}

    def increment(self, delta=1):
        self.counts[self.node_id] += delta

    def value(self) -> int:
        return sum(self.counts.values())

    def merge(self, other: 'GCounter'):
        # Take element-wise maximum — always safe, no conflicts
        for node, count in other.counts.items():
            self.counts[node] = max(self.counts[node], count)

User visits page → Publisher sends bid request (user signals, page context)
                → Ad Exchange (real-time auction)
                → DSPs (Demand-Side Platforms) respond with bids in < 100ms
                → Highest bidder wins, ad served
                → Impression/click events logged

def run_auction(bid_request: BidRequest) -> AdCreative:
    # 1. Retrieve eligible ads for this user+context (< 5ms)
    user_segments = segment_store.get(bid_request.user_id)
    eligible_ads = ad_index.query(
        segments=user_segments,
        page_category=bid_request.page_category,
        format=bid_request.format,
    )

    # 2. Score each ad (relevance × bid price)
    scored_ads = [
        (ad, predict_ctr(ad, bid_request) * ad.max_cpm)
        for ad in eligible_ads
    ]

    # 3. Second-price auction (Vickrey): winner pays second-highest bid + $0.01
    scored_ads.sort(key=lambda x: x[1], reverse=True)
    winner, winning_score = scored_ads[0]
    clearing_price = scored_ads[1][1] + 0.01 if len(scored_ads) > 1 else scored_ads[0][1]

    # 4. Log impression asynchronously
    event_bus.publish("impression", {
        "ad_id": winner.id, "user_id": bid_request.user_id,
        "price": clearing_price, "timestamp": time.time()
    })

    return winner.creative

-- ClickHouse table for events
CREATE TABLE events (
    event_time   DateTime,
    user_id      UInt64,
    event_type   LowCardinality(String),
    session_id   String,
    properties   Map(String, String)
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(event_time)
ORDER BY (event_type, user_id, event_time);

-- Query: DAU for last 30 days
SELECT toDate(event_time) AS date, uniq(user_id) AS dau
FROM events
WHERE event_type = 'session_start'
  AND event_time >= now() - INTERVAL 30 DAY
GROUP BY date ORDER BY date;
-- Runs in < 1 second on billions of rows

{
  "schema_version": "1.0",
  "event_id": "uuid",
  "event_type": "page_view",
  "timestamp": "2024-01-15T10:30:00Z",
  "user_id": 12345,
  "session_id": "abc123",
  "properties": {
    "page": "/product/123",
    "referrer": "google.com",
    "device": "mobile"
  }
}

def delete_user_data(user_id: int):
    # Hard delete PII from all primary stores
    db.execute("DELETE FROM users WHERE id = ?", user_id)
    db.execute("DELETE FROM user_profiles WHERE user_id = ?", user_id)

    # Anonymize historical events (keep for analytics, remove PII)
    db.execute(
        "UPDATE events SET user_id = NULL, ip_address = NULL WHERE user_id = ?",
        user_id
    )

    # Purge from caches
    redis.delete(f"user:{user_id}:*")  # scan and delete all user keys

    # Purge from search indexes
    elasticsearch.delete_by_query({"term": {"user_id": user_id}})

    # Queue purge from backups (harder — flag for exclusion in backup restore)
    deletion_log.record(user_id, requested_at=datetime.utcnow())

# Job definition
{
    "job_id": "send-weekly-digest",
    "cron_expression": "0 9 * * MON",  # every Monday at 9am
    "timezone": "America/New_York",
    "handler": "digest_service.send_weekly",
    "timeout_sec": 300,
    "max_retries": 3,
    "concurrency_policy": "forbid"  # don't run if previous still running
}

def check_and_schedule_jobs():
    now = datetime.utcnow()
    due_jobs = db.query(
        "SELECT * FROM jobs WHERE next_run_at <= ? AND status = 'active'",
        now
    )
    for job in due_jobs:
        # Atomic claim: only one scheduler picks up this job
        updated = db.execute(
            "UPDATE jobs SET status = 'scheduled', next_run_at = ? "
            "WHERE job_id = ? AND status = 'active' AND next_run_at <= ?",
            compute_next_run(job.cron_expression), job.job_id, now
        )
        if updated.rowcount == 1:
            queue.enqueue("job_executions", {"job_id": job.job_id, "trigger_time": now})

Document 1: "the quick brown fox"
Document 2: "the lazy brown dog"

Inverted index:
  "the"   → [1, 2]
  "quick" → [1]
  "brown" → [1, 2]
  "fox"   → [1]
  "lazy"  → [2]
  "dog"   → [2]

Query "brown fox" → intersection of [1,2] and [1] = [1]

from collections import defaultdict
import re

class InvertedIndex:
    def __init__(self):
        self.index: dict[str, set[int]] = defaultdict(set)
        self.documents: dict[int, str] = {}

    def tokenize(self, text: str) -> list[str]:
        # Lowercase, split on non-alpha, remove stopwords
        stopwords = {"the", "a", "an", "is", "are", "was"}
        tokens = re.findall(r'[a-z]+', text.lower())
        return [t for t in tokens if t not in stopwords]

    def add_document(self, doc_id: int, content: str):
        self.documents[doc_id] = content
        for token in set(self.tokenize(content)):
            self.index[token].add(doc_id)

    def search(self, query: str) -> list[int]:
        tokens = self.tokenize(query)
        if not tokens:
            return []
        result = self.index.get(tokens[0], set())
        for token in tokens[1:]:
            result = result & self.index.get(token, set())
        return sorted(result)

class FeatureFlagService:
    def __init__(self, redis_client, db):
        self.redis = redis_client
        self.db = db
        self.local_cache = {}  # in-process cache, TTL 30s

    def is_enabled(self, flag_name: str, user_id: int, context: dict = None) -> bool:
        flag = self._get_flag(flag_name)
        if not flag:
            return False  # unknown flags default to off

        # Killswitch: globally disabled
        if not flag["enabled"]:
            return False

        # User-specific override
        if user_id in flag.get("user_overrides", {}):
            return flag["user_overrides"][user_id]

        # Cohort-based (e.g., beta users, employees)
        if context and context.get("cohort") in flag.get("allowed_cohorts", []):
            return True

        # Percentage rollout (deterministic: same user always gets same result)
        if "rollout_percentage" in flag:
            bucket = hash(f"{flag_name}:{user_id}") % 100
            return bucket < flag["rollout_percentage"]

        return flag.get("default", False)

    def _get_flag(self, name: str) -> dict:
        # L1: in-process cache
        if name in self.local_cache and not self._is_stale(name):
            return self.local_cache[name]

        # L2: Redis
        cached = self.redis.get(f"flag:{name}")
        if cached:
            flag = json.loads(cached)
            self.local_cache[name] = flag
            return flag

        # L3: DB
        flag = self.db.get_flag(name)
        if flag:
            self.redis.setex(f"flag:{name}", 30, json.dumps(flag))
            self.local_cache[name] = flag
        return flag

Latency                          Throughput
─────────────────────────────    ─────────────────────────────────
L1 cache:         0.5 ns         Single-core compute:  ~1B ops/sec
L2 cache:           5 ns         Redis:              ~1M ops/sec
RAM:               100 ns        Postgres writes:    ~5K-10K/sec
SSD random read:   100 µs        Kafka:              ~1M msgs/sec
Network (same DC): 0.5 ms        S3 GET:             ~5,500 req/s
Network (cross):   100 ms        CDN edge:          ~100K req/s

Storage
─────────────────────────────────────────
1B users × 1KB profile  = 1 TB
1B rows × 100 bytes     = 100 GB
1M req/sec × 1KB/req    = 1 GB/sec ingress
100 bytes/msg × 1M msg/s = 100 MB/s Kafka throughput