# Low-Level Design: Inventory Management System (Stock Tracking, Reservations) Source: https://www.techinterview.org/post/3233464456/lld-inventory-management-system/ Updated: 2026-07-05 ยท techinterview.org ## Low-Level Design: Inventory Management System An Inventory Management System (IMS) tracks product stock levels, warehouse locations, reorder triggers, and stock movements. It is asked at Shopify, Amazon, and DoorDash in the context of e-commerce or supply chain design. ## Core Entities ``` from dataclasses import dataclass, field from enum import Enum from datetime import datetime from typing import Optional import uuid class MovementType(Enum): RECEIVE = "receive" # stock added (purchase order received) SELL = "sell" # stock consumed (order fulfillment) ADJUST = "adjust" # manual correction (damage, stocktake) TRANSFER = "transfer" # move between warehouses RESERVE = "reserve" # hold stock for pending order RELEASE = "release" # undo reservation class StockStatus(Enum): NORMAL = "normal" LOW_STOCK = "low_stock" # below reorder_point OUT_OF_STOCK = "out_of_stock" OVERSTOCKED = "overstocked" # above max_stock @dataclass class Product: product_id: str name: str sku: str unit_cost_cents: int reorder_point: int # trigger reorder when quantity falls below reorder_quantity: int # how much to order max_stock: int @dataclass class Warehouse: warehouse_id: str name: str location: str @dataclass class InventoryRecord: inventory_id: str product_id: str warehouse_id: str quantity_on_hand: int # physically present quantity_reserved: int # held for pending orders quantity_available: int # on_hand - reserved @property def status(self) -> StockStatus: if self.quantity_available <= 0: return StockStatus.OUT_OF_STOCK if self.quantity_on_hand < 10: # simplified; use reorder_point return StockStatus.LOW_STOCK return StockStatus.NORMAL @dataclass class StockMovement: movement_id: str product_id: str warehouse_id: str movement_type: MovementType quantity: int # positive for additions, negative for removals reference_id: str # order_id, po_id, etc. created_at: datetime = field(default_factory=datetime.utcnow) notes: str = "" ``` ## Inventory Service ``` import threading class InventoryService: def __init__(self): self._inventory: dict[tuple, InventoryRecord] = {} # (product_id, warehouse_id) self._movements: list[StockMovement] = [] self._products: dict[str, Product] = {} self._lock = threading.Lock() # per-record in production def _key(self, product_id: str, warehouse_id: str) -> tuple: return (product_id, warehouse_id) def _get_record(self, product_id: str, warehouse_id: str) -> InventoryRecord: key = self._key(product_id, warehouse_id) if key not in self._inventory: self._inventory[key] = InventoryRecord( inventory_id=str(uuid.uuid4()), product_id=product_id, warehouse_id=warehouse_id, quantity_on_hand=0, quantity_reserved=0, quantity_available=0, ) return self._inventory[key] def receive_stock(self, product_id: str, warehouse_id: str, quantity: int, po_id: str) -> InventoryRecord: if quantity <= 0: raise ValueError("Quantity must be positive") with self._lock: record = self._get_record(product_id, warehouse_id) record.quantity_on_hand += quantity record.quantity_available += quantity self._log_movement(product_id, warehouse_id, MovementType.RECEIVE, quantity, po_id) return record def reserve_stock(self, product_id: str, warehouse_id: str, quantity: int, order_id: str) -> bool: """Reserve available stock for an order; return False if insufficient.""" with self._lock: record = self._get_record(product_id, warehouse_id) if record.quantity_available < quantity: return False record.quantity_reserved += quantity record.quantity_available -= quantity self._log_movement(product_id, warehouse_id, MovementType.RESERVE, quantity, order_id) self._check_reorder(product_id, record) return True def commit_reservation(self, product_id: str, warehouse_id: str, quantity: int, order_id: str) -> None: """Commit reserved stock: deduct from on_hand after shipment.""" with self._lock: record = self._get_record(product_id, warehouse_id) if record.quantity_reserved < quantity: raise ValueError("Cannot commit more than is reserved") record.quantity_reserved -= quantity record.quantity_on_hand -= quantity self._log_movement(product_id, warehouse_id, MovementType.SHIP, -quantity, order_id) def release_reservation(self, product_id: str, warehouse_id: str, quantity: int, order_id: str) -> None: """Release reserved stock back to available (e.g., order cancelled).""" with self._lock: record = self._get_record(product_id, warehouse_id) if record.quantity_reserved < quantity: raise ValueError("Cannot release more than is reserved") record.quantity_reserved -= quantity record.quantity_available += quantity self._log_movement(product_id, warehouse_id, MovementType.RELEASE, quantity, order_id) def adjust_inventory(self, product_id: str, warehouse_id: str, new_on_hand: int, reason: str) -> InventoryRecord: """Stocktake correction: set on_hand to actual counted quantity.""" with self._lock: record = self._get_record(product_id, warehouse_id) delta = new_on_hand - record.quantity_on_hand record.quantity_on_hand = new_on_hand record.quantity_available = new_on_hand - record.quantity_reserved self._log_movement(product_id, warehouse_id, MovementType.ADJUST, delta, reason) return record def _log_movement(self, product_id, warehouse_id, movement_type, quantity, ref): self._movements.append(StockMovement( movement_id=str(uuid.uuid4()), product_id=product_id, warehouse_id=warehouse_id, movement_type=movement_type, quantity=quantity, reference_id=ref, )) def _check_reorder(self, product_id: str, record: InventoryRecord) -> None: product = self._products.get(product_id) if not product: return if record.quantity_available < product.reorder_point: # In production: emit a reorder event to purchasing system print(f"REORDER ALERT: {product_id} at {record.quantity_available} units " f"(reorder point: {product.reorder_point})") ``` ## Multi-Warehouse Allocation ``` def allocate_from_best_warehouse(self, product_id: str, quantity: int, order_id: str, ship_to_lat: float, ship_to_lng: float) -> Optional[str]: """Find the warehouse with sufficient stock closest to the ship-to location.""" candidates = [] for (pid, wid), record in self._inventory.items(): if pid != product_id: continue if record.quantity_available >= quantity: distance = self._distance(wid, ship_to_lat, ship_to_lng) candidates.append((distance, wid, record)) if not candidates: return None candidates.sort(key=lambda x: x[0]) _, best_wid, _ = candidates[0] self.reserve_stock(product_id, best_wid, quantity, order_id) return best_wid def _distance(self, warehouse_id: str, lat: float, lng: float) -> float: # In production: fetch warehouse coordinates and compute haversine return 0.0 # placeholder ``` ## Design Decisions | Decision | Choice | Rationale | | --- | --- | --- | | Stock tracking | on_hand + reserved + available | Prevents overselling during concurrent orders | | Movement log | Append-only StockMovement records | Audit trail, enables replay, supports stocktake reconciliation | | Concurrency | Lock per operation | For LLD; production uses row-level DB locks or optimistic locking | | Reorder trigger | On receive/reserve check | Lazy evaluation; background job for bulk checks | Asked at: [Shopify Interview Guide](https://techinterview.org/shopify-interview/) Asked at: [DoorDash Interview Guide](https://techinterview.org/doordash-interview/) Asked at: [Stripe Interview Guide](https://techinterview.org/stripe-interview/) Asked at: [Airbnb Interview Guide](https://techinterview.org/airbnb-interview/)