Integration Events y Consistencia Eventual
Eventos de dominio vs integración, idempotencia y entrega at-least-once.
Los domain events son hechos internos a un bounded context: usan el vocabulario del dominio y pueden referenciar tipos internos como Value Objects e IDs tipados. Los integration events son contratos públicos entre contextos o sistemas externos: deben ser estables, versionados, y no filtrar detalles del modelo interno. Un domain event nunca cruza una frontera de contexto directamente — siempre se traduce a un integration event.
La capa de traducción es el puente entre ambos mundos. Cuando un domain event necesita cruzar una frontera, un traductor dedicado lo mapea a un integration event. El integration event es plano, serializable con tipos primitivos, y diseñado para compatibilidad hacia atrás. Esta traducción protege el modelo interno de presiones externas y permite evolucionar el dominio sin romper contratos públicos.
La consistencia eventual es la consecuencia de la comunicación asíncrona entre contextos. Cada contexto es consistente dentro de sí mismo; los contextos convergen a través del tiempo mediante eventos. Los consumidores deben tolerar ver datos levemente desactualizados temporalmente. Esta es una elección de diseño deliberada, no un accidente — debe documentarse el SLA de convergencia esperado.
La entrega at-least-once es el comportamiento por defecto de los brokers de mensajes: garantizan que el mensaje llegará, pero pueden duplicarlo si hay fallos de red o reinicios. Los consumidores deben ser idempotentes — procesar el mismo mensaje dos veces debe tener exactamente el mismo efecto que procesarlo una vez. El event_id es el identificador estándar de deduplicación.
Cuándo NO usar consistencia eventual: si dos contextos necesitan ser consistentes dentro de la misma request, probablemente pertenecen al mismo bounded context. La consistencia eventual es un tradeoff real — el sistema puede mostrar datos desactualizados, la compensación de errores es compleja, y el debugging requiere rastrear eventos a través de múltiples sistemas. No la aceptes porque es más fácil de implementar.
# domain/events/<something_happened>.py — Domain Event (internal)
# application/integration/<something_happened>_integration_event.py — Integration Event (public)
# application/translators/<something>_translator.py — Translator
# application/handlers/<something>_integration_handler.py — Consumer
@dataclass(frozen=True)
class <SomethingHappened>: # Domain Event (internal, rich types)
aggregate_id: <AggregateId>
<value>: <ValueObject>
occurred_at: datetime
@dataclass(frozen=True)
class <SomethingHappened>IntegrationEvent: # Integration Event (public, primitive types)
event_id: str # For idempotency
event_version: str # 'v1'
aggregate_id: str # Primitive, no domain type
<value>: str # Primitive
occurred_at: str # ISO-8601 string
class <Something>Translator:
def to_integration(self, event: <SomethingHappened>) -> <SomethingHappened>IntegrationEvent:
return <SomethingHappened>IntegrationEvent(
event_id=str(uuid4()),
event_version='v1',
aggregate_id=str(event.aggregate_id),
<value>=str(event.<value>),
occurred_at=event.occurred_at.isoformat(),
)
class <Something>IntegrationHandler:
def __init__(self, processed_store: <ProcessedEventStore>, service: <ApplicationService>) -> None:
self._processed = processed_store
self._service = service
def handle(self, event: <SomethingHappened>IntegrationEvent) -> None:
# Idempotency check — skip if already processed
if self._processed.contains(event.event_id):
return
self._service.handle_<something>(event)
self._processed.mark(event.event_id)Debugging lab
Detecta y corrige el error o la violación de diseño.
- 7.5.5.1
# Bug: integration event containing internal ValueObject type imported from the domain package from contexts.catalog.domain.model.price import Price # Internal domain type @dataclass(frozen=True) class ItemPublishedIntegrationEvent: event_id: str aggregate_id: str price: Price # Bug: internal ValueObject leaking into public contract occurred_at: str
- 7.5.5.2
# Bug: integration event consumer without idempotency check class ActionTriggeredIntegrationHandler: def handle(self, event: ActionTriggeredIntegrationEvent) -> None: # Bug: no idempotency check — same event processed twice = double effect target = self._repo.find(TargetId(event.target_id)) target.apply_action(ActionValue(event.action_value)) self._repo.save(target)
- 7.5.5.3
# Bug: Context A directly querying Context B's database table to stay consistent immediately class ContextAService: def get_current_state(self, entity_id: str) -> ContextAView: # Bug: directly querying Context B's database — structural coupling row = self._context_b_db_session.query(ContextBEntity).filter_by(id=entity_id).first() return ContextAView(id=row.id, value=row.internal_field, status=row.internal_status)
- 7.5.5.4
# Bug: integration event handler accessing internal ValueObject attribute from domain of producer class SomethingHappenedIntegrationHandler: def handle(self, event: SomethingHappenedIntegrationEvent) -> None: # Bug: accessing .value attribute of what was an AggregateId VO in the producer # The handler treats event.aggregate_id as if it still has the VO interface internal_id = event.aggregate_id.value # AttributeError: str has no attribute 'value' target = self._repo.find(LocalId(internal_id)) target.react_to_external_event(event.some_field.amount) # Same bug with some_field
- 7.5.5.5
# Bug: two contexts made eventually consistent for an operation that requires atomicity # Requirement: both contexts must reflect the same state — no partial states allowed class ContextAService: def perform_critical_action(self, cmd: CriticalActionCommand) -> None: aggregate = self._repo.find(cmd.aggregate_id) aggregate.perform_action(cmd.data) self._repo.save(aggregate) # Event published to broker — Context B will update eventually # Bug: if Context B fails to process, system is permanently inconsistent