Expand user events PubSub

Author: raulrpearson

lib/claper/events.ex

@@ -6,8 +6,8 @@ defmodule Claper.Events do
   """
 
   import Ecto.Query, warn: false
-  alias Claper.Repo
 
+  alias Claper.{Repo, Accounts}
   alias Claper.Events.{Event, ActivityLeader}
 
   @default_page_size 5
@@ -363,7 +363,10 @@ defmodule Claper.Events do
     |> validate_unique_event()
     |> case do
       {:ok, event} ->
-        Repo.insert(event, returning: [:uuid])
+        with {:ok, event} <- Repo.insert(event, returning: [:uuid]) do
+          broadcast_all_users({:created, event})
+          {:ok, event}
+        end
 
       {:error, changeset} ->
         {:error, %{changeset | action: :insert}}
@@ -402,7 +405,25 @@ defmodule Claper.Events do
     |> validate_unique_event()
     |> case do
       {:ok, event} ->
-        Repo.update(event, returning: [:uuid])
+        with {:ok, event} <- Repo.update(event, returning: [:uuid]) do
+          broadcast_all_users({:updated, event})
+
+          deleted_leaders =
+            attrs
+            |> Map.get("leaders", %{})
+            |> Map.values()
+            |> Enum.filter(fn
+              %{"delete" => "true"} -> true
+              _ -> false
+            end)
+
+          for %{"email" => leader_email} <- deleted_leaders do
+            leader = Accounts.get_user_by_email(leader_email)
+            broadcast_user_events(leader.id, {:updated, event})
+          end
+
+          {:ok, event}
+        end
 
       {:error, changeset} ->
         {:error, %{changeset | action: :update}}
@@ -424,7 +445,9 @@ defmodule Claper.Events do
     |> Repo.update()
     |> case do
       {:ok, event} ->
-        broadcast({:ok, event, event.uuid}, :event_terminated)
+        broadcast_all_users({:updated, event})
+        broadcast_event(event.uuid, {:event_terminated, event.uuid})
+        {:ok, event}
 
       {:error, changeset} ->
         {:error, %{changeset | action: :update}}
@@ -693,7 +716,20 @@ defmodule Claper.Events do
 
   """
   def delete_event(%Event{} = event) do
-    Repo.delete(event)
+    leaders =
+      for %{email: email} <- get_activity_leaders_for_event(event.id) do
+        Accounts.get_user_by_email(email)
+      end
+
+    with {:ok, event} <- Repo.delete(event) do
+      broadcast_user_events(event.user_id, {:deleted, event})
+
+      for leader <- leaders do
+        broadcast_user_events(leader.id, {:deleted, event})
+      end
+
+      {:ok, event}
+    end
   end
 
   @doc """
@@ -709,8 +745,6 @@ defmodule Claper.Events do
     Event.changeset(event, attrs)
   end
 
-  alias Claper.Events.ActivityLeader
-
   @doc """
   Creates a activity leader.
 
@@ -777,13 +811,47 @@ defmodule Claper.Events do
     ActivityLeader.changeset(activity_leader, attrs)
   end
 
-  defp broadcast({:ok, e, event_uuid}, event) do
-    Phoenix.PubSub.broadcast(
-      Claper.PubSub,
-      "event:#{event_uuid}",
-      {event, event_uuid}
-    )
+  @doc """
+  Subscribes to an event's public `Phoenix.PubSub` topic.
 
-    {:ok, e}
+  The broadcasted messages match the pattern:
+
+    * {:terminated, event_uuid}
+
+  """
+  def subscribe_event(event_uuid) when is_binary(event_uuid) do
+    Phoenix.PubSub.subscribe(Claper.PubSub, "event:#{event_uuid}")
+  end
+
+  defp broadcast_event(event_uuid, message) when is_binary(event_uuid) do
+    Phoenix.PubSub.broadcast(Claper.PubSub, "event:#{event_uuid}", message)
+  end
+
+  @doc """
+  Subscribes to a user's events private `Phoenix.PubSub` topic.
+
+  The broadcasted messages match the pattern:
+
+    * {:created, %Event{}}
+    * {:updated, %Event{}}
+    * {:deleted, %Event{}}
+
+  """
+  def subscribe_user_events(user_id) when is_integer(user_id) do
+    Phoenix.PubSub.subscribe(Claper.PubSub, "user:#{user_id}:events")
+  end
+
+  def broadcast_user_events(user_id, message) when is_integer(user_id) do
+    Phoenix.PubSub.broadcast(Claper.PubSub, "user:#{user_id}:events", message)
+  end
+
+  defp broadcast_all_users({_type, %Event{} = event} = message, _opts \\ []) do
+    event = Repo.preload(event, [:leaders])
+    broadcast_user_events(event.user_id, message)
+
+    for %{email: leader_email} <- event.leaders do
+      leader = Accounts.get_user_by_email(leader_email)
+      broadcast_user_events(leader.id, message)
+    end
   end
 end

lib/claper/events/event.ex

@@ -90,8 +90,8 @@ defmodule Claper.Events.Event do
     change(event, expired_at: expiry)
   end
 
-  def subscribe(event_id) do
-    Phoenix.PubSub.subscribe(Claper.PubSub, "event:#{event_id}")
+  def subscribe(event_uuid) do
+    Phoenix.PubSub.subscribe(Claper.PubSub, "event:#{event_uuid}")
   end
 
   def started?(event) do

lib/claper/tasks/converter.ex

@@ -4,6 +4,7 @@ defmodule Claper.Tasks.Converter do
   We use a hash to identify the presentation. A new hash is generated when the conversion is finished and the presentation is being uploaded.
   """
 
+  alias Claper.Events
   alias ExAws.S3
   alias Porcelain.Result
 
@@ -19,11 +20,7 @@ defmodule Claper.Tasks.Converter do
         "status" => "progress"
       })
 
-    Phoenix.PubSub.broadcast(
-      Claper.PubSub,
-      "events:#{user_id}",
-      {:presentation_file_process_done, presentation}
-    )
+    Events.broadcast_user_events(user_id, {:presentation_file_process_done, presentation})
 
     path =
       Path.join([
@@ -167,11 +164,7 @@ defmodule Claper.Tasks.Converter do
            }) do
       if get_presentation_storage() != "local", do: File.rm_rf!(path)
 
-      Phoenix.PubSub.broadcast(
-        Claper.PubSub,
-        "events:#{user_id}",
-        {:presentation_file_process_done, presentation}
-      )
+      Events.broadcast_user_events(user_id, {:presentation_file_process_done, presentation})
     end
   end
 
@@ -182,11 +175,7 @@ defmodule Claper.Tasks.Converter do
            }) do
       File.rm_rf!(path)
 
-      Phoenix.PubSub.broadcast(
-        Claper.PubSub,
-        "events:#{user_id}",
-        {:presentation_file_process_done, presentation}
-      )
+      Events.broadcast_user_events(user_id, {:presentation_file_process_done, presentation})
     end
   end
 

lib/claper_web/live/event_live/index.ex

@@ -22,7 +22,7 @@ defmodule ClaperWeb.EventLive.Index do
       })
 
     if connected?(socket) do
-      Phoenix.PubSub.subscribe(Claper.PubSub, "events:#{socket.assigns.current_user.id}")
+      Events.subscribe_user_events(socket.assigns.current_user.id)
     end
 
     expired_events_count = Events.count_expired_events(socket.assigns.current_user.id)
@@ -57,6 +57,12 @@ defmodule ClaperWeb.EventLive.Index do
      socket |> assign(:events, [event | socket.assigns.events]) |> put_flash(:info, nil)}
   end
 
+  @impl true
+  def handle_info({type, %Events.Event{}}, socket)
+      when type in [:created, :updated, :deleted] do
+    {:noreply, refresh_events(socket)}
+  end
+
   @impl true
   def handle_event("validate", %{"event" => event_params}, socket) do
     changeset =
@@ -245,4 +251,16 @@ defmodule ClaperWeb.EventLive.Index do
       if(socket.assigns.page == 1, do: events, else: socket.assigns.events ++ events)
     )
   end
+
+  defp refresh_events(socket) do
+    expired_events_count = Events.count_expired_events(socket.assigns.current_user.id)
+    invited_events_count = Events.count_managed_events_by(socket.assigns.current_user.email)
+
+    socket
+    |> assign(:has_expired_events, expired_events_count > 0)
+    |> assign(:has_invited_events, invited_events_count > 0)
+    |> assign(:events, [])
+    |> assign(:page, 1)
+    |> load_events()
+  end
 end

lib/claper_web/live/event_live/show.ex

@@ -187,7 +187,7 @@ defmodule ClaperWeb.EventLive.Show do
   end
 
   @impl true
-  def handle_info({:event_terminated, _event}, socket) do
+  def handle_info({:event_terminated, _event_uuid}, socket) do
     {:noreply,
      socket
      |> put_flash(:error, gettext("This event has been terminated"))