Multi-Step Order Processing

When a customer clicks "Place Order", a complex orchestration begins: cart validation, inventory reservation, payment processing, fraud detection, fulfillment triggering, and notification sending. Each step can fail, and failures require careful handling.

The Order Processing Challenge

23%

of orders have inventory issues

3-5%

of payments fail initially

< 3s

expected checkout time

Order Processing Pipeline

Customer clicks "Place Order"
            │
            ▼
    ┌───────────────┐
    │  Validate     │──── Invalid? ──→ Return errors
    │    Cart       │
    └───────┬───────┘
            │ Valid
            ▼
    ┌───────────────┐
    │   Reserve     │──── Out of stock? ──→ Suggest alternatives
    │  Inventory    │
    └───────┬───────┘
            │ Reserved
            ▼
    ┌───────────────┐
    │   Fraud       │──── Suspicious? ──→ Manual review
    │   Check       │
    └───────┬───────┘
            │ Clear
            ▼
    ┌───────────────┐
    │   Process     │──── Failed? ──→ Release inventory, notify
    │   Payment     │
    └───────┬───────┘
            │ Success
            ▼
    ┌───────────────┐
    │   Create      │
    │   Order       │
    └───────┬───────┘
            │
            ▼
    ┌───────────────┐
    │   Trigger     │
    │ Fulfillment   │
    └───────┬───────┘
            │
            ▼
    ┌───────────────┐
    │    Send       │
    │ Confirmation  │
    └───────────────┘

Step 1: Define Order State

from typing import TypedDict, Literal, Optional, Annotated
from datetime import datetime
import operator

class OrderItem(TypedDict):
    sku: str
    name: str
    quantity: int
    unit_price: float
    reserved: bool

class OrderState(TypedDict):
    # Order identification
    order_id: str
    idempotency_key: str  # Prevents duplicate processing

    # Customer info
    customer_id: str
    customer_email: str
    shipping_address: dict
    billing_address: dict

    # Cart
    items: list[OrderItem]
    subtotal: float
    tax: float
    shipping_cost: float
    discount: float
    total: float

    # Payment
    payment_method: dict  # {type: "card", last4: "4242", ...}
    payment_intent_id: Optional[str]
    payment_status: Literal["pending", "processing", "succeeded", "failed"]

    # Inventory
    inventory_reserved: bool
    reservation_id: Optional[str]

    # Fraud
    fraud_score: float
    fraud_check_passed: bool

    # Order status
    status: Literal[
        "cart", "validating", "reserving_inventory",
        "checking_fraud", "processing_payment", "creating_order",
        "fulfilling", "completed", "failed", "cancelled"
    ]
    error_message: Optional[str]

    # Timestamps
    created_at: str
    updated_at: str

    # Audit trail
    events: Annotated[list[dict], operator.add]

Step 2: Implement Processing Nodes

Cart Validation

def validate_cart(state: OrderState) -> dict:
    """Validate cart items and calculate totals"""
    errors = []

    # Check items exist
    if not state["items"]:
        errors.append("Cart is empty")

    # Validate each item
    for item in state["items"]:
        if item["quantity"] <= 0:
            errors.append(f"Invalid quantity for {item['sku']}")
        if item["unit_price"] <= 0:
            errors.append(f"Invalid price for {item['sku']}")

    if errors:
        return {
            "status": "failed",
            "error_message": "; ".join(errors),
            "events": [{"type": "validation_failed", "errors": errors, "at": datetime.now().isoformat()}]
        }

    # Calculate totals
    subtotal = sum(item["quantity"] * item["unit_price"] for item in state["items"])
    tax = subtotal * 0.08  # 8% tax
    shipping = 0 if subtotal > 50 else 5.99  # Free shipping over $50
    total = subtotal + tax + shipping - state.get("discount", 0)

    return {
        "subtotal": subtotal,
        "tax": round(tax, 2),
        "shipping_cost": shipping,
        "total": round(total, 2),
        "status": "validating",
        "events": [{"type": "cart_validated", "total": total, "at": datetime.now().isoformat()}]
    }

Inventory Reservation

import asyncio
from typing import Tuple

# Mock inventory service
INVENTORY = {
    "SKU-001": {"available": 50, "reserved": 0},
    "SKU-002": {"available": 3, "reserved": 2},
    "SKU-003": {"available": 0, "reserved": 0},  # Out of stock
}

async def check_and_reserve_inventory(sku: str, quantity: int) -> Tuple[bool, str]:
    """Reserve inventory for an item"""
    if sku not in INVENTORY:
        return False, f"Unknown SKU: {sku}"

    available = INVENTORY[sku]["available"] - INVENTORY[sku]["reserved"]
    if available < quantity:
        return False, f"Only {available} available for {sku}"

    # Reserve
    INVENTORY[sku]["reserved"] += quantity
    return True, "reserved"

def reserve_inventory(state: OrderState) -> dict:
    """Reserve inventory for all items in the order"""
    reservation_results = []
    all_reserved = True
    failed_items = []

    for item in state["items"]:
        success, message = asyncio.run(
            check_and_reserve_inventory(item["sku"], item["quantity"])
        )
        reservation_results.append({
            "sku": item["sku"],
            "success": success,
            "message": message
        })
        if not success:
            all_reserved = False
            failed_items.append(f"{item['name']}: {message}")

    if not all_reserved:
        return {
            "status": "failed",
            "inventory_reserved": False,
            "error_message": f"Inventory unavailable: {', '.join(failed_items)}",
            "events": [{
                "type": "inventory_reservation_failed",
                "results": reservation_results,
                "at": datetime.now().isoformat()
            }]
        }

    return {
        "status": "reserving_inventory",
        "inventory_reserved": True,
        "reservation_id": f"RES-{state['order_id']}",
        "items": [
            {**item, "reserved": True} for item in state["items"]
        ],
        "events": [{
            "type": "inventory_reserved",
            "reservation_id": f"RES-{state['order_id']}",
            "at": datetime.now().isoformat()
        }]
    }

Fraud Detection

def check_fraud(state: OrderState) -> dict:
    """Run fraud detection on the order"""
    # Simplified fraud scoring - in production use ML model or service like Stripe Radar
    fraud_score = 0.0

    # High-value orders get scrutiny
    if state["total"] > 500:
        fraud_score += 0.2

    # New customers
    # if is_new_customer(state["customer_id"]):
    #     fraud_score += 0.15

    # Shipping != Billing address
    if state["shipping_address"] != state["billing_address"]:
        fraud_score += 0.1

    # Multiple items of same SKU
    for item in state["items"]:
        if item["quantity"] > 5:
            fraud_score += 0.15

    fraud_passed = fraud_score < 0.5

    return {
        "fraud_score": round(fraud_score, 2),
        "fraud_check_passed": fraud_passed,
        "status": "checking_fraud" if fraud_passed else "failed",
        "error_message": None if fraud_passed else "Order flagged for review",
        "events": [{
            "type": "fraud_check_completed",
            "score": fraud_score,
            "passed": fraud_passed,
            "at": datetime.now().isoformat()
        }]
    }

Payment Processing

import uuid

def process_payment(state: OrderState) -> dict:
    """Process payment through payment gateway"""

    # Check idempotency - prevent double charges
    # In production: check if this idempotency_key already has a successful payment

    try:
        # Mock Stripe-like payment processing
        payment_result = mock_stripe_charge(
            amount=int(state["total"] * 100),  # Cents
            currency="usd",
            payment_method=state["payment_method"],
            idempotency_key=state["idempotency_key"]
        )

        if payment_result["status"] == "succeeded":
            return {
                "payment_intent_id": payment_result["id"],
                "payment_status": "succeeded",
                "status": "processing_payment",
                "events": [{
                    "type": "payment_succeeded",
                    "payment_intent_id": payment_result["id"],
                    "amount": state["total"],
                    "at": datetime.now().isoformat()
                }]
            }
        else:
            # Payment failed - need to release inventory
            return {
                "payment_status": "failed",
                "status": "failed",
                "error_message": payment_result.get("error", "Payment failed"),
                "events": [{
                    "type": "payment_failed",
                    "error": payment_result.get("error"),
                    "at": datetime.now().isoformat()
                }]
            }

    except Exception as e:
        return {
            "payment_status": "failed",
            "status": "failed",
            "error_message": str(e),
            "events": [{
                "type": "payment_error",
                "error": str(e),
                "at": datetime.now().isoformat()
            }]
        }

def mock_stripe_charge(amount: int, currency: str, payment_method: dict, idempotency_key: str) -> dict:
    """Mock Stripe payment - replace with actual Stripe SDK"""
    # Simulate 95% success rate
    import random
    if random.random() > 0.05:
        return {
            "id": f"pi_{uuid.uuid4().hex[:24]}",
            "status": "succeeded",
            "amount": amount
        }
    else:
        return {
            "status": "failed",
            "error": "Card declined"
        }

Order Creation & Fulfillment

def create_order(state: OrderState) -> dict:
    """Create the order record in the database"""
    # In production: INSERT into orders table

    order_number = f"ORD-{datetime.now().strftime('%Y%m%d')}-{state['order_id'][-6:]}"

    return {
        "status": "creating_order",
        "events": [{
            "type": "order_created",
            "order_number": order_number,
            "at": datetime.now().isoformat()
        }]
    }

def trigger_fulfillment(state: OrderState) -> dict:
    """Send order to fulfillment system"""
    # In production: Call fulfillment API (ShipStation, ShipBob, etc.)

    fulfillment_request = {
        "order_id": state["order_id"],
        "items": state["items"],
        "shipping_address": state["shipping_address"],
        "shipping_method": "standard"  # or "express" based on customer choice
    }

    # mock_fulfillment_api(fulfillment_request)

    return {
        "status": "fulfilling",
        "events": [{
            "type": "fulfillment_triggered",
            "at": datetime.now().isoformat()
        }]
    }

def send_confirmation(state: OrderState) -> dict:
    """Send order confirmation email"""
    # In production: trigger email via SendGrid, SES, etc.

    # mock_send_email(
    #     to=state["customer_email"],
    #     template="order_confirmation",
    #     data={"order_id": state["order_id"], "total": state["total"]}
    # )

    return {
        "status": "completed",
        "events": [{
            "type": "confirmation_sent",
            "email": state["customer_email"],
            "at": datetime.now().isoformat()
        }]
    }

Compensation: Release Inventory

def release_inventory(state: OrderState) -> dict:
    """Release reserved inventory when order fails"""
    if not state.get("inventory_reserved"):
        return {"events": []}

    for item in state["items"]:
        if item.get("reserved"):
            # Release the reservation
            if item["sku"] in INVENTORY:
                INVENTORY[item["sku"]]["reserved"] -= item["quantity"]

    return {
        "inventory_reserved": False,
        "items": [{**item, "reserved": False} for item in state["items"]],
        "events": [{
            "type": "inventory_released",
            "reservation_id": state.get("reservation_id"),
            "at": datetime.now().isoformat()
        }]
    }

Step 3: Build the Graph

from langgraph.graph import StateGraph, START, END

def should_continue(state: OrderState) -> str:
    """Determine next step based on current status"""
    if state["status"] == "failed":
        if state.get("inventory_reserved"):
            return "release_inventory"
        return "end"
    return "continue"

def route_after_validation(state: OrderState) -> str:
    if state["status"] == "failed":
        return "end"
    return "reserve_inventory"

def route_after_inventory(state: OrderState) -> str:
    if state["status"] == "failed":
        return "end"
    return "check_fraud"

def route_after_fraud(state: OrderState) -> str:
    if not state["fraud_check_passed"]:
        return "release_inventory"  # Release and fail
    return "process_payment"

def route_after_payment(state: OrderState) -> str:
    if state["payment_status"] != "succeeded":
        return "release_inventory"  # Release and fail
    return "create_order"

# Build the graph
workflow = StateGraph(OrderState)

# Add all nodes
workflow.add_node("validate_cart", validate_cart)
workflow.add_node("reserve_inventory", reserve_inventory)
workflow.add_node("check_fraud", check_fraud)
workflow.add_node("process_payment", process_payment)
workflow.add_node("create_order", create_order)
workflow.add_node("trigger_fulfillment", trigger_fulfillment)
workflow.add_node("send_confirmation", send_confirmation)
workflow.add_node("release_inventory", release_inventory)

# Add edges
workflow.add_edge(START, "validate_cart")
workflow.add_conditional_edges("validate_cart", route_after_validation,
    {"end": END, "reserve_inventory": "reserve_inventory"})
workflow.add_conditional_edges("reserve_inventory", route_after_inventory,
    {"end": END, "check_fraud": "check_fraud"})
workflow.add_conditional_edges("check_fraud", route_after_fraud,
    {"release_inventory": "release_inventory", "process_payment": "process_payment"})
workflow.add_conditional_edges("process_payment", route_after_payment,
    {"release_inventory": "release_inventory", "create_order": "create_order"})

# Happy path continues linearly
workflow.add_edge("create_order", "trigger_fulfillment")
workflow.add_edge("trigger_fulfillment", "send_confirmation")
workflow.add_edge("send_confirmation", END)

# Compensation path
workflow.add_edge("release_inventory", END)

# Compile
order_processor = workflow.compile()

Step 4: Process an Order

import uuid

# Create order request
order_request = {
    "order_id": str(uuid.uuid4()),
    "idempotency_key": f"order_{uuid.uuid4()}",
    "customer_id": "CUST-12345",
    "customer_email": "customer@example.com",
    "shipping_address": {
        "line1": "123 Main St",
        "city": "San Francisco",
        "state": "CA",
        "zip": "94102"
    },
    "billing_address": {
        "line1": "123 Main St",
        "city": "San Francisco",
        "state": "CA",
        "zip": "94102"
    },
    "items": [
        {"sku": "SKU-001", "name": "Wireless Headphones", "quantity": 1, "unit_price": 149.99, "reserved": False},
        {"sku": "SKU-002", "name": "Phone Case", "quantity": 2, "unit_price": 24.99, "reserved": False}
    ],
    "subtotal": 0,
    "tax": 0,
    "shipping_cost": 0,
    "discount": 10.00,  # $10 coupon
    "total": 0,
    "payment_method": {"type": "card", "last4": "4242", "brand": "visa"},
    "payment_intent_id": None,
    "payment_status": "pending",
    "inventory_reserved": False,
    "reservation_id": None,
    "fraud_score": 0,
    "fraud_check_passed": False,
    "status": "cart",
    "error_message": None,
    "created_at": datetime.now().isoformat(),
    "updated_at": datetime.now().isoformat(),
    "events": []
}

# Process the order
result = order_processor.invoke(order_request)

print(f"Order Status: {result['status']}")
print(f"Total: ${result['total']:.2f}")
print(f"Payment: {result['payment_status']}")
print(f"\nEvent Timeline:")
for event in result['events']:
    print(f"  - {event['type']} at {event['at']}")

Order Processing Flow Visualization

┌─────────────────────────────────────────────────────────────┐
│                    ORDER PROCESSING FLOW                     │
├─────────────────────────────────────────────────────────────┤
│                                                              │
│   ┌──────────┐   ┌──────────┐   ┌──────────┐                │
│   │ Validate │──▶│ Reserve  │──▶│  Fraud   │                │
│   │   Cart   │   │Inventory │   │  Check   │                │
│   └────┬─────┘   └────┬─────┘   └────┬─────┘                │
│        │              │              │                       │
│        ▼              ▼              ▼                       │
│     [FAIL?]       [FAIL?]       [FAIL?]                     │
│        │              │              │                       │
│        └──────────────┴──────────────┘                      │
│                       │                                      │
│                       ▼                                      │
│              ┌───────────────┐                               │
│              │    Release    │                               │
│              │   Inventory   │──────▶ [END: FAILED]         │
│              └───────────────┘                               │
│                                                              │
│   [SUCCESS PATH]                                             │
│        │                                                     │
│        ▼                                                     │
│   ┌──────────┐   ┌──────────┐   ┌──────────┐   ┌────────┐  │
│   │ Process  │──▶│  Create  │──▶│ Trigger  │──▶│  Send  │  │
│   │ Payment  │   │  Order   │   │Fulfillment│  │ Email  │  │
│   └──────────┘   └──────────┘   └──────────┘   └───┬────┘  │
│                                                     │       │
│                                                     ▼       │
│                                            [END: COMPLETED] │
└─────────────────────────────────────────────────────────────┘

Key Takeaways

Validate early - Check cart and inventory before touching payment
Idempotency keys - Prevent duplicate charges on retries
Compensation pattern - Release inventory when payment fails
Event trail - Log every step for debugging and auditing

Next up: Product Recommendations - Building a recommendation agent with memory and personalization.

Multi-Step Order Processing

The Order Processing Challenge

Order Processing Pipeline

Step 1: Define Order State

Step 2: Implement Processing Nodes

Cart Validation

Inventory Reservation

Fraud Detection

Payment Processing

Order Creation & Fulfillment

Compensation: Release Inventory

Step 3: Build the Graph

Step 4: Process an Order

Order Processing Flow Visualization

Key Takeaways

🧠 Quick Quiz

In an order processing workflow, why should inventory be checked BEFORE payment processing?

What should happen if payment fails in the middle of an order workflow?

What is the purpose of idempotency in order processing?

Launch Offer

Multi-Step Order Processing

The Order Processing Challenge

Order Processing Pipeline

Step 1: Define Order State

Step 2: Implement Processing Nodes

Cart Validation

Inventory Reservation

Fraud Detection

Payment Processing

Order Creation & Fulfillment

Compensation: Release Inventory

Step 3: Build the Graph

Step 4: Process an Order

Order Processing Flow Visualization

Key Takeaways

🧠 Quick Quiz

In an order processing workflow, why should inventory be checked BEFORE payment processing?

What should happen if payment fails in the middle of an order workflow?

What is the purpose of idempotency in order processing?

Lesson Complete!

Get Your Free Resource

Start Your Free Trial

Your Interactive Session is Ready