Terminate interrupted streams (#75)

lib/algora/application.ex

@@ -55,6 +55,7 @@ defmodule Algora.Application do
         start: {Membrane.RTMP.Source.TcpServer, :start_link, [tcp_server_options]}
       },
       Algora.Stargazer,
+      Algora.Terminate,
       ExMarcel.TableWrapper,
       Algora.Youtube.Chat.Supervisor
       # Start a worker by calling: Algora.Worker.start_link(arg)

lib/algora/library.ex

@@ -226,6 +226,43 @@ defmodule Algora.Library do
     |> Repo.one()
   end
 
+  def terminate_stream(video_id) do
+    video = Repo.get!(Video, video_id)
+
+    resp = Finch.build(:get, "#{video.url_root}/g3cFdmlkZW8.m3u8") |> Finch.request(Algora.Finch)
+
+    with {:ok, %Finch.Response{status: 200, body: body}} <- resp,
+         {:ok, duration} <- get_duration(video) do
+      if !String.ends_with?(body, "#EXT-X-ENDLIST\n") do
+        Storage.upload(
+          String.trim_trailing(body) <> "\n#EXT-X-ENDLIST\n",
+          "#{video.uuid}/g3cFdmlkZW8.m3u8",
+          content_type: "application/x-mpegURL"
+        )
+      end
+
+      video
+      |> change()
+      |> put_change(:duration, duration)
+      |> Repo.update()
+    else
+      _missing_manifest ->
+        video
+        |> change()
+        |> put_change(:corrupted, true)
+        |> Repo.update()
+    end
+  end
+
+  def terminate_interrupted_streams() do
+    from(v in Video,
+      where: v.duration == 0 and v.is_live == false and v.format == :hls and v.corrupted == false,
+      select: v.id
+    )
+    |> Repo.all()
+    |> Enum.each(&terminate_stream/1)
+  end
+
   def toggle_streamer_live(%Video{} = video, is_live) do
     video = get_video!(video.id)
     user = Accounts.get_user!(video.user_id)

lib/algora/library/video.ex

@@ -17,6 +17,7 @@ defmodule Algora.Library.Video do
     field :description, :string
     field :type, Ecto.Enum, values: [vod: 1, livestream: 2]
     field :format, Ecto.Enum, values: [mp4: 1, hls: 2, youtube: 3]
+    field :corrupted, :boolean, default: false
     field :is_live, :boolean, default: false
     field :thumbnail_url, :string
     field :vertical_thumbnail_url, :string

lib/algora/terminate.ex

@@ -0,0 +1,26 @@
+defmodule Algora.Terminate do
+  use GenServer
+
+  @terminate_interval :timer.hours(1)
+
+  def start_link(_) do
+    GenServer.start_link(__MODULE__, %{}, name: __MODULE__)
+  end
+
+  @impl true
+  def init(state) do
+    schedule_terminate()
+    {:ok, state}
+  end
+
+  @impl true
+  def handle_info(:terminate, state) do
+    Algora.Library.terminate_interrupted_streams()
+    schedule_terminate()
+    {:noreply, state}
+  end
+
+  defp schedule_terminate() do
+    Process.send_after(self(), :terminate, @terminate_interval)
+  end
+end

priv/repo/migrations/20240831185906_add_corrupted_to_video.exs

@@ -0,0 +1,9 @@
+defmodule Algora.Repo.Local.Migrations.AddCorruptedToVideo do
+  use Ecto.Migration
+
+  def change do
+    alter table(:videos) do
+      add :corrupted, :boolean, default: false
+    end
+  end
+end