Distributed Transactions: 2PC & Saga

-- Outbox pattern: write event atomically with business data
BEGIN;
  -- Business operation
  INSERT INTO orders (id, user_id, status, total)
  VALUES ('ord-123', 'usr-456', 'CONFIRMED', 99.99);

  -- Outbox event (same transaction!)
  INSERT INTO outbox_events (id, aggregate_type, aggregate_id, event_type, payload, created_at)
  VALUES (
    gen_random_uuid(),
    'Order',
    'ord-123',
    'OrderConfirmed',
    '{"orderId":"ord-123","userId":"usr-456","total":99.99}',
    NOW()
  );
COMMIT;

-- Separate outbox publisher process (Debezium or custom poller):
-- SELECT * FROM outbox_events WHERE published = false ORDER BY created_at LIMIT 100;
-- Publish to Kafka, then mark as published.
-- At-least-once delivery — consumers must be idempotent!

from enum import Enum
from dataclasses import dataclass, field
from typing import Callable
import asyncio, logging

class StepState(Enum):
    PENDING = "pending"
    RUNNING = "running"
    DONE = "done"
    FAILED = "failed"
    COMPENSATING = "compensating"
    COMPENSATED = "compensated"

@dataclass
class SagaStep:
    name: str
    action: Callable          # The forward action
    compensation: Callable    # Undo action if later step fails
    state: StepState = StepState.PENDING
    result: dict = field(default_factory=dict)

class SagaOrchestrator:
    def __init__(self, saga_id: str, steps: list[SagaStep]):
        self.saga_id = saga_id
        self.steps = steps
        self.completed = []   # Track for compensation

    async def execute(self) -> bool:
        for step in self.steps:
            step.state = StepState.RUNNING
            try:
                step.result = await step.action()
                step.state = StepState.DONE
                self.completed.append(step)
                logging.info(f"[{self.saga_id}] {step.name}: DONE")
            except Exception as e:
                step.state = StepState.FAILED
                logging.error(f"[{self.saga_id}] {step.name}: FAILED — {e}")
                await self._compensate()
                return False
        return True

    async def _compensate(self):
        """Roll back completed steps in reverse order."""
        for step in reversed(self.completed):
            step.state = StepState.COMPENSATING
            try:
                await step.compensation()
                step.state = StepState.COMPENSATED
                logging.info(f"[{self.saga_id}] {step.name}: COMPENSATED")
            except Exception as e:
                # Compensation failed — requires manual intervention!
                logging.critical(f"[{self.saga_id}] Compensation failed for {step.name}: {e}")
                # Alert on-call, add to dead letter queue

# Usage
async def book_order(order_id: str):
    saga = SagaOrchestrator(
        saga_id=order_id,
        steps=[
            SagaStep("CreateOrder", create_order, cancel_order),
            SagaStep("ProcessPayment", charge_card, refund_card),
            SagaStep("ReserveInventory", reserve_items, release_items),
            SagaStep("ScheduleShipping", schedule_ship, cancel_ship),
        ]
    )
    success = await saga.execute()
    return {"status": "completed" if success else "rolled_back"}