🏛️ Domain Driven Design Pattern¶

# ❌ Problem: Anemic domain model - just a data bag
class Order:
    def __init__(self):
        self.id = None
        self.customer_id = None
        self.items = []
        self.total = 0
        self.status = "pending"
        # ❌ No behavior, just properties

# ❌ Business logic scattered in service
class OrderService:
    async def place_order(self, order_data: dict):
        # ❌ Business rules in service layer
        order = Order()
        order.customer_id = order_data["customer_id"]
        order.items = order_data["items"]

        # ❌ Total calculation logic here
        subtotal = sum(item["price"] * item["quantity"] for item in order.items)
        tax = subtotal * 0.08
        order.total = subtotal + tax

        # ❌ Validation logic here
        if order.total > 1000:
            raise ValueError("Order exceeds maximum amount")

        # ❌ Business rule enforcement here
        if len(order.items) == 0:
            raise ValueError("Order must have items")

        await self._db.save(order)

        # ❌ Events created manually, not from domain
        await self._event_bus.publish({"type": "OrderPlaced", "order_id": order.id})

# ❌ Different service duplicates same logic
class ReportingService:
    async def calculate_revenue(self, orders: List[Order]):
        # ❌ Duplicating total calculation logic
        total_revenue = 0
        for order in orders:
            subtotal = sum(item["price"] * item["quantity"] for item in order.items)
            tax = subtotal * 0.08
            total_revenue += subtotal + tax
        return total_revenue

# ✅ Solution: Rich domain model with behavior
from neuroglia.data.abstractions import Entity
from decimal import Decimal

class Order(Entity):
    """Rich domain entity with business logic and validation"""

    def __init__(self, customer_id: str, items: List[OrderItem]):
        super().__init__()

        # ✅ Business rule validation at construction
        if not items:
            raise ValueError("Order must contain at least one item")

        self.customer_id = customer_id
        self.items = items
        self.status = OrderStatus.PENDING
        self.total = self._calculate_total()  # ✅ Encapsulated calculation

        # ✅ Business rule enforcement
        if self.total > Decimal("1000.00"):
            raise ValueError("Order exceeds maximum allowed amount")

        # ✅ Domain event automatically raised
        self.raise_event(OrderPlacedEvent(
            order_id=self.id,
            customer_id=customer_id,
            total=self.total,
            items=[item.to_dto() for item in items]
        ))

    def _calculate_total(self) -> Decimal:
        """✅ Business rule: Calculate total with tax"""
        subtotal = sum(item.price * item.quantity for item in self.items)
        tax = subtotal * Decimal("0.08")  # 8% tax rate
        return subtotal + tax

    def add_item(self, item: OrderItem):
        """✅ Business operation with validation"""
        if self.status != OrderStatus.PENDING:
            raise InvalidOperationError("Cannot modify confirmed order")

        # ✅ Check business constraint
        new_total = self.total + (item.price * item.quantity * Decimal("1.08"))
        if new_total > Decimal("1000.00"):
            raise ValueError("Adding item would exceed maximum order amount")

        self.items.append(item)
        self.total = self._calculate_total()

        # ✅ Domain event for business occurrence
        self.raise_event(OrderItemAddedEvent(
            order_id=self.id,
            item=item.to_dto()
        ))

    def confirm(self, payment_transaction_id: str):
        """✅ Business workflow encapsulated"""
        if self.status != OrderStatus.PENDING:
            raise InvalidOperationError(f"Cannot confirm order in {self.status} status")

        self.status = OrderStatus.CONFIRMED
        self.payment_transaction_id = payment_transaction_id
        self.confirmed_at = datetime.utcnow()

        # ✅ Domain event for state change
        self.raise_event(OrderConfirmedEvent(
            order_id=self.id,
            transaction_id=payment_transaction_id
        ))

# ✅ Service layer is thin - just orchestration
class PlaceOrderHandler(CommandHandler):
    async def handle_async(self, command: PlaceOrderCommand):
        # ✅ Domain entity handles all business logic
        order = Order(command.customer_id, command.items)

        # ✅ Process payment (external concern)
        payment = await self._payment_service.process_async(order.total)
        if not payment.success:
            return self.bad_request("Payment failed")

        order.confirm(payment.transaction_id)

        # ✅ Persist (infrastructure concern)
        await self._repository.save_async(order)

        # ✅ Events automatically published by framework
        return self.created(self._mapper.map(order, OrderDto))

from neuroglia.data.abstractions import Entity
from datetime import datetime
from typing import List
import uuid

class Pizza(Entity[str]):
    """Pizza entity with business logic and identity"""

    def __init__(self, name: str, price: float, ingredients: List[str], id: str = None):
        super().__init__()
        self.id = id or f"pizza_{uuid.uuid4().hex[:8]}"
        self.name = name
        self.price = price
        self.ingredients = ingredients.copy()
        self.is_available = True
        self.created_at = datetime.now()

        # Business rule validation
        if price <= 0:
            raise ValueError("Pizza price must be positive")
        if not ingredients:
            raise ValueError("Pizza must have at least one ingredient")

    def add_ingredient(self, ingredient: str) -> None:
        """Add ingredient with business rule validation"""
        if ingredient in self.ingredients:
            raise ValueError(f"Ingredient '{ingredient}' already exists")

        self.ingredients.append(ingredient)
        self.price += 2.50  # Business rule: each ingredient adds $2.50
        self.updated_at = datetime.now()

    def remove_ingredient(self, ingredient: str) -> None:
        """Remove ingredient with business validation"""
        if ingredient not in self.ingredients:
            raise ValueError(f"Ingredient '{ingredient}' not found")
        if len(self.ingredients) <= 1:
            raise ValueError("Pizza must have at least one ingredient")

        self.ingredients.remove(ingredient)
        self.price -= 2.50
        self.updated_at = datetime.now()

    def make_unavailable(self, reason: str) -> None:
        """Business operation to make pizza unavailable"""
        self.is_available = False
        self.unavailable_reason = reason
        self.updated_at = datetime.now()

from neuroglia.data.abstractions import DomainEvent
from dataclasses import dataclass
from decimal import Decimal
from datetime import datetime
from typing import List, Dict, Any

@dataclass
class PizzaOrderPlacedEvent(DomainEvent[str]):
    """Domain event representing a pizza order being placed"""
    customer_id: str
    items: List[Dict[str, Any]]
    total_amount: Decimal
    special_instructions: str

    def __init__(self, aggregate_id: str, customer_id: str, items: List[Dict[str, Any]],
                 total_amount: Decimal, special_instructions: str = ""):
        super().__init__(aggregate_id)
        self.customer_id = customer_id
        self.items = items
        self.total_amount = total_amount
        self.special_instructions = special_instructions

@dataclass
class OrderStatusChangedEvent(DomainEvent[str]):
    """Domain event representing order status changes"""
    previous_status: str
    new_status: str
    changed_by: str
    reason: str

    def __init__(self, aggregate_id: str, previous_status: str, new_status: str,
                 changed_by: str, reason: str = ""):
        super().__init__(aggregate_id)
        self.previous_status = previous_status
        self.new_status = new_status
        self.changed_by = changed_by
        self.reason = reason

@dataclass
class PaymentProcessedEvent(DomainEvent[str]):
    """Domain event representing successful payment processing"""
    payment_method: str
    amount: Decimal
    transaction_id: str
    processed_at: datetime

    def __init__(self, aggregate_id: str, payment_method: str, amount: Decimal,
                 transaction_id: str):
        super().__init__(aggregate_id)
        self.payment_method = payment_method
        self.amount = amount
        self.transaction_id = transaction_id
        self.processed_at = datetime.now()

# /src/neuroglia/data/abstractions.py
from abc import ABC
from datetime import datetime
from typing import Generic, List, Type, TypeVar

TKey = TypeVar("TKey")
"""Represents the generic argument used to specify the type of key to use"""

class Identifiable(Generic[TKey], ABC):
    """Defines the fundamentals of an object that can be identified based on a unique identifier"""
    id: TKey

class Entity(Generic[TKey], Identifiable[TKey], ABC):
    """Represents the abstract class inherited by all entities in the application"""

    def __init__(self) -> None:
        super().__init__()
        self.created_at = datetime.now()

    created_at: datetime
    last_modified: datetime

class VersionedState(ABC):
    """Represents the abstract class inherited by all versioned states"""

    def __init__(self):
        self.state_version = 0

    state_version: int = 0

class AggregateState(Generic[TKey], Identifiable[TKey], VersionedState, ABC):
    """Represents the abstract class inherited by all aggregate root states"""

    def __init__(self):
        super().__init__()

    id: TKey
    created_at: datetime
    last_modified: datetime

class DomainEvent(Generic[TKey], ABC):
    """Represents the base class inherited by all domain events"""

    def __init__(self, aggregate_id: TKey):
        self.created_at = datetime.now()
        self.aggregate_id = aggregate_id

    created_at: datetime
    aggregate_id: TKey
    aggregate_version: int

class AggregateRoot(Generic[TState, TKey], Entity[TKey], ABC):
    """Represents the base class for all aggregate roots"""

    _pending_events: List[DomainEvent]

    def __init__(self):
        self.state = object.__new__(self.__orig_bases__[0].__args__[0])
        self.state.__init__()
        self._pending_events = list[DomainEvent]()

    def id(self):
        return self.state.id

    state: TState

    def register_event(self, e: TEvent) -> TEvent:
        """Registers the specified domain event"""
        if not hasattr(self, "_pending_events"):
            self._pending_events = list[DomainEvent]()
        self._pending_events.append(e)
        e.aggregate_version = self.state.state_version + len(self._pending_events)
        return e

    def clear_pending_events(self):
        """Clears all pending domain events"""
        self._pending_events.clear()

from multipledispatch import dispatch
from neuroglia.data.abstractions import AggregateRoot, AggregateState

class BankAccountState(AggregateState[str]):
    """Aggregate state with event handlers using multiple dispatch"""

    def __init__(self):
        super().__init__()
        self.account_number: Optional[str] = None
        self.owner_name: Optional[str] = None
        self.balance: Decimal = Decimal("0.00")
        self.account_type: Optional[str] = None
        self.is_active: bool = True

    @dispatch(AccountCreatedEvent)
    def on(self, event: AccountCreatedEvent):
        """Apply account created event to state"""
        self.id = event.aggregate_id
        self.created_at = event.created_at
        self.account_number = event.account_number
        self.owner_name = event.owner_name
        self.balance = event.initial_balance
        self.account_type = event.account_type
        self.is_active = True

    @dispatch(MoneyDepositedEvent)
    def on(self, event: MoneyDepositedEvent):
        """Apply money deposited event to state"""
        self.balance = event.new_balance
        self.last_modified = event.created_at

    @dispatch(MoneyWithdrawnEvent)
    def on(self, event: MoneyWithdrawnEvent):
        """Apply money withdrawn event to state"""
        self.balance = event.new_balance
        self.last_modified = event.created_at

class BankAccountAggregate(AggregateRoot[BankAccountState, str]):
    """Aggregate root demonstrating event application pattern"""

    def create_account(self, account_number: str, owner_name: str,
                      initial_balance: Decimal, account_type: str):
        """Business operation that applies events to state"""

        # 1. Business rule validation
        if initial_balance < 0:
            raise ValueError("Initial balance cannot be negative")
        if account_type not in ["checking", "savings", "business"]:
            raise ValueError("Invalid account type")

        # 2. Create domain event
        event = AccountCreatedEvent(
            aggregate_id=self.state.id,
            account_number=account_number,
            owner_name=owner_name,
            initial_balance=initial_balance,
            account_type=account_type,
        )

        # 3. Apply event to state AND register for persistence
        self.state.on(event)  # Updates aggregate state immediately
        self.register_event(event)  # Adds to pending events for persistence

    def deposit_money(self, amount: Decimal, transaction_id: str):
        """Deposit operation with event-driven state changes"""

        # Business validation
        if amount <= 0:
            raise ValueError("Deposit amount must be positive")
        if not self.state.is_active:
            raise ValueError("Cannot deposit to inactive account")

        # Calculate new balance
        new_balance = self.state.balance + amount

        # Create and apply event
        event = MoneyDepositedEvent(
            aggregate_id=self.state.id,
            amount=amount,
            new_balance=new_balance,
            transaction_id=transaction_id,
        )

        self.state.on(event)  # State update via event
        self.register_event(event)  # Event registration

# Business method creates event
event = MoneyDepositedEvent(aggregate_id=self.id, amount=100.00, ...)

# State application - Uses @dispatch to find the right handler
self.state.on(event)  # Calls BankAccountState.on(MoneyDepositedEvent)

# Event registration - Adds to pending events collection
self.register_event(event)  # Adds to _pending_events list

@dispatch(AccountCreatedEvent)
def on(self, event: AccountCreatedEvent):
    # Handles AccountCreatedEvent specifically
    self.balance = event.initial_balance

@dispatch(MoneyDepositedEvent)
def on(self, event: MoneyDepositedEvent):
    # Handles MoneyDepositedEvent specifically
    self.balance = event.new_balance

# Python's multiple dispatch automatically routes:
# state.on(AccountCreatedEvent) -> First method
# state.on(MoneyDepositedEvent) -> Second method

def register_event(self, e: TEvent) -> TEvent:
    """Framework method that handles event registration"""

    # Add to pending events collection
    self._pending_events.append(e)

    # Set event version based on current state + pending events
    e.aggregate_version = self.state.state_version + len(self._pending_events)

    return e

# In CommandHandler or Application Service
async def handle_async(self, command: DepositMoneyCommand):

    # Load aggregate
    account = await self.repository.get_by_id_async(command.account_id)

    # Execute business operation (applies events)
    account.deposit_money(command.amount, command.transaction_id)

    # Save aggregate (persists events and updates state)
    await self.repository.save_async(account)
    #                                  ↑
    # Repository implementation will:
    # 1. Append events to event store
    # 2. Update read model/snapshot
    # 3. Publish events to event bus
    # 4. Clear pending events

def deposit_money(self, amount: Decimal):
    # Direct state mutation
    self.balance += amount
    self.last_modified = datetime.now()
    # Lost: WHY the balance changed, WHEN exactly, by WHOM

def deposit_money(self, amount: Decimal, transaction_id: str):
    # Create event with full context
    event = MoneyDepositedEvent(
        aggregate_id=self.id,
        amount=amount,
        new_balance=self.balance + amount,
        transaction_id=transaction_id
    )

    # Apply event to state (predictable, testable)
    self.state.on(event)

    # Register for persistence (audit trail, replay capability)
    self.register_event(event)

class Customer(Entity):
    """Simple entity with state persistence and domain events."""

    def __init__(self, name: str, email: str):
        super().__init__()
        self._id = str(uuid.uuid4())
        self.name = name
        self.email = email

        # Still raises domain events for integration
        self._raise_domain_event(CustomerCreatedEvent(
            customer_id=self.id,
            name=self.name,
            email=self.email
        ))

    def update_email(self, new_email: str) -> None:
        """Business method with validation and events."""
        if not self._is_valid_email(new_email):
            raise ValueError("Invalid email format")

        old_email = self.email
        self.email = new_email

        # Domain event for integration
        self._raise_domain_event(CustomerEmailUpdatedEvent(
            customer_id=self.id,
            old_email=old_email,
            new_email=new_email
        ))

    def _raise_domain_event(self, event: DomainEvent) -> None:
        if not hasattr(self, '_pending_events'):
            self._pending_events = []
        self._pending_events.append(event)

    @property
    def domain_events(self) -> List[DomainEvent]:
        """Required for Unit of Work integration."""
        return getattr(self, '_pending_events', []).copy()

# Usage in handler - same patterns as AggregateRoot
class UpdateCustomerEmailHandler(CommandHandler):
    async def handle_async(self, command: UpdateCustomerEmailCommand):
        customer = await self.customer_repository.get_by_id_async(command.customer_id)
        customer.update_email(command.new_email)  # Business logic + events

        await self.customer_repository.save_async(customer)  # State persistence
        self.unit_of_work.register_aggregate(customer)       # Event dispatching

        return self.ok(CustomerDto.from_entity(customer))

class BankAccount(AggregateRoot[BankAccountState, str]):
    """Complex aggregate with event sourcing and rich business logic."""

    def deposit_money(self, amount: Decimal, transaction_id: str) -> None:
        """Rich business logic with comprehensive validation."""
        # Business rules
        if amount <= 0:
            raise ValueError("Deposit amount must be positive")
        if self.state.is_frozen:
            raise DomainException("Cannot deposit to frozen account")

        # Apply event (changes state + records for replay)
        event = MoneyDepositedEvent(
            aggregate_id=self.id,
            amount=amount,
            new_balance=self.state.balance + amount,
            transaction_id=transaction_id,
            deposited_at=datetime.utcnow()
        )

        self.state.on(event)      # Apply to current state
        self.register_event(event)  # Record for event sourcing

# Usage - same handler patterns
class DepositMoneyHandler(CommandHandler):
    async def handle_async(self, command: DepositMoneyCommand):
        account = await self.account_repository.get_by_id_async(command.account_id)
        account.deposit_money(command.amount, command.transaction_id)

        await self.account_repository.save_async(account)  # Event store persistence
        self.unit_of_work.register_aggregate(account)      # Event dispatching

        return self.ok(AccountDto.from_aggregate(account))

# Domain Events
@dataclass
class BankAccountCreatedDomainEventV1(DomainEvent[str]):
    """Event raised when a bank account is created"""
    owner_id: str
    overdraft_limit: Decimal

@dataclass
class BankAccountTransactionRecordedDomainEventV1(DomainEvent[str]):
    """Event raised when a transaction is recorded"""
    transaction: BankTransactionV1

# Aggregate State with Event Handlers
@map_to(BankAccountDto)
class BankAccountStateV1(AggregateState[str]):
    """Bank account state with multiple dispatch event handlers"""

    def __init__(self):
        super().__init__()
        self.transactions: List[BankTransactionV1] = []
        self.balance: Decimal = Decimal("0.00")
        self.overdraft_limit: Decimal = Decimal("0.00")
        self.owner_id: str = ""

    @dispatch(BankAccountCreatedDomainEventV1)
    def on(self, event: BankAccountCreatedDomainEventV1):
        """Apply account creation event"""
        self.id = event.aggregate_id
        self.created_at = event.created_at
        self.owner_id = event.owner_id
        self.overdraft_limit = event.overdraft_limit

    @dispatch(BankAccountTransactionRecordedDomainEventV1)
    def on(self, event: BankAccountTransactionRecordedDomainEventV1):
        """Apply transaction event and recompute balance"""
        self.last_modified = event.created_at
        self.transactions.append(event.transaction)
        self._compute_balance()

    def _compute_balance(self):
        """Recompute balance from all transactions (event sourcing)"""
        balance = Decimal("0.00")
        for transaction in self.transactions:
            if transaction.type in [BankTransactionTypeV1.DEPOSIT.value,
                                  BankTransactionTypeV1.INTEREST.value]:
                balance += Decimal(transaction.amount)
            elif transaction.type == BankTransactionTypeV1.TRANSFER.value:
                if transaction.to_account_id == self.id:
                    balance += Decimal(transaction.amount)  # Incoming transfer
                else:
                    balance -= Decimal(transaction.amount)  # Outgoing transfer
            else:  # Withdrawal
                balance -= Decimal(transaction.amount)
        self.balance = balance

# Aggregate Root with Business Logic
class BankAccount(AggregateRoot[BankAccountStateV1, str]):
    """Bank account aggregate implementing banking business rules"""

    def __init__(self, owner: Person, overdraft_limit: Decimal = Decimal("0.00")):
        super().__init__()

        # Create account through event application
        event = BankAccountCreatedDomainEventV1(
            aggregate_id=str(uuid.uuid4()).replace('-', ''),
            owner_id=owner.id(),
            overdraft_limit=overdraft_limit
        )

        # Apply event to state AND register for persistence
        self.state.on(event)
        self.register_event(event)

    def get_available_balance(self) -> Decimal:
        """Calculate available balance including overdraft"""
        return self.state.balance + self.state.overdraft_limit

    def try_add_transaction(self, transaction: BankTransactionV1) -> bool:
        """Attempt to add transaction with business rule validation"""

        # Business rule: Check if transaction would cause overdraft
        if (transaction.type not in [BankTransactionTypeV1.DEPOSIT,
                                   BankTransactionTypeV1.INTEREST] and
            not (transaction.type == BankTransactionTypeV1.TRANSFER and
                 transaction.to_account_id == self.id()) and
            transaction.amount > self.get_available_balance()):
            return False  # Transaction rejected

        # Create and apply transaction event
        event = BankAccountTransactionRecordedDomainEventV1(
            aggregate_id=self.id(),
            transaction=transaction
        )

        # Event application pattern
        self.state.on(event)  # Updates state via multiple dispatch
        self.register_event(event)  # Registers for persistence

        return True  # Transaction accepted

# /src/neuroglia/data/infrastructure/event_sourcing/abstractions.py
class Aggregator:
    """Reconstructs aggregates from event streams"""

    def aggregate(self, events: List[EventRecord], aggregate_type: Type) -> AggregateRoot:
        """Rebuild aggregate state from historical events"""

        # 1. Create empty aggregate instance
        aggregate: AggregateRoot = object.__new__(aggregate_type)
        aggregate.state = aggregate.__orig_bases__[0].__args__[0]()

        # 2. Replay all events in sequence
        for event_record in events:
            # Apply each event to state using multiple dispatch
            aggregate.state.on(event_record.data)

            # Update state version to match event version
            aggregate.state.state_version = event_record.data.aggregate_version

        return aggregate

# Application Service using the pattern
class CreateBankAccountHandler(CommandHandler[CreateBankAccountCommand, OperationResult[BankAccountDto]]):

    async def handle_async(self, command: CreateBankAccountCommand) -> OperationResult[BankAccountDto]:

        # 1. Load related aggregate (Person)
        owner = await self.person_repository.get_by_id_async(command.owner_id)

        # 2. Create new aggregate (triggers events)
        account = BankAccount(owner, command.overdraft_limit)
        #                     ↑
        # This constructor:
        # - Creates BankAccountCreatedDomainEventV1
        # - Calls self.state.on(event) → Updates state via @dispatch
        # - Calls self.register_event(event) → Adds to _pending_events

        # 3. Save aggregate (persists events and publishes)
        saved_account = await self.repository.add_async(account)
        #                                                   ↑
        # Repository implementation:
        # - Appends events from _pending_events to event store
        # - Publishes events to event bus for integration
        # - Updates read models/projections
        # - Clears _pending_events

        # 4. Return DTO mapped from aggregate state
        return self.created(self.mapper.map(saved_account.state, BankAccountDto))

from neuroglia.data.abstractions import AggregateRoot, AggregateState
from multipledispatch import dispatch
from enum import Enum
from typing import Optional

class OrderStatus(Enum):
    PENDING = "PENDING"
    CONFIRMED = "CONFIRMED"
    PREPARING = "PREPARING"
    READY = "READY"
    DELIVERED = "DELIVERED"
    CANCELLED = "CANCELLED"

class PizzaOrderState(AggregateState[str]):
    """State for pizza order aggregate"""

    def __init__(self):
        super().__init__()
        self.customer_id = ""
        self.items = []
        self.total_amount = Decimal('0.00')
        self.status = OrderStatus.PENDING
        self.special_instructions = ""
        self.estimated_delivery = None
        self.payment_status = "UNPAID"

    @dispatch(PizzaOrderPlacedEvent)
    def on(self, event: PizzaOrderPlacedEvent):
        """Apply order placed event to state"""
        self.id = event.aggregate_id
        self.customer_id = event.customer_id
        self.items = event.items.copy()
        self.total_amount = event.total_amount
        self.special_instructions = event.special_instructions
        self.created_at = event.created_at

    @dispatch(OrderStatusChangedEvent)
    def on(self, event: OrderStatusChangedEvent):
        """Apply status change event to state"""
        self.status = OrderStatus(event.new_status)
        self.last_modified = event.created_at

    @dispatch(PaymentProcessedEvent)
    def on(self, event: PaymentProcessedEvent):
        """Apply payment processed event to state"""
        self.payment_status = "PAID"
        self.last_modified = event.created_at

class PizzaOrderAggregate(AggregateRoot[PizzaOrderState, str]):
    """Pizza order aggregate root implementing business rules"""

    def __init__(self, order_id: str = None):
        super().__init__()
        if order_id:
            self.state.id = order_id

    def place_order(self, customer_id: str, items: List[Dict[str, Any]],
                   special_instructions: str = "") -> None:
        """Place a new pizza order with business validation"""

        # Business rule validation
        if not items:
            raise ValueError("Order must contain at least one item")

        # Calculate total with business rules
        total = Decimal('0.00')
        for item in items:
            if item['quantity'] <= 0:
                raise ValueError("Item quantity must be positive")
            total += Decimal(str(item['price'])) * item['quantity']

        # Minimum order business rule
        if total < Decimal('10.00'):
            raise ValueError("Minimum order amount is $10.00")

        # Create and apply domain event
        event = PizzaOrderPlacedEvent(
            aggregate_id=self.state.id,
            customer_id=customer_id,
            items=items,
            total_amount=total,
            special_instructions=special_instructions
        )

        self.state.on(event)
        self.register_event(event)

    def confirm_order(self, estimated_delivery: datetime, confirmed_by: str) -> None:
        """Confirm order with business rules"""
        if self.state.status != OrderStatus.PENDING:
            raise ValueError(f"Cannot confirm order in {self.state.status.value} status")

        self.state.estimated_delivery = estimated_delivery

        # Create status change event
        event = OrderStatusChangedEvent(
            aggregate_id=self.state.id,
            previous_status=self.state.status.value,
            new_status=OrderStatus.CONFIRMED.value,
            changed_by=confirmed_by,
            reason="Order confirmed by kitchen"
        )

        self.state.on(event)
        self.register_event(event)

    def process_payment(self, payment_method: str, transaction_id: str) -> None:
        """Process payment with business validation"""
        if self.state.payment_status == "PAID":
            raise ValueError("Order is already paid")

        if self.state.status == OrderStatus.CANCELLED:
            raise ValueError("Cannot process payment for cancelled order")

        event = PaymentProcessedEvent(
            aggregate_id=self.state.id,
            payment_method=payment_method,
            amount=self.state.total_amount,
            transaction_id=transaction_id
        )

        self.state.on(event)
        self.register_event(event)

    def cancel_order(self, reason: str, cancelled_by: str) -> None:
        """Cancel order with business rules"""
        if self.state.status in [OrderStatus.DELIVERED, OrderStatus.CANCELLED]:
            raise ValueError(f"Cannot cancel order in {self.state.status.value} status")

        event = OrderStatusChangedEvent(
            aggregate_id=self.state.id,
            previous_status=self.state.status.value,
            new_status=OrderStatus.CANCELLED.value,
            changed_by=cancelled_by,
            reason=reason
        )

        self.state.on(event)
        self.register_event(event)

from typing import List
from decimal import Decimal

class OrderWithPromotionAggregate(AggregateRoot[PizzaOrderState, str]):
    """Extended order aggregate with promotion handling"""

    def place_order_with_promotion(self, customer_id: str, items: List[Dict[str, Any]],
                                  promotion_code: str = None) -> None:
        """Place order with potential promotion - multiple events in single transaction"""

        # Step 1: Validate and place base order
        self.place_order(customer_id, items)

        # Step 2: Apply promotion if valid
        if promotion_code:
            discount_amount = self._validate_and_calculate_promotion(promotion_code)
            if discount_amount > 0:
                # Create promotion applied event
                promotion_event = PromotionAppliedEvent(
                    aggregate_id=self.state.id,
                    promotion_code=promotion_code,
                    discount_amount=discount_amount,
                    original_amount=self.state.total_amount
                )

                self.state.on(promotion_event)
                self.register_event(promotion_event)

        # Step 3: Check for loyalty points
        loyalty_points = self._calculate_loyalty_points()
        if loyalty_points > 0:
            loyalty_event = LoyaltyPointsEarnedEvent(
                aggregate_id=self.state.id,
                customer_id=customer_id,
                points_earned=loyalty_points,
                transaction_amount=self.state.total_amount
            )

            self.register_event(loyalty_event)

    def _validate_and_calculate_promotion(self, promotion_code: str) -> Decimal:
        """Business logic for promotion validation"""
        promotions = {
            "FIRST10": Decimal('10.00'),
            "STUDENT15": self.state.total_amount * Decimal('0.15')
        }
        return promotions.get(promotion_code, Decimal('0.00'))

    def _calculate_loyalty_points(self) -> int:
        """Business logic for loyalty points calculation"""
        # 1 point per dollar spent
        return int(self.state.total_amount)

@dataclass
class PromotionAppliedEvent(DomainEvent[str]):
    """Domain event for promotion application"""
    promotion_code: str
    discount_amount: Decimal
    original_amount: Decimal

@dataclass
class LoyaltyPointsEarnedEvent(DomainEvent[str]):
    """Domain event for loyalty points"""
    customer_id: str
    points_earned: int
    transaction_amount: Decimal

from neuroglia.eventing import DomainEventHandler
from neuroglia.eventing.cloud_events import CloudEvent
from typing import Dict, Any

class OrderDomainEventHandler(DomainEventHandler[PizzaOrderPlacedEvent]):
    """Handles domain events and publishes integration events"""

    def __init__(self, event_bus: EventBus, mapper: Mapper):
        self.event_bus = event_bus
        self.mapper = mapper

    async def handle_async(self, domain_event: PizzaOrderPlacedEvent) -> None:
        """Convert domain event to integration event (CloudEvent)"""

        # Transform domain event to integration event data
        integration_data = {
            "orderId": domain_event.aggregate_id,
            "customerId": domain_event.customer_id,
            "totalAmount": float(domain_event.total_amount),
            "items": domain_event.items,
            "orderPlacedAt": domain_event.created_at.isoformat()
        }

        # Create CloudEvent for external systems
        cloud_event = CloudEvent(
            source="mario-pizzeria/orders",
            type="com.mario-pizzeria.order.placed.v1",
            data=integration_data,
            datacontenttype="application/json"
        )

        # Publish to external systems
        await self.event_bus.publish_async(cloud_event)

        # Handle internal business workflows
        await self._notify_kitchen(domain_event)
        await self._update_inventory(domain_event)
        await self._send_customer_confirmation(domain_event)

    async def _notify_kitchen(self, event: PizzaOrderPlacedEvent) -> None:
        """Internal business workflow - kitchen notification"""
        kitchen_notification = KitchenOrderReceivedEvent(
            order_id=event.aggregate_id,
            items=event.items,
            special_instructions=event.special_instructions
        )
        await self.event_bus.publish_async(kitchen_notification)

    async def _update_inventory(self, event: PizzaOrderPlacedEvent) -> None:
        """Internal business workflow - inventory management"""
        for item in event.items:
            inventory_event = IngredientReservedEvent(
                pizza_type=item['name'],
                quantity=item['quantity'],
                order_id=event.aggregate_id
            )
            await self.event_bus.publish_async(inventory_event)

class TraditionalOrderService:
    """Traditional CRUD approach - current state only"""

    def __init__(self, repository: Repository[PizzaOrder, str]):
        self.repository = repository

    async def place_order_async(self, command: PlaceOrderCommand) -> PizzaOrder:
        """Traditional approach - direct state mutation"""

        # Create order entity with current state
        order = PizzaOrder(
            customer_id=command.customer_id,
            items=command.items,
            total_amount=self._calculate_total(command.items),
            status=OrderStatus.PENDING,
            created_at=datetime.now()
        )

        # Validate business rules
        self._validate_order(order)

        # Save current state only
        saved_order = await self.repository.add_async(order)

        # Manually trigger side effects
        await self._send_notifications(saved_order)
        await self._update_inventory(saved_order)

        return saved_order

    async def update_status_async(self, order_id: str, new_status: OrderStatus) -> PizzaOrder:
        """Traditional approach - direct state update"""
        order = await self.repository.get_async(order_id)
        if not order:
            raise ValueError("Order not found")

        # Direct state mutation (loses history)
        old_status = order.status
        order.status = new_status
        order.updated_at = datetime.now()

        # Save updated state (old state is lost)
        return await self.repository.update_async(order)

from neuroglia.data.infrastructure.event_sourcing import EventSourcingRepository

class EventSourcedOrderService:
    """Event sourcing approach - complete history preservation"""

    def __init__(self, repository: EventSourcingRepository[PizzaOrderAggregate, str]):
        self.repository = repository

    async def place_order_async(self, command: PlaceOrderCommand) -> PizzaOrderAggregate:
        """Event sourcing approach - event-based state building"""

        # Create new aggregate
        aggregate = PizzaOrderAggregate(f"order_{uuid.uuid4().hex[:8]}")

        # Apply business operation (generates events)
        aggregate.place_order(
            customer_id=command.customer_id,
            items=command.items,
            special_instructions=command.special_instructions
        )

        # Repository saves events and publishes them
        return await self.repository.add_async(aggregate)

    async def update_status_async(self, order_id: str, new_status: OrderStatus,
                                 changed_by: str, reason: str) -> PizzaOrderAggregate:
        """Event sourcing approach - reconstruct from events"""

        # Reconstruct aggregate from events
        aggregate = await self.repository.get_async(order_id)
        if not aggregate:
            raise ValueError("Order not found")

        # Apply business operation (generates new events)
        if new_status == OrderStatus.CONFIRMED:
            aggregate.confirm_order(
                estimated_delivery=datetime.now() + timedelta(minutes=30),
                confirmed_by=changed_by
            )
        elif new_status == OrderStatus.CANCELLED:
            aggregate.cancel_order(reason, changed_by)

        # Save new events (all history preserved)
        return await self.repository.update_async(aggregate)

from neuroglia.mvc import ControllerBase
from neuroglia.mediation import Mediator, Command, OperationResult
from neuroglia.mapping import Mapper

# 1. API Layer - Controllers and DTOs
@dataclass
class PlaceOrderDto:
    """Data Transfer Object for API requests"""
    customer_id: str
    items: List[Dict[str, Any]]
    special_instructions: str = ""

@dataclass
class OrderDto:
    """Data Transfer Object for API responses"""
    id: str
    customer_id: str
    items: List[Dict[str, Any]]
    total_amount: float
    status: str
    created_at: str

class OrdersController(ControllerBase):
    """API Controller handling HTTP requests"""

    @post("/orders", response_model=OrderDto, status_code=201)
    async def place_order(self, dto: PlaceOrderDto) -> OrderDto:
        """🌐 API Layer: Transform DTO to Command"""
        command = self.mapper.map(dto, PlaceOrderCommand)
        result = await self.mediator.execute_async(command)
        return self.process(result)

# 2. Application Layer - Commands and Handlers
@dataclass
class PlaceOrderCommand(Command[OperationResult[OrderDto]]):
    """Application command for placing orders"""
    customer_id: str
    items: List[Dict[str, Any]]
    special_instructions: str = ""

class PlaceOrderHandler(CommandHandler[PlaceOrderCommand, OperationResult[OrderDto]]):
    """💼 Application Layer: Business orchestration"""

    def __init__(self, repository: Repository[PizzaOrderAggregate, str],
                 mapper: Mapper):
        self.repository = repository
        self.mapper = mapper

    async def handle_async(self, command: PlaceOrderCommand) -> OperationResult[OrderDto]:
        """Handle command with domain coordination"""
        try:
            # Create domain aggregate
            aggregate = PizzaOrderAggregate()

            # Apply domain operation
            aggregate.place_order(
                customer_id=command.customer_id,
                items=command.items,
                special_instructions=command.special_instructions
            )

            # Persist with events
            saved_aggregate = await self.repository.add_async(aggregate)

            # Transform back to DTO
            dto = self.mapper.map(saved_aggregate.state, OrderDto)
            return self.created(dto)

        except ValueError as e:
            return self.bad_request(str(e))
        except Exception as e:
            return self.internal_server_error(f"Failed to place order: {str(e)}")

# 3. Integration Layer - Event Handlers
class OrderIntegrationEventHandler(DomainEventHandler[PizzaOrderPlacedEvent]):
    """🔌 Integration Layer: External system coordination"""

    async def handle_async(self, event: PizzaOrderPlacedEvent) -> None:
        """Transform domain events to integration events"""

        # Convert to CloudEvent for external systems
        cloud_event = CloudEvent(
            source="mario-pizzeria/orders",
            type="com.mario-pizzeria.order.placed.v1",
            data={
                "orderId": event.aggregate_id,
                "customerId": event.customer_id,
                "totalAmount": float(event.total_amount),
                "timestamp": event.created_at.isoformat()
            }
        )

        await self.event_bus.publish_async(cloud_event)