Architecture Patterns: Saga Compensation Rate Limits

# io.thecodeforge — System Design tutorial

# Scenario: E-commerce order service.
# This is the monolith version. Notice what you get for FREE
# that microservices make you rebuild from scratch.

from dataclasses import dataclass
from decimal import Decimal
from typing import Optional
import sqlite3

# Single database connection — one transaction wraps everything.
# In a distributed system, this atomicity is GONE unless you implement
# two-phase commit or saga patterns. Both are painful.
DB_PATH = "orders.db"

@dataclass
class OrderItem:
    product_id: str
    quantity: int
    unit_price: Decimal

@dataclass
class Order:
    order_id: str
    customer_id: str
    items: list[OrderItem]
    status: str = "pending"

class InventoryService:
    """In a monolith, this is just a class. In microservices, it's a network call
    that can timeout, return stale data, or be temporarily unavailable."""

    def __init__(self, conn: sqlite3.Connection):
        self.conn = conn

    def reserve_stock(self, product_id: str, quantity: int) -> bool:
        cursor = self.conn.cursor()
        # This SELECT ... FOR UPDATE equivalent in SQLite — serialised write.
        # In a distributed inventory service, this becomes a distributed lock.
        # Redis SETNX, or a dedicated locking service. Both add latency and failure modes.
        cursor.execute(
            "SELECT stock_count FROM inventory WHERE product_id = ?",
            (product_id,)
        )
        row = cursor.fetchone()
        if not row or row[0] < quantity:
            return False  # Not enough stock — fail the whole operation cleanly

        cursor.execute(
            "UPDATE inventory SET stock_count = stock_count - ? WHERE product_id = ?",
            (quantity, product_id)
        )
        return True

class PaymentService:
    """Again — a class. Zero network calls. Zero timeout handling needed.
    The moment this becomes a separate service, you need circuit breakers,
    retry logic, idempotency keys, and dead-letter queues. All of that
    is real engineering effort — not free."""

    def __init__(self, conn: sqlite3.Connection):
        self.conn = conn

    def charge_customer(self, customer_id: str, amount: Decimal) -> bool:
        cursor = self.conn.cursor()
        cursor.execute(
            "SELECT balance FROM accounts WHERE customer_id = ?",
            (customer_id,)
        )
        row = cursor.fetchone()
        if not row or row[0] < amount:
            return False  # Insufficient funds — atomic rollback handles cleanup

        cursor.execute(
            "UPDATE accounts SET balance = balance - ? WHERE customer_id = ?",
            (amount, customer_id)
        )
        return True

class OrderOrchestrator:
    """The core of the monolith advantage: ONE database transaction covers
    inventory reservation, payment, AND order creation. If payment fails,
    inventory is automatically unreserved. No compensation logic. No sagas.
    No 'sorry, your order is stuck in PENDING forever' bugs."""

    def __init__(self):
        self.conn = sqlite3.connect(DB_PATH)
        self.inventory = InventoryService(self.conn)
        self.payment = PaymentService(self.conn)

    def place_order(self, order: Order) -> dict:
        total = sum(
            item.unit_price * item.quantity for item in order.items
        )

        try:
            # BEGIN TRANSACTION — implicit in SQLite when autocommit is off
            self.conn.execute("BEGIN")

            # Step 1: Reserve all stock items.
            # If ANY item fails, we roll back everything. One line of code.
            for item in order.items:
                if not self.inventory.reserve_stock(item.product_id, item.quantity):
                    self.conn.rollback()
                    return {
                        "success": False,
                        "reason": f"Insufficient stock for product {item.product_id}"
                    }

            # Step 2: Charge the customer.
            # In a microservices world, payment already happened in a different
            # service. If inventory reservation then fails, you're issuing refunds.
            # That's a SUPPORT TICKET. Here? It's a rollback.
            if not self.payment.charge_customer(order.customer_id, total):
                self.conn.rollback()
                return {"success": False, "reason": "Payment failed"}

            # Step 3: Persist the order record.
            self.conn.execute(
                "INSERT INTO orders (order_id, customer_id, status, total) VALUES (?, ?, ?, ?)",
                (order.order_id, order.customer_id, "confirmed", str(total))
            )

            self.conn.commit()  # All three operations committed atomically
            return {"success": True, "order_id": order.order_id, "total": str(total)}

        except Exception as e:
            self.conn.rollback()  # Something unexpected? Clean slate.
            raise RuntimeError(f"Order placement failed: {e}") from e


# --- Demonstrate the flow ---
if __name__ == "__main__":
    # Normally you'd have migrations. Simplified for illustration.
    orchestrator = OrderOrchestrator()

    order = Order(
        order_id="ORD-20240315-001",
        customer_id="CUST-789",
        items=[
            OrderItem(product_id="SKU-HEADPHONES-XZ3", quantity=1, unit_price=Decimal("149.99")),
            OrderItem(product_id="SKU-USB-CABLE-C", quantity=2, unit_price=Decimal("12.50"))
        ]
    )

    result = orchestrator.place_order(order)
    print(f"Order result: {result}")

# io.thecodeforge — System Design tutorial

# Scenario: Post-checkout event pipeline.
# The payment is already confirmed. Now we need to notify inventory,
# fraud, shipping, and email — all independently, all without
# the checkout service caring about any of them.

import json
import time
import threading
from dataclasses import dataclass, asdict
from typing import Callable
from collections import defaultdict, deque
from datetime import datetime

@dataclass
class OrderConfirmedEvent:
    event_id: str
    order_id: str
    customer_id: str
    product_ids: list[str]
    total_amount: float
    occurred_at: str  # ISO 8601 — always timestamp your events at creation time

    def to_json(self) -> str:
        return json.dumps(asdict(self))


class InMemoryEventBus:
    """
    Production equivalent: Apache Kafka, AWS SQS+SNS, or RabbitMQ.
    The contract is the same: publish once, consume independently.
    This in-memory version makes the pattern visible without Kafka setup.
    """

    def __init__(self):
        # Each topic maps to a list of independent consumer queues.
        # In Kafka terms: each consumer GROUP gets its own queue.
        # This is what enables independent consumption and replay.
        self._topics: dict[str, list[deque]] = defaultdict(list)
        self._handlers: dict[str, list[Callable]] = defaultdict(list)
        self._lock = threading.Lock()

    def subscribe(self, topic: str, handler: Callable) -> None:
"""Busy-polls the queue. In production Kafka consumers use long-polling
with configurable fetch.min.bytes and fetch.max.wait.ms for efficiency."""
        while True:
            if queue:
                event = queue.popleft()
                try:
                    handler(event)
                except Exception as e:
                    # In production: route to dead-letter queue, alert, do NOT silently swallow.
                    # Silent swallowing here is how you get the 'orders piling up unprocessed' nightmare.
                    print(f"[EventBus] CONSUMER ERROR — routing to DLQ: {e}")
            else:
                time.sleep(0.01)  # Back off when idle — don't spin-burn CPU


# --- CONSUMERS ---
# Each of these would be a separate service in production.
# They share zero state. They don't know about each other.

class InventoryConsumer:
    def handle(self, event: dict) -> None:
        print(f"[Inventory] Reserving stock for order {event['order_id']} "
              f"— products: {event['product_ids']}")
        # In reality: update stock counts, trigger reorder if threshold hit

class FraudDetectionConsumer:
    def handle(self, event: dict) -> None:
        print(f"[Fraud] Running fraud score for customer {event['customer_id']} "
              f"— amount: ${event['total_amount']}")
        # In reality: call ML model, flag order if score > threshold, publish FraudFlaggedEvent

class ShippingConsumer:
    def handle(self, event: dict) -> None:
        print(f"[Shipping] Creating shipment manifest for order {event['order_id']}")
        # In reality: call 3PL API, generate label, store tracking number

class EmailNotificationConsumer:
    def handle(self, event: dict) -> None:
        print(f"[Email] Queueing confirmation email to customer {event['customer_id']} "
              f"for order {event['order_id']}")
        # In reality: render template, call SES/SendGrid, record sent timestamp


# --- PRODUCER ---

class CheckoutService:
    """The checkout service knows about the event bus and the event schema.
    It does NOT know about inventory, fraud, shipping, or email.
    Adding a new downstream consumer requires ZERO changes here."""

    def __init__(self, event_bus: InMemoryEventBus):
        self.event_bus = event_bus

    def complete_checkout(self, queue: deque, order_id: str, customer_id: str, product_ids: list[str], total: float) -> dict:
        # Payment authorisation would happen here synchronously BEFORE this point.
        # EDA handles post-payment side effects — not the payment itself.
        event = OrderConfirmedEvent(
            event_id=f"evt-{order_id}-{int(time.time())}",
            order_id=order_id,
            customer_id=customer_id,
            product_ids=product_ids,
            total_amount=total,
            occurred_at=datetime.utcnow().isoformat() + "Z"
        )

        self.event_bus.publish("order.confirmed", asdict(event))

        # Returns IMMEDIATELY — doesn't wait for inventory, fraud, or shipping.
        # This is your sub-100ms checkout response time.
        return {"status": "confirmed", "order_id": order_id}

# io.thecodeforge — System Design tutorial

# Scenario: Order processing saga using an orchestrator.
# The orchestrator coordinates three microservices: Inventory, Payment, Shipping.
# If any step fails, it runs compensations for all completed steps.

from dataclasses import dataclass, asdict
from typing import Optional, Callable
from enum import Enum
import json
import time

class SagaStepStatus(Enum):
    PENDING = "pending"
    COMPLETED = "completed"
    FAILED = "failed"
    COMPENSATED = "compensated"

@dataclass
class SagaStep:
    name: str
    action: Callable
    compensate: Callable
    status: SagaStepStatus = SagaStepStatus.PENDING
    output: Optional[dict] = None
    error: Optional[str] = None

@dataclass
class SagaContext:
    order_id: str
    customer_id: str
    products: list[dict]
    payment_amount: float
    # Store outputs of each step for potential compensation
    inventory_reservation_id: Optional[str] = None
    payment_transaction_id: Optional[str] = None
    shipping_label_id: Optional[str] = None


class SagaOrchestrator:
    """Central coordinator for the order processing saga.
    In production, this would be a stateful service using a database or event store
    to persist saga state so it can survive restarts."""

    def __init__(self):
        self._steps: list[SagaStep] = []
        self._context: Optional[SagaContext] = None

    def add_step(self, name: str, action: Callable, compensate: Callable) -> None:
        self._steps.append(SagaStep(name=name, action=action, compensate=compensate))

    def execute(self, context: SagaContext) -> dict:
        self._context = context
        completed_steps = []

        print(f"[Saga] Starting saga for order {context.order_id}")

        try:
            for step in self._steps:
                print(f"[Saga] Executing step: {step.name}")
                # Call the action with context; action returns a dict with output
                result = step.action(self._context)
                step.status = SagaStepStatus.COMPLETED
                step.output = result
                completed_steps.append(step)
                print(f"[Saga] Step {step.name} completed: {result}")

            print(f"[Saga] All steps succeeded for order {context.order_id}")
            return {"status": "success", "order_id": context.order_id}

        except Exception as e:
            print(f"[Saga] Step failed: {e}")
            # Compensate completed steps in reverse order
            for step in reversed(completed_steps):
                print(f"[Saga] Compensating step: {step.name}")
                try:
                    step.compensate(self._context, step.output)
                    step.status = SagaStepStatus.COMPENSATED
                except Exception as comp_err:
                    # Non-recoverable compensation failure—requires manual intervention
                    print(f"[Saga] COMPENSATION FAILED for step {step.name}: {comp_err}")
                    # In production: publish to DLQ, alert ops
                    step.error = str(comp_err)
            return {"status": "failed", "order_id": context.order_id, "error": str(e)}


# --- Dummy microservice actions (simulate actual service calls) ---

def reserve_inventory(ctx: SagaContext) -> dict:
    """Call inventory service. Can raise exception on failure."""
    print(f"[Inventory] Reserving stock for order {ctx.order_id}")
    # Simulate success
    ctx.inventory_reservation_id = "INV-RES-12345"
    return {"reservation_id": ctx.inventory_reservation_id}

def compensate_inventory(ctx: SagaContext, step_output: dict) -> None:
    """Cancel inventory reservation."""
    print(f"[Inventory] Compensation: releasing reservation {step_output['reservation_id']}")
    # Simulate success

def process_payment(ctx: SagaContext) -> dict:
    """Call payment gateway. Can raise exception."""
    print(f"[Payment] Charging ${ctx.payment_amount} for order {ctx.order_id}")
    # Simulate success
    ctx.payment_transaction_id = "PAY-TXN-67890"
    return {"transaction_id": ctx.payment_transaction_id}

def compensate_payment(ctx: SagaContext, step_output: dict) -> None:
    """Issue refund."""
    print(f"[Payment] Compensation: refunding transaction {step_output['transaction_id']}")
    # Simulate success

def create_shipment(ctx: SagaContext) -> dict:
    """Create shipment with shipping provider."""
    print(f"[Shipping] Creating shipment for order {ctx.order_id}")
    # Simulate success
    ctx.shipping_label_id = "SHP-LBL-11111"
    return {"label_id": ctx.shipping_label_id}

def compensate_shipment(ctx: SagaContext, step_output: dict) -> None:
    """Cancel shipment."""
    print(f"[Shipping] Compensating: cancelling label {step_output['label_id']}")
    # Simulate success


# --- Wire it up ---

if __name__ == "__main__":
    orchestrator = SagaOrchestrator()

    # Define saga steps in order
    orchestrator.add_step("Reserve Inventory", reserve_inventory, compensate_inventory)
    orchestrator.add_step("Process Payment", process_payment, compensate_payment)
    orchestrator.add_step("Create Shipment", create_shipment, compensate_shipment)

    ctx = SagaContext(
        order_id="ORD-20241105-089",
        customer_id="CUST-321",
        products=[{"sku": "SKU-LAPTOP", "qty": 1}],
        payment_amount=1299.99
    )

    result = orchestrator.execute(ctx)
    print(f"\n[Saga] Final result: {json.dumps(result, indent=2)}")

# io.thecodeforge — System Design tutorial

# Scenario: Product catalogue with very different read/write patterns.
# Write side: admin updates one product at a time, needs validation + atomicity.
# Read side: 10,000 RPS product page loads, need sub-10ms response.
# A single DB model with indexes for both is a bottleneck in both directions.

from dataclasses import dataclass, field
from typing import Optional
import json
import time

# ─── WRITE SIDE (Command Model) ────────────────────────────────────────────
# Normalised. Strict validation. Transactional. Think PostgreSQL.

@dataclass
class ProductWriteModel:
    """The authoritative source of truth. Commands land here.
    This model is optimised for integrity, not query speed."""
    product_id: str
    sku: str
    name: str
    description: str
    price_cents: int       # Store money as integers — never floats. Ever.
    stock_count: int
    category_id: str
    brand_id: str
    is_active: bool = True
    version: int = 1       # Optimistic locking — detect concurrent modification

@dataclass
class UpdateProductPriceCommand:
    product_id: str
    new_price_cents: int
    updated_by: str        # Always audit who changed what in a write model
    reason: str            # Force callers to explain why — reduces lazy changes

@dataclass
class ProductPriceUpdatedEvent:
    product_id: str
    old_price_cents: int
    new_price_cents: int
    occurred_at: float


class ProductCommandHandler:
    """Handles all writes. Returns an event that downstream projections consume
    to update read models. The command handler does NOT update read models directly —
    that's the projection layer's job."""

    def __init__(self, write_store: dict):
        # In production: PostgreSQL with row-level locking
        self.write_store = write_store
        self.event_log: list[ProductPriceUpdatedEvent] = []

    def handle(self, command: UpdateProductPriceCommand) -> ProductPriceUpdatedEvent:
        product = self.write_store.get(command.product_id)
        if not product:
            raise ValueError(f"Product {command.product_id} not found")
        old_price = product.price_cents
        product.price_cents = command.new_price_cents
        product.version += 1
        # Store updated product (in production: UPDATE with WHERE version = old_version)
        self.write_store[command.product_id] = product

        event = ProductPriceUpdatedEvent(
            product_id=command.product_id,
            old_price_cents=old_price,
            new_price_cents=command.new_price_cents,
            occurred_at=time.time()
        )
        self.event_log.append(event)
        return event


# ─── READ SIDE (Projection / Query Model) ──────────────────────────────────
# Denormalised. Pre-joined. Optimised for fast lookups. Think Elasticsearch or Redis.

class ProductReadProjection:
    """Consumes events from the command side and builds a denormalised read model.
    This is eventually consistent. In production, this would run as a separate
    service or job that reads from an event queue."""

    def __init__(self):
        # In-memory dictionary simulating a fast key-value store or search index
        self._store: dict[str, dict] = {}

    def apply(self, event: ProductPriceUpdatedEvent):
        """Update the read model based on the event.
        Idempotent: applying the same event twice should yield same result."""
        # In production: check event idempotency (e.g., event_id dedup)
        if event.product_id not in self._store:
            # Initialise with default fields; in reality, you'd join with other data
            self._store[event.product_id] = {
                "product_id": event.product_id,
                "current_price_cents": event.new_price_cents,
                "last_updated": event.occurred_at
            }
        else:
            self._store[event.product_id]["current_price_cents"] = event.new_price_cents
            self._store[event.product_id]["last_updated"] = event.occurred_at

    def get_product(self, product_id: str) -> Optional[dict]:
        """Query side — instant, no joins, no locks."""
        return self._store.get(product_id)

    def get_all_prices(self) -> dict:
        """Bulk read: returns all current prices. Useful for caching."""
        return {pid: data["current_price_cents"] for pid, data in self._store.items()}


# ─── WIRING ─────────────────────────────────────────────────────────────────
# In a real system, the event would be published to a message broker,
# and the projection would consume asynchronously. Here we do it inline for demo.

if __name__ == "__main__":
    # Seed write store
    write_store = {
        "PROD-001": ProductWriteModel(
            product_id="PROD-001", sku="SKU-123", name="Widget",
            description="A widget", price_cents=1999, stock_count=100,
            category_id="CAT-1", brand_id="BRAND-A"
        )
    }

    command_handler = ProductCommandHandler(write_store)
    projection = ProductReadProjection()

    # Simulate a command
    cmd = UpdateProductPriceCommand(
        product_id="PROD-001",
        new_price_cents=1499,
        updated_by="admin@example.com",
        reason="Seasonal discount"
    )
    event = command_handler.handle(cmd)
    print(f"Command executed: price changed from ${event.old_price_cents/100:.2f} to ${event.new_price_cents/100:.2f}")

    # Apply event to projection
    projection.apply(event)
    print(f"Read model shows price: ${projection.get_product('PROD-001')['current_price_cents']/100:.2f}")

    # Consistency note: in a real async system, there's a delay.
    print("\nNote: In production, the read model update would be asynchronous.")
    print("The command handler does NOT wait for the projection to finish.")

// io.thecodeforge — System Design tutorial

// Simplified example: BFF for mobile vs web

class MobileBFF {
  constructor(orderService, userService, productService) {
    this.orderService = orderService;
    this.userService = userService;
    this.productService = productService;
  }

  // Mobile needs a lightweight order summary
  async getOrderSummary(orderId) {
    const order = await this.orderService.getOrder(orderId);
    // Only return what mobile needs — no full product details
    return {
      orderId: order.id,
      status: order.status,
      total: order.total,
      itemCount: order.items.length,
      lastUpdated: order.updatedAt
    };
  }
}

class WebBFF {
  constructor(orderService, userService, productService) {
    this.orderService = orderService;
    this.userService = userService;
    this.productService = productService;
  }

  // Web needs full details for product pages
  async getOrderDetail(orderId) {
    const order = await this.orderService.getOrder(orderId);
    // Enhance with product details
    const productDetails = await Promise.all(
      order.items.map(item => this.productService.getProduct(item.productId))
    );
    return {
      ...order,
      products: productDetails,
      user: await this.userService.getUser(order.userId)
    };
  }
}

// In production, these BFFs run as separate services,
// each with their own rate limits, auth, and circuit breakers.