# Multi-Tenant Architecture Patterns

## The Production Nightmare: When Tenant A Saw Tenant B's Orders

It was a Friday afternoon (of course it was). I was about to leave for the weekend when Slack lit up. A restaurant manager reported seeing orders from a completely different restaurant in their dashboard. Then another report came in. Then another.

My blood ran cold. This wasn't a UI glitch—this was a catastrophic multi-tenant data leak. Restaurant A could see Restaurant B's orders, revenue data, customer information. Everything.

The root cause? One missing line of code:

```python
# What I wrote (WRONG):
@app.get("/orders")
async def get_orders(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    orders = db.query(Order).offset(skip).limit(limit).all()
    return orders

# What I should have written:
@app.get("/orders")
async def get_orders(
    skip: int = 0, 
    limit: int = 100,
    tenant_id: str = Depends(get_tenant_id),
    db: Session = Depends(get_db)
):
    orders = db.query(Order).filter(Order.tenant_id == tenant_id).offset(skip).limit(limit).all()
    return orders
```

One missing `filter(Order.tenant_id == tenant_id)` exposed every tenant's data to every other tenant. That Friday night, I learned that multi-tenancy isn't a feature—it's a fundamental architectural constraint that must be enforced at every level.

## What is Multi-Tenancy?

Multi-tenancy means serving multiple customers (tenants) from a single application instance while keeping their data completely isolated.

In my POS system, I onboarded several types of clients:

* **Tenant 1:** A local restaurant client
* **Tenant 2:** A fast food chain client
* **Tenant 3:** A retail shop client

Each tenant should:

* See only their own data
* Be completely unaware other tenants exist
* Have isolated configuration and customization
* Never be able to access another tenant's data

## Three Multi-Tenant Isolation Strategies

{% @mermaid/diagram content="graph TB
subgraph "Strategy 1: Separate Database"
APP1\[Application]
DB1\[(Restaurant Client DB)]
DB2\[(Fast Food Client DB)]
DB3\[(Retail Client DB)]

```
    APP1 --> DB1
    APP1 --> DB2
    APP1 --> DB3
end

subgraph "Strategy 2: Shared Database, Separate Schemas"
    APP2[Application]
    DB4[(Shared Database)]
    
    APP2 --> DB4
    
    DB4 --> SCHEMA1[Schema: restaurant_01]
    DB4 --> SCHEMA2[Schema: fastfood_01]
    DB4 --> SCHEMA3[Schema: retail_01]
end

subgraph "Strategy 3: Shared Database, Shared Tables"
    APP3[Application]
    DB5[(Shared Database)]
    
    APP3 --> DB5
    
    DB5 --> TABLE1[Orders Table<br/>tenant_id column]
    DB5 --> TABLE2[Products Table<br/>tenant_id column]
end" %}
```

Let me show you each strategy as implemented in the POS system.

### Strategy 1: Separate Database Per Tenant

**Pros:**

* Strongest isolation (physically separate data)
* Easy backup/restore per tenant
* Can customize schema per tenant
* Simple compliance (delete tenant = drop database)

**Cons:**

* Expensive (one DB instance per tenant)
* Complex connection management
* Hard to do cross-tenant analytics
* Schema migrations run N times

```python
# config/database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from typing import Dict
import os

class MultiTenantDatabaseManager:
    """Manages separate database per tenant"""
    
    _engines: Dict[str, any] = {}
    _sessions: Dict[str, any] = {}
    
    @classmethod
    def get_engine(cls, tenant_id: str):
        """Get or create database engine for tenant"""
        if tenant_id not in cls._engines:
            # Each tenant has separate database
            database_url = f"postgresql://user:password@localhost:5432/{tenant_id}_db"
            
            cls._engines[tenant_id] = create_engine(
                database_url,
                pool_size=5,
                max_overflow=10,
                pool_pre_ping=True
            )
            
            cls._sessions[tenant_id] = sessionmaker(
                autocommit=False,
                autoflush=False,
                bind=cls._engines[tenant_id]
            )
        
        return cls._engines[tenant_id]
    
    @classmethod
    def get_session(cls, tenant_id: str):
        """Get database session for tenant"""
        if tenant_id not in cls._sessions:
            cls.get_engine(tenant_id)
        
        return cls._sessions[tenant_id]()

# Usage in endpoints
from fastapi import Depends, Header

def get_tenant_db(x_tenant_id: str = Header(...)):
    """Dependency to get tenant-specific database session"""
    db = MultiTenantDatabaseManager.get_session(x_tenant_id)
    try:
        yield db
    finally:
        db.close()

@app.get("/orders")
async def get_orders(db: Session = Depends(get_tenant_db)):
    # No need for tenant_id filter - this database only has this tenant's data
    orders = db.query(Order).all()
    return orders
```

**When to use:** Enterprise SaaS where tenants pay premium, need guaranteed isolation, or have compliance requirements.

### Strategy 2: Shared Database, Separate Schemas

**Pros:**

* Good isolation (schema-level separation)
* Cheaper than separate databases
* Easier connection pooling
* Can do cross-tenant queries if needed

**Cons:**

* Still complex migrations
* Need schema switching logic
* Backup/restore more complex

```python
# config/database.py
from sqlalchemy import create_engine, event, text
from sqlalchemy.orm import sessionmaker, Session
from sqlalchemy.pool import NullPool

# Single database with multiple schemas
engine = create_engine(
    "postgresql://user:password@localhost:5432/pos_db",
    poolclass=NullPool  # Don't pool to avoid schema switching issues
)

def get_tenant_session(tenant_id: str):
    """Get session with search_path set to tenant schema"""
    session = Session(bind=engine)
    
    # Set PostgreSQL search_path to tenant schema
    session.execute(text(f"SET search_path TO {tenant_id}, public"))
    
    return session

# Automatic schema creation for new tenants
from sqlalchemy import MetaData

def create_tenant_schema(tenant_id: str):
    """Create schema and tables for new tenant"""
    with engine.connect() as conn:
        # Create schema
        conn.execute(text(f"CREATE SCHEMA IF NOT EXISTS {tenant_id}"))
        conn.commit()
        
        # Create tables in schema
        metadata = MetaData(schema=tenant_id)
        # Define your models with this metadata
        # metadata.create_all(bind=engine)

# Usage
@app.get("/orders")
async def get_orders(x_tenant_id: str = Header(...)):
    db = get_tenant_session(x_tenant_id)
    try:
        # Queries automatically use tenant schema due to search_path
        orders = db.query(Order).all()
        return orders
    finally:
        db.close()
```

**When to use:** Mid-market SaaS with moderate tenant count, need good isolation but not separate databases.

### Strategy 3: Shared Database, Shared Tables (Discriminator Column)

**Pros:**

* Simplest to implement
* Best performance (shared connection pool)
* Easy cross-tenant analytics
* Cheapest option

**Cons:**

* Higher risk of data leaks (my Friday nightmare)
* Must filter every query by tenant\_id
* Indexes include tenant\_id (larger indexes)

```python
# models/base.py
from sqlalchemy import Column, String
from sqlalchemy.ext.declarative import declared_attr

class TenantMixin:
    """Mixin to add tenant_id to all models"""
    
    @declared_attr
    def tenant_id(cls):
        return Column(String(50), nullable=False, index=True)

# models/order.py
from sqlalchemy import Column, Integer, Float, DateTime, JSON
from models.base import TenantMixin, Base

class Order(TenantMixin, Base):
    __tablename__ = "orders"
    
    id = Column(Integer, primary_key=True)
    # tenant_id inherited from TenantMixin
    order_number = Column(String(50), nullable=False)
    total_amount = Column(Float, nullable=False)
    items = Column(JSON, nullable=False)
    
    # Composite index for performance
    __table_args__ = (
        Index('ix_orders_tenant_id_created_at', 'tenant_id', 'created_at'),
    )
```

This is what I use in the POS system. Here's how to do it safely:

## The x-tenant-id Header Flow Through All Services

The tenant ID must flow through every request, to every service, for every operation.

{% @mermaid/diagram content="sequenceDiagram
participant Client
participant Auth
participant POSCore
participant Inventory
participant Payment
participant DB

```
Client->>Auth: POST /auth/login<br/>x-tenant-id: restaurant_01
Auth->>DB: Verify user in restaurant_01 tenant
Auth-->>Client: JWT token (contains tenant_id)

Client->>POSCore: POST /orders<br/>x-tenant-id: restaurant_01<br/>Authorization: Bearer {token}
POSCore->>POSCore: Verify token.tenant_id == x-tenant-id
POSCore->>Inventory: Check stock<br/>x-tenant-id: restaurant_01
Inventory->>DB: SELECT * FROM products<br/>WHERE tenant_id = 'restaurant_01'
Inventory-->>POSCore: Stock available
POSCore->>Payment: Process payment<br/>x-tenant-id: restaurant_01
Payment->>DB: INSERT INTO payments<br/>tenant_id = 'restaurant_01'
Payment-->>POSCore: Payment successful
POSCore->>DB: INSERT INTO orders<br/>tenant_id = 'restaurant_01'
POSCore-->>Client: Order created" %}
```

### Enforcing Tenant Isolation at Every Layer

```python
# 1. Middleware Layer: Extract and validate tenant_id
from fastapi import Request, HTTPException
from starlette.middleware.base import BaseHTTPMiddleware

class TenantMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # Extract tenant_id from header
        tenant_id = request.headers.get("x-tenant-id")
        
        if not tenant_id and request.url.path not in ["/health", "/docs"]:
            raise HTTPException(400, "Missing x-tenant-id header")
        
        # Validate tenant exists (prevent fake tenant_ids)
        if tenant_id and not await self.validate_tenant(tenant_id):
            raise HTTPException(404, "Tenant not found")
        
        # Store in request state
        request.state.tenant_id = tenant_id
        
        response = await call_next(request)
        return response
    
    async def validate_tenant(self, tenant_id: str) -> bool:
        # Check if tenant exists in tenants table
        # Cache this check for performance
        return True  # Simplified

# 2. JWT Token Layer: Verify token contains matching tenant_id
import jwt
from config.settings import settings

class AuthMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        public_paths = ["/health", "/auth/login", "/auth/register"]
        if request.url.path in public_paths:
            return await call_next(request)
        
        # Extract JWT token
        auth_header = request.headers.get("authorization", "")
        if not auth_header.startswith("Bearer "):
            raise HTTPException(401, "Missing authorization header")
        
        token = auth_header.replace("Bearer ", "")
        
        try:
            payload = jwt.decode(token, settings.JWT_SECRET_KEY, algorithms=["HS256"])
        except jwt.InvalidTokenError:
            raise HTTPException(401, "Invalid token")
        
        # Critical: Verify token tenant matches header tenant
        token_tenant = payload.get("tenant_id")
        header_tenant = request.state.tenant_id
        
        if token_tenant != header_tenant:
            raise HTTPException(403, "Tenant ID mismatch - potential security violation")
        
        # Store user info
        request.state.user_id = payload.get("user_id")
        request.state.user_role = payload.get("role")
        
        response = await call_next(request)
        return response

# 3. Repository Layer: Always filter by tenant_id
class OrderRepository:
    def __init__(self, db: Session, tenant_id: str):
        self.db = db
        self.tenant_id = tenant_id
    
    def get_by_id(self, order_id: int):
        return self.db.query(Order).filter(
            Order.id == order_id,
            Order.tenant_id == self.tenant_id  # Never forget this!
        ).first()
    
    def list_all(self, skip: int = 0, limit: int = 100):
        return self.db.query(Order).filter(
            Order.tenant_id == self.tenant_id  # Never forget this!
        ).offset(skip).limit(limit).all()
    
    def create(self, order_data: dict):
        order = Order(
            **order_data,
            tenant_id=self.tenant_id  # Automatically set from repository
        )
        self.db.add(order)
        self.db.commit()
        self.db.refresh(order)
        return order

# 4. Service Layer: Pass tenant_id to repository
class OrderService:
    def __init__(self, db: Session, tenant_id: str):
        self.repository = OrderRepository(db, tenant_id)
    
    def get_order(self, order_id: int):
        order = self.repository.get_by_id(order_id)
        if not order:
            raise HTTPException(404, "Order not found")
        return order

# 5. Endpoint Layer: Inject tenant_id from request state
from fastapi import Depends

def get_tenant_id(request: Request) -> str:
    """Dependency to extract tenant_id from request state"""
    return request.state.tenant_id

def get_order_service(
    tenant_id: str = Depends(get_tenant_id),
    db: Session = Depends(get_db)
) -> OrderService:
    """Dependency to create OrderService with tenant isolation"""
    return OrderService(db, tenant_id)

@app.get("/orders/{order_id}")
async def get_order(
    order_id: int,
    service: OrderService = Depends(get_order_service)
):
    # Service already has tenant_id, no way to forget filtering
    return service.get_order(order_id)
```

## PostgreSQL Row-Level Security (Defense in Depth)

Even with application-level filtering, I added database-level protection:

```sql
-- Enable row-level security on orders table
ALTER TABLE orders ENABLE ROW LEVEL SECURITY;

-- Create policy: users can only see their tenant's data
CREATE POLICY tenant_isolation_policy ON orders
    USING (tenant_id = current_setting('app.current_tenant_id')::text);

-- Create policy for inserts
CREATE POLICY tenant_insert_policy ON orders
    FOR INSERT
    WITH CHECK (tenant_id = current_setting('app.current_tenant_id')::text);

-- Similar for other tables
ALTER TABLE products ENABLE ROW LEVEL SECURITY;
CREATE POLICY tenant_isolation_policy ON products
    USING (tenant_id = current_setting('app.current_tenant_id')::text);

ALTER TABLE payments ENABLE ROW LEVEL SECURITY;
CREATE POLICY tenant_isolation_policy ON payments
    USING (tenant_id = current_setting('app.current_tenant_id')::text);
```

```python
# config/database.py
from sqlalchemy import event, text

@event.listens_for(SessionLocal, "after_begin")
def set_tenant_context(session, transaction, connection):
    """Set tenant context for PostgreSQL RLS"""
    if hasattr(session, 'tenant_id'):
        connection.execute(
            text(f"SET LOCAL app.current_tenant_id = '{session.tenant_id}'")
        )

# Usage
def get_tenant_db(tenant_id: str = Depends(get_tenant_id)):
    db = SessionLocal()
    db.tenant_id = tenant_id  # Set tenant_id on session
    try:
        yield db
    finally:
        db.close()

# Now even if you forget to filter by tenant_id, PostgreSQL won't return other tenants' data
@app.get("/orders")
async def get_orders(db: Session = Depends(get_tenant_db)):
    # Even without explicit filter, RLS ensures only current tenant's orders returned
    orders = db.query(Order).all()
    return orders
```

**This saved me multiple times!** When I forgot to add tenant filters during rapid development, RLS prevented data leaks.

## Redis Cache with Tenant Prefixes

Caching adds another layer where tenant isolation is critical:

```python
# services/cache.py
import redis
from typing import Optional, Any
import json

class TenantAwareCache:
    """Redis cache with automatic tenant prefixing"""
    
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client
    
    def _make_key(self, tenant_id: str, key: str) -> str:
        """Prefix all keys with tenant_id"""
        return f"tenant:{tenant_id}:{key}"
    
    def get(self, tenant_id: str, key: str) -> Optional[Any]:
        """Get value from cache"""
        cache_key = self._make_key(tenant_id, key)
        value = self.redis.get(cache_key)
        
        if value:
            return json.loads(value)
        return None
    
    def set(self, tenant_id: str, key: str, value: Any, ttl: int = 3600):
        """Set value in cache with TTL"""
        cache_key = self._make_key(tenant_id, key)
        self.redis.setex(cache_key, ttl, json.dumps(value))
    
    def delete(self, tenant_id: str, key: str):
        """Delete value from cache"""
        cache_key = self._make_key(tenant_id, key)
        self.redis.delete(cache_key)
    
    def clear_tenant(self, tenant_id: str):
        """Clear all cache for a tenant"""
        pattern = f"tenant:{tenant_id}:*"
        keys = self.redis.keys(pattern)
        if keys:
            self.redis.delete(*keys)

# Usage in services
class ProductService:
    def __init__(self, db: Session, tenant_id: str, cache: TenantAwareCache):
        self.db = db
        self.tenant_id = tenant_id
        self.cache = cache
    
    def get_product(self, product_id: int):
        # Try cache first
        cache_key = f"product:{product_id}"
        cached = self.cache.get(self.tenant_id, cache_key)
        
        if cached:
            return cached
        
        # Not in cache, query database
        product = self.db.query(Product).filter(
            Product.id == product_id,
            Product.tenant_id == self.tenant_id
        ).first()
        
        if product:
            # Cache for 1 hour
            self.cache.set(self.tenant_id, cache_key, product.dict(), ttl=3600)
        
        return product
```

## Preventing Cross-Tenant Data Leaks: Comprehensive Checklist

Here's my checklist to prevent the nightmare I experienced:

### 1. Request Level

* [ ] Extract tenant\_id from header
* [ ] Validate tenant exists
* [ ] Extract tenant\_id from JWT
* [ ] Verify header tenant\_id matches JWT tenant\_id
* [ ] Store tenant\_id in request context

### 2. Service Level

* [ ] Pass tenant\_id to all service constructors
* [ ] Services accept tenant\_id as first parameter
* [ ] Services pass tenant\_id to repositories

### 3. Repository Level

* [ ] Every query includes `WHERE tenant_id = ?`
* [ ] Every insert sets `tenant_id` field
* [ ] Use repository base class to enforce filtering

### 4. Database Level

* [ ] Enable PostgreSQL Row-Level Security
* [ ] Set tenant context in database session
* [ ] Test RLS policies thoroughly

### 5. Cache Level

* [ ] Prefix all cache keys with tenant\_id
* [ ] Never cache cross-tenant data
* [ ] Clear tenant cache on tenant deletion

### 6. Testing Level

* [ ] Test with multiple tenants
* [ ] Verify isolation in integration tests
* [ ] Audit all queries for tenant filtering

### Automated Testing for Tenant Isolation

```python
# tests/test_tenant_isolation.py
import pytest
from fastapi.testclient import TestClient

def test_order_isolation_between_tenants():
    """Ensure tenant A cannot see tenant B's orders"""
    
    # Create order for tenant A
    client_a = TestClient(app)
    response_a = client_a.post(
        "/orders",
        headers={
            "x-tenant-id": "tenant_a",
            "authorization": f"Bearer {tenant_a_token}"
        },
        json={"items": [{"product_id": 1, "quantity": 2}]}
    )
    assert response_a.status_code == 201
    order_a_id = response_a.json()["id"]
    
    # Create order for tenant B
    client_b = TestClient(app)
    response_b = client_b.post(
        "/orders",
        headers={
            "x-tenant-id": "tenant_b",
            "authorization": f"Bearer {tenant_b_token}"
        },
        json={"items": [{"product_id": 2, "quantity": 1}]}
    )
    assert response_b.status_code == 201
    
    # Tenant A tries to access their orders - should only see their order
    response_a_orders = client_a.get(
        "/orders",
        headers={
            "x-tenant-id": "tenant_a",
            "authorization": f"Bearer {tenant_a_token}"
        }
    )
    orders_a = response_a_orders.json()
    assert len(orders_a) == 1
    assert orders_a[0]["id"] == order_a_id
    
    # Tenant A tries to access tenant B's order directly - should fail
    response_forbidden = client_a.get(
        f"/orders/{order_b_id}",
        headers={
            "x-tenant-id": "tenant_a",
            "authorization": f"Bearer {tenant_a_token}"
        }
    )
    assert response_forbidden.status_code == 404  # Not found, not 403
    
    # Tenant A tries to use tenant B's header with their token - should fail
    response_mismatch = client_a.get(
        "/orders",
        headers={
            "x-tenant-id": "tenant_b",  # Wrong tenant!
            "authorization": f"Bearer {tenant_a_token}"  # Tenant A token
        }
    )
    assert response_mismatch.status_code == 403  # Tenant mismatch

def test_cache_isolation():
    """Ensure cached data is tenant-isolated"""
    cache = TenantAwareCache(redis_client)
    
    # Cache data for tenant A
    cache.set("tenant_a", "product:1", {"name": "Pizza", "price": 10.00})
    
    # Cache data for tenant B (same key!)
    cache.set("tenant_b", "product:1", {"name": "Burger", "price": 8.00})
    
    # Verify isolation
    product_a = cache.get("tenant_a", "product:1")
    product_b = cache.get("tenant_b", "product:1")
    
    assert product_a["name"] == "Pizza"
    assert product_b["name"] == "Burger"
```

## The Real Bug: How One Tenant Saw Another's Data

Let me share the exact bug that caused the Friday incident:

```python
# chatbot/orchestrator.py - THE BUG
class ChatbotOrchestrator:
    async def get_dashboard_summary(self, tenant_id: str):
        """Aggregate data from all services"""
        
        # Get orders (CORRECT - includes tenant_id)
        orders = await self.pos_core_service.get_orders(tenant_id)
        
        # Get inventory (BUG - forgot tenant_id!)
        inventory = await self.inventory_service.get_all_products()
        
        # Get payments (CORRECT)
        payments = await self.payment_service.get_payments(tenant_id)
        
        return {
            "orders": orders,
            "inventory": inventory,  # This returned ALL tenants' products!
            "payments": payments
        }

# inventory/service.py - THE MISSING FILTER
class InventoryService:
    def get_all_products(self):
        # BUG: No tenant filtering!
        return self.db.query(Product).all()  # Returns products from ALL tenants

# The fix
class InventoryService:
    def get_all_products(self, tenant_id: str):
        # FIXED: Filter by tenant
        return self.db.query(Product).filter(
            Product.tenant_id == tenant_id
        ).all()
```

The chatbot showed aggregated data, so it wasn't immediately obvious. But when a tenant expanded the inventory section, they saw products from other restaurants.

**Lesson:** When aggregating data from multiple services, verify each service call includes tenant\_id.

## Performance Considerations

Multi-tenancy adds overhead. Here's how to optimize:

### 1. Index Strategy

```sql
-- Composite indexes with tenant_id first
CREATE INDEX ix_orders_tenant_date ON orders(tenant_id, created_at DESC);
CREATE INDEX ix_products_tenant_category ON products(tenant_id, category);

-- Partition large tables by tenant_id (PostgreSQL 10+)
CREATE TABLE orders (
    id INTEGER,
    tenant_id VARCHAR(50),
    order_number VARCHAR(50),
    -- other columns
) PARTITION BY LIST (tenant_id);

-- Create partition per tenant
CREATE TABLE orders_tenant_a PARTITION OF orders FOR VALUES IN ('tenant_a');
CREATE TABLE orders_tenant_b PARTITION OF orders FOR VALUES IN ('tenant_b');
```

### 2. Connection Pooling

```python
# config/database.py
from sqlalchemy.pool import QueuePool

engine = create_engine(
    settings.DATABASE_URL,
    poolclass=QueuePool,
    pool_size=20,  # Shared across all tenants
    max_overflow=40,
    pool_pre_ping=True,
    pool_recycle=3600  # Recycle connections after 1 hour
)

# Don't create separate pools per tenant (unless Strategy 1)
# Shared pool is much more efficient
```

### 3. Query Optimization

```python
# Bad: N+1 query problem
def get_orders_with_products(tenant_id: str):
    orders = db.query(Order).filter(Order.tenant_id == tenant_id).all()
    
    for order in orders:
        # N additional queries!
        order.products = db.query(Product).filter(
            Product.id.in_(order.product_ids),
            Product.tenant_id == tenant_id
        ).all()
    
    return orders

# Good: Single query with join
from sqlalchemy.orm import joinedload

def get_orders_with_products(tenant_id: str):
    return db.query(Order).filter(
        Order.tenant_id == tenant_id
    ).options(
        joinedload(Order.products)
    ).all()
```

## Key Learnings

1. **Tenant isolation must be enforced at every layer**
   * Middleware, services, repositories, database, cache
   * Defense in depth prevents catastrophic leaks
2. **Always verify token tenant matches header tenant**
   * Prevents malicious users from accessing other tenants
   * Simple check, massive security benefit
3. **Use PostgreSQL Row-Level Security as safety net**
   * Catches bugs before they become data breaches
   * Small performance cost, huge security gain
4. **Prefix all cache keys with tenant\_id**
   * Easy to forget, easy to exploit
   * Automated testing is critical
5. **Test with multiple tenants from day one**
   * Don't add multi-tenancy later
   * Design for it from the start

## Common Mistakes

1. **Forgetting tenant filter in queries**
   * Use repository pattern to enforce filtering
   * Enable PostgreSQL RLS
2. **Not validating tenant\_id from JWT**
   * User could send arbitrary tenant\_id in header
   * Always verify against JWT claims
3. **Caching data without tenant prefix**
   * All tenants share Redis
   * Unprefixed keys leak data
4. **Using separate database without connection pooling strategy**
   * Too many connections kill database
   * Need connection pool manager

## When to Use Each Strategy

**Separate Database:**

* Enterprise customers paying premium
* Compliance requirements (HIPAA, GDPR)
* < 100 tenants (connection management gets complex)

**Separate Schemas:**

* Need strong isolation
* Medium tenant count (100-1000)
* Want to balance cost and isolation

**Shared Tables (Discriminator Column):**

* High tenant count (1000+)
* Need best performance
* Can enforce isolation at application layer
* My choice for POS system

## Next Steps

Now that you understand multi-tenant architecture, the next article dives into service layer architecture—how to structure your code within each service for maintainability and testability.

We'll cover:

* Three-layer architecture (Domain, Application, Infrastructure)
* Dependency injection with FastAPI
* Repository pattern implementation
* Why layers matter for testing

**Next Article:** [04-service-layer-architecture.md](https://blog.htunnthuthu.com/architecture-and-design/architecture-and-patterns/software-architecture-101/04-service-layer-architecture) - Learn how to structure services so they're testable, maintainable, and ready to scale.

***

*Remember: Multi-tenancy isn't a feature you add later. It's a fundamental architectural constraint. Design for it from day one, test it thoroughly, and never trust yourself to remember that WHERE clause.*