Twitter/X.

# timeline/fanout_service.py

"""
Fan-Out Service for Twitter Timeline

Distributes tweets to follower timelines using a hybrid
push/pull approach based on follower count.
"""

from dataclasses import dataclass
from typing import List, Set
from enum import Enum
import asyncio


class FanoutStrategy(Enum):
    PUSH = "push"      # Fan-out on write
    PULL = "pull"      # Fan-out on read (celebrities)


@dataclass
class FanoutConfig:
    celebrity_threshold: int = 10_000  # Followers above this = celebrity
    batch_size: int = 1000             # Followers per batch
    max_concurrent_batches: int = 100  # Parallel batch processing
    timeline_max_size: int = 800       # Max tweets in timeline cache


@dataclass
class Tweet:
    tweet_id: str
    author_id: str
    content: str
    timestamp: int
    is_celebrity: bool = False


class FanoutService:
    """
    Hybrid fan-out service using push for regular users,
    pull for celebrities.
    
    Applies concepts:
    - Week 3: Async message processing
    - Week 4: Cache-aside pattern for timelines
    - Week 1: Partitioning by user_id
    """
    
    def __init__(
        self,
        config: FanoutConfig,
        social_graph,
        timeline_cache,
        tweet_storage,
        message_queue
    ):
        self.config = config
        self.social_graph = social_graph
        self.timeline_cache = timeline_cache
        self.tweet_storage = tweet_storage
        self.queue = message_queue
    
    async def process_tweet(self, tweet: Tweet) -> None:
        """
        Process a new tweet for fan-out.
        
        1. Store the tweet
        2. Determine fan-out strategy
        3. If push: queue fan-out jobs
        4. If pull: mark author as celebrity
        """
        # Store tweet first (source of truth)
        await self.tweet_storage.store(tweet)
        
        # Get follower count to determine strategy
        follower_count = await self.social_graph.get_follower_count(
            tweet.author_id
        )
        
        strategy = self._determine_strategy(follower_count)
        
        if strategy == FanoutStrategy.PUSH:
            # Regular user: fan out to all followers
            await self._push_fanout(tweet)
        else:
            # Celebrity: don't fan out, mark for pull
            await self._mark_celebrity_tweet(tweet)
    
    def _determine_strategy(self, follower_count: int) -> FanoutStrategy:
        """Determine push vs pull based on follower count."""
        if follower_count > self.config.celebrity_threshold:
            return FanoutStrategy.PULL
        return FanoutStrategy.PUSH
    
    async def _push_fanout(self, tweet: Tweet) -> None:
        """
        Fan out tweet to all follower timelines.
        
        Uses batching and parallelism for efficiency.
        """
        # Page through the followers (a push-path author's set
        # usually comes back whole, in one page)
        cursor = None
        while True:
            followers, cursor = await self.social_graph.get_followers(
                tweet.author_id, cursor=cursor
            )
            
            # Process in batches for efficiency
            batches = self._create_batches(
                followers, 
                self.config.batch_size
            )
            
            # Queue batch jobs (processed by workers)
            for batch in batches:
                await self.queue.enqueue({
                    "type": "fanout_batch",
                    "tweet_id": tweet.tweet_id,
                    "author_id": tweet.author_id,
                    "timestamp": tweet.timestamp,
                    "follower_ids": batch
                })
            
            if cursor is None:
                break
    
    async def process_fanout_batch(
        self, 
        tweet_id: str,
        author_id: str,
        timestamp: int,
        follower_ids: List[str]
    ) -> None:
        """
        Process a batch of follower timeline updates.
        
        Called by queue workers.
        """
        # Create timeline entry (minimal data)
        entry = {
            "tweet_id": tweet_id,
            "author_id": author_id,
            "timestamp": timestamp
        }
        
        # Update each follower's timeline cache
        tasks = [
            self._update_timeline(follower_id, entry)
            for follower_id in follower_ids
        ]
        
        await asyncio.gather(*tasks, return_exceptions=True)
    
    async def _update_timeline(
        self, 
        user_id: str, 
        entry: dict
    ) -> None:
        """
        Add tweet to user's cached timeline.
        
        Uses Redis list with LPUSH + LTRIM pattern.
        """
        timeline_key = f"timeline:{user_id}"
        
        # Push to front of list
        await self.timeline_cache.lpush(timeline_key, entry)
        
        # Trim to max size (keep only recent 800)
        await self.timeline_cache.ltrim(
            timeline_key, 
            0, 
            self.config.timeline_max_size - 1
        )
    
    async def _mark_celebrity_tweet(self, tweet: Tweet) -> None:
        """
        Mark tweet for celebrity pull-based delivery.
        
        Store in celebrity tweets index for merge at read time.
        """
        await self.timeline_cache.zadd(
            f"celebrity_tweets:{tweet.author_id}",
            {tweet.tweet_id: tweet.timestamp}
        )
        
        # Trim old celebrity tweets (keep last 100)
        await self.timeline_cache.zremrangebyrank(
            f"celebrity_tweets:{tweet.author_id}",
            0,
            -101
        )
    
    def _create_batches(
        self, 
        items: List[str], 
        batch_size: int
    ) -> List[List[str]]:
        """Split list into batches."""
        return [
            items[i:i + batch_size]
            for i in range(0, len(items), batch_size)
        ]

# timeline/timeline_service.py

"""
Timeline Service with Celebrity Merge

Constructs user's home timeline by merging pre-computed
timeline with celebrity tweets fetched at read time.
"""

from dataclasses import dataclass
from typing import List, Optional
import asyncio
import heapq


@dataclass
class TimelineEntry:
    tweet_id: str
    author_id: str
    timestamp: int


@dataclass
class FullTweet:
    tweet_id: str
    author_id: str
    author_name: str
    content: str
    timestamp: int
    likes: int
    retweets: int
    replies: int


class TimelineService:
    """
    Timeline service with hybrid push/pull architecture.
    
    Applies concepts:
    - Week 4: Cache-aside for timeline reads
    - Week 1: Read optimization through pre-computation
    - Week 3: Async parallel fetching
    """
    
    CELEBRITY_THRESHOLD = 10_000
    DEFAULT_TIMELINE_SIZE = 50
    MAX_CELEBRITIES_TO_FETCH = 100
    
    def __init__(
        self,
        timeline_cache,
        tweet_storage,
        social_graph,
        user_service
    ):
        self.timeline_cache = timeline_cache
        self.tweet_storage = tweet_storage
        self.social_graph = social_graph
        self.user_service = user_service
    
    async def get_home_timeline(
        self,
        user_id: str,
        count: int = 50,
        max_id: Optional[str] = None
    ) -> List[FullTweet]:
        """
        Get user's home timeline with celebrity merge.
        
        Merges pre-computed timeline (push) with
        celebrity tweets (pull) at read time.
        """
        # Step 1: Get pre-computed timeline from cache
        cached_timeline = await self._get_cached_timeline(
            user_id, count * 2  # Fetch extra for merge buffer
        )
        
        # Step 2: Get celebrities this user follows
        celebrities = await self._get_followed_celebrities(user_id)
        
        # Step 3: Fetch recent tweets from celebrities
        celebrity_tweets = await self._fetch_celebrity_tweets(
            celebrities
        )
        
        # Step 4: Merge and sort
        merged = self._merge_timelines(
            cached_timeline, 
            celebrity_tweets
        )
        
        # Step 5: Apply pagination
        if max_id:
            merged = [t for t in merged if t.tweet_id < max_id]
        
        # Step 6: Hydrate tweets (get full content)
        tweet_ids = [t.tweet_id for t in merged[:count]]
        full_tweets = await self._hydrate_tweets(tweet_ids)
        
        return full_tweets
    
    async def _get_cached_timeline(
        self,
        user_id: str,
        count: int
    ) -> List[TimelineEntry]:
        """Fetch pre-computed timeline from Redis."""
        timeline_key = f"timeline:{user_id}"
        
        # Check if timeline exists
        exists = await self.timeline_cache.exists(timeline_key)
        
        if not exists:
            # Rebuild timeline for inactive user
            await self._rebuild_timeline(user_id)
        
        # Fetch timeline entries
        entries = await self.timeline_cache.lrange(
            timeline_key, 0, count - 1
        )
        
        return [
            TimelineEntry(**entry) 
            for entry in entries
        ]
    
    async def _get_followed_celebrities(
        self,
        user_id: str
    ) -> List[str]:
        """Get list of celebrities the user follows."""
        # Get all users this person follows
        following = await self.social_graph.get_following(user_id)
        
        # Filter to celebrities (high follower count)
        celebrities = []
        for followed_id in following:
            follower_count = await self.social_graph.get_follower_count(
                followed_id
            )
            if follower_count > self.CELEBRITY_THRESHOLD:
                celebrities.append(followed_id)
                if len(celebrities) >= self.MAX_CELEBRITIES_TO_FETCH:
                    break
        
        return celebrities
    
    async def _fetch_celebrity_tweets(
        self,
        celebrity_ids: List[str]
    ) -> List[TimelineEntry]:
        """Fetch recent tweets from celebrities in parallel."""
        if not celebrity_ids:
            return []
        
        # Parallel fetch from all celebrities
        tasks = [
            self._get_celebrity_recent_tweets(celeb_id)
            for celeb_id in celebrity_ids
        ]
        
        results = await asyncio.gather(*tasks)
        
        # Flatten results
        all_tweets = []
        for tweets in results:
            all_tweets.extend(tweets)
        
        return all_tweets
    
    async def _get_celebrity_recent_tweets(
        self,
        celebrity_id: str,
        count: int = 20
    ) -> List[TimelineEntry]:
        """Get recent tweets from a celebrity."""
        key = f"celebrity_tweets:{celebrity_id}"
        
        # Sorted set: score = timestamp
        tweet_data = await self.timeline_cache.zrevrange(
            key, 0, count - 1, withscores=True
        )
        
        return [
            TimelineEntry(
                tweet_id=tweet_id,
                author_id=celebrity_id,
                timestamp=int(score)
            )
            for tweet_id, score in tweet_data
        ]
    
    def _merge_timelines(
        self,
        cached: List[TimelineEntry],
        celebrity: List[TimelineEntry]
    ) -> List[TimelineEntry]:
        """
        Merge two sorted timelines by timestamp.
        
        Uses heap merge for O(n log k) efficiency.
        """
        # Convert to tuples for heap: (-timestamp, entry)
        # Negative because heapq is min-heap
        all_entries = [
            (-e.timestamp, i, e) 
            for i, e in enumerate(cached + celebrity)
        ]
        
        heapq.heapify(all_entries)
        
        merged = []
        seen_ids = set()
        
        while all_entries and len(merged) < 200:
            _, _, entry = heapq.heappop(all_entries)
            
            # Deduplicate (same tweet might appear twice)
            if entry.tweet_id not in seen_ids:
                seen_ids.add(entry.tweet_id)
                merged.append(entry)
        
        return merged
    
    async def _hydrate_tweets(
        self,
        tweet_ids: List[str]
    ) -> List[FullTweet]:
        """Fetch full tweet content from storage."""
        if not tweet_ids:
            return []
        
        # Batch fetch from tweet storage
        tweets = await self.tweet_storage.mget(tweet_ids)
        
        # Get author info
        author_ids = list(set(t.author_id for t in tweets if t))
        authors = await self.user_service.mget(author_ids)
        author_map = {a.user_id: a for a in authors}
        
        return [
            FullTweet(
                tweet_id=t.tweet_id,
                author_id=t.author_id,
                author_name=author_map.get(t.author_id, {}).get("name", "Unknown"),
                content=t.content,
                timestamp=t.timestamp,
                likes=t.likes,
                retweets=t.retweets,
                replies=t.replies
            )
            for t in tweets if t
        ]
    
    async def _rebuild_timeline(self, user_id: str) -> None:
        """
        Rebuild timeline for inactive user.
        
        Called when user returns after 30+ days of inactivity
        and their timeline has been evicted from cache.
        """
        # Get users they follow
        following = await self.social_graph.get_following(user_id)
        
        # Fetch recent tweets from each (excluding celebrities)
        all_tweets = []
        for followed_id in following:
            follower_count = await self.social_graph.get_follower_count(
                followed_id
            )
            if follower_count <= self.CELEBRITY_THRESHOLD:
                tweets = await self.tweet_storage.get_user_tweets(
                    followed_id, limit=50
                )
                all_tweets.extend(tweets)
        
        # Sort by timestamp
        all_tweets.sort(key=lambda t: t.timestamp, reverse=True)
        
        # Store in cache
        timeline_key = f"timeline:{user_id}"
        for tweet in all_tweets[:800]:
            await self.timeline_cache.rpush(timeline_key, {
                "tweet_id": tweet.tweet_id,
                "author_id": tweet.author_id,
                "timestamp": tweet.timestamp
            })
        
        # Set TTL (30 days)
        await self.timeline_cache.expire(timeline_key, 30 * 86400)

# graph/social_graph_service.py

"""
Social Graph Service

Stores and queries follow relationships using a
hybrid approach: Redis for hot data, distributed
storage for full graph.
"""

from dataclasses import dataclass
from typing import List, Set, Optional, Tuple
from enum import Enum
import time


@dataclass
class FollowRelation:
    follower_id: str
    followee_id: str
    timestamp: int


class SocialGraphService:
    """
    Social graph with optimized follower/following queries.
    
    Uses Redis sets for fast membership checks and
    counts, with Manhattan/Cassandra for full graph storage.
    
    Applies concepts:
    - Week 1: Partitioning by user_id
    - Week 4: Caching hot data (follower counts)
    - Week 5: Eventual consistency for graph updates
    """
    
    def __init__(self, redis_client, graph_storage):
        self.redis = redis_client
        self.storage = graph_storage
    
    async def follow(
        self, 
        follower_id: str, 
        followee_id: str
    ) -> bool:
        """
        Create a follow relationship.
        
        Updates both directions for bidirectional queries.
        """
        # Check if already following
        if await self.is_following(follower_id, followee_id):
            return False
        
        timestamp = int(time.time())
        
        # Store in persistent storage
        await self.storage.put(
            FollowRelation(follower_id, followee_id, timestamp)
        )
        
        # Update Redis caches
        # 1. Add to follower's "following" set
        await self.redis.sadd(
            f"following:{follower_id}", 
            followee_id
        )
        
        # 2. Add to followee's "followers" set
        await self.redis.sadd(
            f"followers:{followee_id}", 
            follower_id
        )
        
        # 3. Increment follower count
        await self.redis.incr(f"follower_count:{followee_id}")
        await self.redis.incr(f"following_count:{follower_id}")
        
        return True
    
    async def unfollow(
        self, 
        follower_id: str, 
        followee_id: str
    ) -> bool:
        """Remove a follow relationship."""
        if not await self.is_following(follower_id, followee_id):
            return False
        
        # Remove from persistent storage
        await self.storage.delete(follower_id, followee_id)
        
        # Update Redis caches
        await self.redis.srem(f"following:{follower_id}", followee_id)
        await self.redis.srem(f"followers:{followee_id}", follower_id)
        await self.redis.decr(f"follower_count:{followee_id}")
        await self.redis.decr(f"following_count:{follower_id}")
        
        return True
    
    async def is_following(
        self, 
        follower_id: str, 
        followee_id: str
    ) -> bool:
        """Check if follower follows followee. O(1) operation."""
        return await self.redis.sismember(
            f"following:{follower_id}", 
            followee_id
        )
    
    async def get_followers(
        self, 
        user_id: str,
        cursor: Optional[str] = None,
        count: int = 1000
    ) -> Tuple[List[str], Optional[str]]:
        """
        Get one page of a user's followers: (follower_ids, next_cursor).
        next_cursor is None when there are no more pages.
        
        For users with many followers, uses cursor-based
        pagination to avoid loading millions of IDs.
        """
        # Small follower sets (up to the 10,000 celebrity threshold,
        # so every push-path author) come back whole, in one page
        follower_count = await self.get_follower_count(user_id)
        
        if follower_count <= 10_000:
            return list(await self.redis.smembers(
                f"followers:{user_id}"
            )), None
        
        # For large sets, use cursor-based scan
        return await self._scan_followers(
            user_id, cursor, count
        )
    
    async def get_following(
        self, 
        user_id: str
    ) -> List[str]:
        """Get all users that user_id follows."""
        return list(await self.redis.smembers(
            f"following:{user_id}"
        ))
    
    async def get_follower_count(self, user_id: str) -> int:
        """Get follower count. O(1) operation."""
        count = await self.redis.get(f"follower_count:{user_id}")
        return int(count) if count else 0
    
    async def get_following_count(self, user_id: str) -> int:
        """Get following count. O(1) operation."""
        count = await self.redis.get(f"following_count:{user_id}")
        return int(count) if count else 0
    
    async def _scan_followers(
        self,
        user_id: str,
        cursor: Optional[str],
        count: int
    ) -> Tuple[List[str], Optional[str]]:
        """Cursor-based scan for large follower sets: one page and the next cursor."""
        start_cursor = int(cursor) if cursor else 0
        
        new_cursor, followers = await self.redis.sscan(
            f"followers:{user_id}",
            cursor=start_cursor,
            count=count
        )
        
        # SSCAN returns cursor 0 when the scan is complete
        return list(followers), (str(new_cursor) if new_cursor else None)

# trending/trending_service.py

"""
Trending Topics Service

Computes trending hashtags and topics using
sliding window counts with decay.
"""

from dataclasses import dataclass
from typing import List, Dict, Tuple, Optional
from collections import defaultdict
import time
import math


@dataclass
class TrendingTopic:
    topic: str
    tweet_count: int
    velocity: float  # Rate of growth
    score: float


class TrendingService:
    """
    Real-time trending topics computation.
    
    Uses time-bucketed counts with exponential decay
    to identify rapidly growing topics.
    
    Applies concepts:
    - Week 8: Stream processing
    - Week 4: Time-windowed aggregation
    - Week 3: Async event processing
    """
    
    BUCKET_SIZE_SECONDS = 60      # 1-minute buckets
    WINDOW_SIZE_BUCKETS = 60      # Look at last 60 minutes
    DECAY_FACTOR = 0.95           # Recent tweets count more
    MIN_TWEETS_THRESHOLD = 100    # Minimum to be trending
    
    def __init__(self, redis_client, location_service):
        self.redis = redis_client
        self.location_service = location_service
    
    async def record_hashtag(
        self, 
        hashtag: str, 
        tweet_id: str,
        location: Optional[str] = None
    ) -> None:
        """
        Record a hashtag occurrence from a tweet.
        
        Updates both global and location-specific counts.
        """
        current_bucket = self._get_current_bucket()
        
        # Increment global count for this bucket
        await self.redis.hincrby(
            f"trending:global:{current_bucket}",
            hashtag.lower(),
            1
        )
        
        # Increment location-specific count
        if location:
            await self.redis.hincrby(
                f"trending:{location}:{current_bucket}",
                hashtag.lower(),
                1
            )
            await self.redis.expire(
                f"trending:{location}:{current_bucket}", 3600 * 2
            )
        
        # Set expiry on bucket (auto-cleanup)
        await self.redis.expire(
            f"trending:global:{current_bucket}",
            3600 * 2  # 2 hours
        )
    
    async def get_trending(
        self, 
        location: Optional[str] = None,
        count: int = 10
    ) -> List[TrendingTopic]:
        """
        Get current trending topics.
        
        Combines raw counts with velocity (growth rate)
        to surface rapidly emerging topics.
        """
        # Get counts from recent buckets
        topic_counts = await self._aggregate_buckets(location)
        
        # Calculate velocity and score
        scored_topics = []
        for topic, counts in topic_counts.items():
            if sum(counts) < self.MIN_TWEETS_THRESHOLD:
                continue
            
            total_count = self._weighted_count(counts)
            velocity = self._calculate_velocity(counts)
            score = self._calculate_score(total_count, velocity)
            
            scored_topics.append(TrendingTopic(
                topic=topic,
                tweet_count=sum(counts),
                velocity=velocity,
                score=score
            ))
        
        # Sort by score and return top N
        scored_topics.sort(key=lambda t: t.score, reverse=True)
        return scored_topics[:count]
    
    async def _aggregate_buckets(
        self, 
        location: Optional[str]
    ) -> Dict[str, List[int]]:
        """Aggregate counts from recent time buckets."""
        current_bucket = self._get_current_bucket()
        prefix = f"trending:{location or 'global'}"
        
        # Collect counts from each bucket
        topic_counts = defaultdict(list)
        
        for i in range(self.WINDOW_SIZE_BUCKETS):
            bucket = current_bucket - i
            bucket_data = await self.redis.hgetall(f"{prefix}:{bucket}")
            
            for topic, count in bucket_data.items():
                if topic not in topic_counts:
                    # First seen in an older bucket: zero-fill the newer
                    # ones so index 0 always means "the current minute"
                    topic_counts[topic] = [0] * i
                topic_counts[topic].append(int(count))
            
            # Pad with zeros if bucket doesn't have this topic
            for topic in topic_counts:
                if len(topic_counts[topic]) <= i:
                    topic_counts[topic].append(0)
        
        return dict(topic_counts)
    
    def _weighted_count(self, counts: List[int]) -> float:
        """
        Calculate weighted count with exponential decay.
        
        More recent buckets count more heavily.
        """
        total = 0.0
        for i, count in enumerate(counts):
            weight = self.DECAY_FACTOR ** i
            total += count * weight
        return total
    
    def _calculate_velocity(self, counts: List[int]) -> float:
        """
        Calculate velocity (rate of change).
        
        Compare recent period to earlier period.
        """
        if len(counts) < 10:
            return 0.0
        
        recent = sum(counts[:5])   # Last 5 minutes
        earlier = sum(counts[5:10]) or 1  # 5-10 minutes ago
        
        return (recent - earlier) / earlier
    
    def _calculate_score(
        self, 
        weighted_count: float, 
        velocity: float
    ) -> float:
        """
        Calculate trending score.
        
        Combines volume with velocity to surface
        both popular and rapidly growing topics.
        """
        # Log scale for volume (diminishing returns)
        volume_score = math.log10(weighted_count + 1)
        
        # Velocity boost for growing topics
        velocity_boost = max(0, velocity) * 2
        
        return volume_score * (1 + velocity_boost)
    
    def _get_current_bucket(self) -> int:
        """Get current time bucket."""
        return int(time.time()) // self.BUCKET_SIZE_SECONDS