HLD: Design Twitter - Timeline and Tweet Distribution

public class TweetService {
    public Tweet postTweet(String userId, String content, List<Media> media) {
        Tweet tweet = new Tweet(userId, content, media);
        tweetRepository.save(tweet);
        
        kafkaTemplate.send("tweet-events", new TweetCreatedEvent(tweet));
        
        return tweet;
    }
}
public class FanoutService {
    public void processTweetEvent(TweetCreatedEvent event) {
        String userId = event.getTweet().getUserId();
        int followerCount = userService.getFollowerCount(userId);
        
        if (followerCount < 10000) {
            List<String> followers = userService.getFollowers(userId);
            for (String followerId : followers) {
                timelineCache.addToTimeline(followerId, event.getTweet());
            }
        }
    }
}

public class TimelineService {
    public List<Tweet> getHomeTimeline(String userId, int limit) {
        List<Tweet> timeline = new ArrayList<>();
        
        timeline.addAll(timelineCache.getTimeline(userId, limit));
        
        List<String> celebrities = userService.getCelebritiesFollowed(userId);
        for (String celeb : celebrities) {
            timeline.addAll(tweetRepository.getRecentTweets(celeb, 20));
        }
        
        timeline.sort((a, b) -> b.getCreatedAt().compareTo(a.getCreatedAt()));
        
        return timeline.stream().limit(limit).collect(Collectors.toList());
    }
}

CREATE TABLE tweets (
    tweet_id BIGINT PRIMARY KEY,
    user_id BIGINT,
    content VARCHAR(280),
    media_urls TEXT[],
    created_at TIMESTAMP,
    like_count INT DEFAULT 0,
    retweet_count INT DEFAULT 0
);
CREATE TABLE followers (
    user_id BIGINT,
    follower_id BIGINT,
    created_at TIMESTAMP,
    PRIMARY KEY (user_id, follower_id)
);
CREATE TABLE user_timeline (
    user_id BIGINT,
    tweet_id BIGINT,
    created_at TIMESTAMP,
    PRIMARY KEY (user_id, created_at, tweet_id)
);

public class TrendingService {
    private RedisTemplate<String, Integer> redis;
    
    public void trackHashtag(String hashtag) {
        String key = "trending:" + getCurrentWindow();
        redis.opsForZSet().incrementScore(key, hashtag, 1);
        redis.expire(key, Duration.ofHours(24));
    }
    
    public List<String> getTrending(int limit) {
        String key = "trending:" + getCurrentWindow();
        return redis.opsForZSet()
            .reverseRange(key, 0, limit - 1)
            .stream()
            .map(Object::toString)
            .collect(Collectors.toList());
    }
}

public class SearchService {
    public List<Tweet> searchTweets(String query, int page, int size) {
        SearchRequest request = new SearchRequest("tweets");
        request.source(new SearchSourceBuilder()
            .query(QueryBuilders.multiMatchQuery(query, "content", "hashtags"))
            .from(page * size)
            .size(size)
            .sort("created_at", SortOrder.DESC)
        );
        
        return elasticsearchClient.search(request);
    }
}