Event-Driven Architecture como estilo arquitetural

Event-Driven Architecture (EDA) não é um padrão único — é um espectro de abordagens que compartilham uma ideia central: componentes do sistema se comunicam produzindo e consumindo eventos, em vez de chamar uns aos outros diretamente. A natureza do desacoplamento e o que um "evento" significa variam significativamente entre as variações. Martin Fowler organizou esse espectro em quatro padrões distintos, e entender a diferença entre eles é essencial para escolher o certo para cada problema.

Os quatro padrões de Fowler

Event Notification. O produtor publica um evento informando que algo aconteceu — sem incluir os dados do que aconteceu. Os consumidores recebem a notificação e, se precisarem de detalhes, consultam o produtor de volta. Exemplo: "OrderShipped" sem os detalhes do pedido no payload. O benefício é desacoplamento temporal. O custo é o "chattiness": cada consumidor precisa de uma chamada adicional para obter os dados.

Event-Carried State Transfer. O produtor publica um evento com todos os dados relevantes do que aconteceu. Consumidores recebem o evento e podem agir sem consultar de volta o produtor. Exemplo: "CustomerUpdated" com o estado completo atualizado do cliente. O benefício é autonomia dos consumidores — eles mantêm uma cópia local do estado e operam mesmo se o produtor estiver indisponível. O custo é o tamanho dos eventos e a necessidade de sincronizar estado entre produtor e cópias locais dos consumidores.

Event Sourcing. O estado do sistema é armazenado como uma sequência imutável de eventos, em vez de como snapshot do estado atual. O estado atual é derivado replaying (aplicando) os eventos em ordem. Este padrão é coberto em profundidade no M07 conceito 13.

CQRS (Command Query Responsibility Segregation). Separação do modelo de escrita (comandos) do modelo de leitura (queries). Frequentemente combinado com Event Sourcing mas independente dele. Coberto em M07 conceito 12.

Choreography vs Orchestration

Quando múltiplos serviços precisam colaborar em um fluxo de negócio, há duas formas de coordenar essa colaboração. A escolha entre elas é uma das decisões mais importantes em sistemas event-driven.

Choreography (coreografia): cada serviço reage a eventos publicados por outros e publica seus próprios eventos. Não há coordenador central. O fluxo emerge da interação entre os participantes. Exemplo: Orders publica OrderPlaced → Inventory consome e publica StockReserved → Payments consome e publica PaymentApproved → Shipping consome e inicia despacho.

O benefício de choreography é o desacoplamento máximo: cada serviço só sabe dos eventos que consome e dos que produz. Adicionar um novo participante é adicionar um novo consumidor sem mudar os outros serviços. O custo é a rastreabilidade: entender o fluxo completo requer correlacionar eventos de múltiplos serviços. Debugar "por que esse pedido não foi despachado?" pode exigir olhar logs de quatro serviços diferentes.

Orchestration (orquestração): um coordenador central (orquestrador ou Saga) dirige o fluxo — chama serviços em sequência e decide o que fazer com os resultados. Exemplo: um OrderSaga que chama Inventory.ReserveStock, depois Payments.Charge, depois Shipping.Schedule, e lida com falhas de cada passo com ações de compensação.

O benefício de orchestration é a visibilidade: o fluxo inteiro está documentado no orquestrador, e o estado da Saga pode ser consultado diretamente. O custo é acoplamento: o orquestrador conhece todos os serviços, tornando-se um ponto de mudança frequente.

Versionamento de eventos

Eventos publicados em um sistema event-driven se tornam parte de uma API pública implícita. Consumidores dependem do schema do evento. Quando o schema muda, você tem um problema de compatibilidade que precisa ser gerenciado explicitamente.

A abordagem mais pragmática é o versionamento por versão no tipo do evento: OrderPlaced.v1 → OrderPlaced.v2. Durante a transição, produtor publica ambas as versões (ou um adapter converte); consumidores migram em seu próprio ritmo.

Uma alternativa é a estratégia "tolerant reader": consumidores ignoram campos que não reconhecem e tratam campos ausentes com defaults. Isso permite adicionar campos sem versionar — mas não permite remover campos ou mudar semântica.

A pior abordagem é a que mais acontece: nenhuma estratégia. Produtor muda o schema; alguns consumidores quebram; há urgência para coordenar um deploy simultâneo de múltiplos serviços. Esse cenário é o "monolito distribuído temporal" — onde o que deveria ser desacoplado no tempo está acoplado pelo schema do evento.

Testabilidade de flows assíncronos

Testar flows event-driven é genuinamente mais difícil que testar chamadas síncronas. Há três estratégias principais:

Teste de unidade por componente. Cada produtor é testado em isolamento: dado esse input, esse evento deve ser publicado. Cada consumidor é testado em isolamento: dado esse evento, essa ação deve acontecer. Os testes não cobrem o fluxo end-to-end, mas cobrem cada componente individualmente com velocidade.

Teste de contrato (Consumer-Driven Contracts). Consumidores definem contratos de quais eventos esperam e como. Produtores verificam que satisfazem esses contratos. Pact é a ferramenta mais popular para isso. O benefício é detectar breaking changes antes de ir para produção sem precisar de um ambiente completo.

Teste de integração com broker local. Subir um Kafka ou RabbitMQ em Docker para testes de integração que verificam o fluxo end-to-end. Testcontainers é a solução canônica aqui — broker em container descartável por suite de teste. Mais lento que unidade, mas verifica a integração real.

Implementação: evento publicado e consumido

// Domain Event publicado pelo aggregate
public record OrderPlacedEvent(
    string OrderId,
    string CustomerId,
    decimal Total,
    DateTimeOffset OccurredAt
) : IDomainEvent;

// MassTransit — publicação via outbox (garante at-least-once)
public class PlaceOrderHandler
{
    private readonly IPublishEndpoint _bus;

    public async Task Handle(PlaceOrderCommand cmd, CancellationToken ct)
    {
        var order = Order.Create(cmd.CustomerId, cmd.Lines);
        await _orders.SaveAsync(order, ct);
        // publicado após commit — MassTransit Outbox garante entrega
        await _bus.Publish(new OrderPlacedEvent(
            order.Id.Value, order.CustomerId.Value, order.Total.Amount, DateTimeOffset.UtcNow
        ), ct);
    }
}

// Consumer no serviço de Inventory — coreografia
public class OrderPlacedConsumer : IConsumer<OrderPlacedEvent>
{
    private readonly IInventoryService _inventory;

    public async Task Consume(ConsumeContext<OrderPlacedEvent> ctx)
    {
        var ev = ctx.Message;
        await _inventory.ReserveStockAsync(ev.OrderId, ev.CustomerId);
        // publica seu próprio evento para continuar o flow
        await ctx.Publish(new StockReservedEvent(ev.OrderId, DateTimeOffset.UtcNow));
    }
}

// Registro no host
services.AddMassTransit(cfg =>
{
    cfg.AddConsumer<OrderPlacedConsumer>();
    cfg.UsingRabbitMq((ctx, bus) =>
    {
        bus.Host("rabbitmq://localhost");
        bus.ConfigureEndpoints(ctx);
    });
});

// Teste de unidade do consumer
[Fact]
public async Task OrderPlacedConsumer_ReservesStock_AndPublishesEvent()
{
    var inventory = Substitute.For<IInventoryService>();
    var consumer = new OrderPlacedConsumer(inventory);
    var ctx = FakeConsumeContext.For(new OrderPlacedEvent("ord-1", "cust-1", 100m, DateTimeOffset.UtcNow));

    await consumer.Consume(ctx);

    await inventory.Received(1).ReserveStockAsync("ord-1", "cust-1");
    ctx.ShouldHavePublished<StockReservedEvent>();
}

from dataclasses import dataclass
from datetime import datetime, timezone
import aio_pika
import json

@dataclass(frozen=True)
class OrderPlacedEvent:
    order_id: str
    customer_id: str
    total: float
    occurred_at: str

    def to_json(self) -> str:
        return json.dumps(self.__dict__)

    @classmethod
    def from_json(cls, data: str) -> "OrderPlacedEvent":
        return cls(**json.loads(data))

# Publicação (produtor)
class EventPublisher:
    def __init__(self, channel: aio_pika.Channel):
        self._channel = channel

    async def publish(self, event: OrderPlacedEvent, exchange: str) -> None:
        x = await self._channel.get_exchange(exchange)
        await x.publish(
            aio_pika.Message(
                body=event.to_json().encode(),
                content_type="application/json",
                headers={"event-type": event.__class__.__name__}
            ),
            routing_key="orders.placed",
        )

# Consumer (coreografia em Inventory)
async def on_order_placed(msg: aio_pika.IncomingMessage) -> None:
    async with msg.process():
        event = OrderPlacedEvent.from_json(msg.body.decode())
        await inventory_service.reserve_stock(event.order_id)
        # publica StockReservedEvent para continuar o flow

# Registro do consumer
queue = await channel.declare_queue("inventory.order-placed")
await queue.consume(on_order_placed)

# Teste do consumer (sem broker)
import pytest

@pytest.mark.asyncio
async def test_on_order_placed_reserves_stock(mocker):
    mock_inventory = mocker.AsyncMock()
    mocker.patch("module.inventory_service", mock_inventory)

    event = OrderPlacedEvent("ord-1", "cust-1", 100.0, datetime.now(timezone.utc).isoformat())
    msg = FakeIncomingMessage(event.to_json())

    await on_order_placed(msg)

    mock_inventory.reserve_stock.assert_awaited_once_with("ord-1")

// events/order_placed.go
package events

import "time"

type OrderPlaced struct {
    OrderID    string    `json:"order_id"`
    CustomerID string    `json:"customer_id"`
    Total      float64   `json:"total"`
    OccurredAt time.Time `json:"occurred_at"`
}

// publisher/publisher.go — publicação via NATS JetStream
package publisher

import (
    "encoding/json"
    "github.com/nats-io/nats.go"
)

type Publisher struct{ js nats.JetStreamContext }

func (p *Publisher) Publish(subject string, event any) error {
    body, err := json.Marshal(event)
    if err != nil { return err }
    _, err = p.js.Publish(subject, body)
    return err
}

// orders/handler.go — produtor
func (h *PlaceOrderHandler) Handle(ctx context.Context, cmd PlaceOrderCommand) error {
    order := domain.NewOrder(cmd.CustomerID, cmd.Lines)
    if err := h.orders.Save(ctx, order); err != nil { return err }
    return h.pub.Publish("orders.placed", events.OrderPlaced{
        OrderID:    order.ID,
        CustomerID: order.CustomerID,
        Total:      order.Total(),
        OccurredAt: time.Now().UTC(),
    })
}

// inventory/consumer.go — consumidor (coreografia)
type OrderPlacedConsumer struct {
    inventory InventoryService
    publisher *publisher.Publisher
}

func (c *OrderPlacedConsumer) Handle(msg *nats.Msg) {
    var ev events.OrderPlaced
    if err := json.Unmarshal(msg.Data, &ev); err != nil {
        msg.Nak(); return
    }
    if err := c.inventory.ReserveStock(context.Background(), ev.OrderID); err != nil {
        msg.Nak(); return
    }
    c.publisher.Publish("inventory.stock-reserved", events.StockReserved{
        OrderID: ev.OrderID,
        At:      time.Now().UTC(),
    })
    msg.Ack()
}

// Teste
func TestOrderPlacedConsumer_ReservesStock(t *testing.T) {
    inventory := &fakeInventory{}
    pub := &fakePublisher{}
    consumer := &OrderPlacedConsumer{inventory: inventory, publisher: pub}

    ev := events.OrderPlaced{OrderID: "ord-1", CustomerID: "cust-1"}
    body, _ := json.Marshal(ev)
    consumer.Handle(&nats.Msg{Data: body})

    assert.True(t, inventory.reserveCalled)
    assert.True(t, pub.publishedStockReserved)
}