Merge pull request #77 from fly-apps/cm-active-users-sync

Active users sync

lib/live_beats/application.ex

@@ -21,10 +21,11 @@ defmodule LiveBeats.Application do
   def start(_type, _args) do
     parent = FLAME.Parent.get()
     LiveBeats.MediaLibrary.attach()
+    whisper_serving? = parent || FLAME.Backend.impl() != FLAME.FlyBackend
 
     children =
       [
-        {Nx.Serving, name: LiveBeats.WhisperServing, serving: load_serving()},
+        whisper_serving? && {Nx.Serving, name: LiveBeats.WhisperServing, serving: load_serving()},
         !parent && {DNSCluster, query: Application.get_env(:wps, :dns_cluster_query) || :ignore},
         {Task.Supervisor, name: LiveBeats.TaskSupervisor},
         # Start the Ecto repository

lib/live_beats/media_library.ex

@@ -23,7 +23,28 @@ defmodule LiveBeats.MediaLibrary do
 
   def handle_execute({Accounts, %Accounts.Events.PublicSettingsChanged{user: user}}) do
     profile = get_profile!(user)
-    broadcast!(user.id, %Events.PublicProfileUpdated{profile: profile})
+
+    active? =
+      Repo.exists?(
+        from(s in Song,
+          inner_join: u in LiveBeats.Accounts.User,
+          on: s.user_id == ^user.id,
+          where: s.status in [:playing],
+          limit: 1
+        )
+      )
+
+    if active? do
+      broadcast_active_profile!(profile)
+    else
+      broadcast_inactive_profile!(profile)
+    end
+
+    broadcast!(user.id, %Events.PublicProfileUpdated{profile: profile, active?: active?})
+  end
+
+  def subscribe_to_active_profiles do
+    Phoenix.PubSub.subscribe(@pubsub, "active_profiles")
   end
 
   def subscribe_to_profile(%Profile{} = profile) do
@@ -53,6 +74,8 @@ defmodule LiveBeats.MediaLibrary do
   end
 
   def play_song(%Song{} = song) do
+    user = Accounts.get_user!(song.user_id)
+
     played_at =
       cond do
         playing?(song) ->
@@ -86,6 +109,7 @@ defmodule LiveBeats.MediaLibrary do
 
     elapsed = elapsed_playback(new_song)
 
+    broadcast_active_profile!(get_profile!(user))
     broadcast!(song.user_id, %Events.Play{song: song, elapsed: elapsed})
 
     new_song
@@ -98,6 +122,8 @@ defmodule LiveBeats.MediaLibrary do
   end
 
   def pause_song(%Song{} = song) do
+    user = Accounts.get_user!(song.user_id)
+
     now = DateTime.truncate(DateTime.utc_now(), :second)
     set = [status: :paused, paused_at: now]
     pause_query = from(s in Song, where: s.id == ^song.id, update: [set: ^set])
@@ -114,6 +140,7 @@ defmodule LiveBeats.MediaLibrary do
       |> Multi.update_all(:now_paused, fn _ -> pause_query end, [])
       |> Repo.transaction()
 
+    broadcast_inactive_profile!(get_profile!(user))
     broadcast!(song.user_id, %Events.Pause{song: song})
   end
 
@@ -554,6 +581,16 @@ defmodule LiveBeats.MediaLibrary do
     Phoenix.PubSub.broadcast_from!(@pubsub, pid, topic(user_id), {__MODULE__, msg})
   end
 
+  defp broadcast_active_profile!(%Profile{} = profile) do
+    msg = %Events.PublicProfileActive{profile: profile}
+    Phoenix.PubSub.broadcast!(@pubsub, "active_profiles", {__MODULE__, msg})
+  end
+
+  defp broadcast_inactive_profile!(%Profile{} = profile) do
+    msg = %Events.PublicProfileInActive{profile: profile}
+    Phoenix.PubSub.broadcast!(@pubsub, "active_profiles", {__MODULE__, msg})
+  end
+
   defp topic(user_id) when is_integer(user_id), do: "profile:#{user_id}"
 
   defp validate_songs_limit(user_songs, new_songs_count) do

lib/live_beats/media_library/events.ex

@@ -8,6 +8,14 @@ defmodule LiveBeats.MediaLibrary.Events do
   end
 
   defmodule PublicProfileUpdated do
+    defstruct profile: nil, active?: false
+  end
+
+  defmodule PublicProfileActive do
+    defstruct profile: nil
+  end
+
+  defmodule PublicProfileInActive do
     defstruct profile: nil
   end
 

lib/live_beats_web/components/core_components.ex

@@ -574,8 +574,8 @@ defmodule LiveBeatsWeb.CoreComponents do
 
   slot :col, required: true do
     attr :label, :string
-    attr :class, :string
+    attr :class, :any
-    attr :class!, :string
+    attr :class!, :any
   end
 
   def table(assigns) do

lib/live_beats_web/components/layouts.ex

@@ -12,18 +12,18 @@ defmodule LiveBeatsWeb.Layouts do
       <h3 class="px-3 text-xs font-semibold text-gray-500 uppercase tracking-wider" id={@id}>
         Active Users
       </h3>
-      <div class="mt-1 space-y-1" role="group" aria-labelledby={@id}>
+      <div id={"#{@id}-stream"} phx-update="stream" class="mt-1 space-y-1" role="group" aria-labelledby={@id}>
-        <%= for user <- @users do %>
+        <.link
-          <.link
+          :for={{id, user} <- @users}
-            navigate={profile_path(user)}
+          id={id}
-            class="group flex items-center px-3 py-2 text-base leading-5 font-medium text-gray-600 rounded-md hover:text-gray-900 hover:bg-gray-50"
+          navigate={profile_path(user)}
-          >
+          class="group flex items-center px-3 py-2 text-base leading-5 font-medium text-gray-600 rounded-md hover:text-gray-900 hover:bg-gray-50"
-            <span class="w-2.5 h-2.5 mr-4 bg-indigo-500 rounded-full" aria-hidden="true"></span>
+        >
-            <span class="truncate">
+          <span class="w-2.5 h-2.5 mr-4 bg-indigo-500 rounded-full" aria-hidden="true"></span>
-              <%= user.username %>
+          <span class="truncate">
-            </span>
+            <%= user.username %>
-          </.link>
+          </span>
-        <% end %>
+        </.link>
       </div>
     </div>
     """

lib/live_beats_web/components/layouts/live.html.heex

@@ -57,7 +57,7 @@
         <%= if @current_user do %>
           <.sidebar_nav_links current_user={@current_user} active_tab={@active_tab} />
         <% end %>
-        <.sidebar_active_users id="desktop-active-users" users={@active_users} />
+        <.sidebar_active_users id="mobile-active-users" users={@streams.mobile_active_users} />
       </nav>
     </div>
   </div>
@@ -121,7 +121,7 @@
           <.sidebar_nav_links current_user={@current_user} active_tab={@active_tab} />
         <% end %>
         <!-- Secondary navigation -->
-        <.sidebar_active_users id="mobile-active-users" users={@active_users} />
+        <.sidebar_active_users id="desktop-active-users" users={@streams.active_users} />
       </nav>
     </div>
   </div>

lib/live_beats_web/endpoint.ex

@@ -7,7 +7,7 @@ defmodule LiveBeatsWeb.Endpoint do
   @session_options [
     store: :cookie,
     key: "_live_beats_key_v1",
-    signing_salt: "9OALgV62"
+    signing_salt: "1a9OALgV62"
   ]
 
   socket "/live", Phoenix.LiveView.Socket,

lib/live_beats_web/live/nav.ex

@@ -6,12 +6,22 @@ defmodule LiveBeatsWeb.Nav do
   alias LiveBeatsWeb.{ProfileLive, SettingsLive}
 
   def on_mount(:default, _params, _session, socket) do
+    if connected?(socket) do
+      MediaLibrary.subscribe_to_active_profiles()
+    end
+
+    active_users = MediaLibrary.list_active_profiles(limit: 20)
+
     {:cont,
      socket
-     |> assign(active_users: MediaLibrary.list_active_profiles(limit: 20))
+     |> stream_configure(:mobile_active_users, dom_id: &"mobile_active-#{&1.user_id}")
+     |> stream_configure(:active_users, dom_id: &"active-#{&1.user_id}")
+     |> stream(:active_users, active_users)
+     |> stream(:mobile_active_users, active_users)
      |> assign(:region, System.get_env("FLY_REGION") || "iad")
      |> attach_hook(:active_tab, :handle_params, &handle_active_tab_params/3)
-     |> attach_hook(:ping, :handle_event, &handle_event/3)}
+     |> attach_hook(:ping, :handle_event, &handle_event/3)
+     |> attach_hook(:active_profile_changes, :handle_info, &handle_info/2)}
   end
 
   defp handle_active_tab_params(params, _url, socket) do
@@ -41,6 +51,16 @@ defmodule LiveBeatsWeb.Nav do
 
   defp handle_event(_, _, socket), do: {:cont, socket}
 
+  defp handle_info({MediaLibrary, %MediaLibrary.Events.PublicProfileActive{} = update}, socket) do
+    {:halt, stream_insert(socket, :active_users, update.profile)}
+  end
+
+  defp handle_info({MediaLibrary, %MediaLibrary.Events.PublicProfileInActive{} = update}, socket) do
+    {:halt, stream_delete(socket, :active_users, update.profile)}
+  end
+
+  defp handle_info(_, socket), do: {:cont, socket}
+
   defp rate_limited_ping_broadcast(socket, %Accounts.User{} = user, rtt) when is_integer(rtt) do
     now = System.system_time(:millisecond)
     last_ping_at = socket.assigns[:last_ping_at]

lib/live_beats_web/live/profile_live/song_entry_component.ex

@@ -17,8 +17,7 @@ defmodule LiveBeatsWeb.ProfileLive.SongEntryComponent do
           <% else %>
             Title
             <span class="text-gray-400">
-              (calculating duration
+              (calculating duration)
-              <.spinner class="inline-block animate-spin h-2.5 w-2.5 text-gray-400" />)
             </span>
           <% end %>
         </label>