Send event creation and event update notifications in a background task

The event update notification is made unique so that repeated changes
only trigger one notificate every 30 minutes

Signed-off-by: Thomas Citharel <tcit@tcit.fr>

lib/federation/activity_pub/types/events.ex

@@ -14,9 +14,10 @@ defmodule Mobilizon.Federation.ActivityPub.Types.Events do
   alias Mobilizon.Service.Formatter.HTML
   alias Mobilizon.Service.LanguageDetection
   alias Mobilizon.Service.Notifications.Scheduler
+  alias Mobilizon.Service.Workers.EventDelayedNotificationWorker
   alias Mobilizon.Share
   alias Mobilizon.Tombstone
-  alias Mobilizon.Web.Email.Group
+  import Mobilizon.Events.Utils, only: [calculate_notification_time: 1]
   import Mobilizon.Federation.ActivityPub.Utils, only: [make_create_data: 2, make_update_data: 2]
   require Logger
 
@@ -29,10 +30,15 @@ defmodule Mobilizon.Federation.ActivityPub.Types.Events do
     args = prepare_args_for_event(args)
 
     case EventsManager.create_event(args) do
-      {:ok, %Event{} = event} ->
+      {:ok, %Event{uuid: event_uuid, begins_on: begins_on} = event} ->
         EventActivity.insert_activity(event, subject: "event_created")
-        # TODO make this async
+
-        Group.notify_of_new_event(event)
+        %{action: :notify_of_new_event, event_uuid: event_uuid}
+        |> EventDelayedNotificationWorker.new(
+          scheduled_at: calculate_notification_time(begins_on)
+        )
+        |> Oban.insert()
+
         event_as_data = Convertible.model_to_as(event)
         audience = Audience.get_audience(event)
         create_data = make_create_data(event_as_data, Map.merge(audience, additional))

lib/mobilizon/events/events.ex

@@ -10,6 +10,7 @@ defmodule Mobilizon.Events do
 
   import Mobilizon.Service.Guards
   import Mobilizon.Storage.Ecto
+  import Mobilizon.Events.Utils, only: [calculate_notification_time: 1]
 
   alias Ecto.{Changeset, Multi}
 
@@ -18,6 +19,7 @@ defmodule Mobilizon.Events do
 
   alias Mobilizon.Events.{
     Event,
+    EventOptions,
     EventParticipantStats,
     FeedToken,
     Participant,
@@ -27,13 +29,12 @@ defmodule Mobilizon.Events do
     Track
   }
 
-  alias Mobilizon.Service.Workers
+  alias Mobilizon.Service.Workers.BuildSearch
+  alias Mobilizon.Service.Workers.EventDelayedNotificationWorker
   alias Mobilizon.Share
   alias Mobilizon.Storage.{Page, Repo}
   alias Mobilizon.Users.{Setting, User}
 
-  alias Mobilizon.Web.Email
-
   defenum(EventVisibility, :event_visibility, [
     :public,
     :unlisted,
@@ -229,7 +230,7 @@ defmodule Mobilizon.Events do
     with {:ok, %{insert: %Event{} = event}} <- do_create_event(attrs),
          %Event{} = event <- Repo.preload(event, @event_preloads) do
       unless event.draft do
-        Workers.BuildSearch.enqueue(:insert_search_event, %{"event_id" => event.id})
+        BuildSearch.enqueue(:insert_search_event, %{"event_id" => event.id})
       end
 
       {:ok, event}
@@ -303,19 +304,63 @@ defmodule Mobilizon.Events do
          %Event{} = new_event <- Repo.preload(new_event, @event_preloads, force: true) do
       Cachex.del(:ics, "event_#{new_event.uuid}")
 
-      Email.Event.calculate_event_diff_and_send_notifications(
+      unless new_event.draft do
-        old_event,
+        %{
-        new_event,
+          action: :notify_of_event_update,
-        changes
+          event_uuid: new_event.uuid,
+          old_event: old_event |> Map.from_struct() |> Map.take(Event.__schema__(:fields)),
+          changes: build_changes(changes)
+        }
+        |> EventDelayedNotificationWorker.new(
+          scheduled_at: calculate_notification_time(new_event.begins_on),
+          replace: [:scheduled_at, :args],
+          unique: [period: 1, keys: [:event_uuid, :action]]
         )
+        |> Oban.insert()
 
-      unless new_event.draft,
+        BuildSearch.enqueue(:update_search_event, %{"event_id" => new_event.id})
-        do: Workers.BuildSearch.enqueue(:update_search_event, %{"event_id" => new_event.id})
+      end
 
       {:ok, new_event}
     end
   end
 
+  @spec build_changes(map()) :: map()
+  defp build_changes(changes) do
+    changes
+    |> Map.take(Event.__schema__(:fields))
+    |> maybe_add_address(changes)
+    |> maybe_add_options(changes)
+  end
+
+  @spec maybe_add_address(map(), map()) :: map()
+  defp maybe_add_address(changes, %{physical_address: %Ecto.Changeset{} = changeset}),
+    do:
+      Map.put(
+        changes,
+        :physical_address,
+        changeset
+        |> Ecto.Changeset.apply_changes()
+        |> Map.from_struct()
+        |> Map.take(Address.__schema__(:fields))
+      )
+
+  defp maybe_add_address(changes, _), do: Map.drop(changes, [:physical_address])
+
+  @spec maybe_add_options(map(), map()) :: map()
+  defp maybe_add_options(changes, %{options: %Ecto.Changeset{} = changeset}),
+    do:
+      Map.put(
+        changes,
+        :options,
+        changeset
+        |> Ecto.Changeset.apply_changes()
+        |> Map.from_struct()
+        |> Map.take(EventOptions.__schema__(:fields))
+      )
+
+  defp maybe_add_options(changes, _), do: Map.drop(changes, [:options])
+
   @doc """
   Deletes an event.
   """

lib/mobilizon/events/utils.ex

@@ -0,0 +1,18 @@
+defmodule Mobilizon.Events.Utils do
+  @moduledoc """
+  Utils related to events
+  """
+
+  @spec calculate_notification_time(DateTime.t()) :: DateTime.t()
+  def calculate_notification_time(begins_on, options \\ []) do
+    now = Keyword.get(options, :now, DateTime.utc_now())
+    notify_at = DateTime.add(now, 1800)
+
+    # If the event begins in less than half an hour, send the notification right now
+    if DateTime.compare(notify_at, begins_on) == :lt do
+      notify_at
+    else
+      now
+    end
+  end
+end

lib/service/workers/event_delayed_notification_worker.ex

@@ -0,0 +1,51 @@
+defmodule Mobilizon.Service.Workers.EventDelayedNotificationWorker do
+  @moduledoc """
+  Worker to send notifications about an event changes a while after they're performed
+  """
+
+  use Oban.Worker, unique: [period: :infinity, keys: [:event_uuid, :action]]
+
+  alias Mobilizon.Events
+  alias Mobilizon.Events.Event
+  alias Mobilizon.Web.Email.Event, as: EventEmail
+  alias Mobilizon.Web.Email.Group
+  alias Oban.Job
+
+  @impl Oban.Worker
+  def perform(%Job{args: %{"action" => "notify_of_new_event", "event_uuid" => event_uuid}}) do
+    case Events.get_event_by_uuid_with_preload(event_uuid) do
+      %Event{} = event ->
+        Group.notify_of_new_event(event)
+
+      nil ->
+        # Event deleted inbetween, no worries, just ignore
+        :ok
+    end
+  end
+
+  @impl Oban.Worker
+  def perform(%Job{
+        args: %{
+          "action" => "notify_of_event_update",
+          "event_uuid" => event_uuid,
+          "old_event" => old_event,
+          "changes" => changes
+        }
+      }) do
+    old_event = for {key, val} <- old_event, into: %{}, do: {String.to_existing_atom(key), val}
+    old_event = struct(Event, old_event)
+
+    case Events.get_event_by_uuid_with_preload(event_uuid) do
+      %Event{draft: false} = new_event ->
+        EventEmail.calculate_event_diff_and_send_notifications(
+          old_event,
+          new_event,
+          changes
+        )
+
+      _ ->
+        # Event deleted inbetween, no worries, just ignore
+        :ok
+    end
+  end
+end

lib/web/email/event.ex

@@ -26,6 +26,7 @@ defmodule Mobilizon.Web.Email.Event do
           Event.t(),
           Event.t(),
           MapSet.t(),
+          String.t(),
           String.t()
         ) ::
           Bamboo.Email.t()
@@ -36,8 +37,8 @@ defmodule Mobilizon.Web.Email.Event do
         %Event{} = old_event,
         %Event{} = event,
         changes,
-        timezone \\ "Etc/UTC",
+        timezone,
-        locale \\ "en"
+        locale
       ) do
     Gettext.put_locale(locale)
 
@@ -70,13 +71,15 @@ defmodule Mobilizon.Web.Email.Event do
         %Event{id: event_id} = event,
         changes
       ) do
-    important = MapSet.new(@important_changes)
+    important = @important_changes |> Enum.map(&to_string/1) |> MapSet.new()
 
     diff =
       changes
       |> Map.keys()
       |> MapSet.new()
       |> MapSet.intersection(important)
+      |> Enum.map(&String.to_existing_atom/1)
+      |> MapSet.new()
 
     if MapSet.size(diff) > 0 do
       Repo.transaction(fn ->
@@ -178,7 +181,7 @@ defmodule Mobilizon.Web.Email.Event do
          locale
        ) do
     email
-    |> Email.Event.event_updated(participant, actor, old_event, event, diff, timezone, locale)
+    |> event_updated(participant, actor, old_event, event, diff, timezone, locale)
     |> Email.Mailer.send_email_later()
   end
 end

lib/web/templates/email/event_updated.text.eex

@@ -1,22 +1,14 @@
 <%= gettext "Event update!" %>
 ==
 <%= gettext "There have been changes for %{title} so we'd thought we'd let you know.", title: @old_event.title %>
-<%= if MapSet.member?(@changes, :status) do %>
+<%= if MapSet.member?(@changes, :status) do %><%= case @event.status do %><% :confirmed -> %><%= gettext "This event has been confirmed" %><% :tentative -> %>
-  <%= case @event.status do %>
+<%= gettext "This event has yet to be confirmed: organizers will let you know if they do confirm it." %><% :cancelled -> %>
-    <% :confirmed -> %>
+<%= gettext "This event has been cancelled by its organizers. Sorry!" %><% end %><% end %><%= if MapSet.member?(@changes, :title) do %>
-    <%= gettext "This event has been confirmed" %>
+<%= gettext "New title: %{title}", title: @event.title %><% end %><%= if MapSet.member?(@changes, :begins_on) do %>
-    <% :tentative -> %>
+<%= gettext "New start date:" %> <%= render("date/event_tz_date.text", event: @event, date: @event.begins_on, timezone: @timezone, locale: @locale) %><% end %><%= if MapSet.member?(@changes, :ends_on) && !is_nil(@event.ends_on) do %>
-    <%= gettext "This event has yet to be confirmed: organizers will let you know if they do confirm it." %>
+<%= gettext "New end date:" %> <%= render("date/event_tz_date.text", event: @event, date: @event.ends_on, timezone: @timezone, locale: @locale) %><% end %><%= if MapSet.member?(@changes, :physical_address) do %>
-    <% :cancelled -> %>
+<%= gettext "New location:" %> <%= Mobilizon.Addresses.Address.representation(@event.physical_address) %><% end %>
-    <%= gettext "This event has been cancelled by its organizers. Sorry!" %>
+
-  <% end %>
-<% end %>
-<%= if MapSet.member?(@changes, :title) do %>
-  <%= gettext "New title: %{title}", title: @event.title %>
-<% end %>
-<%= if MapSet.member?(@changes, :begins_on) do %><%= render("date/event_tz_date.text", event: @event, date: @event.begins_on, timezone: @timezone, locale: @locale) %>
-<% end %>
-<%= if MapSet.member?(@changes, :ends_on) && !is_nil(@event.ends_on) do %><%= render("date/event_tz_date.text", event: @event, date: @event.ends_on, timezone: @timezone, locale: @locale) %>
-<% end %>
 <%= gettext "Visit the updated event page: %{link}", link: Routes.page_url(Mobilizon.Web.Endpoint, :event, @event.uuid) %>
+
 <%= ngettext "Would you wish to cancel your attendance, visit the event page through the link above and click the « Attending » button.", "Would you wish to cancel your attendance to one or several events, visit the event pages through the links above and click the « Attending » button.", 1 %>

test/graphql/resolvers/event_test.exs

@@ -1,6 +1,6 @@
 defmodule Mobilizon.Web.Resolvers.EventTest do
   use Mobilizon.Web.ConnCase
-  use Bamboo.Test
+  use Bamboo.Test, shared: true
   use Oban.Testing, repo: Mobilizon.Storage.Repo
 
   import Mobilizon.Factory
@@ -180,6 +180,8 @@ defmodule Mobilizon.Web.Resolvers.EventTest do
     end
 
     test "create_event/3 creates an event", %{conn: conn, actor: actor, user: user} do
+      begins_on = DateTime.utc_now()
+
       res =
         conn
         |> auth_conn(user)
@@ -188,18 +190,25 @@ defmodule Mobilizon.Web.Resolvers.EventTest do
           variables: %{
             title: "come to my event",
             description: "it will be fine",
-            begins_on: "#{DateTime.utc_now()}",
+            begins_on: "#{DateTime.add(begins_on, 3600 * 24)}",
             organizer_actor_id: "#{actor.id}"
           }
         )
 
       assert res["data"]["createEvent"]["title"] == "come to my event"
-      {id, ""} = res["data"]["createEvent"]["id"] |> Integer.parse()
+      id = String.to_integer(res["data"]["createEvent"]["id"])
+      uuid = res["data"]["createEvent"]["uuid"]
 
       assert_enqueued(
         worker: Workers.BuildSearch,
         args: %{event_id: id, op: :insert_search_event}
       )
+
+      assert_enqueued(
+        worker: Workers.EventDelayedNotificationWorker,
+        args: %{event_uuid: uuid, action: :notify_of_new_event},
+        scheduled_at: {DateTime.add(begins_on, 1800), delta: 5}
+      )
     end
 
     test "create_event/3 creates an event and escapes title", %{
@@ -930,8 +939,10 @@ defmodule Mobilizon.Web.Resolvers.EventTest do
                ["ends_on cannot be set before begins_on"]
     end
 
-    test "update_event/3 updates an event", %{conn: conn, actor: actor, user: user} do
+    test "updates an event", %{conn: conn, actor: actor, user: user} do
+      %Event{uuid: event_uuid, title: event_title} =
         event = insert(:event, organizer_actor: actor)
+
       creator = insert(:participant, event: event, actor: actor, role: :creator)
       participant_user = insert(:user)
       participant_actor = insert(:actor, user: participant_user)
@@ -1005,6 +1016,25 @@ defmodule Mobilizon.Web.Resolvers.EventTest do
         args: %{event_id: event_id_int, op: :update_search_event}
       )
 
+      assert [
+               %{
+                 args: %{
+                   "event_uuid" => ^event_uuid,
+                   "action" => "notify_of_event_update",
+                   "old_event" => %{
+                     "title" => ^event_title
+                   },
+                   "changes" => %{
+                     "description" => "description updated",
+                     "status" => "tentative",
+                     "title" => "my event updated"
+                   }
+                 }
+               }
+             ] = all_enqueued(worker: Workers.EventDelayedNotificationWorker)
+
+      Oban.drain_queue(queue: :default, with_scheduled: true)
+
       {:ok, new_event} = Mobilizon.Events.get_event_with_preload(event.id)
 
       assert_delivered_email(
@@ -1014,7 +1044,9 @@ defmodule Mobilizon.Web.Resolvers.EventTest do
           actor,
           event,
           new_event,
-          MapSet.new([:title, :begins_on, :ends_on, :status, :physical_address])
+          MapSet.new([:title, :begins_on, :ends_on, :status, :physical_address]),
+          "Etc/UTC",
+          "en"
         )
       )
 
@@ -1025,7 +1057,9 @@ defmodule Mobilizon.Web.Resolvers.EventTest do
           participant_actor,
           event,
           new_event,
-          MapSet.new([:title, :begins_on, :ends_on, :status, :physical_address])
+          MapSet.new([:title, :begins_on, :ends_on, :status, :physical_address]),
+          "Etc/UTC",
+          "en"
         )
       )
     end

test/mobilizon/events/utils_test.exs

@@ -0,0 +1,19 @@
+defmodule Mobilizon.Events.UtilsTest do
+  use Mobilizon.DataCase, async: true
+
+  alias Mobilizon.Events.Utils
+
+  @now ~U[2021-11-19T18:17:00Z]
+
+  describe "calculate_notification_time" do
+    test "when the event begins in less than 30 minutes" do
+      begins_on = ~U[2021-11-19T18:27:00Z]
+      assert @now == Utils.calculate_notification_time(begins_on, now: @now)
+    end
+
+    test "when the event begins in more than 30 minutes" do
+      begins_on = ~U[2021-11-19T18:17:00Z]
+      assert begins_on == Utils.calculate_notification_time(begins_on, now: @now)
+    end
+  end
+end

test/service/workers/event_delayed_notification_worker_test.exs

@@ -0,0 +1,43 @@
+defmodule Mobilizon.Service.Workers.EventDelayedNotificationWorkerTest do
+  @moduledoc """
+  Test the event delayed notification worker
+  """
+
+  alias Mobilizon.Events.Event
+  alias Mobilizon.Service.Workers.EventDelayedNotificationWorker
+  alias Oban.Job
+
+  use Mobilizon.DataCase
+  import Mobilizon.Factory
+
+  test "Run notify of new event" do
+    group = insert(:group)
+    event = insert(:event, attributed_to: group)
+
+    assert :ok ==
+             EventDelayedNotificationWorker.perform(%Job{
+               args: %{"action" => "notify_of_new_event", "event_uuid" => event.uuid}
+             })
+  end
+
+  test "Run notify of updates to event" do
+    group = insert(:group)
+    event = insert(:event, attributed_to: group)
+    old_event = %Event{event | title: "Previous title"}
+
+    old_event =
+      for {key, val} <- Map.from_struct(old_event), into: %{}, do: {Atom.to_string(key), val}
+
+    changes = %{"title" => "New title"}
+
+    assert {:ok, :ok} ==
+             EventDelayedNotificationWorker.perform(%Job{
+               args: %{
+                 "action" => "notify_of_event_update",
+                 "event_uuid" => event.uuid,
+                 "old_event" => old_event,
+                 "changes" => changes
+               }
+             })
+  end
+end