System Design Series Part 5: Microservices Architecture

• Small team (< 10 developers)
• Early-stage startup exploring product-market fit
• Simple domain with clear boundaries
• Need rapid development without distributed complexity
• Limited DevOps expertise

# Monolith structure example
my_app/
+-- app/
¦   +-- models/
¦   ¦   +-- user.py
¦   ¦   +-- order.py
¦   ¦   +-- product.py
¦   +-- services/
¦   ¦   +-- user_service.py
¦   ¦   +-- order_service.py
¦   ¦   +-- product_service.py
¦   +-- api/
¦   ¦   +-- routes.py
¦   +-- database.py
+-- tests/
+-- requirements.txt

• Large organization with multiple teams
• Different parts of system have different scaling needs
• Teams need to deploy independently
• Complex domain with clear bounded contexts
• Strong DevOps/infrastructure capabilities

# Microservices structure example
platform/
+-- services/
¦   +-- user-service/
¦   ¦   +-- src/
¦   ¦   +-- Dockerfile
¦   ¦   +-- k8s/
¦   +-- order-service/
¦   ¦   +-- src/
¦   ¦   +-- Dockerfile
¦   ¦   +-- k8s/
¦   +-- payment-service/
¦   ¦   +-- src/
¦   ¦   +-- Dockerfile
¦   ¦   +-- k8s/
¦   +-- notification-service/
¦       +-- src/
¦       +-- Dockerfile
¦       +-- k8s/
+-- api-gateway/
+-- infrastructure/
+-- docker-compose.yml

# Bounded Contexts for E-Commerce Platform

# User Context - User identity and preferences
class UserService:
    """Handles user registration, authentication, profiles"""
    def register_user(self, email, password): pass
    def authenticate(self, email, password): pass
    def get_profile(self, user_id): pass

# Catalog Context - Product information
class CatalogService:
    """Manages product listings, categories, search"""
    def get_product(self, product_id): pass
    def search_products(self, query): pass
    def get_category(self, category_id): pass

# Order Context - Order lifecycle
class OrderService:
    """Handles order creation, status, history"""
    def create_order(self, user_id, items): pass
    def get_order_status(self, order_id): pass
    def cancel_order(self, order_id): pass

# Inventory Context - Stock management
class InventoryService:
    """Manages stock levels, reservations"""
    def check_availability(self, product_id, quantity): pass
    def reserve_stock(self, product_id, quantity): pass
    def release_stock(self, product_id, quantity): pass

# Payment Context - Financial transactions
class PaymentService:
    """Processes payments, refunds"""
    def process_payment(self, order_id, amount): pass
    def refund_payment(self, payment_id): pass

# Shipping Context - Fulfillment
class ShippingService:
    """Handles shipping labels, tracking"""
    def create_shipment(self, order_id): pass
    def get_tracking(self, shipment_id): pass

# Tactical DDD Building Blocks

# Entity: has identity, mutable state
class Order:
    def __init__(self, order_id, customer_id):
        self.order_id = order_id         # Identity
        self.customer_id = customer_id
        self.items = []
        self.status = "pending"
    
    def add_item(self, product_id, quantity, price):
        self.items.append(OrderItem(product_id, quantity, price))
    
    def confirm(self):
        if not self.items:
            raise ValueError("Cannot confirm empty order")
        self.status = "confirmed"

# Value Object: no identity, immutable, compared by attributes
class Money:
    def __init__(self, amount, currency):
        self.amount = amount
        self.currency = currency
    
    def __eq__(self, other):
        return self.amount == other.amount and self.currency == other.currency
    
    def add(self, other):
        assert self.currency == other.currency, "Currency mismatch"
        return Money(self.amount + other.amount, self.currency)

# Aggregate Root: Order is the root, OrderItem is only accessed through it
class OrderItem:
    """Never accessed directly — always through Order aggregate"""
    def __init__(self, product_id, quantity, price):
        self.product_id = product_id
        self.quantity = quantity
        self.price = price  # Money value object

# Domain Event: something that happened in the domain
class OrderConfirmed:
    def __init__(self, order_id, customer_id, total):
        self.order_id = order_id
        self.customer_id = customer_id
        self.total = total
        self.occurred_at = "2025-06-01T10:30:00Z"

print("Aggregate: Order (root) -> OrderItem (child)")
print("Value Object: Money (immutable, no identity)")
print("Domain Event: OrderConfirmed (published to other contexts)")

# Good: Order service owns all order data
class OrderService:
    def create_order(self, user_id, items):
        # All order logic contained within service
        order = Order(user_id=user_id)
        for item in items:
            order.add_item(item)
        order.calculate_total()
        self.db.save(order)
        
        # Emit event for other services (async)
        self.event_bus.publish("order.created", order)
        return order
    
    def get_order_details(self, order_id):
        # No external calls needed
        return self.db.get(order_id)

# Bad: Order service makes synchronous calls for every operation
class OrderService:
    def create_order(self, user_id, items):
        # Synchronous calls create tight coupling
        user = self.user_service.get_user(user_id)  # Network call
        
        for item in items:
            product = self.catalog_service.get_product(item.id)  # Network call
            price = self.pricing_service.get_price(item.id)  # Network call
            stock = self.inventory_service.check(item.id)  # Network call
        
        # Any service failure = order failure
        # Can't deploy independently
        # Latency compounds with each call

Single entry point that routes requests to appropriate services.

# API Gateway responsibilities
class APIGateway:
    def __init__(self):
        self.services = {
            "/users": "user-service:8080",
            "/orders": "order-service:8080",
            "/products": "catalog-service:8080"
        }
    
    def route_request(self, request):
        # 1. Authentication
        user = self.authenticate(request.headers.get("Authorization"))
        
        # 2. Rate limiting
        if self.rate_limiter.is_limited(user.id):
            return Response(status=429)
        
        # 3. Route to service
        service = self.get_service(request.path)
        
        # 4. Protocol translation (REST -> gRPC)
        response = service.forward(request)
        
        # 5. Response aggregation (if needed)
        return response
    
    def aggregate_product_page(self, product_id):
        """Combine data from multiple services"""
        product = self.catalog_service.get(product_id)
        reviews = self.review_service.get_for_product(product_id)
        inventory = self.inventory_service.check(product_id)
        
        return {
            **product,
            "reviews": reviews,
            "in_stock": inventory.available > 0
        }

Separate backend for each client type (web, mobile, IoT).

# BFF Pattern
# Each client gets optimized API

# Mobile BFF - Minimal data, pagination
class MobileBFF:
    def get_product_list(self, page=1, limit=20):
        products = self.catalog.get_products(page, limit)
        return [{
            "id": p.id,
            "name": p.name,
            "price": p.price,
            "thumbnail": p.images[0].url  # Only first image
        } for p in products]

# Web BFF - Rich data, full details
class WebBFF:
    def get_product_list(self):
        products = self.catalog.get_products()
        for product in products:
            product["reviews_summary"] = self.reviews.get_summary(product.id)
            product["availability"] = self.inventory.check(product.id)
            product["related"] = self.recommendations.get_related(product.id)
        return products

Each service owns its data store—no shared databases.

# Each service manages its own database
# Order Service - PostgreSQL for transactions
order_db = PostgreSQL("order-db")

# Catalog Service - MongoDB for flexible product data
catalog_db = MongoDB("catalog-db")

# Search Service - Elasticsearch for full-text search
search_db = Elasticsearch("search-cluster")

# Session Service - Redis for fast access
session_db = Redis("session-cache")

# Analytics Service - ClickHouse for time-series
analytics_db = ClickHouse("analytics-cluster")

Challenge: Cross-service queries require careful design (event sourcing, CQRS, or API composition).

Manage distributed transactions across services using compensating transactions.

# Saga Pattern for Order Creation
class OrderSaga:
    def execute(self, order_data):
        saga_log = []
        
        try:
            # Step 1: Reserve inventory
            reservation = self.inventory.reserve(order_data.items)
            saga_log.append(("inventory", reservation.id))
            
            # Step 2: Process payment
            payment = self.payment.charge(order_data.user_id, order_data.total)
            saga_log.append(("payment", payment.id))
            
            # Step 3: Create order
            order = self.orders.create(order_data)
            saga_log.append(("order", order.id))
            
            # Step 4: Notify user
            self.notifications.send(order_data.user_id, "Order confirmed!")
            
            return order
            
        except Exception as e:
            # Compensating transactions (rollback)
            self.compensate(saga_log)
            raise e
    
    def compensate(self, saga_log):
        """Undo completed steps in reverse order"""
        for service, resource_id in reversed(saga_log):
            if service == "order":
                self.orders.cancel(resource_id)
            elif service == "payment":
                self.payment.refund(resource_id)
            elif service == "inventory":
                self.inventory.release(resource_id)

Prevent cascading failures by stopping calls to failing services.

# Circuit Breaker Implementation
from enum import Enum
import time

class CircuitState(Enum):
    CLOSED = "closed"      # Normal operation
    OPEN = "open"          # Failing, reject calls
    HALF_OPEN = "half_open"  # Testing recovery

class CircuitBreaker:
    def __init__(self, failure_threshold=5, recovery_timeout=30):
        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 self._should_attempt_reset():
                self.state = CircuitState.HALF_OPEN
            else:
                raise CircuitOpenError("Circuit 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
    
    def _should_attempt_reset(self):
        return (time.time() - self.last_failure_time) > self.recovery_timeout

# Usage
payment_circuit = CircuitBreaker(failure_threshold=5, recovery_timeout=30)
result = payment_circuit.call(payment_service.charge, user_id, amount)

# Istio VirtualService for traffic routing
apiVersion: networking.istio.io/v1beta1
kind: VirtualService
metadata:
  name: order-service
spec:
  hosts:
    - order-service
  http:
    - match:
        - headers:
            x-canary:
              exact: "true"
      route:
        - destination:
            host: order-service
            subset: v2
          weight: 100
    - route:
        - destination:
            host: order-service
            subset: v1
          weight: 90
        - destination:
            host: order-service
            subset: v2
          weight: 10  # 10% canary traffic

---
# Istio DestinationRule for circuit breaking
apiVersion: networking.istio.io/v1beta1
kind: DestinationRule
metadata:
  name: order-service
spec:
  host: order-service
  trafficPolicy:
    connectionPool:
      tcp:
        maxConnections: 100
      http:
        h2UpgradePolicy: UPGRADE
        http1MaxPendingRequests: 100
    outlierDetection:
      consecutive5xxErrors: 5
      interval: 30s
      baseEjectionTime: 30s

# API Composition — aggregate data from multiple services
import asyncio

class OrderDetailsComposer:
    """
    Composes a rich order view by querying multiple services
    and joining results in memory.
    """
    def __init__(self, order_svc, user_svc, product_svc):
        self.order_svc = order_svc
        self.user_svc = user_svc
        self.product_svc = product_svc
    
    async def get_order_details(self, order_id):
        # Fetch order first (owns the core data)
        order = await self.order_svc.get(order_id)
        
        # Parallel calls to other services
        user, products = await asyncio.gather(
            self.user_svc.get(order["user_id"]),
            self.product_svc.get_batch(
                [item["product_id"] for item in order["items"]]
            )
        )
        
        # In-memory join
        product_map = {p["id"]: p for p in products}
        return {
            "order_id": order["id"],
            "status": order["status"],
            "customer": {
                "name": user["name"],
                "email": user["email"]
            },
            "items": [
                {
                    **item,
                    "product_name": product_map[item["product_id"]]["name"],
                    "image_url": product_map[item["product_id"]]["image"]
                }
                for item in order["items"]
            ],
            "total": order["total"]
        }

# Typically implemented in the API Gateway or a BFF
print("API Composition: query N services, join in memory")
print("Best for: read-heavy queries spanning 2-4 services")

Write events to an outbox table in the same database transaction as the business data. A separate process reads the outbox and publishes events to the message broker.

# Transactional Outbox Pattern
import json
import uuid

class OrderService:
    def create_order(self, user_id, items, total):
        # Single database transaction
        with self.db.transaction() as tx:
            # 1. Write business data
            order_id = str(uuid.uuid4())
            tx.execute(
                "INSERT INTO orders (id, user_id, total, status) "
                "VALUES (%s, %s, %s, 'created')",
                (order_id, user_id, total)
            )
            for item in items:
                tx.execute(
                    "INSERT INTO order_items (order_id, product_id, qty, price) "
                    "VALUES (%s, %s, %s, %s)",
                    (order_id, item["product_id"], item["qty"], item["price"])
                )
            
            # 2. Write event to outbox (SAME transaction)
            tx.execute(
                "INSERT INTO outbox (id, aggregate_type, aggregate_id, "
                "event_type, payload, created_at) "
                "VALUES (%s, 'Order', %s, 'OrderCreated', %s, NOW())",
                (str(uuid.uuid4()), order_id, json.dumps({
                    "order_id": order_id,
                    "user_id": user_id,
                    "total": total,
                    "items": items
                }))
            )
            # Both writes commit or both roll back — atomicity guaranteed

# Separate process: OutboxRelay reads outbox, publishes to Kafka/RabbitMQ
class OutboxRelay:
    """Polls the outbox table and publishes events to the broker."""
    def relay(self):
        rows = self.db.query(
            "SELECT * FROM outbox WHERE published = FALSE "
            "ORDER BY created_at LIMIT 100"
        )
        for row in rows:
            self.broker.publish(row["event_type"], row["payload"])
            self.db.execute(
                "UPDATE outbox SET published = TRUE WHERE id = %s",
                (row["id"],)
            )

print("Outbox guarantees: DB write + event publish are atomic")
print("Relay methods: polling, CDC (Change Data Capture), log tailing")

In distributed systems, messages can be delivered more than once (at-least-once delivery). An idempotent consumer ensures that processing the same message multiple times produces the same result.

# Idempotent Consumer Pattern
class PaymentEventHandler:
    def __init__(self, db, payment_service):
        self.db = db
        self.payment_service = payment_service
    
    def handle_order_created(self, event):
        message_id = event["message_id"]
        
        # Check if we've already processed this message
        if self.db.exists("processed_messages", message_id):
            print(f"Duplicate message {message_id} — skipping")
            return  # Idempotent: no side effects
        
        # Process the message
        self.payment_service.charge(
            user_id=event["user_id"],
            amount=event["total"],
            order_id=event["order_id"]
        )
        
        # Record that we processed this message
        self.db.insert("processed_messages", {
            "message_id": message_id,
            "processed_at": "2026-03-26T10:00:00Z",
            "event_type": "OrderCreated"
        })

print("Idempotency key: message_id (UUID assigned by producer)")
print("Storage: processed_messages table with TTL cleanup")

# Service Discovery — Client-Side (e.g., with Consul)
import random

class ServiceRegistry:
    """In production, this would be Consul, Eureka, or etcd."""
    def __init__(self):
        self.services = {}
    
    def register(self, service_name, instance_id, host, port):
        self.services.setdefault(service_name, []).append({
            "instance_id": instance_id,
            "host": host,
            "port": port,
            "healthy": True
        })
    
    def deregister(self, service_name, instance_id):
        self.services[service_name] = [
            i for i in self.services[service_name]
            if i["instance_id"] != instance_id
        ]
    
    def get_instances(self, service_name):
        instances = self.services.get(service_name, [])
        return [i for i in instances if i["healthy"]]

class ClientSideLoadBalancer:
    """Client picks an instance from the registry."""
    def __init__(self, registry):
        self.registry = registry
    
    def call(self, service_name, request):
        instances = self.registry.get_instances(service_name)
        if not instances:
            raise Exception(f"No healthy instances for {service_name}")
        
        # Round-robin, random, or weighted selection
        instance = random.choice(instances)
        url = f"http://{instance['host']}:{instance['port']}"
        print(f"Routing to {service_name} at {url}")
        return url

# Kubernetes simplifies this with built-in DNS
# kubectl: order-service.default.svc.cluster.local
registry = ServiceRegistry()
registry.register("order-service", "order-1", "10.0.1.5", 8080)
registry.register("order-service", "order-2", "10.0.1.6", 8080)
registry.register("order-service", "order-3", "10.0.1.7", 8080)

lb = ClientSideLoadBalancer(registry)
lb.call("order-service", "/api/orders/123")

# Consumer-Driven Contract Test with Pact
# Written by the ORDER SERVICE team (consumer)
# Tests what it expects from the USER SERVICE (provider)

# Step 1: Consumer defines expected interactions
def test_get_user_contract():
    """Order service expects user-service to return user by ID."""
    pact = Pact(consumer="order-service", provider="user-service")
    
    # Define expected interaction
    pact.given("user 123 exists").upon_receiving(
        "a request for user 123"
    ).with_request(
        method="GET", path="/api/users/123"
    ).will_respond_with(
        status=200,
        body={
            "id": "123",
            "name": Like("Jane Doe"),      # Any string
            "email": Like("jane@test.com")  # Any string
        }
    )
    
    # Run test against mock provider
    with pact:
        user = order_service_client.get_user("123")
        assert user["id"] == "123"
        assert "name" in user
        assert "email" in user
    
    # Pact file generated → shared with provider team
    # Provider runs this contract in their CI pipeline
    print("Contract published to Pact Broker")
    print("Provider verifies contract on every build")

// Webpack Module Federation — micro-frontend setup
// Each team deploys independently, modules loaded at runtime

// Team A: Product Catalog (host application)
// webpack.config.js
module.exports = {
  plugins: [
    new ModuleFederationPlugin({
      name: "catalog_app",
      remotes: {
        // Other teams' micro-frontends loaded at runtime
        reviews: "reviews_app@https://reviews.cdn.example.com/remoteEntry.js",
        cart: "cart_app@https://cart.cdn.example.com/remoteEntry.js",
        recommendations: "recs_app@https://recs.cdn.example.com/remoteEntry.js"
      },
      shared: ["react", "react-dom"]  // Shared dependencies
    })
  ]
};

// In the host app — compose micro-frontends
// ProductPage.jsx
const ReviewsWidget = React.lazy(
  () => import("reviews/ReviewsWidget")
);
const CartButton = React.lazy(
  () => import("cart/AddToCartButton")
);
const Recommendations = React.lazy(
  () => import("recommendations/RelatedProducts")
);

function ProductPage({ productId }) {
  return (
    <div>
      <ProductDetails id={productId} />
      <Suspense fallback="Loading reviews...">
        <ReviewsWidget productId={productId} />
      </Suspense>
      <Suspense fallback="Loading cart...">
        <CartButton productId={productId} />
      </Suspense>
      <Suspense fallback="Loading recommendations...">
        <Recommendations productId={productId} />
      </Suspense>
    </div>
  );
}

console.log("Each widget deployed independently by different teams");

# AWS Lambda Function - Process uploaded images
import boto3
import json
from PIL import Image
import io

s3 = boto3.client('s3')

def lambda_handler(event, context):
    """Triggered when image uploaded to S3"""
    # Get uploaded file info
    bucket = event['Records'][0]['s3']['bucket']['name']
    key = event['Records'][0]['s3']['object']['key']
    
    # Download image
    response = s3.get_object(Bucket=bucket, Key=key)
    image_content = response['Body'].read()
    
    # Process image (resize)
    image = Image.open(io.BytesIO(image_content))
    thumbnail = image.resize((200, 200))
    
    # Save thumbnail
    buffer = io.BytesIO()
    thumbnail.save(buffer, format='JPEG')
    buffer.seek(0)
    
    thumbnail_key = f"thumbnails/{key}"
    s3.put_object(
        Bucket=bucket,
        Key=thumbnail_key,
        Body=buffer.getvalue()
    )
    
    return {
        'statusCode': 200,
        'body': json.dumps({'thumbnail': thumbnail_key})
    }

Functions triggered by events from various sources:

# serverless.yml - Event triggers
service: order-processor

functions:
  processOrder:
    handler: handler.process_order
    events:
      - sqs:
          arn: arn:aws:sqs:region:account:order-queue
          batchSize: 10
  
  sendNotification:
    handler: handler.send_notification
    events:
      - sns:
          arn: arn:aws:sns:region:account:order-notifications
  
  apiEndpoint:
    handler: handler.api
    events:
      - http:
          path: /orders
          method: post
  
  scheduledReport:
    handler: handler.generate_report
    events:
      - schedule: rate(1 day)

Chain functions using Step Functions or event choreography:

# AWS Step Functions - Order workflow
{
  "Comment": "Order Processing Workflow",
  "StartAt": "ValidateOrder",
  "States": {
    "ValidateOrder": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:...:validateOrder",
      "Next": "CheckInventory"
    },
    "CheckInventory": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:...:checkInventory",
      "Next": "ProcessPayment",
      "Catch": [
        {
          "ErrorEquals": ["OutOfStockError"],
          "Next": "NotifyOutOfStock"
        }
      ]
    },
    "ProcessPayment": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:...:processPayment",
      "Next": "FulfillOrder"
    },
    "FulfillOrder": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:...:fulfillOrder",
      "End": true
    },
    "NotifyOutOfStock": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:...:notifyUser",
      "End": true
    }
  }
}

# Strangler Fig Pattern — incremental monolith decomposition

# Step 1: Intercept requests at the gateway
class StranglerProxy:
    def __init__(self):
        self.migrated_routes = {
            "/api/users": "http://user-service:8080",
            "/api/auth":  "http://auth-service:8080",
        }
        self.legacy_url = "http://monolith:3000"
    
    def route(self, request):
        for prefix, target in self.migrated_routes.items():
            if request.path.startswith(prefix):
                return forward_to(target, request)
        # Not yet migrated — send to monolith
        return forward_to(self.legacy_url, request)

# Step 2: Verify parity with shadow traffic
class ShadowTrafficVerifier:
    def compare(self, request):
        legacy_response = call_legacy(request)
        new_response = call_new_service(request)
        if legacy_response != new_response:
            log_discrepancy(request, legacy_response, new_response)
        return legacy_response  # Serve legacy until verified

# Step 3: Cutover — update routing, decommission legacy module
proxy = StranglerProxy()
print("Migrated routes:", list(proxy.migrated_routes.keys()))
print("Remaining legacy:", proxy.legacy_url)