Design a Video Streaming Platform (Netflix)

Java — Transcoding orchestrator and GPU encode worker

// Transcoding job dispatcher — splits video and enqueues chunk jobs
public class TranscodingOrchestrator {

    private static final int CHUNK_DURATION_SECONDS = 600; // 10 minutes

    private final VideoSplitter splitter;
    private final SqsTemplate sqs;
    private final TranscodingJobRepository jobRepo;

    public void dispatch(UploadCompletedEvent event) {
        String rawS3Key = event.s3Key();
        long durationSecs = event.durationSeconds();
        long titleId = event.titleId();

        // Split into 10-minute chunks
        List<VideoChunk> chunks = splitter.split(rawS3Key, CHUNK_DURATION_SECONDS);

        List<String> targetRenditions = List.of(
            "240p-h264", "360p-h264", "480p-h264",
            "720p-h264", "1080p-h264", "1080p-hevc",
            "2160p-hevc", "2160p-av1"
        );

        TranscodingJob job = jobRepo.create(titleId, chunks.size() * targetRenditions.size());

        for (VideoChunk chunk : chunks) {
            for (String rendition : targetRenditions) {
                EncodeTask task = new EncodeTask(
                    job.id(), titleId, chunk.s3Key(),
                    chunk.index(), rendition
                );
                sqs.send("encode-tasks", task);
            }
        }
    }
}

// GPU worker — receives a single chunk encode task
@SqsListener("encode-tasks")
public void encodeChunk(EncodeTask task) {
    // Download chunk from S3
    Path inputPath = s3.download(task.chunkS3Key());

    // Run ffmpeg with target rendition settings
    RenditionConfig config = RenditionConfig.forProfile(task.rendition());
    Path outputPath = ffmpeg.encode(inputPath, config);

    // Upload encoded chunk to S3
    String outputKey = buildOutputKey(task);
    s3.upload(outputKey, outputPath);

    // Notify orchestrator of completion
    jobRepo.markChunkDone(task.jobId(), task.chunkIndex(), task.rendition(), outputKey);
}

Java — Simplified buffer-based ABR quality selector

// Simplified ABR quality selector (runs in the client player)
public class AbrQualitySelector {

    // Bitrate ladder (bps) — matches server renditions
    private static final int[] BITRATE_LADDER = {
        300_000,    // 240p
        800_000,    // 360p
        1_500_000,  // 480p
        3_000_000,  // 720p
        6_000_000,  // 1080p
        15_000_000, // 4K
    };

    private static final double BUFFER_MAX     = 30.0; // seconds
    private static final double BUFFER_MIN     = 10.0;
    private static final double BUFFER_CRITICAL = 5.0;

    private int currentIndex = 0;

    /**
     * @param bufferSeconds seconds of video buffered ahead
     * @param bandwidthBps  estimated download bandwidth (bits/sec)
     * @return index into BITRATE_LADDER for the next segment
     */
    public int selectQuality(double bufferSeconds, long bandwidthBps) {
        if (bufferSeconds < BUFFER_CRITICAL) {
            // Critical: drop to lowest immediately
            currentIndex = 0;
        } else if (bufferSeconds < BUFFER_MIN) {
            // Low buffer: step down one level
            currentIndex = Math.max(0, currentIndex - 1);
        } else if (bufferSeconds > BUFFER_MAX) {
            // Healthy buffer: step up if bandwidth supports it
            int next = currentIndex + 1;
            if (next < BITRATE_LADDER.length
                    && bandwidthBps > BITRATE_LADDER[next] * 1.2) { // 20% headroom
                currentIndex = next;
            }
        }
        // else: buffer in stable range — hold current quality
        return currentIndex;
    }
}

Java — Playback Service: OCA selection and signed manifest URL

// Playback Service — OCA selection and manifest URL generation
@Service
public class PlaybackService {

    private final OcaSteeringService ocaSteering;
    private final EntitlementService entitlement;
    private final ManifestStore manifestStore;
    private final SigningKeyService signingKeys;

    public PlaybackSession startPlay(long userId, long titleId, String clientIp,
                                     String deviceId) {
        // 1. Validate entitlement (subscription active, concurrent stream limit)
        entitlement.validate(userId, deviceId);

        // 2. Select best OCA for this client
        OcaNode bestOca = ocaSteering.selectOca(clientIp, titleId);

        // 3. Fetch manifest metadata from catalog
        ManifestMetadata meta = manifestStore.getManifest(titleId);

        // 4. Build signed manifest URL pointing to selected OCA
        String manifestUrl = buildSignedUrl(bestOca.hostname(), meta.masterManifestPath(),
            userId, titleId, signingKeys.currentKey());

        // 5. Issue playback token (DRM key delivery auth)
        String playbackToken = signingKeys.issuePlaybackToken(userId, titleId,
            deviceId, Duration.ofHours(8));

        return new PlaybackSession(manifestUrl, playbackToken,
            meta.availableAudioTracks(), meta.availableSubtitles());
    }

    private String buildSignedUrl(String ocaHost, String path,
                                   long userId, long titleId, SigningKey key) {
        long expiresAt = Instant.now().plus(8, ChronoUnit.HOURS).getEpochSecond();
        String policy = String.format("uid=%d&tid=%d&exp=%d", userId, titleId, expiresAt);
        String signature = key.sign(policy);
        return "https://" + ocaHost + path + "?" + policy + "&sig=" + signature;
    }
}

Java — Catalog Service with cache-aside and Kafka-triggered invalidation

// Content Catalog Service — title detail with cache-aside
@Service
public class CatalogService {

    private static final Duration CACHE_TTL = Duration.ofMinutes(5);
    private final RedisTemplate<String, TitleDetail> redis;
    private final TitleRepository titleRepo;
    private final EpisodeRepository episodeRepo;

    public TitleDetail getTitleDetail(long titleId, String locale) {
        String cacheKey = "title:" + titleId + ":" + locale;
        TitleDetail cached = redis.opsForValue().get(cacheKey);
        if (cached != null) return cached;

        // DB fetch — joined query for title + cast + episodes
        Title title = titleRepo.findByIdWithCast(titleId)
            .orElseThrow(() -> new NotFoundException("title", titleId));

        Localisation loc = title.getLocalisation(locale)
            .orElse(title.getLocalisation("en-US").orElseThrow());

        List<Episode> episodes = title.isSeriesType()
            ? episodeRepo.findByTitleIdOrderedBySeasonAndEpisode(titleId)
            : Collections.emptyList();

        TitleDetail detail = TitleDetail.builder()
            .titleId(titleId)
            .displayTitle(loc.displayTitle())
            .synopsis(loc.synopsis())
            .maturityRating(title.maturityRating())
            .releaseYear(title.releaseYear())
            .genres(title.genres())
            .cast(title.castMembers())
            .episodes(episodes)
            .thumbnailUrl(cdnUrl(title.thumbnailS3Key()))
            .build();

        redis.opsForValue().set(cacheKey, detail, CACHE_TTL);
        return detail;
    }

    // Called when catalog is updated — invalidate stale caches
    @KafkaListener(topics = "catalog-updates")
    public void onCatalogUpdate(CatalogUpdatedEvent event) {
        // Invalidate all locale variants for the updated title
        Set<String> keys = redis.keys("title:" + event.titleId() + ":*");
        if (keys != null && !keys.isEmpty()) redis.delete(keys);
    }
}