Sagas vs 2PC: transações distribuídas sem perder o sono

# order_saga_workflow.py — orchestration com Temporal (temporal.io)
from datetime import timedelta
from temporalio import workflow
from temporalio.common import RetryPolicy

@workflow.defn
class OrderSagaWorkflow:
    @workflow.run
    async def run(self, order: dict) -> dict:
        # Passo 1: reservar estoque
        reservation_id = await workflow.execute_activity(
            stock_reserve, order, start_to_close_timeout=timedelta(seconds=30),
            retry_policy=RetryPolicy(maximum_attempts=5),
        )
        try:
            # Passo 2: cobrar cartão
            charge_id = await workflow.execute_activity(
                payment_charge, order, start_to_close_timeout=timedelta(seconds=30),
            )
            try:
                # Passo 3: agendar entrega
                shipment_id = await workflow.execute_activity(
                    shipping_schedule, order, start_to_close_timeout=timedelta(seconds=30),
                )
                return {"status": "ok", "order_id": order["id"]}
            except Exception as e:
                # compensa: reembolsar cartão
                await workflow.execute_activity(
                    payment_refund, charge_id,
                    start_to_close_timeout=timedelta(seconds=60),
                    retry_policy=RetryPolicy(maximum_attempts=10),  # crítico, retria muito
                )
                raise
        except Exception as e:
            # compensa: liberar estoque
            await workflow.execute_activity(
                stock_release, reservation_id,
                start_to_close_timeout=timedelta(seconds=60),
            )
            raise

# order_service.py — outbox garantindo atomicidade
async def create_order(data: dict) -> dict:
    async with db.begin() as tx:
        # 1. Inserir pedido na tabela de domínio
        order = await tx.execute(
            text("INSERT INTO orders (...) VALUES (...) RETURNING *"),
            data,
        )
        order_row = order.one()

        # 2. Inserir evento na outbox — MESMA transação
        await tx.execute(
            text("""
                INSERT INTO outbox (aggregate_id, event_type, payload, created_at)
                VALUES (:id, 'OrderCreated', :payload, NOW())
            """),
            {"id": order_row.id, "payload": json.dumps(data)},
        )
        # commit = ambos salvos ou nenhum

    return {"id": order_row.id}


# relay.py — worker separado publicando
async def relay_loop():
    while True:
        async with db.begin() as tx:
            rows = await tx.execute(text("""
                SELECT id, event_type, payload FROM outbox
                WHERE sent_at IS NULL
                ORDER BY id
                LIMIT 100
                FOR UPDATE SKIP LOCKED
            """))
            events = rows.fetchall()
            for e in events:
                await kafka.publish(topic="orders", key=str(e.id), value=e.payload)
                await tx.execute(
                    text("UPDATE outbox SET sent_at = NOW() WHERE id = :id"),
                    {"id": e.id},
                )
        await asyncio.sleep(0.1)