System Design Interview Questions: How to Answer Them (40 Questions)

import string

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

def encode(n: int) -> str:
    if n == 0:
        return BASE62[0]
    result = []
    while n:
        result.append(BASE62[n % 62])
        n //= 62
    return ''.join(reversed(result))

def decode(s: str) -> int:
    result = 0
    for char in s:
        result = result * 62 + BASE62.index(char)
    return result

# encode(12345678) → "5lurZ" (6 chars covers 56B URLs)

import secrets
import string

def generate_code(length=7):
    alphabet = string.ascii_letters + string.digits
    return ''.join(secrets.choice(alphabet) for _ in range(length))

# On write: generate code, check DB, retry if collision
# Collision probability at 7 chars: negligible until ~3B URLs

CREATE TABLE urls (
    short_code  VARCHAR(10) PRIMARY KEY,
    long_url    TEXT NOT NULL,
    user_id     BIGINT,
    created_at  TIMESTAMP DEFAULT NOW(),
    expires_at  TIMESTAMP,
    click_count BIGINT DEFAULT 0
);

CREATE INDEX idx_long_url ON urls(long_url);  -- for dedup check

Client
  ↓
CDN (cache redirect responses with Cache-Control: 301, max-age=3600)
  ↓
Load Balancer
  ↓
App Servers (stateless, horizontal scale)
  ↓
Redis Cache (short_code → long_url, TTL = 24h)
  ↓ (cache miss)
PostgreSQL Primary (writes) + Read Replicas (reads)

-- Lua script runs atomically on Redis
local key = KEYS[1]
local rate = tonumber(ARGV[1])      -- tokens per second
local capacity = tonumber(ARGV[2])  -- max tokens in bucket
local now = tonumber(ARGV[3])       -- current timestamp (ms)
local requested = tonumber(ARGV[4]) -- tokens needed for this request

local bucket = redis.call("HMGET", key, "tokens", "last_refill")
local tokens = tonumber(bucket[1]) or capacity
local last_refill = tonumber(bucket[2]) or now

-- Refill tokens based on elapsed time
local elapsed = (now - last_refill) / 1000.0
tokens = math.min(capacity, tokens + elapsed * rate)

if tokens >= requested then
    tokens = tokens - requested
    redis.call("HMSET", key, "tokens", tokens, "last_refill", now)
    redis.call("EXPIRE", key, 3600)
    return 1  -- allowed
else
    redis.call("HMSET", key, "tokens", tokens, "last_refill", now)
    redis.call("EXPIRE", key, 3600)
    return 0  -- denied
end

import redis
import time

r = redis.Redis()

def is_allowed(user_id: str, limit: int, window_seconds: int) -> bool:
    now = time.time()
    window_start = now - window_seconds
    key = f"ratelimit:{user_id}"
    
    pipe = r.pipeline()
    # Remove events outside the window
    pipe.zremrangebyscore(key, 0, window_start)
    # Count events in current window
    pipe.zcard(key)
    # Add current event
    pipe.zadd(key, {str(now): now})
    # Set TTL to clean up old keys
    pipe.expire(key, window_seconds * 2)
    results = pipe.execute()
    
    current_count = results[1]
    return current_count < limit

Request → API Gateway
  → Extract identifier (user_id, API key, IP)
  → Rate Limit Service
      → Redis Cluster (consistent hashing, same user always hits same shard)
      → Return Allow/Deny + Retry-After header
  → Forward to backend (if allowed) or return 429 (if denied)

X-RateLimit-Limit: 1000
X-RateLimit-Remaining: 743
X-RateLimit-Reset: 1735689600
Retry-After: 3600  (only on 429)

import time
import threading
from collections import defaultdict
from typing import Optional

class KeyValueStore:
    def __init__(self):
        self._store: dict = {}
        self._expiry: dict = {}
        self._lock = threading.RLock()
    
    def set(self, key: str, value, ttl_seconds: Optional[int] = None):
        with self._lock:
            self._store[key] = value
            if ttl_seconds is not None:
                self._expiry[key] = time.time() + ttl_seconds
            elif key in self._expiry:
                del self._expiry[key]
    
    def get(self, key: str):
        with self._lock:
            if key in self._expiry and time.time() > self._expiry[key]:
                del self._store[key]
                del self._expiry[key]
                return None
            return self._store.get(key)
    
    def delete(self, key: str) -> bool:
        with self._lock:
            if key in self._store:
                del self._store[key]
                self._expiry.pop(key, None)
                return True
            return False

import hashlib
from bisect import bisect, insort

class ConsistentHash:
    def __init__(self, replicas=150):
        self.replicas = replicas
        self.ring = []
        self.nodes = {}
    
    def add_node(self, node: str):
        for i in range(self.replicas):
            virtual_node = f"{node}:{i}"
            h = int(hashlib.md5(virtual_node.encode()).hexdigest(), 16)
            insort(self.ring, h)
            self.nodes[h] = node
    
    def remove_node(self, node: str):
        for i in range(self.replicas):
            virtual_node = f"{node}:{i}"
            h = int(hashlib.md5(virtual_node.encode()).hexdigest(), 16)
            self.ring.remove(h)
            del self.nodes[h]
    
    def get_node(self, key: str) -> str:
        h = int(hashlib.md5(key.encode()).hexdigest(), 16)
        idx = bisect(self.ring, h) % len(self.ring)
        return self.nodes[self.ring[idx]]

Topic: "user-events"
  Partition 0: [e1, e4, e7, e10, ...]
  Partition 1: [e2, e5, e8, e11, ...]
  Partition 2: [e3, e6, e9, e12, ...]

Consumer Group A:
  Consumer A1 → reads Partition 0
  Consumer A2 → reads Partition 1
  Consumer A3 → reads Partition 2

Consumer Group B (independent):
  Consumer B1 → reads all 3 partitions (if only 1 consumer in group)

from kafka import KafkaProducer
import json

producer = KafkaProducer(
    bootstrap_servers=['kafka:9092'],
    value_serializer=lambda v: json.dumps(v).encode('utf-8'),
    acks='all',              # wait for all replicas (durability)
    retries=3,
    compression_type='gzip'
)

def publish_user_event(user_id: str, event: dict):
    # partition_key ensures all events for same user go to same partition
    # → ordering guaranteed per user
    producer.send(
        topic='user-events',
        key=user_id.encode('utf-8'),  # partition key
        value=event
    )

publish_user_event("user-123", {"type": "page_view", "url": "/dashboard"})

from kafka import KafkaConsumer
import json

consumer = KafkaConsumer(
    'user-events',
    bootstrap_servers=['kafka:9092'],
    group_id='analytics-service',
    value_deserializer=lambda m: json.loads(m.decode('utf-8')),
    enable_auto_commit=False,  # manual commit for control
    auto_offset_reset='earliest'
)

for message in consumer:
    try:
        process_event(message.value)
        consumer.commit()  # only commit after successful processing
    except Exception as e:
        # don't commit — message will be redelivered
        log_error(e)

User tweets
  → Write to tweet table (async)
  → Enqueue fan-out job
  → Fan-out worker reads follower list (in batches of 1000)
  → For each follower: append tweet_id to their timeline cache in Redis
  
Read timeline:
  → Fetch tweet_ids from Redis (O(1) per user)
  → Fetch tweet details by IDs (cache-friendly)
  → Return merged timeline

User tweets
  → Write to tweet table
  
Read timeline:
  → Fetch list of accounts user follows
  → Query each account's recent tweets (since last read)
  → Merge and sort by time

Timeline Request:
  1. Fetch pre-computed timeline from Redis
  2. Identify celebrities user follows
  3. For each celebrity: fetch their recent tweets
  4. Merge + deduplicate + sort
  5. Return top N

CREATE TABLE tweets (
    tweet_id    BIGINT PRIMARY KEY,  -- snowflake ID (time-sortable)
    user_id     BIGINT NOT NULL,
    content     TEXT NOT NULL,
    created_at  TIMESTAMP NOT NULL,
    reply_to_id BIGINT,
    retweet_id  BIGINT
);

CREATE TABLE follows (
    follower_id  BIGINT NOT NULL,
    followee_id  BIGINT NOT NULL,
    created_at   TIMESTAMP NOT NULL,
    PRIMARY KEY (follower_id, followee_id)
);

CREATE INDEX idx_tweets_user ON tweets(user_id, created_at DESC);
CREATE INDEX idx_follows_followee ON follows(followee_id);  -- for fan-out

64-bit ID structure:
[41 bits: timestamp ms] [10 bits: machine ID] [12 bits: sequence]

Max throughput: 4096 IDs/ms per machine
IDs are roughly time-sorted — good for feed ordering

1. User uploads raw video (direct to object storage via presigned URL)
2. Upload service validates and enqueues transcoding job
3. Transcoding workers (GPU instances) convert to multiple resolutions:
   - 1080p, 720p, 480p, 360p, 240p
   - Multiple codecs: H.264 (wide compatibility), VP9/AV1 (better compression)
4. Output segments stored as HLS (.m3u8 + .ts chunks) or DASH
5. CDN origins pull from object storage; CDN edges cache popular segments
6. Video metadata written to DB; thumbnail generated
7. Notification sent to subscribers (async, via Kafka)

Player downloads manifest:
  720p: /video/abc/720p/index.m3u8
  480p: /video/abc/480p/index.m3u8
  360p: /video/abc/360p/index.m3u8

Player monitors bandwidth in real time.
If bandwidth drops → switches to lower quality playlist.
If bandwidth improves → switches to higher quality.
User gets smooth playback without buffering.

# Driver app sends location every 4 seconds
# Not every update needs to go to the DB

# Architecture:
# Driver → WebSocket → Location Service → Redis (current location)
#                                       → Kafka (location stream for analytics)
# Writes: ~1M drivers × 15 updates/min = 250K writes/sec to Redis

# Redis Geo commands (built-in geospatial index)
import redis
r = redis.Redis()

# Driver sends location update
r.geoadd("active_drivers", (longitude, latitude, driver_id))

# Rider requests a ride at (lat, lng)
# Find all drivers within 5km
nearby_drivers = r.georadius(
    "active_drivers",
    longitude=rider_lng,
    latitude=rider_lat,
    radius=5,
    unit="km",
    withdist=True,
    sort="ASC",
    count=20
)

1. Find N nearest available drivers (Redis GEO query)
2. Rank by: estimated arrival time + driver rating + acceptance rate
3. Offer to top-ranked driver (timeout: 10 seconds)
4. If declined/timeout → offer to next driver
5. On acceptance: create trip, notify both parties

import h3

def get_cell(lat: float, lng: float, resolution: int = 9) -> str:
    # Resolution 9 ≈ hexagons of ~0.1 km² area
    return h3.geo_to_h3(lat, lng, resolution)

def find_nearby_cells(lat: float, lng: float, rings: int = 2) -> list:
    center = get_cell(lat, lng)
    # k-ring returns all hexagons within k rings of center
    return list(h3.k_ring(center, rings))

Sent (server received) → Delivered (recipient device received) → Read (recipient opened)

Client A                 Server                    Client B
  |                         |                         |
  |--- send(msg, to=B) --→  |                         |
  |                         |-- B online? ----------→ |
  |                         |  ← yes ---------------  |
  |← ack(SENT) ------------ |-- push(msg) ----------→ |
  |                         |← ack(DELIVERED) -------- |
  |← notification(DELIVERED)|                         |
  |                         |← ack(READ) ------------ |  (B opens chat)
  |← notification(READ) ----|                         |

CREATE TABLE messages (
    message_id   BIGINT PRIMARY KEY,  -- snowflake
    conversation_id BIGINT NOT NULL,
    sender_id    BIGINT NOT NULL,
    content      BYTEA NOT NULL,      -- encrypted payload
    sent_at      TIMESTAMP NOT NULL,
    delivered_at TIMESTAMP,
    read_at      TIMESTAMP
);

-- Offline message queue (temporary)
CREATE TABLE pending_deliveries (
    message_id   BIGINT NOT NULL,
    recipient_id BIGINT NOT NULL,
    retry_count  INT DEFAULT 0,
    PRIMARY KEY (message_id, recipient_id)
);

from collections import deque
import threading

class WebCrawler:
    def __init__(self, seed_urls: list, max_depth: int = 3):
        self.queue = deque([(url, 0) for url in seed_urls])
        self.visited = set()
        self.max_depth = max_depth
        self.politeness_delay = {}  # domain → last_crawl_time
    
    def crawl(self):
        while self.queue:
            url, depth = self.queue.popleft()
            if url in self.visited or depth > self.max_depth:
                continue
            
            # Respect robots.txt and crawl delay
            if not self.is_allowed(url):
                continue
            
            content = self.fetch(url)
            self.visited.add(url)
            
            # Extract and store
            text = self.extract_text(content)
            links = self.extract_links(content)
            self.store(url, text)
            
            # Enqueue new URLs
            for link in links:
                if link not in self.visited:
                    self.queue.append((link, depth + 1))

Forward index:  doc_id → [word1, word2, ...]
Inverted index: word → [(doc_id, tf, positions), ...]

Example:
  "python tutorial" query
  → Lookup "python": [(doc1, tf=5, [2,15,89]), (doc3, tf=2, [1,7]), ...]
  → Lookup "tutorial": [(doc1, tf=3, [3,20]), (doc2, tf=8, [1,...]), ...]
  → Intersect: docs containing BOTH terms
  → Score by TF-IDF + PageRank + hundreds of other signals
  → Return top K

def pagerank(graph: dict, damping: float = 0.85, iterations: int = 100):
    N = len(graph)
    rank = {node: 1.0 / N for node in graph}
    
    for _ in range(iterations):
        new_rank = {}
        for node in graph:
            incoming_sum = sum(
                rank[src] / len(graph[src])
                for src in graph
                if node in graph[src]
            )
            new_rank[node] = (1 - damping) / N + damping * incoming_sum
        rank = new_rank
    
    return rank

def get_user(user_id: str) -> dict:
    # 1. Check cache
    cached = redis.get(f"user:{user_id}")
    if cached:
        return json.loads(cached)
    
    # 2. Cache miss — query DB
    user = db.query("SELECT * FROM users WHERE id = %s", user_id)
    
    # 3. Populate cache
    redis.setex(
        f"user:{user_id}",
        3600,  # TTL: 1 hour
        json.dumps(user)
    )
    
    return user

def update_user(user_id: str, data: dict):
    # Write to DB first (source of truth)
    db.execute("UPDATE users SET ... WHERE id = %s", user_id)
    # Write to cache (keep in sync)
    redis.setex(f"user:{user_id}", 3600, json.dumps(data))

import threading

_locks = {}
_locks_mutex = threading.Lock()

def get_with_mutex(key: str, fetch_fn, ttl: int):
    cached = redis.get(key)
    if cached:
        return json.loads(cached)
    
    # Ensure only one request fetches from DB
    with _locks_mutex:
        if key not in _locks:
            _locks[key] = threading.Lock()
        lock = _locks[key]
    
    with lock:
        # Double-check after acquiring lock
        cached = redis.get(key)
        if cached:
            return json.loads(cached)
        
        value = fetch_fn()
        redis.setex(key, ttl, json.dumps(value))
        return value

import math
import random
import time

def get_with_early_expiration(key: str, fetch_fn, ttl: int, beta: float = 1.0):
    data = redis.get(key)
    if data:
        value, expiry = json.loads(data)
        # Probabilistically refresh before expiry
        remaining = expiry - time.time()
        if remaining > 0:
            jitter = -beta * math.log(random.random())
            if remaining > jitter:
                return value
    
    # Fetch and cache
    value = fetch_fn()
    expiry = time.time() + ttl
    redis.setex(key, ttl, json.dumps([value, expiry]))
    return value

import heapq
import threading

class LoadBalancer:
    def __init__(self, servers: list):
        # Min-heap: (active_connections, server_id)
        self.heap = [(0, s) for s in servers]
        heapq.heapify(self.heap)
        self.lock = threading.Lock()
    
    def get_server(self) -> str:
        with self.lock:
            count, server = heapq.heappop(self.heap)
            heapq.heappush(self.heap, (count + 1, server))
            return server
    
    def release(self, server: str):
        with self.lock:
            # Decrement this server's count
            for i, (count, s) in enumerate(self.heap):
                if s == server:
                    self.heap[i] = (count - 1, s)
                    heapq.heapify(self.heap)
                    break

import asyncio
import aiohttp

class HealthChecker:
    def __init__(self, servers: list, interval: int = 10):
        self.servers = servers
        self.healthy = set(servers)
        self.interval = interval
    
    async def check_server(self, server: str):
        try:
            async with aiohttp.ClientSession() as session:
                async with session.get(
                    f"http://{server}/health",
                    timeout=aiohttp.ClientTimeout(total=2)
                ) as resp:
                    if resp.status == 200:
                        self.healthy.add(server)
                    else:
                        self.healthy.discard(server)
        except Exception:
            self.healthy.discard(server)
    
    async def run(self):
        while True:
            await asyncio.gather(*[self.check_server(s) for s in self.servers])
            await asyncio.sleep(self.interval)

63 bits total (signed int64, but using as unsigned)

[1 bit: always 0][41 bits: ms timestamp][10 bits: machine ID][12 bits: sequence]

Timestamp: 41 bits = 2^41 ms = ~69 years from epoch
Machine ID: 10 bits = 1024 machines max
Sequence: 12 bits = 4096 IDs per millisecond per machine
Total: 4096 × 1024 machines × 1000 ms = ~4.2 billion IDs/sec

import time
import threading

class SnowflakeGenerator:
    def __init__(self, machine_id: int, epoch: int = 1609459200000):  # 2021-01-01
        assert 0 <= machine_id < 1024
        self.machine_id = machine_id
        self.epoch = epoch
        self.sequence = 0
        self.last_ms = -1
        self.lock = threading.Lock()
    
    def next_id(self) -> int:
        with self.lock:
            ms = int(time.time() * 1000) - self.epoch
            
            if ms == self.last_ms:
                self.sequence = (self.sequence + 1) & 0xFFF  # 12-bit mask
                if self.sequence == 0:
                    # Sequence exhausted — wait for next millisecond
                    while ms <= self.last_ms:
                        ms = int(time.time() * 1000) - self.epoch
            else:
                self.sequence = 0
            
            self.last_ms = ms
            
            return (
                (ms << 22) |
                (self.machine_id << 12) |
                self.sequence
            )

gen = SnowflakeGenerator(machine_id=42)
id1 = gen.next_id()
id2 = gen.next_id()
assert id1 < id2  # time-sortable

User (Brazil)
  ↓ DNS lookup → Anycast routes to nearest edge
Edge PoP (São Paulo)
  ↓ cache hit? → return immediately (< 5ms latency)
  ↓ cache miss
Regional Cache (US-East)
  ↓ cache hit? → return + populate edge
  ↓ cache miss
Origin Server (your datacenter)
  → return content + populate regional cache + edge cache

Default: method + URL → GET https://cdn.example.com/image.jpg

But be careful with:
- Query strings: is /image.jpg?v=1 the same as /image.jpg?v=2? Usually not.
- Cookies: never include session cookies in cache key (every user gets different content)
- Accept-Encoding: gzip vs. brotli may be served differently
- Vary header: tells CDN to vary cache by specific request headers

class TrieNode:
    def __init__(self):
        self.children: dict = {}
        self.top_searches: list = []  # pre-computed top-K for this prefix
        self.is_end: bool = False

class AutocompleteService:
    def __init__(self, k: int = 5):
        self.root = TrieNode()
        self.k = k
    
    def insert(self, word: str, frequency: int):
        node = self.root
        for char in word.lower():
            if char not in node.children:
                node.children[char] = TrieNode()
            node = node.children[char]
            # Update top-K at each prefix node
            node.top_searches = self._merge_top_k(
                node.top_searches, (word, frequency)
            )
        node.is_end = True
    
    def search(self, prefix: str) -> list:
        node = self.root
        for char in prefix.lower():
            if char not in node.children:
                return []
            node = node.children[char]
        return [word for word, _ in node.top_searches]
    
    def _merge_top_k(self, current: list, new_entry: tuple) -> list:
        current.append(new_entry)
        current.sort(key=lambda x: -x[1])
        return current[:self.k]

# Offline job: compute top queries per prefix
def build_index(top_queries: list):
    pipe = r.pipeline()
    for query, score in top_queries:
        for i in range(1, len(query) + 1):
            prefix = query[:i]
            pipe.zadd(f"autocomplete:{prefix}", {query: score})
    pipe.execute()

# Real-time lookup: < 1ms
def get_suggestions(prefix: str, k: int = 5) -> list:
    return r.zrevrange(f"autocomplete:{prefix}", 0, k - 1)

Event Source (any service)
  → Kafka topic: "notification-events"
  → Notification Service (fan-out per channel)
      → Email Worker → SendGrid / SES
      → Push Worker → FCM (Android) / APNs (iOS)
      → SMS Worker → Twilio
      → In-App Worker → WebSocket / SSE

CREATE TABLE notification_preferences (
    user_id     BIGINT NOT NULL,
    channel     VARCHAR(20) NOT NULL,  -- 'email', 'push', 'sms'
    event_type  VARCHAR(50) NOT NULL,  -- 'new_message', 'payment', etc.
    enabled     BOOLEAN DEFAULT TRUE,
    PRIMARY KEY (user_id, channel, event_type)
);

import firebase_admin
from firebase_admin import messaging
import time

def send_push_with_retry(token: str, title: str, body: str, max_retries: int = 3):
    message = messaging.Message(
        notification=messaging.Notification(title=title, body=body),
        token=token
    )
    
    for attempt in range(max_retries):
        try:
            response = messaging.send(message)
            return {"status": "delivered", "message_id": response}
        except messaging.UnregisteredError:
            # Token is invalid — remove from DB
            invalidate_token(token)
            return {"status": "invalid_token"}
        except Exception as e:
            if attempt < max_retries - 1:
                time.sleep(2 ** attempt)  # exponential backoff
            else:
                # Move to dead letter queue for manual inspection
                dlq.enqueue({"token": token, "title": title, "body": body, "error": str(e)})
                return {"status": "failed"}

def process_payment(
    idempotency_key: str,
    user_id: str,
    amount: int,  # in cents
    currency: str
) -> dict:
    # Check if we've already processed this request
    existing = db.get_payment_by_idempotency_key(idempotency_key)
    if existing:
        return existing  # return same response as original
    
    # Acquire distributed lock (prevents race conditions)
    lock_key = f"payment_lock:{idempotency_key}"
    with redis_lock(lock_key, timeout=30):
        # Double-check (another process may have completed while we waited)
        existing = db.get_payment_by_idempotency_key(idempotency_key)
        if existing:
            return existing
        
        # Process the payment
        result = payment_provider.charge(amount, currency)
        
        # Write result atomically
        payment_record = {
            "idempotency_key": idempotency_key,
            "user_id": user_id,
            "amount": amount,
            "currency": currency,
            "status": result.status,
            "provider_id": result.charge_id
        }
        db.insert_payment(payment_record)
        return payment_record

Phase 1 (Prepare):
  Coordinator → Reserve funds (deduct from wallet, mark as "pending")
  Coordinator → Reserve inventory (decrement stock, mark as "held")
  If both OK → proceed to commit

Phase 2 (Commit):
  Coordinator → Confirm wallet deduction → mark as "completed"
  Coordinator → Confirm inventory hold → mark as "sold"
  
  If either fails in Phase 2 → compensating transactions to reverse

OrderService: create_order() → emit "ORDER_CREATED"
PaymentService: on "ORDER_CREATED" → charge() → emit "PAYMENT_COMPLETED" OR "PAYMENT_FAILED"
InventoryService: on "PAYMENT_COMPLETED" → reserve() → emit "INVENTORY_RESERVED"
ShippingService: on "INVENTORY_RESERVED" → schedule_shipment()

Compensations (if any step fails):
  "INVENTORY_FAILED" → PaymentService: refund()
  "PAYMENT_FAILED" → OrderService: cancel_order()

CREATE TABLE reservations (
    id          BIGINT PRIMARY KEY,
    resource_id BIGINT NOT NULL,  -- hotel room, seat, etc.
    user_id     BIGINT NOT NULL,
    check_in    DATE NOT NULL,
    check_out   DATE NOT NULL,
    status      VARCHAR(20) DEFAULT 'confirmed',
    version     INT DEFAULT 0  -- for optimistic locking
);

-- On update, include version in WHERE clause
UPDATE reservations
SET status = 'confirmed', version = version + 1
WHERE id = 123 AND version = 5;  -- fails if someone else already updated

-- If rows_affected = 0 → conflict! Retry or fail.

-- For seat reservations: a seat can only be booked once per event
CREATE UNIQUE INDEX idx_no_double_book
ON seat_reservations (event_id, seat_id)
WHERE status != 'cancelled';

-- For hotel rooms: exclude date range overlaps
-- PostgreSQL supports exclusion constraints with daterange
ALTER TABLE room_reservations
ADD CONSTRAINT no_overlap
EXCLUDE USING gist (
    room_id WITH =,
    daterange(check_in, check_out) WITH &&
);

1. User selects seats → Server creates PENDING reservation (holds seat for 10 min)
2. User completes payment → Server confirms reservation + charges payment
3. If timeout: PENDING reservation expires → seat released back to available pool

Background job: cleanup expired PENDING reservations every minute

# Distributed crawler architecture

# Components:
# 1. URL Frontier: prioritized queue of URLs to crawl
# 2. Fetcher: downloads URLs, respects robots.txt + crawl delay
# 3. Parser: extracts text + outlinks from HTML
# 4. URL Filter: deduplication, normalization
# 5. Content Store: saves raw + parsed content

class DistributedCrawler:
    def __init__(self):
        self.url_frontier = KafkaProducer(topic='urls_to_crawl')
        self.seen_urls = BloomFilter(capacity=10_000_000_000, error_rate=0.001)
        self.robots_cache = {}  # domain → parsed robots.txt
    
    def should_crawl(self, url: str) -> bool:
        # Bloom filter: fast probabilistic deduplication
        if url in self.seen_urls:
            return False
        # Check robots.txt
        domain = extract_domain(url)
        if not self.is_allowed_by_robots(domain, url):
            return False
        return True
    
    def fetch_and_parse(self, url: str):
        # Respect crawl delay
        self.throttle(extract_domain(url))
        
        response = requests.get(url, timeout=5, headers={'User-Agent': 'MyBot/1.0'})
        
        if response.status_code == 200:
            text = extract_text(response.content)
            links = extract_links(response.content, base_url=url)
            
            # Store
            content_store.save(url, text)
            
            # Enqueue new URLs
            for link in links:
                normalized = normalize_url(link)
                if self.should_crawl(normalized):
                    self.seen_urls.add(normalized)
                    self.url_frontier.send(normalized)

Raw Events → Kafka
  ↓ (real-time path)
  Stream Processor (Flink/Spark Streaming)
    → Computes rolling 1-min, 5-min, 1-hour aggregates
    → Stores in Redis (query latency: < 10ms)
  ↓ (batch path)
  Batch Processor (Spark, runs every hour)
    → Computes complete historical aggregates
    → Stores in ClickHouse/BigQuery (query latency: 1-10s)

Dashboard query:
  Recent data (last hour) → Redis
  Historical data → ClickHouse
  Merge in application layer

-- ClickHouse: columnar storage, vectorized execution
-- Aggregating 1B rows in 2-3 seconds is typical

SELECT
    toStartOfHour(event_time) AS hour,
    country,
    count() AS events,
    uniqExact(user_id) AS unique_users,
    avg(session_duration) AS avg_duration
FROM events
WHERE event_time >= now() - INTERVAL 7 DAY
  AND event_type = 'page_view'
GROUP BY hour, country
ORDER BY hour DESC, events DESC;

-- With proper partitioning (by date) and MergeTree engine,
-- this query scans only the relevant partitions

import redis
import time
import uuid
from contextlib import contextmanager

class RedisLock:
    def __init__(self, redis_client, key: str, timeout: int = 30):
        self.redis = redis_client
        self.key = f"lock:{key}"
        self.timeout = timeout
        self.identifier = str(uuid.uuid4())
    
    def acquire(self) -> bool:
        # SET key value NX PX timeout
        # NX: only set if not exists (atomic)
        # PX: millisecond TTL (prevents deadlock if process dies)
        return self.redis.set(
            self.key,
            self.identifier,
            nx=True,
            px=self.timeout * 1000
        )
    
    def release(self) -> bool:
        # Lua script: only release if we own the lock
        # Prevents accidentally releasing someone else's lock
        script = """
        if redis.call("get", KEYS[1]) == ARGV[1] then
            return redis.call("del", KEYS[1])
        else
            return 0
        end
        """
        return bool(self.redis.eval(script, 1, self.key, self.identifier))
    
    @contextmanager
    def __call__(self):
        acquired = False
        try:
            # Retry acquisition with backoff
            for _ in range(10):
                if self.acquire():
                    acquired = True
                    break
                time.sleep(0.1)
            if not acquired:
                raise TimeoutError(f"Could not acquire lock: {self.key}")
            yield
        finally:
            if acquired:
                self.release()

# Usage:
lock = RedisLock(r, "critical-section")
with lock():
    # Only one process executes this at a time
    process_data()

import hashlib

CHUNK_SIZE = 4 * 1024 * 1024  # 4MB chunks

def upload_file(file_path: str, user_id: str) -> str:
    chunks = []
    with open(file_path, 'rb') as f:
        while True:
            data = f.read(CHUNK_SIZE)
            if not data:
                break
            
            chunk_hash = hashlib.sha256(data).hexdigest()
            chunks.append(chunk_hash)
            
            # Check if chunk already exists in storage (deduplication)
            if not chunk_store.exists(chunk_hash):
                chunk_store.put(chunk_hash, data)
    
    # Store file manifest: ordered list of chunk hashes
    file_hash = hashlib.sha256(''.join(chunks).encode()).hexdigest()
    manifest = {
        "file_hash": file_hash,
        "chunks": chunks,
        "user_id": user_id,
        "created_at": time.time()
    }
    manifest_store.put(file_hash, manifest)
    return file_hash

def sync_file(local_path: str, remote_manifest: dict) -> list:
    # Compare local chunks against remote chunks
    # Only upload chunks that differ or are new
    local_chunks = compute_chunks(local_path)
    remote_chunks = set(remote_manifest.get("chunks", []))
    
    chunks_to_upload = [c for c in local_chunks if c not in remote_chunks]
    return chunks_to_upload

Alice                    Signaling Server              Bob
  | --- offer SDP ----→ |                              |
  |                     | ------- forward offer ----→  |
  |                     | ←------ answer SDP --------- |
  | ←--- forward ans -- |                              |
  |                                                    |
  | ←------------- ICE candidates exchanged ---------- |
  |                                                    |
  | ←===================== Direct P2P ===============→ |
  |                (bypasses servers entirely)         |

Participant A → SFU → Participant B
Participant A → SFU → Participant C
Participant B → SFU → Participant A
Participant B → SFU → Participant C
(etc.)

Upload: 1 stream per participant
Download: N-1 streams per participant (or simulcast + adaptive forwarding)

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

class CollaborativeFilter:
    def __init__(self):
        self.user_item_matrix = None
        self.user_similarities = None
    
    def fit(self, interactions: dict):
        # interactions: {user_id: {item_id: rating}}
        users = list(interactions.keys())
        items = list({item for ratings in interactions.values() for item in ratings})
        
        # Build sparse matrix
        matrix = np.zeros((len(users), len(items)))
        user_idx = {u: i for i, u in enumerate(users)}
        item_idx = {item: i for i, item in enumerate(items)}
        
        for user, ratings in interactions.items():
            for item, rating in ratings.items():
                matrix[user_idx[user]][item_idx[item]] = rating
        
        self.user_item_matrix = matrix
        self.user_similarities = cosine_similarity(matrix)
        self.users = users
        self.items = items
        self.user_idx = user_idx
        self.item_idx = item_idx
    
    def recommend(self, user_id: str, n: int = 10) -> list:
        if user_id not in self.user_idx:
            return []  # cold start problem
        
        idx = self.user_idx[user_id]
        similarities = self.user_similarities[idx]
        
        # Weighted average of ratings from similar users
        scores = np.dot(similarities, self.user_item_matrix)
        
        # Exclude items the user has already interacted with
        already_seen = set(k for k, v in 
                          zip(self.items, self.user_item_matrix[idx]) if v > 0)
        
        recommendations = [
            (self.items[i], scores[i])
            for i in np.argsort(-scores)
            if self.items[i] not in already_seen
        ]
        
        return recommendations[:n]

def compute_features(transaction: dict, user_history: dict) -> dict:
    return {
        # Velocity features
        "tx_count_1h": count_transactions(user_history, hours=1),
        "tx_count_24h": count_transactions(user_history, hours=24),
        "amount_sum_1h": sum_transactions(user_history, hours=1),
        
        # Deviation features
        "amount_vs_avg": transaction["amount"] / (user_history["avg_amount"] + 1),
        "is_new_merchant": transaction["merchant_id"] not in user_history["known_merchants"],
        "is_new_country": transaction["country"] != user_history["home_country"],
        
        # Time features
        "hour_of_day": datetime.fromtimestamp(transaction["ts"]).hour,
        "is_weekend": datetime.fromtimestamp(transaction["ts"]).weekday() >= 5,
        
        # Device/location features
        "device_seen_before": transaction["device_id"] in user_history["known_devices"],
        "ip_reputation_score": ip_reputation_service.score(transaction["ip"]),
        "distance_from_last_tx_km": haversine(
            transaction["location"], 
            user_history["last_location"]
        )
    }

Transaction arrives
  → Feature computation (< 10ms, from Redis feature store)
  → Rule engine (hard rules: block if X, always fast)
  → ML model (gradient boosting, returns fraud probability)
  → Decision:
      probability < 0.1 → ALLOW
      0.1 ≤ probability < 0.7 → STEP UP (require 2FA)
      probability ≥ 0.7 → BLOCK + alert
  → Log decision + features (for model retraining)

import time
from enum import Enum
from threading import Lock

class CircuitState(Enum):
    CLOSED = "closed"      # Normal operation
    OPEN = "open"          # Rejecting requests (failing fast)
    HALF_OPEN = "half_open" # Testing if service recovered

class CircuitBreaker:
    def __init__(self, failure_threshold=5, recovery_timeout=60, success_threshold=2):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.success_threshold = success_threshold
        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.success_count = 0
        self.last_failure_time = None
        self.lock = Lock()
    
    def call(self, func, *args, **kwargs):
        with self.lock:
            if self.state == CircuitState.OPEN:
                if time.time() - self.last_failure_time > self.recovery_timeout:
                    self.state = CircuitState.HALF_OPEN
                    self.success_count = 0
                else:
                    raise Exception("Circuit is OPEN — fast failing")
        
        try:
            result = func(*args, **kwargs)
            self._on_success()
            return result
        except Exception as e:
            self._on_failure()
            raise
    
    def _on_success(self):
        with self.lock:
            self.failure_count = 0
            if self.state == CircuitState.HALF_OPEN:
                self.success_count += 1
                if self.success_count >= self.success_threshold:
                    self.state = CircuitState.CLOSED
    
    def _on_failure(self):
        with self.lock:
            self.failure_count += 1
            self.last_failure_time = time.time()
            if self.failure_count >= self.failure_threshold:
                self.state = CircuitState.OPEN

Source Systems (Postgres, MySQL, APIs)
  → CDC (Change Data Capture) via Debezium → Kafka
  → OR: batch export → S3 (raw zone)

Transformation:
  → dbt (SQL-based transforms, runs in the warehouse)
  → OR: Spark (for complex Python/Scala transforms)

Storage:
  → S3 (raw) → S3 (transformed) → BigQuery / Snowflake / Redshift

Orchestration:
  → Airflow / Dagster (DAGs, scheduling, retries, monitoring)

def load_partition(date: str, source: str):
    """
    Idempotent: safe to run multiple times for same date+source.
    Uses REPLACE INTO / MERGE / truncate+reload semantics.
    """
    data = extract(source, date)
    transformed = transform(data)
    
    # Delete existing data for this partition before loading
    db.execute(f"DELETE FROM facts WHERE date = '{date}' AND source = '{source}'")
    
    # Load fresh data
    db.bulk_insert("facts", transformed)
    
    # Record successful run (for monitoring/auditing)
    db.execute(f"""
        INSERT INTO pipeline_runs (date, source, records, run_at)
        VALUES ('{date}', '{source}', {len(transformed)}, NOW())
        ON CONFLICT (date, source) DO UPDATE SET
            records = {len(transformed)},
            run_at = NOW()
    """)

# G-Counter: only supports increment, automatically merges across replicas
class GCounter:
    def __init__(self, node_id: str, num_nodes: int):
        self.node_id = node_id
        self.counts = [0] * num_nodes
        self.node_index = hash(node_id) % num_nodes
    
    def increment(self, amount: int = 1):
        self.counts[self.node_index] += amount
    
    def value(self) -> int:
        return sum(self.counts)
    
    def merge(self, other: 'GCounter'):
        # Take the maximum count at each position
        self.counts = [max(a, b) for a, b in zip(self.counts, other.counts)]

# This CRDT can be incremented on any node and will always converge
# to the correct total across all nodes, eventually.
# Perfect for: page view counts, like counts, inventory decrements.

import structlog
import time

logger = structlog.get_logger()

def handle_request(request_id: str, user_id: str, endpoint: str):
    start = time.time()
    try:
        result = process(request_id)
        logger.info(
            "request.completed",
            request_id=request_id,
            user_id=user_id,
            endpoint=endpoint,
            duration_ms=(time.time() - start) * 1000,
            status="success"
        )
        return result
    except Exception as e:
        logger.error(
            "request.failed",
            request_id=request_id,
            user_id=user_id,
            endpoint=endpoint,
            duration_ms=(time.time() - start) * 1000,
            error=str(e),
            status="error"
        )
        raise

from opentelemetry import trace
from opentelemetry.propagate import inject, extract

tracer = trace.get_tracer("my-service")

def call_downstream_service(headers: dict, payload: dict):
    with tracer.start_as_current_span("downstream-call") as span:
        span.set_attribute("http.url", DOWNSTREAM_URL)
        
        # Inject trace context into outgoing headers
        # This propagates the trace ID to the downstream service
        inject(headers)
        
        response = requests.post(DOWNSTREAM_URL, headers=headers, json=payload)
        span.set_attribute("http.status_code", response.status_code)
        
        return response

Code push to main
  → Trigger: webhook → CI/CD orchestrator
  → Pipeline:
      1. Checkout code (fast, ~5s)
      2. Build (Docker image, ~2-10 min)
      3. Unit + integration tests (parallel shards, ~5-15 min)
      4. Security scan (SAST, dependency audit)
      5. Push image to registry (tagged with commit SHA)
      6. Deploy to staging
      7. Smoke tests on staging
      8. Manual approval gate (for production)
      9. Rolling deploy to production:
           → Deploy to 5% of instances (canary)
           → Monitor error rate + latency for 10 min
           → If OK: roll out to 100%
           → If anomaly: automatic rollback

def canary_deploy(image_tag: str, canary_percentage: int = 5):
    # Route X% of traffic to new version
    update_load_balancer_weights(
        stable=100 - canary_percentage,
        canary=canary_percentage
    )
    
    # Monitor for 10 minutes
    baseline_error_rate = get_error_rate(version="stable", window_minutes=60)
    
    for minute in range(10):
        time.sleep(60)
        canary_error_rate = get_error_rate(version="canary", window_minutes=5)
        
        if canary_error_rate > baseline_error_rate * 1.5:
            # Error rate spiked — rollback immediately
            update_load_balancer_weights(stable=100, canary=0)
            alert(f"Canary rollback: error rate {canary_error_rate:.2%} vs baseline {baseline_error_rate:.2%}")
            return "ROLLED_BACK"
    
    # All good — full rollout
    update_load_balancer_weights(stable=0, canary=100)
    return "DEPLOYED"

class IdempotentConsumer:
    def __init__(self):
        self.processed_ids = redis  # persistent deduplication store
    
    def process(self, message_id: str, payload: dict):
        # Check if already processed
        if self.processed_ids.get(f"processed:{message_id}"):
            return  # skip duplicate
        
        # Process (your business logic here)
        result = do_business_logic(payload)
        
        # Atomic: mark as processed + apply result in same transaction
        with db.transaction():
            apply_result(result)
            self.processed_ids.setex(f"processed:{message_id}", 86400, "1")

# Shard by user_id (good: most queries are per-user)
def get_shard(user_id: int, num_shards: int = 16) -> int:
    return user_id % num_shards

# But what if you need to query across users? (e.g., "find all orders from Brazil")
# That requires querying ALL shards and merging results.
# This is the tradeoff: optimize for per-user queries vs. cross-user analytics.

import random
import time
import threading

class RaftNode:
    def __init__(self, node_id: str, peers: list):
        self.node_id = node_id
        self.peers = peers
        self.current_term = 0
        self.state = "follower"  # follower | candidate | leader
        self.voted_for = None
        self.votes_received = 0
        self.election_timeout = random.uniform(150, 300)  # ms; randomized to avoid split votes
        self.last_heartbeat = time.time()
    
    def start_election(self):
        self.state = "candidate"
        self.current_term += 1
        self.voted_for = self.node_id
        self.votes_received = 1  # vote for self
        
        # Send RequestVote to all peers
        for peer in self.peers:
            vote_granted = self.request_vote(peer, self.current_term)
            if vote_granted:
                self.votes_received += 1
        
        # If majority votes received → become leader
        if self.votes_received > (len(self.peers) + 1) / 2:
            self.become_leader()
    
    def become_leader(self):
        self.state = "leader"
        # Start sending heartbeats to prevent other nodes from starting elections
        self.send_heartbeats()

# BAD: N+1 queries
orders = db.query("SELECT * FROM orders WHERE user_id = ?", user_id)
for order in orders:
    # This issues 1 query per order!
    items = db.query("SELECT * FROM order_items WHERE order_id = ?", order.id)
    order.items = items
# Total: 1 + N queries

# GOOD: 2 queries total (or 1 with JOIN)
orders = db.query("SELECT * FROM orders WHERE user_id = ?", user_id)
order_ids = [o.id for o in orders]

# Fetch all items in one query
items = db.query("SELECT * FROM order_items WHERE order_id = ANY(?)", order_ids)

# Group items by order_id in Python (no DB roundtrip)
from collections import defaultdict
items_by_order = defaultdict(list)
for item in items:
    items_by_order[item.order_id].append(item)

for order in orders:
    order.items = items_by_order[order.id]

class FeatureFlagService:
    def __init__(self):
        self.flags = {}  # loaded from DB/config service
    
    def is_enabled(self, flag_name: str, user_id: str = None, context: dict = None) -> bool:
        flag = self.flags.get(flag_name)
        if not flag:
            return False
        
        # Global kill switch
        if not flag.get("enabled"):
            return False
        
        # Percentage rollout
        if "rollout_percentage" in flag and user_id:
            user_hash = int(hashlib.md5(f"{flag_name}:{user_id}".encode()).hexdigest(), 16)
            if (user_hash % 100) >= flag["rollout_percentage"]:
                return False
        
        # User allowlist
        if "allowed_users" in flag:
            if user_id in flag["allowed_users"]:
                return True
        
        # Attribute targeting (e.g., only for premium users)
        if "targeting" in flag and context:
            for rule in flag["targeting"]:
                if not evaluate_rule(rule, context):
                    return False
        
        return True

def get_user_recommendations(user_id: str) -> list:
    try:
        # Primary: personalized recommendations from ML service
        return recommendation_service.get(user_id, timeout=200)  # 200ms SLA
    except (TimeoutError, ServiceUnavailableError):
        try:
            # Fallback 1: cached recommendations from last successful call
            cached = redis.get(f"recs:{user_id}")
            if cached:
                return json.loads(cached)
        except Exception:
            pass
        
        try:
            # Fallback 2: trending items (doesn't require user data)
            return trending_service.get_trending(limit=10, timeout=100)
        except Exception:
            pass
        
        # Final fallback: hardcoded popular items (never fails)
        return DEFAULT_POPULAR_ITEMS

    # At no point do we show an error to the user for a non-critical feature

import redis
import json
import time
import uuid
from dataclasses import dataclass

@dataclass
class Task:
    task_id: str
    function: str
    args: list
    kwargs: dict
    priority: int = 5  # 1=highest, 10=lowest
    retry_count: int = 0
    max_retries: int = 3
    eta: float = None  # run_at timestamp (for scheduled tasks)

class TaskQueue:
    def __init__(self, redis_client):
        self.r = redis_client
    
    def enqueue(self, function: str, *args, priority: int = 5, delay_seconds: int = 0, **kwargs) -> str:
        task = Task(
            task_id=str(uuid.uuid4()),
            function=function,
            args=list(args),
            kwargs=kwargs,
            priority=priority,
            eta=time.time() + delay_seconds if delay_seconds else None
        )
        
        if task.eta:
            # Delayed task: add to sorted set with eta as score
            self.r.zadd("delayed_tasks", {json.dumps(task.__dict__): task.eta})
        else:
            # Immediate task: add to priority queue
            score = priority * 1e12 + time.time()  # lower priority = higher score
            self.r.zadd(f"queue:p{priority}", {json.dumps(task.__dict__): score})
        
        return task.task_id
    
    def dequeue(self) -> Task:
        # Check delayed tasks first (move ready ones to main queue)
        self._move_ready_delayed_tasks()
        
        # Pop highest-priority task
        for priority in range(1, 11):
            result = self.r.zpopmin(f"queue:p{priority}", count=1)
            if result:
                return Task(**json.loads(result[0][0]))
        return None
    
    def _move_ready_delayed_tasks(self):
        now = time.time()
        ready = self.r.zrangebyscore("delayed_tasks", 0, now)
        if ready:
            pipe = self.r.pipeline()
            for task_json in ready:
                task = Task(**json.loads(task_json))
                pipe.zrem("delayed_tasks", task_json)
                pipe.zadd(f"queue:p{task.priority}", {task_json: task.priority * 1e12 + now})
            pipe.execute()