Add recipients field to activities.

Also do some very basic checks for AP message insertion.

lib/pleroma/activity.ex

@@ -7,6 +7,7 @@ defmodule Pleroma.Activity do
     field :data, :map
     field :local, :boolean, default: true
     field :actor, :string
+    field :recipients, {:array, :string}
     has_many :notifications, Notification, on_delete: :delete_all
 
     timestamps()

lib/pleroma/web/activity_pub/activity_pub.ex

@@ -1,14 +1,19 @@
 defmodule Pleroma.Web.ActivityPub.ActivityPub do
   alias Pleroma.{Activity, Repo, Object, Upload, User, Notification}
+  alias Pleroma.Web.OStatus
   import Ecto.Query
   import Pleroma.Web.ActivityPub.Utils
   require Logger
 
+  def get_recipients(data) do
+    (data["to"] || []) ++ (data["cc"] || [])
+  end
+
   def insert(map, local \\ true) when is_map(map) do
     with nil <- Activity.get_by_ap_id(map["id"]),
          map <- lazy_put_activity_defaults(map),
          :ok <- insert_full_object(map) do
-      {:ok, activity} = Repo.insert(%Activity{data: map, local: local, actor: map["actor"]})
+      {:ok, activity} = Repo.insert(%Activity{data: map, local: local, actor: map["actor"], recipients: get_recipients(map)})
       Notification.create_notifications(activity)
       stream_out(activity)
       {:ok, activity}
@@ -215,4 +220,16 @@ def upload(file) do
     data = Upload.store(file)
     Repo.insert(%Object{data: data})
   end
+
+  def prepare_incoming(%{"type" => "Create", "object" => %{"type" => "Note"} = object} = data) do
+    with {:ok, user} <- OStatus.find_or_make_user(data["actor"]) do
+      data
+    else
+      _e -> :error
+    end
+  end
+
+  def prepare_incoming(_) do
+    :error
+  end
 end

lib/pleroma/web/activity_pub/activity_pub_controller.ex

@@ -20,7 +20,9 @@ def object(conn, %{"uuid" => uuid}) do
 
   # TODO: Move signature failure halt into plug
   def inbox(%{assigns: %{valid_signature: true}} = conn, params) do
-    {:ok, activity} = ActivityPub.insert(params, false)
+    with {:ok, data} <- ActivityPub.prepare_incoming(params),
+         {:ok, activity} <- ActivityPub.insert(data, false) do
       json(conn, "ok")
     end
   end
+end

priv/repo/migrations/20171212163643_add_recipients_to_activities.exs

@@ -0,0 +1,11 @@
+defmodule Pleroma.Repo.Migrations.AddRecipientsToActivities do
+  use Ecto.Migration
+
+  def change do
+    alter table(:activities) do
+      add :recipients, {:array, :string}
+    end
+
+    create index(:activities, [:recipients], using: :gin)
+  end
+end

priv/repo/migrations/20171212164525_fill_recipients_in_activities.exs

@@ -0,0 +1,21 @@
+defmodule Pleroma.Repo.Migrations.FillRecipientsInActivities do
+  use Ecto.Migration
+  alias Pleroma.{Repo, Activity}
+
+  def up do
+    max = Repo.aggregate(Activity, :max, :id)
+    if max do
+      IO.puts("#{max} activities")
+      chunks = 0..(round(max / 10_000))
+
+      Enum.each(chunks, fn (i) ->
+        min = i * 10_000
+        max = min + 10_000
+        execute("""
+        update activities set recipients = array(select jsonb_array_elements_text(data->'to')) where id > #{min} and id <= #{max};
+        """)
+        |> IO.inspect
+      end)
+    end
+  end
+end

test/web/activity_pub/activity_pub_test.exs

@@ -53,6 +53,7 @@ test "removes doubled 'to' recipients" do
       {:ok, activity} = ActivityPub.create(["user1", "user1", "user2"], %User{ap_id: "1"}, "", %{})
       assert activity.data["to"] == ["user1", "user2"]
       assert activity.actor == "1"
+      assert activity.recipients == ["user1", "user2"]
     end
   end