Event-Driven Architecture

Lexigram is built for the “Event First” world. lexigram-events provides a full CQRS stack that decouples your business domain from transactional infrastructure, allowing for highly scalable and resilient systems.

---

1. Core Concepts

Lexigram separates operations into two distinct paths:

1. Commands: Intentional actions that change state (e.g., PlaceOrder).
2. Events: Historical facts that have already occurred (e.g., OrderPlaced).

graph LR
    User[User/System] -- Dispatch --> CB[Command Bus]
    CB -- Execute --> Agg[Aggregate/Service]
    Agg -- Publish --> EB[Event Bus]
    EB -- Notify --> Sub[Subscribers/Projections]

---

2. Commands & Command Bus

A Command is a simple DTO that implements the Command contract. The CommandBus routes it to exactly one handler.

from lexigram.contracts.domain.base import Command
from lexigram.events.buses.command import CommandBus

class RegisterUser(Command):
    user_id: str
    email: str

# Dispatching from a controller
async def register(bus: CommandBus):
    await bus.dispatch(RegisterUser(user_id="u1", email="admin@lexigram.dev"))

---

3. Events & Event Bus

An Event is shared with the system to notify other components of a state change. Unlike commands, events can have multiple subscribers.

Defining Events

from lexigram.contracts.domain.base import DomainEvent

class UserRegistered(DomainEvent):
    user_id: str
    email: str

Subscribing to Events

Use the @event_handler decorator to respond to specific events.

from lexigram.events.decorators import event_handler

@event_handler(UserRegistered)
async def welcome_new_user(event: UserRegistered):
    # Logic to send email, create profile, etc.
    print(f"Welcoming {event.email}!")

---

4. Encryption for PII

Lexigram provides first-class support for protecting sensitive data in your event stream. By using the @encrypted_event decorator, PII fields are encrypted at rest in the Event Store using Fernet encryption.

from lexigram.events.decorators import encrypted_event

@encrypted_event(key_alias="events/pii-key")
class CustomerOnboarded(DomainEvent):
    customer_id: str
    email: str        # This field will be encrypted in the store
    phone: str        # This field will be encrypted in the store

---

5. Event Sourcing & Projections

For complex domains, you can use the EventStore to persist the full history of changes instead of just the current state.

• Event Store: An append-only log of all domain events.
• Projections: Read models built by subscribing to the event stream and updating a specialized table (e.g., for search or analytics).

from lexigram.events.stores.postgres.event_store import PostgresEventStore

# Appending to the log
await event_store.append(UserRegistered(user_id="u1", email="..."))

# Replaying history to reconstruct state
events = await event_store.get_events(aggregate_id="u1")

---

6. Reliable Delivery (The Outbox)

To prevent “Dual Write” problems (updating DB but failing to publish event), Lexigram implements the Transactional Outbox pattern.

1. Your service saves data and the event to the database in a single transaction.
2. A background process picks up the events from the outbox table and publishes them to the broker.

application.yaml

events:
  outbox_enabled: true
  outbox_poll_interval: 1.0

---

[!IMPORTANT] Always prefer the Command Bus for inter-module communication within the same application. This ensures that modules are decoupled and only communicate through well-defined contracts.