Padrões de Mensageria Assíncrona

Ter uma fila ou Kafka na arquitetura não torna o sistema correto — apenas assincrono. Os bugs mais sutis de sistemas distribuídos aparecem depois que a mensageria é adotada: eventos publicados mas não persistidos, transações distribuídas que ficam parcialmente concluídas, duplicatas processadas duas vezes, estado reconstruído de forma inconsistente. Os padrões deste conceito existem para resolver exatamente esses problemas.

A distinção fundamental antes de qualquer padrão: eventos, comandos e queries são tipos de mensagem com semânticas diferentes. Um evento é um fato passado imutável — OrderPlaced, PaymentFailed. Um comando é uma intenção de ação — PlaceOrder, ReserveInventory. Uma query é uma pergunta sobre estado — GetOrderStatus. Misturar os três no mesmo canal ou tratar eventos como comandos é uma fonte comum de acoplamento inadvertido e semântica confusa. Saga, Outbox e Event Sourcing trabalham especificamente com eventos e comandos. CQRS separa queries do resto.

Problema raiz: atomicidade entre banco de dados e mensageria

O problema mais fundamental de sistemas assíncronos não é desempenho nem escala — é a impossibilidade de escrever em dois sistemas distintos atomicamente sem um protocolo de duas fases. Considere o cenário clássico:

// CÓDIGO ERRADO — dual write sem garantia
async Task PlaceOrder(Order order) {
    await _db.SaveAsync(order);            // persiste no banco
    await _queue.PublishAsync(new OrderPlaced(order.Id)); // publica na fila
}

Entre o SaveAsync e o PublishAsync, o processo pode morrer. O pedido foi salvo mas o evento nunca foi publicado. Os consumidores nunca souberam do pedido. Ou, em outro cenário: o banco falha depois da publicação — o evento foi para a fila mas o pedido não existe no banco. O sistema está inconsistente em ambos os casos, e não há nenhuma transação global que corrija isso.

Duas soluções para esse problema: o Outbox Pattern (escreve evento no banco junto com os dados na mesma transação, publica depois) e o Event Sourcing (o evento é o estado — salvar o evento é a operação primária, não existe banco de domínio separado).

Outbox Pattern

O Outbox resolve o dual-write problem usando a própria transação do banco para persistir eventos antes de publicá-los. A tabela de outbox funciona como buffer transacional: na mesma transação que modifica os dados de negócio, você insere o evento na tabela outbox_events. Um processo separado (relay) lê a outbox e publica no broker. Se a publicação falha, o relay tenta novamente — o evento está salvo e nunca é perdido.

-- Esquema da outbox
CREATE TABLE outbox_events (
    id          UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    aggregate_type  TEXT NOT NULL,   -- 'Order', 'Payment'
    aggregate_id    TEXT NOT NULL,   -- ID do agregado
    event_type      TEXT NOT NULL,   -- 'OrderPlaced', 'PaymentFailed'
    payload         JSONB NOT NULL,
    occurred_at     TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    published_at    TIMESTAMPTZ,     -- NULL = pendente
    sequence_nr     BIGSERIAL        -- para ordenação por agregado
);

CREATE INDEX outbox_pending ON outbox_events (occurred_at)
    WHERE published_at IS NULL;

O relay pode ser implementado de duas formas: polling (consulta periódica de eventos pendentes) ou CDC — Change Data Capture (lê o transaction log do banco via Debezium ou similar). CDC é preferível em produção porque reage imediatamente às mudanças sem polling e sem carga extra no banco.

Variante: Inbox Pattern

O Inbox é o espelho do Outbox, aplicado ao consumidor. Quando uma mensagem chega, você insere na tabela inbox_events e processa em transação separada. A inserção no inbox serve como deduplicação: se a mesma mensagem chegar duas vezes (broker entrega em duplicata), a segunda inserção viola a constraint de unicidade e é ignorada. O processamento acontece exatamente uma vez.

CREATE TABLE inbox_events (
    message_id    TEXT PRIMARY KEY,  -- ID único fornecido pelo broker
    event_type    TEXT NOT NULL,
    payload       JSONB NOT NULL,
    received_at   TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    processed_at  TIMESTAMPTZ
);

Saga Pattern

Uma Saga é uma sequência de transações locais coordenadas por mensagens, onde cada etapa publica eventos/comandos que disparam a próxima etapa. Se uma etapa falha, a saga executa transações compensatórias que desfazem o efeito das etapas anteriores. Sagas são a alternativa a transações distribuídas (2PC) em sistemas assíncronos — em vez de bloquear recursos através de serviços, cada serviço confirma localmente e o sistema converge para consistência eventual.

Existem dois estilos de coordenação:

Coreografia (choreography)

Cada serviço reage a eventos publicados por outros serviços sem um coordenador central. A saga emerge do comportamento coletivo dos participantes.

Order Service                Payment Service           Inventory Service
     │                              │                          │
     │ OrderPlaced ─────────────────►                          │
     │                              │ PaymentProcessed ────────►
     │                              │                          │ InventoryReserved
     │◄─────────────────────────────────────────────────────────
     │ OrderConfirmed               │                          │
     │                              │                          │
     │ (falha: estoque insuficiente)│                          │
     │                              │◄──────── InventoryFailed
     │◄──────── PaymentRefunded     │                          │
     │ OrderCancelled               │                          │

Prós: baixo acoplamento, sem SPOF central, cada serviço é autônomo. Contras: difícil de rastrear o fluxo completo da saga, lógica de compensação espalhada, hard to debug quando algo quebra no meio.

Orquestração (orchestration)

Um orquestrador central (saga orchestrator ou process manager) coordena os participantes enviando comandos e recebendo respostas. O orquestrador sabe em qual estado a saga está e o que fazer a seguir.

OrderSagaOrchestrator
     │
     ├── Envia: ReserveInventory → Inventory Service
     │   └── Recebe: InventoryReserved ou InventoryFailed
     │
     ├── Envia: ProcessPayment → Payment Service
     │   └── Recebe: PaymentProcessed ou PaymentFailed
     │
     ├── Envia: ConfirmOrder → Order Service
     │   └── Recebe: OrderConfirmed
     │
     └── Em falha: Envia compensações na ordem inversa
         ├── ReleaseInventory → Inventory Service
         └── RefundPayment → Payment Service

Prós: fluxo visível em um lugar, fácil de adicionar steps, rastreamento natural, lógica de compensação centralizada. Contras: o orquestrador se torna um componente crítico, pode virar um anti-pattern "God orchestrator" se absorver lógica de negócio dos participantes.

Event Sourcing

Event Sourcing inverte o modelo tradicional de persistência. Em vez de armazenar o estado atual de uma entidade (a linha na tabela orders reflete o estado mais recente), você armazena a sequência de eventos que levou a esse estado. O estado atual é derivado — reconstruído aplicando os eventos em ordem. O event store é a fonte de verdade; o estado atual é uma projeção.

-- Modelo tradicional (state-based)
orders: { id, status: "confirmed", total: 150.00, ... }

-- Event Sourcing (event-based)
order_events:
  { order_id, seq: 1, type: "OrderPlaced",    data: {items: [...], total: 150.00} }
  { order_id, seq: 2, type: "PaymentApplied", data: {method: "credit", amount: 150.00} }
  { order_id, seq: 3, type: "OrderConfirmed", data: {confirmed_at: "..."} }

-- Estado atual: reduce(events, applyEvent, initialState)

Propriedades que emergem do Event Sourcing: audit log completo (cada mudança de estado é rastreada, com quem/quando/por quê), time travel (reconstituir o estado em qualquer ponto no tempo), retroactive corrections (adicionar novos eventos que corrigem interpretações antigas sem destruir histórico), e múltiplas projeções (o mesmo stream de eventos pode gerar leituras otimizadas para diferentes casos de uso).

O preço é real: reconstruir estado de uma sequência longa de eventos é caro. A solução são snapshots — periodicamente persistir o estado atual como checkpoint, e ao reconstruir, partir do snapshot mais recente em vez do início do stream. Um snapshot do evento 1000 permite reconstruir o estado atual carregando apenas os eventos de 1001 em diante.

CQRS — Command Query Responsibility Segregation

CQRS separa o modelo de escrita (commands) do modelo de leitura (queries). Em vez de uma única API que tanto lê quanto escreve usando o mesmo modelo de domínio, você tem dois lados: o command side processa mutações e produz eventos; o query side mantém projeções otimizadas para leitura, construídas a partir desses eventos.

┌─────────────────────────────────────────────────────────┐
│                       Cliente                           │
│                          │                              │
│          ┌───────────────┴───────────────┐              │
│          ▼                               ▼              │
│   Command Handler                  Query Handler        │
│   (PlaceOrder, etc.)               (GetOrder, etc.)     │
│          │                               │              │
│          ▼                               ▼              │
│   Event Store / DB            Read Models (Redis,       │
│   (escrita normalizada)        Elasticsearch, views)    │
│          │                               ▲              │
│          └───── Event Projector ─────────┘              │
│                 (consome eventos, atualiza read models)  │
└─────────────────────────────────────────────────────────┘

O benefício prático: o modelo de leitura é uma projeção desnormalizada, construída especificamente para as consultas que o cliente faz. Não há joins, não há mapeamento ORM complexo. O modelo de escrita é normalizado e reflete o domínio com precisão. Os dois evoluem independentemente. Se o padrão de leitura muda (novo relatório, nova tela), você cria uma nova projeção sem tocar no modelo de escrita.

CQRS tem graus. No mínimo, é só separar os métodos de leitura dos de escrita no mesmo repositório. No máximo, são bancos de dados físicos separados, stores diferentes, e consistência eventual entre o lado de escrita e as projeções de leitura. A consistência eventual é o preço do CQRS completo: após um comando, a projeção de leitura pode estar alguns milissegundos (ou mais) desatualizada.

Event-carried State Transfer

Quando serviços precisam de dados de outros serviços, existem duas opções: perguntar ao serviço dono na hora da necessidade (query síncrona ou assíncrona) ou receber os dados junto com o evento que notifica sobre mudanças. O segundo padrão é Event-Carried State Transfer (ECST).

// Sem ECST: Order Service precisa consultar Customer Service para obter o endereço
{
  "type": "OrderPlaced",
  "order_id": "ord-123",
  "customer_id": "cust-456"  // receptor precisa buscar o endereço separadamente
}

// Com ECST: dados relevantes viajam junto com o evento
{
  "type": "OrderPlaced",
  "order_id": "ord-123",
  "customer_id": "cust-456",
  "shipping_address": {     // receptor tem tudo que precisa
    "street": "Av. Paulista, 1000",
    "city": "São Paulo",
    "postal_code": "01310-100"
  }
}

ECST elimina chamadas síncronas para enriquecer eventos — o receptor é autossuficiente com os dados do evento. O trade-off: os eventos ficam maiores, e se o schema dos dados embutidos mudar, os consumidores podem precisar de migração. Use ECST para dados que raramente mudam e que múltiplos consumidores precisarão — endereços, nomes de usuário, dados de produto no momento da compra (importante: snapshot do dado no momento do evento, não referência mutável).

Transactional Messaging

Transactional messaging é o conjunto de práticas que garante que a publicação de mensagens seja atômica com as operações de negócio. O Outbox é a implementação mais comum, mas existem outras abordagens:

CDC (Change Data Capture): Debezium captura mudanças no transaction log do Postgres/MySQL e as transforma em eventos Kafka. O banco é a fonte de verdade; o Kafka é derivado. Não requer alteração no código de aplicação — mas o schema do banco vira o contrato do evento, o que é perigoso.

Exactly-once com Kafka transactions: Kafka suporta transações nativas desde 0.11 — o produtor pode abrir uma transação, publicar em múltiplos tópicos, e commitar ou abortar atomicamente. Combinado com read_committed no consumidor, garante exactly-once dentro do ecossistema Kafka. A limitação: funciona apenas Kafka-to-Kafka. Se a operação envolve um banco de dados externo, volta ao problema do dual write.

Two-Phase Commit (2PC): Coordina uma transação distribuída através de múltiplos recursos. Funciona, mas tem locking de longa duração, é blocking se o coordenador falha, e poucos brokers de mensagens modernos suportam XA transactions. Evite em sistemas de alta escala.

Comparativo dos padrões

Comparativo entre linguagens — Outbox Pattern

O Outbox é o padrão mais universal — aparece em quase todo sistema que combina banco de dados com mensageria. A implementação difere em como cada ecossistema lida com transações e polling.

// C# — Outbox com EF Core e MassTransit Outbox integrado
// MassTransit tem suporte nativo ao Outbox — sem polling manual

// Program.cs
services.AddMassTransit(x => {
    x.AddEntityFrameworkOutbox<AppDbContext>(o => {
        o.UsePostgres();          // dialeto do banco
        o.UseBusOutbox();         // relay integrado ao bus
        o.QueryDelay = TimeSpan.FromSeconds(5);
        o.QueryTimeout = TimeSpan.FromSeconds(30);
    });
    x.AddConsumer<OrderPlacedConsumer>();
    x.UsingRabbitMq((ctx, cfg) => cfg.ConfigureEndpoints(ctx));
});

// Entidade de outbox gerada pelo EF Core (tabela outbox_message)
// Criada automaticamente via migration

// Handler: a publicação acontece dentro da transação do EF Core
public class PlaceOrderHandler(AppDbContext db, IPublishEndpoint bus) {
    public async Task Handle(PlaceOrderCommand cmd, CancellationToken ct) {
        var order = Order.Place(cmd.CustomerId, cmd.Items);

        db.Orders.Add(order);

        // Publicação vai para a tabela de outbox, não para o RabbitMQ diretamente
        await bus.Publish(new OrderPlaced(order.Id, order.CustomerId), ct);

        // SaveChanges persiste tanto o Order quanto a mensagem do outbox
        // na mesma transação — atomicidade garantida
        await db.SaveChangesAsync(ct);

        // O relay do MassTransit lê a outbox e publica no RabbitMQ em background
    }
}

// Para Event Sourcing com Marten (PostgreSQL event store)
services.AddMarten(opts => {
    opts.Connection(connectionString);
    opts.Events.AddEventType<OrderPlaced>();
    opts.Events.AddEventType<PaymentApplied>();
}).AddAsyncDaemon(DaemonMode.HotCold); // daemon de projeção

public class OrderProjection : MultiStreamProjection<OrderReadModel, Guid> {
    public OrderProjection() {
        Identity<OrderPlaced>(e => e.OrderId);
        Identity<PaymentApplied>(e => e.OrderId);
    }

    public void Apply(OrderReadModel model, OrderPlaced evt) {
        model.Status = "placed";
        model.CustomerId = evt.CustomerId;
    }

    public void Apply(OrderReadModel model, PaymentApplied evt) {
        model.Status = "paid";
        model.PaidAt = evt.PaidAt;
    }
}

# Python — Outbox com SQLAlchemy e polling manual
# Não existe biblioteca equivalente ao MassTransit; polling é o padrão

import asyncio
import json
import uuid
from datetime import datetime, timezone
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import text
import aio_pika

# Tabela de outbox (SQLAlchemy Core ou ORM)
# CREATE TABLE outbox_events (
#     id UUID PRIMARY KEY,
#     aggregate_type TEXT,
#     aggregate_id TEXT,
#     event_type TEXT,
#     payload JSONB,
#     occurred_at TIMESTAMPTZ,
#     published_at TIMESTAMPTZ
# )

async def place_order(session: AsyncSession, command: dict) -> str:
    order_id = str(uuid.uuid4())

    # Operação de negócio
    await session.execute(
        text("INSERT INTO orders (id, customer_id, status) VALUES (:id, :cid, 'placed')"),
        {"id": order_id, "cid": command["customer_id"]}
    )

    # Inserção no outbox NA MESMA transação
    event_payload = json.dumps({
        "order_id": order_id,
        "customer_id": command["customer_id"],
        "items": command["items"],
    })
    await session.execute(
        text("""
            INSERT INTO outbox_events (id, aggregate_type, aggregate_id, event_type, payload, occurred_at)
            VALUES (:id, 'Order', :agg_id, 'OrderPlaced', :payload::jsonb, :now)
        """),
        {
            "id": str(uuid.uuid4()),
            "agg_id": order_id,
            "payload": event_payload,
            "now": datetime.now(timezone.utc),
        }
    )

    await session.commit()  # banco + outbox atomicamente
    return order_id


# Relay: polling da outbox e publicação no broker
async def outbox_relay(session_factory, amqp_url: str):
    connection = await aio_pika.connect_robust(amqp_url)
    channel = await connection.channel()
    exchange = await channel.declare_exchange("domain-events", aio_pika.ExchangeType.TOPIC, durable=True)

    while True:
        async with session_factory() as session:
            rows = await session.execute(
                text("""
                    SELECT id, event_type, payload
                    FROM outbox_events
                    WHERE published_at IS NULL
                    ORDER BY occurred_at
                    LIMIT 100
                    FOR UPDATE SKIP LOCKED  -- evita conflito com outras instâncias do relay
                """)
            )
            events = rows.fetchall()

            for event_id, event_type, payload in events:
                routing_key = event_type.lower().replace(".", "_")
                await exchange.publish(
                    aio_pika.Message(
                        body=json.dumps(payload).encode(),
                        content_type="application/json",
                        message_id=str(event_id),
                        delivery_mode=aio_pika.DeliveryMode.PERSISTENT,
                    ),
                    routing_key=routing_key,
                )
                await session.execute(
                    text("UPDATE outbox_events SET published_at = NOW() WHERE id = :id"),
                    {"id": event_id}
                )

            await session.commit()

        await asyncio.sleep(1)  # intervalo de polling

// Go — Outbox com pgx e watermill como event bus
// watermill tem suporte a Outbox com Postgres publisher

package outbox

import (
    "context"
    "encoding/json"
    "time"

    "github.com/ThreeDotsLabs/watermill"
    "github.com/ThreeDotsLabs/watermill-sql/v2/pkg/sql"
    "github.com/ThreeDotsLabs/watermill/message"
    "github.com/jackc/pgx/v5"
)

type OrderEvent struct {
    OrderID    string `json:"order_id"`
    CustomerID string `json:"customer_id"`
}

// PlaceOrder: persiste pedido e evento no outbox na mesma transação
func PlaceOrder(ctx context.Context, tx pgx.Tx, pub message.Publisher, cmd PlaceOrderCmd) (string, error) {
    orderID := uuid.New().String()

    _, err := tx.Exec(ctx,
        "INSERT INTO orders (id, customer_id, status) VALUES ($1, $2, 'placed')",
        orderID, cmd.CustomerID,
    )
    if err != nil {
        return "", fmt.Errorf("inserir pedido: %w", err)
    }

    payload, _ := json.Marshal(OrderEvent{OrderID: orderID, CustomerID: cmd.CustomerID})
    msg := message.NewMessage(watermill.NewUUID(), payload)

    // Publisher SQL do watermill escreve na tabela de outbox
    // usando a transação pgx existente
    if err := pub.Publish("OrderPlaced", msg); err != nil {
        return "", fmt.Errorf("outbox publish: %w", err)
    }

    return orderID, nil // caller faz tx.Commit()
}

// Setup do publisher outbox (watermill-sql)
func NewOutboxPublisher(db *sql.DB, logger watermill.LoggerAdapter) (message.Publisher, error) {
    return sql.NewPublisher(db, sql.PublisherConfig{
        SchemaAdapter:        sql.DefaultPostgreSQLSchema{},
        AutoInitializeSchema: true,
    }, logger)
}

// Subscriber: relay que lê outbox e publica no AMQP/Kafka
func StartRelay(ctx context.Context, db *sql.DB, amqpPublisher message.Publisher, logger watermill.LoggerAdapter) error {
    subscriber, err := sql.NewSubscriber(db, sql.SubscriberConfig{
        SchemaAdapter:    sql.DefaultPostgreSQLSchema{},
        OffsetsAdapter:   sql.DefaultPostgreSQLOffsetsAdapter{},
        InitializeSchema: true,
    }, logger)
    if err != nil {
        return err
    }

    router, err := message.NewRouter(message.RouterConfig{}, logger)
    if err != nil {
        return err
    }

    // Rota: outbox → AMQP
    router.AddHandler(
        "outbox-relay",
        "OrderPlaced",        // tópico SQL (outbox)
        subscriber,
        "order.placed",       // tópico AMQP de destino
        amqpPublisher,
        func(msg *message.Message) ([]*message.Message, error) {
            return message.Messages{msg}, nil // passthrough
        },
    )

    return router.Run(ctx)
}

Exercícios práticos

1. Implemente o Outbox Pattern do zero sem bibliotecas: crie a tabela outbox_events, escreva o handler que insere evento junto com dados de negócio na mesma transação, e implemente o relay com FOR UPDATE SKIP LOCKED. Simule falha do processo entre o commit e a publicação (kill -9) e valide que o relay recupera e publica todos os eventos pendentes ao reiniciar.
   Critério: nenhum evento é perdido após kill do processo entre commit e publicação; relay reprocessa sem duplicar eventos já publicados.
2. Implemente uma Saga de criação de pedido com orquestração: Order Service → Reserve Inventory → Process Payment → Confirm Order. Inclua compensação para cada etapa (cancelar reserva, estornar pagamento). Use uma máquina de estados explícita no orquestrador com estados PENDING | INVENTORY_RESERVED | PAYMENT_PROCESSED | CONFIRMED | COMPENSATING | FAILED.
   Critério: Quando Payment falha, InventoryReservation é cancelada automaticamente; o estado final da saga é FAILED e o Order não aparece como confirmado.
3. Compare coreografia vs orquestração para o mesmo fluxo de confirmação de pedido. Depois de implementar os dois, adicione o requisito de envio de email após confirmação. Documente: quantos serviços foram modificados em cada abordagem, onde a lógica da saga é visível, e como você rastrearia um fluxo específico em produção.
   Critério: análise escrita com métricas concretas — número de arquivos modificados, onde está a lógica de negócio do fluxo em cada abordagem.
4. Implemente Event Sourcing para o agregado BankAccount com eventos AccountOpened, MoneyDeposited, MoneyWithdrawn. Adicione snapshots a cada 50 eventos. Implemente GetBalanceAt(DateTime point) que reconstrói o saldo exato em qualquer ponto no tempo. Crie duas projeções separadas: saldo atual e histórico do último mês.
   Critério: GetBalanceAt retorna saldo correto para timestamps arbitrários; a projeção de saldo atual é consistente com a reconstrução completa do evento store.
5. Combine CQRS com Event Sourcing: o command side usa event store (Postgres + tabela de eventos), o query side mantém uma projeção Redis com saldo atual atualizada por um projector assíncrono. Meça a latência de consistência entre escrita e leitura. Implemente "read your own writes" usando o número de sequência do último evento do usuário para esperar pela projeção antes de retornar.
   Critério: p99 de consistência da projeção medido e documentado; "read your own writes" funciona corretamente sem polling incondicional.