tribes-plugin-sender/test/sender/playback_api_test.exs

defmodule TribeOne.TribesPlugin.Sender.PlaybackAPITest do
  use Tribes.PluginTest.PageCase, plugin: TribeOne.TribesPlugin.Sender.Plugin

  alias TribeOne.TribesPlugin.Sender.{Stats, Streaming}

  @ash_opts [authorize?: false, context: %{ash_nostr_sync_skip_publish: true}]

  setup do
    :ok = Stats.reset()
  end

  test "returns playback metadata for a live stream", %{conn: conn} do
    %{stream: stream, generation: generation} = live_stream_fixture()

    conn = get(conn, "/plugins-api/tribe-one-sender/streams/#{stream.id}/playback")

    assert decoded_response(conn, 200) == %{
             "generation" => %{
               "id" => generation.id,
               "started_at" => DateTime.to_iso8601(generation.started_at),
               "status" => "live"
             },
             "player" => "videojs-html-v10",
             "sources" => [
               %{
                 "height" => 720,
                 "rendition" => "source",
                 "src" => "/sender/hls/#{stream.id}/#{generation.id}/source/media.m3u8",
                 "type" => "application/vnd.apple.mpegurl",
                 "video_bitrate" => 2_500_000,
                 "width" => 1280
               }
             ],
             "stream" => %{
               "id" => stream.id,
               "latency_mode" => "hls",
               "slug" => stream.slug,
               "title" => stream.title
             }
           }
  end

  test "returns not_live when no active generation exists", %{conn: conn} do
    {:ok, stream} =
      Streaming.create_stream(
        %{slug: "quiet", title: "Quiet stream"},
        @ash_opts
      )

    conn = get(conn, "/plugins-api/tribe-one-sender/streams/#{stream.id}/playback")

    assert decoded_response(conn, 404) == %{"error" => "not_live"}
  end

  test "serves standalone HLS files from the sender media path", %{conn: conn} do
    root =
      Path.join(System.tmp_dir!(), "sender-hls-api-test-#{System.unique_integer([:positive])}")

    path = Path.join([root, "stream", "generation", "source"])
    File.mkdir_p!(path)
    File.write!(Path.join(path, "media.m3u8"), "#EXTM3U\n")

    previous = Application.get_env(:tribe_one_sender, TribeOne.TribesPlugin.Sender.HLS, [])

    Application.put_env(
      :sender,
      TribeOne.TribesPlugin.Sender.HLS,
      Keyword.put(previous, :spool_root, root)
    )

    on_exit(fn ->
      Application.put_env(:tribe_one_sender, TribeOne.TribesPlugin.Sender.HLS, previous)
    end)

    conn =
      conn
      |> Map.put(:remote_ip, {203, 0, 113, 20})
      |> get("/sender/hls/stream/generation/source/media.m3u8")

    assert response(conn, 200) == "#EXTM3U\n"
    assert get_resp_header(conn, "cache-control") == ["no-store"]
    assert [content_type] = get_resp_header(conn, "content-type")
    assert content_type =~ "application/vnd.apple.mpegurl"
    assert Stats.viewer_count("stream", "generation") == 1
  end

  test "counts standalone HLS playlist vsids as viewer sessions", %{conn: conn} do
    root =
      Path.join(System.tmp_dir!(), "sender-hls-vsid-test-#{System.unique_integer([:positive])}")

    path = Path.join([root, "stream", "generation", "source"])
    File.mkdir_p!(path)
    File.write!(Path.join(path, "media.m3u8"), "#EXTM3U\n")

    previous = Application.get_env(:tribe_one_sender, TribeOne.TribesPlugin.Sender.HLS, [])

    Application.put_env(
      :sender,
      TribeOne.TribesPlugin.Sender.HLS,
      Keyword.put(previous, :spool_root, root)
    )

    on_exit(fn ->
      Application.put_env(:tribe_one_sender, TribeOne.TribesPlugin.Sender.HLS, previous)
    end)

    conn
    |> Map.put(:remote_ip, {203, 0, 113, 20})
    |> get("/sender/hls/stream/generation/source/media.m3u8?vsid=browser-1")

    conn
    |> Map.put(:remote_ip, {203, 0, 113, 20})
    |> get("/sender/hls/stream/generation/source/media.m3u8?vsid=browser-2")

    assert Stats.viewer_count("stream", "generation") == 2
  end

  test "serves trusted proxy HLS requests without counting internal viewers", %{conn: conn} do
    root =
      Path.join(System.tmp_dir!(), "sender-hls-proxy-test-#{System.unique_integer([:positive])}")

    path = Path.join([root, "stream", "generation", "source"])
    File.mkdir_p!(path)
    File.write!(Path.join(path, "media.m3u8"), "#EXTM3U\n")

    previous = Application.get_env(:tribe_one_sender, TribeOne.TribesPlugin.Sender.HLS, [])

    Application.put_env(
      :sender,
      TribeOne.TribesPlugin.Sender.HLS,
      Keyword.put(previous, :spool_root, root)
    )

    on_exit(fn ->
      Application.put_env(:tribe_one_sender, TribeOne.TribesPlugin.Sender.HLS, previous)
    end)

    conn =
      conn
      |> Map.put(:remote_ip, {127, 0, 0, 1})
      |> put_req_header("x-forwarded-for", "203.0.113.20")
      |> get("/sender/hls/stream/generation/source/media.m3u8?vsid=browser-1")

    assert response(conn, 200) == "#EXTM3U\n"
    assert Stats.viewer_count("stream", "generation") == 0
  end

  test "counts player heartbeats as viewer sessions" do
    stream_id = Ash.UUID.generate()
    generation_id = Ash.UUID.generate()

    conn =
      Plug.Test.conn(
        "POST",
        "/plugins-api/tribe-one-sender/player-events",
        JSON.encode!(%{
          "event" => "heartbeat",
          "stream_id" => stream_id,
          "generation_id" => generation_id,
          "viewer_session_id" => "browser-1"
        })
      )
      |> Map.put(:remote_ip, {203, 0, 113, 30})
      |> put_req_header("content-type", "application/json")
      |> TribeOne.TribesPlugin.SenderWeb.StreamingAPIPlug.call([])

    assert decoded_response(conn, 202) == %{"ok" => true, "viewer_count" => 1}
    assert Stats.viewer_count(stream_id, generation_id) == 1
  end

  test "serves trusted proxy player events without counting internal viewers" do
    stream_id = Ash.UUID.generate()
    generation_id = Ash.UUID.generate()

    conn =
      Plug.Test.conn(
        "POST",
        "/plugins-api/tribe-one-sender/player-events",
        JSON.encode!(%{
          "event" => "heartbeat",
          "stream_id" => stream_id,
          "generation_id" => generation_id,
          "viewer_session_id" => "browser-1"
        })
      )
      |> Map.put(:remote_ip, {127, 0, 0, 1})
      |> put_req_header("content-type", "application/json")
      |> put_req_header("x-forwarded-for", "203.0.113.30")
      |> TribeOne.TribesPlugin.SenderWeb.StreamingAPIPlug.call([])

    assert decoded_response(conn, 202) == %{"ok" => true}
    assert Stats.viewer_count(stream_id, generation_id) == 0
  end

  test "does not count localhost HLS access as a fallback viewer", %{conn: conn} do
    root =
      Path.join(System.tmp_dir!(), "sender-hls-local-test-#{System.unique_integer([:positive])}")

    path = Path.join([root, "stream", "generation", "source"])
    File.mkdir_p!(path)
    File.write!(Path.join(path, "media.m3u8"), "#EXTM3U\n")

    previous = Application.get_env(:tribe_one_sender, TribeOne.TribesPlugin.Sender.HLS, [])

    Application.put_env(
      :sender,
      TribeOne.TribesPlugin.Sender.HLS,
      Keyword.put(previous, :spool_root, root)
    )

    on_exit(fn ->
      Application.put_env(:tribe_one_sender, TribeOne.TribesPlugin.Sender.HLS, previous)
    end)

    conn = get(conn, "/sender/hls/stream/generation/source/media.m3u8")

    assert response(conn, 200) == "#EXTM3U\n"
    assert Stats.viewer_count("stream", "generation") == 0
  end

  test "does not cache missing standalone HLS files", %{conn: conn} do
    conn = get(conn, "/sender/hls/missing/future.ts")

    assert response(conn, 404) == "Not found"
    assert get_resp_header(conn, "cache-control") == ["no-store"]
  end

  defp live_stream_fixture do
    now = DateTime.utc_now() |> DateTime.truncate(:second)

    {:ok, stream} =
      Streaming.create_stream(
        %{slug: "live", title: "Live stream"},
        @ash_opts
      )

    {:ok, endpoint} =
      Streaming.upsert_media_endpoint(
        %{
          type: :external_origin,
          display_name: "Origin"
        },
        @ash_opts
      )

    {:ok, generation} =
      Streaming.create_stream_generation(
        %{
          stream_id: stream.id,
          status: :live,
          started_at: now,
          primary_origin_endpoint_id: endpoint.id
        },
        @ash_opts
      )

    {:ok, _rendition} =
      Streaming.upsert_rendition(
        %{
          generation_id: generation.id,
          name: "source",
          width: 1280,
          height: 720,
          video_bitrate: 2_500_000,
          playlist_path: "/sender/hls/#{stream.id}/#{generation.id}/source/media.m3u8",
          status: :ready
        },
        @ash_opts
      )

    %{stream: stream, generation: generation, endpoint: endpoint}
  end

  defp decoded_response(conn, status) do
    assert conn.status == status
    JSON.decode!(conn.resp_body)
  end
end