Ingestion Pipeline: Listen

lib/pleroma/web/activity_pub/activity_pub.ex

@@ -314,26 +314,6 @@ defp maybe_schedule_poll_notifications(activity) do
     :ok
   end
 
-  @spec listen(map()) :: {:ok, Activity.t()} | {:error, any()}
-  def listen(%{to: to, actor: actor, context: context, object: object} = params) do
-    additional = params[:additional] || %{}
-    # only accept false as false value
-    local = !(params[:local] == false)
-    published = params[:published]
-
-    listen_data =
-      make_listen_data(
-        %{to: to, actor: actor, published: published, context: context, object: object},
-        additional
-      )
-
-    with {:ok, activity} <- insert(listen_data, local),
-         _ <- notify_and_stream(activity),
-         :ok <- maybe_federate(activity) do
-      {:ok, activity}
-    end
-  end
-
   @spec unfollow(User.t(), User.t(), String.t() | nil, boolean()) ::
           {:ok, Activity.t()} | nil | {:error, any()}
   def unfollow(follower, followed, activity_id \\ nil, local \\ true) do

lib/pleroma/web/activity_pub/builder.ex

@@ -337,4 +337,18 @@ def unpin(%User{} = user, object) do
   defp pinned_url(nickname) when is_binary(nickname) do
     Pleroma.Web.Router.Helpers.activity_pub_url(Pleroma.Web.Endpoint, :pinned, nickname)
   end
+
+  def listen(%{to: to, actor: %{ap_id: actor}, object: object} = params, additional) do
+    {:ok,
+     %{
+       "type" => "Listen",
+       "id" => Utils.generate_activity_id(),
+       "to" => to |> Enum.uniq(),
+       "actor" => actor,
+       "object" => object,
+       "published" => Map.get(params, :published, Utils.make_date()),
+       "context" => params.context
+     }
+     |> Map.merge(additional), []}
+  end
 end

lib/pleroma/web/activity_pub/object_validator.ex

@@ -31,6 +31,7 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidator do
   alias Pleroma.Web.ActivityPub.ObjectValidators.EventValidator
   alias Pleroma.Web.ActivityPub.ObjectValidators.FollowValidator
   alias Pleroma.Web.ActivityPub.ObjectValidators.LikeValidator
+  alias Pleroma.Web.ActivityPub.ObjectValidators.ListenValidator
   alias Pleroma.Web.ActivityPub.ObjectValidators.QuestionValidator
   alias Pleroma.Web.ActivityPub.ObjectValidators.UndoValidator
   alias Pleroma.Web.ActivityPub.ObjectValidators.UpdateValidator
@@ -98,6 +99,19 @@ def validate(
     end
   end
 
+  def validate(
+        %{"type" => "Listen", "object" => %{"type" => "Audio"} = object} = activity,
+        meta
+      ) do
+    with {:ok, activity} <-
+           activity
+           |> ListenValidator.cast_and_validate(meta)
+           |> Ecto.Changeset.apply_action(:insert) do
+      activity = stringify_keys(activity)
+      {:ok, activity, meta}
+    end
+  end
+
   def validate(
         %{"type" => "Create", "object" => %{"type" => objtype} = object} = create_activity,
         meta
@@ -143,7 +157,7 @@ def validate(%{"type" => type} = object, meta)
 
   def validate(%{"type" => type} = object, meta)
       when type in ~w[Accept Reject Follow Update Like EmojiReact Announce
-      ChatMessage Answer] do
+      ChatMessage Answer Add Remove] do
     validator =
       case type do
         "Accept" -> AcceptRejectValidator
@@ -155,6 +169,8 @@ def validate(%{"type" => type} = object, meta)
         "Announce" -> AnnounceValidator
         "ChatMessage" -> ChatMessageValidator
         "Answer" -> AnswerValidator
+        "Add" -> AddRemoveValidator
+        "Remove" -> AddRemoveValidator
       end
 
     with {:ok, object} <-
@@ -166,16 +182,6 @@ def validate(%{"type" => type} = object, meta)
     end
   end
 
-  def validate(%{"type" => type} = object, meta) when type in ~w(Add Remove) do
-    with {:ok, object} <-
-           object
-           |> AddRemoveValidator.cast_and_validate()
-           |> Ecto.Changeset.apply_action(:insert) do
-      object = stringify_keys(object)
-      {:ok, object, meta}
-    end
-  end
-
   def validate(o, m), do: {:error, {:validator_not_set, {o, m}}}
 
   def cast_and_apply(%{"type" => "ChatMessage"} = object) do

lib/pleroma/web/activity_pub/object_validators/listen_validator.ex

@@ -0,0 +1,82 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Web.ActivityPub.ObjectValidators.ListenValidator do
+  use Ecto.Schema
+
+  alias Pleroma.EctoType.ActivityPub.ObjectValidators
+  alias Pleroma.Web.ActivityPub.ObjectValidators.CommonFixes
+  alias Pleroma.Web.ActivityPub.ObjectValidators.CommonValidations
+
+  import Ecto.Changeset
+
+  @primary_key false
+  @derive Jason.Encoder
+
+  embedded_schema do
+    quote do
+      unquote do
+        import Elixir.Pleroma.Web.ActivityPub.ObjectValidators.CommonFields
+        message_fields()
+      end
+    end
+
+    field(:actor, ObjectValidators.ObjectID)
+    field(:published, ObjectValidators.DateTime)
+
+    embeds_one :object, UrlObjectValidator, primary_key: false do
+      field(:type, :string)
+
+      field(:to, ObjectValidators.Recipients, default: [])
+      field(:cc, ObjectValidators.Recipients, default: [])
+      field(:bto, ObjectValidators.Recipients, default: [])
+      field(:bcc, ObjectValidators.Recipients, default: [])
+
+      field(:title, :string)
+      field(:artist, :string)
+      field(:album, :string)
+      field(:length, :integer)
+    end
+  end
+
+  def changeset(struct, data) do
+    struct
+    |> cast(data, __schema__(:fields) -- [:object])
+    |> cast_embed(:object, with: &audio_changeset/2)
+  end
+
+  def audio_changeset(struct, data) do
+    struct
+    |> cast(data, Map.keys(struct) -- [:__struct__])
+    |> validate_inclusion(:type, ["Audio"])
+  end
+
+  def cast_data(data, meta \\ []) do
+    data = fix(data, meta)
+
+    %__MODULE__{}
+    |> changeset(data)
+  end
+
+  def cast_and_validate(data, meta \\ []) do
+    data
+    |> cast_data(meta)
+    |> validate_data(meta)
+  end
+
+  defp fix(data, _meta) do
+    data
+    |> CommonFixes.fix_actor()
+    |> CommonFixes.fix_activity_addressing()
+  end
+
+  defp validate_data(data_cng, _meta) do
+    # TODO: Restrict to Audio objects
+
+    data_cng
+    |> validate_inclusion(:type, ["Listen"])
+    |> validate_required([:id, :type, :object, :actor, :to, :cc, :published])
+    |> CommonValidations.validate_actor_presence()
+  end
+end

lib/pleroma/web/activity_pub/side_effects.ex

@@ -234,6 +234,18 @@ def handle(%{data: %{"type" => "Create"}} = activity, meta) do
     end
   end
 
+  # Tasks this handles
+  # - Actually create object
+  # - Rollback if we couldn't create it
+  @impl true
+  def handle(%{data: %{"type" => "Listen"}} = activity, meta) do
+    with {:ok, _object, meta} <- handle_object_creation(meta[:object_data], activity, meta) do
+      {:ok, activity, meta}
+    else
+      e -> Repo.rollback(e)
+    end
+  end
+
   # Tasks this handles:
   # - Add announce to object
   # - Set up notification

lib/pleroma/web/activity_pub/transmogrifier.ex

@@ -384,37 +384,6 @@ def handle_incoming(%{"id" => ""}, _options), do: :error
   def handle_incoming(%{"id" => id}, _options) when is_binary(id) and byte_size(id) < 8,
     do: :error
 
-  def handle_incoming(
-        %{"type" => "Listen", "object" => %{"type" => "Audio"} = object} = data,
-        options
-      ) do
-    actor = Containment.get_actor(data)
-
-    data =
-      Map.put(data, "actor", actor)
-      |> fix_addressing
-
-    with {:ok, %User{} = user} <- User.get_or_fetch_by_ap_id(data["actor"]) do
-      reply_depth = (options[:depth] || 0) + 1
-      options = Keyword.put(options, :depth, reply_depth)
-      object = fix_object(object, options)
-
-      params = %{
-        to: data["to"],
-        object: object,
-        actor: user,
-        context: nil,
-        local: false,
-        published: data["published"],
-        additional: Map.take(data, ["cc", "id"])
-      }
-
-      ActivityPub.listen(params)
-    else
-      _e -> :error
-    end
-  end
-
   @misskey_reactions %{
     "like" => "👍",
     "love" => "❤️",
@@ -492,6 +461,17 @@ def handle_incoming(
     end
   end
 
+  def handle_incoming(
+        %{"type" => "Listen", "object" => %{"type" => "Audio"}} = data,
+        _options
+      ) do
+    with {:ok, %User{}} <- ObjectValidator.fetch_actor(data),
+         {:ok, activity, _} <-
+           Pipeline.common_pipeline(data, local: false) do
+      {:ok, activity}
+    end
+  end
+
   def handle_incoming(
         %{"type" => "Delete"} = data,
         _options
@@ -694,8 +674,7 @@ def prepare_object(object) do
   #  internal -> Mastodon
   #  """
 
-  def prepare_outgoing(%{"type" => activity_type, "object" => object_id} = data)
-      when activity_type in ["Create", "Listen"] do
+  def prepare_outgoing(%{"type" => "Create", "object" => object_id} = data) do
     object =
       object_id
       |> Object.normalize(fetch: false)
@@ -711,6 +690,20 @@ def prepare_outgoing(%{"type" => activity_type, "object" => object_id} = data)
     {:ok, data}
   end
 
+  def prepare_outgoing(%{"type" => "Listen", "object" => object} = data) do
+    object =
+      object
+      |> prepare_object
+
+    data =
+      data
+      |> Map.put("object", object)
+      |> Map.merge(Utils.make_json_ld_header())
+      |> Map.delete("bcc")
+
+    {:ok, data}
+  end
+
   def prepare_outgoing(%{"type" => "Announce", "actor" => ap_id, "object" => object_id} = data) do
     object =
       object_id

lib/pleroma/web/activity_pub/utils.ex

@@ -23,16 +23,7 @@ defmodule Pleroma.Web.ActivityPub.Utils do
   require Logger
   require Pleroma.Constants
 
-  @supported_object_types [
-    "Article",
-    "Note",
-    "Event",
-    "Video",
-    "Page",
-    "Question",
-    "Answer",
-    "Audio"
-  ]
+  @supported_object_types ~w[Article Note Event Video Page Question Answer Audio]
   @strip_status_report_states ~w(closed resolved)
   @supported_report_states ~w(open closed resolved)
   @valid_visibilities ~w(public unlisted private direct)
@@ -675,21 +666,6 @@ def make_create_data(params, additional) do
     |> Map.merge(additional)
   end
 
-  #### Listen-related helpers
-  def make_listen_data(params, additional) do
-    published = params.published || make_date()
-
-    %{
-      "type" => "Listen",
-      "to" => params.to |> Enum.uniq(),
-      "actor" => params.actor.ap_id,
-      "object" => params.object,
-      "published" => published,
-      "context" => params.context
-    }
-    |> Map.merge(additional)
-  end
-
   #### Flag-related helpers
   @spec make_flag_data(map(), map()) :: map()
   def make_flag_data(%{actor: actor, context: context, content: content} = params, additional) do

lib/pleroma/web/common_api.ex

@@ -391,8 +391,14 @@ def check_expiry_date(expiry_str) do
   end
 
   def listen(user, data) do
-    with {:ok, draft} <- ActivityDraft.listen(user, data) do
-      ActivityPub.listen(draft.changes)
+    with {_, {:ok, %{changes: draft}}} <- {:draft, ActivityDraft.listen(user, data)},
+         {_, {:ok, activity_data, []}} <- {:builder, Builder.listen(draft, draft.additional)},
+         {_, {:ok, activity, _}} <-
+           {:pipeline, Pipeline.common_pipeline(activity_data, local: true)},
+         {_, %Activity{} = activity} <- {:norm, Activity.normalize(activity)} do
+      {:ok, activity}
+    else
+      e -> {:error, e}
     end
   end
 

lib/pleroma/web/pleroma_api/views/scrobble_view.ex

@@ -9,25 +9,35 @@ defmodule Pleroma.Web.PleromaAPI.ScrobbleView do
 
   alias Pleroma.Activity
   alias Pleroma.HTML
-  alias Pleroma.Object
   alias Pleroma.Web.CommonAPI
   alias Pleroma.Web.CommonAPI.Utils
   alias Pleroma.Web.MastodonAPI.AccountView
 
-  def render("show.json", %{activity: %Activity{data: %{"type" => "Listen"}} = activity} = opts) do
-    object = Object.normalize(activity, fetch: false)
-
-    user = CommonAPI.get_user(activity.data["actor"])
-    created_at = Utils.to_masto_date(activity.data["published"])
+  def render(
+        "show.json",
+        %{
+          activity: %Activity{
+            id: id,
+            data: %{
+              "type" => "Listen",
+              "actor" => actor,
+              "published" => published,
+              "object" => object
+            }
+          }
+        } = opts
+      ) do
+    user = CommonAPI.get_user(actor)
+    created_at = Utils.to_masto_date(published)
 
     %{
-      id: activity.id,
+      id: id,
       account: AccountView.render("show.json", %{user: user, for: opts[:for]}),
       created_at: created_at,
-      title: object.data["title"] |> HTML.strip_tags(),
-      artist: object.data["artist"] |> HTML.strip_tags(),
-      album: object.data["album"] |> HTML.strip_tags(),
-      length: object.data["length"]
+      title: object["title"] |> HTML.strip_tags(),
+      artist: object["artist"] |> HTML.strip_tags(),
+      album: object["album"] |> HTML.strip_tags(),
+      length: object["length"]
     }
   end
 

test/pleroma/web/activity_pub/activity_pub_test.exs

@@ -601,30 +601,6 @@ test "adds an id to a given object if it lacks one and is a note and inserts it
   end
 
   describe "listen activities" do
-    test "does not increase user note count" do
-      user = insert(:user)
-
-      {:ok, activity} =
-        ActivityPub.listen(%{
-          to: ["https://www.w3.org/ns/activitystreams#Public"],
-          actor: user,
-          context: "",
-          object: %{
-            "actor" => user.ap_id,
-            "to" => ["https://www.w3.org/ns/activitystreams#Public"],
-            "artist" => "lain",
-            "title" => "lain radio episode 1",
-            "length" => 180_000,
-            "type" => "Audio"
-          }
-        })
-
-      assert activity.actor == user.ap_id
-
-      user = User.get_cached_by_id(user.id)
-      assert user.note_count == 0
-    end
-
     test "can be fetched into a timeline" do
       _listen_activity_1 = insert(:listen)
       _listen_activity_2 = insert(:listen)

test/pleroma/web/activity_pub/transmogrifier/audio_handling_test.exs

@@ -12,39 +12,6 @@ defmodule Pleroma.Web.ActivityPub.Transmogrifier.AudioHandlingTest do
 
   import Pleroma.Factory
 
-  test "it works for incoming listens" do
-    _user = insert(:user, ap_id: "http://mastodon.example.org/users/admin")
-
-    data = %{
-      "@context" => "https://www.w3.org/ns/activitystreams",
-      "to" => ["https://www.w3.org/ns/activitystreams#Public"],
-      "cc" => [],
-      "type" => "Listen",
-      "id" => "http://mastodon.example.org/users/admin/listens/1234/activity",
-      "actor" => "http://mastodon.example.org/users/admin",
-      "object" => %{
-        "type" => "Audio",
-        "to" => ["https://www.w3.org/ns/activitystreams#Public"],
-        "cc" => [],
-        "id" => "http://mastodon.example.org/users/admin/listens/1234",
-        "attributedTo" => "http://mastodon.example.org/users/admin",
-        "title" => "lain radio episode 1",
-        "artist" => "lain",
-        "album" => "lain radio",
-        "length" => 180_000
-      }
-    }
-
-    {:ok, %Activity{local: false} = activity} = Transmogrifier.handle_incoming(data)
-
-    object = Object.normalize(activity, fetch: false)
-
-    assert object.data["title"] == "lain radio episode 1"
-    assert object.data["artist"] == "lain"
-    assert object.data["album"] == "lain radio"
-    assert object.data["length"] == 180_000
-  end
-
   test "Funkwhale Audio object" do
     Tesla.Mock.mock(fn
       %{url: "https://channels.tests.funkwhale.audio/federation/actors/compositions"} ->

test/pleroma/web/activity_pub/transmogrifier/listen_handling_test.exs

@@ -0,0 +1,59 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Web.ActivityPub.Transmogrifier.ListenHandlingTest do
+  use Pleroma.DataCase, async: true
+
+  alias Pleroma.Activity
+  alias Pleroma.Web.ActivityPub.Transmogrifier
+
+  import Pleroma.Factory
+
+  test "it works for incoming listens" do
+    _user = insert(:user, ap_id: "http://mastodon.example.org/users/admin")
+
+    audio_data = %{
+      "type" => "Audio",
+      "to" => ["https://www.w3.org/ns/activitystreams#Public"],
+      "cc" => [],
+      "id" => "http://mastodon.example.org/users/admin/listens/1234",
+      "attributedTo" => "http://mastodon.example.org/users/admin",
+      "title" => "lain radio episode 1",
+      "artist" => "lain",
+      "album" => "lain radio",
+      "length" => 180_000
+    }
+
+    data = %{
+      "@context" => "https://www.w3.org/ns/activitystreams",
+      "to" => ["https://www.w3.org/ns/activitystreams#Public"],
+      "cc" => [],
+      "type" => "Listen",
+      "id" => "http://mastodon.example.org/users/admin/listens/1234/activity",
+      "actor" => "http://mastodon.example.org/users/admin",
+      "object" => audio_data
+    }
+
+    Tesla.Mock.mock(fn
+      %{url: "http://mastodon.example.org/users/admin/listens/1234"} ->
+        %Tesla.Env{
+          status: 200,
+          body: audio_data,
+          headers: HttpRequestMock.activitypub_object_headers()
+        }
+    end)
+
+    {:ok, %Activity{local: false} = activity} = Transmogrifier.handle_incoming(data)
+
+    assert activity.data["type"] == "Listen"
+
+    assert object = activity.data["object"]
+
+    assert object["type"] == "Audio"
+    assert object["title"] == "lain radio episode 1"
+    assert object["artist"] == "lain"
+    assert object["album"] == "lain radio"
+    assert object["length"] == 180_000
+  end
+end

test/pleroma/web/activity_pub/transmogrifier_test.exs

@@ -295,7 +295,7 @@ test "it can handle Listen activities" do
 
       {:ok, activity} = CommonAPI.listen(user, %{"title" => "lain radio episode 1"})
 
-      {:ok, _modified} = Transmogrifier.prepare_outgoing(activity.data)
+      {:ok, _modified} = Transmogrifier.prepare_outgoing(activity)
     end
 
     test "custom emoji urls are URI encoded" do

test/pleroma/web/common_api_test.exs

@@ -1371,9 +1371,9 @@ test "returns a valid activity" do
           length: 180_000
         })
 
-      object = Object.normalize(activity, fetch: false)
+      assert object = activity.data["object"]
 
-      assert object.data["title"] == "lain radio episode 1"
+      assert object["title"] == "lain radio episode 1"
 
       assert Visibility.get_visibility(activity) == "public"
     end
@@ -1381,7 +1381,7 @@ test "returns a valid activity" do
     test "respects visibility=private" do
       user = insert(:user)
 
-      {:ok, activity} =
+      {:ok, %Activity{} = activity} =
         CommonAPI.listen(user, %{
           title: "lain radio episode 1",
           album: "lain radio",
@@ -1390,12 +1390,24 @@ test "respects visibility=private" do
           visibility: "private"
         })
 
-      object = Object.normalize(activity, fetch: false)
+      assert object = activity.data["object"]
 
-      assert object.data["title"] == "lain radio episode 1"
+      assert object["title"] == "lain radio episode 1"
 
       assert Visibility.get_visibility(activity) == "private"
     end
+
+    test "does not increase user note count" do
+      user = insert(:user)
+
+      {:ok, %Activity{actor: actor}} =
+        CommonAPI.listen(user, %{artist: "lain", title: "lain radio episode 1", length: 80_000})
+
+      assert actor == user.ap_id
+
+      user = User.get_cached_by_id(user.id)
+      assert user.note_count == 0
+    end
   end
 
   describe "get_user/1" do

test/support/http_request_mock.ex

@@ -1075,6 +1075,14 @@ def get("http://404.site" <> _, _, _, _) do
      }}
   end
 
+  def get("https://404.site" <> _, _, _, _) do
+    {:ok,
+     %Tesla.Env{
+       status: 404,
+       body: ""
+     }}
+  end
+
   def get(
         "https://zetsubou.xn--q9jyb4c/.well-known/webfinger?resource=acct:lain@zetsubou.xn--q9jyb4c",
         _,