Ingestion Pipeline: Listen

lib/pleroma/web/activity_pub/activity_pub.ex

@@ -20,6 +20,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
   alias Pleroma.Repo
   alias Pleroma.Upload
   alias Pleroma.User
+  alias Pleroma.Web.ActivityPub.Builder
   alias Pleroma.Web.ActivityPub.MRF
   alias Pleroma.Web.ActivityPub.Transmogrifier
   alias Pleroma.Web.Streamer
@@ -310,7 +311,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
     published = params[:published]
 
     listen_data =
-      make_listen_data(
+      Builder.listen(
         %{to: to, actor: actor, published: published, context: context, object: object},
         additional
       )

lib/pleroma/web/activity_pub/builder.ex

@@ -305,4 +305,17 @@ defmodule Pleroma.Web.ActivityPub.Builder 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(params, additional) do
+    %{
+      "type" => "Listen",
+      "id" => Utils.generate_activity_id(),
+      "to" => params.to |> Enum.uniq(),
+      "actor" => params.actor.ap_id,
+      "object" => params.object,
+      "published" => 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
@@ -99,6 +100,21 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidator do
   end
 
   def validate(
+        %{"type" => "Listen", "object" => %{"type" => "Audio"} = object} = activity,
+        meta
+      ) do
+    with {:ok, object_data} <- cast_and_apply(object),
+         meta = Keyword.put(meta, :object_data, object_data |> stringify_keys),
+         {: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 +159,7 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidator do
 
   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 +171,8 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidator do
         "Announce" -> AnnounceValidator
         "ChatMessage" -> ChatMessageValidator
         "Answer" -> AnswerValidator
+        "Add" -> AddRemoveValidator
+        "Remove" -> AddRemoveValidator
       end
 
     with {:ok, object} <-
@@ -166,16 +184,6 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidator do
     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/audio_video_validator.ex

@@ -50,6 +50,12 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidators.AudioVideoValidator do
 
     field(:likes, {:array, ObjectValidators.ObjectID}, default: [])
     field(:announcements, {:array, ObjectValidators.ObjectID}, default: [])
+
+    # Used by Pleroma's Listen-Audio Scrobbler
+    field(:title, :string)
+    field(:artist, :string)
+    field(:album, :string)
+    field(:length, :integer)
   end
 
   def cast_and_apply(data) do

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

@@ -0,0 +1,61 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2021 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
+
+  embedded_schema do
+    field(:id, ObjectValidators.ObjectID, primary_key: true)
+    field(:type, :string)
+    field(:published, ObjectValidators.DateTime)
+    field(:object, ObjectValidators.ObjectID)
+    field(:actor, ObjectValidators.ObjectID)
+    field(:context, :string)
+    field(:to, ObjectValidators.Recipients, default: [])
+    field(:cc, ObjectValidators.Recipients, default: [])
+    field(:bto, ObjectValidators.Recipients, default: [])
+    field(:bcc, ObjectValidators.Recipients, default: [])
+  end
+
+  def changeset(struct, data) do
+    struct
+    |> cast(data, __schema__(:fields))
+  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])
+    |> CommonValidations.validate_actor_presence()
+  end
+end

lib/pleroma/web/activity_pub/side_effects.ex

@@ -231,6 +231,18 @@ defmodule Pleroma.Web.ActivityPub.SideEffects 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], 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 @@ defmodule Pleroma.Web.ActivityPub.Transmogrifier do
   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" => "❤️",
@@ -493,6 +462,17 @@ defmodule Pleroma.Web.ActivityPub.Transmogrifier do
   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
       ) do

lib/pleroma/web/activity_pub/utils.ex

@@ -675,21 +675,6 @@ defmodule Pleroma.Web.ActivityPub.Utils 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

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-2021 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.Object
+  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"
+
+    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
+end