Requirements
Functional: place orders (cart checkout → payment → fulfillment), track order status through lifecycle, handle partial fulfillment (items shipped separately), support cancellations and returns, manage inventory reservation during checkout, send notifications at each status change.
Non-functional: order placement is idempotent, inventory reservation is atomic, order history is immutable (append-only), consistent state even under payment system failures.
Core Entities
from enum import Enum
from dataclasses import dataclass, field
from typing import Optional, List
from datetime import datetime
class OrderStatus(Enum):
PENDING_PAYMENT = "PENDING_PAYMENT"
PAYMENT_FAILED = "PAYMENT_FAILED"
PAID = "PAID"
PROCESSING = "PROCESSING" # warehouse picking
PARTIALLY_SHIPPED = "PARTIALLY_SHIPPED"
SHIPPED = "SHIPPED"
DELIVERED = "DELIVERED"
CANCELLED = "CANCELLED"
RETURN_REQUESTED = "RETURN_REQUESTED"
RETURNED = "RETURNED"
@dataclass
class OrderItem:
item_id: str
product_id: str
variant_id: str
name: str
quantity: int
unit_price_cents: int
fulfillment_status: str # 'PENDING' | 'SHIPPED' | 'DELIVERED'
tracking_number: Optional[str] = None
shipped_at: Optional[datetime] = None
@dataclass
class Order:
order_id: str
user_id: str
items: List[OrderItem]
status: OrderStatus
subtotal_cents: int
shipping_cents: int
tax_cents: int
discount_cents: int
total_cents: int
shipping_address: dict
payment_intent_id: str
idempotency_key: str
placed_at: datetime
updated_at: datetime
notes: str = ''
@dataclass
class OrderEvent:
event_id: str
order_id: str
event_type: str # 'STATUS_CHANGED' | 'ITEM_SHIPPED' | 'PAYMENT_FAILED'
old_status: Optional[str]
new_status: Optional[str]
metadata: dict
occurred_at: datetime
actor: str # 'CUSTOMER' | 'SYSTEM' | 'WAREHOUSE'
Order State Machine
VALID_TRANSITIONS = {
OrderStatus.PENDING_PAYMENT: [OrderStatus.PAID, OrderStatus.PAYMENT_FAILED, OrderStatus.CANCELLED],
OrderStatus.PAYMENT_FAILED: [OrderStatus.PENDING_PAYMENT, OrderStatus.CANCELLED],
OrderStatus.PAID: [OrderStatus.PROCESSING, OrderStatus.CANCELLED],
OrderStatus.PROCESSING: [OrderStatus.PARTIALLY_SHIPPED, OrderStatus.SHIPPED, OrderStatus.CANCELLED],
OrderStatus.PARTIALLY_SHIPPED: [OrderStatus.SHIPPED, OrderStatus.CANCELLED],
OrderStatus.SHIPPED: [OrderStatus.DELIVERED],
OrderStatus.DELIVERED: [OrderStatus.RETURN_REQUESTED],
OrderStatus.RETURN_REQUESTED: [OrderStatus.RETURNED, OrderStatus.DELIVERED],
OrderStatus.CANCELLED: [],
OrderStatus.RETURNED: [],
}
def transition(order: Order, new_status: OrderStatus, actor: str, metadata: dict = None):
if new_status not in VALID_TRANSITIONS[order.status]:
raise ValueError(f"Invalid: {order.status} -> {new_status}")
old_status = order.status
order.status = new_status
order.updated_at = datetime.utcnow()
db.save(order)
# Append-only event log
db.save(OrderEvent(generate_id(), order.order_id, 'STATUS_CHANGED',
old_status.value, new_status.value, metadata or {}, datetime.utcnow(), actor))
emit_notification(order, new_status)
Order Placement Flow
class OrderService:
def place_order(self, user_id: str, cart_id: str, payment_method_id: str,
shipping_address: dict, idempotency_key: str) -> Order:
# Idempotency: return existing order if this checkout was already submitted
existing = db.get_order_by_idempotency_key(idempotency_key)
if existing: return existing
cart = cart_service.get_cart(cart_id)
if not cart.items: raise ValueError("Cart is empty")
# Atomic inventory reservation
reservation_ids = inventory_service.reserve_all(cart.items)
# If any item is out of stock, reserve_all raises and we never create the order
subtotal = sum(i.unit_price_cents * i.quantity for i in cart.items)
tax = tax_service.calculate(subtotal, shipping_address)
shipping = shipping_service.calculate(cart.items, shipping_address)
order = Order(
order_id=generate_id(), user_id=user_id,
items=[OrderItem(...) for i in cart.items],
status=OrderStatus.PENDING_PAYMENT,
subtotal_cents=subtotal, shipping_cents=shipping,
tax_cents=tax, discount_cents=0,
total_cents=subtotal + tax + shipping,
shipping_address=shipping_address,
payment_intent_id='', idempotency_key=idempotency_key,
placed_at=datetime.utcnow(), updated_at=datetime.utcnow(),
)
db.save(order)
cart_service.clear(cart_id)
# Charge payment (outside DB transaction — avoid holding lock during network call)
result = payment_service.charge(payment_method_id, order.total_cents,
idempotency_key=f"pay_{idempotency_key}")
if result.success:
order.payment_intent_id = result.intent_id
transition(order, OrderStatus.PAID, 'SYSTEM')
inventory_service.confirm_reservations(reservation_ids)
fulfillment_service.submit(order)
else:
transition(order, OrderStatus.PAYMENT_FAILED, 'SYSTEM',
{'reason': result.error_code})
inventory_service.release_reservations(reservation_ids)
return order
Partial Fulfillment
class FulfillmentService:
def ship_items(self, order_id: str, item_ids: List[str], tracking_number: str):
order = db.get_order(order_id)
shipped_count = 0
for item in order.items:
if item.item_id in item_ids:
item.fulfillment_status = 'SHIPPED'
item.tracking_number = tracking_number
item.shipped_at = datetime.utcnow()
shipped_count += 1
all_shipped = all(i.fulfillment_status == 'SHIPPED' for i in order.items)
new_status = OrderStatus.SHIPPED if all_shipped else OrderStatus.PARTIALLY_SHIPPED
transition(order, new_status, 'WAREHOUSE',
{'tracking_number': tracking_number, 'item_ids': item_ids})
db.save(order)
Return and Refund Flow
def request_return(order_id: str, user_id: str, item_ids: List[str], reason: str):
order = db.get_order(order_id)
if order.user_id != user_id: raise PermissionError
if order.status != OrderStatus.DELIVERED: raise ValueError("Can only return delivered orders")
days_since_delivery = (datetime.utcnow() - order.delivered_at).days
if days_since_delivery > 30: raise ValueError("Return window expired (30 days)")
transition(order, OrderStatus.RETURN_REQUESTED, 'CUSTOMER',
{'item_ids': item_ids, 'reason': reason})
generate_return_label(order, item_ids)
def process_return(order_id: str, item_ids: List[str]):
order = db.get_order(order_id)
return_amount = sum(i.unit_price_cents * i.quantity
for i in order.items if i.item_id in item_ids)
payment_service.refund(order.payment_intent_id, return_amount)
inventory_service.restock(item_ids)
transition(order, OrderStatus.RETURNED, 'WAREHOUSE',
{'refund_cents': return_amount, 'item_ids': item_ids})
Interview Questions
Q: How do you prevent inventory from going negative under concurrent orders?
Atomic reservation: UPDATE inventory SET reserved = reserved + qty, available = available – qty WHERE product_id = X AND available >= qty. If 0 rows updated, stock is insufficient. The WHERE clause makes this atomic at the database level — two concurrent requests can’t both read available=5 and both deduct 4. For higher throughput, use Redis Lua script: atomically check and decrement available in memory, persist asynchronously. Reservations have a TTL: if payment fails within 10 minutes, release them.
Q: How do you handle a payment timeout when you don’t know if the charge succeeded?
Store the order in PENDING_PAYMENT before charging. If the payment call times out: don’t immediately mark as failed. Query the payment gateway for the intent status using your idempotency key — the gateway can tell you if the charge went through. If confirmed: transition to PAID. If failed: release inventory, transition to PAYMENT_FAILED. A background reconciliation job checks orders stuck in PENDING_PAYMENT for more than 5 minutes and queries gateway status.
Asked at: Shopify Interview Guide
Asked at: Stripe Interview Guide
Asked at: DoorDash Interview Guide
Asked at: Airbnb Interview Guide