Fix notifications query not working for those without

referenced_activity

lib/clacks/activity.ex

@@ -12,8 +12,8 @@ defmodule Clacks.Activity do
   schema "activities" do
     field :data, :map
     field :local, :boolean
-    field :actor, :string
 
+    belongs_to :actor, Clacks.Actor, foreign_key: :actor_ap_id, references: :ap_id, type: :string
     has_one :object, Clacks.Object, on_delete: :nothing, foreign_key: :id
 
     timestamps()
@@ -21,8 +21,8 @@ defmodule Clacks.Activity do
 
   def changeset(%__MODULE__{} = schema, attrs) do
     schema
-    |> cast(attrs, [:data, :local, :actor])
-    |> validate_required([:data, :local, :actor])
+    |> cast(attrs, [:data, :local, :actor_ap_id])
+    |> validate_required([:data, :local, :actor_ap_id])
   end
 
   @spec changeset_for_creating(activity :: map(), local :: boolean()) :: Ecto.Changeset.t()
@@ -30,7 +30,7 @@ defmodule Clacks.Activity do
     changeset(%__MODULE__{}, %{
       data: activity,
       local: local,
-      actor: activity["actor"]
+      actor_ap_id: activity["actor"]
     })
   end
 
@@ -60,14 +60,17 @@ defmodule Clacks.Activity do
     Repo.one(from a in __MODULE__, where: fragment("?->>'id'", a.data) == ^ap_id)
   end
 
-  @spec get_by_object_ap_id(object_id :: String.t()) :: t() | nil
-  def get_by_object_ap_id(object_id) do
+  @spec get_by_object_ap_id(object_id :: String.t(), type :: String.t()) :: t() | nil
+  def get_by_object_ap_id(object_id, type) do
     __MODULE__
     |> where(
       [a],
-      fragment("COALESCE(?->'object'->>'id', ?->>'object')", a.data, a.data) == ^object_id
+      fragment("?->>'type'", a.data) == ^type and
+        fragment("COALESCE(?->'object'->>'id', ?->>'object')", a.data, a.data) == ^object_id
     )
     |> preload_object()
+    # todo: may not need to preload actor by default
+    |> preload_actor()
     |> Repo.one()
   end
 
@@ -90,6 +93,11 @@ defmodule Clacks.Activity do
     |> preload([_a, object: object], object: object)
   end
 
+  @spec preload_actor(Ecto.Queryable.t()) :: Ecto.Query.t()
+  def preload_actor(query) do
+    preload(query, :actor)
+  end
+
   @spec fetch(ap_id :: String.t()) :: t() | nil
   def fetch(ap_id) do
     case Clacks.ActivityPub.Fetcher.fetch_activity(ap_id) do

lib/clacks/activitypub/federator.ex

@@ -18,7 +18,7 @@ defmodule Clacks.ActivityPub.Federator do
     addressed_actors =
       if actor.data["followers"] in addressed do
         addressed = List.delete(addressed, actor.data["followers"])
-        [actor.followers | addressed]
+        actor.followers ++ addressed
       else
         addressed
       end

lib/clacks/inbox.ex

@@ -2,11 +2,21 @@ defmodule Clacks.Inbox do
   require Logger
   alias Clacks.{Repo, Activity, Object, Actor, ActivityPub, Notification}
 
+  @spec store_object(map()) :: {:ok, Object.t()}
+  defp store_object(%{"id" => ap_id} = object) do
+    changeset = Object.changeset(Object.get_cached_by_ap_id(ap_id) || %Object{}, %{data: object})
+    {:ok, object} = Repo.insert_or_update(changeset)
+    {:ok, object}
+  end
+
+  @spec store_activity(map(), boolean()) :: {:ok, Activity.t()} | {:error, term()}
   defp store_activity(%{"actor" => actor, "id" => ap_id} = activity, local \\ false)
        when is_binary(actor) do
+    # remove the embedded object (if there is one) from the activity
     activity_without_embedded_object =
       case Map.get(activity, "object") do
         %{"id" => object_id} ->
+          # todo: this assumes we already have stored the object
           Map.put(activity, "object", object_id)
 
         _ ->
@@ -17,7 +27,7 @@ defmodule Clacks.Inbox do
       Activity.changeset(Activity.get_cached_by_ap_id(ap_id) || %Activity{}, %{
         data: activity_without_embedded_object,
         local: local,
-        actor: actor
+        actor_ap_id: actor
       })
 
     case Repo.insert_or_update(changeset) do
@@ -35,9 +45,8 @@ defmodule Clacks.Inbox do
 
   def handle(%{"type" => "Create", "object" => object} = activity) do
     object = Clacks.Inbox.Transformer.restrict_incoming_object(object)
-    changeset = Object.changeset_for_creating(object)
 
-    with {:ok, _object} <- Repo.insert(changeset),
+    with {:ok, _object} <- store_object(object),
          {:ok, _activity} <- store_activity(activity) do
       :ok
     else
@@ -114,6 +123,33 @@ defmodule Clacks.Inbox do
     end
   end
 
+  def handle(
+        %{
+          "type" => "Undo",
+          "actor" => follower_id,
+          "object" => %{"type" => "Follow", "object" => followee_id}
+        } = activity
+      )
+      when is_binary(followee_id) do
+    followee = Actor.get_by_ap_id(followee_id)
+
+    store_activity(activity)
+
+    changeset =
+      Actor.changeset(followee, %{
+        followers: List.delete(followee.followers, follower_id)
+      })
+
+    case Repo.update(changeset) do
+      {:error, changeset} ->
+        Logger.error("Couldn't store updated followers: #{inspect(changeset)}")
+        {:error, "Couldn't store updated followers"}
+
+      {:ok, _followee} ->
+        :ok
+    end
+  end
+
   # as a fallback, just store the activity
   def handle(activity) do
     Logger.debug("Unhandled activity: #{inspect(activity)}")

lib/clacks/notification.ex

@@ -12,30 +12,37 @@ defmodule Clacks.Notification do
 
     belongs_to :user, Clacks.User
     belongs_to :activity, Clacks.Activity, type: FlakeId.Ecto.Type
+    belongs_to :referenced_activity, Clacks.Activity, type: FlakeId.Ecto.Type
 
     timestamps()
   end
 
   def changeset(%__MODULE__{} = schema, attrs) do
     schema
-    |> cast(attrs, [:type, :user_id, :activity_id])
-    |> validate_required([:type, :user_id])
+    |> cast(attrs, [:type, :user_id, :activity_id, :referenced_activity_id])
+    |> validate_required([:type, :user_id, :activity_id])
     |> validate_inclusion(:type, @valid_types)
   end
 
-  @spec create(type :: String.t(), activity :: Activity.t(), actor :: Actor.t()) ::
+  @spec create(
+          type :: String.t(),
+          activity :: Activity.t(),
+          actor :: Actor.t(),
+          referenced_activity :: Activity.t() | nil
+        ) ::
           {:ok, Notification.t()} | {:error, any()}
 
-  def create(type, _, _) when not (type in @valid_types) do
+  def create(type, _, _, _) when not (type in @valid_types) do
     {:error, "invalid notification type '#{type}'"}
   end
 
-  def create(type, activity, actor) do
+  def create(type, activity, actor, referenced_activity) do
     changeset =
       changeset(%__MODULE__{}, %{
         type: type,
         user_id: actor.user_id,
-        activity_id: activity.id
+        activity_id: activity.id,
+        referenced_activity_id: if(referenced_activity, do: referenced_activity.id, else: nil)
       })
 
     Repo.insert(changeset)
@@ -48,7 +55,7 @@ defmodule Clacks.Notification do
       ) do
     case Actor.get_cached_by_ap_id(followee_ap_id) do
       %Actor{local: true} = followee ->
-        create("follow", activity, followee)
+        create("follow", activity, followee, nil)
 
       _ ->
         :ok
@@ -62,7 +69,7 @@ defmodule Clacks.Notification do
     Enum.each(tags, fn %{"href" => mentioned_actor_id} ->
       case Actor.get_cached_by_ap_id(mentioned_actor_id) do
         %Actor{local: true} = mentioned_actor ->
-          create("mention", activity, mentioned_actor)
+          create("mention", activity, mentioned_actor, nil)
 
         _ ->
           :ok
@@ -74,11 +81,15 @@ defmodule Clacks.Notification do
         %Activity{data: %{"type" => type, "object" => original_object_ap_id}} = activity
       )
       when type in ["Announce", "Like"] do
-    with %Activity{local: true, actor: local_actor_id} <-
-           Activity.get_by_object_ap_id(original_object_ap_id),
-         %Actor{local: true} = original_activity_actor <-
-           Actor.get_cached_by_ap_id(local_actor_id) do
-      create(String.downcase(type), activity, original_activity_actor)
+    with %Activity{local: true, actor: %Actor{local: true} = original_activity_actor} =
+           original_activity <-
+           Activity.get_by_object_ap_id(original_object_ap_id, "Create") do
+      create(
+        String.downcase(type),
+        activity,
+        original_activity_actor,
+        original_activity
+      )
     else
       _ ->
         :ok

lib/clacks/object.ex

@@ -17,6 +17,7 @@ defmodule Clacks.Object do
     schema
     |> cast(attrs, [:data])
     |> validate_required([:data])
+    |> unique_constraint(:ap_id, name: :objects_unique_ap_id_index)
   end
 
   @spec changeset_for_creating(data :: map()) :: Ecto.Changeset.t()
@@ -55,7 +56,7 @@ defmodule Clacks.Object do
 
   @spec fetch(url :: String.t(), synthesize_create :: boolean(), return :: :object | :activity) ::
           t() | Activity.t() | nil
-  def fetch(url, synthesize_create \\ true, return \\ :object) do
+  def fetch(url, synthesize_create \\ true, return \\ :object) when is_binary(url) do
     case Clacks.ActivityPub.Fetcher.fetch_object(url) do
       nil ->
         nil
@@ -74,7 +75,7 @@ defmodule Clacks.Object do
             _ = Clacks.Actor.get_by_ap_id(actor)
 
             activity =
-              case Clacks.Activity.get_by_object_ap_id(ap_id) do
+              case Clacks.Activity.get_by_object_ap_id(ap_id, "Create") do
                 nil ->
                   if synthesize_create do
                     create = Clacks.ActivityPub.synthesized_create(data)

lib/clacks/timeline.ex

@@ -1,5 +1,5 @@
 defmodule Clacks.Timeline do
-  alias Clacks.{Repo, Actor, Activity, User, Notification}
+  alias Clacks.{Repo, Actor, Activity, Object, User, Notification}
   import Clacks.Paginator
   import Ecto.Query
 
@@ -82,34 +82,65 @@ defmodule Clacks.Timeline do
     |> Repo.all()
   end
 
-  @spec notifications(actor :: Actor.t(), params :: map()) :: [
-          {:follow, activity :: Activity.t(), actor :: Actor.t()}
-          | {:mention, activity :: Activity.t(), actor :: Actor.t()}
-          | {:announce, announce :: Activity.t(), announce_actor :: Actor.t(),
-             activity :: Activity.t(), actor :: Actor.t()}
-          | {:like, like :: Activity.t(), like_actor :: Actor.t(), activity :: Activity.t(),
-             actor :: Actor.t()}
-        ]
+  @spec notifications(Actor.t(), map()) :: [Notification.t()]
   def notifications(actor, params) do
     Notification
     |> where([n], n.user_id == ^actor.user_id)
-    |> join(:inner, [n], activity in Activity, on: activity.id == n.activity_id)
-    |> join(:inner, [n, activity], actor in Actor, on: activity.actor == actor.ap_id)
-    |> join(:left, [n, activity, actor], other in Activity,
-      on:
-        n.type in ["announce", "like"] and fragment("?->>'type'", other.data) == "Create" and
-          fragment("?->>'object'", activity.data) == fragment("?->'object'->>'id'", other.data)
-    )
     |> paginate(params)
     |> limit(^Map.get(params, "limit", 20))
+    |> join(:left, [n], a in Activity,
+      as: :activity,
+      on: a.id == n.activity_id
+    )
+    |> join(:left, [n], a in Activity,
+      as: :referenced_activity,
+      on: a.id == n.referenced_activity_id
+    )
+    |> join(:left, [n, activity: a], o in Object,
+      as: :object,
+      on:
+        fragment("?->>'id'", o.data) ==
+          fragment("COALESCE(?->'object'->>'id', ?->>'object')", a.data, a.data)
+    )
+    |> join(:left, [n, referenced_activity: a], o in Object,
+      as: :referenced_object,
+      on:
+        fragment("?->>'id'", o.data) ==
+          fragment("COALESCE(?->'object'->>'id', ?->>'object')", a.data, a.data)
+    )
+    |> join(:left, [n, activity: activity], actor in Actor,
+      as: :activity_actor,
+      on: activity.actor_ap_id == actor.ap_id
+    )
+    # note: we shouldn't need to load the actor for the referenced_activity here,
+    # because notifications for a given actor always reference activities from that same actor
     |> select(
-      [notification, activity, actor, original_activity],
-      {notification, activity, actor, original_activity}
+      [
+        n,
+        activity: activity,
+        referenced_activity: referenced_activity,
+        object: object,
+        referenced_object: referenced_object,
+        activity_actor: activity_actor
+      ],
+      {n, activity, object, activity_actor, referenced_activity, referenced_object}
     )
     |> Repo.all()
-    |> Enum.map(fn {notification, activity, actor, original_activity} ->
-      {String.to_existing_atom(notification.type), activity, actor, original_activity}
+    |> Enum.map(fn {n, activity, object, activity_actor, referenced_activity, referenced_object} ->
+      # unfortunately we can't use the select to set the associations because we need to handle the possibility of referenced_activity being nil
+      referenced_activity =
+        case referenced_activity do
+          nil -> nil
+          a -> %Activity{a | object: referenced_object}
+        end
+
+      %Notification{
+        n
+        | activity: %Activity{activity | object: object, actor: activity_actor},
+          referenced_activity: referenced_activity
+      }
     end)
+    |> IO.inspect()
   end
 
   defp restrict_to_actor(query, actor_id) do

lib/clacks/user_actions_helper.ex

@@ -35,7 +35,7 @@ defmodule Clacks.UserActionsHelper do
     note_changeset = Object.changeset_for_creating(note)
     {:ok, _object} = Repo.insert(note_changeset)
 
-    %{"id" => create_ap_id} = create = ActivityPub.internal_create(note, author.actor.ap_id)
+    %{"id" => create_ap_id} = create = ActivityPub.internal_create(note)
 
     case ActivityPub.Helper.save_and_federate(create, author.actor) do
       {:ok, activity} ->
@@ -48,12 +48,11 @@ defmodule Clacks.UserActionsHelper do
     end
   end
 
-  @spec get_addressed(String.t(), [{String.t(), Actor.t()}], String.t() | nil) ::
+  @spec get_addressed(User.t(), [{String.t(), Actor.t()}], String.t() | nil) ::
           {[String.t()], [String.t()]}
-  defp get_addressed(_author, mentions, in_reply_to_actor) do
+  defp get_addressed(author, mentions, in_reply_to_actor) do
     to = [@public | Enum.map(mentions, fn {_, actor} -> actor.ap_id end)]
-    # todo: followers
-    cc = []
+    cc = [Repo.preload(author, :actor).actor.data["followers"]]
 
     to =
       case in_reply_to_actor do

lib/clacks_web/controllers/objects_controller.ex

@@ -32,7 +32,7 @@ defmodule ClacksWeb.ObjectsController do
   defp redirect_to_status(conn, _params) do
     object_id = current_url(conn)
 
-    case Activity.get_by_object_ap_id(object_id) do
+    case Activity.get_by_object_ap_id(object_id, "Create") do
       nil ->
         conn
         |> resp(404, "Not Found")

lib/clacks_web/templates/frontend/_action_status.html.eex

@@ -11,7 +11,7 @@
       <%= past_tense %> by <a href="<%= local_actor_link(@action_actor) %>"><%= @action_actor.data["preferredUsername"] %></a>
     </p>
     <p class="status-meta-right">
-    <time datetime="<%= @action_activity.data["published"] %>"><%= display_timestamp(@action_activity.data["published"]) %></time>
+    <time datetime="<%= iso_datetime(@action_activity) %>"><%= display_timestamp(@action_activity) %></time>
     </p>
   </div>
   <div class="status-meta">
@@ -26,7 +26,7 @@
       </a>
     </h3>
     <p class="status-meta-right">
-      <time datetime="<%= @original_note["published"] %>"><%= display_timestamp(@original_note["published"]) %></time>
+      <time datetime="<%= iso_datetime(@original_activity) %>"><%= display_timestamp(@original_activity) %></time>
       <a href="<%= @original_note["url"] || @original_note["id"] %>" class="status-permalink">Permalink</a>
     </p>
   </div>

lib/clacks_web/templates/frontend/_follow_notification.html.eex

@@ -4,7 +4,7 @@
       Followed by <a href="<%= local_actor_link(@actor) %>"><%= @actor.data["preferredUsername"] %></a>
     </p>
     <p class="notification-info-right">
-      <time datetime="<%= @activity.data["published"] %>"><%= display_timestamp(@activity) %></time>
+      <time datetime="<%= iso_datetime(@activity) %>"><%= display_timestamp(@activity) %></time>
     </p>
   </div>
 </div>

lib/clacks_web/templates/frontend/_status.html.eex

@@ -13,7 +13,7 @@
       </h3>
     </div>
     <p class="status-meta-right">
-      <time datetime="<%= @note["published"] %>" class="dt-published"><%= display_timestamp(@note["published"]) %></time>
+    <time datetime="<%= iso_datetime(@note["published"]) %>" class="dt-published"><%= display_timestamp(@note["published"]) %></time>
       <a href="<%= @note["url"] || @note["id"] %>" class="status-permalink u-url">Permalink</a>
     </p>
   </div>

lib/clacks_web/templates/frontend/notifications.html.eex

@@ -11,17 +11,17 @@
   <%= for notification <- @notifications do %>
     <li>
       <%= case notification do %>
-        <% {:like, like_activity, actor, original_activity} -> %>
-          <%= render "_action_status.html", class: "notification", conn: @conn, action: :like, action_activity: like_activity, action_actor: actor, original_activity: original_activity, original_note: original_activity.data["object"], original_actor: @current_user.actor %>
+        <% %{type: "like", activity: %{actor: like_actor} = like_activity, referenced_activity: original_activity} -> %>
+          <%= render "_action_status.html", class: "notification", conn: @conn, action: :like, action_activity: like_activity, action_actor: like_actor, original_activity: original_activity, original_note: original_activity.object.data, original_actor: @current_user.actor %>
 
-        <% {:announce, announce_activity, actor, original_activity} -> %>
-          <%= render "_action_status.html", class: "notification", conn: @conn, action: :announce, action_activity: announce_activity, action_actor: actor, original_activity: original_activity, original_note: original_activity.data["object"], original_actor: @current_user.actor %>
+          <% %{type: "announce", activity: %{actor: announce_actor} = announce_activity, referenced_activity: original_activity} -> %>
+          <%= render "_action_status.html", class: "notification", conn: @conn, action: :announce, action_activity: announce_activity, action_actor: announce_actor, original_activity: original_activity, original_note: original_activity.object.data, original_actor: @current_user.actor %>
 
-        <% {:mention, mention_activity, actor, _} -> %>
-          <%= render "_status.html", class: "notification", conn: @conn, author: actor, status: mention_activity, note: mention_activity.data["object"] %>
+        <% %{type: "mention", activity: mention_activity} -> %>
+          <%= render "_status.html", class: "notification", conn: @conn, author: mention_activity.actor, status: mention_activity, note: mention_activity.object.data %>
 
-          <% {:follow, follow_activity, actor, _} -> %>
-            <%= render "_follow_notification.html", activity: follow_activity, actor: actor %>
+        <% %{type: "follow", activity: follow_activity} -> %>
+          <%= render "_follow_notification.html", activity: follow_activity, actor: follow_activity.actor %>
       <% end %>
     </li>
   <% end %>

lib/clacks_web/views/frontend_view.ex

@@ -82,6 +82,22 @@ defmodule ClacksWeb.FrontendView do
     end
   end
 
+  @spec iso_datetime(datetime :: String.t() | DateTime.t() | NaiveDateTime.t() | Activity.t()) ::
+          String.t()
+
+  def iso_datetime(str) when is_binary(str) do
+    str
+  end
+
+  def iso_datetime(%Activity{data: data, inserted_at: inserted_at}) do
+    iso_datetime(Map.get(data, "published", inserted_at))
+  end
+
+  def iso_datetime(%{__struct__: struct} = datetime)
+      when struct == DateTime or struct == NaiveDateTime do
+    Timex.format!(datetime, "{ISO:Extended:Z}")
+  end
+
   @spec prev_page_path(conn :: Plug.Conn.t(), [
           Activity.t() | {Activity.t(), Actor.t()} | Notification.t()
         ]) ::

priv/repo/migrations/20210826220029_more_belongs_to.exs

@@ -0,0 +1,14 @@
+defmodule Clacks.Repo.Migrations.MoreBelongsTo do
+  use Ecto.Migration
+
+  def change do
+    alter table(:notifications) do
+      add :referenced_activity_id, references(:activities, type: :uuid)
+    end
+
+    alter table(:activities) do
+      remove :actor
+      add :actor_ap_id, references(:actors, column: :ap_id, type: :string)
+    end
+  end
+end