temporal_workflows_fixed.py

# temporal_version/workflows.py
from datetime import timedelta
from dataclasses import dataclass
from typing import List
from temporalio import workflow
from temporalio.common import RetryPolicy

import temporal_activities

@dataclass
class ProcessOrderRequest:
    order_id: str
    customer_id: str
    items: List[dict]
    total_amount: float

@workflow.defn
class OrderProcessingWorkflow:
    """
    Temporal workflow for processing e-commerce orders.
    
    This workflow orchestrates the entire order processing pipeline:
    1. Check inventory availability
    2. Process payment
    3. Reserve inventory
    4. Ship order
    5. Send notification
    
    Key Temporal features demonstrated:
    - Automatic state persistence
    - Built-in retry mechanisms
    - Timeout handling
    - Saga pattern for compensation
    - Deterministic execution
    """
    
    @workflow.run
    async def run(self, request: ProcessOrderRequest) -> temporal_activities.OrderDetails:
        """
        Main workflow execution method.
        
        Temporal automatically persists workflow state between each activity.
        If the worker crashes, the workflow will resume from the last completed activity.
        """
        
        # Configure retry policy for activities
        retry_policy = RetryPolicy(
            initial_interval=timedelta(seconds=1),
            maximum_interval=timedelta(seconds=10),
            maximum_attempts=3,
            backoff_coefficient=2.0,
        )
        
        # Activity options with timeouts and retries
        activity_options = {
            "start_to_close_timeout": timedelta(seconds=30),
            "retry_policy": retry_policy,
        }
        
        # Initialize order details
        order = temporal_activities.OrderDetails(
            order_id=request.order_id,
            customer_id=request.customer_id,
            items=request.items,
            total_amount=request.total_amount,
            status="processing"
        )
        
        try:
            # Step 1: Check Inventory
            workflow.logger.info(f"Starting inventory check for order {order.order_id}")
            
            inventory_request = temporal_activities.InventoryCheckRequest(items=request.items)
            inventory_available = await workflow.execute_activity(
                temporal_activities.OrderProcessingActivities.check_inventory,
                inventory_request,
                **activity_options
            )
            
            if not inventory_available:
                order.status = "failed"
                order.failure_reason = "Insufficient inventory"
                return order
            
            workflow.logger.info(f"Inventory check passed for order {order.order_id}")
            
            # Step 2: Process Payment
            workflow.logger.info(f"Processing payment for order {order.order_id}")
            
            payment_request = temporal_activities.PaymentRequest(
                customer_id=request.customer_id,
                amount=request.total_amount,
                order_id=request.order_id
            )
            
            payment_successful = await workflow.execute_activity(
                temporal_activities.OrderProcessingActivities.process_payment,
                payment_request,
                **activity_options
            )
            
            if not payment_successful:
                order.status = "failed"
                order.failure_reason = "Payment processing failed"
                return order
            
            workflow.logger.info(f"Payment processed for order {order.order_id}")
            
            # Step 3: Reserve Inventory
            workflow.logger.info(f"Reserving inventory for order {order.order_id}")
            
            reservation_request = temporal_activities.InventoryReservationRequest(
                items=request.items,
                order_id=request.order_id
            )
            
            try:
                reservation_successful = await workflow.execute_activity(
                    temporal_activities.OrderProcessingActivities.reserve_inventory,
                    reservation_request,
                    **activity_options
                )
                
                if not reservation_successful:
                    # Compensate: rollback payment
                    await self._rollback_payment(request.order_id, request.total_amount)
                    order.status = "failed"
                    order.failure_reason = "Inventory reservation failed"
                    return order
                    
            except Exception as e:
                # Compensate: rollback payment
                await self._rollback_payment(request.order_id, request.total_amount)
                order.status = "failed"
                order.failure_reason = f"Inventory reservation error: {str(e)}"
                return order
            
            workflow.logger.info(f"Inventory reserved for order {order.order_id}")
            
            # Step 4: Ship Order
            workflow.logger.info(f"Shipping order {order.order_id}")
            
            shipping_request = temporal_activities.ShippingRequest(
                order_id=request.order_id,
                customer_id=request.customer_id,
                items=request.items
            )
            
            try:
                shipping_successful = await workflow.execute_activity(
                    temporal_activities.OrderProcessingActivities.ship_order,
                    shipping_request,
                    **activity_options
                )
                
                if not shipping_successful:
                    # Compensate: rollback payment and inventory
                    await self._rollback_payment(request.order_id, request.total_amount)
                    await self._rollback_inventory(request.items, request.order_id)
                    order.status = "failed"
                    order.failure_reason = "Shipping failed"
                    return order
                    
            except Exception as e:
                # Compensate: rollback payment and inventory
                await self._rollback_payment(request.order_id, request.total_amount)
                await self._rollback_inventory(request.items, request.order_id)
                order.status = "failed"
                order.failure_reason = f"Shipping error: {str(e)}"
                return order
            
            workflow.logger.info(f"Order {order.order_id} shipped successfully")
            
            # Step 5: Send Notification (non-critical, don't fail order if this fails)
            workflow.logger.info(f"Sending notification for order {order.order_id}")
            
            notification_request = temporal_activities.NotificationRequest(
                order_id=request.order_id,
                customer_id=request.customer_id,
                message=f"Your order {request.order_id} has been shipped!"
            )
            
            try:
                await workflow.execute_activity(
                    temporal_activities.OrderProcessingActivities.send_notification,
                    notification_request,
                    start_to_close_timeout=timedelta(seconds=10),
                    retry_policy=RetryPolicy(maximum_attempts=2)  # Fewer retries for notifications
                )
                workflow.logger.info(f"Notification sent for order {order.order_id}")
            except Exception as e:
                workflow.logger.warning(f"Notification failed for order {order.order_id}: {e}")
                # Don't fail the order for notification failure
            
            # Order completed successfully
            order.status = "completed"
            workflow.logger.info(f"Order {order.order_id} completed successfully")
            
            return order
            
        except Exception as e:
            workflow.logger.error(f"Workflow failed for order {order.order_id}: {e}")
            order.status = "failed"
            order.failure_reason = str(e)
            return order
    
    async def _rollback_payment(self, order_id: str, amount: float):
        """Compensating transaction: rollback payment"""
        workflow.logger.info(f"Rolling back payment for order {order_id}")
        
        try:
            await workflow.execute_activity(
                temporal_activities.OrderProcessingActivities.rollback_payment,
                {"order_id": order_id, "amount": amount},
                start_to_close_timeout=timedelta(seconds=30),
                retry_policy=RetryPolicy(maximum_attempts=5)  # More retries for rollbacks
            )
        except Exception as e:
            workflow.logger.error(f"Payment rollback failed for order {order_id}: {e}")
            # In production, this might trigger alerts or manual intervention
    
    async def _rollback_inventory(self, items: List[dict], order_id: str):
        """Compensating transaction: rollback inventory reservation"""
        workflow.logger.info(f"Rolling back inventory for order {order_id}")
        
        try:
            await workflow.execute_activity(
                temporal_activities.OrderProcessingActivities.rollback_inventory,
                {"items": items, "order_id": order_id},
                start_to_close_timeout=timedelta(seconds=30),
                retry_policy=RetryPolicy(maximum_attempts=5)  # More retries for rollbacks
            )
        except Exception as e:
            workflow.logger.error(f"Inventory rollback failed for order {order_id}: {e}")
            # In production, this might trigger alerts or manual intervention

@workflow.defn
class OrderApprovalWorkflow:
    """
    Workflow for orders requiring manual approval.
    
    Demonstrates:
    - Human-in-the-loop workflows
    - Signals for external input
    - Timeouts for approval processes
    """
    
    def __init__(self):
        self.approved = False
        self.approval_received = False
    
    @workflow.run
    async def run(self, request: ProcessOrderRequest) -> temporal_activities.OrderDetails:
        """
        Approval workflow with timeout.
        """
        
        workflow.logger.info(f"Order {request.order_id} requires approval")
        
        # Wait for approval signal with timeout
        try:
            await workflow.wait_condition(
                lambda: self.approval_received,
                timeout=timedelta(minutes=30)  # 30 minute approval timeout
            )
            
            if self.approved:
                workflow.logger.info(f"Order {request.order_id} approved, proceeding with processing")
                
                # Delegate to main processing workflow
                processing_workflow = OrderProcessingWorkflow()
                return await processing_workflow.run(request)
            else:
                workflow.logger.info(f"Order {request.order_id} rejected")
                return temporal_activities.OrderDetails(
                    order_id=request.order_id,
                    customer_id=request.customer_id,
                    items=request.items,
                    total_amount=request.total_amount,
                    status="rejected",
                    failure_reason="Order rejected during approval process"
                )
                
        except Exception as e:
            workflow.logger.error(f"Approval timeout for order {request.order_id}")
            return temporal_activities.OrderDetails(
                order_id=request.order_id,
                customer_id=request.customer_id,
                items=request.items,
                total_amount=request.total_amount,
                status="failed",
                failure_reason="Approval timeout - no response within 30 minutes"
            )
    
    @workflow.signal
    async def approve_order(self, approved: bool):
        """Signal handler for order approval"""
        self.approved = approved
        self.approval_received = True
        workflow.logger.info(f"Approval signal received: {approved}")