design-expert

System Design Expert

Expert guidance for system design, software architecture, scalability patterns, and distributed systems.

Core Concepts

Architecture Patterns

• Microservices vs Monolithic
• Event-driven architecture
• CQRS and Event Sourcing
• Layered architecture
• Hexagonal architecture
• Service-oriented architecture (SOA)

Scalability

• Horizontal vs vertical scaling
• Load balancing strategies
• Caching layers
• Database sharding
• Read replicas
• CDN usage

Distributed Systems

• CAP theorem
• Consistency models
• Distributed consensus (Raft, Paxos)
• Message queues
• Service discovery
• Circuit breakers

Design Patterns

# Singleton Pattern
class DatabaseConnection:
    _instance = None
    _lock = threading.Lock()

    def __new__(cls):
        if cls._instance is None:
            with cls._lock:
                if cls._instance is None:
                    cls._instance = super().__new__(cls)
                    cls._instance._initialize()
        return cls._instance

    def _initialize(self):
        self.connection = self._create_connection()

# Factory Pattern
class ShapeFactory:
    @staticmethod
    def create_shape(shape_type: str):
        if shape_type == "circle":
            return Circle()
        elif shape_type == "square":
            return Square()
        raise ValueError(f"Unknown shape: {shape_type}")

# Observer Pattern
class Subject:
    def __init__(self):
        self._observers = []

    def attach(self, observer):
        self._observers.append(observer)

    def notify(self, event):
        for observer in self._observers:
            observer.update(event)

# Strategy Pattern
class PaymentStrategy:
    def pay(self, amount): pass

class CreditCardPayment(PaymentStrategy):
    def pay(self, amount):
        return f"Paid ${amount} via credit card"

class PayPalPayment(PaymentStrategy):
    def pay(self, amount):
        return f"Paid ${amount} via PayPal"

Scalability Patterns

# Circuit Breaker Pattern
from enum import Enum
import time

class CircuitState(Enum):
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"

class CircuitBreaker:
    def __init__(self, failure_threshold=5, timeout=60):
        self.failure_threshold = failure_threshold
        self.timeout = 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.timeout:
                self.state = CircuitState.HALF_OPEN
            else:
                raise Exception("Circuit breaker is OPEN")

        try:
            result = func(*args, **kwargs)
            self.on_success()
            return result
        except Exception as e:
            self.on_failure()
            raise e

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

    def on_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()

        if self.failure_count >= self.failure_threshold:
            self.state = CircuitState.OPEN

# Rate Limiter
from collections import deque
import time

class RateLimiter:
    def __init__(self, max_requests, window_seconds):
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self.requests = deque()

    def allow_request(self, user_id):
        now = time.time()

        # Remove old requests outside window
        while self.requests and self.requests[0][1] < now - self.window_seconds:
            self.requests.popleft()

        # Check if under limit
        user_requests = sum(1 for uid, _ in self.requests if uid == user_id)

        if user_requests < self.max_requests:
            self.requests.append((user_id, now))
            return True

        return False