Fix legacy HLS paths when refreshing media playlists

lib/mave_core/assets.ex

@@ -1019,14 +1019,22 @@ defmodule MaveCore.Assets do
   end
 
   defp rebuild_hls_master_playlist(%Embed{} = embed, region) do
+    variants = hls_video_variant_outputs(embed, region)
+
+    with {:ok, audio_tracks} <- hls_master_audio_tracks(embed, region, variants) do
+      publish_hls_master_playlist(embed, region, variants, audio_tracks)
+    end
+  end
+
+  defp publish_hls_master_playlist(embed, region, variants, audio_tracks) do
     dependency_outputs =
-      (hls_video_variant_outputs(embed, region) ++
+      (variants ++
          [
            {"hls_audio_tracks",
             %{
               "step_type" => "media.package_hls_audio",
               "status" => "ok",
-              "audio_tracks" => manifest_audio_tracks(embed, region)
+              "audio_tracks" => audio_tracks
             }},
            {"hls_subtitles",
             %{
@@ -1051,8 +1059,12 @@ defmodule MaveCore.Assets do
     }
 
     case MediaBuildHlsMasterStep.run(step_definition, context) do
-      {:ok, output, _artifacts} -> {:ok, output}
-      {:error, reason} -> {:error, reason}
+      {:ok, output, _artifacts} ->
+        CdnCache.purge_best_effort(embed.space, region, [output["key"]])
+        {:ok, output}
+
+      {:error, reason} ->
+        {:error, reason}
     end
   end
 
@@ -1081,7 +1093,14 @@ defmodule MaveCore.Assets do
     )
     |> Repo.all()
     |> Enum.map(fn rendition ->
-      playlist_path = relative_playlist_path(rendition.rendition_key, embed.hash, version)
+      legacy? = legacy_rendition_key?(rendition.rendition_key)
+
+      playlist_path =
+        if legacy?,
+          do: "#{rendition.codec}_#{rendition.size}_hls/playlist.m3u8",
+          else: relative_playlist_path(rendition.rendition_key, embed.hash, version)
+
+      prefix = if version > 0, do: "#{embed.hash}/v#{version}/", else: "#{embed.hash}/"
 
       {"hls_#{rendition.codec}_#{rendition.size}",
        %{
@@ -1090,12 +1109,78 @@ defmodule MaveCore.Assets do
          "codec" => rendition.codec,
          "size" => rendition.size,
          "resolution" => hls_variant_resolution(embed.asset.current_video, rendition),
+         "legacy_rendition" => legacy?,
          "playlist_path" => playlist_path,
-         "playlist_uri" => "s3://#{bucket}/#{rendition.rendition_key}"
+         "playlist_uri" => "s3://#{bucket}/#{prefix}#{playlist_path}"
        }}
     end)
   end
 
+  # Older rendition keys identify database records, not storage objects. Those
+  # videos can also have muxed audio without a separate HLS audio rendition.
+  defp legacy_rendition_key?(key), do: Regex.match?(~r/\A[0-9a-fA-F]{32}\z/, key)
+
+  defp hls_master_audio_tracks(embed, region, variants) do
+    tracks = manifest_audio_tracks(embed, region)
+
+    if Enum.any?(variants, fn {_id, output} -> output["legacy_rendition"] end) do
+      existing_hls_audio_tracks(embed, region, tracks, variants)
+    else
+      {:ok, tracks}
+    end
+  end
+
+  defp existing_hls_audio_tracks(embed, region, tracks, variants) do
+    bucket = Storage.bucket_for_space(embed.space.hash, region)
+    version = Embeds.current_video_version(embed)
+    prefix = if version > 0, do: "#{embed.hash}/v#{version}/", else: "#{embed.hash}/"
+
+    Enum.reduce_while(tracks, {:ok, []}, fn track, {:ok, existing} ->
+      case storage_adapter().get(bucket, prefix <> track["hls_playlist"], region) do
+        {:ok, _body} ->
+          {:cont, {:ok, existing ++ [track]}}
+
+        {:error, :not_found} ->
+          missing_audio_result(bucket, prefix, region, variants, existing, track)
+
+        {:error, reason} ->
+          {:halt, {:error, {:audio_playlist_unavailable, reason}}}
+      end
+    end)
+  end
+
+  defp missing_audio_result(bucket, prefix, region, variants, existing, track) do
+    if legacy_variants_have_audio?(bucket, prefix, region, variants) do
+      {:cont, {:ok, existing}}
+    else
+      {:halt, {:error, {:missing_audio_playlist, track["hls_playlist"]}}}
+    end
+  end
+
+  defp legacy_variants_have_audio?(bucket, prefix, region, variants) do
+    variants != [] and
+      Enum.all?(variants, fn {_id, variant} ->
+        key = prefix <> Path.dirname(variant["playlist_path"]) <> "/init.mp4"
+
+        case storage_adapter().get_prefix(bucket, key, region, 1_048_576) do
+          {:ok, bytes} when is_binary(bytes) -> mp4_audio_handler?(bytes)
+          _ -> false
+        end
+      end)
+  end
+
+  defp mp4_audio_handler?(<<size::32, type::binary-size(4), rest::binary>>)
+       when size >= 8 and byte_size(rest) >= size - 8 do
+    data = binary_part(rest, 0, size - 8)
+    tail = binary_part(rest, size - 8, byte_size(rest) - size + 8)
+
+    (type == "hdlr" and match?(<<_::binary-size(8), "soun", _::binary>>, data)) or
+      (type in ["moov", "trak", "mdia"] and mp4_audio_handler?(data)) or
+      mp4_audio_handler?(tail)
+  end
+
+  defp mp4_audio_handler?(_bytes), do: false
+
   # Audio-only originals have no dimensions. Their generated SD waveform does,
   # and must keep them when a track or subtitle edit rebuilds the master playlist.
   defp hls_variant_resolution(
@@ -1164,9 +1249,12 @@ defmodule MaveCore.Assets do
     end)
   end
 
-  defp hls_subtitle_path(%Embed{} = embed, %Subtitle{path: path}, version)
+  defp hls_subtitle_path(%Embed{} = embed, %Subtitle{path: path, language: language}, version)
        when is_binary(path) and path != "" do
     cond do
+      legacy_rendition_key?(path) ->
+        "subtitle_#{language || "en"}.vtt"
+
       String.starts_with?(path, "http://") or String.starts_with?(path, "https://") ->
         path
 

test/mave_core/assets/hls_refresh_test.exs

@@ -0,0 +1,178 @@
+defmodule MaveCore.Assets.HlsRefreshTest do
+  use MaveCore.DataCase, async: false
+
+  alias MaveCore.Assets
+  alias MaveCore.Assets.{Asset, AudioTrack, Rendition, Subtitle, Video}
+  alias MaveCore.Embeds.Embed
+  alias MaveCore.Spaces.Space
+  alias MaveCore.TestSupport.FlowStorageAdapterStub, as: Storage
+
+  defmodule UnavailableStorage do
+    def get(_, _, _), do: {:error, :timeout}
+  end
+
+  setup do
+    previous = Application.get_env(:mave_core, :flow_storage_adapter)
+    previous_purger = Application.get_env(:mave_core, :cdn_cache_purger)
+    Application.put_env(:mave_core, :flow_storage_adapter, Storage)
+    owner = self()
+
+    Application.put_env(:mave_core, :cdn_cache_purger, fn _, _, keys ->
+      send(owner, {:purged, keys})
+      :ok
+    end)
+
+    Storage.reset!()
+
+    on_exit(fn ->
+      Application.put_env(:mave_core, :flow_storage_adapter, previous)
+      Application.put_env(:mave_core, :cdn_cache_purger, previous_purger)
+      Storage.reset!()
+    end)
+
+    :ok
+  end
+
+  test "opaque rendition IDs resolve to stored HLS paths and absent external audio is omitted" do
+    embed = embed_fixture()
+    legacy_rendition(embed)
+    audio_track(embed, "audio.mp3")
+
+    assert :ok = Assets.refresh_hls_master_playlist(embed)
+    body = master(embed)
+    assert body =~ "h264_hd_hls/playlist.m3u8"
+    refute body =~ String.duplicate("a", 32)
+    refute body =~ "TYPE=AUDIO"
+    refute body =~ "AUDIO=\"audio\""
+    assert_received {:purged, ["abcdefghij/playlist.m3u8"]}
+  end
+
+  test "legacy replacement versions retain their version prefix and real alternate audio" do
+    embed = embed_fixture()
+    old = embed.asset.current_video
+    earlier = DateTime.add(old.inserted_at, -60, :second)
+    Repo.insert!(%Video{asset_id: old.asset_id, inserted_at: earlier})
+    legacy_rendition(embed)
+    audio_track(embed, "alternate.mp3")
+    put(embed, "v1/alternate_hls/playlist.m3u8", "#EXTM3U\n")
+
+    assert :ok = Assets.refresh_hls_master_playlist(embed)
+    body = master(embed, "v1/")
+    assert body =~ "h264_hd_hls/playlist.m3u8"
+    assert body =~ ~s(URI="alternate_hls/playlist.m3u8")
+    refute body =~ "abcdefghij/v1/h264"
+  end
+
+  test "modern custom rendition paths are preserved" do
+    embed = embed_fixture()
+    rendition(embed, "#{embed.hash}/custom/video.m3u8")
+
+    assert :ok = Assets.refresh_hls_master_playlist(embed)
+    assert master(embed) =~ "custom/video.m3u8"
+  end
+
+  test "opaque subtitle IDs resolve to the subtitle file when rebuilding its playlist" do
+    embed = embed_fixture()
+    legacy_rendition(embed)
+
+    Repo.insert!(%Subtitle{
+      video_id: embed.asset.current_video.id,
+      language: "fr",
+      path: String.duplicate("b", 32)
+    })
+
+    assert :ok = Assets.refresh_hls_master_playlist(embed)
+
+    assert {:ok, body} =
+             Storage.get("space-abcde", "#{embed.hash}/subtitle_fr_hls/playlist.m3u8", "eu")
+
+    assert body =~ "../subtitle_fr.vtt"
+    refute body =~ String.duplicate("b", 32)
+  end
+
+  test "storage failures do not silently remove audio or overwrite the master" do
+    embed = embed_fixture()
+    legacy_rendition(embed)
+    audio_track(embed, "audio.mp3")
+    put(embed, "playlist.m3u8", "original")
+    Application.put_env(:mave_core, :flow_storage_adapter, UnavailableStorage)
+
+    assert {:error, {:audio_playlist_unavailable, :timeout}} =
+             Assets.refresh_hls_master_playlist(embed)
+
+    assert master(embed) == "original"
+  end
+
+  test "missing separate audio does not mute a video-only variant" do
+    embed = embed_fixture()
+    legacy_rendition(embed)
+    audio_track(embed, "audio.mp3")
+
+    put(
+      embed,
+      "h264_hd_hls/init.mp4",
+      box("moov", box("trak", box("mdia", box("hdlr", <<0::64, "vide">>))))
+    )
+
+    put(embed, "playlist.m3u8", "original")
+
+    assert {:error, {:missing_audio_playlist, "audio_hls/playlist.m3u8"}} =
+             Assets.refresh_hls_master_playlist(embed)
+
+    assert master(embed) == "original"
+  end
+
+  defp embed_fixture do
+    space = Repo.insert!(%Space{hash: "abcde", region: "eu"})
+    asset = Repo.insert!(%Asset{space_id: space.id})
+    video = Repo.insert!(%Video{asset_id: asset.id, duration: 10.0})
+    asset |> Ecto.Changeset.change(current_video_id: video.id) |> Repo.update!()
+
+    Repo.insert!(%Embed{space_id: space.id, asset_id: asset.id, hash: "abcdefghij"})
+    |> Repo.preload([:space, asset: [:current_video]])
+  end
+
+  defp legacy_rendition(embed) do
+    put(
+      embed,
+      "h264_hd_hls/init.mp4",
+      box("moov", box("trak", box("mdia", box("hdlr", <<0::64, "soun">>))))
+    )
+
+    rendition(embed, String.duplicate("a", 32))
+  end
+
+  defp box(type, body), do: <<byte_size(body) + 8::32, type::binary, body::binary>>
+
+  defp rendition(embed, key) do
+    Repo.insert!(%Rendition{
+      video_id: embed.asset.current_video.id,
+      rendition_key: key,
+      type: :video,
+      codec: :h264,
+      container: :hls,
+      size: :hd,
+      progress: 100.0
+    })
+  end
+
+  defp audio_track(embed, filename) do
+    Repo.insert!(%AudioTrack{
+      video_id: embed.asset.current_video.id,
+      filename: filename,
+      codec: "mp3",
+      default: true
+    })
+  end
+
+  defp put(embed, path, body) do
+    Storage.put("space-#{embed.space.hash}", "#{embed.hash}/#{path}", body, "text/plain", "eu")
+  end
+
+  defp master(embed, prefix \\ "") do
+    {:ok, body} =
+      Storage.get("space-#{embed.space.hash}", "#{embed.hash}/#{prefix}playlist.m3u8", "eu")
+
+    body
+  end
+end