Software Architecture Patterns: Trade-offs and Real Decisions

# 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[[dict], None]) -> None:
        """Register a consumer for a topic. Each subscriber gets ALL events
        published to that topic — not round-robin like a work queue."""
        with self._lock:
            consumer_queue: deque = deque()
            self._topics[topic].append(consumer_queue)
            # Start a background thread per consumer — simulates async consumption
            thread = threading.Thread(
                target=self._consume,
                args=(consumer_queue, handler),
                daemon=True  # Dies when main thread dies — fine for demo, use proper lifecycle in prod
            )
            thread.start()

    def publish(self, topic: str, event: dict) -> None:
        """Producer publishes and returns immediately. It does NOT wait for consumers.
        This is the core of EDA's decoupling — and the source of its observability challenges."""
        with self._lock:
            for consumer_queue in self._topics[topic]:
                consumer_queue.append(event)  # Each consumer gets its own copy
        print(f"[EventBus] Published to '{topic}': event_id={event.get('event_id')}")

    def _consume(self, queue: deque, 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, 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}


# --- Wire it up ---

if __name__ == "__main__":
    bus = InMemoryEventBus()

    # Register consumers — order doesn't matter, they're all independent
    inventory = InventoryConsumer()
    fraud = FraudDetectionConsumer()
    shipping = ShippingConsumer()
    email = EmailNotificationConsumer()

    bus.subscribe("order.confirmed", inventory.handle)
    bus.subscribe("order.confirmed", fraud.handle)
    bus.subscribe("order.confirmed", shipping.handle)
    bus.subscribe("order.confirmed", email.handle)

    checkout = CheckoutService(bus)

    # Simulate a completed checkout
    result = checkout.complete_checkout(
        order_id="ORD-20240315-002",
        customer_id="CUST-456",
        product_ids=["SKU-MONITOR-27", "SKU-HDMI-CABLE"],
        total=389.98
    )

    print(f"\n[Checkout] Response: {result}")
    time.sleep(0.1)  # Give async consumers time to process in this demo

# 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_update_price(
        self, command: UpdateProductPriceCommand
    ) -> ProductPriceUpdatedEvent:

        product = self.write_store.get(command.product_id)
        if not product:
            raise ValueError(f"Product {command.product_id} not found")

        if command.new_price_cents <= 0:
            raise ValueError("Price must be positive — write model enforces this invariant")

        old_price = product.price_cents
        product.price_cents = command.new_price_cents
        product.version += 1  # Increment version for optimistic lock detection

        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)  # In production: append to Kafka/outbox table
        print(f"[CommandHandler] Price updated: {command.product_id} "
              f"${old_price/100:.2f} → ${command.new_price_cents/100:.2f} "
              f"(v{product.version}) by {command.updated_by}")
        return event


# ─── READ SIDE (Query Model) ───────────────────────────────────────────────
# Denormalised. Pre-computed. Optimised for the specific query pattern.
# NOT the source of truth — it's a projection of the write side.

@dataclass
class ProductReadModel:
    """Flattened, denormalised view of a product.
    Fields are exactly what the product page API endpoint needs — nothing more.
    In production: this lives in Redis or Elasticsearch, not Postgres."""
    product_id: str
    display_name: str
    price_display: str     # Pre-formatted: '$149.99' — computed at write time, not read time
    category_name: str     # Joined and baked in — no JOIN at query time
    brand_name: str
    is_available: bool
    search_keywords: list[str]  # Pre-extracted for search — no LIKE query at runtime


class ProductReadModelProjection:
    """Listens to events from the write side and updates read models.
    This is the 'eventually consistent' part — it runs async after the command.
    If this falls behind, reads show stale data. Monitor it."""

    def __init__(self, read_store: dict, category_lookup: dict, brand_lookup: dict):
        self.read_store = read_store
        self.category_lookup = category_lookup  # Pre-loaded reference data
        self.brand_lookup = brand_lookup

    def project_price_update(self, event: ProductPriceUpdatedEvent,
                              write_model: ProductWriteModel) -> None:
        """Rebuild only the affected fields in the read model.
        In production with Kafka: this handler is idempotent — if it runs twice
        (at-least-once delivery), the result is the same."""

        existing_read_model = self.read_store.get(event.product_id)
        if not existing_read_model:
            # First time building this read model — construct from write model
            existing_read_model = ProductReadModel(
                product_id=write_model.product_id,
                display_name=write_model.name,
                price_display="",  # Will be set below
                category_name=self.category_lookup.get(write_model.category_id, "Unknown"),
                brand_name=self.brand_lookup.get(write_model.brand_id, "Unknown"),
                is_available=write_model.is_active and write_model.stock_count > 0,
                search_keywords=write_model.name.lower().split()
            )

        # Update ONLY the price field — targeted, not full rebuild
        existing_read_model.price_display = f"${event.new_price_cents / 100:.2f}"
        self.read_store[event.product_id] = existing_read_model

        print(f"[Projection] Read model updated for {event.product_id}: "
              f"price now {existing_read_model.price_display}")


class ProductQueryHandler:
    """Handles all reads. Hits the read store only — never touches the write DB.
    This is why reads are fast: the read store is pre-computed and query-shaped."""

    def __init__(self, read_store: dict):
        self.read_store = read_store

    def get_product_page_data(self, product_id: str) -> Optional[dict]:
        model = self.read_store.get(product_id)
        if not model:
            return None
        # Return exactly what the product page needs — no transformation at read time
        return {
            "id": model.product_id,
            "name": model.display_name,
            "price": model.price_display,
            "category": model.category_name,
            "brand": model.brand_name,
            "available": model.is_available
        }


# ─── Wire it all together ──────────────────────────────────────────────────

if __name__ == "__main__":
    # Simulate backing stores
    write_store = {
        "PROD-MONITOR-4K-32": ProductWriteModel(
            product_id="PROD-MONITOR-4K-32",
            sku="MON-32-4K-BK",
            name="32-inch 4K Monitor",
            description="IPS panel, 144Hz, USB-C PD 96W",
            price_cents=59999,
            stock_count=47,
            category_id="CAT-DISPLAYS",
            brand_id="BRD-DELL"
        )
    }
    read_store = {}
    category_lookup = {"CAT-DISPLAYS": "Monitors & Displays"}
    brand_lookup = {"BRD-DELL": "Dell"}

    command_handler = ProductCommandHandler(write_store)
    projection = ProductReadModelProjection(read_store, category_lookup, brand_lookup)
    query_handler = ProductQueryHandler(read_store)

    # 1. Admin issues a price update command
    update_command = UpdateProductPriceCommand(
        product_id="PROD-MONITOR-4K-32",
        new_price_cents=54999,  # Sale price
        updated_by="admin-jenna@example.com",
        reason="Black Friday promotional pricing"
    )

    event = command_handler.handle_update_price(update_command)

    # 2. Projection layer processes the event (async in prod, sync here for clarity)
    projection.project_price_update(event, write_store["PROD-MONITOR-4K-32"])

    # 3. Query handler serves the product page — hits read store only
    product_page = query_handler.get_product_page_data("PROD-MONITOR-4K-32")
    print(f"\n[QueryHandler] Product page response: {json.dumps(product_page, indent=2)}")

# io.thecodeforge — System Design tutorial

# Architecture Decision Records (ADRs) are the most underused practice
# in software teams. This isn't just documentation — it's the thing that
# stops your team from having the same architectural argument six months later
# when half the context has walked out the door with someone who quit.

# An ADR captures: what decision was made, what alternatives were rejected,
# what problem it solves, and — critically — when to revisit it.
# Format based on Michael Nygard's widely adopted template.

from dataclasses import dataclass, field
from datetime import date
from enum import Enum
from typing import Optional

class ADRStatus(Enum):
    PROPOSED = "Proposed"       # Under discussion — not committed
    ACCEPTED = "Accepted"       # Decision made, team aligned
    DEPRECATED = "Deprecated"   # Superseded by a later ADR
    SUPERSEDED = "Superseded"   # Replaced — link to the new ADR

@dataclass
class Alternative:
    name: str
    why_rejected: str  # Honest, specific reason — not 'it was worse'

@dataclass
class ArchitectureDecisionRecord:
    adr_number: int
    title: str
    status: ADRStatus
    date_decided: date
    deciders: list[str]           # Who was in the room — accountability
    context: str                  # What problem forced a decision?
    decision: str                 # What did you decide? One clear statement.
    consequences_positive: list[str]   # What you gain
    consequences_negative: list[str]   # What you sacrifice — be honest
    alternatives_rejected: list[Alternative]
    revisit_trigger: str          # Specific condition that should re-open this decision
    superseded_by: Optional[int] = None  # ADR number that replaces this one

    def render(self) -> str:
        lines = [
            f"# ADR-{self.adr_number:04d}: {self.title}",
            f"**Status:** {self.status.value}   **Date:** {self.date_decided}   "
            f"**Deciders:** {', '.join(self.deciders)}",
            "",
            "## Context",
            self.context,
            "",
            "## Decision",
            self.decision,
            "",
            "## Consequences",
            "**Gains:**",
        ]
        for item in self.consequences_positive:
            lines.append(f"  + {item}")
        lines.append("**Costs:**")
        for item in self.consequences_negative:
            lines.append(f"  - {item}")
        lines.append("")
        lines.append("## Alternatives Rejected")
        for alt in self.alternatives_rejected:
            lines.append(f"**{alt.name}:** {alt.why_rejected}")
        lines.append("")
        lines.append("## Revisit When")
        lines.append(self.revisit_trigger)
        if self.superseded_by:
            lines.append(f"\n> **Superseded by ADR-{self.superseded_by:04d}**")
        return "\n".join(lines)


# ─── Real example: the decision NOT to go microservices ──────────────────

adr_001 = ArchitectureDecisionRecord(
    adr_number=1,
    title="Deploy as a modular monolith, not microservices",
    status=ADRStatus.ACCEPTED,
    date_decided=date(2024, 3, 15),
    deciders=["sarah-cto@example.com", "marcus-lead@example.com", "priya-senior@example.com"],
    context=(
        "We are a 6-engineer team building a B2B SaaS product with ~200 customers. "
        "Current P99 API latency is 180ms against a 500ms SLA — we have headroom. "
        "Peak load is 150 concurrent users. Engineering asked whether to start with "
        "microservices to 'build for scale' before our Series A."
    ),
    decision=(
        "Ship as a modular monolith with clear internal boundaries: orders/, billing/, "
        "inventory/, notifications/ as distinct Python packages with no cross-package "
        "imports except through defined interface classes. Single PostgreSQL database "
        "with schema-per-module. Single deployment unit."
    ),
    consequences_positive=[
        "Single deployment pipeline — ~20min CI vs estimated 2hr for 8+ microservices",
        "ACID transactions across modules — no saga patterns needed at this scale",
        "Local debugging with a single process — no distributed tracing setup",
        "New engineer onboarding: clone one repo, run one docker-compose",
    ],
    consequences_negative=[
        "A crash in one module affects all modules — mitigated by process-level health checks",
        "Shared database means schema migrations are global — coordinate carefully",
        "Harder to scale modules independently if one becomes a bottleneck",
    ],
    alternatives_rejected=[
        Alternative(
            name="Microservices from day one",
            why_rejected=(
                "At 6 engineers, we don't have the operational bandwidth to run "
                "distributed tracing, per-service deployment pipelines, service meshes, "
                "or manage eventual consistency. The complexity cost exceeds the scaling "
                "benefit at our current load. Revisit when load requires independent scaling."
            )
        ),
        Alternative(
            name="Serverless functions (AWS Lambda)",
            why_rejected=(
                "Cold start latency is incompatible with our interactive API SLA. "
                "Local development story is poor. State management for our transaction-heavy "
                "workflows requires significant workarounds."
            )
        ),
    ],
    revisit_trigger=(
        "Re-evaluate when: (a) any single module exceeds 40% of total DB CPU, "
        "(b) engineering team exceeds 20 engineers, or "
        "(c) two teams are blocked on the same deployment more than 3 times per sprint."
    )
)

if __name__ == "__main__":
    print(adr_001.render())