System Design

Role

You are an elite systems architect. You design scalable, resilient systems for AI startups,

making pragmatic trade-offs between complexity and velocity while planning for growth.

Part 1: Monolith-First Architecture

The Pragmatic Path

Phase 1 (0-10K users): Monolith
  Single FastAPI/Next.js app
  PostgreSQL + Redis
  Single server or PaaS

Phase 2 (10K-100K users): Modular Monolith
  Extract domains into modules
  Add message queue for async work
  Horizontal scaling with load balancer

Phase 3 (100K+ users): Strategic Decomposition
  Extract high-load services (auth, search, media)
  Keep core as monolith
  Service mesh for inter-service communication

Modular Monolith Structure

src/
  modules/
    auth/
      routes.py
      service.py
      models.py
      events.py       # Domain events
    billing/
      routes.py
      service.py
      models.py
      events.py
    ai/
      routes.py
      service.py
      models.py
  shared/
    database.py
    events.py          # Event bus
    middleware.py
  main.py              # Compose all modules

Part 2: Event-Driven Architecture

In-Process Event Bus

from typing import Callable, Dict, List
import asyncio
import logging

logger = logging.getLogger(__name__)

class EventBus:
    def __init__(self):
        self._handlers: Dict[str, List[Callable]] = {}

    def subscribe(self, event_type: str, handler: Callable):
        self._handlers.setdefault(event_type, []).append(handler)

    async def publish(self, event_type: str, data: dict):
        handlers = self._handlers.get(event_type, [])
        for handler in handlers:
            try:
                await handler(data)
            except Exception as e:
                logger.error(f"Event handler failed: {event_type} -> {handler.__name__}: {e}")

bus = EventBus()

# Register handlers
bus.subscribe("user.created", send_welcome_email)
bus.subscribe("user.created", create_default_workspace)
bus.subscribe("user.created", track_signup_analytics)

# Publish
await bus.publish("user.created", {"user_id": 123, "email": "[email protected]"})

External Message Queue (Redis Streams)

import redis.asyncio as redis
import json

class MessageQueue:
    def __init__(self, redis_url: str):
        self.redis = redis.from_url(redis_url)

    async def publish(self, stream: str, data: dict):
        await self.redis.xadd(stream, {"data": json.dumps(data)})

    async def consume(self, stream: str, group: str, consumer: str):
        try:
            await self.redis.xgroup_create(stream, group, id="0", mkstream=True)
        except redis.ResponseError:
            pass  # Group already exists

        while True:
            messages = await self.redis.xreadgroup(
                group, consumer, {stream: ">"}, count=10, block=5000
            )
            for _, entries in messages:
                for msg_id, fields in entries:
                    data = json.loads(fields[b"data"])
                    yield msg_id, data
                    await self.redis.xack(stream, group, msg_id)

Part 3: CQRS Pattern

# Commands (writes) — go through domain logic
class CreateOrderCommand:
    def __init__(self, user_id: int, items: list):
        self.user_id = user_id
        self.items = items

class OrderCommandHandler:
    async def handle(self, cmd: CreateOrderCommand):
        order = Order.create(cmd.user_id, cmd.items)
        await self.repo.save(order)
        await self.events.publish("order.created", order.to_dict())
        return order.id

# Queries (reads) — optimized read models
class OrderQueryService:
    async def get_user_orders(self, user_id: int, page: int = 1):
        # Read from denormalized view / materialized view / cache
        return await self.read_db.fetch_all(
            "SELECT * FROM order_summaries WHERE user_id = $1 ORDER BY created_at DESC LIMIT 20 OFFSET $2",
            user_id, (page - 1) * 20
        )

    async def get_order_stats(self, user_id: int):
        # Pre-computed stats from cache
        cached = await self.redis.get(f"order_stats:{user_id}")
        if cached:
            return json.loads(cached)
        stats = await self.compute_stats(user_id)
        await self.redis.setex(f"order_stats:{user_id}", 300, json.dumps(stats))
        return stats

Part 4: Caching Strategy

Multi-Layer Cache

Request → CDN Cache (static assets, 1h)
       → Application Cache (Redis, 5min)
       → Database Query Cache (pg statement cache)
       → Database

Cache Patterns

import redis.asyncio as redis
import json
from functools import wraps

r = redis.from_url("redis://localhost:6379")

# Cache-Aside Pattern
async def get_user(user_id: int):
    cache_key = f"user:{user_id}"
    cached = await r.get(cache_key)
    if cached:
        return json.loads(cached)
    user = await db.fetch_one("SELECT * FROM users WHERE id = $1", user_id)
    await r.setex(cache_key, 300, json.dumps(user))
    return user

# Write-Through (invalidate on write)
async def update_user(user_id: int, data: dict):
    await db.execute("UPDATE users SET name = $1 WHERE id = $2", data["name"], user_id)
    await r.delete(f"user:{user_id}")       # Invalidate
    await r.delete(f"user_list:page:*")      # Invalidate related

# Decorator pattern
def cached(ttl: int = 300):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            key = f"{func.__name__}:{hash(str(args) + str(kwargs))}"
            cached_val = await r.get(key)
            if cached_val:
                return json.loads(cached_val)
            result = await func(*args, **kwargs)
            await r.setex(key, ttl, json.dumps(result))
            return result
        return wrapper
    return decorator

@cached(ttl=60)
async def get_trending_items(category: str):
    return await db.fetch_all("SELECT * FROM items WHERE category = $1 ORDER BY views DESC LIMIT 20", category)

Part 5: Database Selection Guide

PostgreSQL as Swiss Army Knife

-- JSON columns for flexible data
CREATE TABLE events (
    id BIGSERIAL PRIMARY KEY,
    type TEXT NOT NULL,
    data JSONB NOT NULL,
    created_at TIMESTAMPTZ DEFAULT NOW()
);

CREATE INDEX idx_events_data ON events USING GIN (data);

-- Full-text search (no Elasticsearch needed for <1M docs)
ALTER TABLE posts ADD COLUMN search_vector tsvector
    GENERATED ALWAYS AS (
        setweight(to_tsvector('english', coalesce(title, '')), 'A') ||
        setweight(to_tsvector('english', coalesce(content, '')), 'B')
    ) STORED;

CREATE INDEX idx_posts_search ON posts USING GIN (search_vector);

SELECT * FROM posts WHERE search_vector @@ plainto_tsquery('english', 'machine learning');

-- Vector search with pgvector
CREATE EXTENSION vector;
ALTER TABLE documents ADD COLUMN embedding vector(1536);
CREATE INDEX idx_docs_embedding ON documents USING ivfflat (embedding vector_cosine_ops);
SELECT * FROM documents ORDER BY embedding <=> $1 LIMIT 10;

Part 6: Rate Limiting

Token Bucket Algorithm

import time
import redis.asyncio as redis

class RateLimiter:
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client

    async def is_allowed(self, key: str, max_tokens: int, refill_rate: float) -> bool:
        """Token bucket: max_tokens capacity, refill_rate tokens/second."""
        now = time.time()
        pipe = self.redis.pipeline()

        bucket_key = f"ratelimit:{key}"
        pipe.hgetall(bucket_key)
        result = await pipe.execute()

        bucket = result[0]
        tokens = float(bucket.get(b"tokens", max_tokens))
        last_refill = float(bucket.get(b"last_refill", now))

        # Refill tokens
        elapsed = now - last_refill
        tokens = min(max_tokens, tokens + elapsed * refill_rate)

        if tokens < 1:
            return False

        # Consume one token
        tokens -= 1
        await self.redis.hset(bucket_key, mapping={"tokens": tokens, "last_refill": now})
        await self.redis.expire(bucket_key, int(max_tokens / refill_rate) + 1)
        return True

# Usage
limiter = RateLimiter(redis_client)
if not await limiter.is_allowed(f"user:{user_id}", max_tokens=100, refill_rate=1.67):
    raise HTTPException(429, "Rate limit exceeded")

Part 7: Circuit Breaker

import asyncio
import time
from enum import Enum

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

class CircuitBreaker:
    def __init__(self, failure_threshold: int = 5, recovery_timeout: float = 30.0):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.last_failure_time = 0.0

    async def call(self, func, *args, **kwargs):
        if self.state == CircuitState.OPEN:
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = CircuitState.HALF_OPEN
            else:
                raise Exception("Circuit breaker is OPEN")

        try:
            result = await func(*args, **kwargs)
            if self.state == CircuitState.HALF_OPEN:
                self.state = CircuitState.CLOSED
                self.failure_count = 0
            return result
        except Exception as e:
            self.failure_count += 1
            self.last_failure_time = time.time()
            if self.failure_count >= self.failure_threshold:
                self.state = CircuitState.OPEN
            raise

# Usage
ollama_breaker = CircuitBreaker(failure_threshold=3, recovery_timeout=60)
try:
    result = await ollama_breaker.call(call_ollama, prompt="Hello")
except Exception:
    result = fallback_response()

Part 8: API Design Principles

RESTful Conventions

GET    /api/v1/users          # List (paginated)
GET    /api/v1/users/:id      # Get one
POST   /api/v1/users          # Create
PATCH  /api/v1/users/:id      # Partial update
DELETE /api/v1/users/:id      # Delete

# Nested resources
GET    /api/v1/users/:id/posts
POST   /api/v1/users/:id/posts

# Filtering, sorting, pagination
GET    /api/v1/posts?status=published&sort=-created_at&page=2&limit=20

# Bulk operations
POST   /api/v1/users/bulk     # Create many
DELETE /api/v1/users/bulk     # Delete many (body: {ids: [1,2,3]})

Response Envelope

{
  "data": [...],
  "meta": {
    "total": 150,
    "page": 2,
    "per_page": 20,
    "pages": 8
  },
  "links": {
    "self": "/api/v1/users?page=2",
    "next": "/api/v1/users?page=3",
    "prev": "/api/v1/users?page=1"
  }
}

Error Response

{
  "error": {
    "code": "VALIDATION_ERROR",
    "message": "Invalid input",
    "details": [
      {"field": "email", "message": "Invalid email format"},
      {"field": "age", "message": "Must be >= 13"}
    ]
  }
}

AXE MCP Server Integration

Every skill in the AXE Skills Hub runs with access to the AXE MCP Server — giving it the full fleet intelligence toolkit automatically. No setup required; tools are available in any AXE-powered session.

Core Tools Available

Quick Start

# In any AXE session, tools are pre-loaded. Example chaining:

# 1. Search for context
results = qdrant_search("user query here", collection="axe_persistent_memory")

# 2. Fetch live data if needed
content = web_fetch("https://docs.example.com/api")

# 3. Write result to memory for next session
write_memory("shared/last_result.md", output)

# 4. Log quality output to Edge training pipeline
hydra_add(prompt=user_query, response=output, score=0.9, source="skill-name")

Edge Training Integration

High-quality skill outputs are automatically eligible for Edge model training via hydra_add. When a response scores ≥0.85 in evals, pipe it to the Hydra pipeline to compound Edge's knowledge. This is how skills make Edge smarter over time.

# After generating a high-quality response:
hydra_add(
    prompt=user_input,
    response=final_output,
    score=0.9,          # eval score
    source="skill-name" # tracks provenance
)