Design a Social Media Feed (Twitter/X)

Java — Fan-out worker with celebrity threshold check

// Fan-out worker — pushes tweet ID into follower timelines
public class FanoutWorker {

    private static final int CELEBRITY_THRESHOLD = 1_000_000;
    private static final int MAX_TIMELINE_SIZE   = 800;

    private final FollowerRepository followerRepo;
    private final RedisTemplate<String, String> redis;

    public void fanOut(TweetCreatedEvent event) {
        long authorId  = event.authorId();
        long tweetId   = event.tweetId();
        long timestamp = event.createdAt().toEpochMilli();

        long followerCount = followerRepo.countFollowers(authorId);
        if (followerCount >= CELEBRITY_THRESHOLD) {
            // Skip fan-out; Timeline Service will pull-merge on read
            return;
        }

        // Paginate follower list to avoid OOM for accounts with 100K followers
        int page = 0;
        List<Long> followers;
        do {
            followers = followerRepo.findFollowers(authorId, page++, 5000);
            for (long followerId : followers) {
                String key = "timeline:" + followerId;
                // Add tweet to sorted set; score = timestamp for chronological order
                redis.opsForZSet().add(key, String.valueOf(tweetId), timestamp);
                // Trim to most recent MAX_TIMELINE_SIZE entries
                redis.opsForZSet().removeRange(key, 0, -(MAX_TIMELINE_SIZE + 1));
            }
        } while (followers.size() == 5000);
    }
}

Java — Timeline Service with celebrity pull-merge

// Timeline Service — fetch and hydrate home timeline
public List<TweetDTO> getHomeTimeline(long userId, int limit) {
    String timelineKey = "timeline:" + userId;

    // 1. Fetch tweet IDs from pre-built sorted set (most recent first)
    Set<String> tweetIdStrs = redis.opsForZSet()
        .reverseRange(timelineKey, 0, limit - 1);

    List<Long> tweetIds = tweetIdStrs.stream()
        .map(Long::parseLong)
        .collect(Collectors.toList());

    // 2. Pull celebrity tweets and merge
    List<Long> celebrityFollowees = userGraphService.getCelebrityFollowees(userId);
    for (long celeb : celebrityFollowees) {
        List<Long> celebTweets = tweetStore.getRecentByAuthor(celeb, limit);
        tweetIds.addAll(celebTweets);
    }

    // 3. Sort merged list by tweet ID (Snowflake IDs are time-ordered)
    tweetIds.sort(Comparator.reverseOrder());
    List<Long> topTweets = tweetIds.stream().limit(limit).collect(Collectors.toList());

    // 4. Batch-fetch tweet content and hydrate
    return tweetStore.batchFetch(topTweets).stream()
        .map(tweet -> hydrate(tweet))
        .collect(Collectors.toList());
}

Java — Celebrity detection with Redis-cached follower count

// Detecting celebrity accounts at timeline read time
public class UserGraphService {

    private static final int CELEBRITY_THRESHOLD = 1_000_000;
    // Cached in Redis: "celeb:{userId}" → "1" or "0"
    // Refreshed by FollowerCountUpdater consuming Kafka FollowEvent stream

    public List<Long> getCelebrityFollowees(long userId) {
        List<Long> followees = getFollowing(userId); // from graph store
        return followees.stream()
            .filter(this::isCelebrity)
            .collect(Collectors.toList());
    }

    public boolean isCelebrity(long userId) {
        String cached = redis.opsForValue().get("celeb:" + userId);
        if (cached != null) return "1".equals(cached);

        // Cache miss — query and cache for 10 minutes
        long count = followerRepo.countFollowers(userId);
        boolean celeb = count >= CELEBRITY_THRESHOLD;
        redis.opsForValue().set("celeb:" + userId, celeb ? "1" : "0",
            Duration.ofMinutes(10));
        return celeb;
    }
}

Java — Pre-signed S3 upload + Kafka-triggered transcoding

// Media upload — generate pre-signed S3 URL for direct client upload
@RestController
@RequestMapping("/api/media")
public class MediaController {

    private final AmazonS3 s3;
    private final KafkaTemplate<String, MediaUploadedEvent> kafka;

    @PostMapping("/upload-url")
    public UploadUrlResponse getUploadUrl(@RequestBody UploadRequest req) {
        String key = "uploads/" + UUID.randomUUID() + "/" + req.filename();
        Date expiry = Date.from(Instant.now().plus(15, ChronoUnit.MINUTES));

        GeneratePresignedUrlRequest urlReq = new GeneratePresignedUrlRequest(
            "media-bucket", key)
            .withMethod(HttpMethod.PUT)
            .withExpiration(expiry)
            .withContentType(req.mimeType());

        URL presignedUrl = s3.generatePresignedUrl(urlReq);
        return new UploadUrlResponse(presignedUrl.toString(), key);
    }

    // Called by client after S3 upload completes
    @PostMapping("/confirm")
    public void confirmUpload(@RequestBody ConfirmRequest req) {
        kafka.send("media-uploaded", new MediaUploadedEvent(
            req.mediaKey(), req.mimeType(), req.tweetId()));
    }
}

// Transcoding worker (simplified)
@KafkaListener(topics = "media-uploaded")
public void transcode(MediaUploadedEvent event) {
    if (event.mimeType().startsWith("video/")) {
        ffmpegService.transcodeHLS(event.mediaKey(),
            List.of("360p", "720p", "1080p"));
    } else {
        imageService.resize(event.mediaKey(),
            List.of(150, 600)); // thumbnail, medium
    }
    // Update tweet record with CDN URLs
    tweetStore.attachMedia(event.tweetId(),
        cdnBaseUrl + "/" + event.mediaKey());
}