janus_ws

defmodule Janus.WS do
  @moduledoc File.read!("README.md")
  use WebSockex

  defstruct [:registry, pending_txs: %{}]

  @type client :: WebSockex.client()
  @type session_id :: integer
  @type handle_id :: integer
  @type tx_id :: String.t()
  @type send_result ::
          {:ok, tx_id}
          | {:error,
             %WebSockex.FrameEncodeError{}
             | %WebSockex.ConnError{}
             | %WebSockex.NotConnectedError{}
             | %WebSockex.InvalidFrameError{}}

  @doc """
  Example:

      Registry.start_link(keys: :duplicate, name: Janus.WS.Session.Registry)
      Janus.WS.start_link(url: "ws://localhost:8188", registry: Janus.WS.Session.Registry)

  """
  def start_link(opts) do
    websockex_opts = [
      name: opts[:name],
      extra_headers: [{"Sec-WebSocket-Protocol", "janus-protocol"}],
      debug: opts[:debug]
    ]

    websockex_opts = Enum.reject(websockex_opts, fn {_k, v} -> is_nil(v) end)

    WebSockex.start_link(
      opts[:url],
      __MODULE__,
      opts[:state] || %__MODULE__{registry: opts[:registry]},
      websockex_opts
    )
  end

  @impl true
  def handle_frame(
        {:text, msg},
        %__MODULE__{pending_txs: pending_txs, registry: registry} = state
      ) do
    case Jason.decode!(msg) do
      %{"session_id" => session_id} = msg ->
        if registry, do: _broadcast(registry, session_id, msg)
        {:ok, state}

      %{"transaction" => transaction} = msg ->
        {maybe_from, pending_txs} = Map.pop(pending_txs, transaction)
        if maybe_from, do: send(maybe_from, {:janus_ws, msg})
        {:ok, %{state | pending_txs: pending_txs}}
    end
  end

  @impl true
  def handle_cast({cmd, tx_id, pid}, %__MODULE__{pending_txs: pending_txs} = state) do
    msg =
      case cmd do
        :create_session -> %{"janus" => "create"}
        :info -> %{"janus" => "info"}
      end

    msg = Map.put(msg, "transaction", tx_id)

    {:reply, {:text, Jason.encode!(msg)},
     %{state | pending_txs: Map.put(pending_txs, tx_id, pid)}}
  end

  @doc "To fetch info about the janus instance"
  @spec info(client) :: send_result
  @spec info(client, pid) :: send_result
  def info(client, pid \\ self()) do
    tx_id = tx_id()
    :ok = WebSockex.cast(client, {:info, tx_id, pid})
    {:ok, tx_id}
  end

  @doc "To create a janus session"
  @spec create_session(client) :: send_result
  @spec create_session(client, pid) :: send_result
  def create_session(client, pid \\ self()) do
    tx_id = tx_id()
    :ok = WebSockex.cast(client, {:create_session, tx_id, pid})
    {:ok, tx_id}
  end

  @doc "To destroy a janus session"
  @spec destroy_session(client, session_id) :: send_result
  def destroy_session(client, session_id) do
    _send(client, %{"janus" => "destroy", "session_id" => session_id})
  end

  @doc "To attach a plugin to a janus session"
  @spec attach(client, session_id, String.t()) :: send_result
  def attach(client, session_id, plugin) do
    _send(client, %{"janus" => "attach", "session_id" => session_id, "plugin" => plugin})
  end

  @doc "To detach a plugin from a janus session"
  @spec detach(client, session_id, handle_id) :: send_result
  def detach(client, session_id, handle_id) do
    _send(client, %{"janus" => "detach", "session_id" => session_id, "handle_id" => handle_id})
  end

  @doc "To send a trickle candidate for a session"
  @spec send_trickle_candidate(client, session_id, handle_id, [map]) :: send_result
  def send_trickle_candidate(client, session_id, handle_id, candidates)
      when is_list(candidates) do
    message = %{
      "janus" => "trickle",
      "session_id" => session_id,
      "handle_id" => handle_id,
      "candidates" => candidates
    }

    _send(client, message)
  end

  @spec send_trickle_candidate(client, session_id, handle_id, map) :: send_result
  def send_trickle_candidate(client, session_id, handle_id, candidate) when is_map(candidate) do
    message = %{
      "janus" => "trickle",
      "session_id" => session_id,
      "handle_id" => handle_id,
      "candidate" => candidate
    }

    _send(client, message)
  end

  @doc "To send a message to a handle in a session"
  @spec send_message(client, session_id, handle_id, map) :: send_result
  def send_message(client, session_id, handle_id, data) do
    msg =
      %{
        "janus" => "message",
        "session_id" => session_id,
        "handle_id" => handle_id
      }
      |> Map.merge(Map.take(data, ["body", "jsep"]))

    _send(client, msg)
  end

  @doc "To send a keepalive for a session"
  @spec send_keepalive(client, session_id) :: send_result
  def send_keepalive(client, session_id) do
    _send(client, %{"janus" => "keepalive", "session_id" => session_id})
  end

  @spec _send(client, map) :: send_result
  defp _send(client, msg) do
    tx_id = tx_id()
    msg = Map.put(msg, "transaction", tx_id)

    with :ok <- WebSockex.send_frame(client, {:text, Jason.encode!(msg)}) do
      {:ok, tx_id}
    end
  end

  @spec _broadcast(module, session_id(), map) :: :ok
  defp _broadcast(registry, session_id, message) when not is_nil(registry) do
    Registry.dispatch(registry, session_id, fn entries ->
      Enum.each(entries, fn {pid, _} ->
        send(pid, {:janus_ws, message})
      end)
    end)
  end

  @spec tx_id :: tx_id
  defp tx_id do
    8
    |> :crypto.strong_rand_bytes()
    |> Base.encode64(padding: false)
  end
end