Design Netflix - Video Streaming Platform

public class VideoEncodingService {
    private S3Client s3Client;
    private KafkaProducer kafkaProducer;
    private TranscodingQueue transcodingQueue;
    /**
     * Initiates the video encoding pipeline.
     * Transcodes video into multiple qualities for adaptive streaming.
     */
    public void encodeVideo(String contentId, String originalVideoUrl) {
        System.out.println("Starting encoding for content: " + contentId);
        // 1. Download original video from S3
        byte[] originalVideo = s3Client.download(originalVideoUrl);
        // 2. Define target qualities and bitrates
        EncodingProfile[] profiles = {
            new EncodingProfile("4K", 3840, 2160, 25000, "h265"),
            new EncodingProfile("1080p", 1920, 1080, 5000, "h264"),
            new EncodingProfile("720p", 1280, 720, 3000, "h264"),
            new EncodingProfile("480p", 854, 480, 1500, "h264"),
            new EncodingProfile("360p", 640, 360, 800, "h264")
        };
        // 3. Submit encoding jobs for each quality (parallel processing)
        List<CompletableFuture<EncodingResult>> encodingJobs = new ArrayList<>();
        for (EncodingProfile profile : profiles) {
            CompletableFuture<EncodingResult> job = CompletableFuture.supplyAsync(() -> {
                return transcodeVideo(originalVideo, profile, contentId);
            });
            encodingJobs.add(job);
        }
        // 4. Wait for all encoding jobs to complete
        CompletableFuture.allOf(encodingJobs.toArray(new CompletableFuture[0]))
            .thenRun(() -> {
                System.out.println("All encodings completed for: " + contentId);
                // 5. Generate DASH manifest for adaptive streaming
                generateDashManifest(contentId);
                // 6. Publish completion event
                kafkaProducer.send("video-encoding-completed",
                    new VideoReadyEvent(contentId, profiles.length));
            });
    }
    /**
     * Transcodes video to specific quality using FFmpeg.
     */
    private EncodingResult transcodeVideo(byte[] video, EncodingProfile profile,
                                          String contentId) {
        try {
            System.out.println("Encoding " + profile.quality + " for " + contentId);
            // Simulate transcoding (in reality, calls FFmpeg or cloud encoding service)
            // ffmpeg -i input.mp4 -s 1920x1080 -b:v 5000k -c:v h264 output.mp4
            String outputPath = "videos/" + contentId + "/" + profile.quality + ".mp4";
            // Upload encoded video to S3
            String s3Url = s3Client.upload(outputPath, video);
            // Save metadata to database
            VideoAsset asset = new VideoAsset(
                contentId,
                profile.quality,
                profile.bitrate,
                profile.codec,
                s3Url
            );
            saveVideoAsset(asset);
            return new EncodingResult(profile.quality, s3Url, true);
        } catch (Exception e) {
            System.err.println("Encoding failed for " + profile.quality);
            return new EncodingResult(profile.quality, null, false);
        }
    }
    /**
     * Generates DASH (Dynamic Adaptive Streaming over HTTP) manifest.
     * This manifest tells the video player which qualities are available.
     */
    private void generateDashManifest(String contentId) {
        // Fetch all encoded video assets
        List<VideoAsset> assets = getVideoAssets(contentId);
        StringBuilder manifest = new StringBuilder();
        manifest.append("<?xml version=\"1.0\" encoding=\"UTF-8\"?>\n");
        manifest.append("<MPD xmlns=\"urn:mpeg:dash:schema:mpd:2011\">\n");
        for (VideoAsset asset : assets) {
            manifest.append("  <Representation id=\"" + asset.quality + "\"\n");
            manifest.append("      bandwidth=\"" + asset.bitrate + "\"\n");
            manifest.append("      codecs=\"" + asset.codec + "\">\n");
            manifest.append("    <BaseURL>" + asset.cdnUrl + "</BaseURL>\n");
            manifest.append("  </Representation>\n");
        }
        manifest.append("</MPD>");
        // Upload manifest to S3
        String manifestUrl = "videos/" + contentId + "/manifest.mpd";
        s3Client.upload(manifestUrl, manifest.toString().getBytes());
    }
    static class EncodingProfile {
        String quality;
        int width, height, bitrate;
        String codec;
        EncodingProfile(String quality, int width, int height,
                        int bitrate, String codec) {
            this.quality = quality;
            this.width = width;
            this.height = height;
            this.bitrate = bitrate;
            this.codec = codec;
        }
    }
}

public class AdaptiveBitratePlayer {
    private String manifestUrl;
    private List<VideoQuality> availableQualities;
    private VideoQuality currentQuality;
    private NetworkMonitor networkMonitor;
    private int bufferSeconds = 30; // Target buffer size
    /**
     * Adaptive streaming player that adjusts quality based on network speed.
     */
    public AdaptiveBitratePlayer(String manifestUrl) {
        this.manifestUrl = manifestUrl;
        this.availableQualities = parseManifest(manifestUrl);
        this.networkMonitor = new NetworkMonitor();
        // Start with medium quality
        this.currentQuality = availableQualities.get(2); // 720p
    }
    /**
     * Main playback loop - continuously monitors and adjusts quality.
     */
    public void play() {
        int segmentNumber = 0;
        while (true) {
            // 1. Measure current network bandwidth
            double bandwidthMbps = networkMonitor.measureBandwidth();
            // 2. Check current buffer level
            int currentBuffer = getBufferLevel();
            // 3. Decide if we need to switch quality
            VideoQuality newQuality = selectOptimalQuality(
                bandwidthMbps,
                currentBuffer
            );
            if (!newQuality.equals(currentQuality)) {
                System.out.println("Switching quality: " +
                    currentQuality.resolution + " -> " + newQuality.resolution);
                currentQuality = newQuality;
            }
            // 4. Fetch next video segment at current quality
            String segmentUrl = currentQuality.baseUrl + "/segment_" +
                segmentNumber + ".m4s";
            byte[] segment = downloadSegment(segmentUrl);
            // 5. Add to playback buffer
            addToBuffer(segment);
            segmentNumber++;
            // Wait for next segment (typically 2-10 seconds)
            sleep(2000);
        }
    }
    /**
     * Selects optimal quality based on bandwidth and buffer health.
     */
    private VideoQuality selectOptimalQuality(double bandwidthMbps,
                                               int currentBuffer) {
        // Safety margin: use 80% of measured bandwidth
        double usableBandwidth = bandwidthMbps * 0.8;
        // Buffer-based adjustment
        if (currentBuffer < 10) {
            // Buffer running low - prioritize stability over quality
            usableBandwidth *= 0.6; // Be conservative
        } else if (currentBuffer > 40) {
            // Buffer is healthy - can try higher quality
            usableBandwidth *= 1.2;
        }
        // Find highest quality that fits within bandwidth
        VideoQuality selected = availableQualities.get(0); // Start with lowest
        for (VideoQuality quality : availableQualities) {
            if (quality.bitrateMbps <= usableBandwidth) {
                selected = quality;
            } else {
                break; // Qualities are sorted by bitrate
            }
        }
        return selected;
    }
    /**
     * Parses DASH manifest to extract available qualities.
     */
    private List<VideoQuality> parseManifest(String manifestUrl) {
        // Parse MPD file to extract representations
        // Simplified for demonstration
        return Arrays.asList(
            new VideoQuality("360p", 0.8, "https://cdn.netflix.com/360p"),
            new VideoQuality("480p", 1.5, "https://cdn.netflix.com/480p"),
            new VideoQuality("720p", 3.0, "https://cdn.netflix.com/720p"),
            new VideoQuality("1080p", 5.0, "https://cdn.netflix.com/1080p"),
            new VideoQuality("4K", 25.0, "https://cdn.netflix.com/4k")
        );
    }
    static class VideoQuality {
        String resolution;
        double bitrateMbps;
        String baseUrl;
        VideoQuality(String resolution, double bitrateMbps, String baseUrl) {
            this.resolution = resolution;
            this.bitrateMbps = bitrateMbps;
            this.baseUrl = baseUrl;
        }
        @Override
        public boolean equals(Object obj) {
            if (obj instanceof VideoQuality) {
                return this.resolution.equals(((VideoQuality) obj).resolution);
            }
            return false;
        }
    }
    static class NetworkMonitor {
        /**
         * Measures current network bandwidth by tracking download speeds.
         */
        public double measureBandwidth() {
            // In reality: measure actual download speed of recent segments
            // For demo, simulate varying bandwidth
            return 3.0 + Math.random() * 5.0; // 3-8 Mbps
        }
    }
}

public class RecommendationService {
    private CassandraClient cassandra;
    private RedisCache cache;
    private MLModel collaborativeFilteringModel;
    /**
     * Generates personalized recommendations for a user.
     * Uses collaborative filtering and content-based filtering.
     */
    public List<Recommendation> getRecommendations(long userId, int limit) {
        // 1. Check cache first
        String cacheKey = "recommendations:" + userId;
        List<Recommendation> cached = cache.get(cacheKey);
        if (cached != null) {
            return cached.subList(0, Math.min(limit, cached.size()));
        }
        // 2. Fetch user's viewing history
        List<ViewingHistory> history = getViewingHistory(userId);
        // 3. Collaborative filtering: find similar users
        List<Long> similarUsers = findSimilarUsers(userId, history);
        // 4. Get what similar users watched and liked
        Map<String, Double> candidateScores = new HashMap<>();
        for (Long similarUserId : similarUsers) {
            List<ViewingHistory> theirHistory = getViewingHistory(similarUserId);
            for (ViewingHistory view : theirHistory) {
                // Skip if current user already watched
                if (userHasWatched(userId, view.contentId)) {
                    continue;
                }
                // Calculate recommendation score
                double score = calculateScore(userId, view.contentId,
                                               similarUserId, view);
                candidateScores.merge(view.contentId, score, Double::sum);
            }
        }
        // 5. Content-based filtering: find similar content
        for (ViewingHistory view : history) {
            if (view.completed) {
                List<String> similarContent = findSimilarContent(view.contentId);
                for (String contentId : similarContent) {
                    if (!userHasWatched(userId, contentId)) {
                        double score = 0.5; // Lower weight than collaborative
                        candidateScores.merge(contentId, score, Double::sum);
                    }
                }
            }
        }
        // 6. Sort by score and return top N
        List<Recommendation> recommendations = candidateScores.entrySet()
            .stream()
            .sorted(Map.Entry.<String, Double>comparingByValue().reversed())
            .limit(limit)
            .map(e -> new Recommendation(e.getKey(), e.getValue()))
            .collect(Collectors.toList());
        // 7. Cache results for 1 hour
        cache.setWithExpiry(cacheKey, recommendations, 3600);
        return recommendations;
    }
    /**
     * Finds users with similar viewing patterns using cosine similarity.
     */
    private List<Long> findSimilarUsers(long userId, List<ViewingHistory> userHistory) {
        // In production: use Spark to compute user similarity matrix
        // For demo, simplified approach
        // Get content IDs user has watched
        Set<String> userContentIds = userHistory.stream()
            .map(h -> h.contentId)
            .collect(Collectors.toSet());
        // Query database for users who watched similar content
        List<Long> similarUsers = cassandra.query(
            "SELECT DISTINCT user_id FROM viewing_history " +
            "WHERE content_id IN ? LIMIT 100",
            userContentIds.toArray()
        );
        return similarUsers.stream()
            .filter(uid -> uid != userId)
            .limit(20)
            .collect(Collectors.toList());
    }
    /**
     * Calculates recommendation score for a content item.
     */
    private double calculateScore(long userId, String contentId,
                                   long similarUserId, ViewingHistory view) {
        double score = 1.0;
        // Factor 1: User similarity (computed offline)
        double userSimilarity = getUserSimilarity(userId, similarUserId);
        score *= userSimilarity;
        // Factor 2: Content popularity
        int viewCount = getContentViewCount(contentId);
        double popularityScore = Math.log(viewCount + 1) / 10.0;
        score *= (1.0 + popularityScore);
        // Factor 3: Recency (recent content weighted higher)
        long daysSinceRelease = getDaysSinceRelease(contentId);
        if (daysSinceRelease < 30) {
            score *= 1.5; // Boost new content
        }
        // Factor 4: Completion rate (if similar user finished it)
        if (view.completed) {
            score *= 1.3;
        }
        return score;
    }
    /**
     * Finds content similar to given content based on metadata.
     */
    private List<String> findSimilarContent(String contentId) {
        // Get content metadata
        Content content = getContent(contentId);
        // Find content with overlapping genres, cast, director
        List<String> similar = cassandra.query(
            "SELECT content_id FROM content " +
            "WHERE genres CONTAINS ANY ? " +
            "AND content_id != ? " +
            "LIMIT 10",
            content.genres, contentId
        );
        return similar;
    }
    private boolean userHasWatched(long userId, String contentId) {
        return cassandra.exists(
            "SELECT * FROM viewing_history " +
            "WHERE user_id = ? AND content_id = ?",
            userId, contentId
        );
    }
    static class Recommendation {
        String contentId;
        double score;
        Recommendation(String contentId, double score) {
            this.contentId = contentId;
            this.score = score;
        }
    }
}