Core Entities
Product: product_id, sku, name, category_id, unit_cost, unit_weight_grams, dimensions (JSONB: length, width, height in mm), reorder_point, reorder_quantity, lead_time_days, is_active. Warehouse: warehouse_id, name, address, timezone, capacity_cubic_meters. Location: location_id, warehouse_id, aisle, bay, level, bin, type (BULK, PICK_FACE, OVERFLOW), max_weight_kg, max_volume_cubic_cm. Inventory: inventory_id, product_id, location_id, quantity_on_hand, quantity_reserved, quantity_available (computed: on_hand – reserved), last_counted_at, last_moved_at. InventoryTransaction: txn_id, product_id, location_id, txn_type (RECEIVE, SHIP, ADJUST, TRANSFER, RESERVE, RELEASE), quantity (positive=in, negative=out), reference_id (order_id or PO_id), reference_type, performed_by, created_at. PurchaseOrder: po_id, supplier_id, status (DRAFT, SENT, CONFIRMED, PARTIAL, RECEIVED, CANCELLED), line_items (JSONB), expected_date, created_at.
Stock Reservation and Atomic Allocation
class InventoryService:
def reserve_stock(self, product_id: int, quantity: int,
order_id: int) -> bool:
with db.transaction():
# Atomic read-modify-write with row-level lock
inv = db.query_one(
"SELECT inventory_id, quantity_on_hand, quantity_reserved "
"FROM inventory WHERE product_id = %s "
"FOR UPDATE", # row lock prevents concurrent oversell
product_id
)
if not inv:
raise ProductNotFound(product_id)
available = inv.quantity_on_hand - inv.quantity_reserved
if available < quantity:
raise InsufficientStock(
product_id, available, quantity
)
# Reserve stock
db.execute(
"UPDATE inventory SET quantity_reserved = quantity_reserved + %s "
"WHERE inventory_id = %s",
quantity, inv.inventory_id
)
# Audit trail
db.insert("inventory_transactions", {
"product_id": product_id,
"location_id": inv.location_id,
"txn_type": "RESERVE",
"quantity": -quantity, # negative = outbound reservation
"reference_id": order_id,
"reference_type": "ORDER"
})
return True
def fulfill_reservation(self, product_id: int, quantity: int,
order_id: int):
with db.transaction():
inv = db.query_one(
"SELECT * FROM inventory WHERE product_id = %s FOR UPDATE",
product_id
)
db.execute(
"UPDATE inventory "
"SET quantity_on_hand = quantity_on_hand - %s, "
" quantity_reserved = quantity_reserved - %s "
"WHERE inventory_id = %s",
quantity, quantity, inv.inventory_id
)
db.insert("inventory_transactions", {
"product_id": product_id,
"txn_type": "SHIP",
"quantity": -quantity,
"reference_id": order_id,
"reference_type": "ORDER"
})Receiving and Put-Away
def receive_purchase_order(self, po_id: int,
received_lines: list[dict]) -> Receipt:
po = db.get_po(po_id)
if po.status not in ("CONFIRMED", "PARTIAL"):
raise InvalidPOStatus()
for line in received_lines:
product_id = line["product_id"]
qty_received = line["quantity"]
location_id = line.get("location_id") or
self._suggest_putaway_location(product_id)
with db.transaction():
# Upsert inventory record for this product+location
existing = db.query_one(
"SELECT inventory_id, quantity_on_hand FROM inventory "
"WHERE product_id = %s AND location_id = %s FOR UPDATE",
product_id, location_id
)
if existing:
db.execute(
"UPDATE inventory SET quantity_on_hand = quantity_on_hand + %s "
"WHERE inventory_id = %s",
qty_received, existing.inventory_id
)
else:
db.insert("inventory", {
"product_id": product_id,
"location_id": location_id,
"quantity_on_hand": qty_received,
"quantity_reserved": 0
})
db.insert("inventory_transactions", {
"product_id": product_id,
"location_id": location_id,
"txn_type": "RECEIVE",
"quantity": qty_received,
"reference_id": po_id,
"reference_type": "PO"
})
self._update_po_status(po_id, received_lines)
self._check_replenishment_triggers(received_lines)Replenishment and Reorder Triggers
Trigger: after any stock-reducing transaction (SHIP, ADJUST), check if quantity_available falls below product.reorder_point. If yes: check if an open PO already exists for this product (avoid duplicate orders), then create a PurchaseOrder with quantity = reorder_quantity, expected_date = TODAY + lead_time_days. Auto-replenishment can be enabled per product. Replenishment background job (daily): scans all products WHERE quantity_available – reorder_point = 2 * average_daily_demand * lead_time_days (safety stock formula).
Shopify system design rounds cover inventory management. Review warehouse inventory LLD patterns for Shopify interview: inventory and warehouse management design.
DoorDash system design interviews cover inventory and supply chain systems. See design patterns for DoorDash interview: inventory and supply chain system design.
Stripe system design interviews cover transactional reservation systems. Review atomic inventory patterns for Stripe interview: transactional inventory and reservation systems.