├── models.py          ← every model, everything mixed
├── views.py           ← all routes
├── services.py        ← all business logic
└── database.py        ← shared session, shared schema

user-service/          ← separate repo, separate DB, separate process
order-service/         ← separate repo, separate DB, separate process
payment-service/       ← separate repo, separate DB, separate process

app/
├── modules/
│   ├── auth/          ← bounded module, owns its models, services, DB schema
│   ├── orders/        ← bounded module, communicates via contracts only
│   └── payments/      ← bounded module
├── shared/            ← truly shared utilities only (logging, config, base classes)
└── main.py

auth module        →   owns: users, sessions, mfa_tokens
billing module     →   owns: billing_accounts, subscriptions, invoices
orders module      →   owns: orders, order_items, fulfillment_status
inventory module   →   owns: products, stock_levels, reservations

modules/
├── auth/
├── billing/
├── orders/
└── inventory/

modules/orders/
├── router.py        ← HTTP: request parsing, response formatting, no business logic
├── service.py       ← Business logic: rules, orchestration, decisions
├── repository.py    ← Data access: all SQL lives here, nothing else
├── schemas.py       ← Pydantic: request/response contracts, NOT ORM models
├── models.py        ← DB models: SQLAlchemy or raw table definitions
└── events.py        ← Domain events this module emits

router → service → repository → database

# billing/service.py — WRONG
from app.modules.auth.services import AuthService

class BillingService:
    def __init__(self):
        self.auth = AuthService()  # direct dependency

# modules/auth/contracts.py — the PUBLIC surface of the auth module
from abc import ABC, abstractmethod
from uuid import UUID
from pydantic import BaseModel

# DTO — what crosses the boundary
class UserIdentity(BaseModel):
    user_id: UUID
    email: str
    is_active: bool

# Interface — what operations are available
class IUserService(ABC):
    @abstractmethod
    async def get_user_identity(self, user_id: UUID) -> UserIdentity | None:
        ...

    @abstractmethod
    async def assert_user_exists(self, user_id: UUID) -> None:
        """Raises UserNotFoundError if the user does not exist."""
        ...

# modules/auth/services.py — the PRIVATE implementation
from .contracts import IUserService, UserIdentity

class UserService(IUserService):
    def __init__(self, repository: IUserRepository):
        self._repo = repository

    async def get_user_identity(self, user_id: UUID) -> UserIdentity | None:
        user = await self._repo.find_by_id(user_id)
        if not user:
            return None
        return UserIdentity(
            user_id=user.id,
            email=user.email,
            is_active=user.is_active
        )

# modules/billing/service.py — depends on the INTERFACE, not the implementation
from app.modules.auth.contracts import IUserService

class BillingService:
    def __init__(self, user_service: IUserService, repository: IBillingRepository):
        self._users = user_service
        self._repo = repository

    async def create_billing_account(self, user_id: UUID) -> BillingAccount:
        await self._users.assert_user_exists(user_id)  # uses the contract
        # ... billing logic

# modules/auth/__init__.py — the module's public surface

from .contracts import IUserService, UserIdentity
from .exceptions import UserNotFoundError, AuthenticationError
from .router import router  # FastAPI router, mounted by main.py

# NOT exported:
# - UserService (implementation detail)
# - UserRepository (implementation detail)
# - User ORM model (internal to this module)
# - Any internal utility functions

__all__ = [
    "IUserService",
    "UserIdentity",
    "UserNotFoundError",
    "AuthenticationError",
    "router",
]

from app.modules.auth.services import UserService  # private

class UserIdentity(BaseModel):
    user_id: UUID
    email: str
    is_active: bool

Infrastructure (DB, HTTP, external APIs)
       ↓
   Adapters (repositories, controllers, event handlers)
       ↓
  Application (services, use cases)
       ↓
    Domain (entities, value objects, interfaces)

# domain layer — no external imports
class IUserRepository(ABC):
    @abstractmethod
    async def find_by_id(self, user_id: UUID) -> User | None: ...

    @abstractmethod
    async def save(self, user: User) -> None: ...

# infrastructure layer — implements the interface
class PostgresUserRepository(IUserRepository):
    def __init__(self, connection: asyncpg.Connection):
        self._conn = connection

    async def find_by_id(self, user_id: UUID) -> User | None:
        row = await self._conn.fetchrow(
            "SELECT id, email, password_hash, is_active FROM auth.users WHERE id = $1",
            user_id
        )
        if not row:
            return None
        return User(id=row["id"], email=row["email"], ...)

# orders/service.py
from app.modules.inventory.contracts import IInventoryService  # orders → inventory

# inventory/service.py
from app.modules.orders.contracts import IOrderService  # inventory → orders — CIRCULAR

app/
├── main.py
├── database.py
├── models.py        ← all ORM models, every domain
├── schemas.py       ← all Pydantic models, every domain
└── routers/
    ├── auth.py
    ├── orders.py
    └── billing.py

app/
├── main.py
├── models/
│   ├── user.py
│   ├── order.py
│   └── invoice.py
├── services/
│   ├── auth_service.py
│   ├── order_service.py
│   └── billing_service.py
├── repositories/
│   ├── user_repository.py
│   └── order_repository.py
└── routers/
    ├── auth.py
    └── orders.py

app/
├── main.py
├── core/                    ← system-wide infrastructure
│   ├── config.py
│   ├── database.py
│   ├── exceptions.py
│   └── logging.py
├── shared/                  ← truly shared data structures only
│   ├── base_repository.py
│   ├── base_schema.py
│   └── types.py
└── modules/
    ├── auth/
    ├── orders/
    ├── billing/
    └── inventory/

app/
├── main.py                        ← app factory, router registration, lifespan
├── core/
│   ├── config.py                  ← pydantic-settings, env vars, never hardcoded values
│   ├── database.py                ← asyncpg pool, connection lifecycle
│   ├── exceptions.py              ← base exception classes for the entire system
│   ├── logging.py                 ← structured JSON logging config
│   └── middleware.py              ← correlation ID injection, request timing
├── shared/
│   ├── base_repository.py         ← abstract base with common query helpers
│   ├── base_schema.py             ← common Pydantic config (camelCase, etc.)
│   └── types.py                   ← UUID aliases, KES money type, etc.
└── modules/
    └── orders/                    ← one module, fully self-contained
        ├── __init__.py            ← public contract surface (exports only)
        ├── contracts.py           ← IOrderService interface + DTOs for other modules
        ├── exceptions.py          ← OrderNotFoundError, InvalidOrderStateError, etc.
        ├── router.py              ← FastAPI router, HTTP only
        ├── service.py             ← business logic, orchestration
        ├── repository.py          ← all SQL for this module
        ├── schemas.py             ← Pydantic request/response models
        ├── models.py              ← DB table definitions (SQLAlchemy or raw)
        ├── events.py              ← domain events this module emits
        ├── dependencies.py        ← FastAPI Depends() factories for this module
        └── migrations/            ← Alembic versions scoped to this module
            └── versions/

HTTP Request
    ↓
Router          ← parse, validate input shape, extract path/query params
    ↓
Service         ← business rules, decisions, orchestration
    ↓
Unit of Work    ← transaction boundary opens here
    ↓
Repository      ← executes SQL, returns domain objects
    ↓
PostgreSQL      ← query execution
    ↓
Repository      ← maps rows to domain objects
    ↓
Unit of Work    ← commit or rollback
    ↓
Service         ← maps domain result to response DTO
    ↓
Router          ← returns HTTP response

# modules/orders/router.py
from uuid import UUID
from fastapi import APIRouter, Depends, status
from .schemas import CreateOrderRequest, OrderResponse
from .service import OrderService
from .dependencies import get_order_service
from .exceptions import OrderNotFoundError

router = APIRouter(prefix="/orders", tags=["orders"])

@router.post("/", response_model=OrderResponse, status_code=status.HTTP_201_CREATED)
async def create_order(
    payload: CreateOrderRequest,          # FastAPI validates shape here
    service: OrderService = Depends(get_order_service),
):
    order = await service.create_order(
        user_id=payload.user_id,
        items=payload.items,
    )
    return OrderResponse.model_validate(order)

@router.get("/{order_id}", response_model=OrderResponse)
async def get_order(
    order_id: UUID,
    service: OrderService = Depends(get_order_service),
):
    order = await service.get_order(order_id)
    if order is None:
        raise OrderNotFoundError(order_id)
    return OrderResponse.model_validate(order)

# modules/orders/schemas.py
from uuid import UUID
from decimal import Decimal
from pydantic import BaseModel, field_validator
from shared.base_schema import AppBaseModel

class OrderItemRequest(AppBaseModel):
    product_id: UUID
    quantity: int

    @field_validator("quantity")
    @classmethod
    def quantity_must_be_positive(cls, v: int) -> int:
        if v <= 0:
            raise ValueError("Quantity must be at least 1")
        return v

class CreateOrderRequest(AppBaseModel):
    user_id: UUID
    items: list[OrderItemRequest]

    @field_validator("items")
    @classmethod
    def must_have_items(cls, v: list) -> list:
        if not v:
            raise ValueError("Order must contain at least one item")
        return v

class OrderItemResponse(AppBaseModel):
    product_id: UUID
    product_name: str
    quantity: int
    unit_price: Decimal
    line_total: Decimal

class OrderResponse(AppBaseModel):
    order_id: UUID
    user_id: UUID
    status: str
    items: list[OrderItemResponse]
    total_amount: Decimal
    created_at: datetime

# modules/orders/service.py
from uuid import UUID
from .repository import OrderRepository
from .schemas import OrderItemRequest
from .models import Order, OrderItem
from .events import OrderPlacedEvent
from app.modules.inventory.contracts import IInventoryService
from app.core.events import EventBus

class OrderService:
    def __init__(
        self,
        repository: OrderRepository,
        inventory: IInventoryService,    # interface, not concrete class
        event_bus: EventBus,
    ):
        self._repo = repository
        self._inventory = inventory
        self._events = event_bus

    async def create_order(
        self,
        user_id: UUID,
        items: list[OrderItemRequest],
    ) -> Order:
        # 1. Validate stock availability (calls inventory contract)
        product_ids = [item.product_id for item in items]
        snapshots = await self._inventory.get_product_snapshots(product_ids)

        # 2. Apply business rules
        order_items = []
        for item in items:
            snapshot = snapshots.get(item.product_id)
            if snapshot is None:
                raise ProductNotFoundError(item.product_id)
            if snapshot.stock_quantity < item.quantity:
                raise InsufficientStockError(item.product_id, item.quantity)

            order_items.append(OrderItem(
                product_id=item.product_id,
                product_name=snapshot.name,     # snapshot stored here
                quantity=item.quantity,
                unit_price=snapshot.price,
            ))

        # 3. Persist — repository handles the transaction
        order = await self._repo.create_order(user_id=user_id, items=order_items)

        # 4. Emit domain event
        await self._events.publish(OrderPlacedEvent(
            order_id=order.id,
            user_id=user_id,
            total_amount=order.total_amount,
        ))

        return order

    async def get_order(self, order_id: UUID) -> Order | None:
        return await self._repo.find_by_id(order_id)

# modules/orders/repository.py
import asyncpg
from uuid import UUID, uuid4
from decimal import Decimal
from .models import Order, OrderItem

class OrderRepository:
    def __init__(self, connection: asyncpg.Connection):
        self._conn = connection

    async def create_order(
        self,
        user_id: UUID,
        items: list[OrderItem],
    ) -> Order:
        order_id = uuid4()
        total = sum(item.unit_price * item.quantity for item in items)

        # Transaction managed by Unit of Work — connection is already in transaction
        await self._conn.execute(
            """
            INSERT INTO orders.orders (id, user_id, status, total_amount, created_at)
            VALUES ($1, $2, 'pending', $3, NOW())
            """,
            order_id, user_id, total
        )

        for item in items:
            await self._conn.execute(
                """
                INSERT INTO orders.order_items
                    (id, order_id, product_id, product_name, quantity, unit_price)
                VALUES ($1, $2, $3, $4, $5, $6)
                """,
                uuid4(), order_id, item.product_id,
                item.product_name, item.quantity, item.unit_price
            )

        return await self.find_by_id(order_id)

    async def find_by_id(self, order_id: UUID) -> Order | None:
        row = await self._conn.fetchrow(
            """
            SELECT
                o.id, o.user_id, o.status, o.total_amount, o.created_at,
                json_agg(json_build_object(
                    'product_id', oi.product_id,
                    'product_name', oi.product_name,
                    'quantity', oi.quantity,
                    'unit_price', oi.unit_price
                )) AS items
            FROM orders.orders o
            JOIN orders.order_items oi ON oi.order_id = o.id
            WHERE o.id = $1
            GROUP BY o.id
            """,
            order_id
        )
        if not row:
            return None
        return Order.from_row(row)

# core/database.py
import asyncpg
from contextlib import asynccontextmanager

class UnitOfWork:
    def __init__(self, pool: asyncpg.Pool):
        self._pool = pool
        self.connection: asyncpg.Connection | None = None

    async def __aenter__(self) -> "UnitOfWork":
        self.connection = await self._pool.acquire()
        self._transaction = self.connection.transaction()
        await self._transaction.start()
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if exc_type:
            await self._transaction.rollback()
        else:
            await self._transaction.commit()
        await self._pool.release(self.connection)

# modules/orders/dependencies.py
from fastapi import Depends
import asyncpg
from app.core.database import get_connection
from app.modules.inventory.dependencies import get_inventory_service
from app.core.events import get_event_bus
from .repository import OrderRepository
from .service import OrderService

async def get_order_service(
    conn: asyncpg.Connection = Depends(get_connection),
    inventory: IInventoryService = Depends(get_inventory_service),
    event_bus: EventBus = Depends(get_event_bus),
) -> OrderService:
    repo = OrderRepository(conn)
    return OrderService(
        repository=repo,
        inventory=inventory,
        event_bus=event_bus,
    )

# core/database.py
async def get_connection(request: Request) -> asyncpg.Connection:
    async with request.app.state.pool.acquire() as conn:
        yield conn

# main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
import asyncpg
from app.core.config import settings
from app.core.middleware import CorrelationIDMiddleware
from app.modules.auth import router as auth_router
from app.modules.orders import router as orders_router
from app.modules.billing import router as billing_router

@asynccontextmanager
async def lifespan(app: FastAPI):
    # startup
    app.state.pool = await asyncpg.create_pool(
        dsn=settings.database_url,
        min_size=5,
        max_size=20,
    )
    yield
    # shutdown
    await app.state.pool.close()

app = FastAPI(
    title="KukuFiti API",
    version="1.0.0",
    lifespan=lifespan,
)

app.add_middleware(CorrelationIDMiddleware)

app.include_router(auth_router, prefix="/api/v1")
app.include_router(orders_router, prefix="/api/v1")
app.include_router(billing_router, prefix="/api/v1")

Depends(get_connection)   ← connection acquired, transaction starts
    ↓
service.create_order()    ← business logic runs inside the transaction
    ↓
repo.create_order()       ← SQL executes inside the transaction
repo.find_by_id()         ← SQL executes inside the same transaction
    ↓
return from handler       ← transaction commits, connection released
exception raised          ← transaction rolls back, connection released

auth_db      ← separate Postgres instance
orders_db    ← separate Postgres instance
billing_db   ← separate Postgres instance

CREATE TABLE users (...);           -- auth
CREATE TABLE auth_sessions (...);   -- auth
CREATE TABLE orders (...);          -- orders
CREATE TABLE order_items (...);     -- orders
CREATE TABLE billing_accounts (...); -- billing

CREATE SCHEMA auth;
CREATE SCHEMA orders;
CREATE SCHEMA billing;
CREATE SCHEMA inventory;

auth.users
auth.sessions
auth.mfa_tokens

orders.orders
orders.order_items
orders.product_snapshots

billing.billing_accounts
billing.subscriptions
billing.invoices

inventory.products
inventory.stock_levels

-- Create schemas
CREATE SCHEMA auth;
CREATE SCHEMA orders;
CREATE SCHEMA billing;
CREATE SCHEMA inventory;

-- Create application roles — one per module
CREATE ROLE auth_app WITH LOGIN PASSWORD 'auth_secret';
CREATE ROLE orders_app WITH LOGIN PASSWORD 'orders_secret';
CREATE ROLE billing_app WITH LOGIN PASSWORD 'billing_secret';
CREATE ROLE inventory_app WITH LOGIN PASSWORD 'inventory_secret';

-- Create a read-only reporting role (cross-schema, SELECT only)
CREATE ROLE reporting_reader WITH LOGIN PASSWORD 'reporting_secret';

-- Grant schema usage — each role can only USE its own schema
GRANT USAGE ON SCHEMA auth TO auth_app;
GRANT USAGE ON SCHEMA orders TO orders_app;
GRANT USAGE ON SCHEMA billing TO billing_app;
GRANT USAGE ON SCHEMA inventory TO inventory_app;

-- Grant table-level privileges within each schema
GRANT SELECT, INSERT, UPDATE, DELETE
    ON ALL TABLES IN SCHEMA auth TO auth_app;
GRANT SELECT, INSERT, UPDATE, DELETE
    ON ALL TABLES IN SCHEMA orders TO orders_app;
GRANT SELECT, INSERT, UPDATE, DELETE
    ON ALL TABLES IN SCHEMA billing TO billing_app;
GRANT SELECT, INSERT, UPDATE, DELETE
    ON ALL TABLES IN SCHEMA inventory TO inventory_app;

-- Ensure future tables in each schema are also covered
ALTER DEFAULT PRIVILEGES IN SCHEMA auth
    GRANT SELECT, INSERT, UPDATE, DELETE ON TABLES TO auth_app;
ALTER DEFAULT PRIVILEGES IN SCHEMA orders
    GRANT SELECT, INSERT, UPDATE, DELETE ON TABLES TO orders_app;

-- Reporting role gets SELECT across all schemas
GRANT USAGE ON SCHEMA auth, orders, billing, inventory TO reporting_reader;
GRANT SELECT ON ALL TABLES IN SCHEMA auth TO reporting_reader;
GRANT SELECT ON ALL TABLES IN SCHEMA orders TO reporting_reader;
GRANT SELECT ON ALL TABLES IN SCHEMA billing TO reporting_reader;
GRANT SELECT ON ALL TABLES IN SCHEMA inventory TO reporting_reader;

SELECT * FROM auth.users WHERE id = $1;

ERROR: permission denied for schema auth

# core/config.py
class Settings(BaseSettings):
    auth_db_url: str      # postgres://auth_app:auth_secret@localhost/appdb
    orders_db_url: str    # postgres://orders_app:orders_secret@localhost/appdb
    billing_db_url: str
    inventory_db_url: str

    model_config = SettingsConfigDict(env_file=".env")

-- orders schema referencing auth schema
ALTER TABLE orders.orders
    ADD CONSTRAINT fk_user
    FOREIGN KEY (user_id) REFERENCES auth.users(id);

-- orders schema
CREATE TABLE orders.orders (
    id          UUID        PRIMARY KEY DEFAULT gen_random_uuid(),
    user_id     UUID        NOT NULL,   -- soft reference: no FK to auth.users
    status      TEXT        NOT NULL DEFAULT 'pending',
    total_amount NUMERIC(12,2) NOT NULL,
    created_at  TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at  TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

# orders/service.py
async def create_order(self, user_id: UUID, items: list[OrderItemRequest]) -> Order:
    # Application-layer integrity check — replaces the FK constraint
    identity = await self._user_service.get_user_identity(user_id)
    if identity is None:
        raise UserNotFoundError(user_id)
    if not identity.is_active:
        raise UserInactiveError(user_id)
    # ... proceed with order creation

ALTER TABLE auth.users
    ADD COLUMN deleted_at TIMESTAMPTZ DEFAULT NULL;

-- "Delete" a user
UPDATE auth.users
    SET deleted_at = NOW()
WHERE id = $1;

-- Query active users only
SELECT * FROM auth.users WHERE deleted_at IS NULL;

# orders/event_handlers.py
async def handle_user_deleted(event: UserDeletedEvent, repo: OrderRepository):
    # Cancel all pending orders for the deleted user
    await repo.cancel_orders_for_user(event.user_id)
    # Anonymize completed orders (GDPR compliance)
    await repo.anonymize_user_data(event.user_id)

async def get_order_with_user(self, order_id: UUID) -> OrderDetail:
    order = await self._repo.find_by_id(order_id)
    user = await self._user_service.get_user_identity(order.user_id)

    return OrderDetail(
        order=order,
        user=user,          # None if user deleted — handled gracefully in response
    )

# auth/service.py
async def delete_user(self, user_id: UUID) -> None:
    # Check with other modules via their contracts
    has_active_orders = await self._order_service.has_active_orders(user_id)
    has_outstanding_balance = await self._billing_service.has_outstanding_balance(user_id)

    if has_active_orders:
        raise CannotDeleteUserError("User has active orders")
    if has_outstanding_balance:
        raise CannotDeleteUserError("User has outstanding balance")

    # Safe to soft-delete
    await self._repo.soft_delete(user_id)
    await self._events.publish(UserDeletedEvent(user_id=user_id))

-- orders.orders — soft reference to user
CREATE INDEX idx_orders_user_id ON orders.orders(user_id);

-- orders.orders — status filtering (very common query pattern)
CREATE INDEX idx_orders_status ON orders.orders(status);

-- orders.order_items — joining back to parent order
CREATE INDEX idx_order_items_order_id ON orders.order_items(order_id);

-- Composite — common query: "all pending orders for a user"
CREATE INDEX idx_orders_user_status ON orders.orders(user_id, status);

-- Uses the composite index (leftmost prefix rule)
SELECT * FROM orders.orders WHERE user_id = $1;

-- Also uses the composite index
SELECT * FROM orders.orders WHERE user_id = $1 AND status = 'pending';

-- Does NOT use the composite index efficiently
SELECT * FROM orders.orders WHERE status = 'pending';
-- This needs its own index: idx_orders_status

-- Partial index — only indexes rows where deleted_at IS NULL
CREATE INDEX idx_users_active ON auth.users(email)
    WHERE deleted_at IS NULL;

-- Partial index for pending orders only
-- If 95% of orders are completed, this index is small and fast
CREATE INDEX idx_orders_pending ON orders.orders(user_id, created_at)
    WHERE status = 'pending';

-- Query: fetch order summary for a user (id, status, total, created_at)
-- Covering index includes all projected columns
CREATE INDEX idx_orders_user_summary
    ON orders.orders(user_id, created_at DESC)
    INCLUDE (id, status, total_amount);

SELECT id, status, total_amount, created_at
FROM orders.orders
WHERE user_id = $1
ORDER BY created_at DESC
LIMIT 20;

EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT)
SELECT id, status, total_amount
FROM orders.orders
WHERE user_id = '550e8400-e29b-41d4-a716-446655440000'
  AND status = 'pending';

# Single record fetch
async def find_batch_by_id(self, batch_id: UUID) -> Batch | None:
    row = await self._conn.fetchrow(
        "SELECT id, farm_id, breed, chick_count, placement_date, status "
        "FROM batches.batches WHERE id = $1 AND deleted_at IS NULL",
        batch_id
    )
    return Batch.from_row(row) if row else None

# List with filtering and pagination
async def list_batches(
    self,
    farm_id: UUID,
    status: str | None,
    limit: int,
    offset: int,
) -> list[Batch]:
    query = """
        SELECT id, farm_id, breed, chick_count, placement_date, status
        FROM batches.batches
        WHERE farm_id = $1
          AND deleted_at IS NULL
          AND ($2::text IS NULL OR status = $2)
        ORDER BY placement_date DESC
        LIMIT $3 OFFSET $4
    """
    rows = await self._conn.fetch(query, farm_id, status, limit, offset)
    return [Batch.from_row(row) for row in rows]

# Financial aggregation — never use ORM for this
async def get_batch_financial_summary(self, farm_id: UUID) -> FinancialSummary:
    row = await self._conn.fetchrow(
        """
        SELECT
            COUNT(*) FILTER (WHERE status = 'active')     AS active_batches,
            COUNT(*) FILTER (WHERE status = 'completed')  AS completed_batches,
            SUM(total_revenue) FILTER (WHERE status = 'completed') AS total_revenue,
            SUM(total_cost)                               AS total_cost,
            AVG(fcr) FILTER (WHERE fcr IS NOT NULL)       AS avg_fcr
        FROM batches.batches
        WHERE farm_id = $1
          AND deleted_at IS NULL
        """,
        farm_id
    )
    return FinancialSummary.from_row(row)

# Bulk insert — executemany for efficiency
async def record_mortality_events(
    self,
    events: list[MortalityEvent],
) -> None:
    await self._conn.executemany(
        """
        INSERT INTO health.mortality_events
            (id, batch_id, event_date, count, cause, notes)
        VALUES ($1, $2, $3, $4, $5, $6)
        """,
        [
            (e.id, e.batch_id, e.event_date, e.count, e.cause, e.notes)
            for e in events
        ]
    )

-- Create materialized view: farm performance dashboard
CREATE MATERIALIZED VIEW batches.farm_performance_summary AS
SELECT
    b.farm_id,
    DATE_TRUNC('month', b.placement_date)    AS month,
    COUNT(*)                                  AS total_batches,
    SUM(b.chick_count)                        AS total_chicks,
    SUM(f.total_revenue)                      AS revenue,
    SUM(f.total_cost)                         AS cost,
    SUM(f.total_revenue) - SUM(f.total_cost)  AS profit,
    AVG(b.fcr)                                AS avg_fcr,
    AVG(b.mortality_rate)                     AS avg_mortality_rate
FROM batches.batches b
JOIN finance.batch_financials f ON f.batch_id = b.id
WHERE b.deleted_at IS NULL
GROUP BY b.farm_id, DATE_TRUNC('month', b.placement_date)
WITH DATA;

-- Index the materialized view
CREATE INDEX idx_farm_perf_farm_month
    ON batches.farm_performance_summary(farm_id, month DESC);

-- Refresh — run via cron or background worker
REFRESH MATERIALIZED VIEW CONCURRENTLY batches.farm_performance_summary;

# service.py — N+1 PROBLEM
async def list_orders_with_users(self, farm_id: UUID) -> list[OrderDetail]:
    orders = await self._order_repo.list_orders(farm_id)  # 1 query → returns 50 orders

    result = []
    for order in orders:
        # 1 query PER ORDER — 50 orders = 50 additional queries
        user = await self._user_service.get_user_identity(order.user_id)
        result.append(OrderDetail(order=order, user=user))

    return result

# auth/contracts.py
class IUserService(ABC):
    @abstractmethod
    async def get_user_identities(
        self,
        user_ids: list[UUID]
    ) -> dict[UUID, UserIdentity]:
        """Returns a dict mapping user_id → UserIdentity for all found users."""
        ...

# auth/service.py
async def get_user_identities(
    self,
    user_ids: list[UUID]
) -> dict[UUID, UserIdentity]:
    rows = await self._conn.fetch(
        "SELECT id, email, is_active FROM auth.users WHERE id = ANY($1::uuid[])",
        user_ids
    )
    return {
        row["id"]: UserIdentity(
            user_id=row["id"],
            email=row["email"],
            is_active=row["is_active"]
        )
        for row in rows
    }

# orders/service.py — FIXED: 2 queries total regardless of list size
async def list_orders_with_users(self, farm_id: UUID) -> list[OrderDetail]:
    orders = await self._order_repo.list_orders(farm_id)  # 1 query

    user_ids = list({order.user_id for order in orders})
    user_map = await self._user_service.get_user_identities(user_ids)  # 1 query

    return [
        OrderDetail(order=order, user=user_map.get(order.user_id))
        for order in orders
    ]

-- orders module stores user display name at order creation time
-- No need to query auth module when rendering order history
CREATE TABLE orders.orders (
    id              UUID PRIMARY KEY,
    user_id         UUID NOT NULL,
    user_email      TEXT NOT NULL,    -- denormalized from auth at creation time
    status          TEXT NOT NULL,
    total_amount    NUMERIC(12,2) NOT NULL,
    created_at      TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

# orders/service.py
async def list_orders_with_users(self, farm_id: UUID) -> list[OrderDetail]:
    orders = await self._order_repo.list_orders(farm_id)

    result = []
    for order in orders:
        # Check cache first — Redis with 5-minute TTL
        cache_key = f"user_identity:{order.user_id}"
        cached = await self._cache.get(cache_key)

        if cached:
            user = UserIdentity.model_validate_json(cached)
        else:
            user = await self._user_service.get_user_identity(order.user_id)
            await self._cache.set(cache_key, user.model_dump_json(), ttl=300)

        result.append(OrderDetail(order=order, user=user))

    return result

-- default — one table for entire system
public.alembic_version

auth.alembic_version       -- migration history for auth schema only
orders.alembic_version     -- migration history for orders schema only
billing.alembic_version    -- migration history for billing schema only

# alembic/env.py
from logging.config import fileConfig
from sqlalchemy import engine_from_config, pool, text
from alembic import context
import os

config = context.config
fileConfig(config.config_file_name)

# Determine which module we're migrating
MODULE = os.environ.get("MIGRATION_MODULE", "orders")
SCHEMA = MODULE

# Module-specific connection string
DB_URLS = {
    "auth": os.environ["AUTH_DB_URL"],
    "orders": os.environ["ORDERS_DB_URL"],
    "billing": os.environ["BILLING_DB_URL"],
    "inventory": os.environ["INVENTORY_DB_URL"],
}

def run_migrations_online():
    connectable = engine_from_config(
        {"sqlalchemy.url": DB_URLS[MODULE]},
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

    with connectable.connect() as connection:
        # Ensure schema exists
        connection.execute(text(f"CREATE SCHEMA IF NOT EXISTS {SCHEMA}"))

        context.configure(
            connection=connection,
            target_metadata=target_metadata,
            version_table="alembic_version",
            version_table_schema=SCHEMA,    # version table lives in module's schema
            include_schemas=True,
        )

        with context.begin_transaction():
            context.run_migrations()

# Migrate only the orders module
MIGRATION_MODULE=orders alembic upgrade head

# Create a new migration for the batches module
MIGRATION_MODULE=batches alembic revision --autogenerate -m "add_mortality_threshold_column"

# Rollback one step for billing only
MIGRATION_MODULE=billing alembic downgrade -1

# modules/batches/migrations/versions/001_create_batches_schema.py
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects.postgresql import UUID

def upgrade():
    # Ensure schema exists
    op.execute("CREATE SCHEMA IF NOT EXISTS batches")

    op.create_table(
        "batches",
        sa.Column("id", UUID(as_uuid=True), primary_key=True,
                  server_default=sa.text("gen_random_uuid()")),
        sa.Column("farm_id", UUID(as_uuid=True), nullable=False),
        sa.Column("breed", sa.Text(), nullable=False),
        sa.Column("chick_count", sa.Integer(), nullable=False),
        sa.Column("placement_date", sa.Date(), nullable=False),
        sa.Column("status", sa.Text(), nullable=False,
                  server_default=sa.text("'active'")),
        sa.Column("mortality_rate", sa.Numeric(5, 2), nullable=True),
        sa.Column("fcr", sa.Numeric(5, 3), nullable=True),
        sa.Column("deleted_at", sa.TIMESTAMP(timezone=True), nullable=True),
        sa.Column("created_at", sa.TIMESTAMP(timezone=True), nullable=False,
                  server_default=sa.text("NOW()")),
        sa.Column("updated_at", sa.TIMESTAMP(timezone=True), nullable=False,
                  server_default=sa.text("NOW()")),
        schema="batches",
    )

    op.create_index(
        "idx_batches_farm_id",
        "batches",
        ["farm_id"],
        schema="batches",
    )
    op.create_index(
        "idx_batches_farm_status",
        "batches",
        ["farm_id", "status"],
        schema="batches",
        postgresql_where=sa.text("deleted_at IS NULL"),
    )

def downgrade():
    op.drop_table("batches", schema="batches")
    op.execute("DROP SCHEMA IF EXISTS batches CASCADE")

ALTER TABLE orders.orders OWNER TO orders_app;

# orders/service.py
class OrderService:
    def __init__(
        self,
        repository: OrderRepository,
        inventory: IInventoryService,   # contract — not concrete class
        billing: IBillingService,       # contract — not concrete class
        notification: INotificationService,
    ):
        self._repo = repository
        self._inventory = inventory
        self._billing = billing
        self._notification = notification

    async def create_order(self, user_id: UUID, items: list[OrderItemRequest]) -> Order:
        # Step 1 — validate and reserve stock (inventory module)
        reservation = await self._inventory.reserve_stock(
            items=[StockReservationItem(product_id=i.product_id, quantity=i.quantity)
                   for i in items]
        )

        # Step 2 — persist the order
        order = await self._repo.create_order(user_id=user_id, items=items,
                                               reservation_id=reservation.id)

        # Step 3 — record billing transaction
        await self._billing.record_transaction(
            user_id=user_id,
            order_id=order.id,
            amount=order.total_amount,
        )

        # Step 4 — send confirmation notification
        await self._notification.send_order_confirmation(
            user_id=user_id,
            order_id=order.id,
        )

        return order

# Correct use of direct call — reading before deciding
identity = await self._user_service.get_user_identity(user_id)
if not identity or not identity.is_active:
    raise UserInactiveError(user_id)

async def create_order(self, ...):
    order = await self._repo.create_order(...)        # DB write — succeeds
    await self._billing.record_transaction(...)        # DB write — succeeds
    await self._notification.send_order_confirmation(  # calls external email API
        user_id=user_id,
        order_id=order.id,
    )                                                  # external API is down — raises
    # What happens? The exception propagates upward.
    # If the Unit of Work catches it, the entire order is rolled back.
    # The user gets a 500. Their order does not exist.
    # The email was the problem. The order should have succeeded.

# core/events.py
from dataclasses import dataclass, field
from uuid import UUID
from typing import Any, Callable, Awaitable
from collections import defaultdict

@dataclass
class DomainEvent:
    event_id: UUID = field(default_factory=uuid4)
    occurred_at: datetime = field(default_factory=lambda: datetime.now(UTC))

@dataclass
class OrderPlacedEvent(DomainEvent):
    order_id: UUID
    user_id: UUID
    total_amount: Decimal
    items: list[dict]

class InMemoryEventBus:
    def __init__(self):
        self._handlers: dict[type, list[Callable]] = defaultdict(list)

    def subscribe(self, event_type: type, handler: Callable[..., Awaitable[None]]):
        self._handlers[event_type].append(handler)

    async def publish(self, event: DomainEvent) -> None:
        handlers = self._handlers.get(type(event), [])
        for handler in handlers:
            await handler(event)

# main.py — wire up subscriptions at startup
from app.modules.billing.event_handlers import handle_order_placed_billing
from app.modules.notifications.event_handlers import handle_order_placed_notification
from app.modules.inventory.event_handlers import handle_order_placed_inventory

event_bus = InMemoryEventBus()
event_bus.subscribe(OrderPlacedEvent, handle_order_placed_billing)
event_bus.subscribe(OrderPlacedEvent, handle_order_placed_notification)
event_bus.subscribe(OrderPlacedEvent, handle_order_placed_inventory)

# orders/service.py — no direct knowledge of billing or notifications
async def create_order(self, user_id: UUID, items: list[OrderItemRequest]) -> Order:
    order = await self._repo.create_order(user_id=user_id, items=items)

    await self._event_bus.publish(OrderPlacedEvent(
        order_id=order.id,
        user_id=user_id,
        total_amount=order.total_amount,
        items=[{"product_id": str(i.product_id), "quantity": i.quantity}
               for i in items],
    ))

    return order

1. Order created in database         ✓
2. OrderPlacedEvent published        ✓
3. billing handler starts            ✓
4. Process crashes (OOM, deploy)     ✗
5. billing handler never completes   ✗
6. Order exists, billing not recorded ✗

# Dangerous — two separate writes, no atomicity
async def create_order(self, ...):
    await self._repo.create_order(...)          # Write 1: DB commit
    await self._event_bus.publish(OrderPlaced)  # Write 2: in-memory / message broker
    # Crash here: order exists, event never delivered

# Safe — single atomic transaction
async def create_order(self, ...):
    async with self._uow:
        order = await self._repo.create_order(...)           # Write 1
        await self._outbox.enqueue(OrderPlacedEvent(...))    # Write 2 — same transaction
    # Both committed atomically, or both rolled back
    # Worker will deliver the event from the outbox

CREATE TABLE core.outbox (
    id              UUID            PRIMARY KEY DEFAULT gen_random_uuid(),
    event_type      TEXT            NOT NULL,
    payload         JSONB           NOT NULL,
    status          TEXT            NOT NULL DEFAULT 'pending',
    -- pending → processing → delivered | dead
    created_at      TIMESTAMPTZ     NOT NULL DEFAULT NOW(),
    scheduled_at    TIMESTAMPTZ     NOT NULL DEFAULT NOW(),
    processed_at    TIMESTAMPTZ,
    last_attempt_at TIMESTAMPTZ,
    attempt_count   INTEGER         NOT NULL DEFAULT 0,
    max_attempts    INTEGER         NOT NULL DEFAULT 5,
    last_error      TEXT,
    idempotency_key TEXT            UNIQUE      -- prevents duplicate event delivery
);

-- Worker queries this index constantly — must be fast
CREATE INDEX idx_outbox_pending
    ON core.outbox(scheduled_at ASC)
    WHERE status = 'pending';

-- Monitor dead events
CREATE INDEX idx_outbox_dead
    ON core.outbox(created_at DESC)
    WHERE status = 'dead';

# core/outbox.py
import json
from uuid import UUID, uuid4
from datetime import datetime, UTC
import asyncpg
from app.core.events import DomainEvent

class OutboxWriter:
    def __init__(self, connection: asyncpg.Connection):
        self._conn = connection

    async def enqueue(
        self,
        event: DomainEvent,
        idempotency_key: str | None = None,
        delay_seconds: int = 0,
    ) -> None:
        scheduled_at = datetime.now(UTC)
        if delay_seconds:
            from datetime import timedelta
            scheduled_at += timedelta(seconds=delay_seconds)

        await self._conn.execute(
            """
            INSERT INTO core.outbox
                (id, event_type, payload, status, scheduled_at, idempotency_key)
            VALUES ($1, $2, $3, 'pending', $4, $5)
            ON CONFLICT (idempotency_key) DO NOTHING
            """,
            event.event_id,
            type(event).__name__,
            json.dumps(event.__dict__, default=str),
            scheduled_at,
            idempotency_key or str(event.event_id),
        )

# workers/outbox_worker.py
import asyncio
import asyncpg
import json
import logging
from datetime import datetime, UTC, timedelta
from app.core.config import settings
from app.core.event_registry import EVENT_REGISTRY  # maps event_type → handler

logger = logging.getLogger(__name__)

BATCH_SIZE = 50
POLL_INTERVAL_SECONDS = 2
LOCK_TIMEOUT_SECONDS = 30

async def process_outbox(pool: asyncpg.Pool) -> None:
    """Main worker loop — polls indefinitely."""
    while True:
        try:
            processed = await process_batch(pool)
            if processed == 0:
                await asyncio.sleep(POLL_INTERVAL_SECONDS)
        except Exception as e:
            logger.error("Outbox worker error", exc_info=e)
            await asyncio.sleep(POLL_INTERVAL_SECONDS)

async def process_batch(pool: asyncpg.Pool) -> int:
    async with pool.acquire() as conn:
        # SELECT FOR UPDATE SKIP LOCKED — critical for multiple worker instances
        # Each worker claims its own batch, no two workers process the same row
        rows = await conn.fetch(
            """
            SELECT id, event_type, payload, attempt_count
            FROM core.outbox
            WHERE status = 'pending'
              AND scheduled_at <= NOW()
            ORDER BY scheduled_at ASC
            LIMIT $1
            FOR UPDATE SKIP LOCKED
            """,
            BATCH_SIZE,
        )

        if not rows:
            return 0

        for row in rows:
            await process_single_event(conn, row)

        return len(rows)

async def process_single_event(conn: asyncpg.Connection, row: asyncpg.Record) -> None:
    event_id = row["id"]
    event_type = row["event_type"]
    payload = json.loads(row["payload"])
    attempt_count = row["attempt_count"]

    # Mark as processing immediately — prevents other workers from claiming it
    await conn.execute(
        "UPDATE core.outbox SET status = 'processing', last_attempt_at = NOW(), "
        "attempt_count = attempt_count + 1 WHERE id = $1",
        event_id,
    )

    try:
        handler = EVENT_REGISTRY.get(event_type)
        if handler is None:
            logger.warning(f"No handler registered for event type: {event_type}")
            await mark_delivered(conn, event_id)
            return

        await handler(payload)
        await mark_delivered(conn, event_id)
        logger.info(f"Delivered event {event_type} [{event_id}]")

    except Exception as e:
        logger.error(f"Failed to deliver {event_type} [{event_id}]: {e}")
        next_attempt = attempt_count + 1
        max_attempts = row["max_attempts"] if "max_attempts" in row.keys() else 5

        if next_attempt >= max_attempts:
            # Move to dead letter — requires human intervention
            await conn.execute(
                "UPDATE core.outbox SET status = 'dead', last_error = $2 WHERE id = $1",
                event_id, str(e),
            )
            logger.critical(f"Event {event_type} [{event_id}] moved to dead letter "
                          f"after {next_attempt} attempts")
        else:
            # Exponential backoff — 2^attempt_count * 10 seconds
            backoff = min(2 ** next_attempt * 10, 3600)
            await conn.execute(
                """
                UPDATE core.outbox
                SET status = 'pending',
                    scheduled_at = NOW() + ($2 || ' seconds')::interval,
                    last_error = $3
                WHERE id = $1
                """,
                event_id, str(backoff), str(e),
            )

async def mark_delivered(conn: asyncpg.Connection, event_id: UUID) -> None:
    await conn.execute(
        "UPDATE core.outbox SET status = 'delivered', processed_at = NOW() WHERE id = $1",
        event_id,
    )

# core/event_registry.py
from typing import Callable, Awaitable
from app.modules.billing.event_handlers import handle_order_placed_billing
from app.modules.notifications.event_handlers import handle_order_placed_notification
from app.modules.inventory.event_handlers import handle_order_placed_inventory

EVENT_REGISTRY: dict[str, Callable[[dict], Awaitable[None]]] = {
    "OrderPlacedEvent": handle_order_placed_billing,
    # Fan-out: multiple handlers per event type use a wrapper
    "OrderPlacedEvent_notifications": handle_order_placed_notification,
    "OrderPlacedEvent_inventory": handle_order_placed_inventory,
}

# core/outbox.py
async def enqueue_fan_out(
    self,
    event: DomainEvent,
    handler_names: list[str],
) -> None:
    """Enqueue one outbox row per handler — each processes independently."""
    for handler_name in handler_names:
        await self._conn.execute(
            """
            INSERT INTO core.outbox
                (id, event_type, payload, status, idempotency_key)
            VALUES ($1, $2, $3, 'pending', $4)
            ON CONFLICT (idempotency_key) DO NOTHING
            """,
            uuid4(),
            f"{type(event).__name__}_{handler_name}",
            json.dumps(event.__dict__, default=str),
            f"{event.event_id}_{handler_name}",
        )

1. create_order() begins
   ├── Transaction opens
   ├── INSERT INTO orders.orders          ← domain write
   ├── INSERT INTO core.outbox            ← event write (same transaction)
   └── Transaction commits

2. Worker polls core.outbox
   ├── SELECT ... FOR UPDATE SKIP LOCKED  ← claims the row
   ├── UPDATE status = 'processing'
   └── Calls handler

3a. Handler succeeds
    └── UPDATE status = 'delivered'

3b. Handler fails (transient — network down)
    └── UPDATE status = 'pending', scheduled_at = NOW() + backoff
    └── Worker retries after backoff

3c. Handler fails repeatedly (max_attempts reached)
    └── UPDATE status = 'dead'
    └── Alert fires → human investigation

# billing/service.py — WITHOUT ACL
from app.modules.auth.contracts import UserIdentity

async def calculate_discount(self, user: UserIdentity) -> Decimal:
    # Billing is now making decisions based on auth's data structures
    # auth's "is_active" and "email" are auth concepts, not billing concepts
    if user.is_active and "@enterprise.com" in user.email:
        return Decimal("0.20")  # 20% enterprise discount
    return Decimal("0.00")

# billing/acl.py — the Anti-Corruption Layer
from dataclasses import dataclass
from app.modules.auth.contracts import UserIdentity

@dataclass(frozen=True)
class BillingCustomer:
    """
    Billing's local representation of a customer.
    Expressed entirely in billing domain language.
    This is NOT auth.UserIdentity — it is a billing concept.
    """
    customer_id: UUID
    account_tier: str        # "standard" | "enterprise" | "trial"
    is_billing_eligible: bool

class UserIdentityACL:
    """
    Translates auth.UserIdentity into billing.BillingCustomer.
    All auth→billing translation lives here and only here.
    """

    @staticmethod
    def translate(identity: UserIdentity) -> BillingCustomer:
        # Translation logic — auth concepts converted to billing concepts
        account_tier = UserIdentityACL._determine_tier(identity)
        is_eligible = identity.is_active and account_tier != "suspended"

        return BillingCustomer(
            customer_id=identity.user_id,
            account_tier=account_tier,
            is_billing_eligible=is_eligible,
        )

    @staticmethod
    def _determine_tier(identity: UserIdentity) -> str:
        # Tier determination lives in billing domain, not auth domain
        if identity.email.endswith("@enterprise.com"):
            return "enterprise"
        if identity.created_at and (datetime.now(UTC) - identity.created_at).days < 30:
            return "trial"
        return "standard"

# billing/service.py — WITH ACL
from .acl import UserIdentityACL, BillingCustomer

class BillingService:
    async def calculate_discount(self, user_id: UUID) -> Decimal:
        # Fetch from auth via contract
        identity = await self._user_service.get_user_identity(user_id)
        if identity is None:
            raise CustomerNotFoundError(user_id)

        # Translate through ACL — now working with billing domain concepts
        customer = UserIdentityACL.translate(identity)

        if not customer.is_billing_eligible:
            raise CustomerNotEligibleError(user_id)

        return self._discount_policy.apply(customer.account_tier)

# billing/event_handlers.py
from .acl import OrderPlacedACL

async def handle_order_placed_billing(payload: dict) -> None:
    # Translate external event payload into billing's own representation
    billing_event = OrderPlacedACL.from_payload(payload)

    # Handler works entirely in billing domain language
    await billing_service.record_transaction(
        customer_id=billing_event.customer_id,
        reference=billing_event.billing_reference,
        amount=billing_event.billable_amount,
        currency=billing_event.currency,
    )

# billing/acl.py
@dataclass(frozen=True)
class BillingOrderEvent:
    """Billing's local interpretation of an order event."""
    customer_id: UUID
    billing_reference: str     # billing's name for this concept
    billable_amount: Decimal
    currency: str

class OrderPlacedACL:
    @staticmethod
    def from_payload(payload: dict) -> BillingOrderEvent:
        return BillingOrderEvent(
            customer_id=UUID(payload["user_id"]),
            billing_reference=f"ORD-{payload['order_id'][:8].upper()}",
            billable_amount=Decimal(str(payload["total_amount"])),
            currency=payload.get("currency", "KES"),
        )

# BAD — transaction management scattered across service methods
class OrderService:
    async def create_order(self, user_id: UUID, items: list) -> Order:
        conn = await self._pool.acquire()
        try:
            await conn.execute("BEGIN")
            order = await self._repo.create_order(conn, user_id, items)
            await self._outbox.enqueue(conn, OrderPlacedEvent(...))
            await conn.execute("COMMIT")
            return order
        except Exception:
            await conn.execute("ROLLBACK")
            raise
        finally:
            await self._pool.release(conn)

# core/unit_of_work.py
from __future__ import annotations
import asyncpg
from contextlib import asynccontextmanager
from typing import AsyncGenerator

class UnitOfWork:
    """
    Manages a single database connection and transaction.
    One UoW per request. All repositories share the same connection.
    Commit/rollback handled automatically by context manager.
    """

    def __init__(self, pool: asyncpg.Pool):
        self._pool = pool
        self._connection: asyncpg.Connection | None = None
        self._transaction: asyncpg.transaction.Transaction | None = None

    @property
    def connection(self) -> asyncpg.Connection:
        if self._connection is None:
            raise RuntimeError(
                "UnitOfWork not entered. Use 'async with UnitOfWork(pool) as uow'"
            )
        return self._connection

    async def __aenter__(self) -> UnitOfWork:
        self._connection = await self._pool.acquire()
        self._transaction = self._connection.transaction()
        await self._transaction.start()
        return self

    async def __aexit__(
        self,
        exc_type: type | None,
        exc_val: Exception | None,
        exc_tb,
    ) -> None:
        try:
            if exc_type is not None:
                # Any exception — rollback unconditionally
                await self._transaction.rollback()
            else:
                await self._transaction.commit()
        finally:
            # Connection always released — even if commit/rollback fails
            await self._pool.release(self._connection)
            self._connection = None
            self._transaction = None