Design a Search Autocomplete System

[Data Sources] → [Collection Pipeline] → [Processing] → [Suggestion Store]
     |                                                        ↓
[User Queries] → [Query Analytics] ─────────────────────────────┘
     |
     ↓
[Load Balancer] → [Autocomplete Service] → [Suggestion Cache] → [Response]

# Simplified popularity scoring
def calculate_popularity(query_stats):
    recent_weight = 0.7  # Last 7 days
    medium_weight = 0.2  # 7-30 days ago
    old_weight = 0.1     # 30+ days ago
    
    score = (query_stats.recent_count * recent_weight +
             query_stats.medium_count * medium_weight + 
             query_stats.old_count * old_weight)
    
    return score

public class AutocompleteTrie {
    private TrieNode root = new TrieNode();
    
    private static class TrieNode {
        Map<Character, TrieNode> children = new HashMap<>();
        List<ScoredSuggestion> topSuggestions = new ArrayList<>();
        boolean isEndOfWord = false;
    }
    
    private static class ScoredSuggestion {
        String suggestion;
        double popularity;
        long lastUpdated;
        
        ScoredSuggestion(String suggestion, double popularity) {
            this.suggestion = suggestion;
            this.popularity = popularity;
            this.lastUpdated = System.currentTimeMillis();
        }
    }
    
    // Store top-K suggestions at each node for O(1) retrieval
    public void insertWithPopularity(String word, double popularity) {
        TrieNode current = root;
        
        for (char c : word.toCharArray()) {
            current.children.putIfAbsent(c, new TrieNode());
            current = current.children.get(c);
            
            // Update top suggestions at this prefix
            current.addSuggestion(new ScoredSuggestion(word, popularity));
        }
        current.isEndOfWord = true;
    }
    
    public List<String> getSuggestions(String prefix, int limit) {
        TrieNode current = root;
        
        // Navigate to prefix node
        for (char c : prefix.toCharArray()) {
            if (!current.children.containsKey(c)) {
                return Collections.emptyList();
            }
            current = current.children.get(c);
        }
        
        // Return pre-computed top suggestions
        return current.topSuggestions.stream()
                .sorted((a, b) -> Double.compare(b.popularity, a.popularity))
                .limit(limit)
                .map(s -> s.suggestion)
                .collect(Collectors.toList());
    }
}

"jav" → ["java programming", "javascript tutorial", "java coffee beans"]
"java" → ["java programming", "java coffee beans", "java interview questions"]

class InvertedIndexAutocomplete:
    def __init__(self):
        self.prefix_to_suggestions = {}  # prefix -> sorted list of suggestions
    
    def build_index(self, query_popularity_map):
        for query, popularity in query_popularity_map.items():
            # Generate all prefixes for this query
            for i in range(1, len(query) + 1):
                prefix = query[:i].lower()
                
                if prefix not in self.prefix_to_suggestions:
                    self.prefix_to_suggestions[prefix] = []
                
                self.prefix_to_suggestions[prefix].append({
                    'suggestion': query,
                    'popularity': popularity
                })
        
        # Sort by popularity
        for prefix in self.prefix_to_suggestions:
            self.prefix_to_suggestions[prefix].sort(
                key=lambda x: x['popularity'], reverse=True
            )
    
    def get_suggestions(self, prefix, limit=10):
        prefix = prefix.lower()
        if prefix not in self.prefix_to_suggestions:
            return []
        
        return [item['suggestion'] for item in 
                self.prefix_to_suggestions[prefix][:limit]]

// Edge cache logic
function getCachedSuggestions(prefix, language, region) {
    const cacheKey = `autocomplete:${prefix}:${language}:${region}`;
    const cached = edgeCache.get(cacheKey);
    
    if (cached && !isExpired(cached.timestamp, 6 * 3600)) {
        return cached.suggestions;
    }
    
    return null; // Cache miss, fetch from origin
}

class AutocompleteCache:
    def __init__(self, redis_client):
        self.redis = redis_client
        
    def get_suggestions(self, prefix, user_context=None):
        cache_key = f"ac:{prefix.lower()}"
        
        if user_context and user_context.get('user_id'):
            # Personalized suggestions
            cache_key += f":user:{user_context['user_id']}"
        
        cached = self.redis.get(cache_key)
        if cached:
            return json.loads(cached)
        
        return None
    
    def set_suggestions(self, prefix, suggestions, user_context=None, ttl=300):
        cache_key = f"ac:{prefix.lower()}"
        
        if user_context and user_context.get('user_id'):
            cache_key += f":user:{user_context['user_id']}"
            
        self.redis.setex(cache_key, ttl, json.dumps(suggestions))

public class LocalAutocompleteCache {
    private final Cache<String, List<String>> cache;
    
    public LocalAutocompleteCache() {
        this.cache = Caffeine.newBuilder()
            .maximumSize(50_000)
            .expireAfterWrite(2, TimeUnit.MINUTES)
            .build();
    }
    
    public List<String> getSuggestions(String prefix) {
        return cache.getIfPresent(prefix.toLowerCase());
    }
    
    public void putSuggestions(String prefix, List<String> suggestions) {
        cache.put(prefix.toLowerCase(), suggestions);
    }
}

class GeoAutocompleteService:
    def __init__(self):
        self.regional_stores = {
            'us-east': AutocompleteStore('us-east'),
            'us-west': AutocompleteStore('us-west'),
            'eu-west': AutocompleteStore('eu-west'),
            'asia-pacific': AutocompleteStore('asia-pacific')
        }
    
    def get_suggestions(self, prefix, user_location, limit=10):
        # Route to regional store
        region = self.determine_region(user_location)
        store = self.regional_stores.get(region)
        
        if not store:
            # Fallback to global suggestions
            store = self.regional_stores['us-east']
        
        return store.get_suggestions(prefix, limit)
    
    def determine_region(self, user_location):
        # IP geolocation or user profile data
        if user_location.get('country') in ['US', 'CA']:
            return 'us-east'
        elif user_location.get('country') in ['GB', 'FR', 'DE']:
            return 'eu-west'
        else:
            return 'asia-pacific'

class PersonalizedAutocomplete:
    def __init__(self, global_store, user_profile_service):
        self.global_store = global_store
        self.user_service = user_profile_service
    
    def get_suggestions(self, prefix, user_id, limit=10):
        # Get global popular suggestions
        global_suggestions = self.global_store.get_suggestions(prefix, limit * 2)
        
        if not user_id:
            return global_suggestions[:limit]
        
        # Get user context
        user_profile = self.user_service.get_profile(user_id)
        user_categories = user_profile.get('interest_categories', [])
        search_history = user_profile.get('recent_searches', [])
        
        # Re-rank suggestions based on user context
        scored_suggestions = []
        for suggestion in global_suggestions:
            base_score = suggestion.get('popularity', 0)
            
            # Boost score if matches user interests
            category_boost = self.calculate_category_boost(suggestion, user_categories)
            history_boost = self.calculate_history_boost(suggestion, search_history)
            
            final_score = base_score * (1 + category_boost + history_boost)
            scored_suggestions.append({
                'suggestion': suggestion['text'],
                'score': final_score
            })
        
        # Return top suggestions
        sorted_suggestions = sorted(scored_suggestions, 
                                  key=lambda x: x['score'], reverse=True)
        return [s['suggestion'] for s in sorted_suggestions[:limit]]

import unicodedata
from difflib import SequenceMatcher

class MultiLanguageAutocomplete:
    def __init__(self):
        self.language_specific_stores = {}  # 'en', 'es', 'fr', etc.
    
    def normalize_query(self, query, language='en'):
        # Remove accents and normalize Unicode
        normalized = unicodedata.normalize('NFD', query)
        ascii_query = ''.join(c for c in normalized 
                             if unicodedata.category(c) != 'Mn')
        
        # Language-specific normalization
        if language in ['de', 'fr', 'es']:
            # Handle common character substitutions
            ascii_query = ascii_query.replace('ß', 'ss')
            ascii_query = ascii_query.replace('ñ', 'n')
        
        return ascii_query.lower().strip()
    
    def fuzzy_match_suggestions(self, prefix, suggestions, threshold=0.8):
        fuzzy_matches = []
        
        for suggestion in suggestions:
            similarity = SequenceMatcher(None, prefix, 
                                       suggestion.lower()).ratio()
            
            if similarity >= threshold:
                fuzzy_matches.append({
                    'suggestion': suggestion,
                    'similarity': similarity
                })
        
        return sorted(fuzzy_matches, key=lambda x: x['similarity'], reverse=True)

class HybridSuggestionPipeline:
    def __init__(self):
        self.batch_processor = BatchSuggestionProcessor()
        self.realtime_processor = RealtimeSuggestionProcessor()
        
    def process_query(self, query, user_context):
        # Log for batch processing
        self.batch_processor.log_query(query, user_context)
        
        # Check if this is a trending/breaking topic
        if self.is_trending_topic(query):
            # Process immediately for real-time updates
            self.realtime_processor.update_suggestions(query)
    
    def is_trending_topic(self, query):
        # Simple trending detection
        recent_count = self.get_recent_query_count(query, hours=1)
        historical_avg = self.get_historical_average(query, days=30)
        
        # If current volume is 10x normal, it's trending
        return recent_count > historical_avg * 10

class ContentFilter:
    def __init__(self):
        self.banned_words = self.load_banned_words()
        self.ml_classifier = self.load_toxicity_classifier()
        
    def filter_suggestions(self, suggestions):
        filtered = []
        
        for suggestion in suggestions:
            # Rule-based filtering
            if self.contains_banned_words(suggestion):
                continue
                
            # ML-based toxicity detection
            toxicity_score = self.ml_classifier.predict_toxicity(suggestion)
            if toxicity_score > 0.7:  # High toxicity threshold
                continue
                
            # PII detection
            if self.contains_pii(suggestion):
                continue
                
            filtered.append(suggestion)
        
        return filtered
    
    def contains_banned_words(self, suggestion):
        words = suggestion.lower().split()
        return any(word in self.banned_words for word in words)
    
    def contains_pii(self, suggestion):
        # Simple regex patterns for common PII
        import re
        
        # Phone numbers
        if re.search(r'\d{3}-\d{3}-\d{4}', suggestion):
            return True
            
        # Email addresses
        if re.search(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', suggestion):
            return True
            
        return False