membrane_rtc_engine_ex_webrtc

defmodule Membrane.RTC.Engine.Endpoint.ExWebRTC.SubscriptionManager do
  @moduledoc """
  Manages track subscriptions for ExWebRTC endpoints in both auto and manual modes.
  """

  use Bunch.Access

  alias Membrane.RTC.Engine
  alias Membrane.RTC.Engine.{Endpoint, Track}
  alias Membrane.RTC.Engine.Endpoint.ExWebRTC.Track, as: EndpointTrack

  @type subscribe_mode :: :auto | :manual

  @type t :: %__MODULE__{
          rtc_engine: Endpoint.id(),
          subscribe_mode: subscribe_mode(),
          subscribed_tracks: MapSet.t(Track.id()),
          subscribed_endpoints: MapSet.t(Endpoint.id())
        }
  @enforce_keys [:rtc_engine, :subscribe_mode]

  defstruct @enforce_keys ++ [subscribed_tracks: MapSet.new(), subscribed_endpoints: MapSet.new()]

  @doc """
  Creates a new subscription manager.
  """
  @spec new(Endpoint.id(), subscribe_mode()) :: t()
  def new(rtc_engine, subscribe_mode) do
    %__MODULE__{rtc_engine: rtc_engine, subscribe_mode: subscribe_mode}
  end

  @doc """
  Handles peer subscription in manual mode.
  """
  @spec subscribe_endpoint(t(), String.t()) :: {[Track.t()], t()}
  def subscribe_endpoint(
        %{rtc_engine: rtc_engine, subscribe_mode: :manual} = manager,
        endpoint_id
      ) do
    tracks_to_add =
      rtc_engine
      |> Engine.get_tracks()
      |> Enum.filter(fn t -> t.origin == endpoint_id end)

    manager = %{
      manager
      | subscribed_endpoints: MapSet.put(manager.subscribed_endpoints, endpoint_id)
    }

    {tracks_to_add, manager}
  end

  def subscribe_endpoint(%{subscribe_mode: :auto} = manager, _endpoint_id) do
    {[], manager}
  end

  @doc """
  Handles track subscription in manual mode.
  """
  @spec subscribe_tracks(t(), [Track.id()]) :: {[Track.t()], t()}
  def subscribe_tracks(%{rtc_engine: rtc_engine, subscribe_mode: :manual} = manager, track_ids) do
    tracks_to_add =
      rtc_engine
      |> Engine.get_tracks()
      |> Enum.filter(&Enum.member?(track_ids, &1.id))

    subscribed_tracks = Enum.map(tracks_to_add, fn t -> t.id end)

    manager = %{
      manager
      | subscribed_tracks: MapSet.union(manager.subscribed_tracks, MapSet.new(subscribed_tracks))
    }

    {tracks_to_add, manager}
  end

  def subscribe_tracks(%{subscribe_mode: :auto} = manager, _track_ids) do
    {[], manager}
  end

  @doc """
  Updates known tracks and returns filtered tracks for subscription.
  """
  @spec handle_new_tracks(t(), [Track.t()]) :: {map(), t()}
  def handle_new_tracks(manager, new_tracks) do
    new_tracks_map =
      Map.new(new_tracks, &{&1.id, %EndpointTrack{status: :pending, engine_track: &1}})

    filtered_tracks = filter_subscribed_new_tracks(manager, new_tracks_map)

    {filtered_tracks, manager}
  end

  @doc """
  Removes tracks from the manager.
  """
  @spec remove_tracks(t(), [Track.t()]) :: t()
  def remove_tracks(manager, tracks) do
    track_ids = Enum.map(tracks, & &1.id)

    manager
    |> update_in([:subscribed_tracks], &MapSet.difference(&1, MapSet.new(track_ids)))
  end

  defp filter_subscribed_new_tracks(%{subscribe_mode: :auto}, new_tracks), do: new_tracks

  defp filter_subscribed_new_tracks(%{subscribe_mode: :manual} = manager, new_tracks) do
    new_tracks
    |> Enum.filter(fn {id, %EndpointTrack{engine_track: t}} ->
      MapSet.member?(manager.subscribed_endpoints, t.origin) or
        MapSet.member?(manager.subscribed_tracks, id)
    end)
    |> Map.new()
  end
end