Consistency and Coordination.

# Preview: Consistency levels for different operations

class InventoryService:
    async def check_availability(self, product_id: str) -> int:
        """
        Display inventory count.
        Eventual consistency is fine — cached value OK.
        """
        return await self.cache.get(f"inventory:{product_id}")
    
    async def reserve_inventory(self, product_id: str, quantity: int) -> bool:
        """
        Reserve inventory for checkout.
        STRONG consistency required — must read from primary.
        """
        async with self.db.transaction(isolation="SERIALIZABLE"):
            current = await self.db.fetch_one(
                "SELECT quantity FROM inventory WHERE product_id = $1 FOR UPDATE",
                product_id
            )
            if current['quantity'] >= quantity:
                await self.db.execute(
                    "UPDATE inventory SET quantity = quantity - $2 WHERE product_id = $1",
                    product_id, quantity
                )
                return True
            return False

# Preview: Saga pattern for order processing

class OrderSaga:
    """
    Order processing with compensation.
    
    Steps:
    1. Reserve inventory
    2. Charge payment
    3. Create shipment
    
    If any step fails, compensate previous steps.
    """
    
    async def execute(self, order: Order) -> SagaResult:
        # Step 1: Reserve inventory
        reservation = await self.inventory.reserve(order.items)
        if not reservation.success:
            return SagaResult.failed("Insufficient inventory")
        
        # Step 2: Charge payment
        payment = await self.payment.charge(order.total)
        if not payment.success:
            # COMPENSATE: Release inventory
            await self.inventory.release(reservation.id)
            return SagaResult.failed("Payment failed")
        
        # Step 3: Create shipment
        shipment = await self.shipping.create(order)
        if not shipment.success:
            # COMPENSATE: Refund payment, release inventory
            await self.payment.refund(payment.id)
            await self.inventory.release(reservation.id)
            return SagaResult.failed("Shipment failed")
        
        return SagaResult.success(order_id=order.id)

# Preview: Temporal-style workflow

@workflow.defn
class OrderWorkflow:
    """
    Durable workflow for order processing.
    
    Survives crashes, retries automatically. Workflow progress is
    recorded once; activities are at-least-once, so keep them idempotent.
    """
    
    @workflow.run
    async def run(self, order: Order) -> OrderResult:
        # Each activity is durable — survives crashes
        reservation = await workflow.execute_activity(
            reserve_inventory,
            order.items,
            start_to_close_timeout=timedelta(seconds=30)
        )
        
        try:
            payment = await workflow.execute_activity(
                charge_payment,
                order.total,
                start_to_close_timeout=timedelta(seconds=60)
            )
        except ActivityError:  # e.g. payment declined (activity failures arrive wrapped)
            # Compensation activity
            await workflow.execute_activity(
                release_inventory,
                reservation.id,
                start_to_close_timeout=timedelta(seconds=30)
            )
            raise
        
        # Continue with shipment...

# Preview: CRDT-based shopping cart

class ShoppingCartCRDT:
    """
    Conflict-free shopping cart that works offline.
    
    Uses OR-Set (Observed-Remove Set) for items.
    Automatically merges without conflicts.
    """
    
    def __init__(self):
        self.items = ORSet()  # CRDT set
    
    def add_item(self, product_id: str, quantity: int):
        """Add item — always succeeds, never conflicts."""
        self.items.add((product_id, quantity, unique_tag()))
    
    def remove_item(self, product_id: str):
        """Remove item — only removes observed additions."""
        for item in self.items.lookup(product_id):
            self.items.remove(item)
    
    def merge(self, other: 'ShoppingCartCRDT'):
        """
        Merge two carts — automatic conflict resolution.
        
        If User A adds item X on phone (offline)
        And User B adds item Y on laptop (offline)
        Both items appear after merge!
        """
        self.items.merge(other.items)

# Preview: Safe distributed locking

class FencedLockService:
    """
    Distributed lock with fencing tokens.
    
    Prevents split-brain: If two nodes think they hold the lock,
    only the one with the higher fencing token succeeds.
    """
    
    async def acquire(self, resource: str, holder: str) -> Lock:
        """Acquire lock and get fencing token."""
        lock = await self.redis.set(
            f"lock:{resource}",
            holder,
            nx=True,
            ex=30
        )
        if lock:
            # Get monotonically increasing fencing token
            token = await self.redis.incr(f"fence:{resource}")
            return Lock(resource=resource, holder=holder, fence_token=token)
        return None
    
    async def execute_with_lock(
        self,
        resource: str,
        operation: Callable,
        fence_token: int
    ):
        """
        Execute operation with fencing token.
        
        Storage layer validates token — rejects stale operations.
        """
        await self.storage.execute(
            operation,
            fence_token=fence_token  # Storage validates this
        )