Merge branch 'feature/presence'

assets/js/app.js

@@ -195,6 +195,7 @@ window.addEventListener("phx:page-loading-stop", info => topbar.hide())
 window.addEventListener("phx:page-loading-stop", routeUpdated)
 
 window.addEventListener("js:exec", e => e.target[e.detail.call](...e.detail.args))
+window.addEventListener("phx:remove-el", e => document.getElementById(e.detail.id).remove())
 
 // connect if there are any LiveViews on the page
 liveSocket.getSocket().onOpen(() => execJS("#connection-status", "js-hide"))

lib/live_beats/accounts.ex

@@ -21,6 +21,10 @@ defmodule LiveBeats.Accounts do
     Repo.all(from u in User, limit: ^Keyword.fetch!(opts, :limit))
   end
 
+  def get_users_map(user_ids) when is_list(user_ids) do
+    Repo.all(from u in User, where: u.id in ^user_ids, select: {u.id, u})
+  end
+
   def lists_users_by_active_profile(id, opts) do
     Repo.all(
       from u in User, where: u.active_profile_user_id == ^id, limit: ^Keyword.fetch!(opts, :limit)

lib/live_beats/application.ex

@@ -17,8 +17,16 @@ defmodule LiveBeats.Application do
       LiveBeatsWeb.Telemetry,
       # Start the PubSub system
       {Phoenix.PubSub, name: LiveBeats.PubSub},
+      # start presence
+      LiveBeatsWeb.Presence,
+      {Phoenix.Presence.Client,
+       client: LiveBeats.PresenceClient,
+       pubsub: LiveBeats.PubSub,
+       presence: LiveBeatsWeb.Presence,
+       name: PresenceClient},
       # Start the Endpoint (http/https)
       LiveBeatsWeb.Endpoint
+
       # Start a worker by calling: LiveBeats.Worker.start_link(arg)
       # {LiveBeats.Worker, arg}
     ]

lib/live_beats/presence/phoenix_presence_client.ex

@@ -0,0 +1,203 @@
+defmodule Phoenix.Presence.Client do
+  use GenServer
+
+  @callback init(state :: term) :: {:ok, new_state :: term}
+  @callback handle_join(topic :: String.t(), key :: String.t(), meta :: [map()], state :: term) ::
+              {:ok, term}
+  @callback handle_leave(topic :: String.t(), key :: String.t(), meta :: [map()], state :: term) ::
+              {:ok, term}
+
+  @doc """
+  TODO
+
+  ## Options
+
+    * `:pubsub` - The required name of the pubsub server
+    * `:presence` - The required name of the presence module
+    * `:client` - The required callback module
+  """
+  def start_link(opts) do
+    case Keyword.fetch(opts, :name) do
+      {:ok, name} ->
+        GenServer.start_link(__MODULE__, opts, name: name)
+
+      :error ->
+        GenServer.start_link(__MODULE__, opts)
+    end
+  end
+
+  def track(pid \\ PresenceClient, topic, key, meta) do
+    GenServer.call(pid, {:track, self(), topic, to_string(key), meta})
+  end
+
+  def untrack(pid \\ PresenceClient, topic, key) do
+    GenServer.call(pid, {:untrack, self(), topic, to_string(key)})
+  end
+
+  def init(opts) do
+    client = Keyword.fetch!(opts, :client)
+    {:ok, client_state} = client.init(%{})
+
+    state = %{
+      topics: %{},
+      client: client,
+      pubsub: Keyword.fetch!(opts, :pubsub),
+      presence_mod: Keyword.fetch!(opts, :presence),
+      client_state: client_state
+    }
+
+    {:ok, state}
+  end
+
+  def handle_info(%{topic: topic, event: "presence_diff", payload: diff}, state) do
+    {:noreply, merge_diff(state, topic, diff)}
+  end
+
+  def handle_call(:state, _from, state) do
+    {:reply, state, state}
+  end
+
+  def handle_call({:track, pid, topic, key, meta}, _from, state) do
+    {:reply, :ok, track_pid(state, pid, topic, key, meta)}
+  end
+
+  def handle_call({:untrack, pid, topic, key}, _from, state) do
+    {:reply, :ok, untrack_pid(state, pid, topic, key)}
+  end
+
+  defp track_pid(state, pid, topic, key, meta) do
+    # presences are handled when the presence_diff event is received
+    case Map.fetch(state.topics, topic) do
+      {:ok, _topic_content} ->
+        state.presence_mod.track(pid, topic, key, meta)
+        state
+
+      :error ->
+        # subscribe to topic we weren't yet tracking
+        Phoenix.PubSub.subscribe(state.pubsub, topic)
+        state.presence_mod.track(pid, topic, key, meta)
+        state
+    end
+  end
+
+  defp untrack_pid(state, pid, topic, key) do
+    if Map.has_key?(state.topics, topic) do
+      state.presence_mod.untrack(pid, topic, key)
+    else
+      state
+    end
+  end
+
+  defp merge_diff(state, topic, %{leaves: leaves, joins: joins}) do
+    # add new topic if needed
+    updated_state =
+      if Map.has_key?(state.topics, topic) do
+        state
+      else
+        update_topics_state(:add_new_topic, state, topic)
+      end
+
+    # merge diff into state.topics
+    {updated_state, _topic} = Enum.reduce(joins, {updated_state, topic}, &handle_join/2)
+    {updated_state, _topic} = Enum.reduce(leaves, {updated_state, topic}, &handle_leave/2)
+
+    # if no more presences for given topic, unsubscribe and remove topic
+    if topic_presences_count(updated_state, topic) == 0 do
+      Phoenix.PubSub.unsubscribe(state.pubsub, topic)
+      update_topics_state(:remove_topic, updated_state, topic)
+    else
+      updated_state
+    end
+  end
+
+  defp handle_join({joined_key, meta}, {state, topic}) do
+    joined_meta = Map.get(meta, :metas, [])
+
+    updated_state =
+      update_topics_state(:add_new_presence_or_metas, state, topic, joined_key, joined_meta)
+
+    {:ok, updated_client_state} =
+      state.client.handle_join(topic, joined_key, meta, state.client_state)
+
+    updated_state = Map.put(updated_state, :client_state, updated_client_state)
+
+    {updated_state, topic}
+  end
+
+  defp handle_leave({left_key, meta}, {state, topic}) do
+    updated_state = update_topics_state(:remove_presence_or_metas, state, topic, left_key, meta)
+
+    {:ok, updated_client_state} =
+      state.client.handle_leave(topic, left_key, meta, state.client_state)
+
+    updated_state = Map.put(updated_state, :client_state, updated_client_state)
+
+    {updated_state, topic}
+  end
+
+  defp update_topics_state(:add_new_topic, %{topics: topics} = state, topic) do
+    updated_topics = Map.put_new(topics, topic, %{})
+    Map.put(state, :topics, updated_topics)
+  end
+
+  defp update_topics_state(:remove_topic, %{topics: topics} = state, topic) do
+    updated_topics = Map.delete(topics, topic)
+    Map.put(state, :topics, updated_topics)
+  end
+
+  defp update_topics_state(
+         :add_new_presence_or_metas,
+         %{topics: topics} = state,
+         topic,
+         key,
+         new_metas
+       ) do
+    topic_info = topics[topic]
+
+    updated_topic =
+      case Map.fetch(topic_info, key) do
+        # existing presence, add new metas
+        {:ok, existing_metas} ->
+          remaining_metas = new_metas -- existing_metas
+          updated_metas = existing_metas ++ remaining_metas
+          Map.put(topic_info, key, updated_metas)
+
+        :error ->
+          # there are no presences for that key
+          Map.put(topic_info, key, new_metas)
+      end
+
+    updated_topics = Map.put(topics, topic, updated_topic)
+
+    Map.put(state, :topics, updated_topics)
+  end
+
+  defp update_topics_state(
+         :remove_presence_or_metas,
+         %{topics: topics} = state,
+         topic,
+         key,
+         deleted_metas
+       ) do
+    topic_info = topics[topic]
+
+    state_metas = Map.get(topic_info, key, [])
+    remaining_metas = state_metas -- Map.get(deleted_metas, :metas, [])
+
+    updated_topic =
+      case remaining_metas do
+        # delete presence
+        [] -> Map.delete(topic_info, key)
+        # delete metas
+        _ -> Map.put(topic_info, key, remaining_metas)
+      end
+
+    updated_topics = Map.put(topics, topic, updated_topic)
+
+    Map.put(state, :topics, updated_topics)
+  end
+
+  defp topic_presences_count(state, topic) do
+    map_size(state.topics[topic])
+  end
+end

lib/live_beats/presence/presence_client.ex

@@ -0,0 +1,57 @@
+defmodule LiveBeats.PresenceClient do
+  @behaviour Phoenix.Presence.Client
+
+  @presence LiveBeatsWeb.Presence
+  @pubsub LiveBeats.PubSub
+
+  def list(topic) do
+    @presence.list(topic)
+  end
+
+  @impl Phoenix.Presence.Client
+  def init(_opts) do
+    # user-land state
+    {:ok, %{}}
+  end
+
+  @impl Phoenix.Presence.Client
+  def handle_join(topic, _key, presence, state) do
+    active_users_topic =
+      topic
+      |> profile_identifier()
+      |> active_users_topic()
+
+    Phoenix.PubSub.local_broadcast(
+      @pubsub,
+      active_users_topic,
+      {__MODULE__, %{user_joined: presence}}
+    )
+
+    {:ok, state}
+  end
+
+  @impl Phoenix.Presence.Client
+  def handle_leave(topic, _key, presence, state) do
+    active_users_topic =
+      topic
+      |> profile_identifier()
+      |> active_users_topic()
+
+    Phoenix.PubSub.local_broadcast(
+      @pubsub,
+      active_users_topic,
+      {__MODULE__, %{user_left: presence}}
+    )
+
+    {:ok, state}
+  end
+
+  defp active_users_topic(user_id) do
+    "active_users:#{user_id}"
+  end
+
+  defp profile_identifier(topic) do
+    "active_profile:" <> identifier = topic
+    identifier
+  end
+end

lib/live_beats_web/channels/presence.ex

@@ -5,20 +5,30 @@ defmodule LiveBeatsWeb.Presence do
   See the [`Phoenix.Presence`](http://hexdocs.pm/phoenix/Phoenix.Presence.html)
   docs for more details.
   """
-  use Phoenix.Presence, otp_app: :live_beats,
-                        pubsub_server: LiveBeats.PubSub
+  use Phoenix.Presence,
+    otp_app: :live_beats,
+    pubsub_server: LiveBeats.PubSub
 
   import Phoenix.LiveView.Helpers
   import LiveBeatsWeb.LiveHelpers
+  @pubsub LiveBeats.PubSub
+
+  alias LiveBeats.Accounts
 
   def listening_now(assigns) do
     ~H"""
     <!-- users -->
     <div class="px-4 mt-6 sm:px-6 lg:px-8">
-      <h2 class="text-gray-500 text-xs font-medium uppercase tracking-wide">Who's Listening</h2>
-      <ul role="list" class="grid grid-cols-1 gap-4 sm:gap-4 sm:grid-cols-2 xl:grid-cols-5 mt-3" x-max="1">
+      <h2 class="text-gray-500 text-xs font-medium uppercase tracking-wide">Here now</h2>
+      <ul
+        id="listening-now"
+        phx-update="prepend"
+        role="list"
+        x-max="1"
+        class="grid grid-cols-1 gap-4 sm:gap-4 sm:grid-cols-2 xl:grid-cols-5 mt-3"
+      >
         <%= for presence <- @presences do %>
-          <li class="relative col-span-1 flex shadow-sm rounded-md overflow-hidden">
+          <li id={"presence-#{presence.id}"} class="relative col-span-1 flex shadow-sm rounded-md overflow-hidden">
             <.link navigate={profile_path(presence)} class="flex-1 flex items-center justify-between border-t border-r border-b border-gray-200 bg-white rounded-r-md truncate">
               <img class="w-10 h-10 flex-shrink-0 flex items-center justify-center rounded-l-md bg-purple-600" src={presence.avatar_url} alt="">
               <div class="flex-1 flex items-center justify-between text-gray-900 text-sm font-medium hover:text-gray-600 pl-3">
@@ -31,4 +41,24 @@ defmodule LiveBeatsWeb.Presence do
     </div>
     """
   end
+
+  def fetch(_topic, presences) do
+    users =
+      presences
+      |> Map.keys()
+      |> Accounts.get_users_map()
+      |> Enum.into(%{})
+
+    for {key, %{metas: metas}} <- presences, into: %{} do
+      {key, %{metas: metas, user: users[String.to_integer(key)]}}
+    end
+  end
+
+  def subscribe(user_id) do
+    Phoenix.PubSub.subscribe(@pubsub, topic(user_id))
+  end
+
+  defp topic(profile) do
+    "active_users:#{profile.user_id}"
+  end
 end

lib/live_beats_web/live/player_live.ex

@@ -258,6 +258,8 @@ defmodule LiveBeatsWeb.PlayerLive do
 
   def handle_info({MediaLibrary, _}, socket), do: {:noreply, socket}
 
+  def handle_info(%{event: "presence_diff"}, socket), do: {:noreply, socket}
+
   defp play_song(socket, %Song{} = song, elapsed) do
     socket
     |> push_play(song, elapsed)

lib/live_beats_web/live/profile_live.ex

@@ -87,6 +87,13 @@ defmodule LiveBeatsWeb.ProfileLive do
     if connected?(socket) do
       MediaLibrary.subscribe_to_profile(profile)
       Accounts.subscribe(current_user.id)
+      LiveBeatsWeb.Presence.subscribe(profile)
+
+      Phoenix.Presence.Client.track(
+        topic(profile.user_id),
+        current_user.id,
+        %{}
+      )
     end
 
     active_song_id =
@@ -106,7 +113,7 @@ defmodule LiveBeatsWeb.ProfileLive do
       |> list_songs()
       |> assign_presences()
 
-    {:ok, socket, temporary_assigns: [songs: []]}
+    {:ok, socket, temporary_assigns: [songs: [], presences: []]}
   end
 
   def handle_params(params, _url, socket) do
@@ -142,6 +149,16 @@ defmodule LiveBeatsWeb.ProfileLive do
     {:noreply, socket}
   end
 
+  def handle_info({LiveBeats.PresenceClient, %{user_joined: presence}}, socket) do
+    %{user: user} = presence
+    {:noreply, update(socket, :presences, &[user | &1])}
+  end
+
+  def handle_info({LiveBeats.PresenceClient, %{user_left: presence}}, socket) do
+    %{user: user} = presence
+    {:noreply, push_event(socket, "remove-el", %{id: "presence-#{user.id}"})}
+  end
+
   def handle_info({Accounts, %Accounts.Events.ActiveProfileChanged{} = event}, socket) do
     {:noreply, assign(socket, active_profile_id: event.new_profile_user_id)}
   end
@@ -242,8 +259,13 @@ defmodule LiveBeatsWeb.ProfileLive do
   end
 
   defp assign_presences(socket) do
-    users = Accounts.lists_users_by_active_profile(socket.assigns.profile.user_id, limit: 10)
-    assign(socket, presences: users)
+    presences =
+      socket.assigns.profile.user_id
+      |> topic()
+      |> LiveBeats.PresenceClient.list()
+      |> Enum.map(fn {_key, meta} -> meta.user end)
+
+    assign(socket, presences: presences)
   end
 
   defp url_text(nil), do: ""
@@ -252,4 +274,6 @@ defmodule LiveBeatsWeb.ProfileLive do
     uri = URI.parse(url_str)
     uri.host <> uri.path
   end
+
+  defp topic(user_id) when is_integer(user_id), do: "active_profile:#{user_id}"
 end

mix.exs

@@ -33,7 +33,7 @@ defmodule LiveBeats.MixProject do
   # Type `mix help deps` for examples and options.
   defp deps do
     [
-      {:phoenix, "~> 1.6.0"},
+      {:phoenix, github: "phoenixframework/phoenix", override: true},
       {:phoenix_ecto, "~> 4.4"},
       {:ecto_sql, "~> 3.6"},
       {:postgrex, ">= 0.0.0"},

mix.lock

@@ -20,12 +20,12 @@
   "jason": {:hex, :jason, "1.2.2", "ba43e3f2709fd1aa1dce90aaabfd039d000469c05c56f0b8e31978e03fa39052", [:mix], [{:decimal, "~> 1.0 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: true]}], "hexpm", "18a228f5f0058ee183f29f9eae0805c6e59d61c3b006760668d8d18ff0d12179"},
   "mime": {:hex, :mime, "2.0.2", "0b9e1a4c840eafb68d820b0e2158ef5c49385d17fb36855ac6e7e087d4b1dcc5", [:mix], [], "hexpm", "e6a3f76b4c277739e36c2e21a2c640778ba4c3846189d5ab19f97f126df5f9b7"},
   "mint": {:hex, :mint, "1.3.0", "396b3301102f7b775e103da5a20494b25753aed818d6d6f0ad222a3a018c3600", [:mix], [{:castore, "~> 0.1.0", [hex: :castore, repo: "hexpm", optional: true]}], "hexpm", "a9aac960562e43ca69a77e5176576abfa78b8398cec5543dd4fb4ab0131d5c1e"},
-  "phoenix": {:hex, :phoenix, "1.6.2", "6cbd5c8ed7a797f25a919a37fafbc2fb1634c9cdb12a4448d7a5d0b26926f005", [:mix], [{:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:phoenix_pubsub, "~> 2.0", [hex: :phoenix_pubsub, repo: "hexpm", optional: false]}, {:phoenix_view, "~> 1.0", [hex: :phoenix_view, repo: "hexpm", optional: false]}, {:plug, "~> 1.10", [hex: :plug, repo: "hexpm", optional: false]}, {:plug_cowboy, "~> 2.2", [hex: :plug_cowboy, repo: "hexpm", optional: true]}, {:plug_crypto, "~> 1.2", [hex: :plug_crypto, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "7bbee475acae0c3abc229b7f189e210ea788e63bd168e585f60c299a4b2f9133"},
+  "phoenix": {:git, "https://github.com/phoenixframework/phoenix.git", "8d2b33ac9691bd624ede602088d213f89600d233", []},
   "phoenix_ecto": {:hex, :phoenix_ecto, "4.4.0", "0672ed4e4808b3fbed494dded89958e22fb882de47a97634c0b13e7b0b5f7720", [:mix], [{:ecto, "~> 3.3", [hex: :ecto, repo: "hexpm", optional: false]}, {:phoenix_html, "~> 2.14.2 or ~> 3.0", [hex: :phoenix_html, repo: "hexpm", optional: true]}, {:plug, "~> 1.9", [hex: :plug, repo: "hexpm", optional: false]}], "hexpm", "09864e558ed31ee00bd48fcc1d4fc58ae9678c9e81649075431e69dbabb43cc1"},
   "phoenix_html": {:hex, :phoenix_html, "3.1.0", "0b499df05aad27160d697a9362f0e89fa0e24d3c7a9065c2bd9d38b4d1416c09", [:mix], [{:plug, "~> 1.5", [hex: :plug, repo: "hexpm", optional: true]}], "hexpm", "0c0a98a2cefa63433657983a2a594c7dee5927e4391e0f1bfd3a151d1def33fc"},
   "phoenix_live_dashboard": {:hex, :phoenix_live_dashboard, "0.5.0", "3282d8646e1bfc1ef1218f508d9fcefd48cf47f9081b7667bd9b281b688a49cf", [:mix], [{:ecto, "~> 3.6.2 or ~> 3.7", [hex: :ecto, repo: "hexpm", optional: true]}, {:ecto_psql_extras, "~> 0.6", [hex: :ecto_psql_extras, repo: "hexpm", optional: true]}, {:phoenix_live_view, "~> 0.16.0", [hex: :phoenix_live_view, repo: "hexpm", optional: false]}, {:telemetry_metrics, "~> 0.6.0", [hex: :telemetry_metrics, repo: "hexpm", optional: false]}], "hexpm", "609740be43de94ae0abd2c4300ff0356a6e8a9487bf340e69967643a59fa7ec8"},
   "phoenix_live_reload": {:hex, :phoenix_live_reload, "1.3.3", "3a53772a6118d5679bf50fc1670505a290e32a1d195df9e069d8c53ab040c054", [:mix], [{:file_system, "~> 0.2.1 or ~> 0.3", [hex: :file_system, repo: "hexpm", optional: false]}, {:phoenix, "~> 1.4", [hex: :phoenix, repo: "hexpm", optional: false]}], "hexpm", "766796676e5f558dbae5d1bdb066849673e956005e3730dfd5affd7a6da4abac"},
-  "phoenix_live_view": {:git, "https://github.com/phoenixframework/phoenix_live_view.git", "5409845a27938924c0d9a6267b498438a9103295", []},
+  "phoenix_live_view": {:git, "https://github.com/phoenixframework/phoenix_live_view.git", "d250ad2efd9159c0866def5f5d666bdeeb22ac90", []},
   "phoenix_pubsub": {:hex, :phoenix_pubsub, "2.0.0", "a1ae76717bb168cdeb10ec9d92d1480fec99e3080f011402c0a2d68d47395ffb", [:mix], [], "hexpm", "c52d948c4f261577b9c6fa804be91884b381a7f8f18450c5045975435350f771"},
   "phoenix_view": {:hex, :phoenix_view, "1.0.0", "fea71ecaaed71178b26dd65c401607de5ec22e2e9ef141389c721b3f3d4d8011", [:mix], [{:phoenix_html, "~> 2.14.2 or ~> 3.0", [hex: :phoenix_html, repo: "hexpm", optional: true]}], "hexpm", "82be3e2516f5633220246e2e58181282c71640dab7afc04f70ad94253025db0c"},
   "plug": {:hex, :plug, "1.12.1", "645678c800601d8d9f27ad1aebba1fdb9ce5b2623ddb961a074da0b96c35187d", [:mix], [{:mime, "~> 1.0 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:plug_crypto, "~> 1.1.1 or ~> 1.2", [hex: :plug_crypto, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4.3 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "d57e799a777bc20494b784966dc5fbda91eb4a09f571f76545b72a634ce0d30b"},

test/live_beats/presence/presence_client_test.exs

@@ -0,0 +1,119 @@
+defmodule Phoenix.Presence.ClientTest.Presence do
+  use Phoenix.Presence, otp_app: :live_beats,
+                        pubsub_server: LiveBeats.PubSub
+end
+
+defmodule Phoenix.Presence.ClientTest do
+  use ExUnit.Case
+
+  alias Phoenix.Presence.Client.PresenceMock
+  alias Phoenix.Presence.Client
+
+  @pubsub LiveBeats.PubSub
+  @client Phoenix.Presence.Client.Mock
+  @presence Phoenix.Presence.ClientTest.Presence
+
+  @presence_client_opts [client: @client, pubsub: @pubsub, presence: @presence]
+
+  setup tags do
+    start_supervised!({@presence, []})
+    pid = Ecto.Adapters.SQL.Sandbox.start_owner!(LiveBeats.Repo, shared: not tags[:async])
+    on_exit(fn -> Ecto.Adapters.SQL.Sandbox.stop_owner(pid) end)
+
+    :ok
+  end
+
+  test "A topic key is added to the topics state when a new process is tracked" do
+    presence_key = 1
+    topic = topic(100)
+
+    {:ok, presence_client} = start_supervised({Client, @presence_client_opts})
+    {:ok, presence_process} = start_supervised({PresenceMock, id: presence_key})
+
+    Phoenix.PubSub.subscribe(@pubsub, topic)
+    Process.monitor(presence_process)
+
+    PresenceMock.track(presence_client, presence_process, topic, presence_key)
+    assert_receive %{event: "presence_diff"}
+
+    client_state = :sys.get_state(presence_client)
+    assert %{topics: %{^topic => %{"1" => [%{phx_ref: _ref}]}}} = client_state
+  end
+
+  test "topic is removed from the topics state when there is no more presences" do
+    presence_key = 1
+    topic = topic(100)
+
+    {:ok, presence_client} = start_supervised({Client, @presence_client_opts})
+    {:ok, presence_process} = start_supervised({PresenceMock, id: presence_key})
+
+    Phoenix.PubSub.subscribe(@pubsub, topic)
+    Process.monitor(presence_process)
+
+    PresenceMock.track(presence_client, presence_process, topic, presence_key)
+    assert Process.alive?(presence_process)
+    assert_receive %{event: "presence_diff"}
+
+    client_state = :sys.get_state(presence_client)
+    assert %{topics: %{^topic => %{"1" => [%{phx_ref: _ref}]}}} = client_state
+
+    send(presence_process, :quit)
+    assert_receive {:DOWN, _ref, :process, ^presence_process, _reason}
+
+    client_state = :sys.get_state(presence_client)
+    assert %{topics: %{}} = client_state
+  end
+
+  test "metas are accumulated when there are two presences for the same key" do
+    presence_key = 1
+    topic = topic(100)
+
+    {:ok, presence_client} = start_supervised({Client, @presence_client_opts})
+    {:ok, presence_process_1} = start_supervised({PresenceMock, id: presence_key}, id: :mock_1)
+    {:ok, presence_process_2} = start_supervised({PresenceMock, id: presence_key}, id: :mock_2)
+
+    Phoenix.PubSub.subscribe(@pubsub, topic)
+
+    PresenceMock.track(presence_client, presence_process_1, topic, presence_key, %{m1: :m1})
+    assert_receive %{event: "presence_diff"}
+
+    PresenceMock.track(presence_client, presence_process_2, topic, presence_key, %{m2: :m2})
+    assert_receive %{event: "presence_diff"}
+
+    client_state = :sys.get_state(presence_client)
+
+    assert %{topics: %{^topic => %{"1" => [%{m1: :m1}, %{m2: :m2}]}}} = client_state
+  end
+
+  test "Just one meta is deleted when there are two presences for the same key and one leaves" do
+    presence_key = 1
+    topic = topic(100)
+
+    {:ok, presence_client} = start_supervised({Client, @presence_client_opts})
+    {:ok, presence_process_1} = start_supervised({PresenceMock, id: presence_key}, id: :mock_1)
+    {:ok, presence_process_2} = start_supervised({PresenceMock, id: presence_key}, id: :mock_2)
+
+    Phoenix.PubSub.subscribe(@pubsub, topic)
+    Process.monitor(presence_process_1)
+
+    PresenceMock.track(presence_client, presence_process_1, topic, presence_key, %{m1: :m1})
+    assert_receive %{event: "presence_diff"}
+
+    PresenceMock.track(presence_client, presence_process_2, topic, presence_key, %{m2: :m2})
+    assert_receive %{event: "presence_diff"}
+
+    client_state = :sys.get_state(presence_client)
+    assert %{topics: %{^topic => %{"1" => [%{m1: :m1}, %{m2: :m2}]}}} = client_state
+
+    send(presence_process_1, :quit)
+    assert_receive {:DOWN, _ref, :process, ^presence_process_1, _reason}
+    assert_receive %{event: "presence_diff"}
+
+    client_state = :sys.get_state(presence_client)
+    assert %{topics: %{^topic => %{"1" => [%{m2: :m2}]}}} = client_state
+  end
+
+  defp topic(id) do
+    "mock_topic:#{id}"
+  end
+end

test/live_beats_web/live/profile_live_test.exs

@@ -23,8 +23,8 @@ defmodule LiveBeatsWeb.ProfileLiveTest do
 
       # uploads
       assert lv
-            |> element("#upload-btn")
-            |> render_click()
+             |> element("#upload-btn")
+             |> render_click()
 
       assert render(lv) =~ "Add Songs"
 
@@ -43,26 +43,21 @@ defmodule LiveBeatsWeb.ProfileLiveTest do
       [%{"ref" => ref}] = mp3.entries
 
       refute lv
-            |> form("#song-form")
-            |> render_change(%{
-              "_target" => ["songs", ref, "artist"],
-              "songs" => %{
-                ref => %{"artist" => "Anon", "attribution" => "", "title" => "silence1s"}
-              }
-            }) =~ "can't be blank"
+             |> form("#song-form")
+             |> render_change(%{
+               "_target" => ["songs", ref, "artist"],
+               "songs" => %{
+                 ref => %{"artist" => "Anon", "attribution" => "", "title" => "silence1s"}
+               }
+             }) =~ "can't be blank"
 
-      assert {:ok, new_lv, html} =
-              lv |> form("#song-form") |> render_submit() |> follow_redirect(conn)
-
-      assert_redirected(lv, "/#{current_user.username}")
-      assert html =~ "1 song(s) uploaded"
-
-      assert html =~ "silence1s"
+      assert lv |> form("#song-form") |> render_submit() =~ "silence1s"
+      assert_patch(lv, "/#{current_user.username}")
 
       # deleting songs
 
       song = MediaLibrary.get_first_song(profile)
-      assert new_lv |> element("#delete-modal-#{song.id}-confirm") |> render_click()
+      assert lv |> element("#delete-modal-#{song.id}-confirm") |> render_click()
 
       {:ok, refreshed_lv, _} = live(conn, LiveHelpers.profile_path(current_user))
       refute render(refreshed_lv) =~ "silence1s"

test/support/conn_case.ex

@@ -35,9 +35,23 @@ defmodule LiveBeatsWeb.ConnCase do
     end
   end
 
+  defp wait_for_children(children_lookup) when is_function(children_lookup) do
+    Process.sleep(100)
+
+    for pid <- children_lookup.() do
+      ref = Process.monitor(pid)
+      assert_receive {:DOWN, ^ref, _, _, _}, 1000
+    end
+  end
+
   setup tags do
     pid = Ecto.Adapters.SQL.Sandbox.start_owner!(LiveBeats.Repo, shared: not tags[:async])
     on_exit(fn -> Ecto.Adapters.SQL.Sandbox.stop_owner(pid) end)
+
+    on_exit(fn ->
+      wait_for_children(fn -> LiveBeatsWeb.Presence.fetchers_pids() end)
+    end)
+
     {:ok, conn: Phoenix.ConnTest.build_conn()}
   end
 

test/support/presence/client_mock.ex

@@ -0,0 +1,15 @@
+defmodule Phoenix.Presence.Client.Mock do
+
+  def init(_opts) do
+    {:ok, %{}}
+  end
+
+  def handle_join(_topic, _key, _meta, state) do
+    {:ok, state}
+  end
+
+  def handle_leave(_topic, _key, _meta, state) do
+    {:ok, state}
+  end
+
+end

test/support/presence/presence_mock.ex

@@ -0,0 +1,29 @@
+defmodule Phoenix.Presence.Client.PresenceMock do
+  use GenServer
+  alias Phoenix.Presence.Client
+
+
+  def start_link(opts \\ []) do
+    GenServer.start_link(__MODULE__, opts[:id], opts)
+  end
+
+  @impl true
+  def init(id) do
+    {:ok, %{id: id}}
+  end
+
+  def track(client_pid, pid, topic, key, meta \\ %{}) do
+    GenServer.cast(pid, {:track, client_pid, topic, key, meta})
+  end
+
+  @impl true
+  def handle_info(:quit, state) do
+    {:stop, :normal, state}
+  end
+
+  @impl true
+  def handle_cast({:track, client_pid, topic, key, meta}, state) do
+    Client.track(client_pid, topic, key, meta)
+    {:noreply, state}
+  end
+end