LLD Problems (20 Complete Designs)

Problems List

1. URL Shortener Service
2. Rate Limiter
3. LRU Cache
4. Parking Lot System
5. Elevator System
6. Pub/Sub Messaging
7. Task Scheduler (Cron)
8. Snake & Ladder Game
9. Online Bookstore
10. Splitwise / Expense Tracker
11. Hotel Booking System
12. Movie Ticket Booking (BookMyShow)
13. Logger Framework
14. Notification Service
15. Payment Gateway
16. File System (Linux-like)
17. ATM Machine
18. Vending Machine (State Pattern)
19. Chat Application (LLD)
20. API Rate Limiter + API Gateway Design

---

1. URL Shortener Service

Requirements

• Shorten long URLs to a 7-character alphanumeric code
• Redirect short URL to original with 301/302 response
• Optional custom aliases and expiration dates
• Analytics: track click count, referrer, geo data
• High read throughput (100:1 read-to-write ratio)

Class Diagram

URLShortenerService
├── create_short_url(long_url, custom_alias?, expiry?) -> ShortURL
├── resolve(short_code) -> str
└── get_analytics(short_code) -> Analytics

ShortURL
├── short_code: str
├── long_url: str
├── created_at: datetime
├── expires_at: datetime | None
├── user_id: str | None
└── click_count: int

CodeGenerator (interface)
├── Base62Generator
├── MD5HashGenerator
└── SnowflakeGenerator

URLRepository (interface)
├── save(short_url: ShortURL)
├── find_by_code(code: str) -> ShortURL
└── delete(code: str)

AnalyticsCollector
├── record_click(short_code, metadata)
└── get_stats(short_code) -> Analytics

Python Code

import hashlib
import time
import string
from abc import ABC, abstractmethod
from datetime import datetime, timedelta

class CodeGenerator(ABC):
    @abstractmethod
    def generate(self, long_url: str) -> str:
        pass

class Base62Generator(CodeGenerator):
    CHARS = string.ascii_letters + string.digits
    COUNTER = 0

    def generate(self, long_url: str) -> str:
        Base62Generator.COUNTER += 1
        num = int(hashlib.md5(
            f"{long_url}:{Base62Generator.COUNTER}".encode()
        ).hexdigest()[:12], 16)
        code = []
        for _ in range(7):
            code.append(self.CHARS[num % 62])
            num //= 62
        return "".join(code)

class ShortURL:
    def __init__(self, short_code: str, long_url: str, expires_at=None, user_id=None):
        self.short_code = short_code
        self.long_url = long_url
        self.created_at = datetime.now()
        self.expires_at = expires_at
        self.user_id = user_id
        self.click_count = 0

    def is_expired(self) -> bool:
        if self.expires_at is None:
            return False
        return datetime.now() > self.expires_at

class URLShortenerService:
    def __init__(self):
        self.store = {} # short_code -> ShortURL
        self.generator = Base62Generator()

    def create_short_url(self, long_url: str, custom_alias=None, ttl_days=None):
        code = custom_alias or self.generator.generate(long_url)
        if code in self.store:
            raise ValueError("Alias already exists")
        expires = datetime.now() + timedelta(days=ttl_days) if ttl_days else None
        short_url = ShortURL(code, long_url, expires)
        self.store[code] = short_url
        return short_url

    def resolve(self, short_code: str) -> str:
        entry = self.store.get(short_code)
        if not entry:
            raise KeyError("URL not found")
        if entry.is_expired():
            del self.store[short_code]
            raise KeyError("URL expired")
        entry.click_count += 1
        return entry.long_url

    def get_analytics(self, short_code: str) -> dict:
        entry = self.store.get(short_code)
        if not entry:
            raise KeyError("URL not found")
        return {
            "short_code": entry.short_code,
            "clicks": entry.click_count,
            "created": entry.created_at.isoformat(),
        }

Design Patterns

• Strategy: CodeGenerator allows swapping encoding algorithms
• Repository: URLRepository abstracts persistence (Redis, SQL, DynamoDB)
• Factory: Create ShortURL with builder-style optional params

Evaluation Criteria

• How does the candidate handle collisions in code generation?
• Caching strategy for hot URLs (read-heavy workload)
• Database choice justification (NoSQL for fast lookups)
• How to scale: consistent hashing for distributed storage

---

2. Rate Limiter

Requirements

• Limit requests per user/IP within a configurable time window
• Support Token Bucket and Sliding Window algorithms
• Configurable per-endpoint limits
• Distributed: consistent across multiple server instances
• Return 429 Too Many Requests with Retry-After header

Class Diagram

RateLimiter (interface)
├── allow(client_id: str) -> bool
└── implementations:
    ├── TokenBucketLimiter
    ├── SlidingWindowCounter
    └── FixedWindowCounter

TokenBucket
├── capacity: int
├── tokens: float
├── refill_rate: float (tokens/sec)
├── last_refill: timestamp
└── consume() -> bool

RateLimitConfig
├── endpoint: str
├── max_requests: int
├── window_seconds: int
└── algorithm: str

RateLimiterMiddleware
├── limiters: Map<endpoint, RateLimiter>
└── handle_request(client_id, endpoint) -> Response

Python Code

import time
import threading
from abc import ABC, abstractmethod
from collections import defaultdict

class RateLimiter(ABC):
    @abstractmethod
    def allow(self, client_id: str) -> bool:
        pass

class TokenBucket:
    def __init__(self, capacity: int, refill_rate: float):
        self.capacity = capacity
        self.tokens = capacity
        self.refill_rate = refill_rate
        self.last_refill = time.time()
        self.lock = threading.Lock()

    def consume(self) -> bool:
        with self.lock:
            self._refill()
            if self.tokens >= 1:
                self.tokens -= 1
                return True
            return False

    def _refill(self):
        now = time.time()
        elapsed = now - self.last_refill
        self.tokens = min(self.capacity, self.tokens + elapsed * self.refill_rate)
        self.last_refill = now

class TokenBucketLimiter(RateLimiter):
    def __init__(self, capacity: int, refill_rate: float):
        self.buckets = defaultdict(lambda: TokenBucket(capacity, refill_rate))

    def allow(self, client_id: str) -> bool:
        return self.buckets[client_id].consume()

class SlidingWindowCounter(RateLimiter):
    def __init__(self, max_requests: int, window_seconds: int):
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self.requests = defaultdict(list)
        self.lock = threading.Lock()

    def allow(self, client_id: str) -> bool:
        with self.lock:
            now = time.time()
            cutoff = now - self.window_seconds
            self.requests[client_id] = [
                t for t in self.requests[client_id] if t > cutoff
            ]
            if len(self.requests[client_id]) < self.max_requests:
                self.requests[client_id].append(now)
                return True
            return False

class RateLimiterMiddleware:
    def __init__(self, config: dict):
        self.limiters = {} 
        for endpoint, cfg in config.items():
            if cfg["algorithm"] == "token_bucket":
                self.limiters[endpoint] = TokenBucketLimiter(cfg["capacity"], cfg["refill_rate"])
            else:
                self.limiters[endpoint] = SlidingWindowCounter(cfg["max_requests"], cfg["window_seconds"])

    def handle_request(self, client_id: str, endpoint: str) -> dict:
        limiter = self.limiters.get(endpoint)
        if limiter and not limiter.allow(client_id):
            return {"status": 429, "headers": {"Retry-After": "60"}}
        return {"status": 200}

Design Patterns

• Strategy: Interchangeable rate limiting algorithms
• Factory: Middleware creates limiter based on config
• Decorator: Wraps request handlers transparently

Evaluation Criteria

• Thread safety for concurrent requests
• Token Bucket vs Sliding Window trade-offs (burst vs smooth)
• Distributed version with Redis (INCR + EXPIRE atomicity)
• Handling clock drift across distributed nodes

---

3. LRU Cache

Requirements

• O(1) get and put operations
• Fixed capacity with least-recently-used eviction
• Thread-safe for concurrent access
• Optional TTL (time-to-live) per entry
• Eviction callback to notify external systems

Class Diagram

LRUCache
├── capacity: int
├── cache: HashMap<key, Node>
├── head: Node (most recent)
├── tail: Node (least recent)
├── get(key) -> value
├── put(key, value, ttl?) -> void
└── _evict() -> void

Node (Doubly Linked List)
├── key: K
├── value: V
├── prev: Node
├── next: Node
└── expires_at: float | None

ThreadSafeLRUCache
├── wraps LRUCache
└── lock: threading.Lock

Python Code

import time
import threading

class Node:
    def __init__(self, key=None, value=None):
        self.key = key
        self.value = value
        self.prev = None
        self.next = None
        self.expires_at = None

class LRUCache:
    def __init__(self, capacity: int, on_evict=None):
        self.capacity = capacity
        self.cache = {} 
        self.on_evict = on_evict
        self.head = Node()  # dummy most-recent
        self.tail = Node()  # dummy least-recent
        self.head.next = self.tail
        self.tail.prev = self.head

    def get(self, key):
        if key not in self.cache:
            return -1
        node = self.cache[key]
        if node.expires_at and time.time() > node.expires_at:
            self._remove_node(node)
            del self.cache[key]
            return -1
        self._remove_node(node)
        self._add_to_front(node)
        return node.value

    def put(self, key, value, ttl_seconds=None):
        if key in self.cache:
            self._remove_node(self.cache[key])
        node = Node(key, value)
        if ttl_seconds:
            node.expires_at = time.time() + ttl_seconds
        self.cache[key] = node
        self._add_to_front(node)
        if len(self.cache) > self.capacity:
            self._evict()

    def _evict(self):
        lru_node = self.tail.prev
        self._remove_node(lru_node)
        del self.cache[lru_node.key]
        if self.on_evict:
            self.on_evict(lru_node.key, lru_node.value)

    def _add_to_front(self, node):
        node.prev = self.head
        node.next = self.head.next
        self.head.next.prev = node
        self.head.next = node

    def _remove_node(self, node):
        node.prev.next = node.next
        node.next.prev = node.prev

class ThreadSafeLRUCache:
    def __init__(self, capacity: int):
        self._cache = LRUCache(capacity)
        self._lock = threading.Lock()

    def get(self, key):
        with self._lock:
            return self._cache.get(key)

    def put(self, key, value, ttl_seconds=None):
        with self._lock:
            self._cache.put(key, value, ttl_seconds)

Design Patterns

• Proxy: ThreadSafeLRUCache wraps LRUCache adding thread safety
• Observer: on_evict callback for external notification
• Template Method: Subclass to change eviction policy (LFU, FIFO)

Evaluation Criteria

• Understanding DLL + HashMap for O(1) operations
• Correct pointer manipulation (prev/next updates)
• Thread safety: RWLock vs mutex vs lock-free structures
• TTL: lazy expiration on access + background cleaner thread

---

4. Parking Lot System

Requirements

• Multi-floor parking with slot types: Compact, Regular, Large
• Vehicle types: Motorcycle, Car, Truck
• Issue ticket on entry, calculate fee on exit
• Track available slots per floor in real time
• Pluggable pricing: hourly, flat-rate, surge pricing

Class Diagram

ParkingLot (Singleton)
├── floors: List<ParkingFloor>
├── active_tickets: Map<str, Ticket>
├── pricing: PricingStrategy
├── park_vehicle(vehicle) -> Ticket
└── unpark_vehicle(ticket_id) -> float

ParkingFloor
├── floor_number: int
├── slots: List<ParkingSlot>
└── find_available_slot(slot_type) -> ParkingSlot

ParkingSlot
├── slot_id: str
├── slot_type: SlotType (COMPACT | REGULAR | LARGE)
├── is_occupied: bool
└── vehicle: Vehicle | None

Vehicle
├── license_plate: str
└── vehicle_type: VehicleType

PricingStrategy (interface)
├── HourlyPricing
└── FlatRatePricing

Python Code

from abc import ABC, abstractmethod
from enum import Enum
from datetime import datetime
import uuid, math

class VehicleType(Enum):
    MOTORCYCLE = 1
    CAR = 2
    TRUCK = 3

class SlotType(Enum):
    COMPACT = 1
    REGULAR = 2
    LARGE = 3

VEHICLE_SLOT_MAP = {
    VehicleType.MOTORCYCLE: SlotType.COMPACT,
    VehicleType.CAR: SlotType.REGULAR,
    VehicleType.TRUCK: SlotType.LARGE,
}

class Vehicle:
    def __init__(self, plate: str, vtype: VehicleType):
        self.license_plate = plate
        self.vehicle_type = vtype

class ParkingSlot:
    def __init__(self, slot_id: str, slot_type: SlotType):
        self.slot_id = slot_id
        self.slot_type = slot_type
        self.is_occupied = False
        self.vehicle = None

    def park(self, vehicle: Vehicle):
        self.vehicle = vehicle
        self.is_occupied = True

    def unpark(self):
        self.vehicle = None
        self.is_occupied = False

class PricingStrategy(ABC):
    @abstractmethod
    def calculate_fee(self, hours: float, vtype: VehicleType) -> float:
        pass

class HourlyPricing(PricingStrategy):
    RATES = {VehicleType.MOTORCYCLE: 10, VehicleType.CAR: 20, VehicleType.TRUCK: 30}
    def calculate_fee(self, hours, vtype):
        return math.ceil(hours) * self.RATES[vtype]

class Ticket:
    def __init__(self, vehicle: Vehicle, slot: ParkingSlot):
        self.ticket_id = str(uuid.uuid4())
        self.vehicle = vehicle
        self.slot = slot
        self.entry_time = datetime.now()

class ParkingFloor:
    def __init__(self, floor_num: int, slots: list):
        self.floor_number = floor_num
        self.slots = slots

    def find_available_slot(self, slot_type: SlotType):
        for s in self.slots:
            if s.slot_type == slot_type and not s.is_occupied:
                return s
        return None

class ParkingLot:
    _instance = None
    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance

    def __init__(self):
        if not hasattr(self, '_initialized'):
            self.floors = []
            self.active_tickets = {} 
            self.pricing = HourlyPricing()
            self._initialized = True

    def park_vehicle(self, vehicle: Vehicle) -> Ticket:
        slot_type = VEHICLE_SLOT_MAP[vehicle.vehicle_type]
        for floor in self.floors:
            slot = floor.find_available_slot(slot_type)
            if slot:
                slot.park(vehicle)
                ticket = Ticket(vehicle, slot)
                self.active_tickets[ticket.ticket_id] = ticket
                return ticket
        raise Exception("No available slot")

    def unpark_vehicle(self, ticket_id: str) -> float:
        ticket = self.active_tickets.pop(ticket_id)
        ticket.slot.unpark()
        hours = (datetime.now() - ticket.entry_time).total_seconds() / 3600
        return self.pricing.calculate_fee(hours, ticket.vehicle.vehicle_type)

Design Patterns

• Singleton: ParkingLot single instance for the system
• Strategy: PricingStrategy swappable pricing algorithms
• Factory: VehicleFactory for creating vehicles from input

Evaluation Criteria

• Core entity identification and relationships
• Flexible slot assignment (extensible for EV charging slots)
• Thread safety for concurrent entry/exit
• Extensibility: new vehicle types, new pricing models

---

5. Elevator System

Requirements

• Multiple elevators in a building with N floors
• External up/down buttons per floor, internal floor-select panel
• Optimal scheduling to minimize average wait time
• States: IDLE, MOVING_UP, MOVING_DOWN, DOOR_OPEN
• Emergency stop and overweight detection

Class Diagram

ElevatorSystem
├── elevators: List<Elevator>
├── scheduler: ElevatorScheduler
└── request_elevator(floor, direction)

Elevator
├── id: int
├── current_floor: int
├── state: ElevatorState
├── up_stops: SortedList
└── down_stops: SortedList

ElevatorScheduler
├── assign(request, elevators) -> Elevator
└── _score(elevator, request) -> int

Request
├── floor: int
├── direction: Direction
└── timestamp: datetime

Python Code

from enum import Enum
from sortedcontainers import SortedList

class Direction(Enum):
    UP = 1
    DOWN = 2

class ElevatorState(Enum):
    IDLE = 1
    MOVING_UP = 2
    MOVING_DOWN = 3

class Request:
    def __init__(self, floor: int, direction: Direction):
        self.floor = floor
        self.direction = direction

class Elevator:
    def __init__(self, eid: int):
        self.id = eid
        self.current_floor = 0
        self.state = ElevatorState.IDLE
        self.up_stops = SortedList()
        self.down_stops = SortedList()

    def add_destination(self, floor: int):
        if floor > self.current_floor:
            self.up_stops.add(floor)
        elif floor < self.current_floor:
            self.down_stops.add(floor)

    def move(self):
        if self.state == ElevatorState.MOVING_UP and self.up_stops:
            self.current_floor = self.up_stops.pop(0)
        elif self.state == ElevatorState.MOVING_DOWN and self.down_stops:
            self.current_floor = self.down_stops.pop(-1)
        if not self.up_stops and not self.down_stops:
            self.state = ElevatorState.IDLE

class ElevatorScheduler:
    def assign(self, request: Request, elevators: list) -> Elevator:
        best, best_score = None, float('inf')
        for e in elevators:
            score = self._score(e, request)
            if score < best_score:
                best_score = score
                best = e
        return best

    def _score(self, elevator: Elevator, request: Request) -> int:
        dist = abs(elevator.current_floor - request.floor)
        if elevator.state == ElevatorState.IDLE:
            return dist
        if (elevator.state == ElevatorState.MOVING_UP and
            request.direction == Direction.UP and
            elevator.current_floor <= request.floor):
            return dist
        if (elevator.state == ElevatorState.MOVING_DOWN and
            request.direction == Direction.DOWN and
            elevator.current_floor >= request.floor):
            return dist
        return dist + 100

class ElevatorSystem:
    def __init__(self, num_elevators: int):
        self.elevators = [Elevator(i) for i in range(num_elevators)]
        self.scheduler = ElevatorScheduler()

    def request_elevator(self, floor: int, direction: Direction):
        req = Request(floor, direction)
        elevator = self.scheduler.assign(req, self.elevators)
        elevator.add_destination(floor)
        if elevator.state == ElevatorState.IDLE:
            elevator.state = (ElevatorState.MOVING_UP if floor > elevator.current_floor
                            else ElevatorState.MOVING_DOWN)

Design Patterns

• State: Elevator behavior governed by current state
• Strategy: Pluggable scheduling algorithms (LOOK, SCAN, zone-based)
• Observer: Floor displays subscribe to position changes

Evaluation Criteria

• Scheduling algorithm handles edge cases (all busy, same floor)
• Correct state transitions
• Extensibility: express elevators, VIP priority, weight limits
• Concurrency: multiple simultaneous requests

---

6. Pub/Sub Messaging System

Requirements

• Publishers send messages to named topics
• Subscribers register to topics and receive all messages
• Support multiple subscribers per topic
• Message ordering guarantee within a topic
• At-least-once delivery with retry on failure

Class Diagram

MessageBroker
├── topics: Map<str, Topic>
├── create_topic(name) -> Topic
├── publish(topic_name, payload)
└── subscribe(topic_name, subscriber)

Topic
├── name: str
├── subscribers: List<Subscriber>
├── messages: Deque<Message>
└── publish(message)

Message
├── id: str
├── payload: Any
├── timestamp: datetime
└── headers: dict

Subscriber (interface)
├── on_message(message) -> None
├── PrintSubscriber
└── WebhookSubscriber

Python Code

import uuid
import threading
from abc import ABC, abstractmethod
from collections import deque
from datetime import datetime

class Message:
    def __init__(self, payload, headers=None):
        self.id = str(uuid.uuid4())
        self.payload = payload
        self.headers = headers or {} 
        self.timestamp = datetime.now()

class Subscriber(ABC):
    def __init__(self, sub_id: str):
        self.id = sub_id

    @abstractmethod
    def on_message(self, message: Message):
        pass

class PrintSubscriber(Subscriber):
    def on_message(self, message: Message):
        print(f"[{self.id}] {message.payload}")

class Topic:
    def __init__(self, name: str):
        self.name = name
        self.subscribers = []
        self.messages = deque(maxlen=10000)
        self.lock = threading.Lock()

    def add_subscriber(self, subscriber: Subscriber):
        with self.lock:
            self.subscribers.append(subscriber)

    def publish(self, message: Message):
        with self.lock:
            self.messages.append(message)
            for sub in self.subscribers:
                try:
                    sub.on_message(message)
                except Exception as e:
                    print(f"Delivery failed to {sub.id}: {e}")

class MessageBroker:
    def __init__(self):
        self.topics = {} 
        self.lock = threading.Lock()

    def create_topic(self, name: str) -> Topic:
        with self.lock:
            if name not in self.topics:
                self.topics[name] = Topic(name)
            return self.topics[name]

    def publish(self, topic_name: str, payload):
        topic = self.topics.get(topic_name)
        if not topic:
            raise ValueError(f"Topic '{topic_name}' not found")
        topic.publish(Message(payload))

    def subscribe(self, topic_name: str, subscriber: Subscriber):
        topic = self.topics.get(topic_name)
        if not topic:
            raise ValueError(f"Topic '{topic_name}' not found")
        topic.add_subscriber(subscriber)

Design Patterns

• Observer: Core pub/sub pattern topic notifies subscribers
• Strategy: Delivery modes (push/pull, sync/async)
• Decorator: Add retry, logging, metrics around delivery

Evaluation Criteria

• Thread safety in publish/subscribe
• Handling slow subscribers (async delivery, backpressure)
• Dead letter queue for persistent failures
• Consumer groups for load-balanced consumption

---

7. Task Scheduler (Cron)

Requirements

• Schedule tasks to run at specified intervals or cron expressions
• Support one-time and recurring tasks
• Handle task failures with configurable retry policy
• Concurrent execution with thread pool
• Priority-based execution when tasks compete for resources

Class Diagram

TaskScheduler
├── task_queue: PriorityQueue<ScheduledTask>
├── thread_pool: ThreadPoolExecutor
├── schedule(task, trigger) -> str
├── cancel(task_id) -> bool
└── _run_loop()

ScheduledTask
├── task_id: str
├── task: Callable
├── trigger: Trigger
├── priority: int
├── next_run: datetime
├── retry_policy: RetryPolicy
└── status: TaskStatus

Trigger (interface)
├── IntervalTrigger(seconds)
├── CronTrigger(expression)
└── OneTimeTrigger(run_at)

RetryPolicy
├── max_retries: int
├── backoff_seconds: float
└── should_retry(attempt) -> bool

Python Code

import heapq
import threading
import time
import uuid
from abc import ABC, abstractmethod
from datetime import datetime, timedelta
from concurrent.futures import ThreadPoolExecutor
from enum import Enum

class TaskStatus(Enum):
    PENDING = 1
    RUNNING = 2
    COMPLETED = 3
    FAILED = 4

class Trigger(ABC):
    @abstractmethod
    def next_fire_time(self, after: datetime) -> datetime:
        pass

class IntervalTrigger(Trigger):
    def __init__(self, seconds: int):
        self.interval = timedelta(seconds=seconds)

    def next_fire_time(self, after: datetime) -> datetime:
        return after + self.interval

class OneTimeTrigger(Trigger):
    def __init__(self, run_at: datetime):
        self.run_at = run_at
        self._fired = False

    def next_fire_time(self, after: datetime) -> datetime:
        if self._fired:
            return None
        self._fired = True
        return self.run_at

class RetryPolicy:
    def __init__(self, max_retries=3, backoff=2.0):
        self.max_retries = max_retries
        self.backoff = backoff

    def should_retry(self, attempt: int) -> bool:
        return attempt < self.max_retries

    def get_delay(self, attempt: int) -> float:
        return self.backoff ** attempt

class ScheduledTask:
    def __init__(self, func, trigger: Trigger, priority=0):
        self.task_id = str(uuid.uuid4())
        self.func = func
        self.trigger = trigger
        self.priority = priority
        self.next_run = trigger.next_fire_time(datetime.now())
        self.retry_policy = RetryPolicy()
        self.status = TaskStatus.PENDING

    def __lt__(self, other):
        if self.next_run == other.next_run:
            return self.priority > other.priority
        return self.next_run < other.next_run

class TaskScheduler:
    def __init__(self, max_workers=4):
        self.queue = []
        self.pool = ThreadPoolExecutor(max_workers=max_workers)
        self.lock = threading.Lock()
        self.running = False
        self.tasks = {} 

    def schedule(self, func, trigger: Trigger, priority=0) -> str:
        task = ScheduledTask(func, trigger, priority)
        with self.lock:
            heapq.heappush(self.queue, task)
            self.tasks[task.task_id] = task
        return task.task_id

    def cancel(self, task_id: str) -> bool:
        with self.lock:
            if task_id in self.tasks:
                self.tasks[task_id].status = TaskStatus.COMPLETED
                return True
            return False

    def start(self):
        self.running = True
        threading.Thread(target=self._run_loop, daemon=True).start()

    def _run_loop(self):
        while self.running:
            with self.lock:
                if not self.queue:
                    time.sleep(0.1)
                    continue
                task = self.queue[0]
                if task.next_run > datetime.now():
                    time.sleep(0.1)
                    continue
                heapq.heappop(self.queue)
            if task.status == TaskStatus.COMPLETED:
                continue
            self.pool.submit(self._execute_task, task)

    def _execute_task(self, task: ScheduledTask):
        task.status = TaskStatus.RUNNING
        try:
            task.func()
            task.status = TaskStatus.COMPLETED
            next_time = task.trigger.next_fire_time(datetime.now())
            if next_time:
                task.next_run = next_time
                task.status = TaskStatus.PENDING
                with self.lock:
                    heapq.heappush(self.queue, task)
        except Exception:
            task.status = TaskStatus.FAILED

Design Patterns

• Strategy: Trigger implementations define scheduling behavior
• Command: Tasks encapsulate executable operations
• Observer: Notify on task completion/failure

Evaluation Criteria

• Priority queue for efficient next-task lookup
• Thread pool sizing and task isolation
• Retry with exponential backoff
• Distributed scheduler: leader election, task sharding

---

8. Snake & Ladder Game

Requirements

• Board with configurable size (default 100 squares)
• Configurable snakes (head > tail) and ladders (bottom < top)
• 2-4 players taking turns rolling a single die
• Player must land exactly on last square to win
• Track game history and moves

Class Diagram

Game
├── board: Board
├── players: List<Player>
├── dice: Dice
├── current_turn: int
├── play_turn() -> MoveResult
└── is_game_over() -> bool

Board
├── size: int
├── snakes: Map<int, int> (head -> tail)
├── ladders: Map<int, int> (bottom -> top)
└── get_final_position(pos) -> int

Player
├── name: str
├── position: int
└── move(steps) -> int

Dice
├── faces: int
└── roll() -> int

Python Code

import random
from dataclasses import dataclass

class Dice:
    def __init__(self, faces=6):
        self.faces = faces

    def roll(self) -> int:
        return random.randint(1, self.faces)

class Player:
    def __init__(self, name: str):
        self.name = name
        self.position = 0

@dataclass
class MoveResult:
    player: str
    roll: int
    old_pos: int
    new_pos: int
    hit_snake: bool = False
    hit_ladder: bool = False

class Board:
    def __init__(self, size=100, snakes=None, ladders=None):
        self.size = size
        self.snakes = snakes or {} # {head: tail}
        self.ladders = ladders or {} # {bottom: top}
        self._validate()

    def _validate(self):
        for head, tail in self.snakes.items():
            assert head > tail, "Snake head must be above tail"
        for bottom, top in self.ladders.items():
            assert bottom < top, "Ladder bottom must be below top"

    def get_final_position(self, pos: int) -> tuple:
        hit_snake, hit_ladder = False, False
        if pos in self.snakes:
            pos = self.snakes[pos]
            hit_snake = True
        elif pos in self.ladders:
            pos = self.ladders[pos]
            hit_ladder = True
        return pos, hit_snake, hit_ladder

class Game:
    def __init__(self, players: list, board: Board):
        self.board = board
        self.players = [Player(name) for name in players]
        self.dice = Dice()
        self.current_turn = 0
        self.winner = None
        self.history = []

    def play_turn(self) -> MoveResult:
        player = self.players[self.current_turn % len(self.players)]
        roll = self.dice.roll()
        old_pos = player.position
        new_pos = old_pos + roll

        if new_pos > self.board.size:
            # Must land exactly on last square
            new_pos = old_pos
        else:
            new_pos, hit_snake, hit_ladder = self.board.get_final_position(new_pos)
            player.position = new_pos

        result = MoveResult(player.name, roll, old_pos, new_pos)
        self.history.append(result)

        if new_pos == self.board.size:
            self.winner = player

        self.current_turn += 1
        return result

    def is_game_over(self) -> bool:
        return self.winner is not None

    def play(self):
        while not self.is_game_over():
            result = self.play_turn()
            print(f"{result.player} rolled {result.roll}: "
                  f"{result.old_pos} -> {result.new_pos}")
        print(f"Winner: {self.winner.name}")

Design Patterns

• Strategy: Dice can be swapped (loaded dice for testing)
• Template Method: Game loop structure fixed, steps customizable
• Builder: Board configuration with fluent API

Evaluation Criteria

• Clean separation of Board, Game logic, and Players
• Edge cases: exact landing, multiple snakes/ladders on same path
• Extensibility: power-ups, multiple dice, special squares
• Testability: injectable dice for deterministic tests

---

9. Online Bookstore

Requirements

• Search books by title, author, ISBN, category
• Shopping cart with add/remove/update quantity
• Inventory management with stock tracking
• Order placement with payment integration
• User accounts with order history

Class Diagram

BookStore
├── catalog: BookCatalog
├── inventory: InventoryManager
├── users: Map<str, User>
└── place_order(user_id, cart) -> Order

Book
├── isbn: str
├── title: str
├── author: str
├── price: float
├── category: Category
└── published_year: int

BookCatalog
├── books: List<Book>
├── search(query, field) -> List<Book>
└── get_by_isbn(isbn) -> Book

Cart
├── user_id: str
├── items: Map<str, CartItem>
├── add_item(book, qty)
├── remove_item(isbn)
└── get_total() -> float

Order
├── order_id: str
├── items: List<OrderItem>
├── total: float
├── status: OrderStatus
└── placed_at: datetime

Python Code

import uuid
from enum import Enum
from datetime import datetime
from dataclasses import dataclass, field

class Category(Enum):
    FICTION = 1
    NON_FICTION = 2
    SCIENCE = 3
    TECHNOLOGY = 4

class OrderStatus(Enum):
    PENDING = 1
    CONFIRMED = 2
    SHIPPED = 3
    DELIVERED = 4
    CANCELLED = 5

@dataclass
class Book:
    isbn: str
    title: str
    author: str
    price: float
    category: Category

@dataclass
class CartItem:
    book: Book
    quantity: int

class Cart:
    def __init__(self, user_id: str):
        self.user_id = user_id
        self.items: dict = {} 

    def add_item(self, book: Book, qty: int = 1):
        if book.isbn in self.items:
            self.items[book.isbn].quantity += qty
        else:
            self.items[book.isbn] = CartItem(book, qty)

    def remove_item(self, isbn: str):
        self.items.pop(isbn, None)

    def get_total(self) -> float:
        return sum(item.book.price * item.quantity for item in self.items.values())

    def clear(self):
        self.items.clear()

class InventoryManager:
    def __init__(self):
        self.stock = {} # isbn -> quantity

    def add_stock(self, isbn: str, qty: int):
        self.stock[isbn] = self.stock.get(isbn, 0) + qty

    def check_availability(self, isbn: str, qty: int) -> bool:
        return self.stock.get(isbn, 0) >= qty

    def reserve(self, isbn: str, qty: int) -> bool:
        if self.check_availability(isbn, qty):
            self.stock[isbn] -= qty
            return True
        return False

class Order:
    def __init__(self, user_id: str, items: list, total: float):
        self.order_id = str(uuid.uuid4())
        self.user_id = user_id
        self.items = items
        self.total = total
        self.status = OrderStatus.PENDING
        self.placed_at = datetime.now()

class BookStore:
    def __init__(self):
        self.books = {} # isbn -> Book
        self.inventory = InventoryManager()
        self.orders = []

    def add_book(self, book: Book, stock: int):
        self.books[book.isbn] = book
        self.inventory.add_stock(book.isbn, stock)

    def search(self, query: str) -> list:
        query_lower = query.lower()
        return [b for b in self.books.values()
                if query_lower in b.title.lower() or query_lower in b.author.lower()]

    def place_order(self, user_id: str, cart: Cart) -> Order:
        for isbn, item in cart.items.items():
            if not self.inventory.reserve(isbn, item.quantity):
                raise Exception(f"Insufficient stock for {isbn}")
        order = Order(user_id, list(cart.items.values()), cart.get_total())
        order.status = OrderStatus.CONFIRMED
        self.orders.append(order)
        cart.clear()
        return order

Design Patterns

• Repository: BookCatalog, InventoryManager abstract data access
• State: Order status transitions with validation
• Observer: Notify user on order status changes

Evaluation Criteria

• Inventory reservation and race condition handling
• Cart persistence (session vs database)
• Search optimization (indexing, full-text search)
• Order state machine with valid transitions only

---

10. Splitwise / Expense Tracker

Requirements

• Users create groups and add expenses
• Split types: equal, exact amounts, percentage-based
• Track balances between pairs of users
• Simplify debts (minimize number of transactions)
• Expense history and settlement tracking

Class Diagram

ExpenseService
├── users: Map<str, User>
├── groups: Map<str, Group>
├── add_expense(payer, amount, participants, split_type)
├── get_balance(user_id) -> Map<str, float>
└── simplify_debts(group_id) -> List<Transaction>

Expense
├── expense_id: str
├── payer: User
├── amount: float
├── participants: List<User>
├── split_strategy: SplitStrategy
└── created_at: datetime

SplitStrategy (interface)
├── EqualSplit
├── ExactSplit(amounts: Map)
└── PercentSplit(percentages: Map)

Group
├── group_id: str
├── name: str
├── members: List<User>
└── expenses: List<Expense>

Python Code

import uuid
from abc import ABC, abstractmethod
from collections import defaultdict

class SplitStrategy(ABC):
    @abstractmethod
    def calculate_shares(self, amount: float, participants: list) -> dict:
        """Returns {user_id: amount_owed}"""
        pass

class EqualSplit(SplitStrategy):
    def calculate_shares(self, amount, participants):
        share = round(amount / len(participants), 2)
        return {p: share for p in participants}

class ExactSplit(SplitStrategy):
    def __init__(self, exact_amounts: dict):
        self.exact_amounts = exact_amounts

    def calculate_shares(self, amount, participants):
        total = sum(self.exact_amounts.values())
        assert abs(total - amount) < 0.01, "Exact amounts must sum to total"
        return self.exact_amounts

class PercentSplit(SplitStrategy):
    def __init__(self, percentages: dict):
        self.percentages = percentages

    def calculate_shares(self, amount, participants):
        assert abs(sum(self.percentages.values()) - 100) < 0.01
        return {uid: round(amount * pct / 100, 2)
                for uid, pct in self.percentages.items()}

class Expense:
    def __init__(self, payer_id: str, amount: float, participants: list, strategy: SplitStrategy):
        self.expense_id = str(uuid.uuid4())
        self.payer_id = payer_id
        self.amount = amount
        self.participants = participants
        self.strategy = strategy
        self.shares = strategy.calculate_shares(amount, participants)

class ExpenseService:
    def __init__(self):
        self.balances = defaultdict(lambda: defaultdict(float))
        self.expenses = []

    def add_expense(self, payer_id: str, amount: float, participants: list, strategy: SplitStrategy):
        expense = Expense(payer_id, amount, participants, strategy)
        self.expenses.append(expense)
        for user_id, share in expense.shares.items():
            if user_id != payer_id:
                self.balances[user_id][payer_id] += share
                self.balances[payer_id][user_id] -= share

    def get_balance(self, user_id: str) -> dict:
        """Returns net balance with each user (positive = they owe you)."""
        return dict(self.balances.get(user_id, {} ))

    def simplify_debts(self) -> list:
        """Minimize transactions using net settlement."""
        net = defaultdict(float)
        for user, debts in self.balances.items():
            for other, amount in debts.items():
                net[user] += amount
        creditors = [(uid, amt) for uid, amt in net.items() if amt < 0]
        debtors = [(uid, amt) for uid, amt in net.items() if amt > 0]
        creditors.sort(key=lambda x: x[1])
        debtors.sort(key=lambda x: x[1], reverse=True)
        transactions = []
        i, j = 0, 0
        while i < len(debtors) and j < len(creditors):
            debtor, debt = debtors[i]
            creditor, credit = creditors[j]
            settle = min(debt, -credit)
            transactions.append((debtor, creditor, round(settle, 2)))
            debtors[i] = (debtor, debt - settle)
            creditors[j] = (creditor, credit + settle)
            if debtors[i][1] < 0.01:
                i += 1
            if creditors[j][1] > -0.01:
                j += 1
        return transactions

Design Patterns

• Strategy: SplitStrategy for different split algorithms
• Observer: Notify users when balance changes
• Repository: Expense storage abstraction

Evaluation Criteria

• Correct balance tracking between user pairs
• Debt simplification algorithm (greedy approach)
• Handling floating-point precision in money calculations
• Concurrency: simultaneous expense additions

---

11. Hotel Booking System

Requirements

• Search available rooms by date range and room type
• Room types: Standard, Deluxe, Suite with different pricing
• Reservation with check-in/check-out management
• Handle overbooking prevention with concurrency control
• Cancellation with refund policy (full/partial/none by days)

Class Diagram

Hotel
├── rooms: List<Room>
├── reservations: List<Reservation>
├── search_available(check_in, check_out, room_type) -> List<Room>
├── book_room(guest, room, dates) -> Reservation
└── cancel_reservation(reservation_id) -> Refund

Room
├── room_id: str
├── room_type: RoomType
├── floor: int
├── price_per_night: float
└── is_available(check_in, check_out) -> bool

Reservation
├── reservation_id: str
├── guest: Guest
├── room: Room
├── check_in: date
├── check_out: date
├── status: ReservationStatus
└── total_cost: float

Guest
├── guest_id: str
├── name: str
├── email: str
└── phone: str

Python Code

import uuid
import threading
from enum import Enum
from datetime import date, timedelta

class RoomType(Enum):
    STANDARD = 1
    DELUXE = 2
    SUITE = 3

class ReservationStatus(Enum):
    CONFIRMED = 1
    CHECKED_IN = 2
    CHECKED_OUT = 3
    CANCELLED = 4

class Room:
    def __init__(self, room_id: str, room_type: RoomType, price: float):
        self.room_id = room_id
        self.room_type = room_type
        self.price_per_night = price
        self.booked_dates = set()  # set of dates

    def is_available(self, check_in: date, check_out: date) -> bool:
        nights = (check_out - check_in).days
        for i in range(nights):
            if check_in + timedelta(days=i) in self.booked_dates:
                return False
        return True

    def reserve_dates(self, check_in: date, check_out: date):
        nights = (check_out - check_in).days
        for i in range(nights):
            self.booked_dates.add(check_in + timedelta(days=i))

    def release_dates(self, check_in: date, check_out: date):
        nights = (check_out - check_in).days
        for i in range(nights):
            self.booked_dates.discard(check_in + timedelta(days=i))

class Guest:
    def __init__(self, name: str, email: str):
        self.guest_id = str(uuid.uuid4())
        self.name = name
        self.email = email

class Reservation:
    def __init__(self, guest: Guest, room: Room, check_in: date, check_out: date):
        self.reservation_id = str(uuid.uuid4())
        self.guest = guest
        self.room = room
        self.check_in = check_in
        self.check_out = check_out
        self.status = ReservationStatus.CONFIRMED
        nights = (check_out - check_in).days
        self.total_cost = nights * room.price_per_night

class Hotel:
    def __init__(self, name: str):
        self.name = name
        self.rooms = []
        self.reservations = {} 
        self.lock = threading.Lock()

    def add_room(self, room: Room):
        self.rooms.append(room)

    def search_available(self, check_in: date, check_out: date, room_type: RoomType = None):
        results = []
        for room in self.rooms:
            if room_type and room.room_type != room_type:
                continue
            if room.is_available(check_in, check_out):
                results.append(room)
        return results

    def book_room(self, guest: Guest, room_id: str, check_in: date, check_out: date) -> Reservation:
        with self.lock:
            room = next((r for r in self.rooms if r.room_id == room_id), None)
            if not room:
                raise ValueError("Room not found")
            if not room.is_available(check_in, check_out):
                raise ValueError("Room not available for these dates")
            room.reserve_dates(check_in, check_out)
            reservation = Reservation(guest, room, check_in, check_out)
            self.reservations[reservation.reservation_id] = reservation
            return reservation

    def cancel_reservation(self, reservation_id: str) -> float:
        with self.lock:
            res = self.reservations.get(reservation_id)
            if not res:
                raise ValueError("Reservation not found")
            res.room.release_dates(res.check_in, res.check_out)
            res.status = ReservationStatus.CANCELLED
            days_until = (res.check_in - date.today()).days
            if days_until > 7:
                return res.total_cost  # full refund
            elif days_until > 2:
                return res.total_cost * 0.5
            return 0.0

Design Patterns

• Repository: Room and Reservation storage abstraction
• Strategy: Cancellation refund policy strategies
• State: Reservation status transitions

Evaluation Criteria

• Concurrency control for double-booking prevention
• Date range overlap detection efficiency
• Refund policy flexibility
• Scaling: how to handle multiple hotels, chains

---

12. Movie Ticket Booking (BookMyShow)

Requirements

• Browse movies, theaters, and show timings
• Seat selection with real-time availability display
• Temporary seat lock during payment (5-minute hold)
• Payment processing and ticket generation
• Concurrency: prevent double-booking of same seat

Class Diagram

BookingService
├── search_movies(city, date) -> List<Movie>
├── get_shows(movie_id, city) -> List<Show>
├── get_available_seats(show_id) -> List<Seat>
├── hold_seats(show_id, seat_ids, user_id) -> HoldToken
├── confirm_booking(hold_token, payment) -> Ticket
└── release_expired_holds()

Show
├── show_id: str
├── movie: Movie
├── theater: Theater
├── screen: Screen
├── start_time: datetime
└── seats: Map<str, SeatStatus>

Seat
├── seat_id: str
├── row: str
├── number: int
├── category: SeatCategory (REGULAR | PREMIUM | VIP)
└── price: float

SeatStatus (AVAILABLE | HELD | BOOKED)
HoldToken {token, seats, expires_at}

Python Code

import uuid
import threading
import time
from enum import Enum
from datetime import datetime, timedelta

class SeatCategory(Enum):
    REGULAR = 1
    PREMIUM = 2
    VIP = 3

class SeatStatus(Enum):
    AVAILABLE = 1
    HELD = 2
    BOOKED = 3

class Seat:
    def __init__(self, seat_id: str, row: str, number: int, category: SeatCategory, price: float):
        self.seat_id = seat_id
        self.row = row
        self.number = number
        self.category = category
        self.price = price

class Show:
    def __init__(self, show_id: str, movie_name: str, start_time: datetime):
        self.show_id = show_id
        self.movie_name = movie_name
        self.start_time = start_time
        self.seat_status = {} # seat_id -> SeatStatus
        self.seat_holds = {} # seat_id -> (user_id, expires_at)
        self.lock = threading.Lock()

    def initialize_seats(self, seats: list):
        for seat in seats:
            self.seat_status[seat.seat_id] = SeatStatus.AVAILABLE

    def get_available_seats(self) -> list:
        return [sid for sid, status in self.seat_status.items()
                if status == SeatStatus.AVAILABLE]

class HoldToken:
    def __init__(self, user_id: str, seat_ids: list, show_id: str):
        self.token = str(uuid.uuid4())
        self.user_id = user_id
        self.seat_ids = seat_ids
        self.show_id = show_id
        self.expires_at = datetime.now() + timedelta(minutes=5)

    def is_expired(self) -> bool:
        return datetime.now() > self.expires_at

class BookingService:
    def __init__(self):
        self.shows = {} # show_id -> Show
        self.holds = {} # token -> HoldToken
        self.bookings = []

    def hold_seats(self, show_id: str, seat_ids: list, user_id: str) -> HoldToken:
        show = self.shows[show_id]
        with show.lock:
            for sid in seat_ids:
                if show.seat_status.get(sid) != SeatStatus.AVAILABLE:
                    raise ValueError(f"Seat {sid} not available")
            for sid in seat_ids:
                show.seat_status[sid] = SeatStatus.HELD
                show.seat_holds[sid] = (user_id, datetime.now() + timedelta(minutes=5))
            hold = HoldToken(user_id, seat_ids, show_id)
            self.holds[hold.token] = hold
            return hold

    def confirm_booking(self, token: str) -> dict:
        hold = self.holds.pop(token, None)
        if not hold:
            raise ValueError("Invalid hold token")
        if hold.is_expired():
            self._release_hold(hold)
            raise ValueError("Hold expired")
        show = self.shows[hold.show_id]
        with show.lock:
            for sid in hold.seat_ids:
                show.seat_status[sid] = SeatStatus.BOOKED
        booking_id = str(uuid.uuid4())
        self.bookings.append({"id": booking_id, "seats": hold.seat_ids})
        return {"booking_id": booking_id, "seats": hold.seat_ids}

    def _release_hold(self, hold: HoldToken):
        show = self.shows[hold.show_id]
        with show.lock:
            for sid in hold.seat_ids:
                if show.seat_status.get(sid) == SeatStatus.HELD:
                    show.seat_status[sid] = SeatStatus.AVAILABLE

Design Patterns

• State: Seat transitions (AVAILABLE → HELD → BOOKED)
• Repository: Show and Booking data access abstraction
• Observer: Notify waitlisted users when seats become available

Evaluation Criteria

• Concurrency: lock granularity (per-show vs per-seat)
• Temporary hold mechanism with expiry
• Race condition prevention for same seat
• Scaling: distributed locks for multi-server deployment

---

13. Logger Framework

Requirements

• Multiple log levels: DEBUG, INFO, WARN, ERROR, FATAL
• Multiple output sinks: Console, File, Remote (HTTP/Syslog)
• Configurable format: timestamp, level, message, context
• Log rotation by size or time
• Thread-safe logging with minimal performance impact

Class Diagram

Logger (Singleton per name)
├── name: str
├── level: LogLevel
├── handlers: List<LogHandler>
├── formatter: LogFormatter
├── log(level, message, **context)
├── debug/info/warn/error/fatal(message)
└── add_handler(handler)

LogHandler (interface)
├── ConsoleHandler
├── FileHandler(path, max_bytes, backup_count)
└── HttpHandler(endpoint)

LogFormatter (interface)
├── SimpleFormatter: "[LEVEL] message"
├── JSONFormatter: {"ts":..., "level":..., "msg":...}
└── format(record) -> str

LogRecord
├── timestamp: datetime
├── level: LogLevel
├── message: str
├── logger_name: str
└── context: dict

Python Code

import threading
from abc import ABC, abstractmethod
from datetime import datetime
from enum import IntEnum
import json

class LogLevel(IntEnum):
    DEBUG = 10
    INFO = 20
    WARN = 30
    ERROR = 40
    FATAL = 50

class LogRecord:
    def __init__(self, level: LogLevel, message: str, logger_name: str, **context):
        self.timestamp = datetime.now()
        self.level = level
        self.message = message
        self.logger_name = logger_name
        self.context = context

class LogFormatter(ABC):
    @abstractmethod
    def format(self, record: LogRecord) -> str:
        pass

class SimpleFormatter(LogFormatter):
    def format(self, record: LogRecord) -> str:
        ts = record.timestamp.strftime("%Y-%m-%d %H:%M:%S")
        return f"[{ts}] [{record.level.name}] {record.logger_name}: {record.message}"

class JSONFormatter(LogFormatter):
    def format(self, record: LogRecord) -> str:
        return json.dumps({
            "timestamp": record.timestamp.isoformat(),
            "level": record.level.name,
            "logger": record.logger_name,
            "message": record.message,
            **record.context
        })

class LogHandler(ABC):
    def __init__(self, formatter: LogFormatter = None):
        self.formatter = formatter or SimpleFormatter()

    @abstractmethod
    def emit(self, record: LogRecord):
        pass

class ConsoleHandler(LogHandler):
    def emit(self, record: LogRecord):
        print(self.formatter.format(record))

class FileHandler(LogHandler):
    def __init__(self, filepath: str, max_bytes=10_000_000, formatter=None):
        super().__init__(formatter)
        self.filepath = filepath
        self.max_bytes = max_bytes
        self.lock = threading.Lock()

    def emit(self, record: LogRecord):
        with self.lock:
            with open(self.filepath, "a") as f:
                f.write(self.formatter.format(record) + "\n")

class Logger:
    _instances = {} 
    _lock = threading.Lock()

    @classmethod
    def get_logger(cls, name: str) -> "Logger":
        with cls._lock:
            if name not in cls._instances:
                cls._instances[name] = Logger(name)
            return cls._instances[name]

    def __init__(self, name: str):
        self.name = name
        self.level = LogLevel.DEBUG
        self.handlers = []

    def add_handler(self, handler: LogHandler):
        self.handlers.append(handler)

    def log(self, level: LogLevel, message: str, **context):
        if level < self.level:
            return
        record = LogRecord(level, message, self.name, **context)
        for handler in self.handlers:
            handler.emit(record)

    def debug(self, msg, **ctx): self.log(LogLevel.DEBUG, msg, **ctx)
    def info(self, msg, **ctx): self.log(LogLevel.INFO, msg, **ctx)
    def warn(self, msg, **ctx): self.log(LogLevel.WARN, msg, **ctx)
    def error(self, msg, **ctx): self.log(LogLevel.ERROR, msg, **ctx)
    def fatal(self, msg, **ctx): self.log(LogLevel.FATAL, msg, **ctx)

Design Patterns

• Singleton: Logger.get_logger returns shared instance per name
• Strategy: LogFormatter and LogHandler are pluggable
• Chain of Responsibility: Multiple handlers process each record
• Observer: Handlers observe log events

Evaluation Criteria

• Thread safety with minimal contention
• Separation of formatting from output
• Log rotation implementation details
• Async logging for performance (queue-based)

---

14. Notification Service

Requirements

• Support channels: Email, SMS, Push, In-App
• Template-based notification content
• User preference management (opt-in/out per channel)
• Priority levels: critical (immediate), normal (batched), low (digest)
• Retry failed deliveries with exponential backoff

Class Diagram

NotificationService
├── send(user_id, notification_type, data)
├── channels: Map<ChannelType, NotificationChannel>
├── preferences: UserPreferenceStore
└── template_engine: TemplateEngine

NotificationChannel (interface)
├── EmailChannel
├── SMSChannel
├── PushChannel
└── InAppChannel
    └── send(recipient, content) -> DeliveryResult

Notification
├── id: str
├── user_id: str
├── channel: ChannelType
├── content: str
├── priority: Priority
├── status: DeliveryStatus
└── attempts: int

Template
├── template_id: str
├── channel: ChannelType
├── body_template: str
└── render(data) -> str

Python Code

import uuid
from abc import ABC, abstractmethod
from enum import Enum
from string import Template
from collections import defaultdict

class ChannelType(Enum):
    EMAIL = 1
    SMS = 2
    PUSH = 3
    IN_APP = 4

class Priority(Enum):
    CRITICAL = 1
    NORMAL = 2
    LOW = 3

class DeliveryStatus(Enum):
    PENDING = 1
    SENT = 2
    FAILED = 3

class NotificationChannel(ABC):
    @abstractmethod
    def send(self, recipient: str, content: str) -> bool:
        pass

class EmailChannel(NotificationChannel):
    def send(self, recipient: str, content: str) -> bool:
        print(f"Email to {recipient}: {content}")
        return True

class SMSChannel(NotificationChannel):
    def send(self, recipient: str, content: str) -> bool:
        print(f"SMS to {recipient}: {content[:160]}")
        return True

class PushChannel(NotificationChannel):
    def send(self, recipient: str, content: str) -> bool:
        print(f"Push to {recipient}: {content}")
        return True

class TemplateEngine:
    def __init__(self):
        self.templates = {} 

    def register(self, name: str, template_str: str):
        self.templates[name] = Template(template_str)

    def render(self, name: str, data: dict) -> str:
        return self.templates[name].safe_substitute(data)

class UserPreferences:
    def __init__(self):
        self.prefs = defaultdict(lambda: {ch: True for ch in ChannelType})

    def is_enabled(self, user_id: str, channel: ChannelType) -> bool:
        return self.prefs[user_id].get(channel, True)

    def opt_out(self, user_id: str, channel: ChannelType):
        self.prefs[user_id][channel] = False

class Notification:
    def __init__(self, user_id: str, channel: ChannelType, content: str, priority: Priority):
        self.id = str(uuid.uuid4())
        self.user_id = user_id
        self.channel = channel
        self.content = content
        self.priority = priority
        self.status = DeliveryStatus.PENDING
        self.attempts = 0

class NotificationService:
    def __init__(self):
        self.channels = {
            ChannelType.EMAIL: EmailChannel(),
            ChannelType.SMS: SMSChannel(),
            ChannelType.PUSH: PushChannel(),
        }
        self.preferences = UserPreferences()
        self.templates = TemplateEngine()
        self.queue = []

    def send(self, user_id: str, template_name: str, data: dict,
             channels: list = None, priority: Priority = Priority.NORMAL):
        channels = channels or [ChannelType.EMAIL, ChannelType.PUSH]
        content = self.templates.render(template_name, data)
        for ch_type in channels:
            if not self.preferences.is_enabled(user_id, ch_type):
                continue
            notification = Notification(user_id, ch_type, content, priority)
            self._deliver(notification)

    def _deliver(self, notification: Notification, max_retries=3):
        channel = self.channels.get(notification.channel)
        if not channel:
            return
        for attempt in range(max_retries):
            notification.attempts = attempt + 1
            if channel.send(notification.user_id, notification.content):
                notification.status = DeliveryStatus.SENT
                return
        notification.status = DeliveryStatus.FAILED

Design Patterns

• Strategy: NotificationChannel implementations per channel type
• Template Method: Template rendering with variable substitution
• Observer: Users subscribe/unsubscribe to notification types
• Decorator: Retry logic wraps channel delivery

Evaluation Criteria

• Clean separation of channel, template, preferences
• Retry mechanism with exponential backoff
• Handling user preferences and opt-out
• Async delivery for non-critical notifications

---

15. Payment Gateway

Requirements

• Process payments via multiple providers (Stripe, PayPal, UPI)
• Support payment methods: credit card, debit card, wallet, UPI
• Transaction lifecycle: initiate → authorize → capture → settle
• Refund processing (full and partial)
• Idempotency: prevent duplicate charges on retry

Class Diagram

PaymentGateway
├── process_payment(order, method, amount) -> Transaction
├── refund(transaction_id, amount?) -> RefundResult
├── get_transaction(tx_id) -> Transaction
└── providers: Map<str, PaymentProvider>

PaymentProvider (interface)
├── StripeProvider
├── PayPalProvider
└── UPIProvider
    ├── authorize(amount, method) -> AuthResult
    ├── capture(auth_id, amount) -> CaptureResult
    └── refund(tx_id, amount) -> RefundResult

Transaction
├── transaction_id: str
├── idempotency_key: str
├── amount: float
├── currency: str
├── status: TransactionStatus
├── provider: str
└── created_at: datetime

TransactionStatus: INITIATED | AUTHORIZED | CAPTURED | SETTLED | REFUNDED | FAILED

Python Code

import uuid
import threading
from abc import ABC, abstractmethod
from enum import Enum
from datetime import datetime

class TransactionStatus(Enum):
    INITIATED = 1
    AUTHORIZED = 2
    CAPTURED = 3
    SETTLED = 4
    REFUNDED = 5
    FAILED = 6

class PaymentMethod(Enum):
    CREDIT_CARD = 1
    DEBIT_CARD = 2
    UPI = 3
    WALLET = 4

class Transaction:
    def __init__(self, amount: float, currency: str, idempotency_key: str):
        self.transaction_id = str(uuid.uuid4())
        self.idempotency_key = idempotency_key
        self.amount = amount
        self.currency = currency
        self.status = TransactionStatus.INITIATED
        self.provider = None
        self.created_at = datetime.now()

class PaymentProvider(ABC):
    @abstractmethod
    def authorize(self, amount: float, method_details: dict) -> dict:
        pass

    @abstractmethod
    def capture(self, auth_id: str, amount: float) -> dict:
        pass

    @abstractmethod
    def refund(self, transaction_id: str, amount: float) -> dict:
        pass

class StripeProvider(PaymentProvider):
    def authorize(self, amount, method_details):
        # Simulate Stripe API call
        return {"auth_id": str(uuid.uuid4()), "success": True}

    def capture(self, auth_id, amount):
        return {"capture_id": str(uuid.uuid4()), "success": True}

    def refund(self, transaction_id, amount):
        return {"refund_id": str(uuid.uuid4()), "success": True}

class PaymentGateway:
    def __init__(self):
        self.providers = {"stripe": StripeProvider()}
        self.transactions = {} 
        self.idempotency_store = {} # key -> transaction_id
        self.lock = threading.Lock()

    def process_payment(self, amount: float, currency: str, method: PaymentMethod,
                        method_details: dict, provider_name: str,
                        idempotency_key: str) -> Transaction:
        with self.lock:
            # Idempotency check
            if idempotency_key in self.idempotency_store:
                existing_tx_id = self.idempotency_store[idempotency_key]
                return self.transactions[existing_tx_id]

            tx = Transaction(amount, currency, idempotency_key)
            provider = self.providers[provider_name]
            tx.provider = provider_name

            # Authorize
            auth_result = provider.authorize(amount, method_details)
            if not auth_result["success"]:
                tx.status = TransactionStatus.FAILED
                self.transactions[tx.transaction_id] = tx
                return tx

            tx.status = TransactionStatus.AUTHORIZED

            # Capture
            capture_result = provider.capture(auth_result["auth_id"], amount)
            if capture_result["success"]:
                tx.status = TransactionStatus.CAPTURED
            else:
                tx.status = TransactionStatus.FAILED

            self.transactions[tx.transaction_id] = tx
            self.idempotency_store[idempotency_key] = tx.transaction_id
            return tx

    def refund(self, transaction_id: str, amount: float = None) -> dict:
        tx = self.transactions.get(transaction_id)
        if not tx or tx.status != TransactionStatus.CAPTURED:
            raise ValueError("Invalid transaction for refund")
        refund_amount = amount or tx.amount
        provider = self.providers[tx.provider]
        result = provider.refund(transaction_id, refund_amount)
        if result["success"]:
            tx.status = TransactionStatus.REFUNDED
        return result

Design Patterns

• Strategy: PaymentProvider implementations for each gateway
• State: Transaction status machine with valid transitions
• Factory: Select provider based on payment method/region
• Idempotency Key: Prevent duplicate processing on retries

Evaluation Criteria

• Idempotency handling for network retries
• Transaction state machine correctness
• Provider abstraction for easy addition of new gateways
• Error handling: partial captures, timeout scenarios

---

16. File System (Linux-like)

Requirements

• Hierarchical directory structure with files and folders
• Operations: create, read, write, delete, move, list
• Path resolution (absolute and relative)
• Permissions model (read/write/execute for owner/group/others)
• Support file metadata (size, created, modified timestamps)

Class Diagram

FileSystem
├── root: Directory
├── create_file(path, content) -> File
├── create_directory(path) -> Directory
├── read_file(path) -> str
├── write_file(path, content)
├── delete(path)
├── list_directory(path) -> List<FSEntry>
└── move(src, dst)

FSEntry (abstract)
├── name: str
├── parent: Directory
├── created_at: datetime
├── modified_at: datetime
├── permissions: Permissions
├── subclasses: File, Directory

File extends FSEntry
├── content: str
├── size: int
└── read() / write(content)

Directory extends FSEntry
├── children: Map<str, FSEntry>
├── add_child(entry)
├── remove_child(name)
└── get_child(name) -> FSEntry

Python Code

from abc import ABC
from datetime import datetime

class FSEntry(ABC):
    def __init__(self, name: str, parent=None):
        self.name = name
        self.parent = parent
        self.created_at = datetime.now()
        self.modified_at = datetime.now()

    def get_path(self) -> str:
        if self.parent is None:
            return "/"
        parent_path = self.parent.get_path()
        if parent_path == "/":
            return f"/{self.name}"
        return f"{parent_path}/{self.name}"

class File(FSEntry):
    def __init__(self, name: str, parent=None, content=""):
        super().__init__(name, parent)
        self.content = content

    @property
    def size(self) -> int:
        return len(self.content)

    def read(self) -> str:
        return self.content

    def write(self, content: str):
        self.content = content
        self.modified_at = datetime.now()

class Directory(FSEntry):
    def __init__(self, name: str, parent=None):
        super().__init__(name, parent)
        self.children = {} 

    def add_child(self, entry: FSEntry):
        if entry.name in self.children:
            raise FileExistsError(f"'{entry.name}' already exists")
        entry.parent = self
        self.children[entry.name] = entry

    def remove_child(self, name: str):
        if name not in self.children:
            raise FileNotFoundError(f"'{name}' not found")
        del self.children[name]

    def get_child(self, name: str) -> FSEntry:
        if name not in self.children:
            raise FileNotFoundError(f"'{name}' not found")
        return self.children[name]

    def list_children(self) -> list:
        return list(self.children.keys())

class FileSystem:
    def __init__(self):
        self.root = Directory("")

    def _resolve_path(self, path: str) -> FSEntry:
        if path == "/":
            return self.root
        parts = [p for p in path.strip("/").split("/") if p]
        current = self.root
        for part in parts:
            if not isinstance(current, Directory):
                raise NotADirectoryError(f"'{current.name}' is not a directory")
            current = current.get_child(part)
        return current

    def _resolve_parent(self, path: str):
        parts = [p for p in path.strip("/").split("/") if p]
        name = parts[-1]
        parent_path = "/" + "/".join(parts[:-1]) if len(parts) > 1 else "/"
        parent = self._resolve_path(parent_path)
        if not isinstance(parent, Directory):
            raise NotADirectoryError("Parent is not a directory")
        return parent, name

    def create_file(self, path: str, content: str = "") -> File:
        parent, name = self._resolve_parent(path)
        f = File(name, parent, content)
        parent.add_child(f)
        return f

    def create_directory(self, path: str) -> Directory:
        parent, name = self._resolve_parent(path)
        d = Directory(name, parent)
        parent.add_child(d)
        return d

    def read_file(self, path: str) -> str:
        entry = self._resolve_path(path)
        if not isinstance(entry, File):
            raise IsADirectoryError("Cannot read a directory")
        return entry.read()

    def write_file(self, path: str, content: str):
        entry = self._resolve_path(path)
        if not isinstance(entry, File):
            raise IsADirectoryError("Cannot write to a directory")
        entry.write(content)

    def delete(self, path: str):
        parent, name = self._resolve_parent(path)
        parent.remove_child(name)

    def list_directory(self, path: str) -> list:
        entry = self._resolve_path(path)
        if not isinstance(entry, Directory):
            raise NotADirectoryError("Not a directory")
        return entry.list_children()

Design Patterns

• Composite: File and Directory share FSEntry interface; Directory contains children
• Iterator: Traversal of directory tree
• Template Method: Path resolution reused across operations

Evaluation Criteria

• Composite pattern for file/directory hierarchy
• Path resolution correctness (edge cases: trailing slashes, relative paths)
• Permissions model design
• Handling symbolic links, hard links as extensions

---

17. ATM Machine

Requirements

• Authenticate user with card + PIN
• Operations: check balance, withdraw, deposit, transfer
• Dispense cash using available denominations (greedy algorithm)
• Transaction limits per day
• Handle hardware states: card reader, cash dispenser, receipt printer

Class Diagram

ATM
├── card_reader: CardReader
├── cash_dispenser: CashDispenser
├── state: ATMState
├── authenticate(card, pin) -> Account
├── withdraw(amount) -> CashBundle
├── deposit(amount)
├── check_balance() -> float
└── transfer(to_account, amount)

ATMState (State Pattern)
├── IdleState
├── CardInsertedState
├── AuthenticatedState
└── TransactionState

CashDispenser
├── denominations: Map<int, int> (value -> count)
├── dispense(amount) -> Map<int, int>
└── can_dispense(amount) -> bool

Account
├── account_id: str
├── pin_hash: str
├── balance: float
├── daily_withdrawn: float
└── daily_limit: float

Python Code

from abc import ABC, abstractmethod
from enum import Enum

class ATMStateType(Enum):
    IDLE = 1
    CARD_INSERTED = 2
    AUTHENTICATED = 3
    TRANSACTION = 4

class Account:
    def __init__(self, account_id: str, pin: str, balance: float, daily_limit=50000):
        self.account_id = account_id
        self.pin = pin
        self.balance = balance
        self.daily_limit = daily_limit
        self.daily_withdrawn = 0.0

    def can_withdraw(self, amount: float) -> bool:
        return (self.balance >= amount and
                self.daily_withdrawn + amount <= self.daily_limit)

    def debit(self, amount: float):
        self.balance -= amount
        self.daily_withdrawn += amount

    def credit(self, amount: float):
        self.balance += amount

class CashDispenser:
    def __init__(self, denominations: dict):
        # {2000: 10, 500: 20, 200: 50, 100: 100}
        self.denominations = denominations

    def can_dispense(self, amount: int) -> bool:
        remaining = amount
        for denom in sorted(self.denominations.keys(), reverse=True):
            count = min(remaining // denom, self.denominations[denom])
            remaining -= count * denom
        return remaining == 0

    def dispense(self, amount: int) -> dict:
        if not self.can_dispense(amount):
            raise ValueError("Cannot dispense exact amount")
        result = {} 
        remaining = amount
        for denom in sorted(self.denominations.keys(), reverse=True):
            count = min(remaining // denom, self.denominations[denom])
            if count > 0:
                result[denom] = count
                self.denominations[denom] -= count
                remaining -= count * denom
        return result

class ATMState(ABC):
    @abstractmethod
    def insert_card(self, atm, card_number: str): pass
    @abstractmethod
    def enter_pin(self, atm, pin: str): pass
    @abstractmethod
    def withdraw(self, atm, amount: float): pass
    @abstractmethod
    def eject_card(self, atm): pass

class IdleState(ATMState):
    def insert_card(self, atm, card_number):
        atm.current_card = card_number
        atm.set_state(CardInsertedState())

    def enter_pin(self, atm, pin):
        raise Exception("Insert card first")

    def withdraw(self, atm, amount):
        raise Exception("Insert card first")

    def eject_card(self, atm):
        raise Exception("No card inserted")

class CardInsertedState(ATMState):
    def insert_card(self, atm, card_number):
        raise Exception("Card already inserted")

    def enter_pin(self, atm, pin):
        account = atm.accounts.get(atm.current_card)
        if account and account.pin == pin:
            atm.current_account = account
            atm.set_state(AuthenticatedState())
        else:
            atm.set_state(IdleState())
            raise Exception("Invalid PIN")

    def withdraw(self, atm, amount):
        raise Exception("Enter PIN first")

    def eject_card(self, atm):
        atm.current_card = None
        atm.set_state(IdleState())

class AuthenticatedState(ATMState):
    def insert_card(self, atm, card_number):
        raise Exception("Card already inserted")

    def enter_pin(self, atm, pin):
        raise Exception("Already authenticated")

    def withdraw(self, atm, amount):
        account = atm.current_account
        if not account.can_withdraw(amount):
            raise Exception("Insufficient funds or daily limit exceeded")
        cash = atm.dispenser.dispense(int(amount))
        account.debit(amount)
        return cash

    def eject_card(self, atm):
        atm.current_card = None
        atm.current_account = None
        atm.set_state(IdleState())

class ATM:
    def __init__(self, denominations: dict):
        self.dispenser = CashDispenser(denominations)
        self.state = IdleState()
        self.accounts = {} # card_number -> Account
        self.current_card = None
        self.current_account = None

    def set_state(self, state: ATMState):
        self.state = state

    def insert_card(self, card_number: str):
        self.state.insert_card(self, card_number)

    def enter_pin(self, pin: str):
        self.state.enter_pin(self, pin)

    def withdraw(self, amount: float):
        return self.state.withdraw(self, amount)

    def eject_card(self):
        self.state.eject_card(self)

Design Patterns

• State: ATM behavior changes with current state (core pattern)
• Strategy: Cash dispensing algorithm (greedy, can swap to DP)
• Command: Each transaction type as a command object

Evaluation Criteria

• State pattern implementation with valid transitions
• Cash dispensing greedy algorithm correctness
• Handling edge cases: insufficient denominations, daily limits
• Thread safety for concurrent ATM access

---

18. Vending Machine (State Pattern)

Requirements

• Accept coins/notes and track inserted amount
• Display product catalog with prices and stock
• Dispense product when sufficient payment received
• Return change using available denominations
• Handle out-of-stock and insufficient payment gracefully

Class Diagram

VendingMachine
├── inventory: Map<str, Product>
├── state: VendingState
├── inserted_amount: float
├── insert_money(amount)
├── select_product(code) -> Product
├── dispense() -> Product
└── cancel() -> float (refund)

VendingState (interface)
├── IdleState
├── HasMoneyState
├── DispensingState
└── each defines: insert_money, select, dispense, cancel

Product
├── code: str
├── name: str
├── price: float
└── quantity: int

Python Code

from abc import ABC, abstractmethod

class Product:
    def __init__(self, code: str, name: str, price: float, quantity: int):
        self.code = code
        self.name = name
        self.price = price
        self.quantity = quantity

class VendingState(ABC):
    @abstractmethod
    def insert_money(self, machine, amount: float): pass
    @abstractmethod
    def select_product(self, machine, code: str): pass
    @abstractmethod
    def dispense(self, machine): pass
    @abstractmethod
    def cancel(self, machine) -> float: pass

class IdleState(VendingState):
    def insert_money(self, machine, amount):
        machine.inserted_amount += amount
        machine.set_state(HasMoneyState())
        print(f"Inserted: {amount}. Total: {machine.inserted_amount}")

    def select_product(self, machine, code):
        raise Exception("Insert money first")

    def dispense(self, machine):
        raise Exception("Insert money first")

    def cancel(self, machine):
        print("Nothing to cancel")
        return 0.0

class HasMoneyState(VendingState):
    def insert_money(self, machine, amount):
        machine.inserted_amount += amount
        print(f"Total inserted: {machine.inserted_amount}")

    def select_product(self, machine, code):
        product = machine.inventory.get(code)
        if not product:
            raise Exception("Invalid product code")
        if product.quantity <= 0:
            raise Exception(f"{product.name} is out of stock")
        if machine.inserted_amount < product.price:
            shortfall = product.price - machine.inserted_amount
            raise Exception(f"Insert {shortfall} more")
        machine.selected_product = product
        machine.set_state(DispensingState())

    def dispense(self, machine):
        raise Exception("Select a product first")

    def cancel(self, machine):
        refund = machine.inserted_amount
        machine.inserted_amount = 0
        machine.set_state(IdleState())
        print(f"Refunded: {refund}")
        return refund

class DispensingState(VendingState):
    def insert_money(self, machine, amount):
        raise Exception("Please wait, dispensing...")

    def select_product(self, machine, code):
        raise Exception("Already dispensing")

    def dispense(self, machine):
        product = machine.selected_product
        product.quantity -= 1
        change = machine.inserted_amount - product.price
        machine.inserted_amount = 0
        machine.selected_product = None
        machine.set_state(IdleState())
        print(f"Dispensed: {product.name}. Change: {change}")
        return product, change

    def cancel(self, machine):
        raise Exception("Cannot cancel during dispensing")

class VendingMachine:
    def __init__(self):
        self.inventory = {} 
        self.state = IdleState()
        self.inserted_amount = 0.0
        self.selected_product = None

    def set_state(self, state: VendingState):
        self.state = state

    def add_product(self, product: Product):
        self.inventory[product.code] = product

    def insert_money(self, amount: float):
        self.state.insert_money(self, amount)

    def select_product(self, code: str):
        self.state.select_product(self, code)

    def dispense(self):
        return self.state.dispense(self)

    def cancel(self) -> float:
        return self.state.cancel(self)

Design Patterns

• State: Core pattern machine behavior changes with state
• Strategy: Change-making algorithm (can be greedy or DP)
• Singleton: Machine instance per physical unit

Evaluation Criteria

• State pattern with correct transitions and guards
• Handling all edge cases per state (invalid actions)
• Change calculation with available denominations
• Concurrency: multiple users (physical constraint makes this simpler)

---

19. Chat Application (LLD)

Requirements

• One-to-one and group messaging
• Message types: text, image, file attachment
• Online/offline status and last-seen tracking
• Message delivery status: sent, delivered, read
• Message history with pagination

Class Diagram

ChatService
├── users: Map<str, User>
├── conversations: Map<str, Conversation>
├── send_message(sender_id, conv_id, content) -> Message
├── create_group(creator, name, members) -> Conversation
├── get_messages(conv_id, page, size) -> List<Message>
└── mark_as_read(user_id, conv_id, message_id)

Conversation
├── conversation_id: str
├── type: ConversationType (DIRECT | GROUP)
├── participants: List<str>
├── messages: List<Message>
└── last_message: Message

Message
├── message_id: str
├── sender_id: str
├── content: MessageContent
├── timestamp: datetime
├── status: Map<user_id, DeliveryStatus>

MessageContent (interface)
├── TextContent(text: str)
├── ImageContent(url: str, thumbnail: str)
└── FileContent(url: str, filename: str, size: int)

User
├── user_id: str
├── name: str
├── status: OnlineStatus
└── last_seen: datetime

Python Code

import uuid
from abc import ABC
from enum import Enum
from datetime import datetime
from collections import defaultdict

class OnlineStatus(Enum):
    ONLINE = 1
    OFFLINE = 2
    AWAY = 3

class DeliveryStatus(Enum):
    SENT = 1
    DELIVERED = 2
    READ = 3

class ConversationType(Enum):
    DIRECT = 1
    GROUP = 2

class MessageContent(ABC):
    pass

class TextContent(MessageContent):
    def __init__(self, text: str):
        self.text = text

class ImageContent(MessageContent):
    def __init__(self, url: str, thumbnail: str = None):
        self.url = url
        self.thumbnail = thumbnail

class User:
    def __init__(self, user_id: str, name: str):
        self.user_id = user_id
        self.name = name
        self.status = OnlineStatus.OFFLINE
        self.last_seen = datetime.now()

class Message:
    def __init__(self, sender_id: str, content: MessageContent):
        self.message_id = str(uuid.uuid4())
        self.sender_id = sender_id
        self.content = content
        self.timestamp = datetime.now()
        self.delivery_status = {} # user_id -> DeliveryStatus

class Conversation:
    def __init__(self, conv_type: ConversationType, participants: list):
        self.conversation_id = str(uuid.uuid4())
        self.type = conv_type
        self.participants = participants
        self.messages = []
        self.last_message = None

    def add_message(self, message: Message):
        self.messages.append(message)
        self.last_message = message
        for pid in self.participants:
            if pid != message.sender_id:
                message.delivery_status[pid] = DeliveryStatus.SENT

class ChatService:
    def __init__(self):
        self.users = {} 
        self.conversations = {} 
        self.user_conversations = defaultdict(list) # user_id -> [conv_ids]

    def register_user(self, user_id: str, name: str) -> User:
        user = User(user_id, name)
        self.users[user_id] = user
        return user

    def create_direct_conversation(self, user1_id: str, user2_id: str) -> Conversation:
        conv = Conversation(ConversationType.DIRECT, [user1_id, user2_id])
        self.conversations[conv.conversation_id] = conv
        self.user_conversations[user1_id].append(conv.conversation_id)
        self.user_conversations[user2_id].append(conv.conversation_id)
        return conv

    def create_group(self, creator_id: str, name: str, member_ids: list) -> Conversation:
        participants = [creator_id] + member_ids
        conv = Conversation(ConversationType.GROUP, participants)
        self.conversations[conv.conversation_id] = conv
        for uid in participants:
            self.user_conversations[uid].append(conv.conversation_id)
        return conv

    def send_message(self, sender_id: str, conv_id: str, content: MessageContent) -> Message:
        conv = self.conversations.get(conv_id)
        if not conv:
            raise ValueError("Conversation not found")
        if sender_id not in conv.participants:
            raise PermissionError("Not a participant")
        message = Message(sender_id, content)
        conv.add_message(message)
        return message

    def mark_as_read(self, user_id: str, conv_id: str, message_id: str):
        conv = self.conversations.get(conv_id)
        for msg in conv.messages:
            if msg.message_id == message_id:
                msg.delivery_status[user_id] = DeliveryStatus.READ
                break

    def get_messages(self, conv_id: str, page: int = 1, size: int = 20) -> list:
        conv = self.conversations.get(conv_id)
        if not conv:
            return []
        start = max(0, len(conv.messages) - page * size)
        end = len(conv.messages) - (page - 1) * size
        return conv.messages[start:end]

    def set_user_online(self, user_id: str):
        self.users[user_id].status = OnlineStatus.ONLINE

    def set_user_offline(self, user_id: str):
        user = self.users[user_id]
        user.status = OnlineStatus.OFFLINE
        user.last_seen = datetime.now()

Design Patterns

• Observer: Real-time message delivery to online users
• Strategy: MessageContent subtypes for different media
• Repository: Conversation and Message persistence
• Mediator: ChatService coordinates between users and conversations

Evaluation Criteria

• Message delivery guarantees (at-least-once for offline users)
• Efficient message pagination (cursor-based vs offset)
• Online status propagation to contacts
• Group message fan-out optimization

---

20. API Rate Limiter + API Gateway Design

Requirements

• Unified gateway for routing, authentication, and rate limiting
• Per-client API keys with configurable quotas
• Route-based throttling (different limits per endpoint)
• Request/response transformation and logging
• Circuit breaker for downstream service failures

Class Diagram

APIGateway
├── router: Router
├── auth: AuthMiddleware
├── rate_limiter: RateLimiterMiddleware
├── circuit_breaker: CircuitBreaker
├── handle_request(request) -> Response
└── register_route(path, backend, config)

Router
├── routes: Map<str, RouteConfig>
└── resolve(path) -> BackendService

CircuitBreaker
├── state: CircuitState (CLOSED | OPEN | HALF_OPEN)
├── failure_count: int
├── threshold: int
├── timeout: int
├── call(func) -> result
└── record_result(success: bool)

APIKey
├── key: str
├── client_id: str
├── tier: Tier (FREE | PRO | ENTERPRISE)
├── rate_limit: int (requests/minute)
└── is_active: bool

Middleware (Chain of Responsibility)
├── AuthMiddleware -> RateLimitMiddleware -> LoggingMiddleware -> Router

Python Code

import time
import threading
from enum import Enum
from abc import ABC, abstractmethod
from collections import defaultdict

class CircuitState(Enum):
    CLOSED = 1
    OPEN = 2
    HALF_OPEN = 3

class Tier(Enum):
    FREE = 1       # 100 req/min
    PRO = 2        # 1000 req/min
    ENTERPRISE = 3 # 10000 req/min

TIER_LIMITS = {Tier.FREE: 100, Tier.PRO: 1000, Tier.ENTERPRISE: 10000}

class APIKey:
    def __init__(self, key: str, client_id: str, tier: Tier):
        self.key = key
        self.client_id = client_id
        self.tier = tier
        self.rate_limit = TIER_LIMITS[tier]
        self.is_active = True

class CircuitBreaker:
    def __init__(self, threshold=5, timeout=30):
        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.threshold = threshold
        self.timeout = timeout
        self.last_failure_time = 0
        self.lock = threading.Lock()

    def can_execute(self) -> bool:
        with self.lock:
            if self.state == CircuitState.CLOSED:
                return True
            if self.state == CircuitState.OPEN:
                if time.time() - self.last_failure_time > self.timeout:
                    self.state = CircuitState.HALF_OPEN
                    return True
                return False
            return True  # HALF_OPEN allows one request

    def record_success(self):
        with self.lock:
            self.failure_count = 0
            self.state = CircuitState.CLOSED

    def record_failure(self):
        with self.lock:
            self.failure_count += 1
            self.last_failure_time = time.time()
            if self.failure_count >= self.threshold:
                self.state = CircuitState.OPEN

class RateLimiterPerKey:
    def __init__(self):
        self.windows = defaultdict(list)  # api_key -> [timestamps]
        self.lock = threading.Lock()

    def allow(self, api_key: APIKey) -> bool:
        with self.lock:
            now = time.time()
            window_start = now - 60  # 1-minute window
            key = api_key.key
            self.windows[key] = [t for t in self.windows[key] if t > window_start]
            if len(self.windows[key]) < api_key.rate_limit:
                self.windows[key].append(now)
                return True
            return False

class Middleware(ABC):
    def __init__(self, next_handler=None):
        self.next = next_handler

    @abstractmethod
    def handle(self, request: dict) -> dict:
        pass

class AuthMiddleware(Middleware):
    def __init__(self, api_keys: dict, next_handler=None):
        super().__init__(next_handler)
        self.api_keys = api_keys

    def handle(self, request: dict) -> dict:
        key = request.get("api_key")
        if not key or key not in self.api_keys:
            return {"status": 401, "body": "Unauthorized"}
        api_key = self.api_keys[key]
        if not api_key.is_active:
            return {"status": 403, "body": "API key disabled"}
        request["_api_key_obj"] = api_key
        return self.next.handle(request) if self.next else request

class RateLimitMiddleware(Middleware):
    def __init__(self, limiter: RateLimiterPerKey, next_handler=None):
        super().__init__(next_handler)
        self.limiter = limiter

    def handle(self, request: dict) -> dict:
        api_key = request.get("_api_key_obj")
        if api_key and not self.limiter.allow(api_key):
            return {"status": 429, "body": "Rate limit exceeded",
                    "headers": {"Retry-After": "60"}}
        return self.next.handle(request) if self.next else request

class APIGateway:
    def __init__(self):
        self.api_keys = {} 
        self.limiter = RateLimiterPerKey()
        self.circuit_breakers = defaultdict(CircuitBreaker)
        self.routes = {} # path -> backend_url
        self._build_chain()

    def _build_chain(self):
        self.rate_mw = RateLimitMiddleware(self.limiter)
        self.auth_mw = AuthMiddleware(self.api_keys, self.rate_mw)

    def register_key(self, api_key: APIKey):
        self.api_keys[api_key.key] = api_key
        self._build_chain()

    def register_route(self, path: str, backend: str):
        self.routes[path] = backend

    def handle_request(self, request: dict) -> dict:
        result = self.auth_mw.handle(request)
        if result.get("status") in (401, 403, 429):
            return result
        # Route to backend with circuit breaker
        path = request.get("path", "/")
        cb = self.circuit_breakers[path]
        if not cb.can_execute():
            return {"status": 503, "body": "Service unavailable (circuit open)"}
        try:
            response = {"status": 200, "body": f"Routed to {self.routes.get(path, 'default')}"} 
            cb.record_success()
            return response
        except Exception:
            cb.record_failure()
            return {"status": 502, "body": "Backend error"}

Design Patterns

• Chain of Responsibility: Middleware pipeline (Auth → Rate Limit → Route)
• Circuit Breaker: Protect against cascading downstream failures
• Strategy: Different rate limiting per API tier
• Proxy: Gateway acts as reverse proxy to backend services

Evaluation Criteria

• Middleware chain design and extensibility
• Circuit breaker state transitions (CLOSED → OPEN → HALF_OPEN)
• Per-client tiered rate limiting
• Distributed deployment: sticky sessions, shared state via Redis
• Observability: request logging, metrics, tracing

---