Design a Video Streaming Platform

Java — Upload Service: initiate and complete chunked multipart upload

// Upload Service — initiate multipart upload and return pre-signed part URLs
@RestController
@RequestMapping("/uploads")
public class UploadController {

    private static final long CHUNK_SIZE_BYTES = 100L * 1024 * 1024; // 100 MB

    private final S3PresignClient s3Presign;
    private final S3Client s3;
    private final RedisTemplate<String, UploadState> redis;
    private final KafkaTemplate<String, VideoUploaded> kafka;

    @PostMapping
    public InitiateUploadResponse initiateUpload(@RequestBody @Valid InitiateUploadRequest req,
                                                  @AuthenticationPrincipal Creator creator) {
        // 1. Create multipart upload session in S3
        String rawKey = "raw/" + UUID.randomUUID() + "/" + req.filename();
        CreateMultipartUploadResponse mpResponse = s3.createMultipartUpload(b -> b
            .bucket("video-raw")
            .key(rawKey)
            .contentType(req.contentType())
            .metadata(Map.of("creatorId", creator.id().toString())));

        String s3UploadId = mpResponse.uploadId();
        int totalParts = (int) Math.ceil((double) req.fileSizeBytes() / CHUNK_SIZE_BYTES);

        // 2. Pre-sign a URL for each part
        List<String> partUrls = new ArrayList<>();
        for (int part = 1; part <= totalParts; part++) {
            UploadPartPresignRequest presignReq = UploadPartPresignRequest.builder()
                .signatureDuration(Duration.ofHours(2))
                .uploadPartRequest(b -> b
                    .bucket("video-raw")
                    .key(rawKey)
                    .uploadId(s3UploadId)
                    .partNumber(part))
                .build();
            partUrls.add(s3Presign.presignUploadPart(presignReq).url().toString());
        }

        // 3. Persist upload state to Redis (TTL = 72 hours)
        String uploadId = UUID.randomUUID().toString();
        UploadState state = new UploadState(uploadId, s3UploadId, rawKey,
            totalParts, new ArrayList<>(), creator.id(), req.title());
        redis.opsForValue().set("upload:" + uploadId, state, Duration.ofHours(72));

        return new InitiateUploadResponse(uploadId, partUrls);
    }

    @PostMapping("/{uploadId}/complete")
    public CompleteUploadResponse completeUpload(@PathVariable String uploadId,
                                                  @RequestBody CompleteUploadRequest req) {
        UploadState state = redis.opsForValue().get("upload:" + uploadId);
        if (state == null) throw new NotFoundException("Upload session expired or not found");

        // 4. Complete S3 multipart upload
        List<CompletedPart> parts = req.parts().stream()
            .map(p -> CompletedPart.builder()
                .partNumber(p.partNumber()).eTag(p.etag()).build())
            .toList();
        s3.completeMultipartUpload(b -> b
            .bucket("video-raw")
            .key(state.rawKey())
            .uploadId(state.s3UploadId())
            .multipartUpload(u -> u.parts(parts)));

        // 5. Publish event to trigger transcoding pipeline
        String videoId = UUID.randomUUID().toString();
        kafka.send("video-uploaded", new VideoUploaded(
            videoId, state.rawKey(), state.creatorId(), state.title(),
            List.of("360p", "720p", "1080p", "2160p")));

        redis.delete("upload:" + uploadId);
        return new CompleteUploadResponse(videoId, "PROCESSING");
    }
}

Java — Transcode worker: chunk-parallel FFmpeg encoding with HLS segmentation

// Transcode worker — processes one rendition job
@Component
public class TranscodeWorker {

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

    private final S3Client s3;
    private final FFmpegRunner ffmpeg;
    private final KafkaTemplate<String, Object> kafka;

    @KafkaListener(topics = "transcode-tasks", concurrency = "8")
    public void processTask(TranscodeTask task) {
        Path workDir = Files.createTempDirectory("transcode-" + task.taskId());
        try {
            // 1. Download raw source from S3
            Path rawFile = workDir.resolve("source.mp4");
            s3.getObject(b -> b.bucket("video-raw").key(task.rawS3Key()),
                ResponseTransformer.toFile(rawFile));

            // 2. Split into 10-minute chunks for parallel processing
            List<Path> chunks = ffmpeg.splitIntoChunks(rawFile, CHUNK_DURATION_SECS, workDir);

            // 3. Encode each chunk at the target rendition
            RenditionConfig config = RenditionConfig.forProfile(task.rendition());
            List<Path> encodedChunks = new ArrayList<>();
            for (int i = 0; i < chunks.size(); i++) {
                Path encoded = workDir.resolve("encoded_chunk_" + i + ".ts");
                ffmpeg.encode(chunks.get(i), encoded, config);
                encodedChunks.add(encoded);
            }

            // 4. Concatenate encoded chunks into a single stream
            Path concatenated = workDir.resolve("full_" + task.rendition() + ".ts");
            ffmpeg.concat(encodedChunks, concatenated);

            // 5. Segment into HLS .ts files (4-second segments)
            Path segmentDir = workDir.resolve("segments");
            Files.createDirectories(segmentDir);
            String segmentPattern = segmentDir.resolve("seg%03d.ts").toString();
            String playlistPath  = segmentDir.resolve("playlist.m3u8").toString();
            ffmpeg.segmentToHls(concatenated, segmentPattern, playlistPath, 4);

            // 6. Upload segments and playlist to S3
            String s3Prefix = "processed/" + task.videoId() + "/" + task.rendition() + "/";
            Files.list(segmentDir).forEach(file ->
                s3.putObject(b -> b.bucket("video-processed").key(s3Prefix + file.getFileName()),
                    RequestBody.fromFile(file)));

            // 7. Notify coordinator
            kafka.send("transcode-completed",
                new TranscodeCompleted(task.videoId(), task.rendition(),
                    s3Prefix + "playlist.m3u8"));

        } finally {
            FileUtils.deleteDirectory(workDir.toFile());
        }
    }
}

// Rendition config factory
public record RenditionConfig(String scale, String codec, int crf,
                               String preset, String targetBitrate) {
    public static RenditionConfig forProfile(String profile) {
        return switch (profile) {
            case "360p"  -> new RenditionConfig("scale=-2:360",  "libx264", 28, "fast",   "800k");
            case "720p"  -> new RenditionConfig("scale=-2:720",  "libx264", 23, "medium", "2500k");
            case "1080p" -> new RenditionConfig("scale=-2:1080", "libx264", 20, "medium", "5000k");
            case "2160p" -> new RenditionConfig("scale=-2:2160", "libx265", 22, "slow",   "15000k");
            default -> throw new IllegalArgumentException("Unknown rendition: " + profile);
        };
    }
}

Java — HLS master playlist generator with per-rendition metadata

// Manifest generator — builds master M3U8 and per-rendition playlists
@Component
public class HlsManifestGenerator {

    private final S3Client s3;

    /**
     * Called by TranscodeCoordinator once all renditions are complete.
     * Reads each rendition's playlist from S3 and builds the master playlist.
     */
    public String generateMasterPlaylist(String videoId,
                                          List<RenditionMeta> renditions) {
        StringBuilder master = new StringBuilder();
        master.append("#EXTM3U
");
        master.append("#EXT-X-VERSION:3

");

        for (RenditionMeta r : renditions) {
            master.append(String.format(
                "#EXT-X-STREAM-INF:BANDWIDTH=%d,RESOLUTION=%s,CODECS="%s"
",
                r.bandwidthBps(), r.resolution(), r.codec()));
            // Relative path — CDN resolves against the videoId prefix
            master.append(r.rendition() + "/playlist.m3u8
");
        }

        String masterKey = "processed/" + videoId + "/master.m3u8";
        s3.putObject(
            b -> b.bucket("video-processed").key(masterKey)
                  .contentType("application/vnd.apple.mpegurl")
                  // short TTL on manifest so clients pick up new renditions quickly
                  .cacheControl("public, max-age=60"),
            RequestBody.fromString(master.toString()));

        return masterKey;
    }
}

// Rendition metadata used during manifest generation
public record RenditionMeta(
    String rendition,    // e.g. "720p"
    String resolution,   // e.g. "1280x720"
    long   bandwidthBps, // e.g. 2_500_000
    String codec         // e.g. "avc1.64001F"
) {}

Java — ABR controller with EWMA bandwidth estimation and buffer-based guard rails

// Client-side ABR quality selector (runs in the browser/native player)
public class AbrController {

    private static final double EWMA_ALPHA      = 0.3;
    private static final double BANDWIDTH_MARGIN = 0.8; // use 80% of estimate
    private static final double BUFFER_STEP_UP   = 30.0;
    private static final double BUFFER_HOLD_LOW  = 15.0;
    private static final double BUFFER_STEP_DOWN = 10.0;
    private static final double BUFFER_PANIC     = 5.0;

    // Rendition bitrate ladder (bits per second)
    private final long[] bitrateladder = {
        400_000L,    // 360p
        1_500_000L,  // 720p
        4_000_000L,  // 1080p
        12_000_000L, // 4K
    };

    private double estimatedBandwidthBps = bitrateladder[0]; // start conservative
    private int    currentLevel          = 0;

    /**
     * Called after each segment download completes.
     *
     * @param segmentBytes   bytes received for the last segment
     * @param downloadMs     wall-clock milliseconds to download
     * @param bufferSeconds  decoded seconds ahead of the playhead
     * @return next rendition index to request
     */
    public int selectNextLevel(long segmentBytes, long downloadMs, double bufferSeconds) {
        // 1. Update bandwidth estimate (EWMA)
        double instantBw = (segmentBytes * 8.0) / (downloadMs / 1000.0);
        estimatedBandwidthBps = EWMA_ALPHA * instantBw
            + (1 - EWMA_ALPHA) * estimatedBandwidthBps;

        double effectiveBw = estimatedBandwidthBps * BANDWIDTH_MARGIN;

        // 2. Buffer-based guard rails
        if (bufferSeconds < BUFFER_PANIC) {
            currentLevel = 0; // emergency drop
            return currentLevel;
        }

        if (bufferSeconds < BUFFER_STEP_DOWN) {
            currentLevel = Math.max(0, currentLevel - 1);
            return currentLevel;
        }

        // 3. Throughput-based selection within buffer-healthy range
        if (bufferSeconds >= BUFFER_STEP_UP) {
            // Try to step up if bandwidth supports the next level
            int desired = currentLevel;
            for (int i = bitrateladder.length - 1; i >= 0; i--) {
                if (effectiveBw >= bitrateladder[i]) {
                    desired = i;
                    break;
                }
            }
            // Only step up one level at a time to avoid overloading
            currentLevel = Math.min(currentLevel + 1, desired);
        }
        // else: BUFFER_HOLD_LOW <= bufferSeconds < BUFFER_STEP_UP → hold

        return currentLevel;
    }
}

Java — CDN invalidation worker: CloudFront purge + S3 batch delete on video deletion

// CDN Invalidation Worker — consumes VideoDeleted events and purges CDN + schedules S3 cleanup
@Component
public class CdnInvalidationWorker {

    private final CloudFrontClient cloudFront;
    private final S3Client s3;
    private final VideoMetadataRepository metadataRepo;

    @Value("${cdn.cloudfront.distribution-id}")
    private String distributionId;

    @KafkaListener(topics = "video-deleted")
    public void onVideoDeleted(VideoDeleted event) {
        String videoId = event.videoId();

        // 1. Issue CloudFront invalidation for all paths under this video
        cloudFront.createInvalidation(b -> b
            .distributionId(distributionId)
            .invalidationBatch(ib -> ib
                .callerReference("delete-" + videoId + "-" + System.currentTimeMillis())
                .paths(p -> p
                    .quantity(1)
                    .items("/processed/" + videoId + "/*"))));

        log.info("CDN invalidation issued for videoId={}", videoId);

        // 2. Schedule async S3 deletion via S3 Batch Operations manifest
        // Build a CSV manifest of all S3 keys under this video
        List<String> segmentKeys = listAllSegmentKeys("video-processed",
            "processed/" + videoId + "/");

        String manifestKey = "delete-manifests/" + videoId + ".csv";
        String csvContent  = segmentKeys.stream()
            .map(k -> "video-processed," + k)
            .collect(Collectors.joining("
"));

        s3.putObject(b -> b.bucket("ops-manifests").key(manifestKey),
            RequestBody.fromString(csvContent));

        // Trigger S3 Batch Operations job (ARN configured externally)
        log.info("S3 batch delete manifest written for videoId={} ({} objects)",
            videoId, segmentKeys.size());

        // 3. Update metadata status (belt-and-suspenders — already set by Metadata Service)
        metadataRepo.setStatus(videoId, VideoStatus.DELETED);
    }

    private List<String> listAllSegmentKeys(String bucket, String prefix) {
        List<String> keys = new ArrayList<>();
        ListObjectsV2Request req = ListObjectsV2Request.builder()
            .bucket(bucket).prefix(prefix).build();
        s3.listObjectsV2Paginator(req)
          .forEach(page -> page.contents().forEach(obj -> keys.add(obj.key())));
        return keys;
    }
}