設計問題: 動画配信プラットフォーム — YouTube を設計する

# app/controllers/uploads_controller.rb
class UploadsController < ApplicationController
  before_action :authenticate_user!
 
  # Step 1: アップロード開始（Multipart Upload IDを発行）
  def initiate
    video = Video.create!(
      user: current_user,
      title: params[:title],
      status: 'uploading',
      filename: params[:filename],
      content_type: params[:content_type]
    )
 
    response = S3_CLIENT.create_multipart_upload(
      bucket: ENV['S3_RAW_BUCKET'],
      key: "raw/#{video.id}/#{video.filename}",
      content_type: params[:content_type],
      metadata: {
        'video-id' => video.id.to_s,
        'user-id'  => current_user.id.to_s
      }
    )
 
    render json: {
      video_id:  video.id,
      upload_id: response.upload_id,
      key:       "raw/#{video.id}/#{video.filename}"
    }
  end
 
  # Step 2: チャンクごとのPresigned URL発行
  def presigned_url
    url = S3_CLIENT.presigned_url(
      :upload_part,
      bucket:      ENV['S3_RAW_BUCKET'],
      key:         params[:key],
      upload_id:   params[:upload_id],
      part_number: params[:part_number].to_i,
      expires_in:  3600
    )
    render json: { url: url }
  end
 
  # Step 3: アップロード完了通知 → トランスコードキューへ
  def complete
    S3_CLIENT.complete_multipart_upload(
      bucket:           ENV['S3_RAW_BUCKET'],
      key:              params[:key],
      upload_id:        params[:upload_id],
      multipart_upload: { parts: params[:parts] }
    )
 
    video = Video.find(params[:video_id])
    video.update!(status: 'queued')
 
    # SQS経由でトランスコードをキューに積む
    SQS_CLIENT.send_message(
      queue_url:    ENV['TRANSCODE_QUEUE_URL'],
      message_body: { video_id: video.id, key: params[:key] }.to_json
    )
 
    render json: { status: 'queued', video_id: video.id }
  end
 
  # アップロード中断（チャンクの削除）
  def abort
    S3_CLIENT.abort_multipart_upload(
      bucket:    ENV['S3_RAW_BUCKET'],
      key:       params[:key],
      upload_id: params[:upload_id]
    )
    Video.find(params[:video_id]).update!(status: 'aborted')
    head :no_content
  end
end

// internal/uploader/multipart.go
package uploader
 
import (
    "bytes"
    "context"
    "fmt"
    "io"
    "sort"
    "sync"
 
    "github.com/aws/aws-sdk-go-v2/aws"
    "github.com/aws/aws-sdk-go-v2/service/s3"
    "github.com/aws/aws-sdk-go-v2/service/s3/types"
)
 
const chunkSize = 5 * 1024 * 1024 // 5MB（S3の最小チャンクサイズ）
const maxParallel = 10
 
type MultipartUploader struct {
    client *s3.Client
    bucket string
}
 
func (u *MultipartUploader) Upload(ctx context.Context, key string, r io.Reader) error {
    // Multipart Upload 開始
    createOut, err := u.client.CreateMultipartUpload(ctx, &s3.CreateMultipartUploadInput{
        Bucket:      aws.String(u.bucket),
        Key:         aws.String(key),
        ContentType: aws.String("video/mp4"),
    })
    if err != nil {
        return fmt.Errorf("create multipart upload: %w", err)
    }
    uploadID := *createOut.UploadId
 
    // チャンクを並列アップロード
    sem := make(chan struct{}, maxParallel)
    var mu sync.Mutex
    var parts []types.CompletedPart
    var uploadErr error
    var wg sync.WaitGroup
 
    for partNum := int32(1); ; partNum++ {
        buf := make([]byte, chunkSize)
        n, readErr := io.ReadFull(r, buf)
        if n == 0 {
            break
        }
        if readErr != nil && readErr != io.ErrUnexpectedEOF {
            uploadErr = readErr
            break
        }
        chunk := buf[:n]
        pn := partNum
 
        wg.Add(1)
        sem <- struct{}{}
        go func() {
            defer wg.Done()
            defer func() { <-sem }()
 
            out, err := u.client.UploadPart(ctx, &s3.UploadPartInput{
                Bucket:     aws.String(u.bucket),
                Key:        aws.String(key),
                UploadId:   aws.String(uploadID),
                PartNumber: aws.Int32(pn),
                Body:       bytes.NewReader(chunk),
            })
            if err != nil {
                mu.Lock()
                uploadErr = err
                mu.Unlock()
                return
            }
            mu.Lock()
            parts = append(parts, types.CompletedPart{
                ETag:       out.ETag,
                PartNumber: aws.Int32(pn),
            })
            mu.Unlock()
        }()
    }
    wg.Wait()
 
    if uploadErr != nil {
        // 失敗時はアップロードを中断してリソースを解放
        u.client.AbortMultipartUpload(ctx, &s3.AbortMultipartUploadInput{
            Bucket:   aws.String(u.bucket),
            Key:      aws.String(key),
            UploadId: aws.String(uploadID),
        })
        return uploadErr
    }
 
    // パーツ番号でソートしてから完了通知（順序が必要）
    sort.Slice(parts, func(i, j int) bool {
        return *parts[i].PartNumber < *parts[j].PartNumber
    })
 
    _, err = u.client.CompleteMultipartUpload(ctx, &s3.CompleteMultipartUploadInput{
        Bucket:          aws.String(u.bucket),
        Key:             aws.String(key),
        UploadId:        aws.String(uploadID),
        MultipartUpload: &types.CompletedMultipartUpload{Parts: parts},
    })
    return err
}

# app/jobs/transcode_job.rb
class TranscodeJob < ApplicationJob
  queue_as :transcoding
  retry_on Aws::MediaConvert::Errors::ServiceError,
           wait: :exponentially_longer, attempts: 3
 
  QUALITY_PROFILES = [
    { name: '360p',  height: 360,  bitrate: 1_000_000,  codec: 'H_264', preset: 'FASTER' },
    { name: '720p',  height: 720,  bitrate: 3_000_000,  codec: 'H_264', preset: 'DEFAULT' },
    { name: '1080p', height: 1080, bitrate: 6_000_000,  codec: 'H_264', preset: 'SLOWER' },
    { name: '4k',    height: 2160, bitrate: 20_000_000, codec: 'H_265', preset: 'SLOWER' }
  ].freeze
 
  def perform(video_id)
    video = Video.find(video_id)
    video.update!(status: 'transcoding', transcoding_started_at: Time.current)
 
    # 入力ファイルの解像度を確認して不要なプロファイルをスキップ
    profiles = filter_profiles_by_resolution(video)
 
    response = MEDIACONVERT_CLIENT.create_job(
      role:     ENV['MEDIACONVERT_ROLE_ARN'],
      settings: build_job_settings(video, profiles)
    )
 
    video.update!(transcode_job_id: response.job.id)
    Rails.logger.info "Transcode started: #{response.job.id} for video #{video_id}"
  rescue ActiveRecord::RecordNotFound => e
    Rails.logger.error "Video #{video_id} not found: #{e.message}"
  end
 
  private
 
  def filter_profiles_by_resolution(video)
    return QUALITY_PROFILES unless video.original_height
    QUALITY_PROFILES.select { |p| p[:height] <= video.original_height }
  end
 
  def build_job_settings(video, profiles)
    input_key     = "raw/#{video.id}/#{video.filename}"
    output_prefix = "processed/#{video.id}/"
    {
      inputs: [{
        file_input: "s3://#{ENV['S3_RAW_BUCKET']}/#{input_key}",
        audio_selectors: { 'Audio Selector 1' => { default_selection: 'DEFAULT' } }
      }],
      output_groups: [
        hls_output_group(output_prefix, profiles),
        thumbnail_output_group(output_prefix)
      ]
    }
  end
 
  def hls_output_group(prefix, profiles)
    {
      name: 'HLS Group',
      output_group_settings: {
        type: 'HLS_GROUP_SETTINGS',
        hls_group_settings: {
          destination:        "s3://#{ENV['S3_PROCESSED_BUCKET']}/#{prefix}hls/",
          segment_length:     6,
          min_segment_length: 0
        }
      },
      outputs: profiles.map { |p| hls_output(p) }
    }
  end
 
  def hls_output(profile)
    {
      name_modifier: "_#{profile[:name]}",
      container_settings: { container: 'M3U8' },
      video_description: {
        height: profile[:height],
        codec_settings: {
          codec: profile[:codec],
          h264_settings: {
            bitrate:           profile[:bitrate],
            rate_control_mode: 'CBR',
            codec_profile:     'MAIN',
            codec_level:       'AUTO'
          }
        }
      },
      audio_descriptions: [{
        audio_source_name: 'Audio Selector 1',
        codec_settings: {
          codec: 'AAC',
          aac_settings: {
            bitrate:     128_000,
            coding_mode: 'CODING_MODE_2_0',
            sample_rate: 48_000
          }
        }
      }]
    }
  end
 
  def thumbnail_output_group(prefix)
    {
      name: 'Thumbnails',
      output_group_settings: {
        type: 'FILE_GROUP_SETTINGS',
        file_group_settings: {
          destination: "s3://#{ENV['S3_PROCESSED_BUCKET']}/#{prefix}thumbnails/"
        }
      },
      outputs: [{
        container_settings: { container: 'RAW' },
        video_description: {
          codec_settings: {
            codec: 'FRAME_CAPTURE',
            frame_capture_settings: {
              framerate_numerator:   1,
              framerate_denominator: 10,
              max_captures:          10
            }
          }
        }
      }]
    }
  end
end

# HLSマスターマニフェスト（master.m3u8）の内容
#EXTM3U
#EXT-X-VERSION:3
 
# 360p: 低帯域向け
#EXT-X-STREAM-INF:BANDWIDTH=1000000,RESOLUTION=640x360,CODECS="avc1.42c01e,mp4a.40.2"
_360p.m3u8
 
# 720p: 標準
#EXT-X-STREAM-INF:BANDWIDTH=3000000,RESOLUTION=1280x720,CODECS="avc1.4d401f,mp4a.40.2"
_720p.m3u8
 
# 1080p: 高画質
#EXT-X-STREAM-INF:BANDWIDTH=6000000,RESOLUTION=1920x1080,CODECS="avc1.640028,mp4a.40.2"
_1080p.m3u8

// internal/streaming/handler.go
package streaming
 
import (
    "fmt"
    "net/http"
    "time"
 
    "github.com/aws/aws-sdk-go-v2/aws"
    "github.com/aws/aws-sdk-go-v2/service/s3"
)
 
type StreamHandler struct {
    presigner *s3.PresignClient
    bucket    string
}
 
// マニフェスト・セグメントへのプリサインドURLを発行してリダイレクト
func (h *StreamHandler) ServeSegment(w http.ResponseWriter, r *http.Request) {
    videoID := r.PathValue("video_id")
    path    := r.PathValue("path") // 例: hls/_720p_00001.ts
 
    key := fmt.Sprintf("processed/%s/%s", videoID, path)
 
    presigned, err := h.presigner.PresignGetObject(r.Context(), &s3.GetObjectInput{
        Bucket: aws.String(h.bucket),
        Key:    aws.String(key),
    }, s3.WithPresignExpires(15*time.Minute))
    if err != nil {
        http.Error(w, "failed to presign", http.StatusInternalServerError)
        return
    }
 
    // CDNキャッシュを活かすためリダイレクト（302）
    http.Redirect(w, r, presigned.URL, http.StatusFound)
}

# app/jobs/thumbnail_generation_job.rb
class ThumbnailGenerationJob < ApplicationJob
  queue_as :thumbnails
 
  TIMESTAMP_RATIOS = [0.1, 0.3, 0.5, 0.7, 0.9].freeze
 
  def perform(video_id)
    video      = Video.find(video_id)
    local_path = download_raw_sample(video)
    duration   = probe_duration(local_path)
    return unless duration&.positive?
 
    TIMESTAMP_RATIOS.each_with_index do |ratio, index|
      timestamp  = duration * ratio
      thumb_path = "/tmp/thumb_#{video_id}_#{index}.jpg"
 
      system(
        'ffmpeg', '-ss', timestamp.to_s,
        '-i', local_path,
        '-vframes', '1',
        '-q:v', '2',
        '-vf', 'scale=1280:720:force_original_aspect_ratio=decrease',
        thumb_path, '-y'
      )
 
      S3_CLIENT.put_object(
        bucket:       ENV['S3_PROCESSED_BUCKET'],
        key:          "processed/#{video_id}/thumbnails/thumb_#{index}.jpg",
        body:         File.binread(thumb_path),
        content_type: 'image/jpeg'
      )
    end
 
    video.update!(
      thumbnail_url: "https://cdn.example.com/processed/#{video_id}/thumbnails/thumb_2.jpg",
      status:        'ready'
    )
  end
 
  private
 
  def probe_duration(path)
    out = `ffprobe -v error -show_entries format=duration \
           -of default=noprint_wrappers=1:nokey=1 #{path} 2>&1`
    out.strip.to_f
  end
 
  def download_raw_sample(video)
    path = "/tmp/raw_#{video.id}_sample.mp4"
    # 先頭50MBのみダウンロード（duration取得なら十分）
    S3_CLIENT.get_object(
      bucket: ENV['S3_RAW_BUCKET'],
      key:    "raw/#{video.id}/#{video.filename}",
      range:  'bytes=0-52428800'
    ) { |chunk| File.open(path, 'ab') { |f| f.write(chunk) } }
    path
  end
end

# CloudFront ディストリビューション設定（概念）
distribution:
  origins:
    - id: s3-processed
      domain: my-processed-bucket.s3.ap-northeast-1.amazonaws.com
      s3_origin_config:
        origin_access_control: true  # OACでS3を非公開に
 
  cache_behaviors:
    # HLSセグメント（.ts）: 不変なので長期キャッシュ
    - path_pattern: "*/hls/*.ts"
      ttl_default: 2592000   # 30日
      ttl_max:     2592000
      compress: false         # 動画は圧縮不要（エンコード済み）
      viewer_protocol: redirect-to-https
 
    # HLSマニフェスト（.m3u8）: 短めに設定
    - path_pattern: "*/hls/*.m3u8"
      ttl_default: 300        # 5分（ライブ更新のケースに備えて短く）
      ttl_max:     3600
      cache_policy: CachingOptimized
 
    # サムネイル: 中程度
    - path_pattern: "*/thumbnails/*.jpg"
      ttl_default: 86400      # 24時間
      ttl_max:     604800     # 7日
 
  # オリジンフェイルオーバー（高可用性）
  origin_groups:
    - id: s3-failover
      members:
        - s3-processed           # プライマリ
        - s3-processed-replica   # リプリカ（別リージョン）
      failover_criteria:
        status_codes: [500, 502, 503, 504]
 
  price_class: PriceClass_All    # 全PoP（400+）を利用
  http_version: http2and3        # HTTP/3 (QUIC) で低レイテンシ

アップロードパス:
  Route53:    upload.example.com → ALB
  ALB:        HTTPS終端、ヘルスチェック
  ECS Fargate:
    - Upload API (Rails)
    - 2vCPU / 4GB RAM
    - Auto Scaling: CPU 70%でスケールアウト
  S3 Raw Bucket:
    - 暗号化: SSE-S3
    - ライフサイクル: 30日後 Glacier へ
    - AbortIncompleteMultipartUpload: 7日後
 
トランスコードパス:
  SQS:
    - キュー: transcode-jobs.fifo
    - 可視性タイムアウト: 30分
    - DLQ: 3回失敗後に移動
  ECS Worker (Go):
    - MediaConvertジョブを投入して完了を待機
  AWS MediaConvert:
    - マネージドトランスコード
    - オンデマンド課金
    - 並列ジョブ上限: 20（増加申請可能）
  Lambda（完了通知）:
    - EventBridge で MediaConvert 完了イベントを受信
    - Aurora の動画ステータスを 'ready' に更新
 
配信パス:
  CloudFront: 400+ PoP、HTTP/3対応、OAC
  S3 Processed:
    - Intelligent-Tiering（自動ティア移行）
    - クロスリージョンレプリケーション（DR対応）
 
データストア:
  Aurora PostgreSQL:
    - 動画メタデータ（videos, users, channels）
    - Read Replica x 2（読み取り分散）
  ElastiCache Redis:
    - 動画メタデータキャッシュ（TTL: 1時間）
    - 再生位置保存（視聴再開機能）
  OpenSearch:
    - 全文検索（タイトル・説明文・タグ）
    - DynamoDBストリームで自動インデックス更新
  DynamoDB:
    - 視聴履歴（ユーザーID × 動画ID）
    - 再生数カウンター（アトミック更新）

# app/jobs/storage_tiering_job.rb
# 古い動画をGlacierに移行するジョブ（週次実行）
class StorageTieringJob < ApplicationJob
  queue_as :maintenance
 
  def perform
    old_videos = Video.where(status: 'ready')
                      .where('last_viewed_at < ?', 90.days.ago)
                      .where(archived: false)
 
    old_videos.find_each do |video|
      archive_to_glacier(video)
      video.update!(storage_class: 'GLACIER', archived: true)
    end
 
    Rails.logger.info "Archived #{old_videos.count} videos to Glacier"
  end
 
  private
 
  def archive_to_glacier(video)
    prefix = "processed/#{video.id}/"
 
    S3_CLIENT.list_objects_v2(
      bucket: ENV['S3_PROCESSED_BUCKET'],
      prefix: prefix
    ).contents.each do |object|
      S3_CLIENT.copy_object(
        bucket:             ENV['S3_PROCESSED_BUCKET'],
        copy_source:        "#{ENV['S3_PROCESSED_BUCKET']}/#{object.key}",
        key:                object.key,
        storage_class:      'GLACIER',
        metadata_directive: 'COPY'
      )
    end
  end
end

# app/services/copyright_detector.rb
class CopyrightDetector
  # pHash（知覚的ハッシュ）: 色調が似ていれば同じハッシュ値になる
  # 完全一致ではなくハミング距離で類似判定（改変にも対応）
  HAMMING_DISTANCE_THRESHOLD = 10
 
  def detect(video_id)
    video       = Video.find(video_id)
    fingerprint = extract_fingerprint(video)
 
    matches = query_fingerprint_db(fingerprint)
    return :clean if matches.empty?
 
    best_match = matches.min_by { |m| hamming_distance(fingerprint, m[:fingerprint]) }
    similarity = 1.0 - best_match[:distance].to_f / 64
 
    handle_copyright_match(video, best_match[:rights_holder]) if similarity >= 0.90
  end
 
  private
 
  def extract_fingerprint(video)
    # FFmpegで1秒ごとにフレームを抽出してpHashを計算
    frames = extract_frames(video, interval: 1)
    frames.map { |frame| phash(frame) }
  end
 
  def hamming_distance(hash1, hash2)
    (hash1 ^ hash2).to_s(2).count('1')
  end
 
  def handle_copyright_match(video, rights_holder)
    case rights_holder[:policy]
    when 'block'
      video.update!(status: 'blocked', block_reason: 'copyright')
    when 'monetize'
      video.update!(monetize_for: rights_holder[:id])
    end
  end
end

# config/initializers/cloudwatch_metrics.rb
module Metrics
  def self.record_transcode_result(video_id, success:, duration_sec:)
    CLOUDWATCH_CLIENT.put_metric_data(
      namespace:   'VideoApp/Transcode',
      metric_data: [
        {
          metric_name: 'JobCompleted',
          value:       success ? 1.0 : 0.0,
          unit:        'Count',
          dimensions:  [{ name: 'Environment', value: Rails.env }]
        },
        {
          metric_name: 'JobDurationSeconds',
          value:       duration_sec,
          unit:        'Seconds'
        }
      ]
    )
  end
 
  def self.record_upload_complete(file_size_mb:)
    CLOUDWATCH_CLIENT.put_metric_data(
      namespace:   'VideoApp/Upload',
      metric_data: [{
        metric_name: 'FileSizeMB',
        value:       file_size_mb,
        unit:        'Count'
      }]
    )
  end
end

# CloudWatch アラーム設定（概念）
alarms:
  - name: TranscodeFailureRate
    metric: JobCompleted (sum 0) / total jobs
    threshold: 2%   # 2%以上失敗したらSlack通知
    period: 300sec
 
  - name: CDNCacheHitRateLow
    metric: CloudFront CacheHitRate
    threshold: 80%  # 80%を下回ったらコスト増のサイン
    period: 600sec
 
  - name: TimeToViewableHigh
    metric: VideoApp/Pipeline/TimeToViewableSeconds P95
    threshold: 300  # 5分以内が目標
    period: 300sec