Backend & APIs Topic
E-commerce System Design
System Requirements & Scale
Functional Requirements
-
User Management: Registration, authentication, profiles
-
Product Catalog: Search, browse, categories, filters
-
Shopping Cart: Add/remove items, save for later
-
Order Management: Checkout, payment, order tracking
-
Inventory Management: Stock updates, reservations
-
Recommendation Engine: Personalized suggestions
-
Reviews & Ratings: Product reviews, seller ratings
-
Wishlist: Save favorite products
-
Notifications: Order updates, promotions
-
Analytics: Sales, user behavior, inventory
Non-Functional Requirements
-
Scalability: Handle 10M+ daily users, 1M+ concurrent
-
Availability: 99.99% uptime (max 52.5 mins downtime/year)
-
Performance: < 200ms page load, < 50ms API response
-
Consistency: Strong consistency for orders/inventory
-
Durability: Zero data loss for orders
-
Security: PCI DSS compliance for payments
-
Fault Tolerance: Graceful degradation, circuit breakers
High-Level Architecture
text
┌─────────────────────────────────────────────────────────────────────────┐
│ Client Layer │
├─────────────────────────────────────────────────────────────────────────┤
│ • Web (React/Next.js) • Mobile Apps (iOS/Android) • API Consumers │
│ • CDN (CloudFront/Akamai)• PWA • AMP Pages • Third-party │
└─────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────┐
│ API Gateway / Load Balancer │
│ • Request Routing • Rate Limiting • Authentication │
│ • API Versioning • Caching • Request/Response │
│ • SSL Termination • Logging • Transformations │
└─────────────────────────────────────────────────────────────────────────┘
│
┌───────────────────┼───────────────────┐
│ │ │
▼ ▼ ▼
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│ Microservices │ │ Microservices │ │ Microservices │
│ (Read-heavy) │ │ (Write-heavy) │ │ (Mixed) │
└─────────────────┘ └─────────────────┘ └─────────────────┘
│ │ │
└───────────────────┼───────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────┐
│ Data Layer │
├─────────────────────────────────────────────────────────────────────────┤
│ • SQL DB (Orders) • NoSQL (Catalog) • Cache (Redis) │
│ • Search (Elastic) • Queue (Kafka) • Object Store (S3) │
│ • Graph DB (Social) • Columnar (Analytics) • CDN (Media) │
└─────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────┐
│ External Services │
├─────────────────────────────────────────────────────────────────────────┤
│ • Payment Gateways • SMS/Email • Logistics │
│ • Fraud Detection • Analytics • Cloud Storage │
│ • ML Services • CDN Providers • Monitoring │
└─────────────────────────────────────────────────────────────────────────┘
Detailed Component Design
1. User Service
python
# app/services/user_service.py
class UserService:
"""Handles user registration, authentication, and profiles"""
def __init__(self):
self.user_db = PostgreSQL(shards=100) # Sharded by user_id
self.session_store = RedisCluster() # For session management
self.auth_cache = Redis() # JWT token cache
async def register_user(self, user_data: UserCreate) -> User:
"""Register new user with validation"""
# Check if user exists
existing = await self.user_db.execute("""
SELECT 1 FROM users
WHERE email = %s OR phone = %s
FOR UPDATE SKIP LOCKED
""", (user_data.email, user_data.phone))
if existing:
raise UserAlreadyExists()
# Hash password
hashed_password = bcrypt.hash(user_data.password)
# Generate user ID (snowflake ID for sharding)
user_id = self.generate_snowflake_id()
# Create user in database
async with self.user_db.transaction():
user = await self.user_db.execute("""
INSERT INTO users (id, email, phone, password_hash,
created_at, updated_at)
VALUES (%s, %s, %s, %s, NOW(), NOW())
RETURNING id, email, phone, created_at
""", (user_id, user_data.email, user_data.phone, hashed_password))
# Create user profile
await self.user_db.execute("""
INSERT INTO user_profiles
(user_id, first_name, last_name, date_of_birth)
VALUES (%s, %s, %s, %s)
""", (user_id, user_data.first_name, user_data.last_name,
user_data.date_of_birth))
# Send welcome email (async)
await self.queue.publish('email_queue', {
'type': 'welcome',
'user_id': user_id,
'email': user_data.email
})
return user
async def authenticate(self, identifier: str, password: str) -> AuthToken:
"""Authenticate user and generate JWT"""
# Rate limiting
attempts_key = f"auth_attempts:{identifier}"
attempts = await self.auth_cache.incr(attempts_key, expire=300)
if attempts > 5:
raise TooManyAttempts()
# Get user with password hash
user = await self.user_db.execute("""
SELECT id, email, password_hash, is_active,
mfa_enabled, role
FROM users
WHERE email = %s OR phone = %s
""", (identifier, identifier))
if not user or not bcrypt.verify(password, user.password_hash):
raise InvalidCredentials()
# Reset attempts on success
await self.auth_cache.delete(attempts_key)
# Generate JWT
token = jwt.encode({
'user_id': user.id,
'email': user.email,
'role': user.role,
'exp': datetime.utcnow() + timedelta(days=7)
}, SECRET_KEY)
# Store session
session_id = uuid.uuid4()
await self.session_store.setex(
f"session:{session_id}",
604800, # 7 days
json.dumps({
'user_id': user.id,
'token': token,
'last_active': datetime.utcnow().isoformat()
})
)
return AuthToken(
access_token=token,
refresh_token=self.generate_refresh_token(user.id),
session_id=session_id,
expires_in=604800
)
async def get_user_profile(self, user_id: str) -> UserProfile:
"""Get complete user profile with preferences"""
# Try cache first
cache_key = f"user_profile:{user_id}"
cached = await self.auth_cache.get(cache_key)
if cached:
return json.loads(cached)
# Query from database
profile = await self.user_db.execute("""
SELECT
u.id, u.email, u.phone, u.created_at,
up.first_name, up.last_name, up.date_of_birth,
up.profile_picture, up.bio,
a.address_line1, a.address_line2, a.city,
a.state, a.country, a.zip_code,
p.preferred_payment_method, p.notification_preferences,
p.language, p.currency
FROM users u
LEFT JOIN user_profiles up ON u.id = up.user_id
LEFT JOIN addresses a ON u.id = a.user_id AND a.is_default = true
LEFT JOIN user_preferences p ON u.id = p.user_id
WHERE u.id = %s
""", (user_id,))
if profile:
# Cache for 5 minutes
await self.auth_cache.setex(
cache_key, 300, json.dumps(profile)
)
return profile
def generate_snowflake_id(self) -> int:
"""Generate distributed unique ID"""
# Twitter Snowflake format: timestamp(41) + datacenter(5) + worker(5) + sequence(12)
timestamp = int(time.time() * 1000) - 1288834974657
worker_id = self.get_worker_id()
sequence = self.get_next_sequence()
return (timestamp << 22) | (worker_id << 17) | sequence
2. Product Catalog Service
python
# app/services/catalog_service.py
class CatalogService:
"""Handles product catalog, search, and inventory"""
def __init__(self):
self.mongo_db = MongoDB(sharded=True) # Product details
self.elastic = ElasticsearchCluster() # Search index
self.redis = RedisCluster() # Cache
self.inventory_db = Cassandra() # Real-time inventory
self.cdn = CloudFront() # Product images
async def search_products(self, query: str, filters: dict,
page: int = 1, size: int = 20) -> SearchResult:
"""Search products with filters and pagination"""
# Build Elasticsearch query
es_query = {
"query": {
"bool": {
"must": [
{
"multi_match": {
"query": query,
"fields": [
"title^3",
"description^2",
"brand^2",
"category^1.5"
],
"fuzziness": "AUTO"
}
}
],
"filter": self._build_filters(filters)
}
},
"sort": self._get_sort_criteria(filters),
"from": (page - 1) * size,
"size": size,
"aggs": {
"categories": {"terms": {"field": "category_id"}},
"brands": {"terms": {"field": "brand_id"}},
"price_ranges": {
"range": {
"field": "price",
"ranges": [
{"to": 500},
{"from": 500, "to": 1000},
{"from": 1000, "to": 5000},
{"from": 5000}
]
}
}
}
}
# Execute search
result = await self.elastic.search(
index="products_v1",
body=es_query
)
# Get product details from cache/DB
product_ids = [hit['_id'] for hit in result['hits']['hits']]
products = await self._get_products_by_ids(product_ids)
return SearchResult(
products=products,
total=result['hits']['total']['value'],
page=page,
size=size,
aggregations=result['aggregations']
)
async def _get_products_by_ids(self, product_ids: List[str]) -> List[Product]:
"""Get multiple products with caching"""
products = []
uncached_ids = []
# Try cache first
for pid in product_ids:
cache_key = f"product:{pid}"
cached = await self.redis.get(cache_key)
if cached:
products.append(json.loads(cached))
else:
uncached_ids.append(pid)
# Fetch uncached products from DB
if uncached_ids:
db_products = await self.mongo_db.products.find({
"_id": {"$in": uncached_ids},
"status": "ACTIVE"
}).to_list(length=None)
# Cache results
for product in db_products:
cache_key = f"product:{product['_id']}"
await self.redis.setex(
cache_key,
300, # 5 minutes TTL
json.dumps(product)
)
products.append(product)
# Enrich with inventory and pricing
enriched_products = await self._enrich_products(products)
return enriched_products
async def _enrich_products(self, products: List[dict]) -> List[dict]:
"""Enrich products with real-time data"""
# Batch get inventory
product_ids = [p['_id'] for p in products]
inventory_data = await self.inventory_db.execute_batch("""
SELECT product_id, warehouse_id, quantity,
reserved_quantity, available_quantity
FROM inventory
WHERE product_id IN ? AND available_quantity > 0
""", [product_ids])
# Batch get pricing
pricing_data = await self.redis.mget(
[f"price:{pid}" for pid in product_ids]
)
# Enrich each product
for product in products:
pid = product['_id']
# Add inventory info
product['inventory'] = {
'total_available': sum(
inv['available_quantity']
for inv in inventory_data.get(pid, [])
),
'warehouses': inventory_data.get(pid, [])
}
# Add pricing
product['current_price'] = pricing_data.get(pid) or product['base_price']
# Add image URLs with CDN
if 'images' in product:
product['images'] = [
self.cdn.generate_signed_url(img_url)
for img_url in product['images']
]
return products
async def update_inventory(self, product_id: str, quantity_change: int,
warehouse_id: str, reason: str):
"""Update inventory atomically"""
# Use Cassandra lightweight transactions for consistency
updated = await self.inventory_db.execute("""
UPDATE inventory
SET quantity = quantity + ?,
available_quantity = available_quantity + ?,
updated_at = toTimestamp(now())
WHERE product_id = ? AND warehouse_id = ?
IF quantity + ? >= 0
""", (quantity_change, quantity_change, product_id,
warehouse_id, quantity_change))
if not updated:
raise InsufficientStock()
# Publish inventory update event
await self.kafka.produce('inventory_updates', {
'event_id': str(uuid.uuid4()),
'event_type': 'INVENTORY_UPDATED',
'product_id': product_id,
'warehouse_id': warehouse_id,
'quantity_change': quantity_change,
'reason': reason,
'timestamp': datetime.utcnow().isoformat()
})
# Update cache
cache_key = f"inventory:{product_id}:{warehouse_id}"
await self.redis.delete(cache_key)
3. Shopping Cart Service
python
# app/services/cart_service.py
class CartService:
"""Distributed shopping cart management"""
def __init__(self):
self.redis = RedisCluster() # Primary cart storage
self.db = PostgreSQL() # Persistent backup
self.catalog_service = CatalogService() # Product info
self.inventory_service = InventoryService()
async def add_to_cart(self, user_id: str, item: CartItem) -> Cart:
"""Add item to cart with validation"""
# Validate product exists and is active
product = await self.catalog_service.get_product(item.product_id)
if not product or product['status'] != 'ACTIVE':
raise ProductNotFound()
# Check inventory
available = await self.inventory_service.check_availability(
item.product_id, item.quantity
)
if not available:
raise InsufficientStock()
# Get current cart
cart = await self.get_cart(user_id)
# Check if item already in cart
existing_item = next(
(i for i in cart.items if i.product_id == item.product_id),
None
)
if existing_item:
# Update quantity
new_quantity = existing_item.quantity + item.quantity
# Validate new quantity doesn't exceed max
if new_quantity > product['max_per_order']:
raise MaximumQuantityExceeded()
existing_item.quantity = new_quantity
existing_item.updated_at = datetime.utcnow()
else:
# Add new item
cart_item = CartItem(
product_id=item.product_id,
sku=product['sku'],
quantity=item.quantity,
unit_price=product['current_price'],
added_at=datetime.utcnow(),
updated_at=datetime.utcnow()
)
cart.items.append(cart_item)
# Update cart totals
await self._update_cart_totals(cart)
# Save to Redis with TTL (30 days)
cart_key = f"cart:{user_id}"
await self.redis.setex(
cart_key,
2592000, # 30 days
json.dumps(cart.dict())
)
# Backup to database async
await self._persist_cart_backup(user_id, cart)
return cart
async def get_cart(self, user_id: str) -> Cart:
"""Get user's shopping cart"""
cart_key = f"cart:{user_id}"
# Try Redis first
cached_cart = await self.redis.get(cart_key)
if cached_cart:
cart_data = json.loads(cached_cart)
# Validate cart items are still available
validated_cart = await self._validate_cart_items(cart_data)
return Cart(**validated_cart)
# Fallback to database
cart_data = await self.db.execute("""
SELECT cart_data, updated_at
FROM cart_backups
WHERE user_id = %s
ORDER BY updated_at DESC
LIMIT 1
""", (user_id,))
if cart_data:
# Restore to Redis
await self.redis.setex(
cart_key, 2592000, json.dumps(cart_data)
)
return Cart(**cart_data)
# Return empty cart
return Cart(user_id=user_id, items=[], total_items=0, total_price=0)
async def _validate_cart_items(self, cart_data: dict) -> dict:
"""Validate cart items (prices, availability, etc.)"""
items = cart_data.get('items', [])
validated_items = []
for item in items:
try:
# Check product still exists and is active
product = await self.catalog_service.get_product(
item['product_id']
)
if not product or product['status'] != 'ACTIVE':
continue # Remove invalid product
# Check inventory
available = await self.inventory_service.check_availability(
item['product_id'], item['quantity']
)
if not available:
item['out_of_stock'] = True
item['max_available'] = available
else:
item['out_of_stock'] = False
# Update price if changed
current_price = product['current_price']
if item['unit_price'] != current_price:
item['unit_price'] = current_price
item['price_changed'] = True
validated_items.append(item)
except Exception as e:
logger.error(f"Error validating cart item: {e}")
continue
cart_data['items'] = validated_items
return cart_data
async def checkout_cart(self, user_id: str, checkout_data: CheckoutData) -> Order:
"""Convert cart to order"""
# Get and lock cart
cart = await self.get_cart(user_id)
cart_lock = await self.redis.acquire_lock(f"cart_lock:{user_id}", ttl=30)
try:
# Validate all items
for item in cart.items:
# Check inventory with lock
inventory_lock = await self.inventory_service.reserve_inventory(
item.product_id, item.quantity
)
if not inventory_lock:
raise InsufficientStock(
f"Product {item.product_id} out of stock"
)
# Calculate totals with taxes
order_total = await self._calculate_order_total(cart, checkout_data)
# Create order
order_service = OrderService()
order = await order_service.create_order(
user_id=user_id,
cart=cart,
checkout_data=checkout_data,
total=order_total
)
# Clear cart
await self.redis.delete(f"cart:{user_id}")
# Release inventory locks (they'll be consumed by order fulfillment)
return order
finally:
await cart_lock.release()
async def _calculate_order_total(self, cart: Cart, checkout_data: CheckoutData) -> OrderTotal:
"""Calculate order total with taxes and discounts"""
subtotal = sum(item.unit_price * item.quantity for item in cart.items)
# Apply discounts
discounts = await self._apply_discounts(cart, checkout_data.coupons)
discount_amount = sum(d.amount for d in discounts)
# Calculate shipping
shipping = await self._calculate_shipping(
cart, checkout_data.shipping_address
)
# Calculate taxes
taxes = await self._calculate_taxes(
cart, checkout_data.shipping_address
)
total = subtotal - discount_amount + shipping + taxes
return OrderTotal(
subtotal=subtotal,
discount_amount=discount_amount,
shipping=shipping,
taxes=taxes,
total=total,
currency="INR"
)
4. Order Service
python
# app/services/order_service.py
class OrderService:
"""Order management and processing"""
def __init__(self):
self.order_db = PostgreSQL(sharded=True) # Order data
self.payments = PaymentService() # Payment processing
self.inventory = InventoryService() # Inventory management
self.kafka = KafkaProducer() # Event streaming
self.s3 = S3Bucket() # Order documents
async def create_order(self, user_id: str, cart: Cart,
checkout_data: CheckoutData, total: OrderTotal) -> Order:
"""Create a new order from cart"""
# Generate order ID
order_id = self.generate_order_id()
async with self.order_db.transaction():
# Create order record
order = await self.order_db.execute("""
INSERT INTO orders (
id, user_id, status, total_amount, currency,
shipping_address_id, billing_address_id,
payment_method, created_at, updated_at
) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, NOW(), NOW())
RETURNING *
""", (
order_id, user_id, 'PENDING_PAYMENT',
total.total, total.currency,
checkout_data.shipping_address_id,
checkout_data.billing_address_id,
checkout_data.payment_method
))
# Create order items
order_items = []
for cart_item in cart.items:
item = await self.order_db.execute("""
INSERT INTO order_items (
order_id, product_id, sku, quantity,
unit_price, total_price, created_at
) VALUES (%s, %s, %s, %s, %s, %s, NOW())
RETURNING *
""", (
order_id, cart_item.product_id, cart_item.sku,
cart_item.quantity, cart_item.unit_price,
cart_item.unit_price * cart_item.quantity
))
order_items.append(item)
# Create order summary
await self.order_db.execute("""
INSERT INTO order_summaries (
order_id, subtotal, discount_amount,
shipping_amount, tax_amount, total_amount,
created_at
) VALUES (%s, %s, %s, %s, %s, %s, NOW())
""", (
order_id, total.subtotal, total.discount_amount,
total.shipping, total.taxes, total.total
))
# Generate invoice
invoice_url = await self._generate_invoice(order, order_items, total)
# Update order with invoice
await self.order_db.execute("""
UPDATE orders
SET invoice_url = %s
WHERE id = %s
""", (invoice_url, order_id))
# Publish order created event
await self.kafka.produce('order_events', {
'event_id': str(uuid.uuid4()),
'event_type': 'ORDER_CREATED',
'order_id': order_id,
'user_id': user_id,
'amount': total.total,
'timestamp': datetime.utcnow().isoformat()
})
# Initiate payment
payment = await self.payments.initiate_payment(
order_id=order_id,
amount=total.total,
currency=total.currency,
method=checkout_data.payment_method,
user_id=user_id
)
return Order(
id=order_id,
user_id=user_id,
status='PENDING_PAYMENT',
items=order_items,
total=total,
invoice_url=invoice_url,
payment=payment
)
async def process_payment_webhook(self, payment_data: dict):
"""Handle payment gateway webhook"""
order_id = payment_data['order_id']
payment_status = payment_data['status']
# Get order with lock
order = await self.order_db.execute("""
SELECT * FROM orders
WHERE id = %s FOR UPDATE
""", (order_id,))
if not order:
raise OrderNotFound()
# Update order status based on payment
if payment_status == 'SUCCESS':
new_status = 'PAYMENT_RECEIVED'
# Reserve inventory permanently
await self._reserve_inventory_for_order(order_id)
# Trigger fulfillment
await self._trigger_fulfillment(order_id)
elif payment_status == 'FAILED':
new_status = 'PAYMENT_FAILED'
# Release any reserved inventory
await self._release_inventory_for_order(order_id)
elif payment_status == 'PENDING':
new_status = 'PENDING_PAYMENT'
# Update order
await self.order_db.execute("""
UPDATE orders
SET status = %s, payment_status = %s,
payment_id = %s, updated_at = NOW()
WHERE id = %s
""", (new_status, payment_status,
payment_data['payment_id'], order_id))
# Publish payment event
await self.kafka.produce('order_events', {
'event_id': str(uuid.uuid4()),
'event_type': 'PAYMENT_' + payment_status,
'order_id': order_id,
'timestamp': datetime.utcnow().isoformat()
})
# Send notification
await self._send_order_status_update(order['user_id'], order_id, new_status)
async def _trigger_fulfillment(self, order_id: str):
"""Trigger order fulfillment process"""
# Get order items
items = await self.order_db.execute("""
SELECT product_id, sku, quantity, warehouse_id
FROM order_items oi
JOIN inventory i ON oi.product_id = i.product_id
WHERE order_id = %s AND i.available_quantity >= oi.quantity
""", (order_id,))
# Group by warehouse for optimal picking
warehouse_groups = {}
for item in items:
warehouse_id = item['warehouse_id']
if warehouse_id not in warehouse_groups:
warehouse_groups[warehouse_id] = []
warehouse_groups[warehouse_id].append(item)
# Create fulfillment tasks for each warehouse
for warehouse_id, warehouse_items in warehouse_groups.items():
await self.kafka.produce('fulfillment_tasks', {
'task_id': str(uuid.uuid4()),
'type': 'PICK_AND_PACK',
'order_id': order_id,
'warehouse_id': warehouse_id,
'items': warehouse_items,
'priority': 'HIGH' if any(item['quantity'] > 10 for item in warehouse_items) else 'NORMAL'
})
async def update_order_status(self, order_id: str, status: str,
notes: str = None):
"""Update order status with state machine validation"""
valid_transitions = {
'PENDING_PAYMENT': ['PAYMENT_RECEIVED', 'CANCELLED'],
'PAYMENT_RECEIVED': ['PROCESSING', 'CANCELLED'],
'PROCESSING': ['SHIPPED', 'CANCELLED'],
'SHIPPED': ['OUT_FOR_DELIVERY', 'DELIVERED', 'RETURN_REQUESTED'],
'OUT_FOR_DELIVERY': ['DELIVERED', 'DELIVERY_FAILED'],
'DELIVERED': ['RETURN_REQUESTED', 'COMPLETED'],
'RETURN_REQUESTED': ['RETURN_APPROVED', 'RETURN_REJECTED'],
'RETURN_APPROVED': ['RETURN_RECEIVED'],
'RETURN_RECEIVED': ['REFUND_INITIATED'],
'REFUND_INITIATED': ['REFUNDED'],
'CANCELLED': ['REFUND_INITIATED'],
}
# Get current status
current = await self.order_db.execute("""
SELECT status FROM orders WHERE id = %s
""", (order_id,))
if not current:
raise OrderNotFound()
# Validate transition
if status not in valid_transitions.get(current['status'], []):
raise InvalidStatusTransition()
# Update status
await self.order_db.execute("""
UPDATE orders
SET status = %s, updated_at = NOW()
WHERE id = %s
""", (status, order_id))
# Log status change
await self.order_db.execute("""
INSERT INTO order_status_history (
order_id, old_status, new_status, notes, changed_at
) VALUES (%s, %s, %s, %s, NOW())
""", (order_id, current['status'], status, notes))
# Publish event
await self.kafka.produce('order_events', {
'event_id': str(uuid.uuid4()),
'event_type': f'ORDER_{status}',
'order_id': order_id,
'timestamp': datetime.utcnow().isoformat()
})
# Trigger side effects
await self._handle_status_side_effects(order_id, status)
def generate_order_id(self) -> str:
"""Generate unique order ID"""
# Format: ORD{timestamp}{random}{shard}
timestamp = int(time.time() * 1000)
random_part = random.randint(1000, 9999)
shard_id = hash(self.worker_id) % 100
return f"ORD{timestamp}{random_part}{shard_id:02d}"
5. Recommendation Service
python
# app/services/recommendation_service.py
class RecommendationService:
"""Personalized product recommendations"""
def __init__(self):
self.redis = RedisCluster() # Feature store & cache
self.kafka = KafkaConsumer() # User behavior streaming
self.ml_serving = TensorFlowServing() # Model serving
self.cassandra = Cassandra() # User vectors
self.ann = FAISSIndex() # Approximate nearest neighbors
async def get_recommendations(self, user_id: str, context: dict = None) -> List[Recommendation]:
"""Get personalized recommendations"""
# Try cache first
cache_key = f"recs:{user_id}:{self._get_context_key(context)}"
cached = await self.redis.get(cache_key)
if cached:
return json.loads(cached)
# Get multiple recommendation types
recommendations = await asyncio.gather(
self._get_collaborative_filtering_recs(user_id),
self._get_content_based_recs(user_id),
self._get_trending_recs(context),
self._get_frequently_bought_together(user_id)
)
# Merge and rank recommendations
merged = self._merge_recommendations(
recommendations[0], recommendations[1],
recommendations[2], recommendations[3]
)
# Apply business rules
filtered = await self._apply_business_rules(merged, user_id)
# Cache results
await self.redis.setex(cache_key, 300, json.dumps(filtered[:20]))
return filtered[:20] # Return top 20
async def _get_collaborative_filtering_recs(self, user_id: str) -> List[Recommendation]:
"""Collaborative filtering using matrix factorization"""
# Get user embedding from Cassandra
user_vector = await self.cassandra.execute("""
SELECT embedding FROM user_embeddings
WHERE user_id = %s
""", (user_id,))
if not user_vector:
# Cold start: return popular items
return await self._get_popular_items()
# Find similar users using ANN
similar_users = await self.ann.search(
index='user_embeddings',
query_vector=user_vector['embedding'],
k=50
)
# Get items liked by similar users
similar_user_ids = [su['user_id'] for su in similar_users]
liked_items = await self.cassandra.execute("""
SELECT product_id, COUNT(*) as score
FROM user_interactions
WHERE user_id IN ? AND interaction_type = 'PURCHASE'
GROUP BY product_id
ORDER BY score DESC
LIMIT 100
""", [similar_user_ids])
return [
Recommendation(
product_id=item['product_id'],
score=item['score'] / len(similar_users),
type='collaborative',
reason='Users similar to you bought this'
)
for item in liked_items
]
async def _get_content_based_recs(self, user_id: str) -> List[Recommendation]:
"""Content-based recommendations using product embeddings"""
# Get user's recent interactions
recent_items = await self.cassandra.execute("""
SELECT product_id, interaction_type, timestamp
FROM user_interactions
WHERE user_id = %s
ORDER BY timestamp DESC
LIMIT 20
""", (user_id,))
if not recent_items:
return []
# Get embeddings for recent items
item_ids = [item['product_id'] for item in recent_items]
item_embeddings = await self.cassandra.execute("""
SELECT product_id, embedding
FROM product_embeddings
WHERE product_id IN ?
""", [item_ids])
# Calculate weighted average of item embeddings
user_embedding = self._calculate_user_embedding(
recent_items, item_embeddings
)
# Find similar products using ANN
similar_products = await self.ann.search(
index='product_embeddings',
query_vector=user_embedding,
k=100,
filters={'status': 'ACTIVE', 'in_stock': True}
)
return [
Recommendation(
product_id=product['product_id'],
score=product['similarity'],
type='content',
reason='Similar to items you viewed'
)
for product in similar_products
]
async def _get_frequently_bought_together(self, user_id: str) -> List[Recommendation]:
"""Get items frequently bought together with cart items"""
# Get current cart items
cart_service = CartService()
cart = await cart_service.get_cart(user_id)
if not cart.items:
return []
# Get associations from Redis
recommendations = []
for item in cart.items:
associations = await self.redis.get(
f"fbt:{item.product_id}"
)
if associations:
fbt_items = json.loads(associations)
recommendations.extend([
Recommendation(
product_id=fbt_item['product_id'],
score=fbt_item['confidence'],
type='frequently_bought_together',
reason=f'Frequently bought with {item.product_id}'
)
for fbt_item in fbt_items[:5]
])
return recommendations
async def process_user_interaction(self, event: UserInteractionEvent):
"""Process user interaction for real-time recommendations"""
# Store interaction in Cassandra
await self.cassandra.execute("""
INSERT INTO user_interactions (
user_id, product_id, interaction_type,
timestamp, context
) VALUES (?, ?, ?, ?, ?)
""", (
event.user_id, event.product_id,
event.interaction_type, event.timestamp,
json.dumps(event.context)
))
# Update user embedding asynchronously
await self.kafka.produce('embedding_updates', {
'user_id': event.user_id,
'interaction': event.dict(),
'timestamp': datetime.utcnow().isoformat()
})
# Update association rules
if event.interaction_type == 'PURCHASE':
await self._update_association_rules(
event.user_id, event.product_id
)
async def _update_association_rules(self, user_id: str, product_id: str):
"""Update frequently bought together rules"""
# Get user's recent purchases
recent_purchases = await self.cassandra.execute("""
SELECT product_id
FROM user_interactions
WHERE user_id = %s AND interaction_type = 'PURCHASE'
AND timestamp > %s
ORDER BY timestamp DESC
LIMIT 10
""", (user_id, datetime.utcnow() - timedelta(days=30)))
if len(recent_purchases) < 2:
return
# Update association counts in Redis
for purchase in recent_purchases:
if purchase['product_id'] != product_id:
# Increment association count
association_key = f"assoc:{product_id}:{purchase['product_id']}"
await self.redis.incr(association_key)
# Update TTL to 90 days
await self.redis.expire(association_key, 7776000)
# Periodically rebuild FBT cache
if random.random() < 0.01: # 1% chance
await self._rebuild_fbt_cache()
def _merge_recommendations(self, *recommendation_lists) -> List[Recommendation]:
"""Merge and re-rank multiple recommendation lists"""
# Create combined dictionary
combined = {}
for rec_list in recommendation_lists:
for rec in rec_list:
if rec.product_id not in combined:
combined[rec.product_id] = rec
else:
# Weighted combination
existing = combined[rec.product_id]
weights = {
'collaborative': 0.4,
'content': 0.3,
'trending': 0.2,
'frequently_bought_together': 0.1
}
new_score = existing.score + (rec.score * weights.get(rec.type, 0.1))
existing.score = new_score
existing.reasons.append(rec.reason)
# Sort by score
sorted_recs = sorted(
combined.values(),
key=lambda x: x.score,
reverse=True
)
# Apply diversity (don't show too many similar items)
diversified = self._apply_diversity(sorted_recs, max_similar=3)
return diversified
6. Inventory Service
python
# app/services/inventory_service.py
class InventoryService:
"""Real-time inventory management"""
def __init__(self):
self.cassandra = Cassandra() # Inventory state
self.redis = RedisCluster() # Inventory cache & locks
self.kafka = KafkaProducer() # Inventory events
self.zk = ZooKeeper() # Coordination
async def check_availability(self, product_id: str,
quantity: int) -> bool:
"""Check if product is available in any warehouse"""
# Try cache first
cache_key = f"inventory:{product_id}"
cached = await self.redis.get(cache_key)
if cached:
inventory_data = json.loads(cached)
total_available = sum(
inv['available_quantity']
for inv in inventory_data.values()
)
return total_available >= quantity
# Query Cassandra
inventory = await self.cassandra.execute("""
SELECT warehouse_id, available_quantity
FROM inventory
WHERE product_id = %s AND available_quantity > 0
ALLOW FILTERING
""", (product_id,))
if not inventory:
return False
# Cache result
inventory_dict = {
item['warehouse_id']: item['available_quantity']
for item in inventory
}
await self.redis.setex(cache_key, 30, json.dumps(inventory_dict))
total_available = sum(inventory_dict.values())
return total_available >= quantity
async def reserve_inventory(self, product_id: str,
quantity: int, ttl: int = 300) -> Optional[str]:
"""Reserve inventory for checkout (returns reservation ID)"""
reservation_id = str(uuid.uuid4())
# Try to reserve from warehouses with sufficient stock
warehouses = await self.cassandra.execute("""
SELECT warehouse_id, available_quantity
FROM inventory
WHERE product_id = %s AND available_quantity > 0
ORDER BY available_quantity DESC
""", (product_id,))
remaining = quantity
reservations = []
for warehouse in warehouses:
if remaining <= 0:
break
available = warehouse['available_quantity']
reserve_amount = min(available, remaining)
# Try to reserve using lightweight transaction
reserved = await self.cassandra.execute("""
UPDATE inventory
SET available_quantity = available_quantity - %s,
reserved_quantity = reserved_quantity + %s,
updated_at = toTimestamp(now())
WHERE product_id = %s AND warehouse_id = %s
IF available_quantity >= %s
""", (reserve_amount, reserve_amount,
product_id, warehouse['warehouse_id'], reserve_amount))
if reserved:
reservations.append({
'warehouse_id': warehouse['warehouse_id'],
'quantity': reserve_amount
})
remaining -= reserve_amount
if remaining > 0:
# Failed to reserve enough, rollback
await self._rollback_reservations(product_id, reservations)
return None
# Store reservation
await self.redis.setex(
f"reservation:{reservation_id}",
ttl,
json.dumps({
'product_id': product_id,
'quantity': quantity,
'reservations': reservations,
'created_at': datetime.utcnow().isoformat()
})
)
# Invalidate cache
await self.redis.delete(f"inventory:{product_id}")
return reservation_id
async def consume_reservation(self, reservation_id: str):
"""Convert reservation to permanent allocation"""
# Get reservation
reservation_data = await self.redis.get(f"reservation:{reservation_id}")
if not reservation_data:
raise ReservationNotFound()
reservation = json.loads(reservation_data)
# Mark as consumed in Cassandra
for res in reservation['reservations']:
await self.cassandra.execute("""
UPDATE inventory
SET reserved_quantity = reserved_quantity - %s,
quantity = quantity - %s,
updated_at = toTimestamp(now())
WHERE product_id = %s AND warehouse_id = %s
""", (res['quantity'], res['quantity'],
reservation['product_id'], res['warehouse_id']))
# Delete reservation
await self.redis.delete(f"reservation:{reservation_id}")
# Publish inventory consumed event
await self.kafka.produce('inventory_events', {
'event_id': str(uuid.uuid4()),
'event_type': 'INVENTORY_CONSUMED',
'product_id': reservation['product_id'],
'reservation_id': reservation_id,
'quantity': reservation['quantity'],
'timestamp': datetime.utcnow().isoformat()
})
async def update_inventory_batch(self, updates: List[InventoryUpdate]):
"""Batch update inventory (for stock replenishment)"""
# Group by warehouse for efficient updates
warehouse_updates = {}
for update in updates:
key = (update.product_id, update.warehouse_id)
if key not in warehouse_updates:
warehouse_updates[key] = 0
warehouse_updates[key] += update.quantity_change
# Execute batch updates
batch = self.cassandra.batch()
for (product_id, warehouse_id), quantity_change in warehouse_updates.items():
batch.add("""
UPDATE inventory
SET quantity = quantity + %s,
available_quantity = available_quantity + %s,
updated_at = toTimestamp(now())
WHERE product_id = %s AND warehouse_id = %s
""", (quantity_change, quantity_change, product_id, warehouse_id))
await batch.execute()
# Invalidate caches
for product_id in set(update.product_id for update in updates):
await self.redis.delete(f"inventory:{product_id}")
# Publish events
for update in updates:
await self.kafka.produce('inventory_events', {
'event_id': str(uuid.uuid4()),
'event_type': 'INVENTORY_UPDATED',
'product_id': update.product_id,
'warehouse_id': update.warehouse_id,
'quantity_change': update.quantity_change,
'reason': update.reason,
'timestamp': datetime.utcnow().isoformat()
})
async def get_inventory_snapshot(self, product_id: str = None,
warehouse_id: str = None) -> List[dict]:
"""Get inventory snapshot for reporting"""
if product_id and warehouse_id:
# Specific product in specific warehouse
inventory = await self.cassandra.execute("""
SELECT * FROM inventory
WHERE product_id = %s AND warehouse_id = %s
""", (product_id, warehouse_id))
elif product_id:
# All warehouses for product
inventory = await self.cassandra.execute("""
SELECT * FROM inventory
WHERE product_id = %s
""", (product_id,))
else:
# All inventory (paginated)
inventory = await self.cassandra.execute("""
SELECT * FROM inventory
LIMIT 1000
""")
return inventory
async def monitor_low_stock(self, threshold: int = 10):
"""Monitor and alert for low stock items"""
# Get items below threshold
low_stock = await self.cassandra.execute("""
SELECT product_id, warehouse_id, available_quantity
FROM inventory
WHERE available_quantity < %s
ALLOW FILTERING
""", (threshold,))
# Group by product
product_stock = {}
for item in low_stock:
if item['product_id'] not in product_stock:
product_stock[item['product_id']] = {
'total_available': 0,
'warehouses': []
}
product_stock[item['product_id']]['total_available'] += item['available_quantity']
product_stock[item['product_id']]['warehouses'].append(item)
# Alert for products with total below threshold
alerts = []
for product_id, stock_info in product_stock.items():
if stock_info['total_available'] < threshold:
alerts.append({
'product_id': product_id,
'total_available': stock_info['total_available'],
'warehouses': stock_info['warehouses'],
'alert_level': 'CRITICAL' if stock_info['total_available'] == 0 else 'WARNING'
})
# Send alerts
for alert in alerts:
await self.kafka.produce('inventory_alerts', alert)
# Also send to notification service
await self.kafka.produce('notifications', {
'type': 'LOW_STOCK_ALERT',
'product_id': alert['product_id'],
'available_quantity': alert['total_available'],
'timestamp': datetime.utcnow().isoformat()
})
Database Schema Design
1. PostgreSQL Schema (Orders, Users, Payments)
sql
-- Users sharded by user_id
CREATE TABLE users (
id BIGINT PRIMARY KEY,
email VARCHAR(255) UNIQUE NOT NULL,
phone VARCHAR(20) UNIQUE,
password_hash VARCHAR(255) NOT NULL,
is_active BOOLEAN DEFAULT true,
is_verified BOOLEAN DEFAULT false,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
last_login_at TIMESTAMP,
INDEX idx_email (email),
INDEX idx_phone (phone),
INDEX idx_created_at (created_at)
) PARTITION BY HASH(id);
CREATE TABLE user_profiles (
user_id BIGINT PRIMARY KEY REFERENCES users(id),
first_name VARCHAR(100),
last_name VARCHAR(100),
date_of_birth DATE,
profile_picture_url TEXT,
bio TEXT,
gender VARCHAR(20),
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
-- Orders sharded by order_id
CREATE TABLE orders (
id VARCHAR(50) PRIMARY KEY,
user_id BIGINT NOT NULL,
status VARCHAR(50) NOT NULL,
total_amount DECIMAL(10, 2) NOT NULL,
currency VARCHAR(3) DEFAULT 'INR',
shipping_address_id BIGINT,
billing_address_id BIGINT,
payment_method VARCHAR(50),
payment_status VARCHAR(50),
payment_id VARCHAR(100),
invoice_url TEXT,
notes TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
INDEX idx_user_id (user_id),
INDEX idx_status (status),
INDEX idx_created_at (created_at),
INDEX idx_payment_status (payment_status)
) PARTITION BY HASH(id);
CREATE TABLE order_items (
id BIGSERIAL PRIMARY KEY,
order_id VARCHAR(50) NOT NULL REFERENCES orders(id),
product_id VARCHAR(50) NOT NULL,
sku VARCHAR(100) NOT NULL,
quantity INTEGER NOT NULL CHECK (quantity > 0),
unit_price DECIMAL(10, 2) NOT NULL,
total_price DECIMAL(10, 2) NOT NULL,
tax_amount DECIMAL(10, 2) DEFAULT 0,
discount_amount DECIMAL(10, 2) DEFAULT 0,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
INDEX idx_order_id (order_id),
INDEX idx_product_id (product_id),
UNIQUE (order_id, product_id, sku)
);
-- Payments
CREATE TABLE payments (
id VARCHAR(100) PRIMARY KEY,
order_id VARCHAR(50) NOT NULL REFERENCES orders(id),
user_id BIGINT NOT NULL,
amount DECIMAL(10, 2) NOT NULL,
currency VARCHAR(3) DEFAULT 'INR',
method VARCHAR(50) NOT NULL,
status VARCHAR(50) NOT NULL,
gateway_transaction_id VARCHAR(200),
gateway_response JSONB,
refund_amount DECIMAL(10, 2) DEFAULT 0,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
INDEX idx_order_id (order_id),
INDEX idx_user_id (user_id),
INDEX idx_status (status),
INDEX idx_created_at (created_at)
);
-- Addresses
CREATE TABLE addresses (
id BIGSERIAL PRIMARY KEY,
user_id BIGINT NOT NULL REFERENCES users(id),
address_type VARCHAR(20) DEFAULT 'SHIPPING', -- SHIPPING/BILLING
is_default BOOLEAN DEFAULT false,
recipient_name VARCHAR(200) NOT NULL,
phone VARCHAR(20) NOT NULL,
address_line1 VARCHAR(500) NOT NULL,
address_line2 VARCHAR(500),
city VARCHAR(100) NOT NULL,
state VARCHAR(100) NOT NULL,
country VARCHAR(100) NOT NULL,
zip_code VARCHAR(20) NOT NULL,
landmark VARCHAR(200),
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
INDEX idx_user_id (user_id),
INDEX idx_city (city),
INDEX idx_state (state),
UNIQUE (user_id, address_line1, city, zip_code)
);
2. MongoDB Schema (Product Catalog)
javascript
// Products collection - sharded by category_id
{
"_id": "PROD_123456789", // Snowflake ID
"sku": "FLIPKART-APPLE-IPHONE15-128GB-BLACK",
"title": "Apple iPhone 15 (128 GB) - Black",
"slug": "apple-iphone-15-128-gb-black",
"description": "Latest iPhone with A16 Bionic chip...",
"brand_id": "BRAND_APPLE",
"brand_name": "Apple",
"category_id": "CAT_MOBILES",
"category_path": ["Electronics", "Mobiles", "Smartphones"],
"attributes": {
"color": "Black",
"storage": "128 GB",
"ram": "6 GB",
"display": "6.1 inches",
"camera": "48 MP + 12 MP",
"battery": "3349 mAh",
"processor": "A16 Bionic"
},
"specifications": [
{"key": "Model", "value": "iPhone 15"},
{"key": "Network", "value": "5G"}
],
"images": [
{
"url": "https://cdn.flipkart.com/images/product/iphone15_1.jpg",
"type": "MAIN",
"order": 1
}
],
"videos": [
{
"url": "https://youtube.com/embed/iphone15",
"thumbnail": "https://img.youtube.com/iphone15.jpg"
}
],
"pricing": {
"base_price": 79999.00,
"selling_price": 74999.00,
"currency": "INR",
"discount_percent": 6,
"max_per_order": 2
},
"inventory": {
"total_quantity": 1500,
"available_quantity": 850,
"low_stock_threshold": 50
},
"seller_id": "SELLER_FLIPKART",
"seller_info": {
"name": "Flipkart SmartBuy",
"rating": 4.5,
"total_ratings": 125000
},
"ratings": {
"average": 4.6,
"total": 12500,
"distribution": {
"5": 8500,
"4": 2500,
"3": 1000,
"2": 300,
"1": 200
}
},
"reviews_summary": {
"total": 12500,
"with_media": 4500,
"with_text": 9800
},
"tags": ["smartphone", "apple", "iphone", "5g", "premium"],
"search_keywords": ["iphone 15", "apple phone", "smartphone"],
"status": "ACTIVE", // ACTIVE, INACTIVE, OUT_OF_STOCK, COMING_SOON
"created_at": ISODate("2024-01-15T10:00:00Z"),
"updated_at": ISODate("2024-01-20T15:30:00Z"),
"published_at": ISODate("2024-01-16T09:00:00Z"),
"seo": {
"meta_title": "Buy Apple iPhone 15 (128GB) - Black | Flipkart.com",
"meta_description": "Buy Apple iPhone 15 online at best prices...",
"meta_keywords": "iPhone 15, Apple iPhone, 5G phone"
}
}
// Categories collection
{
"_id": "CAT_MOBILES",
"name": "Mobiles",
"slug": "mobiles",
"parent_id": "CAT_ELECTRONICS",
"path": ["Electronics", "Mobiles"],
"level": 2,
"attributes": [
{
"name": "Brand",
"type": "MULTI_SELECT",
"values": ["Apple", "Samsung", "OnePlus", "Xiaomi"]
},
{
"name": "RAM",
"type": "SINGLE_SELECT",
"values": ["4 GB", "6 GB", "8 GB", "12 GB"]
}
],
"image_url": "https://cdn.flipkart.com/category/mobiles.jpg",
"is_active": true,
"product_count": 12500,
"created_at": ISODate("2024-01-01T00:00:00Z")
}
// Brands collection
{
"_id": "BRAND_APPLE",
"name": "Apple",
"slug": "apple",
"logo_url": "https://cdn.flipkart.com/brands/apple.jpg",
"description": "American multinational technology company...",
"website": "https://www.apple.com",
"product_count": 150,
"is_featured": true,
"is_active": true,
"created_at": ISODate("2024-01-01T00:00:00Z")
}
3. Cassandra Schema (Inventory, User Interactions)
cql
-- Inventory table - partitioned by product_id, clustered by warehouse_id
CREATE TABLE inventory (
product_id TEXT,
warehouse_id TEXT,
quantity INT,
available_quantity INT,
reserved_quantity INT,
minimum_stock_level INT,
maximum_stock_level INT,
reorder_point INT,
last_restocked_at TIMESTAMP,
updated_at TIMESTAMP,
PRIMARY KEY (product_id, warehouse_id)
) WITH CLUSTERING ORDER BY (warehouse_id ASC);
-- Create materialized view for warehouse-centric queries
CREATE MATERIALIZED VIEW inventory_by_warehouse AS
SELECT * FROM inventory
WHERE warehouse_id IS NOT NULL AND product_id IS NOT NULL
PRIMARY KEY (warehouse_id, product_id);
-- User interactions for recommendations
CREATE TABLE user_interactions (
user_id TEXT,
interaction_id TIMEUUID,
product_id TEXT,
interaction_type TEXT, -- VIEW, ADD_TO_CART, PURCHASE, RATE, REVIEW
value DECIMAL, -- For ratings
duration_seconds INT, -- For view duration
context MAP<TEXT, TEXT>, -- Page URL, device, location, etc.
timestamp TIMESTAMP,
PRIMARY KEY (user_id, interaction_id)
) WITH CLUSTERING ORDER BY (interaction_id DESC);
-- User embeddings for collaborative filtering
CREATE TABLE user_embeddings (
user_id TEXT PRIMARY KEY,
embedding LIST<FLOAT>, -- 100-dimensional embedding
updated_at TIMESTAMP,
version INT
);
-- Product embeddings for content-based filtering
CREATE TABLE product_embeddings (
product_id TEXT PRIMARY KEY,
embedding LIST<FLOAT>,
updated_at TIMESTAMP,
version INT
);
-- Session data
CREATE TABLE user_sessions (
session_id TEXT PRIMARY KEY,
user_id TEXT,
device_id TEXT,
device_type TEXT, -- MOBILE, DESKTOP, TABLET
browser TEXT,
os TEXT,
ip_address TEXT,
location TEXT,
started_at TIMESTAMP,
last_active_at TIMESTAMP,
expires_at TIMESTAMP,
session_data MAP<TEXT, TEXT>
);
-- Shopping cart (persistent backup)
CREATE TABLE cart_backups (
user_id TEXT,
cart_id TIMEUUID,
cart_data TEXT, -- JSON serialized cart
created_at TIMESTAMP,
updated_at TIMESTAMP,
PRIMARY KEY (user_id, cart_id)
) WITH CLUSTERING ORDER BY (cart_id DESC);
4. Elasticsearch Index (Search)
json
{
"settings": {
"number_of_shards": 10,
"number_of_replicas": 2,
"analysis": {
"analyzer": {
"hindi_analyzer": {
"type": "custom",
"tokenizer": "standard",
"filter": ["lowercase", "hindi_normalization"]
},
"english_analyzer": {
"type": "custom",
"tokenizer": "standard",
"filter": ["lowercase", "english_stop", "english_stemmer"]
}
}
}
},
"mappings": {
"properties": {
"product_id": {"type": "keyword"},
"title": {
"type": "text",
"analyzer": "english_analyzer",
"fields": {
"hindi": {"type": "text", "analyzer": "hindi_analyzer"},
"keyword": {"type": "keyword"}
}
},
"description": {"type": "text", "analyzer": "english_analyzer"},
"brand": {"type": "keyword"},
"category": {"type": "keyword"},
"category_path": {"type": "keyword"},
"attributes": {
"type": "nested",
"properties": {
"name": {"type": "keyword"},
"value": {"type": "keyword"}
}
},
"specifications": {
"type": "nested",
"properties": {
"key": {"type": "keyword"},
"value": {"type": "text"}
}
},
"price": {"type": "float"},
"selling_price": {"type": "float"},
"discount_percent": {"type": "integer"},
"ratings_average": {"type": "float"},
"ratings_count": {"type": "integer"},
"seller_rating": {"type": "float"},
"tags": {"type": "keyword"},
"search_keywords": {"type": "text"},
"status": {"type": "keyword"},
"in_stock": {"type": "boolean"},
"available_quantity": {"type": "integer"},
"created_at": {"type": "date"},
"updated_at": {"type": "date"},
"boost_score": {"type": "float"}, // For relevance boosting
"location": {
"type": "geo_point"
}
}
}
}
Infrastructure & Deployment
1. Kubernetes Deployment
yaml
# deployment/user-service.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
name: user-service
namespace: ecommerce
spec:
replicas: 10
selector:
matchLabels:
app: user-service
strategy:
rollingUpdate:
maxSurge: 2
maxUnavailable: 1
type: RollingUpdate
template:
metadata:
labels:
app: user-service
version: v1.2.0
spec:
containers:
- name: user-service
image: registry.flipkart.com/user-service:v1.2.0
ports:
- containerPort: 8080
env:
- name: ENVIRONMENT
value: "production"
- name: DB_HOST
valueFrom:
configMapKeyRef:
name: db-config
key: postgres_host
- name: REDIS_HOST
valueFrom:
configMapKeyRef:
name: redis-config
key: redis_host
resources:
requests:
memory: "512Mi"
cpu: "250m"
limits:
memory: "1Gi"
cpu: "500m"
livenessProbe:
httpGet:
path: /health
port: 8080
initialDelaySeconds: 30
periodSeconds: 10
readinessProbe:
httpGet:
path: /ready
port: 8080
initialDelaySeconds: 5
periodSeconds: 5
---
apiVersion: v1
kind: Service
metadata:
name: user-service
namespace: ecommerce
spec:
selector:
app: user-service
ports:
- port: 80
targetPort: 8080
protocol: TCP
type: ClusterIP
---
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
name: user-service-hpa
namespace: ecommerce
spec:
scaleTargetRef:
apiVersion: apps/v1
kind: Deployment
name: user-service
minReplicas: 5
maxReplicas: 50
metrics:
- type: Resource
resource:
name: cpu
target:
type: Utilization
averageUtilization: 70
- type: Resource
resource:
name: memory
target:
type: Utilization
averageUtilization: 80
- type: Pods
pods:
metric:
name: requests_per_second
target:
type: AverageValue
averageValue: 1000
2. Service Mesh (Istio) Configuration
yaml
# istio/user-service-virtual-service.yaml
apiVersion: networking.istio.io/v1beta1
kind: VirtualService
metadata:
name: user-service
namespace: ecommerce
spec:
hosts:
- user-service.ecommerce.svc.cluster.local
- api.flipkart.com
gateways:
- mesh
- flipkart-gateway
http:
- match:
- uri:
prefix: /api/v1/users
route:
- destination:
host: user-service.ecommerce.svc.cluster.local
port:
number: 80
weight: 100
timeout: 5s
retries:
attempts: 3
perTryTimeout: 2s
retryOn: gateway-error,connect-failure,refused-stream
corsPolicy:
allowOrigin:
- "*.flipkart.com"
allowMethods:
- GET
- POST
- PUT
- DELETE
- OPTIONS
allowHeaders:
- authorization
- content-type
headers:
request:
set:
x-request-id: "%REQ(x-request-id)%"
x-b3-traceid: "%REQ(x-b3-traceid)%"
x-b3-spanid: "%REQ(x-b3-spanid)%"
x-b3-parentspanid: "%REQ(x-b3-parentspanid)%"
x-b3-sampled: "%REQ(x-b3-sampled)%"
x-b3-flags: "%REQ(x-b3-flags)%"
x-ot-span-context: "%REQ(x-ot-span-context)%"
---
apiVersion: networking.istio.io/v1beta1
kind: DestinationRule
metadata:
name: user-service
namespace: ecommerce
spec:
host: user-service.ecommerce.svc.cluster.local
trafficPolicy:
connectionPool:
tcp:
maxConnections: 100
connectTimeout: 30ms
http:
http1MaxPendingRequests: 1000
http2MaxRequests: 1000
maxRequestsPerConnection: 10
maxRetries: 3
outlierDetection:
consecutive5xxErrors: 10
interval: 30s
baseEjectionTime: 30s
maxEjectionPercent: 100
subsets:
- name: v1
labels:
version: v1.2.0
- name: v2
labels:
version: v2.0.0
---
apiVersion: security.istio.io/v1beta1
kind: AuthorizationPolicy
metadata:
name: user-service-auth
namespace: ecommerce
spec:
selector:
matchLabels:
app: user-service
rules:
- from:
- source:
principals: ["cluster.local/ns/istio-system/sa/istio-ingressgateway-service-account"]
to:
- operation:
methods: ["GET", "POST", "PUT", "DELETE"]
paths: ["/api/v1/users/*"]
when:
- key: request.auth.claims[iss]
values: ["https://auth.flipkart.com/"]
3. Database Sharding Strategy
python
# database/sharding.py
class ShardingManager:
"""Manages database sharding for scale"""
def __init__(self):
self.shard_map = {} # Maps shard key to physical shard
self.zk = ZooKeeperClient()
async def get_shard_for_user(self, user_id: int) -> DatabaseConnection:
"""Get database shard for user"""
# Consistent hashing for shard distribution
shard_id = user_id % self.total_shards
# Check if shard exists in cache
if shard_id in self.shard_cache:
return self.shard_cache[shard_id]
# Get shard info from ZooKeeper
shard_info = await self.zk.get(f"/shards/{shard_id}")
# Create connection pool for shard
conn_pool = await self._create_connection_pool(shard_info)
# Cache connection
self.shard_cache[shard_id] = conn_pool
return conn_pool
async def create_user_shard(self, user_id: int, user_data: dict):
"""Create user in appropriate shard"""
shard = await self.get_shard_for_user(user_id)
async with shard.transaction():
# Insert user
await shard.execute("""
INSERT INTO users (id, email, phone, ...)
VALUES (%s, %s, %s, ...)
""", (user_id, user_data['email'], user_data['phone']))
# Insert profile
await shard.execute("""
INSERT INTO user_profiles (user_id, first_name, ...)
VALUES (%s, %s, ...)
""", (user_id, user_data['first_name']))
async def query_across_shards(self, query: str, params: tuple = None):
"""Query across all shards (for admin/reporting)"""
results = []
# Fan out to all shards
tasks = []
for shard_id in range(self.total_shards):
shard = await self.get_shard_for_shard_id(shard_id)
task = asyncio.create_task(
shard.execute(query, params)
)
tasks.append(task)
# Gather results
shard_results = await asyncio.gather(*tasks)
# Combine results
for result in shard_results:
results.extend(result)
return results
async def rebalance_shards(self):
"""Rebalance data across shards"""
# Monitor shard sizes
shard_sizes = await self.get_shard_sizes()
# Identify overloaded shards
avg_size = sum(shard_sizes.values()) / len(shard_sizes)
overloaded = [
shard_id for shard_id, size in shard_sizes.items()
if size > avg_size * 1.5
]
# Rebalance data
for shard_id in overloaded:
await self._move_data_from_shard(shard_id)
async def add_new_shard(self):
"""Add new shard to cluster"""
new_shard_id = self.total_shards
# Update total shards
self.total_shards += 1
# Update ZooKeeper
await self.zk.create(f"/shards/{new_shard_id}", json.dumps({
'host': f'db-shard-{new_shard_id}.flipkart.com',
'port': 5432,
'database': f'ecommerce_shard_{new_shard_id}',
'status': 'ACTIVE',
'created_at': datetime.utcnow().isoformat()
}))
# Update consistent hash ring
self._update_hash_ring()
4. Caching Strategy
python
# cache/strategy.py
class CacheManager:
"""Multi-layer caching strategy"""
def __init__(self):
self.l1_cache = RedisCluster() # In-memory, fast, limited
self.l2_cache = RedisCluster() # Persistent, larger
self.local_cache = LRUCache(maxsize=10000) # Process local
self.cache_config = self._load_cache_config()
async def get(self, key: str, loader: callable = None, ttl: int = None):
"""Get value from cache with fallback"""
# Try local cache first
value = self.local_cache.get(key)
if value is not None:
return value
# Try L1 cache
value = await self.l1_cache.get(key)
if value is not None:
# Update local cache
self.local_cache[key] = value
return value
# Try L2 cache
value = await self.l2_cache.get(key)
if value is not None:
# Update L1 and local cache
await self.l1_cache.setex(key, min(ttl or 300, 300), value)
self.local_cache[key] = value
return value
# Load from source if loader provided
if loader:
value = await loader()
if value is not None:
# Store in all cache layers
cache_ttl = ttl or self._get_default_ttl(key)
await self._set_with_consistent_hashing(key, value, cache_ttl)
self.local_cache[key] = value
return value
return None
async def _set_with_consistent_hashing(self, key: str, value: any, ttl: int):
"""Store value using consistent hashing for sharding"""
# Determine which Redis node to use
node = self._get_redis_node_for_key(key)
# Store with appropriate TTL
await node.setex(key, ttl, json.dumps(value))
# Set cache tags if configured
if 'tags' in self.cache_config.get(key, {}):
tags = self.cache_config[key]['tags']
await self._set_cache_tags(key, tags)
async def invalidate(self, key: str = None, pattern: str = None, tag: str = None):
"""Invalidate cache entries"""
if key:
# Invalidate specific key
await self.l1_cache.delete(key)
await self.l2_cache.delete(key)
self.local_cache.pop(key, None)
elif pattern:
# Invalidate by pattern
keys = await self.l1_cache.keys(pattern)
if keys:
await self.l1_cache.delete(*keys)
keys = await self.l2_cache.keys(pattern)
if keys:
await self.l2_cache.delete(*keys)
# Clear local cache (brute force for patterns)
self.local_cache.clear()
elif tag:
# Invalidate by tag
tag_keys = await self.l1_cache.smembers(f"cache_tag:{tag}")
if tag_keys:
await self.l1_cache.delete(*tag_keys)
await self.l1_cache.delete(f"cache_tag:{tag}")
tag_keys = await self.l2_cache.smembers(f"cache_tag:{tag}")
if tag_keys:
await self.l2_cache.delete(*tag_keys)
await self.l2_cache.delete(f"cache_tag:{tag}")
async def warm_cache(self, keys: List[str]):
"""Warm cache with frequently accessed data"""
# Preload product data
product_ids = self._get_hot_product_ids()
# Batch load from database
products = await self.product_service.get_products_by_ids(product_ids)
# Store in cache
for product in products:
key = f"product:{product['id']}"
await self.l1_cache.setex(key, 600, json.dumps(product))
await self.l2_cache.setex(key, 3600, json.dumps(product))
# Preload category data
categories = await self.category_service.get_all_categories()
for category in categories:
key = f"category:{category['id']}"
await self.l1_cache.setex(key, 3600, json.dumps(category))
def _get_default_ttl(self, key: str) -> int:
"""Get default TTL based on key pattern"""
patterns = {
'product:': 300, # 5 minutes
'user:': 600, # 10 minutes
'cart:': 2592000, # 30 days
'category:': 3600, # 1 hour
'inventory:': 30, # 30 seconds
'price:': 60, # 1 minute
'recommendation:': 300, # 5 minutes
}
for pattern, ttl in patterns.items():
if key.startswith(pattern):
return ttl
return 300 # Default 5 minutes
5. Message Queue & Event Streaming
python
# messaging/kafka_manager.py
class KafkaEventManager:
"""Event streaming with Kafka"""
def __init__(self):
self.producer = KafkaProducer(
bootstrap_servers=['kafka1.flipkart.com:9092',
'kafka2.flipkart.com:9092',
'kafka3.flipkart.com:9092'],
acks='all',
retries=3,
compression_type='snappy',
max_in_flight_requests_per_connection=1,
batch_size=16384,
linger_ms=5
)
self.consumers = {}
self.schema_registry = SchemaRegistryClient()
async def produce_event(self, topic: str, event: dict, key: str = None):
"""Produce event to Kafka with schema validation"""
# Validate event against schema
schema = await self.schema_registry.get_latest_schema(topic)
if not schema.validate(event):
raise InvalidEventSchema()
# Add metadata
enriched_event = {
'event_id': str(uuid.uuid4()),
'event_type': event['type'],
'timestamp': datetime.utcnow().isoformat(),
'source': 'ecommerce-api',
'version': '1.0',
'payload': event
}
# Determine partition key
partition_key = key or self._get_partition_key(event)
# Produce to Kafka
try:
await self.producer.send(
topic=topic,
key=partition_key.encode(),
value=json.dumps(enriched_event).encode(),
headers=[
('event_type', event['type'].encode()),
('source', 'ecommerce-api'.encode()),
('correlation_id', event.get('correlation_id', '').encode())
]
)
except KafkaError as e:
logger.error(f"Failed to produce event to {topic}: {e}")
# Fallback to dead letter queue
await self._send_to_dlq(topic, enriched_event, str(e))
def _get_partition_key(self, event: dict) -> str:
"""Get partition key for event"""
event_type = event['type']
if event_type.startswith('ORDER_'):
return event['payload']['order_id']
elif event_type.startswith('USER_'):
return event['payload']['user_id']
elif event_type.startswith('PRODUCT_'):
return event['payload']['product_id']
elif event_type.startswith('PAYMENT_'):
return event['payload']['payment_id']
# Default: round-robin
return str(uuid.uuid4())
async def consume_events(self, topic: str, group_id: str,
processor: callable):
"""Consume and process events from Kafka"""
consumer = KafkaConsumer(
topic,
bootstrap_servers=['kafka1.flipkart.com:9092',
'kafka2.flipkart.com:9092'],
group_id=group_id,
auto_offset_reset='earliest',
enable_auto_commit=False,
max_poll_records=100,
max_poll_interval_ms=300000,
session_timeout_ms=10000,
heartbeat_interval_ms=3000
)
self.consumers[group_id] = consumer
while True:
try:
# Poll for messages
batch = await consumer.poll(timeout_ms=1000)
for topic_partition, messages in batch.items():
for message in messages:
try:
# Process message
event = json.loads(message.value.decode())
await processor(event)
# Commit offset
await consumer.commit({
topic_partition: OffsetAndMetadata(
message.offset + 1, None
)
})
except Exception as e:
logger.error(f"Failed to process message: {e}")
# Send to dead letter queue
await self._send_to_dlq(
topic,
message.value.decode(),
str(e)
)
# Commit anyway to avoid reprocessing
await consumer.commit({
topic_partition: OffsetAndMetadata(
message.offset + 1, None
)
})
except Exception as e:
logger.error(f"Consumer error: {e}")
await asyncio.sleep(5)
async def create_topic(self, topic: str, partitions: int = 10,
replication_factor: int = 3):
"""Create Kafka topic with appropriate configuration"""
admin_client = KafkaAdminClient(
bootstrap_servers=['kafka1.flipkart.com:9092']
)
topic_config = {
'cleanup.policy': 'compact,delete',
'retention.ms': '604800000', # 7 days
'retention.bytes': '1073741824', # 1GB
'segment.bytes': '268435456', # 256MB
'compression.type': 'snappy',
'max.message.bytes': '10485760', # 10MB
'delete.retention.ms': '86400000' # 1 day
}
try:
await admin_client.create_topics([
NewTopic(
name=topic,
num_partitions=partitions,
replication_factor=replication_factor,
topic_configs=topic_config
)
])
except TopicAlreadyExistsError:
logger.info(f"Topic {topic} already exists")
Monitoring & Observability
1. Metrics Collection
python
# monitoring/metrics.py
class MetricsCollector:
"""Collect and expose system metrics"""
def __init__(self):
self.prometheus_client = PrometheusClient()
self.statsd = StatsDClient()
self.datadog = DatadogClient()
async def record_api_metrics(self, endpoint: str, method: str,
status_code: int, duration: float):
"""Record API metrics"""
# Prometheus metrics
self.prometheus_client.counter(
'http_requests_total',
labels={
'endpoint': endpoint,
'method': method,
'status': str(status_code)
}
).inc()
self.prometheus_client.histogram(
'http_request_duration_seconds',
labels={'endpoint': endpoint, 'method': method}
).observe(duration)
# StatsD metrics
self.statsd.increment(
f'api.{endpoint}.{method}.{status_code}'
)
self.statsd.timing(
f'api.{endpoint}.{method}.duration',
duration * 1000 # Convert to milliseconds
)
# Business metrics
if endpoint == '/api/v1/orders' and method == 'POST':
self.prometheus_client.counter(
'orders_created_total'
).inc()
if endpoint == '/api/v1/payments/webhook' and status_code == 200:
self.prometheus_client.counter(
'payments_processed_total'
).inc()
async def record_database_metrics(self, query: str, duration: float,
success: bool):
"""Record database metrics"""
# Extract query type
query_type = self._extract_query_type(query)
self.prometheus_client.histogram(
'database_query_duration_seconds',
labels={'query_type': query_type}
).observe(duration)
if not success:
self.prometheus_client.counter(
'database_errors_total',
labels={'query_type': query_type}
).inc()
async def record_cache_metrics(self, operation: str, cache_type: str,
hit: bool, duration: float):
"""Record cache metrics"""
self.prometheus_client.counter(
'cache_operations_total',
labels={
'operation': operation,
'cache_type': cache_type,
'hit': str(hit)
}
).inc()
self.prometheus_client.histogram(
'cache_operation_duration_seconds',
labels={'operation': operation, 'cache_type': cache_type}
).observe(duration)
# Cache hit rate
self.prometheus_client.gauge(
'cache_hit_rate',
labels={'cache_type': cache_type}
).set(self._calculate_hit_rate(cache_type))
async def record_business_metrics(self):
"""Record business KPIs"""
# Real-time revenue
revenue = await self._calculate_hourly_revenue()
self.prometheus_client.gauge(
'revenue_hourly'
).set(revenue)
# Active users
active_users = await self._count_active_users()
self.prometheus_client.gauge(
'active_users'
).set(active_users)
# Conversion rate
conversions = await self._count_conversions()
sessions = await self._count_sessions()
conversion_rate = conversions / sessions if sessions > 0 else 0
self.prometheus_client.gauge(
'conversion_rate'
).set(conversion_rate)
# Average order value
aov = await self._calculate_aov()
self.prometheus_client.gauge(
'average_order_value'
).set(aov)
async def export_metrics(self):
"""Export metrics to monitoring systems"""
# Export to Prometheus
await self.prometheus_client.push_to_gateway(
'prometheus.flipkart.com:9091',
job='ecommerce-api'
)
# Export to Datadog
await self.datadog.export_metrics(
self._get_all_metrics()
)
# Export to CloudWatch
await self._export_to_cloudwatch()
2. Distributed Tracing
python
# monitoring/tracing.py
class TracingManager:
"""Distributed tracing with OpenTelemetry"""
def __init__(self):
self.tracer_provider = TracerProvider()
self.tracer = self.tracer_provider.get_tracer("ecommerce")
# Configure exporters
self.jaeger_exporter = JaegerExporter(
agent_host_name='jaeger.flipkart.com',
agent_port=6831
)
self.otlp_exporter = OTLPSpanExporter(
endpoint='otlp.flipkart.com:4317'
)
# Configure span processors
self.tracer_provider.add_span_processor(
BatchSpanProcessor(self.jaeger_exporter)
)
self.tracer_provider.add_span_processor(
BatchSpanProcessor(self.otlp_exporter)
)
# Auto-instrumentation
instrument_requests()
instrument_sqlalchemy()
instrument_redis()
instrument_kafka()
def trace_request(self, request, endpoint: str):
"""Trace HTTP request"""
span_name = f"{request.method} {endpoint}"
with self.tracer.start_as_current_span(span_name) as span:
# Add request attributes
span.set_attributes({
"http.method": request.method,
"http.url": str(request.url),
"http.route": endpoint,
"http.user_agent": request.headers.get("user-agent", ""),
"http.client_ip": request.client.host,
})
# Add custom business attributes
user_id = request.headers.get("x-user-id")
if user_id:
span.set_attribute("user.id", user_id)
# Set span as active
token = attach(set_span_in_context(span))
try:
yield span
finally:
detach(token)
def trace_database_query(self, query: str, params: tuple = None):
"""Trace database query"""
with self.tracer.start_as_current_span("database.query") as span:
span.set_attributes({
"db.system": "postgresql",
"db.statement": query,
"db.params": str(params) if params else "",
})
yield span
def trace_external_call(self, service: str, endpoint: str):
"""Trace external service call"""
with self.tracer.start_as_current_span(f"{service}.call") as span:
span.set_attributes({
"rpc.system": "http",
"rpc.service": service,
"rpc.method": endpoint,
})
# Inject trace context into headers
headers = {}
inject(headers)
yield span, headers
def record_exception(self, exception: Exception):
"""Record exception in current span"""
current_span = get_current_span()
if current_span:
current_span.record_exception(exception)
current_span.set_status(StatusCode.ERROR)
def add_business_attributes(self, attributes: dict):
"""Add business attributes to current span"""
current_span = get_current_span()
if current_span:
for key, value in attributes.items():
current_span.set_attribute(f"business.{key}", value)
3. Logging Infrastructure
python
# monitoring/logging.py
class StructuredLogger:
"""Structured logging with correlation"""
def __init__(self):
self.logger = structlog.get_logger()
# Configure processors
structlog.configure(
processors=[
structlog.stdlib.filter_by_level,
structlog.stdlib.add_logger_name,
structlog.stdlib.add_log_level,
structlog.stdlib.PositionalArgumentsFormatter(),
structlog.processors.TimeStamper(fmt="iso"),
structlog.processors.StackInfoRenderer(),
structlog.processors.format_exc_info,
structlog.processors.UnicodeDecoder(),
self._add_correlation_id,
self._add_service_context,
structlog.processors.JSONRenderer()
],
context_class=dict,
logger_factory=structlog.PrintLoggerFactory(),
cache_logger_on_first_use=True,
)
def _add_correlation_id(self, logger, method_name, event_dict):
"""Add correlation ID to logs"""
correlation_id = self._get_correlation_id()
if correlation_id:
event_dict['correlation_id'] = correlation_id
return event_dict
def _add_service_context(self, logger, method_name, event_dict):
"""Add service context to logs"""
event_dict.update({
'service': 'ecommerce-api',
'environment': os.getenv('ENVIRONMENT', 'development'),
'version': os.getenv('VERSION', '1.0.0'),
'host': socket.gethostname(),
'pod': os.getenv('POD_NAME', 'unknown'),
})
return event_dict
async def log_api_request(self, request, response, duration: float):
"""Log API request/response"""
self.logger.info(
"api_request",
method=request.method,
endpoint=request.url.path,
status_code=response.status_code,
duration_ms=duration * 1000,
user_agent=request.headers.get("user-agent"),
client_ip=request.client.host,
query_params=dict(request.query_params),
user_id=request.headers.get("x-user-id"),
)
async def log_business_event(self, event_type: str, event_data: dict):
"""Log business events"""
self.logger.info(
"business_event",
event_type=event_type,
**event_data
)
async def log_error(self, error: Exception, context: dict = None):
"""Log errors with context"""
self.logger.error(
"error_occurred",
error_type=type(error).__name__,
error_message=str(error),
stack_trace=traceback.format_exc(),
**(context or {})
)
async def log_performance_issue(self, metric: str, value: float,
threshold: float, context: dict):
"""Log performance issues"""
self.logger.warning(
"performance_issue",
metric=metric,
value=value,
threshold=threshold,
**context
)
def _get_correlation_id(self):
"""Get correlation ID from context"""
# Try to get from OpenTelemetry context
span = get_current_span()
if span:
span_context = span.get_span_context()
if span_context.trace_id:
return format_trace_id(span_context.trace_id)
# Try to get from request headers
try:
from fastapi import Request
request = Request()
return request.headers.get("x-correlation-id")
except:
return None
Security Implementation
1. Authentication & Authorization
python
# security/auth.py
class AuthManager:
"""Authentication and authorization manager"""
def __init__(self):
self.jwt_secret = os.getenv("JWT_SECRET")
self.jwt_algorithm = "HS256"
self.oauth_providers = {
'google': GoogleOAuth(),
'facebook': FacebookOAuth(),
'apple': AppleOAuth(),
}
self.rbac = RoleBasedAccessControl()
async def authenticate_jwt(self, token: str) -> dict:
"""Authenticate JWT token"""
try:
# Decode token
payload = jwt.decode(
token,
self.jwt_secret,
algorithms=[self.jwt_algorithm]
)
# Check if token is blacklisted
if await self._is_token_blacklisted(token):
raise TokenRevoked()
# Check expiration
if datetime.fromtimestamp(payload['exp']) < datetime.utcnow():
raise TokenExpired()
# Get user from database
user = await self.user_service.get_user(payload['user_id'])
if not user or not user['is_active']:
raise UserNotFound()
return {
'user_id': user['id'],
'email': user['email'],
'roles': user['roles'],
'permissions': await self.rbac.get_permissions(user['roles'])
}
except jwt.PyJWTError as e:
raise InvalidToken(str(e))
async def authorize(self, user_ctx: dict, resource: str,
action: str) -> bool:
"""Authorize user action on resource"""
# Get user permissions
permissions = user_ctx.get('permissions', [])
# Check permission
required_permission = f"{resource}:{action}"
if required_permission not in permissions:
# Check role hierarchy
user_roles = user_ctx.get('roles', [])
for role in user_roles:
role_permissions = await self.rbac.get_role_permissions(role)
if required_permission in role_permissions:
return True
return False
return True
async def authenticate_oauth(self, provider: str, code: str) -> dict:
"""Authenticate using OAuth provider"""
if provider not in self.oauth_providers:
raise InvalidProvider()
oauth_client = self.oauth_providers[provider]
# Exchange code for token
token_data = await oauth_client.exchange_code(code)
# Get user info
user_info = await oauth_client.get_user_info(token_data['access_token'])
# Find or create user
user = await self._find_or_create_oauth_user(provider, user_info)
# Generate JWT
jwt_token = self.generate_jwt(user)
return {
'access_token': jwt_token,
'refresh_token': self.generate_refresh_token(user['id']),
'user': user
}
async def enforce_rate_limit(self, identifier: str,
action: str, limit: int = 10) -> bool:
"""Enforce rate limiting"""
key = f"rate_limit:{identifier}:{action}"
# Use Redis for rate limiting
current = await self.redis.incr(key)
if current == 1:
# Set expiration
await self.redis.expire(key, 60) # 1 minute
return current <= limit
async def audit_log(self, user_id: str, action: str,
resource: str, details: dict = None):
"""Log security audit trail"""
await self.audit_db.execute("""
INSERT INTO audit_logs (
user_id, action, resource, details,
ip_address, user_agent, timestamp
) VALUES (%s, %s, %s, %s, %s, %s, NOW())
""", (
user_id, action, resource,
json.dumps(details or {}),
self._get_client_ip(),
self._get_user_agent()
))
2. Payment Security (PCI DSS Compliance)
python
# security/payments.py
class PaymentSecurity:
"""PCI DSS compliant payment handling"""
def __init__(self):
self.vault = HashiCorpVault() # For secret management
self.hsm = HardwareSecurityModule() # For key management
self.encryption = AESGCMEncryption()
async def tokenize_card(self, card_data: dict) -> str:
"""Tokenize card data (PCI DSS requirement)"""
# Validate card data
if not self._validate_card_data(card_data):
raise InvalidCardData()
# Generate token
token = str(uuid.uuid4())
# Encrypt card data with HSM-managed key
encrypted_data = await self.hsm.encrypt(
json.dumps(card_data).encode()
)
# Store in secure vault
await self.vault.store_secret(
path=f"cards/{token}",
data=encrypted_data,
metadata={
'created_at': datetime.utcnow().isoformat(),
'last_used': None,
'usage_count': 0
}
)
return token
async def detokenize_card(self, token: str) -> dict:
"""Detokenize card data"""
# Get from vault
encrypted_data = await self.vault.get_secret(f"cards/{token}")
if not encrypted_data:
raise TokenNotFound()
# Decrypt with HSM
decrypted = await self.hsm.decrypt(encrypted_data)
# Parse card data
card_data = json.loads(decrypted.decode())
# Update usage metadata
await self.vault.update_metadata(
path=f"cards/{token}",
metadata={'last_used': datetime.utcnow().isoformat()}
)
return card_data
async def process_payment(self, payment_data: dict) -> dict:
"""Process payment securely"""
# Validate payment data
self._validate_payment_data(payment_data)
# Check for fraud
fraud_score = await self._check_fraud(payment_data)
if fraud_score > 0.8:
raise SuspectedFraud()
# Tokenize if card data present
if 'card_number' in payment_data:
token = await self.tokenize_card({
'number': payment_data['card_number'],
'expiry': payment_data['card_expiry'],
'cvv': payment_data['card_cvv']
})
payment_data['card_token'] = token
del payment_data['card_number']
del payment_data['card_expiry']
del payment_data['card_cvv']
# Call payment gateway
gateway_response = await self._call_payment_gateway(payment_data)
# Log payment attempt (without sensitive data)
await self._log_payment_attempt(payment_data, gateway_response)
return gateway_response
async def _check_fraud(self, payment_data: dict) -> float:
"""Check for fraudulent patterns"""
fraud_indicators = []
# Check velocity (too many attempts)
attempts = await self.redis.get(
f"payment_attempts:{payment_data.get('user_id')}"
)
if attempts and int(attempts) > 5:
fraud_indicators.append('HIGH_VELOCITY')
# Check IP geolocation
ip_location = await self._get_ip_location(payment_data.get('ip_address'))
user_location = payment_data.get('billing_address', {}).get('country')
if ip_location != user_location:
fraud_indicators.append('LOCATION_MISMATCH')
# Check card BIN
card_bin = payment_data.get('card_number', '')[:6]
bin_country = await self._get_bin_country(card_bin)
if bin_country != user_location:
fraud_indicators.append('BIN_MISMATCH')
# Calculate fraud score
score = len(fraud_indicators) * 0.2
# Check with ML fraud detection service
ml_score = await self._call_fraud_detection_service(payment_data)
score = max(score, ml_score)
return score
Performance Optimization
1. CDN & Edge Caching
python
# cdn/manager.py
class CDNManager:
"""Manage CDN for static assets and dynamic content"""
def __init__(self):
self.cloudfront = CloudFrontClient()
self.akamai = AkamaiClient()
self.fastly = FastlyClient()
self.s3 = S3Client()
async def distribute_product_images(self, product_id: str, images: List[dict]):
"""Distribute product images to CDN"""
processed_images = []
for image in images:
# Original image
original_key = f"products/{product_id}/original/{image['filename']}"
await self.s3.upload_file(
image['path'],
'flipkart-assets',
original_key
)
# Generate variants
variants = await self._generate_image_variants(image['path'])
for variant_name, variant_data in variants.items():
variant_key = f"products/{product_id}/{variant_name}/{image['filename']}"
await self.s3.upload_fileobj(
variant_data,
'flipkart-assets',
variant_key
)
# Get CDN URL
cdn_url = self.cloudfront.generate_url(variant_key)
processed_images.append({
'type': image['type'],
'variant': variant_name,
'url': cdn_url,
'width': variant_data.get('width'),
'height': variant_data.get('height'),
'size': len(variant_data)
})
# Create image manifest
manifest = {
'product_id': product_id,
'images': processed_images,
'generated_at': datetime.utcnow().isoformat()
}
manifest_key = f"products/{product_id}/manifest.json"
await self.s3.put_object(
'flipkart-assets',
manifest_key,
json.dumps(manifest)
)
# Invalidate CDN cache
await self.cloudfront.create_invalidation(
distribution_id='E1A2B3C4D5E6F7',
paths=[f'/products/{product_id}/*']
)
return processed_images
async def cache_dynamic_content(self, path: str, content: str,
ttl: int = 300):
"""Cache dynamic content at edge"""
# Generate cache key
cache_key = hashlib.md5(path.encode()).hexdigest()
# Store in edge cache
await self.fastly.purge_key(cache_key)
await self.fastly.set(
key=cache_key,
value=content,
ttl=ttl,
tags=['dynamic', 'product-page']
)
# Set cache headers
headers = {
'Cache-Control': f'public, max-age={ttl}, s-maxage={ttl}',
'CDN-Cache-Control': f'max-age={ttl}',
'Vary': 'Accept-Encoding, User-Agent',
}
return headers
async def prefetch_assets(self, user_id: str, session_data: dict):
"""Prefetch assets based on user behavior"""
# Analyze user behavior
predicted_products = await self._predict_next_products(user_id)
# Prefetch product images
for product_id in predicted_products[:5]:
manifest_key = f"products/{product_id}/manifest.json"
# Check if manifest exists in edge cache
cached = await self.fastly.get(manifest_key)
if not cached:
# Fetch from origin and cache
manifest = await self.s3.get_object(
'flipkart-assets', manifest_key
)
await self.fastly.set(
key=manifest_key,
value=manifest,
ttl=3600
)
# Prefetch images
for image in manifest.get('images', []):
if image['variant'] == 'thumbnail':
await self.fastly.prefetch(image['url'])
async def optimize_delivery(self, request):
"""Optimize content delivery based on client"""
client_capabilities = self._detect_client_capabilities(request)
optimizations = {
'image_format': 'webp' if client_capabilities.get('webp') else 'jpeg',
'image_quality': 85 if client_capabilities.get('high_dpi') else 70,
'video_codec': 'h265' if client_capabilities.get('h265') else 'h264',
'compression': 'br' if client_capabilities.get('brotli') else 'gzip',
}
# Adaptive bitrate for videos
if client_capabilities.get('bandwidth'):
bandwidth = client_capabilities['bandwidth']
if bandwidth > 5000000: # 5 Mbps
optimizations['video_bitrate'] = 'high'
elif bandwidth > 1000000: # 1 Mbps
optimizations['video_bitrate'] = 'medium'
else:
optimizations['video_bitrate'] = 'low'
return optimizations
2. Database Optimization
python
# database/optimization.py
class DatabaseOptimizer:
"""Database performance optimization"""
def __init__(self):
self.pg_stat_statements = PostgreSQLStats()
self.query_analyzer = QueryAnalyzer()
self.index_manager = IndexManager()
async def analyze_performance(self):
"""Analyze database performance"""
# Get slow queries
slow_queries = await self.pg_stat_statements.get_slow_queries(
threshold=100 # 100ms
)
recommendations = []
for query in slow_queries:
# Analyze query plan
plan = await self.query_analyzer.explain_query(query['query'])
# Generate recommendations
recs = self._generate_recommendations(query, plan)
recommendations.extend(recs)
# Check for missing indexes
missing_indexes = await self.index_manager.find_missing_indexes(query)
if missing_indexes:
recommendations.append({
'type': 'CREATE_INDEX',
'query': query['query'],
'indexes': missing_indexes,
'estimated_improvement': '70-90%'
})
return recommendations
async def optimize_queries(self):
"""Optimize frequently run queries"""
# Get top N queries by execution time
top_queries = await self.pg_stat_statements.get_top_queries(
limit=50,
order_by='total_time'
)
optimized_queries = []
for query in top_queries:
optimized = await self._optimize_single_query(query)
if optimized:
optimized_queries.append(optimized)
# Create optimized query cache
await self._create_query_cache(optimized_queries)
return optimized_queries
async def manage_indexes(self):
"""Manage database indexes"""
# Analyze index usage
index_usage = await self.index_manager.get_index_usage()
# Identify unused indexes
unused_indexes = [
idx for idx, usage in index_usage.items()
if usage['scan'] == 0 and usage['read'] == 0
]
# Identify duplicate indexes
duplicate_indexes = await self.index_manager.find_duplicate_indexes()
# Create missing indexes (based on query analysis)
missing_indexes = await self.index_manager.find_missing_indexes_from_workload()
actions = {
'drop_unused': unused_indexes[:10], # Limit to 10 at a time
'drop_duplicate': duplicate_indexes,
'create': missing_indexes[:5] # Limit to 5 at a time
}
# Execute actions
await self._execute_index_actions(actions)
return actions
async def partition_tables(self):
"""Partition large tables"""
# Identify candidate tables for partitioning
large_tables = await self._get_large_tables(threshold_gb=50)
partitioning_plans = []
for table in large_tables:
# Determine partitioning strategy
strategy = self._determine_partitioning_strategy(table)
if strategy:
plan = {
'table': table['name'],
'strategy': strategy['type'],
'column': strategy['column'],
'estimated_improvement': '60-80%'
}
partitioning_plans.append(plan)
# Execute partitioning in background
if partitioning_plans:
await self._execute_partitioning(partitioning_plans)
return partitioning_plans
async def vacuum_and_analyze(self):
"""Perform vacuum and analyze operations"""
# Identify tables needing vacuum
tables_needing_vacuum = await self._get_tables_needing_vacuum()
# Perform aggressive vacuum on critical tables
critical_tables = ['orders', 'users', 'products']
for table in critical_tables:
if table in tables_needing_vacuum:
await self._vacuum_table(table, aggressive=True)
# Perform analyze to update statistics
await self._analyze_tables(['orders', 'order_items', 'users'])
# Update query planner statistics
await self._update_planner_stats()
Scalability Patterns
1. Auto-scaling Configuration
python
# scaling/autoscaler.py
class AutoScaler:
"""Auto-scaling based on metrics"""
def __init__(self):
self.prometheus = PrometheusClient()
self.k8s = KubernetesClient()
self.cloudwatch = CloudWatchClient()
async def scale_based_on_metrics(self):
"""Scale services based on metrics"""
services = await self._get_services_to_scale()
for service in services:
# Get metrics
metrics = await self._get_service_metrics(service)
# Calculate desired replica count
desired_replicas = self._calculate_desired_replicas(
service, metrics
)
# Get current replica count
current_replicas = await self.k8s.get_replica_count(service)
# Scale if needed
if desired_replicas != current_replicas:
await self.k8s.scale_service(
service, desired_replicas
)
logger.info(
f"Scaled {service} from {current_replicas} to {desired_replicas}"
)
async def scale_for_special_events(self):
"""Scale for special events (Big Billion Days, etc.)"""
# Check for upcoming events
upcoming_events = await self._get_upcoming_events()
for event in upcoming_events:
# Calculate expected load
expected_load = self._calculate_expected_load(event)
# Scale services preemptively
services_to_scale = self._get_services_for_event(event)
for service in services_to_scale:
desired_replicas = self._calculate_event_replicas(
service, expected_load
)
await self.k8s.scale_service(service, desired_replicas)
# Warm up instances
await self._warm_up_service(service, desired_replicas)
async def scale_based_on_business_metrics(self):
"""Scale based on business metrics (orders, revenue, etc.)"""
# Get current business metrics
metrics = await self._get_business_metrics()
# Calculate scaling factors
scaling_factors = {
'order_rate': metrics['orders_per_minute'] / 100,
'revenue_rate': metrics['revenue_per_minute'] / 10000,
'user_count': metrics['active_users'] / 1000,
}
# Apply to services
for service, base_replicas in self.service_base_replicas.items():
# Determine which metric to use
metric = self.service_metrics.get(service, 'order_rate')
# Calculate replicas
desired_replicas = max(
base_replicas,
int(base_replicas * scaling_factors[metric])
)
# Cap at maximum
desired_replicas = min(desired_replicas, self.service_max_replicas[service])
# Scale
await self.k8s.scale_service(service, desired_replicas)
def _calculate_desired_replicas(self, service: str, metrics: dict) -> int:
"""Calculate desired number of replicas"""
# Get configuration
config = self.scaling_config[service]
# Calculate based on CPU
cpu_replicas = self._calculate_cpu_based_replicas(
metrics['cpu'], config
)
# Calculate based on memory
memory_replicas = self._calculate_memory_based_replicas(
metrics['memory'], config
)
# Calculate based on queue length
queue_replicas = self._calculate_queue_based_replicas(
metrics.get('queue_length', 0), config
)
# Calculate based on latency
latency_replicas = self._calculate_latency_based_replicas(
metrics.get('latency_p99', 0), config
)
# Take maximum of all calculations
desired_replicas = max(
cpu_replicas, memory_replicas,
queue_replicas, latency_replicas
)
# Apply bounds
desired_replicas = max(config['min'], desired_replicas)
desired_replicas = min(config['max'], desired_replicas)
return desired_replicas
Disaster Recovery & Backup
1. Backup Strategy
python
# backup/manager.py
class BackupManager:
"""Manage database backups and recovery"""
def __init__(self):
self.s3 = S3Client()
self.rds = RDSClient()
self.mongodb = MongoDBClient()
self.cassandra = CassandraClient()
async def perform_backups(self):
"""Perform scheduled backups"""
# PostgreSQL backups
await self._backup_postgresql()
# MongoDB backups
await self._backup_mongodb()
# Cassandra backups
await self._backup_cassandra()
# Redis backups
await self._backup_redis()
# Verify backups
await self._verify_backups()
async def _backup_postgresql(self):
"""Backup PostgreSQL databases"""
databases = ['users', 'orders', 'payments', 'catalog']
for db in databases:
# Create backup
backup_file = f"postgresql-{db}-{datetime.now().strftime('%Y%m%d-%H%M%S')}.sql.gz"
# Use pg_dump with compression
await self.rds.create_snapshot(
db_instance=db,
snapshot_id=backup_file
)
# Copy to S3 for long-term storage
await self.s3.upload_file(
f"/backups/{backup_file}",
'flipkart-backups',
f"postgresql/daily/{backup_file}"
)
# Create WAL archive for point-in-time recovery
await self._archive_wal_files(db)
async def setup_disaster_recovery(self):
"""Setup disaster recovery procedures"""
# Setup cross-region replication
await self._setup_cross_region_replication()
# Setup read replicas
await self._setup_read_replicas()
# Setup failover configuration
await self._setup_failover()
# Test disaster recovery
await self._test_disaster_recovery()
async def perform_recovery(self, point_in_time: datetime = None):
"""Perform disaster recovery"""
if point_in_time:
# Point-in-time recovery
await self._recover_to_point_in_time(point_in_time)
else:
# Latest recovery
await self._recover_latest()
# Verify recovery
await self._verify_recovery()
# Switch traffic
await self._switch_traffic_to_recovery()
Cost Optimization
1. Resource Optimization
python
# cost/optimizer.py
class CostOptimizer:
"""Optimize cloud costs"""
def __init__(self):
self.aws_cost_explorer = AWSCostExplorer()
self.gcp_billing = GCPBilling()
self.azure_cost_management = AzureCostManagement()
async def analyze_costs(self):
"""Analyze and optimize costs"""
# Get cost breakdown
costs = await self.aws_cost_explorer.get_cost_breakdown()
recommendations = []
# Analyze EC2 instances
ec2_recommendations = await self._optimize_ec2_instances()
recommendations.extend(ec2_recommendations)
# Analyze RDS instances
rds_recommendations = await self._optimize_rds_instances()
recommendations.extend(rds_recommendations)
# Analyze S3 storage
s3_recommendations = await self._optimize_s3_storage()
recommendations.extend(s3_recommendations)
# Analyze data transfer costs
transfer_recommendations = await self._optimize_data_transfer()
recommendations.extend(transfer_recommendations)
return recommendations
async def _optimize_ec2_instances(self):
"""Optimize EC2 instance usage"""
instances = await self.aws_cost_explorer.get_ec2_instances()
recommendations = []
for instance in instances:
# Check utilization
utilization = await self._get_instance_utilization(instance['id'])
if utilization['cpu'] < 20 and utilization['memory'] < 30:
# Underutilized - recommend downsizing
recommendations.append({
'type': 'DOWN_SIZE_INSTANCE',
'instance_id': instance['id'],
'current_type': instance['type'],
'recommended_type': self._get_smaller_type(instance['type']),
'estimated_savings': self._calculate_savings(instance)
})
elif utilization['cpu'] > 80 or utilization['memory'] > 90:
# Overutilized - recommend upsizing
recommendations.append({
'type': 'UP_SIZE_INSTANCE',
'instance_id': instance['id'],
'current_type': instance['type'],
'recommended_type': self._get_larger_type(instance['type'])
})
return recommendations
This comprehensive system design for a Flipkart-like e-commerce platform covers:
Key Takeaways:
1. Architecture Principles:
-
Microservices: Independent, scalable services
-
Event-Driven: Loose coupling with Kafka
-
CQRS: Separate read/write models
-
Caching: Multi-layer caching strategy
-
CDN: Edge caching for static assets
2. Scalability Patterns:
-
Sharding: Horizontal scaling of databases
-
Replication: Read replicas for heavy reads
-
Partitioning: Time-based and hash-based
-
Auto-scaling: Based on metrics and events
3. Reliability Measures:
-
Circuit Breakers: Prevent cascading failures
-
Retries: With exponential backoff
-
Fallbacks: Graceful degradation
-
Monitoring: Comprehensive observability
4. Security Layers:
-
Authentication: JWT with refresh tokens
-
Authorization: RBAC with fine-grained permissions
-
Encryption: Data at rest and in transit
-
Auditing: Comprehensive audit trails
5. Performance Optimizations:
-
CDN: Global asset distribution
-
Caching: Redis for hot data
-
Database: Indexing and query optimization
-
Compression: Gzip, Brotli, WebP
6. Cost Optimization:
-
Spot Instances: For non-critical workloads
-
Reserved Instances: For predictable loads
-
Auto-scaling: Right-size resources
-
Storage Tiers: Move cold data to cheaper storage
This design can handle 10M+ daily users, 1M+ concurrent users, and 100K+ orders per minute during peak events like Big Billion Days.
Comments (0)
No comments yet. Be the first to comment!