Choreography Saga — Decentralized Distributed Transaction Coordination

# choreography_principles.py
# Choreography saga fundamentals

def choreography_principles():
    print("Choreography Saga Principles")
    print("=" * 40)
    print()
    
    principles = [
        {
            "principle": "Event-Driven Coordination",
            "desc": "Each service emits events that trigger the next service. No central coordinator.",
            "example": "OrderPlaced -> Payment Service starts processing"
        },
        {
            "principle": "Local Transactions Only",
            "desc": "Each service updates its own database independently within a local ACID transaction.",
            "example": "Order service saves order + publishes OrderPlaced in one DB transaction"
        },
        {
            "principle": "Compensating Events",
            "desc": "If a step fails, previous steps emit compensating events to undo their work.",
            "example": "PaymentFailed event triggers OrderCancelled in order service"
        },
        {
            "principle": "Eventual Consistency",
            "desc": "The system becomes consistent over time as events propagate. There is a window of inconsistency.",
            "example": "Order shows 'confirmed' for 100ms before inventory confirms stock"
        },
        {
            "principle": "Idempotent Handlers",
            "desc": "Event handlers must handle duplicate events safely.",
            "example": "Processing PaymentReceived twice should not charge the customer twice"
        },
    ]
    
    for p in principles:
        print(f"{p['principle']:30s}")
        print(f"  {p['desc']}")
        print(f"  Example: {p['example']}")
        print()

choreography_principles()

# saga_implementation.py
# Choreography saga implementation

def saga_implementation():
    print("Choreography Saga - Order Processing")
    print("=" * 45)
    print()
    
    saga_code = """
import json

class OrderService:
    """Step 1: Place the order."""
    
    def place_order(self, order_data):
        with self.db.transaction():
            order = self.save_order(order_data)
            self.event_bus.publish("order.placed", {
                "order_id": order.id,
                "customer_id": order.customer_id,
                "total": order.total,
                "items": order.items
            })
        return order
    
    def handle_payment_failed(self, event):
        """Compensating action: cancel the order."""
        with self.db.transaction():
            self.cancel_order(event["data"]["order_id"])
            self.event_bus.publish("order.cancelled", event["data"])

class PaymentService:
    """Step 2: Process payment."""
    
    def handle_order_placed(self, event):
        data = event["data"]
        try:
            payment = self.process_payment(
                data["customer_id"], data["total"]
            )
            with self.db.transaction():
                self.save_payment(payment)
                self.event_bus.publish("payment.received", {
                    "order_id": data["order_id"],
                    "payment_id": payment.id,
                    "amount": data["total"]
                })
        except PaymentError as e:
            with self.db.transaction():
                self.event_bus.publish("payment.failed", {
                    "order_id": data["order_id"],
                    "reason": str(e)
                })

class InventoryService:
    """Step 3: Reserve inventory."""
    
    def handle_payment_received(self, event):
        data = event["data"]
        try:
            for item in data.get("items", []):
                self.reserve_stock(item["product_id"], item["quantity"])
            with self.db.transaction():
                self.event_bus.publish("inventory.reserved", data)
        except OutOfStockError as e:
            with self.db.transaction():
                self.event_bus.publish("inventory.failed", {
                    **data, "reason": str(e)
                })
    
    def handle_order_cancelled(self, event):
        """Compensating action: release reserved stock."""
        for item in event["data"].get("items", []):
            self.release_stock(item["product_id"], item["quantity"])
"""
    print(saga_code)

saga_implementation()

# failure_scenarios.py
# Saga failure handling

def failure_scenarios():
    print("Saga Failure Scenarios and Compensation")
    print("=" * 45)
    print()
    
    scenarios = [
        {
            "step": "Order Placed",
            "failure": "Payment declined",
            "compensation": "Order cancelled (automatic)",
            "description": "Payment service emits payment.failed. Order service cancels the order."
        },
        {
            "step": "Payment Received",
            "failure": "Out of stock",
            "compensation": "Refund payment + cancel order",
            "description": "Inventory service emits inventory.failed. Payment service refunds. Order service cancels."
        },
        {
            "step": "Inventory Reserved",
            "failure": "Shipping unavailable",
            "compensation": "Release inventory + refund + cancel",
            "description": "Shipping service unavailable. Inventory releases stock. Payment refunds. Order cancels."
        },
        {
            "step": "Shipped",
            "failure": "Delivery failed",
            "compensation": "Return to sender + refund",
            "description": "After shipping, delivery fails. Return process initiated, refund issued."
        },
    ]
    
    for s in scenarios:
        print(f"Step:     {s['step']}")
        print(f"Failure:  {s['failure']}")
        print(f"Rollback: {s['compensation']}")
        print(f"Details:  {s['description']}")
        print()

failure_scenarios()

# vs_orchestration.py
# Choreography vs orchestration

def compare_saga_approaches():
    print("Choreography vs Orchestration Saga")
    print("="  * 40)
    print()
    
    comparisons = [
        ("Coordination", "Event-driven, decentralized", "Command-driven, central coordinator"),
        ("Coupling", "Loose (services know events only)", "Tighter (services know orchestrator)"),
        ("Complexity", "Higher (tracking event flow)", "Lower (single flow definition)"),
        ("Visibility", "Harder to trace flow", "Easy to see all steps"),
        ("Single Point of Failure", "No central point", "Orchestrator is SPOF"),
        ("Best For", "Simple chains, few services", "Complex flows, many services"),
        ("Testing", "Harder (distributed)", "Easier (orchestrator testable)"),
    ]
    
    print(f"{'Aspect':25s} {'Choreography':35s} {'Orchestration':35s}")
    print("-" * 95)
    for aspect, choreo, orchs in comparisons:
        print(f"{aspect:25s} {choreo:35s} {orchs:35s}")

compare_saga_approaches()

def subscription_saga():
    print("SaaS Subscription - Choreography Saga")
    print("=" * 45)
    print()
    print("Events Flow:")
    print()
    print("  1. Billing Service")
    print("     Event: subscription.created")
    print("     Comp:  subscription.cancelled (refund)")
    print()
    print("  2. Provisioning Service")
    print("     Event: resources.provisioned")
    print("     Comp:  resources.deprovisioned")
    print()
    print("  3. Notification Service")
    print("     Event: welcome.sent")
    print("     Comp:  N/A (no side effect)")
    print()
    print("  4. Analytics Service")
    print("     Event: subscription.tracked")
    print("     Comp:  N/A (logging only)")
    print()
    print("Failure: Provisioning fails after billing")
    print("  -> Provisioning emits provisioning.failed")
    print("  -> Billing handles: subscription.cancelled (refund)")
    print("  -> Notification handles: welcome.cancelled")

subscription_saga()