Designing Twitter/X: Social Media Feed at Scale

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

-- Follow relationship table
CREATE TABLE follows (
    follower_id BIGINT NOT NULL,
    followee_id BIGINT NOT NULL,
    created_at TIMESTAMP DEFAULT NOW(),
    PRIMARY KEY (follower_id, followee_id)
);
 
-- Index for efficient "who do I follow?" queries
CREATE INDEX idx_follower ON follows(follower_id);
 
-- Index for "who follows me?" queries  
CREATE INDEX idx_followee ON follows(followee_id);
 
-- Query Alice's following list
SELECT followee_id FROM follows 
WHERE follower_id = :alice_id;

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

-- Tweets table with author index
CREATE TABLE tweets (
    id BIGINT PRIMARY KEY,
    author_id BIGINT NOT NULL,
    content TEXT NOT NULL,
    created_at TIMESTAMP NOT NULL,
    -- ... other fields
);
 
-- Critical index: tweets by author, sorted by time
CREATE INDEX idx_author_time ON tweets(author_id, created_at DESC);
 
-- Query a user's recent tweets
SELECT * FROM tweets 
WHERE author_id = :user_id 
ORDER BY created_at DESC 
LIMIT 20;

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50

import heapq
from dataclasses import dataclass
from typing import List, Iterator
 
@dataclass
class Tweet:
    id: str
    author_id: str
    content: str
    created_at: datetime
 
def merge_timelines(
    tweet_sources: List[Iterator[Tweet]], 
    limit: int = 50
) -> List[Tweet]:
    """
    K-way merge of tweet streams from multiple users.
    Each source is assumed to be sorted by created_at DESC.
    
    Time complexity: O(N * log K) where:
    - K = number of sources (followed users)
    - N = limit (tweets to return)
    """
    # Min-heap entries: (negative_timestamp, source_index, tweet)
    # Negative timestamp for max-heap behavior (most recent first)
    heap = []
    
    # Initialize heap with first tweet from each source
    for idx, source in enumerate(tweet_sources):
        try:
            tweet = next(source)
            # Use negative timestamp for max-heap behavior
            heapq.heappush(heap, (-tweet.created_at.timestamp(), idx, tweet))
        except StopIteration:
            continue  # Source is empty
    
    result = []
    while heap and len(result) < limit:
        neg_ts, source_idx, tweet = heapq.heappop(heap)
        result.append(tweet)
        
        # Try to get next tweet from same source
        try:
            next_tweet = next(tweet_sources[source_idx])
            heapq.heappush(heap, 
                (-next_tweet.created_at.timestamp(), source_idx, next_tweet))
        except StopIteration:
            continue  # Source exhausted
    
    return result

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63

"""
Timeline Cache Design using Redis Sorted Sets
 
Each user's timeline is a sorted set where:
- Member = tweet_id
- Score = timestamp (for chronological ordering)
 
This allows:
- O(log N) insertion of new tweets
- O(log N + M) retrieval of M tweets from any position
- Automatic ordering by timestamp
- Efficient range queries for pagination
"""
 
import redis
from typing import List, Tuple
 
class TimelineCache:
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client
        self.MAX_TIMELINE_SIZE = 800  # Keep last 800 tweets per user
    
    def push_tweet(self, user_id: str, tweet_id: str, timestamp: float):
        """
        Add a tweet to user's pre-computed timeline.
        Called during fanout when someone they follow tweets.
        """
        key = f"timeline:{user_id}"
        
        # Add tweet with timestamp as score
        self.redis.zadd(key, {tweet_id: timestamp})
        
        # Trim to keep only most recent tweets (memory management)
        self.redis.zremrangebyrank(key, 0, -self.MAX_TIMELINE_SIZE - 1)
    
    def get_timeline(
        self, 
        user_id: str, 
        offset: int = 0, 
        limit: int = 50
    ) -> List[str]:
        """
        Retrieve user's timeline. O(log N + limit) operation.
        No merge required—timeline is pre-computed!
        """
        key = f"timeline:{user_id}"
        
        # Get tweet IDs in reverse chronological order
        tweet_ids = self.redis.zrevrange(
            key, 
            offset, 
            offset + limit - 1
        )
        
        return tweet_ids
    
    def remove_tweet(self, user_id: str, tweet_id: str):
        """
        Remove a tweet (e.g., when author deletes it).
        Must be called for ALL followers—expensive operation.
        """
        key = f"timeline:{user_id}"
        self.redis.zrem(key, tweet_id)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65

"""
Fanout Service: Distributes tweets to followers' timelines
 
Key design considerations:
1. Async processing - don't block tweet creation
2. Parallelism - fan out to many followers concurrently
3. Backpressure - handle celebrity tweets without overload
4. Idempotency - safe to retry on failure
"""
 
import asyncio
from dataclasses import dataclass
from typing import List
import aioredis
 
@dataclass
class FanoutJob:
    tweet_id: str
    author_id: str
    timestamp: float
    follower_ids: List[str]
 
class FanoutService:
    def __init__(
        self, 
        redis_pool: aioredis.Redis,
        batch_size: int = 1000,
        max_concurrent: int = 100
    ):
        self.redis = redis_pool
        self.batch_size = batch_size
        self.semaphore = asyncio.Semaphore(max_concurrent)
    
    async def fanout_tweet(self, job: FanoutJob):
        """
        Push tweet to all followers' timelines.
        Batches operations for efficiency.
        """
        # Process followers in batches  
        for i in range(0, len(job.follower_ids), self.batch_size):
            batch = job.follower_ids[i:i + self.batch_size]
            await self._process_batch(job, batch)
    
    async def _process_batch(self, job: FanoutJob, follower_ids: List[str]):
        """Process a batch of followers concurrently."""
        async with self.semaphore:
            # Use Redis pipeline for batch efficiency
            pipe = self.redis.pipeline()
            
            for follower_id in follower_ids:
                key = f"timeline:{follower_id}"
                pipe.zadd(key, {job.tweet_id: job.timestamp})
                # Trim to prevent unbounded growth
                pipe.zremrangebyrank(key, 0, -801)
            
            await pipe.execute()
    
    def estimate_fanout_time(self, follower_count: int) -> float:
        """
        Estimate time to complete fanout.
        Used for monitoring and capacity planning.
        """
        batches = (follower_count + self.batch_size - 1) // self.batch_size
        time_per_batch_ms = 10  # Empirical estimate
        return batches * time_per_batch_ms / 1000  # seconds

Celebrity Tweet Fanout Analysis
================================
 
Elon Musk's followers: 170,000,000
Tweet data to push: ~100 bytes (tweet_id + metadata)
 
Operations required:
- 170,000,000 Redis ZADD operations
- Even at 100,000 ops/second, this takes ~28 minutes
 
Storage impact:
- 170,000,000 × 100 bytes = 17 GB of cache updates
- Each tweet from Elon = 17 GB of write amplification
 
If Elon tweets 20 times per day:
- 20 × 17 GB = 340 GB of daily writes just for Elon
- 20 × 28 minutes = 9+ hours of fanout processing
 
Problem cascade:
1. Tweet appears delayed for many followers
2. Fanout queue backs up
3. Regular users' tweets also delayed
4. System-wide latency spike
 
This is unsustainable at scale.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132

"""
Hybrid Timeline Service
 
Combines push-based (for regular users) and pull-based (for celebrities)
to optimize both read and write paths.
"""
 
from dataclasses import dataclass
from typing import List, Set
import asyncio
 
@dataclass
class UserClassification:
    REGULAR_THRESHOLD = 10_000  # Users with > 10K followers are celebrities
    
    @staticmethod
    def is_celebrity(follower_count: int) -> bool:
        return follower_count > UserClassification.REGULAR_THRESHOLD
 
class HybridTimelineService:
    def __init__(
        self,
        timeline_cache,      # Redis for pre-computed timelines
        tweet_store,         # Primary tweet storage
        social_graph,        # Follow relationships
        user_service         # User metadata (follower counts)
    ):
        self.timeline_cache = timeline_cache
        self.tweet_store = tweet_store
        self.social_graph = social_graph
        self.user_service = user_service
    
    async def on_tweet_posted(self, tweet_id: str, author_id: str):
        """
        Called when a user posts a tweet.
        Decides whether to fanout based on author's follower count.
        """
        follower_count = await self.user_service.get_follower_count(author_id)
        
        if UserClassification.is_celebrity(follower_count):
            # Celebrity: store only, no fanout
            # Their tweets will be pulled at read time
            return {"fanout": False, "reason": "celebrity"}
        
        # Regular user: fanout to all followers
        followers = await self.social_graph.get_followers(author_id)
        await self._fanout_to_followers(tweet_id, followers)
        
        return {"fanout": True, "count": len(followers)}
    
    async def get_home_timeline(
        self, 
        user_id: str, 
        limit: int = 50
    ) -> List[dict]:
        """
        Assemble user's home timeline using hybrid approach.
        
        1. Get pre-computed timeline (regular users' tweets)
        2. Get list of followed celebrities
        3. Pull celebrities' recent tweets
        4. Merge and return
        """
        # Step 1: Get pre-computed timeline from cache
        cached_tweet_ids = await self.timeline_cache.get_timeline(
            user_id, 
            limit=limit * 2  # Fetch extra for merging
        )
        
        # Step 2: Identify followed celebrities
        following = await self.social_graph.get_following(user_id)
        celebrity_ids = await self._filter_celebrities(following)
        
        # Step 3: Pull recent tweets from celebrities
        celebrity_tweets = await self._fetch_celebrity_tweets(
            celebrity_ids, 
            limit=50  # Recent tweets from each celebrity
        )
        
        # Step 4: Merge cached and pulled tweets
        all_tweet_ids = set(cached_tweet_ids) | set(celebrity_tweets)
        
        # Fetch full tweet objects
        tweets = await self.tweet_store.get_tweets(list(all_tweet_ids))
        
        # Sort by timestamp and return top N
        sorted_tweets = sorted(
            tweets, 
            key=lambda t: t['created_at'], 
            reverse=True
        )
        
        return sorted_tweets[:limit]
    
    async def _filter_celebrities(
        self, 
        user_ids: List[str]
    ) -> List[str]:
        """Filter list to only celebrity accounts."""
        celebrity_ids = []
        
        # Batch lookup for efficiency
        follower_counts = await self.user_service.get_follower_counts_batch(
            user_ids
        )
        
        for user_id, count in zip(user_ids, follower_counts):
            if UserClassification.is_celebrity(count):
                celebrity_ids.append(user_id)
        
        return celebrity_ids
    
    async def _fetch_celebrity_tweets(
        self, 
        celebrity_ids: List[str],
        limit: int
    ) -> List[str]:
        """Fetch recent tweets from celebrity accounts."""
        # Fetch in parallel for speed
        tasks = [
            self.tweet_store.get_user_tweets(uid, limit=limit)
            for uid in celebrity_ids
        ]
        
        results = await asyncio.gather(*tasks)
        
        # Flatten results
        all_tweet_ids = []
        for tweets in results:
            all_tweet_ids.extend([t['id'] for t in tweets])
        
        return all_tweet_ids

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64

"""
Celebrity Tweet Cache
 
Since celebrity tweets are requested by millions of users, we cache them
aggressively. The cache key is just the celebrity's user_id.
 
This transforms N individual database queries into N cache lookups,
dramatically reducing read latency.
"""
 
class CelebrityTweetCache:
    def __init__(self, redis_client, ttl_seconds: int = 60):
        self.redis = redis_client
        self.ttl = ttl_seconds
    
    async def get_recent_tweets(
        self, 
        celebrity_id: str, 
        limit: int = 50
    ) -> List[str]:
        """
        Get celebrity's recent tweets, with caching.
        
        Cache strategy:
        - TTL of 60 seconds (tweets are near-real-time)
        - Store as sorted set for efficient range queries
        - Refresh cache on miss
        """
        cache_key = f"celebrity_tweets:{celebrity_id}"
        
        # Try cache first
        cached = await self.redis.zrevrange(cache_key, 0, limit - 1)
        
        if cached:
            return cached
        
        # Cache miss: fetch from database
        tweets = await self.tweet_store.get_user_tweets(
            celebrity_id, 
            limit=100  # Cache more than needed
        )
        
        # Populate cache
        if tweets:
            pipe = self.redis.pipeline()
            for tweet in tweets:
                pipe.zadd(cache_key, {tweet['id']: tweet['created_at']})
            pipe.expire(cache_key, self.ttl)
            await pipe.execute()
        
        return [t['id'] for t in tweets[:limit]]
    
    async def invalidate_on_new_tweet(self, celebrity_id: str, tweet: dict):
        """
        When celebrity posts, update their cache.
        Called synchronously after tweet storage.
        """
        cache_key = f"celebrity_tweets:{celebrity_id}"
        
        # Add new tweet to cache (don't wait for TTL)
        await self.redis.zadd(cache_key, {tweet['id']: tweet['created_at']})
        
        # Trim to maintain cache size
        await self.redis.zremrangebyrank(cache_key, 0, -101)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

"""
Active User Priority Fanout
 
Instead of fanning out to ALL followers equally, prioritize active users.
Inactive users' timelines can be lazily computed on their next login.
"""
 
class ActiveUserFanout:
    ACTIVE_THRESHOLD_DAYS = 7  # Users active in last 7 days
    
    async def smart_fanout(
        self, 
        tweet_id: str, 
        author_id: str,
        follower_ids: List[str]
    ):
        # Partition followers by activity
        active_followers = []
        inactive_followers = []
        
        activity_data = await self.user_service.get_last_active_batch(
            follower_ids
        )
        
        now = datetime.utcnow()
        threshold = now - timedelta(days=self.ACTIVE_THRESHOLD_DAYS)
        
        for follower_id, last_active in zip(follower_ids, activity_data):
            if last_active > threshold:
                active_followers.append(follower_id)
            else:
                inactive_followers.append(follower_id)
        
        # Fanout immediately to active users
        await self._fanout_batch(tweet_id, active_followers)
        
        # Queue inactive users for background processing
        # (or skip entirely—rebuild timeline on next login)
        if len(inactive_followers) > 10000:
            # Too many inactive: skip fanout, rebuild on login
            await self._mark_for_rebuild(inactive_followers)
        else:
            # Manageable: slow background fanout
            await self._queue_background_fanout(tweet_id, inactive_followers)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89

"""
Timeline Cache Warming
 
When a user's timeline cache is cold (empty or stale), we need to
rebuild it. This happens on first login or after extended inactivity.
"""
 
class TimelineWarmer:
    async def ensure_timeline_ready(self, user_id: str) -> bool:
        """
        Check if user's timeline cache is valid. 
        If not, trigger rebuild.
        """
        cache_key = f"timeline:{user_id}"
        cache_meta_key = f"timeline_meta:{user_id}"
        
        # Check cache freshness
        meta = await self.redis.hgetall(cache_meta_key)
        
        if not meta:
            # No timeline exists - full rebuild
            await self._full_timeline_rebuild(user_id)
            return True
        
        last_updated = datetime.fromisoformat(meta['last_updated'])
        gap_duration = datetime.utcnow() - last_updated
        
        if gap_duration > timedelta(hours=24):
            # Stale timeline - incremental update
            await self._incremental_timeline_update(
                user_id, 
                since=last_updated
            )
        
        return True
    
    async def _full_timeline_rebuild(self, user_id: str):
        """
        Rebuild timeline from scratch.
        Uses pull-based approach for all followed accounts.
        """
        following = await self.social_graph.get_following(user_id)
        
        # Fetch recent tweets from all followed accounts
        all_tweets = []
        
        for followee_id in following:
            tweets = await self.tweet_store.get_user_tweets(
                followee_id, 
                limit=20  # Recent tweets only
            )
            all_tweets.extend(tweets)
        
        # Sort and take top 800
        sorted_tweets = sorted(
            all_tweets, 
            key=lambda t: t['created_at'], 
            reverse=True
        )[:800]
        
        # Populate cache
        await self._populate_timeline_cache(user_id, sorted_tweets)
    
    async def _incremental_timeline_update(
        self, 
        user_id: str, 
        since: datetime
    ):
        """
        Update timeline with tweets since last update.
        More efficient than full rebuild for short gaps.
        """
        following = await self.social_graph.get_following(user_id)
        
        new_tweets = []
        for followee_id in following:
            tweets = await self.tweet_store.get_user_tweets_since(
                followee_id, 
                since=since
            )
            new_tweets.extend(tweets)
        
        # Merge into existing cache
        for tweet in new_tweets:
            await self.timeline_cache.push_tweet(
                user_id, 
                tweet['id'], 
                tweet['created_at']
            )