No OneTemporary
Actions

Size

42 KB

Referenced Files

None

Subscribers

None

View Options

	diff --git a/lib/pleroma/notification.ex b/lib/pleroma/notification.ex
	index 654c9c98d..a80279fa6 100644
	--- a/lib/pleroma/notification.ex
	+++ b/lib/pleroma/notification.ex
	@@ -1,746 +1,753 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Notification do
	use Ecto.Schema

	alias Ecto.Multi
	alias Pleroma.Activity
	alias Pleroma.FollowingRelationship
	alias Pleroma.Marker
	alias Pleroma.Notification
	alias Pleroma.Object
	alias Pleroma.Pagination
	alias Pleroma.Repo
	alias Pleroma.ThreadMute
	alias Pleroma.User
	alias Pleroma.Web.CommonAPI
	alias Pleroma.Web.CommonAPI.Utils
	alias Pleroma.Web.Push
	alias Pleroma.Web.Streamer

	import Ecto.Query
	import Ecto.Changeset

	require Logger

	@type t :: %__MODULE__{}

	@include_muted_option :with_muted

	schema "notifications" do
	field(:seen, :boolean, default: false)
	# This is an enum type in the database. If you add a new notification type,
	# remember to add a migration to add it to the `notifications_type` enum
	# as well.
	field(:type, :string)
	belongs_to(:user, User, type: FlakeId.Ecto.CompatType)
	belongs_to(:activity, Activity, type: FlakeId.Ecto.CompatType)

	timestamps()
	end

	def update_notification_type(user, activity) do
	with %__MODULE__{} = notification <-
	Repo.get_by(__MODULE__, user_id: user.id, activity_id: activity.id) do
	type =
	activity
	\|> type_from_activity()

	notification
	\|> changeset(%{type: type})
	\|> Repo.update()
	end
	end

	@spec unread_notifications_count(User.t()) :: integer()
	def unread_notifications_count(%User{id: user_id}) do
	from(q in __MODULE__,
	where: q.user_id == ^user_id and q.seen == false
	)
	\|> Repo.aggregate(:count, :id)
	end

	@notification_types ~w{
	favourite
	follow
	follow_request
	mention
	move
	pleroma:chat_mention
	pleroma:emoji_reaction
	pleroma:report
	reblog
	poll
	}

	def changeset(%Notification{} = notification, attrs) do
	notification
	\|> cast(attrs, [:seen, :type])
	\|> validate_inclusion(:type, @notification_types)
	end

	@spec last_read_query(User.t()) :: Ecto.Queryable.t()
	def last_read_query(user) do
	from(q in Pleroma.Notification,
	where: q.user_id == ^user.id,
	where: q.seen == true,
	select: type(q.id, :string),
	limit: 1,
	order_by: fragment("? desc nulls last", q.id)
	)
	end

	defp for_user_query_ap_id_opts(user, opts) do
	ap_id_relationships =
	[:block] ++
	if opts[@include_muted_option], do: [], else: [:notification_mute]

	preloaded_ap_ids = User.outgoing_relationships_ap_ids(user, ap_id_relationships)

	exclude_blocked_opts = Map.merge(%{blocked_users_ap_ids: preloaded_ap_ids[:block]}, opts)

	exclude_notification_muted_opts =
	Map.merge(%{notification_muted_users_ap_ids: preloaded_ap_ids[:notification_mute]}, opts)

	{exclude_blocked_opts, exclude_notification_muted_opts}
	end

	def for_user_query(user, opts \\ %{}) do
	{exclude_blocked_opts, exclude_notification_muted_opts} =
	for_user_query_ap_id_opts(user, opts)

	Notification
	\|> where(user_id: ^user.id)
	\|> join(:inner, [n], activity in assoc(n, :activity))
	\|> join(:left, [n, a], object in Object,
	on:
	fragment(
	"(?->>'id') = associated_object_id(?)",
	object.data,
	a.data
	)
	)
	\|> join(:inner, [_n, a], u in User, on: u.ap_id == a.actor, as: :user_actor)
	\|> preload([n, a, o], activity: {a, object: o})
	\|> where([user_actor: user_actor], user_actor.is_active)
	\|> exclude_notification_muted(user, exclude_notification_muted_opts)
	\|> exclude_blocked(user, exclude_blocked_opts)
	\|> exclude_blockers(user)
	\|> exclude_filtered(user)
	\|> exclude_visibility(opts)
	end

	# Excludes blocked users and non-followed domain-blocked users
	defp exclude_blocked(query, user, opts) do
	blocked_ap_ids = opts[:blocked_users_ap_ids] \|\| User.blocked_users_ap_ids(user)

	query
	\|> where([..., user_actor: user_actor], user_actor.ap_id not in ^blocked_ap_ids)
	\|> FollowingRelationship.keep_following_or_not_domain_blocked(user)
	end

	defp exclude_blockers(query, user) do
	if Pleroma.Config.get([:activitypub, :blockers_visible]) == true do
	query
	else
	blocker_ap_ids = User.incoming_relationships_ungrouped_ap_ids(user, [:block])

	query
	\|> where([..., user_actor: user_actor], user_actor.ap_id not in ^blocker_ap_ids)
	end
	end

	defp exclude_notification_muted(query, _, %{@include_muted_option => true}) do
	query
	end

	defp exclude_notification_muted(query, user, opts) do
	notification_muted_ap_ids =
	opts[:notification_muted_users_ap_ids] \|\| User.notification_muted_users_ap_ids(user)

	query
	\|> where([..., user_actor: user_actor], user_actor.ap_id not in ^notification_muted_ap_ids)
	\|> join(:left, [n, a], tm in ThreadMute,
	on: tm.user_id == ^user.id and tm.context == fragment("?->>'context'", a.data),
	as: :thread_mute
	)
	\|> where([thread_mute: thread_mute], is_nil(thread_mute.user_id))
	end

	defp exclude_filtered(query, user) do
	case Pleroma.Filter.compose_regex(user) do
	nil ->
	query

	regex ->
	from([_n, a, o] in query,
	where:
	fragment("not(?->>'content' ~* ?)", o.data, ^regex) or
	fragment("?->>'content' is null", o.data) or
	fragment("?->>'actor' = ?", o.data, ^user.ap_id)
	)
	end
	end

	@valid_visibilities ~w[direct unlisted public private]

	defp exclude_visibility(query, %{exclude_visibilities: visibility})
	when is_list(visibility) do
	if Enum.all?(visibility, &(&1 in @valid_visibilities)) do
	query
	\|> join(:left, [n, a], mutated_activity in Pleroma.Activity,
	on:
	fragment(
	"associated_object_id(?)",
	a.data
	) ==
	fragment(
	"associated_object_id(?)",
	mutated_activity.data
	) and
	fragment("(?->>'type' = 'Like' or ?->>'type' = 'Announce')", a.data, a.data) and
	fragment("?->>'type'", mutated_activity.data) == "Create",
	as: :mutated_activity
	)
	\|> where(
	[n, a, mutated_activity: mutated_activity],
	not fragment(
	"""
	CASE WHEN (?->>'type') = 'Like' or (?->>'type') = 'Announce'
	THEN (activity_visibility(?, ?, ?) = ANY (?))
	ELSE (activity_visibility(?, ?, ?) = ANY (?)) END
	""",
	a.data,
	a.data,
	mutated_activity.actor,
	mutated_activity.recipients,
	mutated_activity.data,
	^visibility,
	a.actor,
	a.recipients,
	a.data,
	^visibility
	)
	)
	else
	Logger.error("Could not exclude visibility to #{visibility}")
	query
	end
	end

	defp exclude_visibility(query, %{exclude_visibilities: visibility})
	when visibility in @valid_visibilities do
	exclude_visibility(query, [visibility])
	end

	defp exclude_visibility(query, %{exclude_visibilities: visibility})
	when visibility not in @valid_visibilities do
	Logger.error("Could not exclude visibility to #{visibility}")
	query
	end

	defp exclude_visibility(query, _visibility), do: query

	def for_user(user, opts \\ %{}) do
	user
	\|> for_user_query(opts)
	\|> Pagination.fetch_paginated(opts)
	end

	@doc """
	Returns notifications for user received since given date.

	## Examples

	iex> Pleroma.Notification.for_user_since(%Pleroma.User{}, ~N[2019-04-13 11:22:33])
	[%Pleroma.Notification{}, %Pleroma.Notification{}]

	iex> Pleroma.Notification.for_user_since(%Pleroma.User{}, ~N[2019-04-15 11:22:33])
	[]
	"""
	@spec for_user_since(Pleroma.User.t(), NaiveDateTime.t()) :: [t()]
	def for_user_since(user, date) do
	from(n in for_user_query(user),
	where: n.updated_at > ^date
	)
	\|> Repo.all()
	end

	def set_read_up_to(%{id: user_id} = user, id) do
	query =
	from(
	n in Notification,
	where: n.user_id == ^user_id,
	where: n.id <= ^id,
	where: n.seen == false,
	# Ideally we would preload object and activities here
	# but Ecto does not support preloads in update_all
	select: n.id
	)

	{:ok, %{ids: {_, notification_ids}}} =
	Multi.new()
	\|> Multi.update_all(:ids, query, set: [seen: true, updated_at: NaiveDateTime.utc_now()])
	\|> Marker.multi_set_last_read_id(user, "notifications")
	\|> Repo.transaction()

	for_user_query(user)
	\|> where([n], n.id in ^notification_ids)
	\|> Repo.all()
	end

	@spec read_one(User.t(), String.t()) ::
	{:ok, Notification.t()} \| {:error, Ecto.Changeset.t()} \| nil
	def read_one(%User{} = user, notification_id) do
	with {:ok, %Notification{} = notification} <- get(user, notification_id) do
	Multi.new()
	\|> Multi.update(:update, changeset(notification, %{seen: true}))
	\|> Marker.multi_set_last_read_id(user, "notifications")
	\|> Repo.transaction()
	\|> case do
	{:ok, %{update: notification}} -> {:ok, notification}
	{:error, :update, changeset, _} -> {:error, changeset}
	end
	end
	end

	def get(%{id: user_id} = _user, id) do
	query =
	from(
	n in Notification,
	where: n.id == ^id,
	join: activity in assoc(n, :activity),
	preload: [activity: activity]
	)

	notification = Repo.one(query)

	case notification do
	%{user_id: ^user_id} ->
	{:ok, notification}

	_ ->
	{:error, "Cannot get notification"}
	end
	end

	def clear(user) do
	from(n in Notification, where: n.user_id == ^user.id)
	\|> Repo.delete_all()
	end

	def destroy_multiple(%{id: user_id} = _user, ids) do
	from(n in Notification,
	where: n.id in ^ids,
	where: n.user_id == ^user_id
	)
	\|> Repo.delete_all()
	end

	def dismiss(%Pleroma.Activity{} = activity) do
	Notification
	\|> where([n], n.activity_id == ^activity.id)
	\|> Repo.delete_all()
	\|> case do
	{_, notifications} -> {:ok, notifications}
	_ -> {:error, "Cannot dismiss notification"}
	end
	end

	def dismiss(%{id: user_id} = _user, id) do
	notification = Repo.get(Notification, id)

	case notification do
	%{user_id: ^user_id} ->
	Repo.delete(notification)

	_ ->
	{:error, "Cannot dismiss notification"}
	end
	end

	@spec create_notifications(Activity.t()) :: {:ok, [Notification.t()] \| []}
	def create_notifications(activity)

	def create_notifications(%Activity{data: %{"to" => _, "type" => "Create"}} = activity) do
	object = Object.normalize(activity, fetch: false)

	if object && object.data["type"] == "Answer" do
	{:ok, []}
	else
	do_create_notifications(activity)
	end
	end

	def create_notifications(%Activity{data: %{"type" => type}} = activity)
	when type in ["Follow", "Like", "Announce", "Move", "EmojiReact", "Flag", "Update"] do
	do_create_notifications(activity)
	end

	def create_notifications(_), do: {:ok, []}

	defp do_create_notifications(%Activity{} = activity) do
	enabled_receivers = get_notified_from_activity(activity)

	notifications =
	Enum.map(enabled_receivers, fn user ->
	create_notification(activity, user)
	end)
	\|> Enum.reject(&is_nil/1)

	{:ok, notifications}
	end

	defp type_from_activity(%{data: %{"type" => type}} = activity) do
	case type do
	"Follow" ->
	if Activity.follow_accepted?(activity) do
	"follow"
	else
	"follow_request"
	end

	"Announce" ->
	"reblog"

	"Like" ->
	"favourite"

	"Move" ->
	"move"

	"EmojiReact" ->
	"pleroma:emoji_reaction"

	"Flag" ->
	"pleroma:report"

	# Compatibility with old reactions
	"EmojiReaction" ->
	"pleroma:emoji_reaction"

	"Create" ->
	activity
	\|> type_from_activity_object()

	"Update" ->
	"update"

	t ->
	raise "No notification type for activity type #{t}"
	end
	end

	defp type_from_activity_object(%{data: %{"type" => "Create", "object" => %{}}}), do: "mention"

	defp type_from_activity_object(%{data: %{"type" => "Create"}} = activity) do
	object = Object.get_by_ap_id(activity.data["object"])

	case object && object.data["type"] do
	"ChatMessage" -> "pleroma:chat_mention"
	_ -> "mention"
	end
	end

	# TODO move to sql, too.
	def create_notification(%Activity{} = activity, %User{} = user, opts \\ []) do
	type = Keyword.get(opts, :type, type_from_activity(activity))

	unless skip?(activity, user, opts) do
	{:ok, %{notification: notification}} =
	Multi.new()
	\|> Multi.insert(:notification, %Notification{
	user_id: user.id,
	activity: activity,
	seen: mark_as_read?(activity, user),
	type: type
	})
	\|> Marker.multi_set_last_read_id(user, "notifications")
	\|> Repo.transaction()

	notification
	end
	end

	def create_poll_notifications(%Activity{} = activity) do
	with %Object{data: %{"type" => "Question", "actor" => actor} = data} <-
	Object.normalize(activity) do
	voters =
	case data do
	%{"voters" => voters} when is_list(voters) -> voters
	_ -> []
	end

	notifications =
	Enum.reduce([actor \| voters], [], fn ap_id, acc ->
	with %User{local: true} = user <- User.get_by_ap_id(ap_id) do
	[create_notification(activity, user, type: "poll") \| acc]
	else
	_ -> acc
	end
	end)

	{:ok, notifications}
	end
	end

	@doc """
	Returns a tuple with 2 elements:
	{notification-enabled receivers, currently disabled receivers (blocking / [thread] muting)}

	NOTE: might be called for FAKE Activities, see ActivityPub.Utils.get_notified_from_object/1
	"""
	@spec get_notified_from_activity(Activity.t(), boolean()) :: {list(User.t()), list(User.t())}
	def get_notified_from_activity(activity, local_only \\ true)

	def get_notified_from_activity(%Activity{data: %{"type" => type}} = activity, local_only)
	when type in [
	"Create",
	"Like",
	"Announce",
	"Follow",
	"Move",
	"EmojiReact",
	"Flag",
	"Update"
	] do
	potential_receiver_ap_ids = get_potential_receiver_ap_ids(activity)

	potential_receivers =
	User.get_users_from_set(potential_receiver_ap_ids, local_only: local_only)

	notification_enabled_ap_ids =
	potential_receiver_ap_ids
	\|> exclude_domain_blocker_ap_ids(activity, potential_receivers)
	\|> exclude_relationship_restricted_ap_ids(activity)
	\|> exclude_thread_muter_ap_ids(activity)

	Enum.filter(potential_receivers, fn u -> u.ap_id in notification_enabled_ap_ids end)
	end

	def get_notified_from_activity(_, _local_only), do: {[], []}

	# For some activities, only notify the author of the object
	def get_potential_receiver_ap_ids(%{data: %{"type" => type, "object" => object_id}})
	when type in ~w{Like Announce EmojiReact} do
	case Object.get_cached_by_ap_id(object_id) do
	%Object{data: %{"actor" => actor}} ->
	[actor]

	_ ->
	[]
	end
	end

	def get_potential_receiver_ap_ids(%{data: %{"type" => "Follow", "object" => object_id}}) do
	[object_id]
	end

	def get_potential_receiver_ap_ids(%{data: %{"type" => "Flag", "actor" => actor}}) do
	(User.all_users_with_privilege(:reports_manage_reports)
	\|> Enum.map(fn user -> user.ap_id end)) --
	[actor]
	end

	# Update activity: notify all who repeated this
	def get_potential_receiver_ap_ids(%{data: %{"type" => "Update", "actor" => actor}} = activity) do
	with %Object{data: %{"id" => object_id}} <- Object.normalize(activity, fetch: false) do
	repeaters =
	Activity.Queries.by_type("Announce")
	\|> Activity.Queries.by_object_id(object_id)
	\|> Activity.with_joined_user_actor()
	\|> where([a, u], u.local)
	\|> select([a, u], u.ap_id)
	\|> Repo.all()

	repeaters -- [actor]
	end
	end

	def get_potential_receiver_ap_ids(activity) do
	[]
	\|> Utils.maybe_notify_to_recipients(activity)
	\|> Utils.maybe_notify_mentioned_recipients(activity)
	\|> Utils.maybe_notify_subscribers(activity)
	\|> Utils.maybe_notify_followers(activity)
	\|> Enum.uniq()
	end

	@doc "Filters out AP IDs domain-blocking and not following the activity's actor"
	def exclude_domain_blocker_ap_ids(ap_ids, activity, preloaded_users \\ [])

	def exclude_domain_blocker_ap_ids([], _activity, _preloaded_users), do: []

	def exclude_domain_blocker_ap_ids(ap_ids, %Activity{} = activity, preloaded_users) do
	activity_actor_domain = activity.actor && URI.parse(activity.actor).host

	users =
	ap_ids
	\|> Enum.map(fn ap_id ->
	Enum.find(preloaded_users, &(&1.ap_id == ap_id)) \|\|
	User.get_cached_by_ap_id(ap_id)
	end)
	\|> Enum.filter(& &1)

	domain_blocker_ap_ids = for u <- users, activity_actor_domain in u.domain_blocks, do: u.ap_id

	domain_blocker_follower_ap_ids =
	if Enum.any?(domain_blocker_ap_ids) do
	activity
	\|> Activity.user_actor()
	\|> FollowingRelationship.followers_ap_ids(domain_blocker_ap_ids)
	else
	[]
	end

	ap_ids
	\|> Kernel.--(domain_blocker_ap_ids)
	\|> Kernel.++(domain_blocker_follower_ap_ids)
	end

	@doc "Filters out AP IDs of users basing on their relationships with activity actor user"
	def exclude_relationship_restricted_ap_ids([], _activity), do: []

	def exclude_relationship_restricted_ap_ids(ap_ids, %Activity{} = activity) do
	relationship_restricted_ap_ids =
	activity
	\|> Activity.user_actor()
	\|> User.incoming_relationships_ungrouped_ap_ids([
	:block,
	:notification_mute
	])

	Enum.uniq(ap_ids) -- relationship_restricted_ap_ids
	end

	@doc "Filters out AP IDs of users who mute activity thread"
	def exclude_thread_muter_ap_ids([], _activity), do: []

	def exclude_thread_muter_ap_ids(ap_ids, %Activity{} = activity) do
	thread_muter_ap_ids = ThreadMute.muter_ap_ids(activity.data["context"])

	Enum.uniq(ap_ids) -- thread_muter_ap_ids
	end

	def skip?(activity, user, opts \\ [])

	@spec skip?(Activity.t(), User.t(), Keyword.t()) :: boolean()
	def skip?(%Activity{} = activity, %User{} = user, opts) do
	[
	:self,
	+ :internal,
	:invisible,
	:block_from_strangers,
	:recently_followed,
	:filtered
	]
	\|> Enum.find(&skip?(&1, activity, user, opts))
	end

	def skip?(_activity, _user, _opts), do: false

	@spec skip?(atom(), Activity.t(), User.t(), Keyword.t()) :: boolean()
	def skip?(:self, %Activity{} = activity, %User{} = user, opts) do
	cond do
	opts[:type] == "poll" -> false
	activity.data["actor"] == user.ap_id -> true
	true -> false
	end
	end

	+ def skip?(:internal, %Activity{} = activity, _user, _opts) do
	+ actor = activity.data["actor"]
	+ user = User.get_cached_by_ap_id(actor)
	+ User.internal?(user)
	+ end
	+
	def skip?(:invisible, %Activity{} = activity, _user, _opts) do
	actor = activity.data["actor"]
	user = User.get_cached_by_ap_id(actor)
	User.invisible?(user)
	end

	def skip?(
	:block_from_strangers,
	%Activity{} = activity,
	%User{notification_settings: %{block_from_strangers: true}} = user,
	opts
	) do
	actor = activity.data["actor"]
	follower = User.get_cached_by_ap_id(actor)

	cond do
	opts[:type] == "poll" -> false
	user.ap_id == actor -> false
	!User.following?(user, follower) -> true
	true -> false
	end
	end

	# To do: consider defining recency in hours and checking FollowingRelationship with a single SQL
	def skip?(
	:recently_followed,
	%Activity{data: %{"type" => "Follow"}} = activity,
	%User{} = user,
	_opts
	) do
	actor = activity.data["actor"]

	Notification.for_user(user)
	\|> Enum.any?(fn
	%{activity: %{data: %{"type" => "Follow", "actor" => ^actor}}} -> true
	_ -> false
	end)
	end

	def skip?(:filtered, %{data: %{"type" => type}}, _user, _opts) when type in ["Follow", "Move"],
	do: false

	def skip?(:filtered, activity, user, _opts) do
	object = Object.normalize(activity, fetch: false)

	cond do
	is_nil(object) ->
	false

	object.data["actor"] == user.ap_id ->
	false

	not is_nil(regex = Pleroma.Filter.compose_regex(user, :re)) ->
	Regex.match?(regex, object.data["content"])

	true ->
	false
	end
	end

	def skip?(_type, _activity, _user, _opts), do: false

	def mark_as_read?(activity, target_user) do
	user = Activity.user_actor(activity)
	User.mutes_user?(target_user, user) \|\| CommonAPI.thread_muted?(target_user, activity)
	end

	def for_user_and_activity(user, activity) do
	from(n in __MODULE__,
	where: n.user_id == ^user.id,
	where: n.activity_id == ^activity.id
	)
	\|> Repo.one()
	end

	@spec mark_context_as_read(User.t(), String.t()) :: {integer(), nil \| [term()]}
	def mark_context_as_read(%User{id: id}, context) do
	from(
	n in Notification,
	join: a in assoc(n, :activity),
	where: n.user_id == ^id,
	where: n.seen == false,
	where: fragment("?->>'context'", a.data) == ^context
	)
	\|> Repo.update_all(set: [seen: true])
	end

	@spec send(list(Notification.t())) :: :ok
	def send(notifications) do
	Enum.each(notifications, fn notification ->
	Streamer.stream(["user", "user:notification"], notification)
	Push.send(notification)
	end)
	end
	end
	diff --git a/lib/pleroma/web/activity_pub/side_effects.ex b/lib/pleroma/web/activity_pub/side_effects.ex
	index 982927e16..7ae16eb57 100644
	--- a/lib/pleroma/web/activity_pub/side_effects.ex
	+++ b/lib/pleroma/web/activity_pub/side_effects.ex
	@@ -1,640 +1,633 @@
	# 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.SideEffects do
	@moduledoc """
	This module looks at an inserted object and executes the side effects that it
	implies. For example, a `Like` activity will increase the like count on the
	liked object, a `Follow` activity will add the user to the follower
	collection, and so on.
	"""
	alias Pleroma.Activity
	alias Pleroma.Chat
	alias Pleroma.Chat.MessageReference
	alias Pleroma.FollowingRelationship
	alias Pleroma.Notification
	alias Pleroma.Object
	alias Pleroma.Repo
	alias Pleroma.User
	alias Pleroma.Web.ActivityPub.ActivityPub
	alias Pleroma.Web.ActivityPub.Builder
	alias Pleroma.Web.ActivityPub.Pipeline
	alias Pleroma.Web.ActivityPub.Utils
	alias Pleroma.Web.Streamer
	alias Pleroma.Workers.PollWorker

	require Pleroma.Constants
	require Logger

	@cachex Pleroma.Config.get([:cachex, :provider], Cachex)
	@logger Pleroma.Config.get([:side_effects, :logger], Logger)

	@behaviour Pleroma.Web.ActivityPub.SideEffects.Handling

	defp ap_streamer, do: Pleroma.Config.get([:side_effects, :ap_streamer], ActivityPub)

	@impl true
	def handle(object, meta \\ [])

	# Task this handles
	# - Follows
	# - Sends a notification
	@impl true
	def handle(
	%{
	data: %{
	"actor" => actor,
	"type" => "Accept",
	"object" => follow_activity_id
	}
	} = object,
	meta
	) do
	with %Activity{actor: follower_id} = follow_activity <-
	Activity.get_by_ap_id(follow_activity_id),
	%User{} = followed <- User.get_cached_by_ap_id(actor),
	%User{} = follower <- User.get_cached_by_ap_id(follower_id),
	{:ok, follow_activity} <- Utils.update_follow_state_for_all(follow_activity, "accept"),
	{:ok, _follower, followed} <-
	FollowingRelationship.update(follower, followed, :follow_accept) do
	Notification.update_notification_type(followed, follow_activity)
	end

	{:ok, object, meta}
	end

	# Task this handles
	# - Rejects all existing follow activities for this person
	# - Updates the follow state
	# - Dismisses notification
	@impl true
	def handle(
	%{
	data: %{
	"actor" => actor,
	"type" => "Reject",
	"object" => follow_activity_id
	}
	} = object,
	meta
	) do
	with %Activity{actor: follower_id} = follow_activity <-
	Activity.get_by_ap_id(follow_activity_id),
	%User{} = followed <- User.get_cached_by_ap_id(actor),
	%User{} = follower <- User.get_cached_by_ap_id(follower_id),
	{:ok, _follow_activity} <- Utils.update_follow_state_for_all(follow_activity, "reject") do
	FollowingRelationship.update(follower, followed, :follow_reject)
	Notification.dismiss(follow_activity)
	end

	{:ok, object, meta}
	end

	# Tasks this handle
	# - Follows if possible
	# - Sends a notification
	# - Generates accept or reject if appropriate
	@impl true
	def handle(
	%{
	data: %{
	"id" => follow_id,
	"type" => "Follow",
	"object" => followed_user,
	"actor" => following_user
	}
	} = object,
	meta
	) do
	with %User{} = follower <- User.get_cached_by_ap_id(following_user),
	%User{} = followed <- User.get_cached_by_ap_id(followed_user),
	{_, {:ok, _, _}, _, _} <-
	{:following, User.follow(follower, followed, :follow_pending), follower, followed} do
	if followed.local && !followed.is_locked do
	{:ok, accept_data, _} = Builder.accept(followed, object)
	{:ok, _activity, _} = Pipeline.common_pipeline(accept_data, local: true)
	end
	else
	{:following, {:error, _}, _follower, followed} ->
	{:ok, reject_data, _} = Builder.reject(followed, object)
	{:ok, _activity, _} = Pipeline.common_pipeline(reject_data, local: true)

	_ ->
	nil
	end

	{:ok, notifications} = Notification.create_notifications(object)

	meta =
	meta
	\|> add_notifications(notifications)

	updated_object = Activity.get_by_ap_id(follow_id)

	{:ok, updated_object, meta}
	end

	# Tasks this handles:
	# - Unfollow and block
	@impl true
	def handle(
	%{data: %{"type" => "Block", "object" => blocked_user, "actor" => blocking_user}} =
	object,
	meta
	) do
	with %User{} = blocker <- User.get_cached_by_ap_id(blocking_user),
	%User{} = blocked <- User.get_cached_by_ap_id(blocked_user) do
	User.block(blocker, blocked)
	end

	{:ok, object, meta}
	end

	# Tasks this handles:
	# - Update the user
	# - Update a non-user object (Note, Question, etc.)
	#
	# For a local user, we also get a changeset with the full information, so we
	# can update non-federating, non-activitypub settings as well.
	@impl true
	def handle(%{data: %{"type" => "Update", "object" => updated_object}} = object, meta) do
	updated_object_id = updated_object["id"]

	with {_, true} <- {:has_id, is_binary(updated_object_id)},
	%{"type" => type} <- updated_object,
	{_, is_user} <- {:is_user, type in Pleroma.Constants.actor_types()} do
	if is_user do
	handle_update_user(object, meta)
	else
	handle_update_object(object, meta)
	end
	else
	_ ->
	{:ok, object, meta}
	end
	end

	# Tasks this handles:
	# - Add like to object
	# - Set up notification
	@impl true
	def handle(%{data: %{"type" => "Like"}} = object, meta) do
	liked_object = Object.get_by_ap_id(object.data["object"])
	Utils.add_like_to_object(object, liked_object)

	{:ok, notifications} = Notification.create_notifications(object)

	meta =
	meta
	\|> add_notifications(notifications)

	{:ok, object, meta}
	end

	# Tasks this handles
	# - Actually create object
	# - Rollback if we couldn't create it
	# - Increase the user note count
	# - Increase the reply count
	# - Increase replies count
	# - Set up ActivityExpiration
	# - Set up notifications
	# - Index incoming posts for search (if needed)
	@impl true
	def handle(%{data: %{"type" => "Create"}} = activity, meta) do
	with {:ok, object, meta} <- handle_object_creation(meta[:object_data], activity, meta),
	%User{} = user <- User.get_cached_by_ap_id(activity.data["actor"]) do
	{:ok, notifications} = Notification.create_notifications(activity)
	{:ok, _user} = ActivityPub.increase_note_count_if_public(user, object)
	{:ok, _user} = ActivityPub.update_last_status_at_if_public(user, object)

	if in_reply_to = object.data["type"] != "Answer" && object.data["inReplyTo"] do
	Object.increase_replies_count(in_reply_to)
	end

	if quote_url = object.data["quoteUrl"] do
	Object.increase_quotes_count(quote_url)
	end

	reply_depth = (meta[:depth] \|\| 0) + 1

	# FIXME: Force inReplyTo to replies
	if Pleroma.Web.Federator.allowed_thread_distance?(reply_depth) and
	object.data["replies"] != nil do
	for reply_id <- object.data["replies"] do
	Pleroma.Workers.RemoteFetcherWorker.enqueue("fetch_remote", %{
	"id" => reply_id,
	"depth" => reply_depth
	})
	end
	end

	ConcurrentLimiter.limit(Pleroma.Web.RichMedia.Helpers, fn ->
	Task.start(fn -> Pleroma.Web.RichMedia.Helpers.fetch_data_for_activity(activity) end)
	end)

	Pleroma.Search.add_to_index(Map.put(activity, :object, object))

	Utils.maybe_handle_group_posts(activity)

	meta =
	meta
	\|> add_notifications(notifications)

	ap_streamer().stream_out(activity)

	{:ok, activity, meta}
	else
	e -> Repo.rollback(e)
	end
	end

	# Tasks this handles:
	# - Add announce to object
	# - Set up notification
	# - Stream out the announce
	@impl true
	def handle(%{data: %{"type" => "Announce"}} = object, meta) do
	announced_object = Object.get_by_ap_id(object.data["object"])
	user = User.get_cached_by_ap_id(object.data["actor"])

	Utils.add_announce_to_object(object, announced_object)

	- notifications =
	- if !User.is_internal_user?(user) do
	- {:ok, notifications} = Notification.create_notifications(object)
	-
	- ap_streamer().stream_out(object)
	+ {:ok, notifications} = Notification.create_notifications(object)

	- notifications
	- else
	- []
	- end
	+ if !User.internal?(user), do: ap_streamer().stream_out(object)

	meta =
	meta
	\|> add_notifications(notifications)

	{:ok, object, meta}
	end

	@impl true
	def handle(%{data: %{"type" => "Undo", "object" => undone_object}} = object, meta) do
	with undone_object <- Activity.get_by_ap_id(undone_object),
	:ok <- handle_undoing(undone_object) do
	{:ok, object, meta}
	end
	end

	# Tasks this handles:
	# - Add reaction to object
	# - Set up notification
	@impl true
	def handle(%{data: %{"type" => "EmojiReact"}} = object, meta) do
	reacted_object = Object.get_by_ap_id(object.data["object"])
	Utils.add_emoji_reaction_to_object(object, reacted_object)

	{:ok, notifications} = Notification.create_notifications(object)

	meta =
	meta
	\|> add_notifications(notifications)

	{:ok, object, meta}
	end

	# Tasks this handles:
	# - Delete and unpins the create activity
	# - Replace object with Tombstone
	# - Reduce the user note count
	# - Reduce the reply count
	# - Stream out the activity
	# - Removes posts from search index (if needed)
	@impl true
	def handle(%{data: %{"type" => "Delete", "object" => deleted_object}} = object, meta) do
	deleted_object =
	Object.normalize(deleted_object, fetch: false) \|\|
	User.get_cached_by_ap_id(deleted_object)

	result =
	case deleted_object do
	%Object{} ->
	with {_, {:ok, deleted_object, _activity}} <- {:object, Object.delete(deleted_object)},
	{_, actor} when is_binary(actor) <- {:actor, deleted_object.data["actor"]},
	{_, %User{} = user} <- {:user, User.get_cached_by_ap_id(actor)} do
	User.remove_pinned_object_id(user, deleted_object.data["id"])

	{:ok, user} = ActivityPub.decrease_note_count_if_public(user, deleted_object)

	if in_reply_to = deleted_object.data["inReplyTo"] do
	Object.decrease_replies_count(in_reply_to)
	end

	if quote_url = deleted_object.data["quoteUrl"] do
	Object.decrease_quotes_count(quote_url)
	end

	MessageReference.delete_for_object(deleted_object)

	ap_streamer().stream_out(object)
	ap_streamer().stream_out_participations(deleted_object, user)
	:ok
	else
	{:actor, _} ->
	@logger.error("The object doesn't have an actor: #{inspect(deleted_object)}")
	:no_object_actor

	{:user, _} ->
	@logger.error(
	"The object's actor could not be resolved to a user: #{inspect(deleted_object)}"
	)

	:no_object_user

	{:object, _} ->
	@logger.error("The object could not be deleted: #{inspect(deleted_object)}")
	{:error, object}
	end

	%User{} ->
	with {:ok, _} <- User.delete(deleted_object) do
	:ok
	end
	end

	if result == :ok do
	# Only remove from index when deleting actual objects, not users or anything else
	with %Pleroma.Object{} <- deleted_object do
	Pleroma.Search.remove_from_index(deleted_object)
	end

	{:ok, object, meta}
	else
	{:error, result}
	end
	end

	# Tasks this handles:
	# - adds pin to user
	# - removes expiration job for pinned activity, if was set for expiration
	@impl true
	def handle(%{data: %{"type" => "Add"} = data} = object, meta) do
	with %User{} = user <- User.get_cached_by_ap_id(data["actor"]),
	{:ok, _user} <- User.add_pinned_object_id(user, data["object"]) do
	# if pinned activity was scheduled for deletion, we remove job
	if expiration = Pleroma.Workers.PurgeExpiredActivity.get_expiration(meta[:activity_id]) do
	Oban.cancel_job(expiration.id)
	end

	{:ok, object, meta}
	else
	nil ->
	{:error, :user_not_found}

	{:error, changeset} ->
	if changeset.errors[:pinned_objects] do
	{:error, :pinned_statuses_limit_reached}
	else
	changeset.errors
	end
	end
	end

	# Tasks this handles:
	# - removes pin from user
	# - removes corresponding Add activity
	# - if activity had expiration, recreates activity expiration job
	@impl true
	def handle(%{data: %{"type" => "Remove"} = data} = object, meta) do
	with %User{} = user <- User.get_cached_by_ap_id(data["actor"]),
	{:ok, _user} <- User.remove_pinned_object_id(user, data["object"]) do
	data["object"]
	\|> Activity.add_by_params_query(user.ap_id, user.featured_address)
	\|> Repo.delete_all()

	# if pinned activity was scheduled for deletion, we reschedule it for deletion
	if meta[:expires_at] do
	# MRF.ActivityExpirationPolicy used UTC timestamps for expires_at in original implementation
	{:ok, expires_at} =
	Pleroma.EctoType.ActivityPub.ObjectValidators.DateTime.cast(meta[:expires_at])

	Pleroma.Workers.PurgeExpiredActivity.enqueue(%{
	activity_id: meta[:activity_id],
	expires_at: expires_at
	})
	end

	{:ok, object, meta}
	else
	nil -> {:error, :user_not_found}
	error -> error
	end
	end

	# Nothing to do
	@impl true
	def handle(object, meta) do
	{:ok, object, meta}
	end

	defp handle_update_user(
	%{data: %{"type" => "Update", "object" => updated_object}} = object,
	meta
	) do
	if changeset = Keyword.get(meta, :user_update_changeset) do
	changeset
	\|> User.update_and_set_cache()
	else
	{:ok, new_user_data} = ActivityPub.user_data_from_user_object(updated_object)

	User.get_by_ap_id(updated_object["id"])
	\|> User.remote_user_changeset(new_user_data)
	\|> User.update_and_set_cache()
	end

	{:ok, object, meta}
	end

	defp handle_update_object(
	%{data: %{"type" => "Update", "object" => updated_object}} = object,
	meta
	) do
	orig_object_ap_id = updated_object["id"]
	orig_object = Object.get_by_ap_id(orig_object_ap_id)
	orig_object_data = orig_object.data

	updated_object =
	if meta[:local] do
	# If this is a local Update, we don't process it by transmogrifier,
	# so we use the embedded object as-is.
	updated_object
	else
	meta[:object_data]
	end

	if orig_object_data["type"] in Pleroma.Constants.updatable_object_types() do
	{:ok, _, updated} =
	Object.Updater.do_update_and_invalidate_cache(orig_object, updated_object)

	if updated do
	object
	\|> Activity.normalize()
	\|> ActivityPub.notify_and_stream()
	end
	end

	{:ok, object, meta}
	end

	def handle_object_creation(%{"type" => "ChatMessage"} = object, _activity, meta) do
	with {:ok, object, meta} <- Pipeline.common_pipeline(object, meta) do
	actor = User.get_cached_by_ap_id(object.data["actor"])
	recipient = User.get_cached_by_ap_id(hd(object.data["to"]))

	streamables =
	[[actor, recipient], [recipient, actor]]
	\|> Enum.uniq()
	\|> Enum.map(fn [user, other_user] ->
	if user.local do
	{:ok, chat} = Chat.bump_or_create(user.id, other_user.ap_id)
	{:ok, cm_ref} = MessageReference.create(chat, object, user.ap_id != actor.ap_id)

	@cachex.put(
	:chat_message_id_idempotency_key_cache,
	cm_ref.id,
	meta[:idempotency_key]
	)

	{
	["user", "user:pleroma_chat"],
	{user, %{cm_ref \| chat: chat, object: object}}
	}
	end
	end)
	\|> Enum.filter(& &1)

	meta =
	meta
	\|> add_streamables(streamables)

	{:ok, object, meta}
	end
	end

	def handle_object_creation(%{"type" => "Question"} = object, activity, meta) do
	with {:ok, object, meta} <- Pipeline.common_pipeline(object, meta) do
	PollWorker.schedule_poll_end(activity)
	{:ok, object, meta}
	end
	end

	def handle_object_creation(%{"type" => "Answer"} = object_map, _activity, meta) do
	with {:ok, object, meta} <- Pipeline.common_pipeline(object_map, meta) do
	Object.increase_vote_count(
	object.data["inReplyTo"],
	object.data["name"],
	object.data["actor"]
	)

	{:ok, object, meta}
	end
	end

	def handle_object_creation(%{"type" => objtype} = object, _activity, meta)
	when objtype in ~w[Audio Video Image Event Article Note Page] do
	with {:ok, object, meta} <- Pipeline.common_pipeline(object, meta) do
	{:ok, object, meta}
	end
	end

	# Nothing to do
	def handle_object_creation(object, _activity, meta) do
	{:ok, object, meta}
	end

	defp undo_like(nil, object), do: delete_object(object)

	defp undo_like(%Object{} = liked_object, object) do
	with {:ok, _} <- Utils.remove_like_from_object(object, liked_object) do
	delete_object(object)
	end
	end

	def handle_undoing(%{data: %{"type" => "Like"}} = object) do
	object.data["object"]
	\|> Object.get_by_ap_id()
	\|> undo_like(object)
	end

	def handle_undoing(%{data: %{"type" => "EmojiReact"}} = object) do
	with %Object{} = reacted_object <- Object.get_by_ap_id(object.data["object"]),
	{:ok, _} <- Utils.remove_emoji_reaction_from_object(object, reacted_object),
	{:ok, _} <- Repo.delete(object) do
	:ok
	end
	end

	def handle_undoing(%{data: %{"type" => "Announce"}} = object) do
	with %Object{} = liked_object <- Object.get_by_ap_id(object.data["object"]),
	{:ok, _} <- Utils.remove_announce_from_object(object, liked_object),
	{:ok, _} <- Repo.delete(object) do
	:ok
	end
	end

	def handle_undoing(
	%{data: %{"type" => "Block", "actor" => blocker, "object" => blocked}} = object
	) do
	with %User{} = blocker <- User.get_cached_by_ap_id(blocker),
	%User{} = blocked <- User.get_cached_by_ap_id(blocked),
	{:ok, _} <- User.unblock(blocker, blocked),
	{:ok, _} <- Repo.delete(object) do
	:ok
	end
	end

	def handle_undoing(object), do: {:error, ["don't know how to handle", object]}

	@spec delete_object(Activity.t()) :: :ok \| {:error, Ecto.Changeset.t()}
	defp delete_object(object) do
	with {:ok, _} <- Repo.delete(object), do: :ok
	end

	defp send_notifications(meta) do
	Keyword.get(meta, :notifications, [])
	\|> Notification.send()

	meta
	end

	defp send_streamables(meta) do
	Keyword.get(meta, :streamables, [])
	\|> Enum.each(fn {topics, items} ->
	Streamer.stream(topics, items)
	end)

	meta
	end

	defp add_streamables(meta, streamables) do
	existing = Keyword.get(meta, :streamables, [])

	meta
	\|> Keyword.put(:streamables, streamables ++ existing)
	end

	defp add_notifications(meta, notifications) do
	existing = Keyword.get(meta, :notifications, [])

	meta
	\|> Keyword.put(:notifications, notifications ++ existing)
	end

	@impl true
	def handle_after_transaction(meta) do
	meta
	\|> send_notifications()
	\|> send_streamables()
	end
	end

File Metadata

Mime Type: text/x-diff
Expires: Mon, Nov 25, 6:02 AM (1 d, 11 h)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 39655
Default Alt Text: (42 KB)

No OneTemporaryActions

View Options

File Metadata

Event Timeline

No OneTemporary
Actions