Domain Events: Communicating Between Aggregates

When Things Happen in the Domain

Business processes are rarely isolated. When a customer places an order, several things need to happen: inventory must be reserved, a confirmation email must be sent, the sales team's dashboard must update, and a fraud check must run.

If you put all of this logic inside the Order aggregate's place_order method, you've created a massive, tightly coupled blob that violates the Single Responsibility Principle and makes the aggregate aware of every other system in the company.

Domain Events as the Solution

A Domain Event is a record of something significant that happened in the domain — something the business cares about. It is named in the past tense because it represents a fact that has already occurred.

Examples: OrderPlaced, PaymentProcessed, ItemShipped, AccountSuspended.

The pattern works like this:

1. An aggregate performs an action and records a domain event internally.
2. After the aggregate is saved, the application layer publishes those events to a message bus (e.g., Kafka, RabbitMQ, AWS SNS).
3. Other Bounded Contexts subscribe to the events they care about and react independently.

# The event — an immutable record of what happened
@dataclass(frozen=True)
class OrderPlaced:
    event_id: str
    order_id: str
    customer_id: str
    total_amount: float
    currency: str
    occurred_at: str  # ISO 8601 timestamp

class Order:
    def __init__(self, order_id: str, customer_id: str):
        self.order_id = order_id
        self.customer_id = customer_id
        self._items = []
        self.status = "PENDING"
        self._events = []  # Uncommitted domain events

    def place_order(self):
        if not self._items:
            raise ValueError("Cannot place an empty order")
        self.status = "PLACED"
        # Record the event — don't publish it yet
        self._events.append(OrderPlaced(
            event_id=generate_uuid(),
            order_id=self.order_id,
            customer_id=self.customer_id,
            total_amount=self.calculate_total().amount,
            currency="USD",
            occurred_at=now_iso8601()
        ))

    def pop_events(self):
        events = list(self._events)
        self._events.clear()
        return events

After saving the Order to the database, the application layer calls order.pop_events() and publishes each event to the message bus. The Inventory service, Email service, and Fraud service each subscribe independently and react in their own time.

This decouples the Bounded Contexts completely. The Order aggregate doesn't know or care that an Inventory service exists.