Capstone: saga distribuída ponta a ponta

# Temporal workflow — state machine durável
from temporalio import workflow, activity
from datetime import timedelta

@activity.defn
async def create_order(input: OrderInput) -> str: ...

@activity.defn
async def cancel_order(order_id: str) -> None: ...

@activity.defn
async def charge_payment(order_id: str, amount: int, idempotency_key: str) -> str: ...

@activity.defn
async def refund_payment(charge_id: str, idempotency_key: str) -> None: ...

@activity.defn
async def reserve_inventory(order_id: str, items: list) -> str: ...

@activity.defn
async def release_inventory(reservation_id: str) -> None: ...

@workflow.defn
class CheckoutSaga:
    @workflow.run
    async def run(self, input: CheckoutInput) -> str:
        compensations = []
        try:
            order_id = await workflow.execute_activity(
                create_order, input,
                start_to_close_timeout=timedelta(seconds=10),
                retry_policy={"maximum_attempts": 3},
            )
            compensations.append(lambda: workflow.execute_activity(cancel_order, order_id))

            charge_id = await workflow.execute_activity(
                charge_payment, order_id, input.amount, input.idempotency_key,
                start_to_close_timeout=timedelta(seconds=15),
            )
            compensations.append(lambda: workflow.execute_activity(
                refund_payment, charge_id, input.idempotency_key
            ))

            reservation_id = await workflow.execute_activity(
                reserve_inventory, order_id, input.items,
                start_to_close_timeout=timedelta(seconds=10),
            )
            compensations.append(lambda: workflow.execute_activity(release_inventory, reservation_id))

            # ... mais steps
            return order_id

        except Exception as e:
            # Compensação na ordem inversa
            for compensate in reversed(compensations):
                try:
                    await compensate()
                except Exception as comp_error:
                    # Compensação falhou — escala pro humano
                    await workflow.execute_activity(
                        alert_ops, str(comp_error),
                        start_to_close_timeout=timedelta(seconds=5),
                    )
            raise

-- Tabela de idempotency por serviço
CREATE TABLE idempotency_keys (
  key            TEXT PRIMARY KEY,
  request_hash   TEXT NOT NULL,           -- SHA do payload
  response       JSONB NOT NULL,
  status_code    INT NOT NULL,
  created_at     TIMESTAMPTZ DEFAULT now(),
  expires_at     TIMESTAMPTZ NOT NULL     -- TTL, ex: 24h
);

CREATE INDEX idx_idempotency_expires ON idempotency_keys (expires_at);

-- Lógica no handler:
-- 1. SELECT * FROM idempotency_keys WHERE key = $1
--    Se encontrou: retorna response salvo (comparar hash pra detectar payload diff)
--    Se não: processa + INSERT response + retorna

async def charge_payment_handler(request):
    idempotency_key = request.headers.get("Idempotency-Key")
    if not idempotency_key:
        return Response(400, "Missing Idempotency-Key header")

    # Lookup
    existing = await db.fetchone(
        "SELECT response, status_code, request_hash FROM idempotency_keys WHERE key = $1",
        idempotency_key,
    )
    if existing:
        # Se payload diferente com mesma key → erro (cliente bugado)
        if existing["request_hash"] != hash_payload(request.body):
            return Response(422, "Idempotency key reused with different payload")
        return Response(existing["status_code"], existing["response"])

    # Processa de verdade
    result = await stripe.charge(request.body)

    # Salva response
    await db.execute(
        "INSERT INTO idempotency_keys (key, request_hash, response, status_code, expires_at) "
        "VALUES ($1, $2, $3, $4, $5)",
        idempotency_key, hash_payload(request.body), result, 200, now() + timedelta(hours=24),
    )
    return Response(200, result)