System Integration

import time
import hashlib
import threading
import heapq
from typing import Dict, List, Set, Optional, Tuple
from collections import defaultdict, deque
from dataclasses import dataclass
from concurrent.futures import ThreadPoolExecutor
import asyncio

# Scalable Rate Limiter Design
class TokenBucketRateLimiter:
    """Token bucket rate limiter for handling burst traffic"""
    
    def __init__(self, capacity: int, refill_rate: float):
        self.capacity = capacity
        self.refill_rate = refill_rate  # tokens per second
        self.buckets: Dict[str, 'TokenBucket'] = {}
        self.lock = threading.RLock()
    
    def is_allowed(self, user_id: str) -> bool:
        with self.lock:
            if user_id not in self.buckets:
                self.buckets[user_id] = TokenBucket(self.capacity, self.refill_rate)
            return self.buckets[user_id].consume()
    
    def reset(self, user_id: str) -> None:
        with self.lock:
            self.buckets.pop(user_id, None)

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()
    
    def consume(self) -> bool:
        self._refill()
        if self.tokens > 0:
            self.tokens -= 1
            return True
        return False
    
    def _refill(self) -> None:
        now = time.time()
        time_passed = now - self.last_refill
        tokens_to_add = time_passed * self.refill_rate
        
        self.tokens = min(self.capacity, self.tokens + tokens_to_add)
        self.last_refill = now

# Distributed Cache with Consistent Hashing
class ConsistentHashRing:
    """Consistent hashing for distributed cache"""
    
    def __init__(self, virtual_nodes: int = 150):
        self.virtual_nodes = virtual_nodes
        self.ring: Dict[int, str] = {}
        self.sorted_hashes: List[int] = []
    
    def add_server(self, server_id: str) -> None:
        for i in range(self.virtual_nodes):
            key = f"{server_id}:{i}"
            hash_value = self._hash(key)
            self.ring[hash_value] = server_id
        
        self.sorted_hashes = sorted(self.ring.keys())
    
    def remove_server(self, server_id: str) -> None:
        for i in range(self.virtual_nodes):
            key = f"{server_id}:{i}"
            hash_value = self._hash(key)
            if hash_value in self.ring:
                del self.ring[hash_value]
        
        self.sorted_hashes = sorted(self.ring.keys())
    
    def get_server(self, key: str) -> Optional[str]:
        if not self.sorted_hashes:
            return None
        
        hash_value = self._hash(key)
        
        # Binary search for the first server hash >= key hash
        idx = self._binary_search(hash_value)
        server_hash = self.sorted_hashes[idx % len(self.sorted_hashes)]
        
        return self.ring[server_hash]
    
    def _hash(self, key: str) -> int:
        return int(hashlib.md5(key.encode()).hexdigest(), 16)
    
    def _binary_search(self, target: int) -> int:
        left, right = 0, len(self.sorted_hashes)
        while left < right:
            mid = (left + right) // 2
            if self.sorted_hashes[mid] < target:
                left = mid + 1
            else:
                right = mid
        return left

# Scalable Search System
@dataclass
class SearchResult:
    doc_id: int
    content: str
    score: float

class SearchIndex:
    """Inverted index for scalable search"""
    
    def __init__(self):
        self.inverted_index: Dict[str, Set[int]] = defaultdict(set)
        self.documents: Dict[int, str] = {}
        self.doc_count = 0
        self.lock = threading.RLock()
    
    def add_document(self, content: str) -> int:
        with self.lock:
            doc_id = self.doc_count
            self.doc_count += 1
            self.documents[doc_id] = content
            
            tokens = self._tokenize(content)
            for token in tokens:
                self.inverted_index[token].add(doc_id)
            
            return doc_id
    
    def search(self, query: str, limit: int = 10) -> List[SearchResult]:
        query_tokens = self._tokenize(query)
        if not query_tokens:
            return []
        
        # Get candidate documents (intersection of all terms)
        candidates = None
        for token in query_tokens:
            docs = self.inverted_index.get(token, set())
            if not docs:
                return []  # No results if any token not found
            
            if candidates is None:
                candidates = docs.copy()
            else:
                candidates &= docs
        
        if not candidates:
            return []
        
        # Score and rank results
        results = []
        for doc_id in candidates:
            score = self._calculate_tf_idf(doc_id, query_tokens)
            results.append(SearchResult(
                doc_id=doc_id,
                content=self.documents[doc_id],
                score=score
            ))
        
        return sorted(results, key=lambda x: x.score, reverse=True)[:limit]
    
    def _tokenize(self, text: str) -> List[str]:
        import re
        return [token.lower() for token in re.findall(r'\b\w+\b', text)]
    
    def _calculate_tf_idf(self, doc_id: int, query_tokens: List[str]) -> float:
        doc_tokens = self._tokenize(self.documents[doc_id])
        doc_length = len(doc_tokens)
        
        score = 0.0
        for token in query_tokens:
            tf = doc_tokens.count(token) / doc_length
            df = len(self.inverted_index[token])
            idf = math.log(len(self.documents) / (df + 1))
            score += tf * idf
        
        return score

# Load Balancer with Health Checks
class Server:
    def __init__(self, server_id: str, weight: int = 1):
        self.id = server_id
        self.weight = weight
        self.healthy = True
        self.last_health_check = time.time()
    
    async def handle_request(self, request) -> dict:
        # Simulate request processing
        await asyncio.sleep(0.1)  # Simulate work
        return {"server_id": self.id, "response": "OK"}
    
    def is_healthy(self) -> bool:
        # Simulate health check
        if time.time() - self.last_health_check > 30:  # Check every 30 seconds
            self.healthy = random.random() > 0.05  # 95% healthy
            self.last_health_check = time.time()
        return self.healthy

class LoadBalancer:
    def __init__(self):
        self.servers: List[Server] = []
        self.current_index = 0
        self.lock = threading.Lock()
    
    def add_server(self, server: Server) -> None:
        with self.lock:
            self.servers.append(server)
    
    def remove_server(self, server_id: str) -> None:
        with self.lock:
            self.servers = [s for s in self.servers if s.id != server_id]
    
    def get_healthy_server(self) -> Optional[Server]:
        """Round robin with health checks"""
        healthy_servers = [s for s in self.servers if s.is_healthy()]
        if not healthy_servers:
            return None
        
        with self.lock:
            server = healthy_servers[self.current_index % len(healthy_servers)]
            self.current_index += 1
            return server
    
    def get_weighted_server(self) -> Optional[Server]:
        """Weighted round robin"""
        weighted_servers = []
        for server in self.servers:
            if server.is_healthy():
                weighted_servers.extend([server] * server.weight)
        
        if not weighted_servers:
            return None
        
        with self.lock:
            server = weighted_servers[self.current_index % len(weighted_servers)]
            self.current_index += 1
            return server

# Circuit Breaker Pattern for Fault Tolerance
class CircuitBreaker:
    def __init__(self, failure_threshold: int = 5, timeout: int = 60):
        self.failure_threshold = failure_threshold
        self.timeout = timeout
        self.failure_count = 0
        self.last_failure_time = None
        self.state = 'CLOSED'  # CLOSED, OPEN, HALF_OPEN
        self.lock = threading.Lock()
    
    def call(self, func, *args, **kwargs):
        with self.lock:
            if self.state == 'OPEN':
                if time.time() - self.last_failure_time > self.timeout:
                    self.state = 'HALF_OPEN'
                else:
                    raise Exception("Circuit breaker is OPEN")
            
            try:
                result = func(*args, **kwargs)
                self._on_success()
                return result
            except Exception as e:
                self._on_failure()
                raise e
    
    def _on_success(self):
        self.failure_count = 0
        self.state = 'CLOSED'
    
    def _on_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        
        if self.failure_count >= self.failure_threshold:
            self.state = 'OPEN'

# Database Sharding Strategy
class ShardManager:
    def __init__(self, num_shards: int):
        self.num_shards = num_shards
        self.shards = {i: f"shard_{i}" for i in range(num_shards)}
    
    def get_shard(self, key: str) -> str:
        """Hash-based sharding"""
        hash_value = int(hashlib.md5(key.encode()).hexdigest(), 16)
        shard_id = hash_value % self.num_shards
        return self.shards[shard_id]
    
    def get_range_shard(self, key: int, ranges: List[Tuple[int, int]]) -> str:
        """Range-based sharding"""
        for i, (start, end) in enumerate(ranges):
            if start <= key <= end:
                return self.shards[i]
        raise ValueError(f"Key {key} not in any range")
    
    def add_shard(self) -> None:
        """Add new shard (requires resharding)"""
        new_shard_id = self.num_shards
        self.shards[new_shard_id] = f"shard_{new_shard_id}"
        self.num_shards += 1
        # In practice, this would trigger data migration