No OneTemporary
Actions

Size

190 KB

Referenced Files

None

Subscribers

None

View Options

	diff --git a/lib/pleroma/object/fetcher.ex b/lib/pleroma/object/fetcher.ex
	index e81938536..df49a959c 100644
	--- a/lib/pleroma/object/fetcher.ex
	+++ b/lib/pleroma/object/fetcher.ex
	@@ -1,226 +1,229 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Object.Fetcher do
	alias Pleroma.HTTP
	alias Pleroma.Object
	alias Pleroma.Object.Containment
	alias Pleroma.Repo
	alias Pleroma.Signature
	alias Pleroma.Web.ActivityPub.InternalFetchActor
	alias Pleroma.Web.ActivityPub.Transmogrifier
	alias Pleroma.Web.Federator
	alias Pleroma.Web.FedSockets

	require Logger
	require Pleroma.Constants

	defp touch_changeset(changeset) do
	updated_at =
	NaiveDateTime.utc_now()
	\|> NaiveDateTime.truncate(:second)

	Ecto.Changeset.put_change(changeset, :updated_at, updated_at)
	end

	defp maybe_reinject_internal_fields(data, %{data: %{} = old_data}) do
	internal_fields = Map.take(old_data, Pleroma.Constants.object_internal_fields())

	Map.merge(data, internal_fields)
	end

	defp maybe_reinject_internal_fields(data, _), do: data

	@spec reinject_object(struct(), map()) :: {:ok, Object.t()} \| {:error, any()}
	defp reinject_object(struct, data) do
	Logger.debug("Reinjecting object #{data["id"]}")

	with data <- Transmogrifier.fix_object(data),
	data <- maybe_reinject_internal_fields(data, struct),
	changeset <- Object.change(struct, %{data: data}),
	changeset <- touch_changeset(changeset),
	{:ok, object} <- Repo.insert_or_update(changeset),
	{:ok, object} <- Object.set_cache(object) do
	{:ok, object}
	else
	e ->
	Logger.error("Error while processing object: #{inspect(e)}")
	{:error, e}
	end
	end

	def refetch_object(%Object{data: %{"id" => id}} = object) do
	with {:local, false} <- {:local, Object.local?(object)},
	{:ok, data} <- fetch_and_contain_remote_object_from_id(id),
	{:ok, object} <- reinject_object(object, data) do
	{:ok, object}
	else
	{:local, true} -> {:ok, object}
	e -> {:error, e}
	end
	end

	# Note: will create a Create activity, which we need internally at the moment.
	def fetch_object_from_id(id, options \\ []) do
	with {_, nil} <- {:fetch_object, Object.get_cached_by_ap_id(id)},
	{_, true} <- {:allowed_depth, Federator.allowed_thread_distance?(options[:depth])},
	{_, {:ok, data}} <- {:fetch, fetch_and_contain_remote_object_from_id(id)},
	{_, nil} <- {:normalize, Object.normalize(data, false)},
	params <- prepare_activity_params(data),
	{_, :ok} <- {:containment, Containment.contain_origin(id, params)},
	{_, {:ok, activity}} <-
	{:transmogrifier, Transmogrifier.handle_incoming(params, options)},
	{_, _data, %Object{} = object} <-
	{:object, data, Object.normalize(activity, false)} do
	{:ok, object}
	else
	{:allowed_depth, false} ->
	{:error, "Max thread distance exceeded."}

	{:containment, _} ->
	{:error, "Object containment failed."}

	{:transmogrifier, {:error, {:reject, nil}}} ->
	{:reject, nil}

	{:transmogrifier, _} ->
	{:error, "Transmogrifier failure."}

	{:object, data, nil} ->
	reinject_object(%Object{}, data)

	{:normalize, object = %Object{}} ->
	{:ok, object}

	{:fetch_object, %Object{} = object} ->
	{:ok, object}

	{:fetch, {:error, error}} ->
	{:error, error}

	e ->
	e
	end
	end

	defp prepare_activity_params(data) do
	%{
	"type" => "Create",
	"to" => data["to"],
	"cc" => data["cc"],
	# Should we seriously keep this attributedTo thing?
	"actor" => data["actor"] \|\| data["attributedTo"],
	"object" => data
	}
	end

	def fetch_object_from_id!(id, options \\ []) do
	with {:ok, object} <- fetch_object_from_id(id, options) do
	object
	else
	{:error, %Tesla.Mock.Error{}} ->
	nil

	{:error, "Object has been deleted"} ->
	nil

	e ->
	Logger.error("Error while fetching #{id}: #{inspect(e)}")
	nil
	end
	end

	defp make_signature(id, date) do
	uri = URI.parse(id)

	signature =
	InternalFetchActor.get_actor()
	\|> Signature.sign(%{
	"(request-target)": "get #{uri.path}",
	host: uri.host,
	date: date
	})

	[{:Signature, signature}]
	end

	defp sign_fetch(headers, id, date) do
	if Pleroma.Config.get([:activitypub, :sign_object_fetches]) do
	headers ++ make_signature(id, date)
	else
	headers
	end
	end

	defp maybe_date_fetch(headers, date) do
	if Pleroma.Config.get([:activitypub, :sign_object_fetches]) do
	headers ++ [{:Date, date}]
	else
	headers
	end
	end

	- def fetch_and_contain_remote_object_from_id(id) when is_binary(id) do
	+ def fetch_and_contain_remote_object_from_id(prm, opts \\ [])
	+
	+ def fetch_and_contain_remote_object_from_id(%{"id" => id}, opts),
	+ do: fetch_and_contain_remote_object_from_id(id, opts)
	+
	+ def fetch_and_contain_remote_object_from_id(id, opts) when is_binary(id) do
	Logger.debug("Fetching object #{id} via AP")

	with {:scheme, true} <- {:scheme, String.starts_with?(id, "http")},
	- {:ok, body} <- get_object(id),
	+ {:ok, body} <- get_object(id, opts),
	{:ok, data} <- safe_json_decode(body),
	:ok <- Containment.contain_origin_from_id(id, data) do
	{:ok, data}
	else
	{:scheme, _} ->
	{:error, "Unsupported URI scheme"}

	{:error, e} ->
	{:error, e}

	e ->
	{:error, e}
	end
	end

	- def fetch_and_contain_remote_object_from_id(%{"id" => id}),
	- do: fetch_and_contain_remote_object_from_id(id)
	-
	- def fetch_and_contain_remote_object_from_id(_id), do: {:error, "id must be a string"}
	+ def fetch_and_contain_remote_object_from_id(_id, _opts),
	+ do: {:error, "id must be a string"}

	- defp get_object(id) do
	- case FedSockets.get_or_create_fed_socket(id) do
	- {:ok, fedsocket} ->
	- FedSockets.fetch(fedsocket, id)
	-
	- _ ->
	+ defp get_object(id, opts) do
	+ with false <- Keyword.get(opts, :force_http, false),
	+ {:ok, fedsocket} <- FedSockets.get_or_create_fed_socket(id) do
	+ IO.inspect(id, label: "#{inspect(self())} - fetching via fedsocket")
	+ FedSockets.fetch(fedsocket, id)
	+ else
	+ _other ->
	+ IO.inspect(id, label: "#{inspect(self())} - fetching via http")
	get_object_http(id)
	end
	end

	defp get_object_http(id) do
	date = Pleroma.Signature.signed_date()

	headers =
	[{:Accept, "application/activity+json"}]
	\|> maybe_date_fetch(date)
	\|> sign_fetch(id, date)

	- Logger.debug("Fetch headers: #{inspect(headers)}")
	-
	case HTTP.get(id, headers) do
	{:ok, %{body: body, status: code}} when code in 200..299 ->
	{:ok, body}

	{:ok, %{status: code}} when code in [404, 410] ->
	{:error, "Object has been deleted"}

	{:error, e} ->
	{:error, e}

	e ->
	{:error, e}
	end
	end

	defp safe_json_decode(nil), do: {:ok, nil}
	defp safe_json_decode(json), do: Jason.decode(json)
	end
	diff --git a/lib/pleroma/signature.ex b/lib/pleroma/signature.ex
	index 6b0b2c969..6b2fe53b7 100644
	--- a/lib/pleroma/signature.ex
	+++ b/lib/pleroma/signature.ex
	@@ -1,62 +1,62 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Signature do
	@behaviour HTTPSignatures.Adapter

	alias Pleroma.Keys
	alias Pleroma.User
	alias Pleroma.Web.ActivityPub.ActivityPub

	def key_id_to_actor_id(key_id) do
	uri =
	URI.parse(key_id)
	\|> Map.put(:fragment, nil)

	uri =
	if not is_nil(uri.path) and String.ends_with?(uri.path, "/publickey") do
	Map.put(uri, :path, String.replace(uri.path, "/publickey", ""))
	else
	uri
	end

	URI.to_string(uri)
	end

	def fetch_public_key(conn) do
	with %{"keyId" => kid} <- HTTPSignatures.signature_for_conn(conn),
	actor_id <- key_id_to_actor_id(kid),
	- {:ok, public_key} <- User.get_public_key_for_ap_id(actor_id) do
	+ {:ok, public_key} <- User.get_public_key_for_ap_id(actor_id, force_http: true) do
	{:ok, public_key}
	else
	e ->
	{:error, e}
	end
	end

	def refetch_public_key(conn) do
	with %{"keyId" => kid} <- HTTPSignatures.signature_for_conn(conn),
	actor_id <- key_id_to_actor_id(kid),
	{:ok, _user} <- ActivityPub.make_user_from_ap_id(actor_id),
	- {:ok, public_key} <- User.get_public_key_for_ap_id(actor_id) do
	+ {:ok, public_key} <- User.get_public_key_for_ap_id(actor_id, force_http: true) do
	{:ok, public_key}
	else
	e ->
	{:error, e}
	end
	end

	def sign(%User{} = user, headers) do
	with {:ok, %{keys: keys}} <- User.ensure_keys_present(user),
	{:ok, private_key, _} <- Keys.keys_from_pem(keys) do
	HTTPSignatures.sign(private_key, user.ap_id <> "#main-key", headers)
	end
	end

	def signed_date, do: signed_date(NaiveDateTime.utc_now())

	def signed_date(%NaiveDateTime{} = date) do
	Timex.format!(date, "{WDshort}, {0D} {Mshort} {YYYY} {h24}:{m}:{s} GMT")
	end
	end
	diff --git a/lib/pleroma/user.ex b/lib/pleroma/user.ex
	index 911dde6e2..9fce0efe9 100644
	--- a/lib/pleroma/user.ex
	+++ b/lib/pleroma/user.ex
	@@ -1,2059 +1,2070 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.User do
	use Ecto.Schema

	import Ecto.Changeset
	import Ecto.Query
	import Ecto, only: [assoc: 2]

	alias Comeonin.Pbkdf2
	alias Ecto.Multi
	alias Pleroma.Activity
	alias Pleroma.Config
	alias Pleroma.Conversation.Participation
	alias Pleroma.Delivery
	alias Pleroma.FollowingRelationship
	alias Pleroma.HTML
	alias Pleroma.Keys
	alias Pleroma.Notification
	alias Pleroma.Object
	alias Pleroma.Registration
	alias Pleroma.Repo
	alias Pleroma.RepoStreamer
	alias Pleroma.User
	alias Pleroma.UserRelationship
	alias Pleroma.Web
	alias Pleroma.Web.ActivityPub.ActivityPub
	alias Pleroma.Web.ActivityPub.Utils
	alias Pleroma.Web.CommonAPI
	alias Pleroma.Web.CommonAPI.Utils, as: CommonUtils
	alias Pleroma.Web.OAuth
	alias Pleroma.Web.RelMe
	alias Pleroma.Workers.BackgroundWorker

	require Logger

	@type t :: %__MODULE__{}
	@type account_status :: :active \| :deactivated \| :password_reset_pending \| :confirmation_pending
	@primary_key {:id, FlakeId.Ecto.CompatType, autogenerate: true}

	# credo:disable-for-next-line Credo.Check.Readability.MaxLineLength
	@email_regex ~r/^[a-zA-Z0-9.!#$%&'+\/=?^_`{\|}~-]+@[a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?(?:\.[a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?)$/

	@strict_local_nickname_regex ~r/^[a-zA-Z\d]+$/
	@extended_local_nickname_regex ~r/^[a-zA-Z\d_-]+$/

	# AP ID user relationships (blocks, mutes etc.)
	# Format: [rel_type: [outgoing_rel: :outgoing_rel_target, incoming_rel: :incoming_rel_source]]
	@user_relationships_config [
	block: [
	blocker_blocks: :blocked_users,
	blockee_blocks: :blocker_users
	],
	mute: [
	muter_mutes: :muted_users,
	mutee_mutes: :muter_users
	],
	reblog_mute: [
	reblog_muter_mutes: :reblog_muted_users,
	reblog_mutee_mutes: :reblog_muter_users
	],
	notification_mute: [
	notification_muter_mutes: :notification_muted_users,
	notification_mutee_mutes: :notification_muter_users
	],
	# Note: `inverse_subscription` relationship is inverse: subscriber acts as relationship target
	inverse_subscription: [
	subscribee_subscriptions: :subscriber_users,
	subscriber_subscriptions: :subscribee_users
	]
	]

	schema "users" do
	field(:bio, :string)
	field(:email, :string)
	field(:name, :string)
	field(:nickname, :string)
	field(:password_hash, :string)
	field(:password, :string, virtual: true)
	field(:password_confirmation, :string, virtual: true)
	field(:keys, :string)
	field(:ap_id, :string)
	field(:avatar, :map)
	field(:local, :boolean, default: true)
	field(:follower_address, :string)
	field(:following_address, :string)
	field(:search_rank, :float, virtual: true)
	field(:search_type, :integer, virtual: true)
	field(:tags, {:array, :string}, default: [])
	field(:last_refreshed_at, :naive_datetime_usec)
	field(:last_digest_emailed_at, :naive_datetime)
	field(:banner, :map, default: %{})
	field(:background, :map, default: %{})
	field(:source_data, :map, default: %{})
	field(:note_count, :integer, default: 0)
	field(:follower_count, :integer, default: 0)
	field(:following_count, :integer, default: 0)
	field(:locked, :boolean, default: false)
	field(:confirmation_pending, :boolean, default: false)
	field(:password_reset_pending, :boolean, default: false)
	field(:confirmation_token, :string, default: nil)
	field(:default_scope, :string, default: "public")
	field(:domain_blocks, {:array, :string}, default: [])
	field(:deactivated, :boolean, default: false)
	field(:no_rich_text, :boolean, default: false)
	field(:ap_enabled, :boolean, default: false)
	field(:is_moderator, :boolean, default: false)
	field(:is_admin, :boolean, default: false)
	field(:show_role, :boolean, default: true)
	field(:settings, :map, default: nil)
	field(:magic_key, :string, default: nil)
	field(:uri, :string, default: nil)
	field(:hide_followers_count, :boolean, default: false)
	field(:hide_follows_count, :boolean, default: false)
	field(:hide_followers, :boolean, default: false)
	field(:hide_follows, :boolean, default: false)
	field(:hide_favorites, :boolean, default: true)
	field(:unread_conversation_count, :integer, default: 0)
	field(:pinned_activities, {:array, :string}, default: [])
	field(:email_notifications, :map, default: %{"digest" => false})
	field(:mascot, :map, default: nil)
	field(:emoji, {:array, :map}, default: [])
	field(:pleroma_settings_store, :map, default: %{})
	field(:fields, {:array, :map}, default: [])
	field(:raw_fields, {:array, :map}, default: [])
	field(:discoverable, :boolean, default: false)
	field(:invisible, :boolean, default: false)
	field(:allow_following_move, :boolean, default: true)
	field(:skip_thread_containment, :boolean, default: false)
	field(:actor_type, :string, default: "Person")
	field(:also_known_as, {:array, :string}, default: [])

	embeds_one(
	:notification_settings,
	Pleroma.User.NotificationSetting,
	on_replace: :update
	)

	has_many(:notifications, Notification)
	has_many(:registrations, Registration)
	has_many(:deliveries, Delivery)

	has_many(:outgoing_relationships, UserRelationship, foreign_key: :source_id)
	has_many(:incoming_relationships, UserRelationship, foreign_key: :target_id)

	for {relationship_type,
	[
	{outgoing_relation, outgoing_relation_target},
	{incoming_relation, incoming_relation_source}
	]} <- @user_relationships_config do
	# Definitions of `has_many :blocker_blocks`, `has_many :muter_mutes` etc.
	has_many(outgoing_relation, UserRelationship,
	foreign_key: :source_id,
	where: [relationship_type: relationship_type]
	)

	# Definitions of `has_many :blockee_blocks`, `has_many :mutee_mutes` etc.
	has_many(incoming_relation, UserRelationship,
	foreign_key: :target_id,
	where: [relationship_type: relationship_type]
	)

	# Definitions of `has_many :blocked_users`, `has_many :muted_users` etc.
	has_many(outgoing_relation_target, through: [outgoing_relation, :target])

	# Definitions of `has_many :blocker_users`, `has_many :muter_users` etc.
	has_many(incoming_relation_source, through: [incoming_relation, :source])
	end

	# `:blocks` is deprecated (replaced with `blocked_users` relation)
	field(:blocks, {:array, :string}, default: [])
	# `:mutes` is deprecated (replaced with `muted_users` relation)
	field(:mutes, {:array, :string}, default: [])
	# `:muted_reblogs` is deprecated (replaced with `reblog_muted_users` relation)
	field(:muted_reblogs, {:array, :string}, default: [])
	# `:muted_notifications` is deprecated (replaced with `notification_muted_users` relation)
	field(:muted_notifications, {:array, :string}, default: [])
	# `:subscribers` is deprecated (replaced with `subscriber_users` relation)
	field(:subscribers, {:array, :string}, default: [])

	timestamps()
	end

	for {_relationship_type, [{_outgoing_relation, outgoing_relation_target}, _]} <-
	@user_relationships_config do
	# Definitions of `blocked_users_relation/1`, `muted_users_relation/1`, etc.
	def unquote(:"#{outgoing_relation_target}_relation")(user, restrict_deactivated? \\ false) do
	target_users_query = assoc(user, unquote(outgoing_relation_target))

	if restrict_deactivated? do
	restrict_deactivated(target_users_query)
	else
	target_users_query
	end
	end

	# Definitions of `blocked_users/1`, `muted_users/1`, etc.
	def unquote(outgoing_relation_target)(user, restrict_deactivated? \\ false) do
	__MODULE__
	\|> apply(unquote(:"#{outgoing_relation_target}_relation"), [
	user,
	restrict_deactivated?
	])
	\|> Repo.all()
	end

	# Definitions of `blocked_users_ap_ids/1`, `muted_users_ap_ids/1`, etc.
	def unquote(:"#{outgoing_relation_target}_ap_ids")(user, restrict_deactivated? \\ false) do
	__MODULE__
	\|> apply(unquote(:"#{outgoing_relation_target}_relation"), [
	user,
	restrict_deactivated?
	])
	\|> select([u], u.ap_id)
	\|> Repo.all()
	end
	end

	@doc "Returns status account"
	@spec account_status(User.t()) :: account_status()
	def account_status(%User{deactivated: true}), do: :deactivated
	def account_status(%User{password_reset_pending: true}), do: :password_reset_pending

	def account_status(%User{confirmation_pending: true}) do
	case Config.get([:instance, :account_activation_required]) do
	true -> :confirmation_pending
	_ -> :active
	end
	end

	def account_status(%User{}), do: :active

	@spec visible_for?(User.t(), User.t() \| nil) :: boolean()
	def visible_for?(user, for_user \\ nil)

	def visible_for?(%User{invisible: true}, _), do: false

	def visible_for?(%User{id: user_id}, %User{id: for_id}) when user_id == for_id, do: true

	def visible_for?(%User{} = user, for_user) do
	account_status(user) == :active \|\| superuser?(for_user)
	end

	def visible_for?(_, _), do: false

	@spec superuser?(User.t()) :: boolean()
	def superuser?(%User{local: true, is_admin: true}), do: true
	def superuser?(%User{local: true, is_moderator: true}), do: true
	def superuser?(_), do: false

	@spec invisible?(User.t()) :: boolean()
	def invisible?(%User{invisible: true}), do: true
	def invisible?(_), do: false

	def avatar_url(user, options \\ []) do
	case user.avatar do
	%{"url" => [%{"href" => href} \| _]} -> href
	_ -> !options[:no_default] && "#{Web.base_url()}/images/avi.png"
	end
	end

	def banner_url(user, options \\ []) do
	case user.banner do
	%{"url" => [%{"href" => href} \| _]} -> href
	_ -> !options[:no_default] && "#{Web.base_url()}/images/banner.png"
	end
	end

	def profile_url(%User{source_data: %{"url" => url}}), do: url
	def profile_url(%User{ap_id: ap_id}), do: ap_id
	def profile_url(_), do: nil

	def ap_id(%User{nickname: nickname}), do: "#{Web.base_url()}/users/#{nickname}"

	def ap_followers(%User{follower_address: fa}) when is_binary(fa), do: fa
	def ap_followers(%User{} = user), do: "#{ap_id(user)}/followers"

	@spec ap_following(User.t()) :: Sring.t()
	def ap_following(%User{following_address: fa}) when is_binary(fa), do: fa
	def ap_following(%User{} = user), do: "#{ap_id(user)}/following"

	def follow_state(%User{} = user, %User{} = target) do
	case Utils.fetch_latest_follow(user, target) do
	%{data: %{"state" => state}} -> state
	# Ideally this would be nil, but then Cachex does not commit the value
	_ -> false
	end
	end

	def get_cached_follow_state(user, target) do
	key = "follow_state:#{user.ap_id}\|#{target.ap_id}"
	Cachex.fetch!(:user_cache, key, fn _ -> {:commit, follow_state(user, target)} end)
	end

	@spec set_follow_state_cache(String.t(), String.t(), String.t()) :: {:ok \| :error, boolean()}
	def set_follow_state_cache(user_ap_id, target_ap_id, state) do
	Cachex.put(:user_cache, "follow_state:#{user_ap_id}\|#{target_ap_id}", state)
	end

	@spec restrict_deactivated(Ecto.Query.t()) :: Ecto.Query.t()
	def restrict_deactivated(query) do
	from(u in query, where: u.deactivated != ^true)
	end

	defdelegate following_count(user), to: FollowingRelationship

	defp truncate_fields_param(params) do
	if Map.has_key?(params, :fields) do
	Map.put(params, :fields, Enum.map(params[:fields], &truncate_field/1))
	else
	params
	end
	end

	defp truncate_if_exists(params, key, max_length) do
	if Map.has_key?(params, key) and is_binary(params[key]) do
	{value, _chopped} = String.split_at(params[key], max_length)
	Map.put(params, key, value)
	else
	params
	end
	end

	def remote_user_creation(params) do
	bio_limit = Pleroma.Config.get([:instance, :user_bio_length], 5000)
	name_limit = Pleroma.Config.get([:instance, :user_name_length], 100)

	params =
	params
	\|> truncate_if_exists(:name, name_limit)
	\|> truncate_if_exists(:bio, bio_limit)
	\|> truncate_fields_param()

	changeset =
	%User{local: false}
	\|> cast(
	params,
	[
	:bio,
	:name,
	:ap_id,
	:nickname,
	:avatar,
	:ap_enabled,
	:source_data,
	:banner,
	:locked,
	:magic_key,
	:uri,
	:hide_followers,
	:hide_follows,
	:hide_followers_count,
	:hide_follows_count,
	:follower_count,
	:fields,
	:following_count,
	:discoverable,
	:invisible,
	:actor_type,
	:also_known_as
	]
	)
	\|> validate_required([:name, :ap_id])
	\|> unique_constraint(:nickname)
	\|> validate_format(:nickname, @email_regex)
	\|> validate_length(:bio, max: bio_limit)
	\|> validate_length(:name, max: name_limit)
	\|> validate_fields(true)

	case params[:source_data] do
	%{"followers" => followers, "following" => following} ->
	changeset
	\|> put_change(:follower_address, followers)
	\|> put_change(:following_address, following)

	_ ->
	followers = ap_followers(%User{nickname: get_field(changeset, :nickname)})
	put_change(changeset, :follower_address, followers)
	end
	end

	def update_changeset(struct, params \\ %{}) do
	bio_limit = Pleroma.Config.get([:instance, :user_bio_length], 5000)
	name_limit = Pleroma.Config.get([:instance, :user_name_length], 100)

	struct
	\|> cast(
	params,
	[
	:bio,
	:name,
	:avatar,
	:locked,
	:no_rich_text,
	:default_scope,
	:banner,
	:hide_follows,
	:hide_followers,
	:hide_followers_count,
	:hide_follows_count,
	:hide_favorites,
	:allow_following_move,
	:background,
	:show_role,
	:skip_thread_containment,
	:fields,
	:raw_fields,
	:pleroma_settings_store,
	:discoverable,
	:actor_type,
	:also_known_as
	]
	)
	\|> unique_constraint(:nickname)
	\|> validate_format(:nickname, local_nickname_regex())
	\|> validate_length(:bio, max: bio_limit)
	\|> validate_length(:name, min: 1, max: name_limit)
	\|> validate_fields(false)
	end

	def upgrade_changeset(struct, params \\ %{}, remote? \\ false) do
	bio_limit = Pleroma.Config.get([:instance, :user_bio_length], 5000)
	name_limit = Pleroma.Config.get([:instance, :user_name_length], 100)

	params = Map.put(params, :last_refreshed_at, NaiveDateTime.utc_now())

	params = if remote?, do: truncate_fields_param(params), else: params

	struct
	\|> cast(
	params,
	[
	:bio,
	:name,
	:follower_address,
	:following_address,
	:avatar,
	:last_refreshed_at,
	:ap_enabled,
	:source_data,
	:banner,
	:locked,
	:magic_key,
	:follower_count,
	:following_count,
	:hide_follows,
	:fields,
	:hide_followers,
	:allow_following_move,
	:discoverable,
	:hide_followers_count,
	:hide_follows_count,
	:actor_type,
	:also_known_as
	]
	)
	\|> unique_constraint(:nickname)
	\|> validate_format(:nickname, local_nickname_regex())
	\|> validate_length(:bio, max: bio_limit)
	\|> validate_length(:name, max: name_limit)
	\|> validate_fields(remote?)
	end

	def password_update_changeset(struct, params) do
	struct
	\|> cast(params, [:password, :password_confirmation])
	\|> validate_required([:password, :password_confirmation])
	\|> validate_confirmation(:password)
	\|> put_password_hash()
	\|> put_change(:password_reset_pending, false)
	end

	@spec reset_password(User.t(), map) :: {:ok, User.t()} \| {:error, Ecto.Changeset.t()}
	def reset_password(%User{id: user_id} = user, data) do
	multi =
	Multi.new()
	\|> Multi.update(:user, password_update_changeset(user, data))
	\|> Multi.delete_all(:tokens, OAuth.Token.Query.get_by_user(user_id))
	\|> Multi.delete_all(:auth, OAuth.Authorization.delete_by_user_query(user))

	case Repo.transaction(multi) do
	{:ok, %{user: user} = _} -> set_cache(user)
	{:error, _, changeset, _} -> {:error, changeset}
	end
	end

	def update_password_reset_pending(user, value) do
	user
	\|> change()
	\|> put_change(:password_reset_pending, value)
	\|> update_and_set_cache()
	end

	def force_password_reset_async(user) do
	BackgroundWorker.enqueue("force_password_reset", %{"user_id" => user.id})
	end

	@spec force_password_reset(User.t()) :: {:ok, User.t()} \| {:error, Ecto.Changeset.t()}
	def force_password_reset(user), do: update_password_reset_pending(user, true)

	def register_changeset(struct, params \\ %{}, opts \\ []) do
	bio_limit = Pleroma.Config.get([:instance, :user_bio_length], 5000)
	name_limit = Pleroma.Config.get([:instance, :user_name_length], 100)

	need_confirmation? =
	if is_nil(opts[:need_confirmation]) do
	Pleroma.Config.get([:instance, :account_activation_required])
	else
	opts[:need_confirmation]
	end

	struct
	\|> confirmation_changeset(need_confirmation: need_confirmation?)
	\|> cast(params, [:bio, :email, :name, :nickname, :password, :password_confirmation])
	\|> validate_required([:name, :nickname, :password, :password_confirmation])
	\|> validate_confirmation(:password)
	\|> unique_constraint(:email)
	\|> unique_constraint(:nickname)
	\|> validate_exclusion(:nickname, Pleroma.Config.get([User, :restricted_nicknames]))
	\|> validate_format(:nickname, local_nickname_regex())
	\|> validate_format(:email, @email_regex)
	\|> validate_length(:bio, max: bio_limit)
	\|> validate_length(:name, min: 1, max: name_limit)
	\|> maybe_validate_required_email(opts[:external])
	\|> put_password_hash
	\|> put_ap_id()
	\|> unique_constraint(:ap_id)
	\|> put_following_and_follower_address()
	end

	def maybe_validate_required_email(changeset, true), do: changeset

	def maybe_validate_required_email(changeset, _) do
	if Pleroma.Config.get([:instance, :account_activation_required]) do
	validate_required(changeset, [:email])
	else
	changeset
	end
	end

	defp put_ap_id(changeset) do
	ap_id = ap_id(%User{nickname: get_field(changeset, :nickname)})
	put_change(changeset, :ap_id, ap_id)
	end

	defp put_following_and_follower_address(changeset) do
	followers = ap_followers(%User{nickname: get_field(changeset, :nickname)})

	changeset
	\|> put_change(:follower_address, followers)
	end

	defp autofollow_users(user) do
	candidates = Pleroma.Config.get([:instance, :autofollowed_nicknames])

	autofollowed_users =
	User.Query.build(%{nickname: candidates, local: true, deactivated: false})
	\|> Repo.all()

	follow_all(user, autofollowed_users)
	end

	@doc "Inserts provided changeset, performs post-registration actions (confirmation email sending etc.)"
	def register(%Ecto.Changeset{} = changeset) do
	with {:ok, user} <- Repo.insert(changeset) do
	post_register_action(user)
	end
	end

	def post_register_action(%User{} = user) do
	with {:ok, user} <- autofollow_users(user),
	{:ok, user} <- set_cache(user),
	{:ok, _} <- User.WelcomeMessage.post_welcome_message_to_user(user),
	{:ok, _} <- try_send_confirmation_email(user) do
	{:ok, user}
	end
	end

	def try_send_confirmation_email(%User{} = user) do
	if user.confirmation_pending &&
	Pleroma.Config.get([:instance, :account_activation_required]) do
	user
	\|> Pleroma.Emails.UserEmail.account_confirmation_email()
	\|> Pleroma.Emails.Mailer.deliver_async()

	{:ok, :enqueued}
	else
	{:ok, :noop}
	end
	end

	def try_send_confirmation_email(users) do
	Enum.each(users, &try_send_confirmation_email/1)
	end

	def needs_update?(%User{local: true}), do: false

	def needs_update?(%User{local: false, last_refreshed_at: nil}), do: true

	def needs_update?(%User{local: false} = user) do
	NaiveDateTime.diff(NaiveDateTime.utc_now(), user.last_refreshed_at) >= 86_400
	end

	def needs_update?(_), do: true

	@spec maybe_direct_follow(User.t(), User.t()) :: {:ok, User.t()} \| {:error, String.t()}
	def maybe_direct_follow(%User{} = follower, %User{local: true, locked: true} = followed) do
	follow(follower, followed, "pending")
	end

	def maybe_direct_follow(%User{} = follower, %User{local: true} = followed) do
	follow(follower, followed)
	end

	def maybe_direct_follow(%User{} = follower, %User{} = followed) do
	if not ap_enabled?(followed) do
	follow(follower, followed)
	else
	{:ok, follower}
	end
	end

	@doc "A mass follow for local users. Respects blocks in both directions but does not create activities."
	@spec follow_all(User.t(), list(User.t())) :: {atom(), User.t()}
	def follow_all(follower, followeds) do
	followeds
	\|> Enum.reject(fn followed -> blocks?(follower, followed) \|\| blocks?(followed, follower) end)
	\|> Enum.each(&follow(follower, &1, "accept"))

	set_cache(follower)
	end

	defdelegate following(user), to: FollowingRelationship

	def follow(%User{} = follower, %User{} = followed, state \\ "accept") do
	deny_follow_blocked = Pleroma.Config.get([:user, :deny_follow_blocked])

	cond do
	followed.deactivated ->
	{:error, "Could not follow user: #{followed.nickname} is deactivated."}

	deny_follow_blocked and blocks?(followed, follower) ->
	{:error, "Could not follow user: #{followed.nickname} blocked you."}

	true ->
	FollowingRelationship.follow(follower, followed, state)

	{:ok, _} = update_follower_count(followed)

	follower
	\|> update_following_count()
	\|> set_cache()
	end
	end

	def unfollow(%User{ap_id: ap_id}, %User{ap_id: ap_id}) do
	{:error, "Not subscribed!"}
	end

	def unfollow(%User{} = follower, %User{} = followed) do
	case get_follow_state(follower, followed) do
	state when state in ["accept", "pending"] ->
	FollowingRelationship.unfollow(follower, followed)
	{:ok, followed} = update_follower_count(followed)

	{:ok, follower} =
	follower
	\|> update_following_count()
	\|> set_cache()

	{:ok, follower, Utils.fetch_latest_follow(follower, followed)}

	nil ->
	{:error, "Not subscribed!"}
	end
	end

	defdelegate following?(follower, followed), to: FollowingRelationship

	def get_follow_state(%User{} = follower, %User{} = following) do
	following_relationship = FollowingRelationship.get(follower, following)

	case {following_relationship, following.local} do
	{nil, false} ->
	case Utils.fetch_latest_follow(follower, following) do
	%{data: %{"state" => state}} when state in ["pending", "accept"] -> state
	_ -> nil
	end

	{%{state: state}, _} ->
	state

	{nil, _} ->
	nil
	end
	end

	def locked?(%User{} = user) do
	user.locked \|\| false
	end

	def get_by_id(id) do
	Repo.get_by(User, id: id)
	end

	def get_by_ap_id(ap_id) do
	Repo.get_by(User, ap_id: ap_id)
	end

	def get_all_by_ap_id(ap_ids) do
	from(u in __MODULE__,
	where: u.ap_id in ^ap_ids
	)
	\|> Repo.all()
	end

	def get_all_by_ids(ids) do
	from(u in __MODULE__, where: u.id in ^ids)
	\|> Repo.all()
	end

	# This is mostly an SPC migration fix. This guesses the user nickname by taking the last part
	# of the ap_id and the domain and tries to get that user
	def get_by_guessed_nickname(ap_id) do
	domain = URI.parse(ap_id).host
	name = List.last(String.split(ap_id, "/"))
	nickname = "#{name}@#{domain}"

	get_cached_by_nickname(nickname)
	end

	def set_cache({:ok, user}), do: set_cache(user)
	def set_cache({:error, err}), do: {:error, err}

	def set_cache(%User{} = user) do
	Cachex.put(:user_cache, "ap_id:#{user.ap_id}", user)
	Cachex.put(:user_cache, "nickname:#{user.nickname}", user)
	{:ok, user}
	end

	def update_and_set_cache(struct, params) do
	struct
	\|> update_changeset(params)
	\|> update_and_set_cache()
	end

	def update_and_set_cache(changeset) do
	with {:ok, user} <- Repo.update(changeset, stale_error_field: :id) do
	set_cache(user)
	end
	end

	def invalidate_cache(user) do
	Cachex.del(:user_cache, "ap_id:#{user.ap_id}")
	Cachex.del(:user_cache, "nickname:#{user.nickname}")
	end

	@spec get_cached_by_ap_id(String.t()) :: User.t() \| nil
	def get_cached_by_ap_id(ap_id) do
	key = "ap_id:#{ap_id}"

	with {:ok, nil} <- Cachex.get(:user_cache, key),
	user when not is_nil(user) <- get_by_ap_id(ap_id),
	{:ok, true} <- Cachex.put(:user_cache, key, user) do
	user
	else
	{:ok, user} -> user
	nil -> nil
	end
	end

	def get_cached_by_id(id) do
	key = "id:#{id}"

	ap_id =
	Cachex.fetch!(:user_cache, key, fn _ ->
	user = get_by_id(id)

	if user do
	Cachex.put(:user_cache, "ap_id:#{user.ap_id}", user)
	{:commit, user.ap_id}
	else
	{:ignore, ""}
	end
	end)

	get_cached_by_ap_id(ap_id)
	end

	def get_cached_by_nickname(nickname) do
	key = "nickname:#{nickname}"

	Cachex.fetch!(:user_cache, key, fn ->
	case get_or_fetch_by_nickname(nickname) do
	{:ok, user} -> {:commit, user}
	{:error, _error} -> {:ignore, nil}
	end
	end)
	end

	def get_cached_by_nickname_or_id(nickname_or_id, opts \\ []) do
	restrict_to_local = Pleroma.Config.get([:instance, :limit_to_local_content])

	cond do
	is_integer(nickname_or_id) or FlakeId.flake_id?(nickname_or_id) ->
	get_cached_by_id(nickname_or_id) \|\| get_cached_by_nickname(nickname_or_id)

	restrict_to_local == false or not String.contains?(nickname_or_id, "@") ->
	get_cached_by_nickname(nickname_or_id)

	restrict_to_local == :unauthenticated and match?(%User{}, opts[:for]) ->
	get_cached_by_nickname(nickname_or_id)

	true ->
	nil
	end
	end

	def get_by_nickname(nickname) do
	Repo.get_by(User, nickname: nickname) \|\|
	if Regex.match?(~r(@#{Pleroma.Web.Endpoint.host()})i, nickname) do
	Repo.get_by(User, nickname: local_nickname(nickname))
	end
	end

	def get_by_email(email), do: Repo.get_by(User, email: email)

	def get_by_nickname_or_email(nickname_or_email) do
	get_by_nickname(nickname_or_email) \|\| get_by_email(nickname_or_email)
	end

	def fetch_by_nickname(nickname), do: ActivityPub.make_user_from_nickname(nickname)

	def get_or_fetch_by_nickname(nickname) do
	with %User{} = user <- get_by_nickname(nickname) do
	{:ok, user}
	else
	_e ->
	with [_nick, _domain] <- String.split(nickname, "@"),
	{:ok, user} <- fetch_by_nickname(nickname) do
	{:ok, user}
	else
	_e -> {:error, "not found " <> nickname}
	end
	end
	end

	@spec get_followers_query(User.t(), pos_integer() \| nil) :: Ecto.Query.t()
	def get_followers_query(%User{} = user, nil) do
	User.Query.build(%{followers: user, deactivated: false})
	end

	def get_followers_query(user, page) do
	user
	\|> get_followers_query(nil)
	\|> User.Query.paginate(page, 20)
	end

	@spec get_followers_query(User.t()) :: Ecto.Query.t()
	def get_followers_query(user), do: get_followers_query(user, nil)

	@spec get_followers(User.t(), pos_integer() \| nil) :: {:ok, list(User.t())}
	def get_followers(user, page \\ nil) do
	user
	\|> get_followers_query(page)
	\|> Repo.all()
	end

	@spec get_external_followers(User.t(), pos_integer() \| nil) :: {:ok, list(User.t())}
	def get_external_followers(user, page \\ nil) do
	user
	\|> get_followers_query(page)
	\|> User.Query.build(%{external: true})
	\|> Repo.all()
	end

	def get_followers_ids(user, page \\ nil) do
	user
	\|> get_followers_query(page)
	\|> select([u], u.id)
	\|> Repo.all()
	end

	@spec get_friends_query(User.t(), pos_integer() \| nil) :: Ecto.Query.t()
	def get_friends_query(%User{} = user, nil) do
	User.Query.build(%{friends: user, deactivated: false})
	end

	def get_friends_query(user, page) do
	user
	\|> get_friends_query(nil)
	\|> User.Query.paginate(page, 20)
	end

	@spec get_friends_query(User.t()) :: Ecto.Query.t()
	def get_friends_query(user), do: get_friends_query(user, nil)

	def get_friends(user, page \\ nil) do
	user
	\|> get_friends_query(page)
	\|> Repo.all()
	end

	def get_friends_ap_ids(user) do
	user
	\|> get_friends_query(nil)
	\|> select([u], u.ap_id)
	\|> Repo.all()
	end

	def get_friends_ids(user, page \\ nil) do
	user
	\|> get_friends_query(page)
	\|> select([u], u.id)
	\|> Repo.all()
	end

	defdelegate get_follow_requests(user), to: FollowingRelationship

	def increase_note_count(%User{} = user) do
	User
	\|> where(id: ^user.id)
	\|> update([u], inc: [note_count: 1])
	\|> select([u], u)
	\|> Repo.update_all([])
	\|> case do
	{1, [user]} -> set_cache(user)
	_ -> {:error, user}
	end
	end

	def decrease_note_count(%User{} = user) do
	User
	\|> where(id: ^user.id)
	\|> update([u],
	set: [
	note_count: fragment("greatest(0, note_count - 1)")
	]
	)
	\|> select([u], u)
	\|> Repo.update_all([])
	\|> case do
	{1, [user]} -> set_cache(user)
	_ -> {:error, user}
	end
	end

	def update_note_count(%User{} = user, note_count \\ nil) do
	note_count =
	note_count \|\|
	from(
	a in Object,
	where: fragment("?->>'actor' = ? and ?->>'type' = 'Note'", a.data, ^user.ap_id, a.data),
	select: count(a.id)
	)
	\|> Repo.one()

	user
	\|> cast(%{note_count: note_count}, [:note_count])
	\|> update_and_set_cache()
	end

	@spec maybe_fetch_follow_information(User.t()) :: User.t()
	def maybe_fetch_follow_information(user) do
	with {:ok, user} <- fetch_follow_information(user) do
	user
	else
	e ->
	Logger.error("Follower/Following counter update for #{user.ap_id} failed.\n#{inspect(e)}")

	user
	end
	end

	def fetch_follow_information(user) do
	with {:ok, info} <- ActivityPub.fetch_follow_information_for_user(user) do
	user
	\|> follow_information_changeset(info)
	\|> update_and_set_cache()
	end
	end

	defp follow_information_changeset(user, params) do
	user
	\|> cast(params, [
	:hide_followers,
	:hide_follows,
	:follower_count,
	:following_count,
	:hide_followers_count,
	:hide_follows_count
	])
	end

	def update_follower_count(%User{} = user) do
	if user.local or !Pleroma.Config.get([:instance, :external_user_synchronization]) do
	follower_count_query =
	User.Query.build(%{followers: user, deactivated: false})
	\|> select([u], %{count: count(u.id)})

	User
	\|> where(id: ^user.id)
	\|> join(:inner, [u], s in subquery(follower_count_query))
	\|> update([u, s],
	set: [follower_count: s.count]
	)
	\|> select([u], u)
	\|> Repo.update_all([])
	\|> case do
	{1, [user]} -> set_cache(user)
	_ -> {:error, user}
	end
	else
	{:ok, maybe_fetch_follow_information(user)}
	end
	end

	@spec update_following_count(User.t()) :: User.t()
	def update_following_count(%User{local: false} = user) do
	if Pleroma.Config.get([:instance, :external_user_synchronization]) do
	maybe_fetch_follow_information(user)
	else
	user
	end
	end

	def update_following_count(%User{local: true} = user) do
	following_count = FollowingRelationship.following_count(user)

	user
	\|> follow_information_changeset(%{following_count: following_count})
	\|> Repo.update!()
	end

	def set_unread_conversation_count(%User{local: true} = user) do
	unread_query = Participation.unread_conversation_count_for_user(user)

	User
	\|> join(:inner, [u], p in subquery(unread_query))
	\|> update([u, p],
	set: [unread_conversation_count: p.count]
	)
	\|> where([u], u.id == ^user.id)
	\|> select([u], u)
	\|> Repo.update_all([])
	\|> case do
	{1, [user]} -> set_cache(user)
	_ -> {:error, user}
	end
	end

	def set_unread_conversation_count(user), do: {:ok, user}

	def increment_unread_conversation_count(conversation, %User{local: true} = user) do
	unread_query =
	Participation.unread_conversation_count_for_user(user)
	\|> where([p], p.conversation_id == ^conversation.id)

	User
	\|> join(:inner, [u], p in subquery(unread_query))
	\|> update([u, p],
	inc: [unread_conversation_count: 1]
	)
	\|> where([u], u.id == ^user.id)
	\|> where([u, p], p.count == 0)
	\|> select([u], u)
	\|> Repo.update_all([])
	\|> case do
	{1, [user]} -> set_cache(user)
	_ -> {:error, user}
	end
	end

	def increment_unread_conversation_count(_, user), do: {:ok, user}

	@spec get_users_from_set([String.t()], boolean()) :: [User.t()]
	def get_users_from_set(ap_ids, local_only \\ true) do
	criteria = %{ap_id: ap_ids, deactivated: false}
	criteria = if local_only, do: Map.put(criteria, :local, true), else: criteria

	User.Query.build(criteria)
	\|> Repo.all()
	end

	@spec get_recipients_from_activity(Activity.t()) :: [User.t()]
	def get_recipients_from_activity(%Activity{recipients: to}) do
	User.Query.build(%{recipients_from_activity: to, local: true, deactivated: false})
	\|> Repo.all()
	end

	@spec mute(User.t(), User.t(), boolean()) ::
	{:ok, list(UserRelationship.t())} \| {:error, String.t()}
	def mute(%User{} = muter, %User{} = mutee, notifications? \\ true) do
	add_to_mutes(muter, mutee, notifications?)
	end

	def unmute(%User{} = muter, %User{} = mutee) do
	remove_from_mutes(muter, mutee)
	end

	def subscribe(%User{} = subscriber, %User{} = target) do
	deny_follow_blocked = Pleroma.Config.get([:user, :deny_follow_blocked])

	if blocks?(target, subscriber) and deny_follow_blocked do
	{:error, "Could not subscribe: #{target.nickname} is blocking you"}
	else
	# Note: the relationship is inverse: subscriber acts as relationship target
	UserRelationship.create_inverse_subscription(target, subscriber)
	end
	end

	def subscribe(%User{} = subscriber, %{ap_id: ap_id}) do
	with %User{} = subscribee <- get_cached_by_ap_id(ap_id) do
	subscribe(subscriber, subscribee)
	end
	end

	def unsubscribe(%User{} = unsubscriber, %User{} = target) do
	# Note: the relationship is inverse: subscriber acts as relationship target
	UserRelationship.delete_inverse_subscription(target, unsubscriber)
	end

	def unsubscribe(%User{} = unsubscriber, %{ap_id: ap_id}) do
	with %User{} = user <- get_cached_by_ap_id(ap_id) do
	unsubscribe(unsubscriber, user)
	end
	end

	def block(%User{} = blocker, %User{} = blocked) do
	# sever any follow relationships to prevent leaks per activitypub (Pleroma issue #213)
	blocker =
	if following?(blocker, blocked) do
	{:ok, blocker, _} = unfollow(blocker, blocked)
	blocker
	else
	blocker
	end

	# clear any requested follows as well
	blocked =
	case CommonAPI.reject_follow_request(blocked, blocker) do
	{:ok, %User{} = updated_blocked} -> updated_blocked
	nil -> blocked
	end

	unsubscribe(blocked, blocker)

	if following?(blocked, blocker), do: unfollow(blocked, blocker)

	{:ok, blocker} = update_follower_count(blocker)
	{:ok, blocker, _} = Participation.mark_all_as_read(blocker, blocked)
	add_to_block(blocker, blocked)
	end

	# helper to handle the block given only an actor's AP id
	def block(%User{} = blocker, %{ap_id: ap_id}) do
	block(blocker, get_cached_by_ap_id(ap_id))
	end

	def unblock(%User{} = blocker, %User{} = blocked) do
	remove_from_block(blocker, blocked)
	end

	# helper to handle the block given only an actor's AP id
	def unblock(%User{} = blocker, %{ap_id: ap_id}) do
	unblock(blocker, get_cached_by_ap_id(ap_id))
	end

	def mutes?(nil, _), do: false
	def mutes?(%User{} = user, %User{} = target), do: mutes_user?(user, target)

	def mutes_user?(%User{} = user, %User{} = target) do
	UserRelationship.mute_exists?(user, target)
	end

	@spec muted_notifications?(User.t() \| nil, User.t() \| map()) :: boolean()
	def muted_notifications?(nil, _), do: false

	def muted_notifications?(%User{} = user, %User{} = target),
	do: UserRelationship.notification_mute_exists?(user, target)

	def blocks?(nil, _), do: false

	def blocks?(%User{} = user, %User{} = target) do
	blocks_user?(user, target) \|\|
	(!User.following?(user, target) && blocks_domain?(user, target))
	end

	def blocks_user?(%User{} = user, %User{} = target) do
	UserRelationship.block_exists?(user, target)
	end

	def blocks_user?(_, _), do: false

	def blocks_domain?(%User{} = user, %User{} = target) do
	domain_blocks = Pleroma.Web.ActivityPub.MRF.subdomains_regex(user.domain_blocks)
	%{host: host} = URI.parse(target.ap_id)
	Pleroma.Web.ActivityPub.MRF.subdomain_match?(domain_blocks, host)
	end

	def blocks_domain?(_, _), do: false

	def subscribed_to?(%User{} = user, %User{} = target) do
	# Note: the relationship is inverse: subscriber acts as relationship target
	UserRelationship.inverse_subscription_exists?(target, user)
	end

	def subscribed_to?(%User{} = user, %{ap_id: ap_id}) do
	with %User{} = target <- get_cached_by_ap_id(ap_id) do
	subscribed_to?(user, target)
	end
	end

	@doc """
	Returns map of outgoing (blocked, muted etc.) relations' user AP IDs by relation type.
	E.g. `outgoing_relations_ap_ids(user, [:block])` -> `%{block: ["https://some.site/users/userapid"]}`
	"""
	@spec outgoing_relations_ap_ids(User.t(), list(atom())) :: %{atom() => list(String.t())}
	def outgoing_relations_ap_ids(_, []), do: %{}

	def outgoing_relations_ap_ids(%User{} = user, relationship_types)
	when is_list(relationship_types) do
	db_result =
	user
	\|> assoc(:outgoing_relationships)
	\|> join(:inner, [user_rel], u in assoc(user_rel, :target))
	\|> where([user_rel, u], user_rel.relationship_type in ^relationship_types)
	\|> select([user_rel, u], [user_rel.relationship_type, fragment("array_agg(?)", u.ap_id)])
	\|> group_by([user_rel, u], user_rel.relationship_type)
	\|> Repo.all()
	\|> Enum.into(%{}, fn [k, v] -> {k, v} end)

	Enum.into(
	relationship_types,
	%{},
	fn rel_type -> {rel_type, db_result[rel_type] \|\| []} end
	)
	end

	def deactivate_async(user, status \\ true) do
	BackgroundWorker.enqueue("deactivate_user", %{"user_id" => user.id, "status" => status})
	end

	def deactivate(user, status \\ true)

	def deactivate(users, status) when is_list(users) do
	Repo.transaction(fn ->
	for user <- users, do: deactivate(user, status)
	end)
	end

	def deactivate(%User{} = user, status) do
	with {:ok, user} <- set_activation_status(user, status) do
	user
	\|> get_followers()
	\|> Enum.filter(& &1.local)
	\|> Enum.each(fn follower ->
	follower \|> update_following_count() \|> set_cache()
	end)

	# Only update local user counts, remote will be update during the next pull.
	user
	\|> get_friends()
	\|> Enum.filter(& &1.local)
	\|> Enum.each(&update_follower_count/1)

	{:ok, user}
	end
	end

	def update_notification_settings(%User{} = user, settings) do
	user
	\|> cast(%{notification_settings: settings}, [])
	\|> cast_embed(:notification_settings)
	\|> validate_required([:notification_settings])
	\|> update_and_set_cache()
	end

	def delete(users) when is_list(users) do
	for user <- users, do: delete(user)
	end

	def delete(%User{} = user) do
	BackgroundWorker.enqueue("delete_user", %{"user_id" => user.id})
	end

	def perform(:force_password_reset, user), do: force_password_reset(user)

	@spec perform(atom(), User.t()) :: {:ok, User.t()}
	def perform(:delete, %User{} = user) do
	{:ok, _user} = ActivityPub.delete(user)

	# Remove all relationships
	user
	\|> get_followers()
	\|> Enum.each(fn follower ->
	ActivityPub.unfollow(follower, user)
	unfollow(follower, user)
	end)

	user
	\|> get_friends()
	\|> Enum.each(fn followed ->
	ActivityPub.unfollow(user, followed)
	unfollow(user, followed)
	end)

	delete_user_activities(user)
	invalidate_cache(user)
	Repo.delete(user)
	end

	def perform(:deactivate_async, user, status), do: deactivate(user, status)

	@spec perform(atom(), User.t(), list()) :: list() \| {:error, any()}
	def perform(:blocks_import, %User{} = blocker, blocked_identifiers)
	when is_list(blocked_identifiers) do
	Enum.map(
	blocked_identifiers,
	fn blocked_identifier ->
	with {:ok, %User{} = blocked} <- get_or_fetch(blocked_identifier),
	{:ok, _user_block} <- block(blocker, blocked),
	{:ok, _} <- ActivityPub.block(blocker, blocked) do
	blocked
	else
	err ->
	Logger.debug("blocks_import failed for #{blocked_identifier} with: #{inspect(err)}")
	err
	end
	end
	)
	end

	def perform(:follow_import, %User{} = follower, followed_identifiers)
	when is_list(followed_identifiers) do
	Enum.map(
	followed_identifiers,
	fn followed_identifier ->
	with {:ok, %User{} = followed} <- get_or_fetch(followed_identifier),
	{:ok, follower} <- maybe_direct_follow(follower, followed),
	{:ok, _} <- ActivityPub.follow(follower, followed) do
	followed
	else
	err ->
	Logger.debug("follow_import failed for #{followed_identifier} with: #{inspect(err)}")
	err
	end
	end
	)
	end

	@spec external_users_query() :: Ecto.Query.t()
	def external_users_query do
	User.Query.build(%{
	external: true,
	active: true,
	order_by: :id
	})
	end

	@spec external_users(keyword()) :: [User.t()]
	def external_users(opts \\ []) do
	query =
	external_users_query()
	\|> select([u], struct(u, [:id, :ap_id]))

	query =
	if opts[:max_id],
	do: where(query, [u], u.id > ^opts[:max_id]),
	else: query

	query =
	if opts[:limit],
	do: limit(query, ^opts[:limit]),
	else: query

	Repo.all(query)
	end

	def blocks_import(%User{} = blocker, blocked_identifiers) when is_list(blocked_identifiers) do
	BackgroundWorker.enqueue("blocks_import", %{
	"blocker_id" => blocker.id,
	"blocked_identifiers" => blocked_identifiers
	})
	end

	def follow_import(%User{} = follower, followed_identifiers)
	when is_list(followed_identifiers) do
	BackgroundWorker.enqueue("follow_import", %{
	"follower_id" => follower.id,
	"followed_identifiers" => followed_identifiers
	})
	end

	def delete_user_activities(%User{ap_id: ap_id}) do
	ap_id
	\|> Activity.Queries.by_actor()
	\|> RepoStreamer.chunk_stream(50)
	\|> Stream.each(fn activities -> Enum.each(activities, &delete_activity/1) end)
	\|> Stream.run()
	end

	defp delete_activity(%{data: %{"type" => "Create"}} = activity) do
	activity
	\|> Object.normalize()
	\|> ActivityPub.delete()
	end

	defp delete_activity(%{data: %{"type" => "Like"}} = activity) do
	object = Object.normalize(activity)

	activity.actor
	\|> get_cached_by_ap_id()
	\|> ActivityPub.unlike(object)
	end

	defp delete_activity(%{data: %{"type" => "Announce"}} = activity) do
	object = Object.normalize(activity)

	activity.actor
	\|> get_cached_by_ap_id()
	\|> ActivityPub.unannounce(object)
	end

	defp delete_activity(_activity), do: "Doing nothing"

	def html_filter_policy(%User{no_rich_text: true}) do
	Pleroma.HTML.Scrubber.TwitterText
	end

	def html_filter_policy(_), do: Pleroma.Config.get([:markup, :scrub_policy])

	- def fetch_by_ap_id(ap_id), do: ActivityPub.make_user_from_ap_id(ap_id)
	+ def fetch_by_ap_id(ap_id, opts \\ []), do: ActivityPub.make_user_from_ap_id(ap_id, opts)

	- def get_or_fetch_by_ap_id(ap_id) do
	+ def get_or_fetch_by_ap_id(ap_id, opts \\ []) do
	user = get_cached_by_ap_id(ap_id)

	if !is_nil(user) and !needs_update?(user) do
	{:ok, user}
	else
	- fetch_by_ap_id(ap_id)
	+ # Whether to fetch initial posts for the user (if it's a new user & the fetching is enabled)
	+ should_fetch_initial = is_nil(user) and Pleroma.Config.get([:fetch_initial_posts, :enabled])
	+
	+ resp = fetch_by_ap_id(ap_id, opts)
	+
	+ if should_fetch_initial do
	+ with {:ok, %User{} = user} <- resp do
	+ fetch_initial_posts(user)
	+ end
	+ end
	+
	+ resp
	end
	end

	@doc """
	Creates an internal service actor by URI if missing.
	Optionally takes nickname for addressing.
	"""
	@spec get_or_create_service_actor_by_ap_id(String.t(), String.t()) :: User.t() \| nil
	def get_or_create_service_actor_by_ap_id(uri, nickname) do
	{_, user} =
	case get_cached_by_ap_id(uri) do
	nil ->
	with {:error, %{errors: errors}} <- create_service_actor(uri, nickname) do
	Logger.error("Cannot create service actor: #{uri}/.\n#{inspect(errors)}")
	{:error, nil}
	end

	%User{invisible: false} = user ->
	set_invisible(user)

	user ->
	{:ok, user}
	end

	user
	end

	@spec set_invisible(User.t()) :: {:ok, User.t()}
	defp set_invisible(user) do
	user
	\|> change(%{invisible: true})
	\|> update_and_set_cache()
	end

	@spec create_service_actor(String.t(), String.t()) ::
	{:ok, User.t()} \| {:error, Ecto.Changeset.t()}
	defp create_service_actor(uri, nickname) do
	%User{
	invisible: true,
	local: true,
	ap_id: uri,
	nickname: nickname,
	follower_address: uri <> "/followers"
	}
	\|> change
	\|> unique_constraint(:nickname)
	\|> Repo.insert()
	\|> set_cache()
	end

	# AP style
	def public_key(%{source_data: %{"publicKey" => %{"publicKeyPem" => public_key_pem}}}) do
	key =
	public_key_pem
	\|> :public_key.pem_decode()
	\|> hd()
	\|> :public_key.pem_entry_decode()

	{:ok, key}
	end

	def public_key(_), do: {:error, "not found key"}

	- def get_public_key_for_ap_id(ap_id) do
	- with {:ok, %User{} = user} <- get_or_fetch_by_ap_id(ap_id),
	+ def get_public_key_for_ap_id(ap_id, opts \\ []) do
	+ with {:ok, %User{} = user} <- get_or_fetch_by_ap_id(ap_id, opts),
	{:ok, public_key} <- public_key(user) do
	{:ok, public_key}
	else
	_ -> :error
	end
	end

	defp blank?(""), do: nil
	defp blank?(n), do: n

	def insert_or_update_user(data) do
	data
	\|> Map.put(:name, blank?(data[:name]) \|\| data[:nickname])
	\|> remote_user_creation()
	\|> Repo.insert(on_conflict: {:replace_all_except, [:id]}, conflict_target: :nickname)
	\|> set_cache()
	end

	def ap_enabled?(%User{local: true}), do: true
	def ap_enabled?(%User{ap_enabled: ap_enabled}), do: ap_enabled
	def ap_enabled?(_), do: false

	@doc "Gets or fetch a user by uri or nickname."
	@spec get_or_fetch(String.t()) :: {:ok, User.t()} \| {:error, String.t()}
	def get_or_fetch("http" <> _host = uri), do: get_or_fetch_by_ap_id(uri)
	def get_or_fetch(nickname), do: get_or_fetch_by_nickname(nickname)

	# wait a period of time and return newest version of the User structs
	# this is because we have synchronous follow APIs and need to simulate them
	# with an async handshake
	def wait_and_refresh(_, %User{local: true} = a, %User{local: true} = b) do
	with %User{} = a <- get_cached_by_id(a.id),
	%User{} = b <- get_cached_by_id(b.id) do
	{:ok, a, b}
	else
	nil -> :error
	end
	end

	def wait_and_refresh(timeout, %User{} = a, %User{} = b) do
	with :ok <- :timer.sleep(timeout),
	%User{} = a <- get_cached_by_id(a.id),
	%User{} = b <- get_cached_by_id(b.id) do
	{:ok, a, b}
	else
	nil -> :error
	end
	end

	def parse_bio(bio) when is_binary(bio) and bio != "" do
	bio
	\|> CommonUtils.format_input("text/plain", mentions_format: :full)
	\|> elem(0)
	end

	def parse_bio(_), do: ""

	def parse_bio(bio, user) when is_binary(bio) and bio != "" do
	# TODO: get profile URLs other than user.ap_id
	profile_urls = [user.ap_id]

	bio
	\|> CommonUtils.format_input("text/plain",
	mentions_format: :full,
	rel: &RelMe.maybe_put_rel_me(&1, profile_urls)
	)
	\|> elem(0)
	end

	def parse_bio(_, _), do: ""

	def tag(user_identifiers, tags) when is_list(user_identifiers) do
	Repo.transaction(fn ->
	for user_identifier <- user_identifiers, do: tag(user_identifier, tags)
	end)
	end

	def tag(nickname, tags) when is_binary(nickname),
	do: tag(get_by_nickname(nickname), tags)

	def tag(%User{} = user, tags),
	do: update_tags(user, Enum.uniq((user.tags \|\| []) ++ normalize_tags(tags)))

	def untag(user_identifiers, tags) when is_list(user_identifiers) do
	Repo.transaction(fn ->
	for user_identifier <- user_identifiers, do: untag(user_identifier, tags)
	end)
	end

	def untag(nickname, tags) when is_binary(nickname),
	do: untag(get_by_nickname(nickname), tags)

	def untag(%User{} = user, tags),
	do: update_tags(user, (user.tags \|\| []) -- normalize_tags(tags))

	defp update_tags(%User{} = user, new_tags) do
	{:ok, updated_user} =
	user
	\|> change(%{tags: new_tags})
	\|> update_and_set_cache()

	updated_user
	end

	defp normalize_tags(tags) do
	[tags]
	\|> List.flatten()
	\|> Enum.map(&String.downcase/1)
	end

	defp local_nickname_regex do
	if Pleroma.Config.get([:instance, :extended_nickname_format]) do
	@extended_local_nickname_regex
	else
	@strict_local_nickname_regex
	end
	end

	def local_nickname(nickname_or_mention) do
	nickname_or_mention
	\|> full_nickname()
	\|> String.split("@")
	\|> hd()
	end

	def full_nickname(nickname_or_mention),
	do: String.trim_leading(nickname_or_mention, "@")

	def error_user(ap_id) do
	%User{
	name: ap_id,
	ap_id: ap_id,
	nickname: "erroruser@example.com",
	inserted_at: NaiveDateTime.utc_now()
	}
	end

	@spec all_superusers() :: [User.t()]
	def all_superusers do
	User.Query.build(%{super_users: true, local: true, deactivated: false})
	\|> Repo.all()
	end

	def showing_reblogs?(%User{} = user, %User{} = target) do
	not UserRelationship.reblog_mute_exists?(user, target)
	end

	@doc """
	The function returns a query to get users with no activity for given interval of days.
	Inactive users are those who didn't read any notification, or had any activity where
	the user is the activity's actor, during `inactivity_threshold` days.
	Deactivated users will not appear in this list.

	## Examples

	iex> Pleroma.User.list_inactive_users()
	%Ecto.Query{}
	"""
	@spec list_inactive_users_query(integer()) :: Ecto.Query.t()
	def list_inactive_users_query(inactivity_threshold \\ 7) do
	negative_inactivity_threshold = -inactivity_threshold
	now = NaiveDateTime.truncate(NaiveDateTime.utc_now(), :second)
	# Subqueries are not supported in `where` clauses, join gets too complicated.
	has_read_notifications =
	from(n in Pleroma.Notification,
	where: n.seen == true,
	group_by: n.id,
	having: max(n.updated_at) > datetime_add(^now, ^negative_inactivity_threshold, "day"),
	select: n.user_id
	)
	\|> Pleroma.Repo.all()

	from(u in Pleroma.User,
	left_join: a in Pleroma.Activity,
	on: u.ap_id == a.actor,
	where: not is_nil(u.nickname),
	where: u.deactivated != ^true,
	where: u.id not in ^has_read_notifications,
	group_by: u.id,
	having:
	max(a.inserted_at) < datetime_add(^now, ^negative_inactivity_threshold, "day") or
	is_nil(max(a.inserted_at))
	)
	end

	@doc """
	Enable or disable email notifications for user

	## Examples

	iex> Pleroma.User.switch_email_notifications(Pleroma.User{email_notifications: %{"digest" => false}}, "digest", true)
	Pleroma.User{email_notifications: %{"digest" => true}}

	iex> Pleroma.User.switch_email_notifications(Pleroma.User{email_notifications: %{"digest" => true}}, "digest", false)
	Pleroma.User{email_notifications: %{"digest" => false}}
	"""
	@spec switch_email_notifications(t(), String.t(), boolean()) ::
	{:ok, t()} \| {:error, Ecto.Changeset.t()}
	def switch_email_notifications(user, type, status) do
	User.update_email_notifications(user, %{type => status})
	end

	@doc """
	Set `last_digest_emailed_at` value for the user to current time
	"""
	@spec touch_last_digest_emailed_at(t()) :: t()
	def touch_last_digest_emailed_at(user) do
	now = NaiveDateTime.truncate(NaiveDateTime.utc_now(), :second)

	{:ok, updated_user} =
	user
	\|> change(%{last_digest_emailed_at: now})
	\|> update_and_set_cache()

	updated_user
	end

	@spec toggle_confirmation(User.t()) :: {:ok, User.t()} \| {:error, Changeset.t()}
	def toggle_confirmation(%User{} = user) do
	user
	\|> confirmation_changeset(need_confirmation: !user.confirmation_pending)
	\|> update_and_set_cache()
	end

	@spec toggle_confirmation([User.t()]) :: [{:ok, User.t()} \| {:error, Changeset.t()}]
	def toggle_confirmation(users) do
	Enum.map(users, &toggle_confirmation/1)
	end

	def get_mascot(%{mascot: %{} = mascot}) when not is_nil(mascot) do
	mascot
	end

	def get_mascot(%{mascot: mascot}) when is_nil(mascot) do
	# use instance-default
	config = Pleroma.Config.get([:assets, :mascots])
	default_mascot = Pleroma.Config.get([:assets, :default_mascot])
	mascot = Keyword.get(config, default_mascot)

	%{
	"id" => "default-mascot",
	"url" => mascot[:url],
	"preview_url" => mascot[:url],
	"pleroma" => %{
	"mime_type" => mascot[:mime_type]
	}
	}
	end

	def ensure_keys_present(%{keys: keys} = user) when not is_nil(keys), do: {:ok, user}

	def ensure_keys_present(%User{} = user) do
	with {:ok, pem} <- Keys.generate_rsa_pem() do
	user
	\|> cast(%{keys: pem}, [:keys])
	\|> validate_required([:keys])
	\|> update_and_set_cache()
	end
	end

	def get_ap_ids_by_nicknames(nicknames) do
	from(u in User,
	where: u.nickname in ^nicknames,
	select: u.ap_id
	)
	\|> Repo.all()
	end

	defdelegate search(query, opts \\ []), to: User.Search

	defp put_password_hash(
	%Ecto.Changeset{valid?: true, changes: %{password: password}} = changeset
	) do
	change(changeset, password_hash: Pbkdf2.hashpwsalt(password))
	end

	defp put_password_hash(changeset), do: changeset

	def is_internal_user?(%User{nickname: nil}), do: true
	def is_internal_user?(%User{local: true, nickname: "internal." <> _}), do: true
	def is_internal_user?(_), do: false

	# A hack because user delete activities have a fake id for whatever reason
	# TODO: Get rid of this
	def get_delivered_users_by_object_id("pleroma:fake_object_id"), do: []

	def get_delivered_users_by_object_id(object_id) do
	from(u in User,
	inner_join: delivery in assoc(u, :deliveries),
	where: delivery.object_id == ^object_id
	)
	\|> Repo.all()
	end

	def change_email(user, email) do
	user
	\|> cast(%{email: email}, [:email])
	\|> validate_required([:email])
	\|> unique_constraint(:email)
	\|> validate_format(:email, @email_regex)
	\|> update_and_set_cache()
	end

	# Internal function; public one is `deactivate/2`
	defp set_activation_status(user, deactivated) do
	user
	\|> cast(%{deactivated: deactivated}, [:deactivated])
	\|> update_and_set_cache()
	end

	def update_banner(user, banner) do
	user
	\|> cast(%{banner: banner}, [:banner])
	\|> update_and_set_cache()
	end

	def update_background(user, background) do
	user
	\|> cast(%{background: background}, [:background])
	\|> update_and_set_cache()
	end

	def update_source_data(user, source_data) do
	user
	\|> cast(%{source_data: source_data}, [:source_data])
	\|> update_and_set_cache()
	end

	def roles(%{is_moderator: is_moderator, is_admin: is_admin}) do
	%{
	admin: is_admin,
	moderator: is_moderator
	}
	end

	# ``fields`` is an array of mastodon profile field, containing ``{"name": "…", "value": "…"}``.
	# For example: [{"name": "Pronoun", "value": "she/her"}, …]
	def fields(%{fields: nil, source_data: %{"attachment" => attachment}}) do
	limit = Pleroma.Config.get([:instance, :max_remote_account_fields], 0)

	attachment
	\|> Enum.filter(fn %{"type" => t} -> t == "PropertyValue" end)
	\|> Enum.map(fn fields -> Map.take(fields, ["name", "value"]) end)
	\|> Enum.take(limit)
	end

	def fields(%{fields: nil}), do: []

	def fields(%{fields: fields}), do: fields

	def validate_fields(changeset, remote? \\ false) do
	limit_name = if remote?, do: :max_remote_account_fields, else: :max_account_fields
	limit = Pleroma.Config.get([:instance, limit_name], 0)

	changeset
	\|> validate_length(:fields, max: limit)
	\|> validate_change(:fields, fn :fields, fields ->
	if Enum.all?(fields, &valid_field?/1) do
	[]
	else
	[fields: "invalid"]
	end
	end)
	end

	defp valid_field?(%{"name" => name, "value" => value}) do
	name_limit = Pleroma.Config.get([:instance, :account_field_name_length], 255)
	value_limit = Pleroma.Config.get([:instance, :account_field_value_length], 255)

	is_binary(name) && is_binary(value) && String.length(name) <= name_limit &&
	String.length(value) <= value_limit
	end

	defp valid_field?(_), do: false

	defp truncate_field(%{"name" => name, "value" => value}) do
	{name, _chopped} =
	String.split_at(name, Pleroma.Config.get([:instance, :account_field_name_length], 255))

	{value, _chopped} =
	String.split_at(value, Pleroma.Config.get([:instance, :account_field_value_length], 255))

	%{"name" => name, "value" => value}
	end

	def admin_api_update(user, params) do
	user
	\|> cast(params, [
	:is_moderator,
	:is_admin,
	:show_role
	])
	\|> update_and_set_cache()
	end

	@doc "Signs user out of all applications"
	def global_sign_out(user) do
	OAuth.Authorization.delete_user_authorizations(user)
	OAuth.Token.delete_user_tokens(user)
	end

	def mascot_update(user, url) do
	user
	\|> cast(%{mascot: url}, [:mascot])
	\|> validate_required([:mascot])
	\|> update_and_set_cache()
	end

	def mastodon_settings_update(user, settings) do
	user
	\|> cast(%{settings: settings}, [:settings])
	\|> validate_required([:settings])
	\|> update_and_set_cache()
	end

	@spec confirmation_changeset(User.t(), keyword()) :: Changeset.t()
	def confirmation_changeset(user, need_confirmation: need_confirmation?) do
	params =
	if need_confirmation? do
	%{
	confirmation_pending: true,
	confirmation_token: :crypto.strong_rand_bytes(32) \|> Base.url_encode64()
	}
	else
	%{
	confirmation_pending: false,
	confirmation_token: nil
	}
	end

	cast(user, params, [:confirmation_pending, :confirmation_token])
	end

	def add_pinnned_activity(user, %Pleroma.Activity{id: id}) do
	if id not in user.pinned_activities do
	max_pinned_statuses = Pleroma.Config.get([:instance, :max_pinned_statuses], 0)
	params = %{pinned_activities: user.pinned_activities ++ [id]}

	user
	\|> cast(params, [:pinned_activities])
	\|> validate_length(:pinned_activities,
	max: max_pinned_statuses,
	message: "You have already pinned the maximum number of statuses"
	)
	else
	change(user)
	end
	\|> update_and_set_cache()
	end

	def remove_pinnned_activity(user, %Pleroma.Activity{id: id}) do
	params = %{pinned_activities: List.delete(user.pinned_activities, id)}

	user
	\|> cast(params, [:pinned_activities])
	\|> update_and_set_cache()
	end

	def update_email_notifications(user, settings) do
	email_notifications =
	user.email_notifications
	\|> Map.merge(settings)
	\|> Map.take(["digest"])

	params = %{email_notifications: email_notifications}
	fields = [:email_notifications]

	user
	\|> cast(params, fields)
	\|> validate_required(fields)
	\|> update_and_set_cache()
	end

	defp set_domain_blocks(user, domain_blocks) do
	params = %{domain_blocks: domain_blocks}

	user
	\|> cast(params, [:domain_blocks])
	\|> validate_required([:domain_blocks])
	\|> update_and_set_cache()
	end

	def block_domain(user, domain_blocked) do
	set_domain_blocks(user, Enum.uniq([domain_blocked \| user.domain_blocks]))
	end

	def unblock_domain(user, domain_blocked) do
	set_domain_blocks(user, List.delete(user.domain_blocks, domain_blocked))
	end

	@spec add_to_block(User.t(), User.t()) ::
	{:ok, UserRelationship.t()} \| {:error, Ecto.Changeset.t()}
	defp add_to_block(%User{} = user, %User{} = blocked) do
	UserRelationship.create_block(user, blocked)
	end

	@spec add_to_block(User.t(), User.t()) ::
	{:ok, UserRelationship.t()} \| {:ok, nil} \| {:error, Ecto.Changeset.t()}
	defp remove_from_block(%User{} = user, %User{} = blocked) do
	UserRelationship.delete_block(user, blocked)
	end

	defp add_to_mutes(%User{} = user, %User{} = muted_user, notifications?) do
	with {:ok, user_mute} <- UserRelationship.create_mute(user, muted_user),
	{:ok, user_notification_mute} <-
	(notifications? && UserRelationship.create_notification_mute(user, muted_user)) \|\|
	{:ok, nil} do
	{:ok, Enum.filter([user_mute, user_notification_mute], & &1)}
	end
	end

	defp remove_from_mutes(user, %User{} = muted_user) do
	with {:ok, user_mute} <- UserRelationship.delete_mute(user, muted_user),
	{:ok, user_notification_mute} <-
	UserRelationship.delete_notification_mute(user, muted_user) do
	{:ok, [user_mute, user_notification_mute]}
	end
	end

	def set_invisible(user, invisible) do
	params = %{invisible: invisible}

	user
	\|> cast(params, [:invisible])
	\|> validate_required([:invisible])
	\|> update_and_set_cache()
	end

	def sanitize_html(%User{} = user) do
	sanitize_html(user, nil)
	end

	# User data that mastodon isn't filtering (treated as plaintext):
	# - field name
	# - display name
	def sanitize_html(%User{} = user, filter) do
	fields =
	user
	\|> User.fields()
	\|> Enum.map(fn %{"name" => name, "value" => value} ->
	%{
	"name" => name,
	"value" => HTML.filter_tags(value, Pleroma.HTML.Scrubber.LinksOnly)
	}
	end)

	user
	\|> Map.put(:bio, HTML.filter_tags(user.bio, filter))
	\|> Map.put(:fields, fields)
	end
	end
	diff --git a/lib/pleroma/web/activity_pub/activity_pub.ex b/lib/pleroma/web/activity_pub/activity_pub.ex
	index d9f74b6a4..37d9e1f44 100644
	--- a/lib/pleroma/web/activity_pub/activity_pub.ex
	+++ b/lib/pleroma/web/activity_pub/activity_pub.ex
	@@ -1,1552 +1,1554 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Web.ActivityPub.ActivityPub do
	alias Pleroma.Activity
	alias Pleroma.Activity.Ir.Topics
	alias Pleroma.Config
	alias Pleroma.Constants
	alias Pleroma.Conversation
	alias Pleroma.Conversation.Participation
	alias Pleroma.Notification
	alias Pleroma.Object
	alias Pleroma.Object.Containment
	alias Pleroma.Object.Fetcher
	alias Pleroma.Pagination
	alias Pleroma.Repo
	alias Pleroma.Upload
	alias Pleroma.User
	alias Pleroma.Web.ActivityPub.MRF
	alias Pleroma.Web.ActivityPub.Transmogrifier
	alias Pleroma.Web.ActivityPub.Utils
	alias Pleroma.Web.Streamer
	alias Pleroma.Web.WebFinger
	alias Pleroma.Workers.BackgroundWorker

	import Ecto.Query
	import Pleroma.Web.ActivityPub.Utils
	import Pleroma.Web.ActivityPub.Visibility

	require Logger
	require Pleroma.Constants

	# For Announce activities, we filter the recipients based on following status for any actors
	# that match actual users. See issue #164 for more information about why this is necessary.
	defp get_recipients(%{"type" => "Announce"} = data) do
	to = Map.get(data, "to", [])
	cc = Map.get(data, "cc", [])
	bcc = Map.get(data, "bcc", [])
	actor = User.get_cached_by_ap_id(data["actor"])

	recipients =
	Enum.filter(Enum.concat([to, cc, bcc]), fn recipient ->
	case User.get_cached_by_ap_id(recipient) do
	nil -> true
	user -> User.following?(user, actor)
	end
	end)

	{recipients, to, cc}
	end

	defp get_recipients(%{"type" => "Create"} = data) do
	to = Map.get(data, "to", [])
	cc = Map.get(data, "cc", [])
	bcc = Map.get(data, "bcc", [])
	actor = Map.get(data, "actor", [])
	recipients = [to, cc, bcc, [actor]] \|> Enum.concat() \|> Enum.uniq()
	{recipients, to, cc}
	end

	defp get_recipients(data) do
	to = Map.get(data, "to", [])
	cc = Map.get(data, "cc", [])
	bcc = Map.get(data, "bcc", [])
	recipients = Enum.concat([to, cc, bcc])
	{recipients, to, cc}
	end

	defp check_actor_is_active(actor) do
	if not is_nil(actor) do
	with user <- User.get_cached_by_ap_id(actor),
	false <- user.deactivated do
	true
	else
	_e -> false
	end
	else
	true
	end
	end

	defp check_remote_limit(%{"object" => %{"content" => content}}) when not is_nil(content) do
	limit = Config.get([:instance, :remote_limit])
	String.length(content) <= limit
	end

	defp check_remote_limit(_), do: true

	def increase_note_count_if_public(actor, object) do
	if is_public?(object), do: User.increase_note_count(actor), else: {:ok, actor}
	end

	def decrease_note_count_if_public(actor, object) do
	if is_public?(object), do: User.decrease_note_count(actor), else: {:ok, actor}
	end

	def increase_replies_count_if_reply(%{
	"object" => %{"inReplyTo" => reply_ap_id} = object,
	"type" => "Create"
	}) do
	if is_public?(object) do
	Object.increase_replies_count(reply_ap_id)
	end
	end

	def increase_replies_count_if_reply(_create_data), do: :noop

	def decrease_replies_count_if_reply(%Object{
	data: %{"inReplyTo" => reply_ap_id} = object
	}) do
	if is_public?(object) do
	Object.decrease_replies_count(reply_ap_id)
	end
	end

	def decrease_replies_count_if_reply(_object), do: :noop

	def increase_poll_votes_if_vote(%{
	"object" => %{"inReplyTo" => reply_ap_id, "name" => name},
	"type" => "Create"
	}) do
	Object.increase_vote_count(reply_ap_id, name)
	end

	def increase_poll_votes_if_vote(_create_data), do: :noop

	@spec insert(map(), boolean(), boolean(), boolean()) :: {:ok, Activity.t()} \| {:error, any()}
	def insert(map, local \\ true, fake \\ false, bypass_actor_check \\ false) when is_map(map) do
	with nil <- Activity.normalize(map),
	map <- lazy_put_activity_defaults(map, fake),
	true <- bypass_actor_check \|\| check_actor_is_active(map["actor"]),
	{_, true} <- {:remote_limit_error, check_remote_limit(map)},
	{:ok, map} <- MRF.filter(map),
	{recipients, _, _} = get_recipients(map),
	{:fake, false, map, recipients} <- {:fake, fake, map, recipients},
	{:containment, :ok} <- {:containment, Containment.contain_child(map)},
	{:ok, map, object} <- insert_full_object(map) do
	{:ok, activity} =
	Repo.insert(%Activity{
	data: map,
	local: local,
	actor: map["actor"],
	recipients: recipients
	})

	# Splice in the child object if we have one.
	activity =
	if not is_nil(object) do
	Map.put(activity, :object, object)
	else
	activity
	end

	BackgroundWorker.enqueue("fetch_data_for_activity", %{"activity_id" => activity.id})

	Notification.create_notifications(activity)

	conversation = create_or_bump_conversation(activity, map["actor"])
	participations = get_participations(conversation)
	stream_out(activity)
	stream_out_participations(participations)
	{:ok, activity}
	else
	%Activity{} = activity ->
	{:ok, activity}

	{:fake, true, map, recipients} ->
	activity = %Activity{
	data: map,
	local: local,
	actor: map["actor"],
	recipients: recipients,
	id: "pleroma:fakeid"
	}

	Pleroma.Web.RichMedia.Helpers.fetch_data_for_activity(activity)
	{:ok, activity}

	error ->
	{:error, error}
	end
	end

	defp create_or_bump_conversation(activity, actor) do
	with {:ok, conversation} <- Conversation.create_or_bump_for(activity),
	%User{} = user <- User.get_cached_by_ap_id(actor),
	Participation.mark_as_read(user, conversation) do
	{:ok, conversation}
	end
	end

	defp get_participations({:ok, conversation}) do
	conversation
	\|> Repo.preload(:participations, force: true)
	\|> Map.get(:participations)
	end

	defp get_participations(_), do: []

	def stream_out_participations(participations) do
	participations =
	participations
	\|> Repo.preload(:user)

	Streamer.stream("participation", participations)
	end

	def stream_out_participations(%Object{data: %{"context" => context}}, user) do
	with %Conversation{} = conversation <- Conversation.get_for_ap_id(context),
	conversation = Repo.preload(conversation, :participations),
	last_activity_id =
	fetch_latest_activity_id_for_context(conversation.ap_id, %{
	"user" => user,
	"blocking_user" => user
	}) do
	if last_activity_id do
	stream_out_participations(conversation.participations)
	end
	end
	end

	def stream_out_participations(_, _), do: :noop

	def stream_out(%Activity{data: %{"type" => data_type}} = activity)
	when data_type in ["Create", "Announce", "Delete"] do
	activity
	\|> Topics.get_activity_topics()
	\|> Streamer.stream(activity)
	end

	def stream_out(_activity) do
	:noop
	end

	@spec create(map(), boolean()) :: {:ok, Activity.t()} \| {:error, any()}
	def create(params, fake \\ false) do
	with {:ok, result} <- Repo.transaction(fn -> do_create(params, fake) end) do
	result
	end
	end

	defp do_create(%{to: to, actor: actor, context: context, object: object} = params, fake) do
	additional = params[:additional] \|\| %{}
	# only accept false as false value
	local = !(params[:local] == false)
	published = params[:published]
	quick_insert? = Config.get([:env]) == :benchmark

	with create_data <-
	make_create_data(
	%{to: to, actor: actor, published: published, context: context, object: object},
	additional
	),
	{:ok, activity} <- insert(create_data, local, fake),
	{:fake, false, activity} <- {:fake, fake, activity},
	_ <- increase_replies_count_if_reply(create_data),
	_ <- increase_poll_votes_if_vote(create_data),
	{:quick_insert, false, activity} <- {:quick_insert, quick_insert?, activity},
	{:ok, _actor} <- increase_note_count_if_public(actor, activity),
	:ok <- maybe_federate(activity) do
	{:ok, activity}
	else
	{:quick_insert, true, activity} ->
	{:ok, activity}

	{:fake, true, activity} ->
	{:ok, activity}

	{:error, message} ->
	Repo.rollback(message)
	end
	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]

	with listen_data <-
	make_listen_data(
	%{to: to, actor: actor, published: published, context: context, object: object},
	additional
	),
	{:ok, activity} <- insert(listen_data, local),
	:ok <- maybe_federate(activity) do
	{:ok, activity}
	end
	end

	@spec accept(map()) :: {:ok, Activity.t()} \| {:error, any()}
	def accept(params) do
	accept_or_reject("Accept", params)
	end

	@spec reject(map()) :: {:ok, Activity.t()} \| {:error, any()}
	def reject(params) do
	accept_or_reject("Reject", params)
	end

	@spec accept_or_reject(String.t(), map()) :: {:ok, Activity.t()} \| {:error, any()}
	def accept_or_reject(type, %{to: to, actor: actor, object: object} = params) do
	local = Map.get(params, :local, true)
	activity_id = Map.get(params, :activity_id, nil)

	with data <-
	%{"to" => to, "type" => type, "actor" => actor.ap_id, "object" => object}
	\|> Utils.maybe_put("id", activity_id),
	{:ok, activity} <- insert(data, local),
	:ok <- maybe_federate(activity) do
	{:ok, activity}
	end
	end

	@spec update(map()) :: {:ok, Activity.t()} \| {:error, any()}
	def update(%{to: to, cc: cc, actor: actor, object: object} = params) do
	local = !(params[:local] == false)
	activity_id = params[:activity_id]

	with data <- %{
	"to" => to,
	"cc" => cc,
	"type" => "Update",
	"actor" => actor,
	"object" => object
	},
	data <- Utils.maybe_put(data, "id", activity_id),
	{:ok, activity} <- insert(data, local),
	:ok <- maybe_federate(activity) do
	{:ok, activity}
	end
	end

	@spec react_with_emoji(User.t(), Object.t(), String.t(), keyword()) ::
	{:ok, Activity.t(), Object.t()} \| {:error, any()}
	def react_with_emoji(user, object, emoji, options \\ []) do
	with {:ok, result} <-
	Repo.transaction(fn -> do_react_with_emoji(user, object, emoji, options) end) do
	result
	end
	end

	defp do_react_with_emoji(user, object, emoji, options) do
	with local <- Keyword.get(options, :local, true),
	activity_id <- Keyword.get(options, :activity_id, nil),
	true <- Pleroma.Emoji.is_unicode_emoji?(emoji),
	reaction_data <- make_emoji_reaction_data(user, object, emoji, activity_id),
	{:ok, activity} <- insert(reaction_data, local),
	{:ok, object} <- add_emoji_reaction_to_object(activity, object),
	:ok <- maybe_federate(activity) do
	{:ok, activity, object}
	else
	false -> {:error, false}
	{:error, error} -> Repo.rollback(error)
	end
	end

	@spec unreact_with_emoji(User.t(), String.t(), keyword()) ::
	{:ok, Activity.t(), Object.t()} \| {:error, any()}
	def unreact_with_emoji(user, reaction_id, options \\ []) do
	with {:ok, result} <-
	Repo.transaction(fn -> do_unreact_with_emoji(user, reaction_id, options) end) do
	result
	end
	end

	defp do_unreact_with_emoji(user, reaction_id, options) do
	with local <- Keyword.get(options, :local, true),
	activity_id <- Keyword.get(options, :activity_id, nil),
	user_ap_id <- user.ap_id,
	%Activity{actor: ^user_ap_id} = reaction_activity <- Activity.get_by_ap_id(reaction_id),
	object <- Object.normalize(reaction_activity),
	unreact_data <- make_undo_data(user, reaction_activity, activity_id),
	{:ok, activity} <- insert(unreact_data, local),
	{:ok, object} <- remove_emoji_reaction_from_object(reaction_activity, object),
	:ok <- maybe_federate(activity) do
	{:ok, activity, object}
	else
	{:error, error} -> Repo.rollback(error)
	end
	end

	# TODO: This is weird, maybe we shouldn't check here if we can make the activity.
	@spec like(User.t(), Object.t(), String.t() \| nil, boolean()) ::
	{:ok, Activity.t(), Object.t()} \| {:error, any()}
	def like(user, object, activity_id \\ nil, local \\ true) do
	with {:ok, result} <- Repo.transaction(fn -> do_like(user, object, activity_id, local) end) do
	result
	end
	end

	defp do_like(
	%User{ap_id: ap_id} = user,
	%Object{data: %{"id" => _}} = object,
	activity_id,
	local
	) do
	with nil <- get_existing_like(ap_id, object),
	like_data <- make_like_data(user, object, activity_id),
	{:ok, activity} <- insert(like_data, local),
	{:ok, object} <- add_like_to_object(activity, object),
	:ok <- maybe_federate(activity) do
	{:ok, activity, object}
	else
	%Activity{} = activity ->
	{:ok, activity, object}

	{:error, error} ->
	Repo.rollback(error)
	end
	end

	@spec unlike(User.t(), Object.t(), String.t() \| nil, boolean()) ::
	{:ok, Activity.t(), Activity.t(), Object.t()} \| {:ok, Object.t()} \| {:error, any()}
	def unlike(%User{} = actor, %Object{} = object, activity_id \\ nil, local \\ true) do
	with {:ok, result} <-
	Repo.transaction(fn -> do_unlike(actor, object, activity_id, local) end) do
	result
	end
	end

	defp do_unlike(actor, object, activity_id, local) do
	with %Activity{} = like_activity <- get_existing_like(actor.ap_id, object),
	unlike_data <- make_unlike_data(actor, like_activity, activity_id),
	{:ok, unlike_activity} <- insert(unlike_data, local),
	{:ok, _activity} <- Repo.delete(like_activity),
	{:ok, object} <- remove_like_from_object(like_activity, object),
	:ok <- maybe_federate(unlike_activity) do
	{:ok, unlike_activity, like_activity, object}
	else
	nil -> {:ok, object}
	{:error, error} -> Repo.rollback(error)
	end
	end

	@spec announce(User.t(), Object.t(), String.t() \| nil, boolean(), boolean()) ::
	{:ok, Activity.t(), Object.t()} \| {:error, any()}
	def announce(
	%User{ap_id: _} = user,
	%Object{data: %{"id" => _}} = object,
	activity_id \\ nil,
	local \\ true,
	public \\ true
	) do
	with {:ok, result} <-
	Repo.transaction(fn -> do_announce(user, object, activity_id, local, public) end) do
	result
	end
	end

	defp do_announce(user, object, activity_id, local, public) do
	with true <- is_announceable?(object, user, public),
	announce_data <- make_announce_data(user, object, activity_id, public),
	{:ok, activity} <- insert(announce_data, local),
	{:ok, object} <- add_announce_to_object(activity, object),
	:ok <- maybe_federate(activity) do
	{:ok, activity, object}
	else
	false -> {:error, false}
	{:error, error} -> Repo.rollback(error)
	end
	end

	@spec unannounce(User.t(), Object.t(), String.t() \| nil, boolean()) ::
	{:ok, Activity.t(), Object.t()} \| {:ok, Object.t()} \| {:error, any()}
	def unannounce(
	%User{} = actor,
	%Object{} = object,
	activity_id \\ nil,
	local \\ true
	) do
	with {:ok, result} <-
	Repo.transaction(fn -> do_unannounce(actor, object, activity_id, local) end) do
	result
	end
	end

	defp do_unannounce(actor, object, activity_id, local) do
	with %Activity{} = announce_activity <- get_existing_announce(actor.ap_id, object),
	unannounce_data <- make_unannounce_data(actor, announce_activity, activity_id),
	{:ok, unannounce_activity} <- insert(unannounce_data, local),
	:ok <- maybe_federate(unannounce_activity),
	{:ok, _activity} <- Repo.delete(announce_activity),
	{:ok, object} <- remove_announce_from_object(announce_activity, object) do
	{:ok, unannounce_activity, object}
	else
	nil -> {:ok, object}
	{:error, error} -> Repo.rollback(error)
	end
	end

	@spec follow(User.t(), User.t(), String.t() \| nil, boolean()) ::
	{:ok, Activity.t()} \| {:error, any()}
	def follow(follower, followed, activity_id \\ nil, local \\ true) do
	with {:ok, result} <-
	Repo.transaction(fn -> do_follow(follower, followed, activity_id, local) end) do
	result
	end
	end

	defp do_follow(follower, followed, activity_id, local) do
	with data <- make_follow_data(follower, followed, activity_id),
	{:ok, activity} <- insert(data, local),
	:ok <- maybe_federate(activity),
	_ <- User.set_follow_state_cache(follower.ap_id, followed.ap_id, activity.data["state"]) do
	{:ok, activity}
	else
	{:error, error} -> Repo.rollback(error)
	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
	with {:ok, result} <-
	Repo.transaction(fn -> do_unfollow(follower, followed, activity_id, local) end) do
	result
	end
	end

	defp do_unfollow(follower, followed, activity_id, local) do
	with %Activity{} = follow_activity <- fetch_latest_follow(follower, followed),
	{:ok, follow_activity} <- update_follow_state(follow_activity, "cancelled"),
	unfollow_data <- make_unfollow_data(follower, followed, follow_activity, activity_id),
	{:ok, activity} <- insert(unfollow_data, local),
	:ok <- maybe_federate(activity) do
	{:ok, activity}
	else
	nil -> nil
	{:error, error} -> Repo.rollback(error)
	end
	end

	@spec delete(User.t() \| Object.t(), keyword()) :: {:ok, User.t() \| Object.t()} \| {:error, any()}
	def delete(entity, options \\ []) do
	with {:ok, result} <- Repo.transaction(fn -> do_delete(entity, options) end) do
	result
	end
	end

	defp do_delete(%User{ap_id: ap_id, follower_address: follower_address} = user, _) do
	with data <- %{
	"to" => [follower_address],
	"type" => "Delete",
	"actor" => ap_id,
	"object" => %{"type" => "Person", "id" => ap_id}
	},
	{:ok, activity} <- insert(data, true, true, true),
	:ok <- maybe_federate(activity) do
	{:ok, user}
	end
	end

	defp do_delete(%Object{data: %{"id" => id, "actor" => actor}} = object, options) do
	local = Keyword.get(options, :local, true)
	activity_id = Keyword.get(options, :activity_id, nil)
	actor = Keyword.get(options, :actor, actor)

	user = User.get_cached_by_ap_id(actor)
	to = (object.data["to"] \|\| []) ++ (object.data["cc"] \|\| [])

	with create_activity <- Activity.get_create_by_object_ap_id(id),
	data <-
	%{
	"type" => "Delete",
	"actor" => actor,
	"object" => id,
	"to" => to,
	"deleted_activity_id" => create_activity && create_activity.id
	}
	\|> maybe_put("id", activity_id),
	{:ok, activity} <- insert(data, local, false),
	{:ok, object, _create_activity} <- Object.delete(object),
	stream_out_participations(object, user),
	_ <- decrease_replies_count_if_reply(object),
	{:ok, _actor} <- decrease_note_count_if_public(user, object),
	:ok <- maybe_federate(activity) do
	{:ok, activity}
	else
	{:error, error} ->
	Repo.rollback(error)
	end
	end

	@spec block(User.t(), User.t(), String.t() \| nil, boolean()) ::
	{:ok, Activity.t()} \| {:error, any()}
	def block(blocker, blocked, activity_id \\ nil, local \\ true) do
	with {:ok, result} <-
	Repo.transaction(fn -> do_block(blocker, blocked, activity_id, local) end) do
	result
	end
	end

	defp do_block(blocker, blocked, activity_id, local) do
	outgoing_blocks = Config.get([:activitypub, :outgoing_blocks])
	unfollow_blocked = Config.get([:activitypub, :unfollow_blocked])

	if unfollow_blocked do
	follow_activity = fetch_latest_follow(blocker, blocked)
	if follow_activity, do: unfollow(blocker, blocked, nil, local)
	end

	with true <- outgoing_blocks,
	block_data <- make_block_data(blocker, blocked, activity_id),
	{:ok, activity} <- insert(block_data, local),
	:ok <- maybe_federate(activity) do
	{:ok, activity}
	else
	{:error, error} -> Repo.rollback(error)
	end
	end

	@spec unblock(User.t(), User.t(), String.t() \| nil, boolean()) ::
	{:ok, Activity.t()} \| {:error, any()} \| nil
	def unblock(blocker, blocked, activity_id \\ nil, local \\ true) do
	with {:ok, result} <-
	Repo.transaction(fn -> do_unblock(blocker, blocked, activity_id, local) end) do
	result
	end
	end

	defp do_unblock(blocker, blocked, activity_id, local) do
	with %Activity{} = block_activity <- fetch_latest_block(blocker, blocked),
	unblock_data <- make_unblock_data(blocker, blocked, block_activity, activity_id),
	{:ok, activity} <- insert(unblock_data, local),
	:ok <- maybe_federate(activity) do
	{:ok, activity}
	else
	nil -> nil
	{:error, error} -> Repo.rollback(error)
	end
	end

	@spec flag(map()) :: {:ok, Activity.t()} \| {:error, any()}
	def flag(
	%{
	actor: actor,
	context: _context,
	account: account,
	statuses: statuses,
	content: content
	} = params
	) do
	# only accept false as false value
	local = !(params[:local] == false)
	forward = !(params[:forward] == false)

	additional = params[:additional] \|\| %{}

	additional =
	if forward do
	Map.merge(additional, %{"to" => [], "cc" => [account.ap_id]})
	else
	Map.merge(additional, %{"to" => [], "cc" => []})
	end

	with flag_data <- make_flag_data(params, additional),
	{:ok, activity} <- insert(flag_data, local),
	{:ok, stripped_activity} <- strip_report_status_data(activity),
	:ok <- maybe_federate(stripped_activity) do
	User.all_superusers()
	\|> Enum.filter(fn user -> not is_nil(user.email) end)
	\|> Enum.each(fn superuser ->
	superuser
	\|> Pleroma.Emails.AdminEmail.report(actor, account, statuses, content)
	\|> Pleroma.Emails.Mailer.deliver_async()
	end)

	{:ok, activity}
	end
	end

	@spec move(User.t(), User.t(), boolean()) :: {:ok, Activity.t()} \| {:error, any()}
	def move(%User{} = origin, %User{} = target, local \\ true) do
	params = %{
	"type" => "Move",
	"actor" => origin.ap_id,
	"object" => origin.ap_id,
	"target" => target.ap_id
	}

	with true <- origin.ap_id in target.also_known_as,
	{:ok, activity} <- insert(params, local) do
	maybe_federate(activity)

	BackgroundWorker.enqueue("move_following", %{
	"origin_id" => origin.id,
	"target_id" => target.id
	})

	{:ok, activity}
	else
	false -> {:error, "Target account must have the origin in `alsoKnownAs`"}
	err -> err
	end
	end

	defp fetch_activities_for_context_query(context, opts) do
	public = [Constants.as_public()]

	recipients =
	if opts["user"],
	do: [opts["user"].ap_id \| User.following(opts["user"])] ++ public,
	else: public

	from(activity in Activity)
	\|> maybe_preload_objects(opts)
	\|> maybe_preload_bookmarks(opts)
	\|> maybe_set_thread_muted_field(opts)
	\|> restrict_blocked(opts)
	\|> restrict_recipients(recipients, opts["user"])
	\|> where(
	[activity],
	fragment(
	"?->>'type' = ? and ?->>'context' = ?",
	activity.data,
	"Create",
	activity.data,
	^context
	)
	)
	\|> exclude_poll_votes(opts)
	\|> exclude_id(opts)
	\|> order_by([activity], desc: activity.id)
	end

	@spec fetch_activities_for_context(String.t(), keyword() \| map()) :: [Activity.t()]
	def fetch_activities_for_context(context, opts \\ %{}) do
	context
	\|> fetch_activities_for_context_query(opts)
	\|> Repo.all()
	end

	@spec fetch_latest_activity_id_for_context(String.t(), keyword() \| map()) ::
	FlakeId.Ecto.CompatType.t() \| nil
	def fetch_latest_activity_id_for_context(context, opts \\ %{}) do
	context
	\|> fetch_activities_for_context_query(Map.merge(%{"skip_preload" => true}, opts))
	\|> limit(1)
	\|> select([a], a.id)
	\|> Repo.one()
	end

	@spec fetch_public_activities(map(), Pagination.type()) :: [Activity.t()]
	def fetch_public_activities(opts \\ %{}, pagination \\ :keyset) do
	opts = Map.drop(opts, ["user"])

	[Constants.as_public()]
	\|> fetch_activities_query(opts)
	\|> restrict_unlisted()
	\|> Pagination.fetch_paginated(opts, pagination)
	end

	@valid_visibilities ~w[direct unlisted public private]

	defp restrict_visibility(query, %{visibility: visibility})
	when is_list(visibility) do
	if Enum.all?(visibility, &(&1 in @valid_visibilities)) do
	query =
	from(
	a in query,
	where:
	fragment(
	"activity_visibility(?, ?, ?) = ANY (?)",
	a.actor,
	a.recipients,
	a.data,
	^visibility
	)
	)

	query
	else
	Logger.error("Could not restrict visibility to #{visibility}")
	end
	end

	defp restrict_visibility(query, %{visibility: visibility})
	when visibility in @valid_visibilities do
	from(
	a in query,
	where:
	fragment("activity_visibility(?, ?, ?) = ?", a.actor, a.recipients, a.data, ^visibility)
	)
	end

	defp restrict_visibility(_query, %{visibility: visibility})
	when visibility not in @valid_visibilities do
	Logger.error("Could not restrict visibility to #{visibility}")
	end

	defp restrict_visibility(query, _visibility), do: query

	defp exclude_visibility(query, %{"exclude_visibilities" => visibility})
	when is_list(visibility) do
	if Enum.all?(visibility, &(&1 in @valid_visibilities)) do
	from(
	a in query,
	where:
	not fragment(
	"activity_visibility(?, ?, ?) = ANY (?)",
	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
	from(
	a in query,
	where:
	not fragment(
	"activity_visibility(?, ?, ?) = ?",
	a.actor,
	a.recipients,
	a.data,
	^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

	defp restrict_thread_visibility(query, _, %{skip_thread_containment: true} = _),
	do: query

	defp restrict_thread_visibility(
	query,
	%{"user" => %User{skip_thread_containment: true}},
	_
	),
	do: query

	defp restrict_thread_visibility(query, %{"user" => %User{ap_id: ap_id}}, _) do
	from(
	a in query,
	where: fragment("thread_visibility(?, (?)->>'id') = true", ^ap_id, a.data)
	)
	end

	defp restrict_thread_visibility(query, _, _), do: query

	def fetch_user_abstract_activities(user, reading_user, params \\ %{}) do
	params =
	params
	\|> Map.put("user", reading_user)
	\|> Map.put("actor_id", user.ap_id)

	recipients =
	user_activities_recipients(%{
	"godmode" => params["godmode"],
	"reading_user" => reading_user
	})

	fetch_activities(recipients, params)
	\|> Enum.reverse()
	end

	def fetch_user_activities(user, reading_user, params \\ %{}) do
	params =
	params
	\|> Map.put("type", ["Create", "Announce"])
	\|> Map.put("user", reading_user)
	\|> Map.put("actor_id", user.ap_id)
	\|> Map.put("pinned_activity_ids", user.pinned_activities)

	params =
	if User.blocks?(reading_user, user) do
	params
	else
	params
	\|> Map.put("blocking_user", reading_user)
	\|> Map.put("muting_user", reading_user)
	end

	recipients =
	user_activities_recipients(%{
	"godmode" => params["godmode"],
	"reading_user" => reading_user
	})

	fetch_activities(recipients, params)
	\|> Enum.reverse()
	end

	def fetch_statuses(reading_user, params) do
	params =
	params
	\|> Map.put("type", ["Create", "Announce"])

	recipients =
	user_activities_recipients(%{
	"godmode" => params["godmode"],
	"reading_user" => reading_user
	})

	fetch_activities(recipients, params, :offset)
	\|> Enum.reverse()
	end

	defp user_activities_recipients(%{"godmode" => true}) do
	[]
	end

	defp user_activities_recipients(%{"reading_user" => reading_user}) do
	if reading_user do
	[Constants.as_public()] ++ [reading_user.ap_id \| User.following(reading_user)]
	else
	[Constants.as_public()]
	end
	end

	defp restrict_since(query, %{"since_id" => ""}), do: query

	defp restrict_since(query, %{"since_id" => since_id}) do
	from(activity in query, where: activity.id > ^since_id)
	end

	defp restrict_since(query, _), do: query

	defp restrict_tag_reject(_query, %{"tag_reject" => _tag_reject, "skip_preload" => true}) do
	raise "Can't use the child object without preloading!"
	end

	defp restrict_tag_reject(query, %{"tag_reject" => tag_reject})
	when is_list(tag_reject) and tag_reject != [] do
	from(
	[_activity, object] in query,
	where: fragment("not (?)->'tag' \\?\| (?)", object.data, ^tag_reject)
	)
	end

	defp restrict_tag_reject(query, _), do: query

	defp restrict_tag_all(_query, %{"tag_all" => _tag_all, "skip_preload" => true}) do
	raise "Can't use the child object without preloading!"
	end

	defp restrict_tag_all(query, %{"tag_all" => tag_all})
	when is_list(tag_all) and tag_all != [] do
	from(
	[_activity, object] in query,
	where: fragment("(?)->'tag' \\?& (?)", object.data, ^tag_all)
	)
	end

	defp restrict_tag_all(query, _), do: query

	defp restrict_tag(_query, %{"tag" => _tag, "skip_preload" => true}) do
	raise "Can't use the child object without preloading!"
	end

	defp restrict_tag(query, %{"tag" => tag}) when is_list(tag) do
	from(
	[_activity, object] in query,
	where: fragment("(?)->'tag' \\?\| (?)", object.data, ^tag)
	)
	end

	defp restrict_tag(query, %{"tag" => tag}) when is_binary(tag) do
	from(
	[_activity, object] in query,
	where: fragment("(?)->'tag' \\? (?)", object.data, ^tag)
	)
	end

	defp restrict_tag(query, _), do: query

	defp restrict_recipients(query, [], _user), do: query

	defp restrict_recipients(query, recipients, nil) do
	from(activity in query, where: fragment("? && ?", ^recipients, activity.recipients))
	end

	defp restrict_recipients(query, recipients, user) do
	from(
	activity in query,
	where: fragment("? && ?", ^recipients, activity.recipients),
	or_where: activity.actor == ^user.ap_id
	)
	end

	defp restrict_local(query, %{"local_only" => true}) do
	from(activity in query, where: activity.local == true)
	end

	defp restrict_local(query, _), do: query

	defp restrict_actor(query, %{"actor_id" => actor_id}) do
	from(activity in query, where: activity.actor == ^actor_id)
	end

	defp restrict_actor(query, _), do: query

	defp restrict_type(query, %{"type" => type}) when is_binary(type) do
	from(activity in query, where: fragment("?->>'type' = ?", activity.data, ^type))
	end

	defp restrict_type(query, %{"type" => type}) do
	from(activity in query, where: fragment("?->>'type' = ANY(?)", activity.data, ^type))
	end

	defp restrict_type(query, _), do: query

	defp restrict_state(query, %{"state" => state}) do
	from(activity in query, where: fragment("?->>'state' = ?", activity.data, ^state))
	end

	defp restrict_state(query, _), do: query

	defp restrict_favorited_by(query, %{"favorited_by" => ap_id}) do
	from(
	[_activity, object] in query,
	where: fragment("(?)->'likes' \\? (?)", object.data, ^ap_id)
	)
	end

	defp restrict_favorited_by(query, _), do: query

	defp restrict_media(_query, %{"only_media" => _val, "skip_preload" => true}) do
	raise "Can't use the child object without preloading!"
	end

	defp restrict_media(query, %{"only_media" => val}) when val == "true" or val == "1" do
	from(
	[_activity, object] in query,
	where: fragment("not (?)->'attachment' = (?)", object.data, ^[])
	)
	end

	defp restrict_media(query, _), do: query

	defp restrict_replies(query, %{"exclude_replies" => val}) when val == "true" or val == "1" do
	from(
	[_activity, object] in query,
	where: fragment("?->>'inReplyTo' is null", object.data)
	)
	end

	defp restrict_replies(query, _), do: query

	defp restrict_reblogs(query, %{"exclude_reblogs" => val}) when val == "true" or val == "1" do
	from(activity in query, where: fragment("?->>'type' != 'Announce'", activity.data))
	end

	defp restrict_reblogs(query, _), do: query

	defp restrict_muted(query, %{"with_muted" => val}) when val in [true, "true", "1"], do: query

	defp restrict_muted(query, %{"muting_user" => %User{} = user} = opts) do
	mutes = opts["muted_users_ap_ids"] \|\| User.muted_users_ap_ids(user)

	query =
	from([activity] in query,
	where: fragment("not (? = ANY(?))", activity.actor, ^mutes),
	where: fragment("not (?->'to' \\?\| ?)", activity.data, ^mutes)
	)

	unless opts["skip_preload"] do
	from([thread_mute: tm] in query, where: is_nil(tm.user_id))
	else
	query
	end
	end

	defp restrict_muted(query, _), do: query

	defp restrict_blocked(query, %{"blocking_user" => %User{} = user} = opts) do
	blocked_ap_ids = opts["blocked_users_ap_ids"] \|\| User.blocked_users_ap_ids(user)
	domain_blocks = user.domain_blocks \|\| []

	following_ap_ids = User.get_friends_ap_ids(user)

	query =
	if has_named_binding?(query, :object), do: query, else: Activity.with_joined_object(query)

	from(
	[activity, object: o] in query,
	where: fragment("not (? = ANY(?))", activity.actor, ^blocked_ap_ids),
	where: fragment("not (? && ?)", activity.recipients, ^blocked_ap_ids),
	where:
	fragment(
	"not (?->>'type' = 'Announce' and ?->'to' \\?\| ?)",
	activity.data,
	activity.data,
	^blocked_ap_ids
	),
	where:
	fragment(
	"(not (split_part(?, '/', 3) = ANY(?))) or ? = ANY(?)",
	activity.actor,
	^domain_blocks,
	activity.actor,
	^following_ap_ids
	),
	where:
	fragment(
	"(not (split_part(?->>'actor', '/', 3) = ANY(?))) or (?->>'actor') = ANY(?)",
	o.data,
	^domain_blocks,
	o.data,
	^following_ap_ids
	)
	)
	end

	defp restrict_blocked(query, _), do: query

	defp restrict_unlisted(query) do
	from(
	activity in query,
	where:
	fragment(
	"not (coalesce(?->'cc', '{}'::jsonb) \\?\| ?)",
	activity.data,
	^[Constants.as_public()]
	)
	)
	end

	defp restrict_pinned(query, %{"pinned" => "true", "pinned_activity_ids" => ids}) do
	from(activity in query, where: activity.id in ^ids)
	end

	defp restrict_pinned(query, _), do: query

	defp restrict_muted_reblogs(query, %{"muting_user" => %User{} = user} = opts) do
	muted_reblogs = opts["reblog_muted_users_ap_ids"] \|\| User.reblog_muted_users_ap_ids(user)

	from(
	activity in query,
	where:
	fragment(
	"not ( ?->>'type' = 'Announce' and ? = ANY(?))",
	activity.data,
	activity.actor,
	^muted_reblogs
	)
	)
	end

	defp restrict_muted_reblogs(query, _), do: query

	defp restrict_instance(query, %{"instance" => instance}) do
	users =
	from(
	u in User,
	select: u.ap_id,
	where: fragment("? LIKE ?", u.nickname, ^"%@#{instance}")
	)
	\|> Repo.all()

	from(activity in query, where: activity.actor in ^users)
	end

	defp restrict_instance(query, _), do: query

	defp exclude_poll_votes(query, %{"include_poll_votes" => true}), do: query

	defp exclude_poll_votes(query, _) do
	if has_named_binding?(query, :object) do
	from([activity, object: o] in query,
	where: fragment("not(?->>'type' = ?)", o.data, "Answer")
	)
	else
	query
	end
	end

	defp exclude_id(query, %{"exclude_id" => id}) when is_binary(id) do
	from(activity in query, where: activity.id != ^id)
	end

	defp exclude_id(query, _), do: query

	defp maybe_preload_objects(query, %{"skip_preload" => true}), do: query

	defp maybe_preload_objects(query, _) do
	query
	\|> Activity.with_preloaded_object()
	end

	defp maybe_preload_bookmarks(query, %{"skip_preload" => true}), do: query

	defp maybe_preload_bookmarks(query, opts) do
	query
	\|> Activity.with_preloaded_bookmark(opts["user"])
	end

	defp maybe_preload_report_notes(query, %{"preload_report_notes" => true}) do
	query
	\|> Activity.with_preloaded_report_notes()
	end

	defp maybe_preload_report_notes(query, _), do: query

	defp maybe_set_thread_muted_field(query, %{"skip_preload" => true}), do: query

	defp maybe_set_thread_muted_field(query, opts) do
	query
	\|> Activity.with_set_thread_muted_field(opts["muting_user"] \|\| opts["user"])
	end

	defp maybe_order(query, %{order: :desc}) do
	query
	\|> order_by(desc: :id)
	end

	defp maybe_order(query, %{order: :asc}) do
	query
	\|> order_by(asc: :id)
	end

	defp maybe_order(query, _), do: query

	defp fetch_activities_query_ap_ids_ops(opts) do
	source_user = opts["muting_user"]
	ap_id_relations = if source_user, do: [:mute, :reblog_mute], else: []

	ap_id_relations =
	ap_id_relations ++
	if opts["blocking_user"] && opts["blocking_user"] == source_user do
	[:block]
	else
	[]
	end

	preloaded_ap_ids = User.outgoing_relations_ap_ids(source_user, ap_id_relations)

	restrict_blocked_opts = Map.merge(%{"blocked_users_ap_ids" => preloaded_ap_ids[:block]}, opts)
	restrict_muted_opts = Map.merge(%{"muted_users_ap_ids" => preloaded_ap_ids[:mute]}, opts)

	restrict_muted_reblogs_opts =
	Map.merge(%{"reblog_muted_users_ap_ids" => preloaded_ap_ids[:reblog_mute]}, opts)

	{restrict_blocked_opts, restrict_muted_opts, restrict_muted_reblogs_opts}
	end

	def fetch_activities_query(recipients, opts \\ %{}) do
	{restrict_blocked_opts, restrict_muted_opts, restrict_muted_reblogs_opts} =
	fetch_activities_query_ap_ids_ops(opts)

	config = %{
	skip_thread_containment: Config.get([:instance, :skip_thread_containment])
	}

	Activity
	\|> maybe_preload_objects(opts)
	\|> maybe_preload_bookmarks(opts)
	\|> maybe_preload_report_notes(opts)
	\|> maybe_set_thread_muted_field(opts)
	\|> maybe_order(opts)
	\|> restrict_recipients(recipients, opts["user"])
	\|> restrict_tag(opts)
	\|> restrict_tag_reject(opts)
	\|> restrict_tag_all(opts)
	\|> restrict_since(opts)
	\|> restrict_local(opts)
	\|> restrict_actor(opts)
	\|> restrict_type(opts)
	\|> restrict_state(opts)
	\|> restrict_favorited_by(opts)
	\|> restrict_blocked(restrict_blocked_opts)
	\|> restrict_muted(restrict_muted_opts)
	\|> restrict_media(opts)
	\|> restrict_visibility(opts)
	\|> restrict_thread_visibility(opts, config)
	\|> restrict_replies(opts)
	\|> restrict_reblogs(opts)
	\|> restrict_pinned(opts)
	\|> restrict_muted_reblogs(restrict_muted_reblogs_opts)
	\|> restrict_instance(opts)
	\|> Activity.restrict_deactivated_users()
	\|> exclude_poll_votes(opts)
	\|> exclude_visibility(opts)
	end

	def fetch_activities(recipients, opts \\ %{}, pagination \\ :keyset) do
	list_memberships = Pleroma.List.memberships(opts["user"])

	fetch_activities_query(recipients ++ list_memberships, opts)
	\|> Pagination.fetch_paginated(opts, pagination)
	\|> Enum.reverse()
	\|> maybe_update_cc(list_memberships, opts["user"])
	end

	@doc """
	Fetch favorites activities of user with order by sort adds to favorites
	"""
	@spec fetch_favourites(User.t(), map(), Pagination.type()) :: list(Activity.t())
	def fetch_favourites(user, params \\ %{}, pagination \\ :keyset) do
	user.ap_id
	\|> Activity.Queries.by_actor()
	\|> Activity.Queries.by_type("Like")
	\|> Activity.with_joined_object()
	\|> Object.with_joined_activity()
	\|> select([_like, object, activity], %{activity \| object: object})
	\|> order_by([like, _, _], desc: like.id)
	\|> Pagination.fetch_paginated(
	Map.merge(params, %{"skip_order" => true}),
	pagination,
	:object_activity
	)
	end

	defp maybe_update_cc(activities, list_memberships, %User{ap_id: user_ap_id})
	when is_list(list_memberships) and length(list_memberships) > 0 do
	Enum.map(activities, fn
	%{data: %{"bcc" => bcc}} = activity when is_list(bcc) and length(bcc) > 0 ->
	if Enum.any?(bcc, &(&1 in list_memberships)) do
	update_in(activity.data["cc"], &[user_ap_id \| &1])
	else
	activity
	end

	activity ->
	activity
	end)
	end

	defp maybe_update_cc(activities, _, _), do: activities

	def fetch_activities_bounded_query(query, recipients, recipients_with_public) do
	from(activity in query,
	where:
	fragment("? && ?", activity.recipients, ^recipients) or
	(fragment("? && ?", activity.recipients, ^recipients_with_public) and
	^Constants.as_public() in activity.recipients)
	)
	end

	def fetch_activities_bounded(
	recipients,
	recipients_with_public,
	opts \\ %{},
	pagination \\ :keyset
	) do
	fetch_activities_query([], opts)
	\|> fetch_activities_bounded_query(recipients, recipients_with_public)
	\|> Pagination.fetch_paginated(opts, pagination)
	\|> Enum.reverse()
	end

	@spec upload(Upload.source(), keyword()) :: {:ok, Object.t()} \| {:error, any()}
	def upload(file, opts \\ []) do
	with {:ok, data} <- Upload.store(file, opts) do
	obj_data =
	if opts[:actor] do
	Map.put(data, "actor", opts[:actor])
	else
	data
	end

	Repo.insert(%Object{data: obj_data})
	end
	end

	defp object_to_user_data(data) do
	avatar =
	data["icon"]["url"] &&
	%{
	"type" => "Image",
	"url" => [%{"href" => data["icon"]["url"]}]
	}

	banner =
	data["image"]["url"] &&
	%{
	"type" => "Image",
	"url" => [%{"href" => data["image"]["url"]}]
	}

	fields =
	data
	\|> Map.get("attachment", [])
	\|> Enum.filter(fn %{"type" => t} -> t == "PropertyValue" end)
	\|> Enum.map(fn fields -> Map.take(fields, ["name", "value"]) end)

	locked = data["manuallyApprovesFollowers"] \|\| false
	data = Transmogrifier.maybe_fix_user_object(data)
	discoverable = data["discoverable"] \|\| false
	invisible = data["invisible"] \|\| false
	actor_type = data["type"] \|\| "Person"

	user_data = %{
	ap_id: data["id"],
	ap_enabled: true,
	source_data: data,
	banner: banner,
	fields: fields,
	locked: locked,
	discoverable: discoverable,
	invisible: invisible,
	avatar: avatar,
	name: data["name"],
	follower_address: data["followers"],
	following_address: data["following"],
	bio: data["summary"],
	actor_type: actor_type,
	also_known_as: Map.get(data, "alsoKnownAs", [])
	}

	# nickname can be nil because of virtual actors
	user_data =
	if data["preferredUsername"] do
	Map.put(
	user_data,
	:nickname,
	"#{data["preferredUsername"]}@#{URI.parse(data["id"]).host}"
	)
	else
	Map.put(user_data, :nickname, nil)
	end

	{:ok, user_data}
	end

	def fetch_follow_information_for_user(user) do
	with {:ok, following_data} <-
	- Fetcher.fetch_and_contain_remote_object_from_id(user.following_address),
	+ Fetcher.fetch_and_contain_remote_object_from_id(user.following_address,
	+ force_http: true
	+ ),
	{:ok, hide_follows} <- collection_private(following_data),
	{:ok, followers_data} <-
	Fetcher.fetch_and_contain_remote_object_from_id(user.follower_address),
	{:ok, hide_followers} <- collection_private(followers_data) do
	{:ok,
	%{
	hide_follows: hide_follows,
	follower_count: normalize_counter(followers_data["totalItems"]),
	following_count: normalize_counter(following_data["totalItems"]),
	hide_followers: hide_followers
	}}
	else
	{:error, _} = e -> e
	e -> {:error, e}
	end
	end

	defp normalize_counter(counter) when is_integer(counter), do: counter
	defp normalize_counter(_), do: 0

	defp maybe_update_follow_information(data) do
	with {:enabled, true} <- {:enabled, Config.get([:instance, :external_user_synchronization])},
	{:ok, info} <- fetch_follow_information_for_user(data) do
	info = Map.merge(data[:info] \|\| %{}, info)
	Map.put(data, :info, info)
	else
	{:enabled, false} ->
	data

	e ->
	Logger.error(
	"Follower/Following counter update for #{data.ap_id} failed.\n" <> inspect(e)
	)

	data
	end
	end

	defp collection_private(%{"first" => %{"type" => type}})
	when type in ["CollectionPage", "OrderedCollectionPage"],
	do: {:ok, false}

	defp collection_private(%{"first" => first}) do
	with {:ok, %{"type" => type}} when type in ["CollectionPage", "OrderedCollectionPage"] <-
	Fetcher.fetch_and_contain_remote_object_from_id(first) do
	{:ok, false}
	else
	{:error, {:ok, %{status: code}}} when code in [401, 403] -> {:ok, true}
	{:error, _} = e -> e
	e -> {:error, e}
	end
	end

	defp collection_private(_data), do: {:ok, true}

	def user_data_from_user_object(data) do
	with {:ok, data} <- MRF.filter(data),
	{:ok, data} <- object_to_user_data(data) do
	{:ok, data}
	else
	e -> {:error, e}
	end
	end

	- def fetch_and_prepare_user_from_ap_id(ap_id) do
	- with {:ok, data} <- Fetcher.fetch_and_contain_remote_object_from_id(ap_id),
	+ def fetch_and_prepare_user_from_ap_id(ap_id, opts \\ []) do
	+ with {:ok, data} <- Fetcher.fetch_and_contain_remote_object_from_id(ap_id, opts),
	{:ok, data} <- user_data_from_user_object(data),
	data <- maybe_update_follow_information(data) do
	{:ok, data}
	else
	{:error, "Object has been deleted"} = e ->
	Logger.debug("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
	{:error, e}

	e ->
	Logger.error("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
	{:error, e}
	end
	end

	- def make_user_from_ap_id(ap_id) do
	+ def make_user_from_ap_id(ap_id, opts \\ []) do
	if _user = User.get_cached_by_ap_id(ap_id) do
	Transmogrifier.upgrade_user_from_ap_id(ap_id)
	else
	- with {:ok, data} <- fetch_and_prepare_user_from_ap_id(ap_id) do
	+ with {:ok, data} <- fetch_and_prepare_user_from_ap_id(ap_id, opts) do
	User.insert_or_update_user(data)
	else
	e -> {:error, e}
	end
	end
	end

	def make_user_from_nickname(nickname) do
	with {:ok, %{"ap_id" => ap_id}} when not is_nil(ap_id) <- WebFinger.finger(nickname) do
	make_user_from_ap_id(ap_id)
	else
	_e -> {:error, "No AP id in WebFinger"}
	end
	end

	# filter out broken threads
	def contain_broken_threads(%Activity{} = activity, %User{} = user) do
	entire_thread_visible_for_user?(activity, user)
	end

	# do post-processing on a specific activity
	def contain_activity(%Activity{} = activity, %User{} = user) do
	contain_broken_threads(activity, user)
	end

	def fetch_direct_messages_query do
	Activity
	\|> restrict_type(%{"type" => "Create"})
	\|> restrict_visibility(%{visibility: "direct"})
	\|> order_by([activity], asc: activity.id)
	end
	end
	diff --git a/lib/pleroma/web/activity_pub/publisher.ex b/lib/pleroma/web/activity_pub/publisher.ex
	index b8ec67d7c..1c6a1728d 100644
	--- a/lib/pleroma/web/activity_pub/publisher.ex
	+++ b/lib/pleroma/web/activity_pub/publisher.ex
	@@ -1,287 +1,289 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Web.ActivityPub.Publisher do
	alias Pleroma.Activity
	alias Pleroma.Config
	alias Pleroma.Delivery
	alias Pleroma.HTTP
	alias Pleroma.Instances
	alias Pleroma.Object
	alias Pleroma.Repo
	alias Pleroma.User
	alias Pleroma.Web.ActivityPub.Relay
	alias Pleroma.Web.ActivityPub.Transmogrifier
	alias Pleroma.Web.FedSockets

	require Pleroma.Constants

	import Pleroma.Web.ActivityPub.Visibility

	@behaviour Pleroma.Web.Federator.Publisher

	require Logger

	@moduledoc """
	ActivityPub outgoing federation module.
	"""

	@doc """
	Determine if an activity can be represented by running it through Transmogrifier.
	"""
	def is_representable?(%Activity{} = activity) do
	with {:ok, _data} <- Transmogrifier.prepare_outgoing(activity.data) do
	true
	else
	_e ->
	false
	end
	end

	@doc """
	Publish a single message to a peer. Takes a struct with the following
	parameters set:

	* `inbox`: the inbox to publish to
	* `json`: the JSON message body representing the ActivityPub message
	* `actor`: the actor which is signing the message
	* `id`: the ActivityStreams URI of the message
	"""
	def publish_one(%{inbox: inbox, json: json, actor: %User{} = actor, id: id} = params) do
	Logger.debug("Federating #{id} to #{inbox}")

	case FedSockets.get_or_create_fed_socket(inbox) do
	{:ok, fedsocket} ->
	+ IO.inspect(inbox, label: "#{inspect(self())} - publishing via fedsockets")
	FedSockets.publish(fedsocket, json)

	_ ->
	+ IO.inspect(inbox, label: "#{inspect(self())} - publishing via http")
	http_publish(inbox, actor, json, params)
	end
	end

	def publish_one(%{actor_id: actor_id} = params) do
	actor = User.get_cached_by_id(actor_id)

	params
	\|> Map.delete(:actor_id)
	\|> Map.put(:actor, actor)
	\|> publish_one()
	end

	defp http_publish(inbox, actor, json, params) do
	%{host: host, path: path} = URI.parse(inbox)
	digest = "SHA-256=" <> (:crypto.hash(:sha256, json) \|> Base.encode64())

	date = Pleroma.Signature.signed_date()

	signature =
	Pleroma.Signature.sign(actor, %{
	"(request-target)": "post #{path}",
	host: host,
	"content-length": byte_size(json),
	digest: digest,
	date: date
	})

	with {:ok, %{status: code}} when code in 200..299 <-
	result =
	HTTP.post(
	inbox,
	json,
	[
	{"Content-Type", "application/activity+json"},
	{"Date", date},
	{"signature", signature},
	{"digest", digest}
	]
	) do
	if !Map.has_key?(params, :unreachable_since) \|\| params[:unreachable_since],
	do: Instances.set_reachable(inbox)

	result
	else
	{_post_result, response} ->
	unless params[:unreachable_since], do: Instances.set_unreachable(inbox)
	{:error, response}
	end
	end

	defp should_federate?(inbox, public) do
	if public do
	true
	else
	%{host: host} = URI.parse(inbox)

	quarantined_instances =
	Config.get([:instance, :quarantined_instances], [])
	\|> Pleroma.Web.ActivityPub.MRF.subdomains_regex()

	!Pleroma.Web.ActivityPub.MRF.subdomain_match?(quarantined_instances, host)
	end
	end

	@spec recipients(User.t(), Activity.t()) :: list(User.t()) \| []
	defp recipients(actor, activity) do
	followers =
	if actor.follower_address in activity.recipients do
	User.get_external_followers(actor)
	else
	[]
	end

	fetchers =
	with %Activity{data: %{"type" => "Delete"}} <- activity,
	%Object{id: object_id} <- Object.normalize(activity),
	fetchers <- User.get_delivered_users_by_object_id(object_id),
	_ <- Delivery.delete_all_by_object_id(object_id) do
	fetchers
	else
	_ ->
	[]
	end

	Pleroma.Web.Federator.Publisher.remote_users(actor, activity) ++ followers ++ fetchers
	end

	defp get_cc_ap_ids(ap_id, recipients) do
	host = Map.get(URI.parse(ap_id), :host)

	recipients
	\|> Enum.filter(fn %User{ap_id: ap_id} -> Map.get(URI.parse(ap_id), :host) == host end)
	\|> Enum.map(& &1.ap_id)
	end

	defp maybe_use_sharedinbox(%User{source_data: data}),
	do: (is_map(data["endpoints"]) && Map.get(data["endpoints"], "sharedInbox")) \|\| data["inbox"]

	@doc """
	Determine a user inbox to use based on heuristics. These heuristics
	are based on an approximation of the ``sharedInbox`` rules in the
	[ActivityPub specification][ap-sharedinbox].

	Please do not edit this function (or its children) without reading
	the spec, as editing the code is likely to introduce some breakage
	without some familiarity.

	[ap-sharedinbox]: https://www.w3.org/TR/activitypub/#shared-inbox-delivery
	"""
	def determine_inbox(
	%Activity{data: activity_data},
	%User{source_data: data} = user
	) do
	to = activity_data["to"] \|\| []
	cc = activity_data["cc"] \|\| []
	type = activity_data["type"]

	cond do
	type == "Delete" ->
	maybe_use_sharedinbox(user)

	Pleroma.Constants.as_public() in to \|\| Pleroma.Constants.as_public() in cc ->
	maybe_use_sharedinbox(user)

	length(to) + length(cc) > 1 ->
	maybe_use_sharedinbox(user)

	true ->
	data["inbox"]
	end
	end

	@doc """
	Publishes an activity with BCC to all relevant peers.
	"""

	def publish(%User{} = actor, %{data: %{"bcc" => bcc}} = activity)
	when is_list(bcc) and bcc != [] do
	public = is_public?(activity)
	{:ok, data} = Transmogrifier.prepare_outgoing(activity.data)

	recipients = recipients(actor, activity)

	inboxes =
	recipients
	\|> Enum.filter(&User.ap_enabled?/1)
	\|> Enum.map(fn %{source_data: data} -> data["inbox"] end)
	\|> Enum.filter(fn inbox -> should_federate?(inbox, public) end)
	\|> Instances.filter_reachable()

	Repo.checkout(fn ->
	Enum.each(inboxes, fn {inbox, unreachable_since} ->
	%User{ap_id: ap_id} =
	Enum.find(recipients, fn %{source_data: data} -> data["inbox"] == inbox end)

	# Get all the recipients on the same host and add them to cc. Otherwise, a remote
	# instance would only accept a first message for the first recipient and ignore the rest.
	cc = get_cc_ap_ids(ap_id, recipients)

	json =
	data
	\|> Map.put("cc", cc)
	\|> Jason.encode!()

	Pleroma.Web.Federator.Publisher.enqueue_one(__MODULE__, %{
	inbox: inbox,
	json: json,
	actor_id: actor.id,
	id: activity.data["id"],
	unreachable_since: unreachable_since
	})
	end)
	end)
	end

	@doc """
	Publishes an activity to all relevant peers.
	"""
	def publish(%User{} = actor, %Activity{} = activity) do
	public = is_public?(activity)

	if public && Config.get([:instance, :allow_relay]) do
	Logger.debug(fn -> "Relaying #{activity.data["id"]} out" end)
	Relay.publish(activity)
	end

	{:ok, data} = Transmogrifier.prepare_outgoing(activity.data)
	json = Jason.encode!(data)

	recipients(actor, activity)
	\|> Enum.filter(fn user -> User.ap_enabled?(user) end)
	\|> Enum.map(fn %User{} = user ->
	determine_inbox(activity, user)
	end)
	\|> Enum.uniq()
	\|> Enum.filter(fn inbox -> should_federate?(inbox, public) end)
	\|> Instances.filter_reachable()
	\|> Enum.each(fn {inbox, unreachable_since} ->
	Pleroma.Web.Federator.Publisher.enqueue_one(
	__MODULE__,
	%{
	inbox: inbox,
	json: json,
	actor_id: actor.id,
	id: activity.data["id"],
	unreachable_since: unreachable_since
	}
	)
	end)
	end

	def gather_webfinger_links(%User{} = user) do
	[
	%{"rel" => "self", "type" => "application/activity+json", "href" => user.ap_id},
	%{
	"rel" => "self",
	"type" => "application/ld+json; profile=\"https://www.w3.org/ns/activitystreams\"",
	"href" => user.ap_id
	},
	%{
	"rel" => "http://ostatus.org/schema/1.0/subscribe",
	"template" => "#{Pleroma.Web.base_url()}/ostatus_subscribe?acct={uri}"
	}
	]
	end

	def gather_nodeinfo_protocol_names, do: ["activitypub"]
	end
	diff --git a/lib/pleroma/web/activity_pub/transmogrifier.ex b/lib/pleroma/web/activity_pub/transmogrifier.ex
	index 9cd3de705..dfefa858b 100644
	--- a/lib/pleroma/web/activity_pub/transmogrifier.ex
	+++ b/lib/pleroma/web/activity_pub/transmogrifier.ex
	@@ -1,1247 +1,1247 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Web.ActivityPub.Transmogrifier do
	@moduledoc """
	A module to handle coding from internal to wire ActivityPub and back.
	"""
	alias Pleroma.Activity
	alias Pleroma.FollowingRelationship
	alias Pleroma.Object
	alias Pleroma.Object.Containment
	alias Pleroma.Repo
	alias Pleroma.User
	alias Pleroma.Web.ActivityPub.ActivityPub
	alias Pleroma.Web.ActivityPub.Utils
	alias Pleroma.Web.ActivityPub.Visibility
	alias Pleroma.Web.Federator
	alias Pleroma.Workers.TransmogrifierWorker

	import Ecto.Query

	require Logger
	require Pleroma.Constants

	@doc """
	Modifies an incoming AP object (mastodon format) to our internal format.
	"""
	def fix_object(object, options \\ []) do
	object
	\|> strip_internal_fields
	\|> fix_actor
	\|> fix_url
	\|> fix_attachments
	\|> fix_context
	\|> fix_in_reply_to(options)
	\|> fix_emoji
	\|> fix_tag
	\|> fix_content_map
	\|> fix_addressing
	\|> fix_summary
	\|> fix_type(options)
	end

	def fix_summary(%{"summary" => nil} = object) do
	Map.put(object, "summary", "")
	end

	def fix_summary(%{"summary" => _} = object) do
	# summary is present, nothing to do
	object
	end

	def fix_summary(object), do: Map.put(object, "summary", "")

	def fix_addressing_list(map, field) do
	cond do
	is_binary(map[field]) ->
	Map.put(map, field, [map[field]])

	is_nil(map[field]) ->
	Map.put(map, field, [])

	true ->
	map
	end
	end

	def fix_explicit_addressing(
	%{"to" => to, "cc" => cc} = object,
	explicit_mentions,
	follower_collection
	) do
	explicit_to = Enum.filter(to, fn x -> x in explicit_mentions end)

	explicit_cc = Enum.filter(to, fn x -> x not in explicit_mentions end)

	final_cc =
	(cc ++ explicit_cc)
	\|> Enum.reject(fn x -> String.ends_with?(x, "/followers") and x != follower_collection end)
	\|> Enum.uniq()

	object
	\|> Map.put("to", explicit_to)
	\|> Map.put("cc", final_cc)
	end

	def fix_explicit_addressing(object, _explicit_mentions, _followers_collection), do: object

	# if directMessage flag is set to true, leave the addressing alone
	def fix_explicit_addressing(%{"directMessage" => true} = object), do: object

	def fix_explicit_addressing(object) do
	explicit_mentions = Utils.determine_explicit_mentions(object)

	%User{follower_address: follower_collection} =
	object
	\|> Containment.get_actor()
	\|> User.get_cached_by_ap_id()

	explicit_mentions =
	explicit_mentions ++
	[
	Pleroma.Constants.as_public(),
	follower_collection
	]

	fix_explicit_addressing(object, explicit_mentions, follower_collection)
	end

	# if as:Public is addressed, then make sure the followers collection is also addressed
	# so that the activities will be delivered to local users.
	def fix_implicit_addressing(%{"to" => to, "cc" => cc} = object, followers_collection) do
	recipients = to ++ cc

	if followers_collection not in recipients do
	cond do
	Pleroma.Constants.as_public() in cc ->
	to = to ++ [followers_collection]
	Map.put(object, "to", to)

	Pleroma.Constants.as_public() in to ->
	cc = cc ++ [followers_collection]
	Map.put(object, "cc", cc)

	true ->
	object
	end
	else
	object
	end
	end

	def fix_implicit_addressing(object, _), do: object

	def fix_addressing(object) do
	{:ok, %User{} = user} = User.get_or_fetch_by_ap_id(object["actor"])
	followers_collection = User.ap_followers(user)

	object
	\|> fix_addressing_list("to")
	\|> fix_addressing_list("cc")
	\|> fix_addressing_list("bto")
	\|> fix_addressing_list("bcc")
	\|> fix_explicit_addressing()
	\|> fix_implicit_addressing(followers_collection)
	end

	def fix_actor(%{"attributedTo" => actor} = object) do
	Map.put(object, "actor", Containment.get_actor(%{"actor" => actor}))
	end

	def fix_in_reply_to(object, options \\ [])

	def fix_in_reply_to(%{"inReplyTo" => in_reply_to} = object, options)
	when not is_nil(in_reply_to) do
	in_reply_to_id = prepare_in_reply_to(in_reply_to)
	object = Map.put(object, "inReplyToAtomUri", in_reply_to_id)
	depth = (options[:depth] \|\| 0) + 1

	if Federator.allowed_thread_distance?(depth) do
	with {:ok, replied_object} <- get_obj_helper(in_reply_to_id, options),
	%Activity{} <- Activity.get_create_by_object_ap_id(replied_object.data["id"]) do
	object
	\|> Map.put("inReplyTo", replied_object.data["id"])
	\|> Map.put("inReplyToAtomUri", object["inReplyToAtomUri"] \|\| in_reply_to_id)
	\|> Map.put("conversation", replied_object.data["context"] \|\| object["conversation"])
	\|> Map.put("context", replied_object.data["context"] \|\| object["conversation"])
	else
	e ->
	Logger.error("Couldn't fetch #{inspect(in_reply_to_id)}, error: #{inspect(e)}")
	object
	end
	else
	object
	end
	end

	def fix_in_reply_to(object, _options), do: object

	defp prepare_in_reply_to(in_reply_to) do
	cond do
	is_bitstring(in_reply_to) ->
	in_reply_to

	is_map(in_reply_to) && is_bitstring(in_reply_to["id"]) ->
	in_reply_to["id"]

	is_list(in_reply_to) && is_bitstring(Enum.at(in_reply_to, 0)) ->
	Enum.at(in_reply_to, 0)

	true ->
	""
	end
	end

	def fix_context(object) do
	context = object["context"] \|\| object["conversation"] \|\| Utils.generate_context_id()

	object
	\|> Map.put("context", context)
	\|> Map.put("conversation", context)
	end

	def fix_attachments(%{"attachment" => attachment} = object) when is_list(attachment) do
	attachments =
	Enum.map(attachment, fn data ->
	media_type = data["mediaType"] \|\| data["mimeType"]
	href = data["url"] \|\| data["href"]
	url = [%{"type" => "Link", "mediaType" => media_type, "href" => href}]

	data
	\|> Map.put("mediaType", media_type)
	\|> Map.put("url", url)
	end)

	Map.put(object, "attachment", attachments)
	end

	def fix_attachments(%{"attachment" => attachment} = object) when is_map(attachment) do
	object
	\|> Map.put("attachment", [attachment])
	\|> fix_attachments()
	end

	def fix_attachments(object), do: object

	def fix_url(%{"url" => url} = object) when is_map(url) do
	Map.put(object, "url", url["href"])
	end

	def fix_url(%{"type" => "Video", "url" => url} = object) when is_list(url) do
	first_element = Enum.at(url, 0)

	link_element = Enum.find(url, fn x -> is_map(x) and x["mimeType"] == "text/html" end)

	object
	\|> Map.put("attachment", [first_element])
	\|> Map.put("url", link_element["href"])
	end

	def fix_url(%{"type" => object_type, "url" => url} = object)
	when object_type != "Video" and is_list(url) do
	first_element = Enum.at(url, 0)

	url_string =
	cond do
	is_bitstring(first_element) -> first_element
	is_map(first_element) -> first_element["href"] \|\| ""
	true -> ""
	end

	Map.put(object, "url", url_string)
	end

	def fix_url(object), do: object

	def fix_emoji(%{"tag" => tags} = object) when is_list(tags) do
	emoji =
	tags
	\|> Enum.filter(fn data -> data["type"] == "Emoji" and data["icon"] end)
	\|> Enum.reduce(%{}, fn data, mapping ->
	name = String.trim(data["name"], ":")

	Map.put(mapping, name, data["icon"]["url"])
	end)

	# we merge mastodon and pleroma emoji into a single mapping, to allow for both wire formats
	emoji = Map.merge(object["emoji"] \|\| %{}, emoji)

	Map.put(object, "emoji", emoji)
	end

	def fix_emoji(%{"tag" => %{"type" => "Emoji"} = tag} = object) do
	name = String.trim(tag["name"], ":")
	emoji = %{name => tag["icon"]["url"]}

	Map.put(object, "emoji", emoji)
	end

	def fix_emoji(object), do: object

	def fix_tag(%{"tag" => tag} = object) when is_list(tag) do
	tags =
	tag
	\|> Enum.filter(fn data -> data["type"] == "Hashtag" and data["name"] end)
	\|> Enum.map(fn data -> String.slice(data["name"], 1..-1) end)

	Map.put(object, "tag", tag ++ tags)
	end

	def fix_tag(%{"tag" => %{"type" => "Hashtag", "name" => hashtag} = tag} = object) do
	combined = [tag, String.slice(hashtag, 1..-1)]

	Map.put(object, "tag", combined)
	end

	def fix_tag(%{"tag" => %{} = tag} = object), do: Map.put(object, "tag", [tag])

	def fix_tag(object), do: object

	# content map usually only has one language so this will do for now.
	def fix_content_map(%{"contentMap" => content_map} = object) do
	content_groups = Map.to_list(content_map)
	{_, content} = Enum.at(content_groups, 0)

	Map.put(object, "content", content)
	end

	def fix_content_map(object), do: object

	def fix_type(object, options \\ [])

	def fix_type(%{"inReplyTo" => reply_id, "name" => _} = object, options)
	when is_binary(reply_id) do
	with true <- Federator.allowed_thread_distance?(options[:depth]),
	{:ok, %{data: %{"type" => "Question"} = _} = _} <- get_obj_helper(reply_id, options) do
	Map.put(object, "type", "Answer")
	else
	_ -> object
	end
	end

	def fix_type(object, _), do: object

	defp mastodon_follow_hack(%{"id" => id, "actor" => follower_id}, followed) do
	with true <- id =~ "follows",
	%User{local: true} = follower <- User.get_cached_by_ap_id(follower_id),
	%Activity{} = activity <- Utils.fetch_latest_follow(follower, followed) do
	{:ok, activity}
	else
	_ -> {:error, nil}
	end
	end

	defp mastodon_follow_hack(_, _), do: {:error, nil}

	defp get_follow_activity(follow_object, followed) do
	with object_id when not is_nil(object_id) <- Utils.get_ap_id(follow_object),
	{_, %Activity{} = activity} <- {:activity, Activity.get_by_ap_id(object_id)} do
	{:ok, activity}
	else
	# Can't find the activity. This might a Mastodon 2.3 "Accept"
	{:activity, nil} ->
	mastodon_follow_hack(follow_object, followed)

	_ ->
	{:error, nil}
	end
	end

	# Reduce the object list to find the reported user.
	defp get_reported(objects) do
	Enum.reduce_while(objects, nil, fn ap_id, _ ->
	with %User{} = user <- User.get_cached_by_ap_id(ap_id) do
	{:halt, user}
	else
	_ -> {:cont, nil}
	end
	end)
	end

	def handle_incoming(data, options \\ [])

	# Flag objects are placed ahead of the ID check because Mastodon 2.8 and earlier send them
	# with nil ID.
	def handle_incoming(%{"type" => "Flag", "object" => objects, "actor" => actor} = data, _options) do
	with context <- data["context"] \|\| Utils.generate_context_id(),
	content <- data["content"] \|\| "",
	%User{} = actor <- User.get_cached_by_ap_id(actor),
	# Reduce the object list to find the reported user.
	%User{} = account <- get_reported(objects),
	# Remove the reported user from the object list.
	statuses <- Enum.filter(objects, fn ap_id -> ap_id != account.ap_id end) do
	%{
	actor: actor,
	context: context,
	account: account,
	statuses: statuses,
	content: content,
	additional: %{"cc" => [account.ap_id]}
	}
	\|> ActivityPub.flag()
	end
	end

	# disallow objects with bogus IDs
	def handle_incoming(%{"id" => nil}, _options), do: :error
	def handle_incoming(%{"id" => ""}, _options), do: :error
	# length of https:// = 8, should validate better, but good enough for now.
	def handle_incoming(%{"id" => id}, _options) when is_binary(id) and byte_size(id) < 8,
	do: :error

	# TODO: validate those with a Ecto scheme
	# - tags
	# - emoji
	def handle_incoming(
	%{"type" => "Create", "object" => %{"type" => objtype} = object} = data,
	options
	)
	when objtype in ["Article", "Event", "Note", "Video", "Page", "Question", "Answer"] do
	actor = Containment.get_actor(data)

	data =
	Map.put(data, "actor", actor)
	\|> fix_addressing

	with nil <- Activity.get_create_by_object_ap_id(object["id"]),
	{:ok, %User{} = user} <- User.get_or_fetch_by_ap_id(data["actor"]) do
	object = fix_object(object, options)

	params = %{
	to: data["to"],
	object: object,
	actor: user,
	context: object["conversation"],
	local: false,
	published: data["published"],
	additional:
	Map.take(data, [
	"cc",
	"directMessage",
	"id"
	])
	}

	with {:ok, created_activity} <- ActivityPub.create(params) do
	reply_depth = (options[:depth] \|\| 0) + 1

	if Federator.allowed_thread_distance?(reply_depth) do
	for reply_id <- replies(object) do
	Pleroma.Workers.RemoteFetcherWorker.enqueue("fetch_remote", %{
	"id" => reply_id,
	"depth" => reply_depth
	})
	end
	end

	{:ok, created_activity}
	end
	else
	%Activity{} = activity -> {:ok, activity}
	_e -> :error
	end
	end

	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

	def handle_incoming(
	%{"type" => "Follow", "object" => followed, "actor" => follower, "id" => id} = data,
	_options
	) do
	with %User{local: true} = followed <-
	User.get_cached_by_ap_id(Containment.get_actor(%{"actor" => followed})),
	{:ok, %User{} = follower} <-
	User.get_or_fetch_by_ap_id(Containment.get_actor(%{"actor" => follower})),
	{:ok, activity} <- ActivityPub.follow(follower, followed, id, false) do
	with deny_follow_blocked <- Pleroma.Config.get([:user, :deny_follow_blocked]),
	{_, false} <- {:user_blocked, User.blocks?(followed, follower) && deny_follow_blocked},
	{_, false} <- {:user_locked, User.locked?(followed)},
	{_, {:ok, follower}} <- {:follow, User.follow(follower, followed)},
	{_, {:ok, _}} <-
	{:follow_state_update, Utils.update_follow_state_for_all(activity, "accept")},
	{:ok, _relationship} <- FollowingRelationship.update(follower, followed, "accept") do
	ActivityPub.accept(%{
	to: [follower.ap_id],
	actor: followed,
	object: data,
	local: true
	})
	else
	{:user_blocked, true} ->
	{:ok, _} = Utils.update_follow_state_for_all(activity, "reject")
	{:ok, _relationship} = FollowingRelationship.update(follower, followed, "reject")

	ActivityPub.reject(%{
	to: [follower.ap_id],
	actor: followed,
	object: data,
	local: true
	})

	{:follow, {:error, _}} ->
	{:ok, _} = Utils.update_follow_state_for_all(activity, "reject")
	{:ok, _relationship} = FollowingRelationship.update(follower, followed, "reject")

	ActivityPub.reject(%{
	to: [follower.ap_id],
	actor: followed,
	object: data,
	local: true
	})

	{:user_locked, true} ->
	{:ok, _relationship} = FollowingRelationship.update(follower, followed, "pending")
	:noop
	end

	{:ok, activity}
	else
	_e ->
	:error
	end
	end

	def handle_incoming(
	%{"type" => "Accept", "object" => follow_object, "actor" => _actor, "id" => id} = data,
	_options
	) do
	with actor <- Containment.get_actor(data),
	{:ok, %User{} = followed} <- User.get_or_fetch_by_ap_id(actor),
	{:ok, follow_activity} <- get_follow_activity(follow_object, followed),
	{:ok, follow_activity} <- Utils.update_follow_state_for_all(follow_activity, "accept"),
	%User{local: true} = follower <- User.get_cached_by_ap_id(follow_activity.data["actor"]),
	{:ok, _relationship} <- FollowingRelationship.update(follower, followed, "accept") do
	ActivityPub.accept(%{
	to: follow_activity.data["to"],
	type: "Accept",
	actor: followed,
	object: follow_activity.data["id"],
	local: false,
	activity_id: id
	})
	else
	_e -> :error
	end
	end

	def handle_incoming(
	%{"type" => "Reject", "object" => follow_object, "actor" => _actor, "id" => id} = data,
	_options
	) do
	with actor <- Containment.get_actor(data),
	{:ok, %User{} = followed} <- User.get_or_fetch_by_ap_id(actor),
	{:ok, follow_activity} <- get_follow_activity(follow_object, followed),
	{:ok, follow_activity} <- Utils.update_follow_state_for_all(follow_activity, "reject"),
	%User{local: true} = follower <- User.get_cached_by_ap_id(follow_activity.data["actor"]),
	{:ok, _relationship} <- FollowingRelationship.update(follower, followed, "reject"),
	{:ok, activity} <-
	ActivityPub.reject(%{
	to: follow_activity.data["to"],
	type: "Reject",
	actor: followed,
	object: follow_activity.data["id"],
	local: false,
	activity_id: id
	}) do
	{:ok, activity}
	else
	_e -> :error
	end
	end

	@misskey_reactions %{
	"like" => "👍",
	"love" => "❤️",
	"laugh" => "😆",
	"hmm" => "🤔",
	"surprise" => "😮",
	"congrats" => "🎉",
	"angry" => "💢",
	"confused" => "😥",
	"rip" => "😇",
	"pudding" => "🍮",
	"star" => "⭐"
	}

	@doc "Rewrite misskey likes into EmojiReacts"
	def handle_incoming(
	%{
	"type" => "Like",
	"_misskey_reaction" => reaction
	} = data,
	options
	) do
	data
	\|> Map.put("type", "EmojiReact")
	\|> Map.put("content", @misskey_reactions[reaction] \|\| reaction)
	\|> handle_incoming(options)
	end

	def handle_incoming(
	%{"type" => "Like", "object" => object_id, "actor" => _actor, "id" => id} = data,
	_options
	) do
	with actor <- Containment.get_actor(data),
	{:ok, %User{} = actor} <- User.get_or_fetch_by_ap_id(actor),
	{:ok, object} <- get_obj_helper(object_id),
	{:ok, activity, _object} <- ActivityPub.like(actor, object, id, false) do
	{:ok, activity}
	else
	_e -> :error
	end
	end

	def handle_incoming(
	%{
	"type" => "EmojiReact",
	"object" => object_id,
	"actor" => _actor,
	"id" => id,
	"content" => emoji
	} = data,
	_options
	) do
	with actor <- Containment.get_actor(data),
	{:ok, %User{} = actor} <- User.get_or_fetch_by_ap_id(actor),
	{:ok, object} <- get_obj_helper(object_id),
	{:ok, activity, _object} <-
	ActivityPub.react_with_emoji(actor, object, emoji, activity_id: id, local: false) do
	{:ok, activity}
	else
	_e -> :error
	end
	end

	def handle_incoming(
	%{"type" => "Announce", "object" => object_id, "actor" => _actor, "id" => id} = data,
	_options
	) do
	with actor <- Containment.get_actor(data),
	{:ok, %User{} = actor} <- User.get_or_fetch_by_ap_id(actor),
	{:ok, object} <- get_embedded_obj_helper(object_id, actor),
	public <- Visibility.is_public?(data),
	{:ok, activity, _object} <- ActivityPub.announce(actor, object, id, false, public) do
	{:ok, activity}
	else
	_e -> :error
	end
	end

	def handle_incoming(
	%{"type" => "Update", "object" => %{"type" => object_type} = object, "actor" => actor_id} =
	data,
	_options
	)
	when object_type in [
	"Person",
	"Application",
	"Service",
	"Organization"
	] do
	with %User{ap_id: ^actor_id} = actor <- User.get_cached_by_ap_id(object["id"]) do
	{:ok, new_user_data} = ActivityPub.user_data_from_user_object(object)

	actor
	\|> User.upgrade_changeset(new_user_data, true)
	\|> User.update_and_set_cache()

	ActivityPub.update(%{
	local: false,
	to: data["to"] \|\| [],
	cc: data["cc"] \|\| [],
	object: object,
	actor: actor_id,
	activity_id: data["id"]
	})
	else
	e ->
	Logger.error(e)
	:error
	end
	end

	# TODO: We presently assume that any actor on the same origin domain as the object being
	# deleted has the rights to delete that object. A better way to validate whether or not
	# the object should be deleted is to refetch the object URI, which should return either
	# an error or a tombstone. This would allow us to verify that a deletion actually took
	# place.
	def handle_incoming(
	%{"type" => "Delete", "object" => object_id, "actor" => actor, "id" => id} = data,
	_options
	) do
	object_id = Utils.get_ap_id(object_id)

	with actor <- Containment.get_actor(data),
	{:ok, %User{} = actor} <- User.get_or_fetch_by_ap_id(actor),
	{:ok, object} <- get_obj_helper(object_id),
	:ok <- Containment.contain_origin(actor.ap_id, object.data),
	{:ok, activity} <-
	ActivityPub.delete(object, local: false, activity_id: id, actor: actor.ap_id) do
	{:ok, activity}
	else
	nil ->
	case User.get_cached_by_ap_id(object_id) do
	%User{ap_id: ^actor} = user ->
	User.delete(user)

	nil ->
	:error
	end

	_e ->
	:error
	end
	end

	def handle_incoming(
	%{
	"type" => "Undo",
	"object" => %{"type" => "Announce", "object" => object_id},
	"actor" => _actor,
	"id" => id
	} = data,
	_options
	) do
	with actor <- Containment.get_actor(data),
	{:ok, %User{} = actor} <- User.get_or_fetch_by_ap_id(actor),
	{:ok, object} <- get_obj_helper(object_id),
	{:ok, activity, _} <- ActivityPub.unannounce(actor, object, id, false) do
	{:ok, activity}
	else
	_e -> :error
	end
	end

	def handle_incoming(
	%{
	"type" => "Undo",
	"object" => %{"type" => "Follow", "object" => followed},
	"actor" => follower,
	"id" => id
	} = _data,
	_options
	) do
	with %User{local: true} = followed <- User.get_cached_by_ap_id(followed),
	{:ok, %User{} = follower} <- User.get_or_fetch_by_ap_id(follower),
	{:ok, activity} <- ActivityPub.unfollow(follower, followed, id, false) do
	User.unfollow(follower, followed)
	{:ok, activity}
	else
	_e -> :error
	end
	end

	def handle_incoming(
	%{
	"type" => "Undo",
	"object" => %{"type" => "EmojiReact", "id" => reaction_activity_id},
	"actor" => _actor,
	"id" => id
	} = data,
	_options
	) do
	with actor <- Containment.get_actor(data),
	{:ok, %User{} = actor} <- User.get_or_fetch_by_ap_id(actor),
	{:ok, activity, _} <-
	ActivityPub.unreact_with_emoji(actor, reaction_activity_id,
	activity_id: id,
	local: false
	) do
	{:ok, activity}
	else
	_e -> :error
	end
	end

	def handle_incoming(
	%{
	"type" => "Undo",
	"object" => %{"type" => "Block", "object" => blocked},
	"actor" => blocker,
	"id" => id
	} = _data,
	_options
	) do
	with %User{local: true} = blocked <- User.get_cached_by_ap_id(blocked),
	{:ok, %User{} = blocker} <- User.get_or_fetch_by_ap_id(blocker),
	{:ok, activity} <- ActivityPub.unblock(blocker, blocked, id, false) do
	User.unblock(blocker, blocked)
	{:ok, activity}
	else
	_e -> :error
	end
	end

	def handle_incoming(
	%{"type" => "Block", "object" => blocked, "actor" => blocker, "id" => id} = _data,
	_options
	) do
	with %User{local: true} = blocked = User.get_cached_by_ap_id(blocked),
	{:ok, %User{} = blocker} = User.get_or_fetch_by_ap_id(blocker),
	{:ok, activity} <- ActivityPub.block(blocker, blocked, id, false) do
	User.unfollow(blocker, blocked)
	User.block(blocker, blocked)
	{:ok, activity}
	else
	_e -> :error
	end
	end

	def handle_incoming(
	%{
	"type" => "Undo",
	"object" => %{"type" => "Like", "object" => object_id},
	"actor" => _actor,
	"id" => id
	} = data,
	_options
	) do
	with actor <- Containment.get_actor(data),
	{:ok, %User{} = actor} <- User.get_or_fetch_by_ap_id(actor),
	{:ok, object} <- get_obj_helper(object_id),
	{:ok, activity, _, _} <- ActivityPub.unlike(actor, object, id, false) do
	{:ok, activity}
	else
	_e -> :error
	end
	end

	# For Undos that don't have the complete object attached, try to find it in our database.
	def handle_incoming(
	%{
	"type" => "Undo",
	"object" => object
	} = activity,
	options
	)
	when is_binary(object) do
	with %Activity{data: data} <- Activity.get_by_ap_id(object) do
	activity
	\|> Map.put("object", data)
	\|> handle_incoming(options)
	else
	_e -> :error
	end
	end

	def handle_incoming(
	%{
	"type" => "Move",
	"actor" => origin_actor,
	"object" => origin_actor,
	"target" => target_actor
	},
	_options
	) do
	with %User{} = origin_user <- User.get_cached_by_ap_id(origin_actor),
	{:ok, %User{} = target_user} <- User.get_or_fetch_by_ap_id(target_actor),
	true <- origin_actor in target_user.also_known_as do
	ActivityPub.move(origin_user, target_user, false)
	else
	_e -> :error
	end
	end

	def handle_incoming(_, _), do: :error

	@spec get_obj_helper(String.t(), Keyword.t()) :: {:ok, Object.t()} \| nil
	def get_obj_helper(id, options \\ []) do
	case Object.normalize(id, true, options) do
	%Object{} = object -> {:ok, object}
	_ -> nil
	end
	end

	@spec get_embedded_obj_helper(String.t() \| Object.t(), User.t()) :: {:ok, Object.t()} \| nil
	def get_embedded_obj_helper(%{"attributedTo" => attributed_to, "id" => object_id} = data, %User{
	ap_id: ap_id
	})
	when attributed_to == ap_id do
	with {:ok, activity} <-
	handle_incoming(%{
	"type" => "Create",
	"to" => data["to"],
	"cc" => data["cc"],
	"actor" => attributed_to,
	"object" => data
	}) do
	{:ok, Object.normalize(activity)}
	else
	_ -> get_obj_helper(object_id)
	end
	end

	def get_embedded_obj_helper(object_id, _) do
	get_obj_helper(object_id)
	end

	def set_reply_to_uri(%{"inReplyTo" => in_reply_to} = object) when is_binary(in_reply_to) do
	with false <- String.starts_with?(in_reply_to, "http"),
	{:ok, %{data: replied_to_object}} <- get_obj_helper(in_reply_to) do
	Map.put(object, "inReplyTo", replied_to_object["external_url"] \|\| in_reply_to)
	else
	_e -> object
	end
	end

	def set_reply_to_uri(obj), do: obj

	@doc """
	Serialized Mastodon-compatible `replies` collection containing _self-replies_.
	Based on Mastodon's ActivityPub::NoteSerializer#replies.
	"""
	def set_replies(obj_data) do
	replies_uris =
	with limit when limit > 0 <-
	Pleroma.Config.get([:activitypub, :note_replies_output_limit], 0),
	%Object{} = object <- Object.get_cached_by_ap_id(obj_data["id"]) do
	object
	\|> Object.self_replies()
	\|> select([o], fragment("?->>'id'", o.data))
	\|> limit(^limit)
	\|> Repo.all()
	else
	_ -> []
	end

	set_replies(obj_data, replies_uris)
	end

	defp set_replies(obj, []) do
	obj
	end

	defp set_replies(obj, replies_uris) do
	replies_collection = %{
	"type" => "Collection",
	"items" => replies_uris
	}

	Map.merge(obj, %{"replies" => replies_collection})
	end

	def replies(%{"replies" => %{"first" => %{"items" => items}}}) when not is_nil(items) do
	items
	end

	def replies(%{"replies" => %{"items" => items}}) when not is_nil(items) do
	items
	end

	def replies(_), do: []

	# Prepares the object of an outgoing create activity.
	def prepare_object(object) do
	object
	\|> set_sensitive
	\|> add_hashtags
	\|> add_mention_tags
	\|> add_emoji_tags
	\|> add_attributed_to
	\|> prepare_attachments
	\|> set_conversation
	\|> set_reply_to_uri
	\|> set_replies
	\|> strip_internal_fields
	\|> strip_internal_tags
	\|> set_type
	end

	# @doc
	# """
	# internal -> Mastodon
	# """

	def prepare_outgoing(%{"type" => activity_type, "object" => object_id} = data)
	when activity_type in ["Create", "Listen"] do
	object =
	object_id
	\|> Object.normalize()
	\|> Map.get(:data)
	\|> 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
	\|> Object.normalize()

	data =
	if Visibility.is_private?(object) && object.data["actor"] == ap_id do
	data \|> Map.put("object", object \|> Map.get(:data) \|> prepare_object)
	else
	data \|> maybe_fix_object_url
	end

	data =
	data
	\|> strip_internal_fields
	\|> Map.merge(Utils.make_json_ld_header())
	\|> Map.delete("bcc")

	{:ok, data}
	end

	# Mastodon Accept/Reject requires a non-normalized object containing the actor URIs,
	# because of course it does.
	def prepare_outgoing(%{"type" => "Accept"} = data) do
	with follow_activity <- Activity.normalize(data["object"]) do
	object = %{
	"actor" => follow_activity.actor,
	"object" => follow_activity.data["object"],
	"id" => follow_activity.data["id"],
	"type" => "Follow"
	}

	data =
	data
	\|> Map.put("object", object)
	\|> Map.merge(Utils.make_json_ld_header())

	{:ok, data}
	end
	end

	def prepare_outgoing(%{"type" => "Reject"} = data) do
	with follow_activity <- Activity.normalize(data["object"]) do
	object = %{
	"actor" => follow_activity.actor,
	"object" => follow_activity.data["object"],
	"id" => follow_activity.data["id"],
	"type" => "Follow"
	}

	data =
	data
	\|> Map.put("object", object)
	\|> Map.merge(Utils.make_json_ld_header())

	{:ok, data}
	end
	end

	def prepare_outgoing(%{"type" => _type} = data) do
	data =
	data
	\|> strip_internal_fields
	\|> maybe_fix_object_url
	\|> Map.merge(Utils.make_json_ld_header())

	{:ok, data}
	end

	def maybe_fix_object_url(%{"object" => object} = data) when is_binary(object) do
	with false <- String.starts_with?(object, "http"),
	{:fetch, {:ok, relative_object}} <- {:fetch, get_obj_helper(object)},
	%{data: %{"external_url" => external_url}} when not is_nil(external_url) <-
	relative_object do
	Map.put(data, "object", external_url)
	else
	{:fetch, e} ->
	Logger.error("Couldn't fetch #{object} #{inspect(e)}")
	data

	_ ->
	data
	end
	end

	def maybe_fix_object_url(data), do: data

	def add_hashtags(object) do
	tags =
	(object["tag"] \|\| [])
	\|> Enum.map(fn
	# Expand internal representation tags into AS2 tags.
	tag when is_binary(tag) ->
	%{
	"href" => Pleroma.Web.Endpoint.url() <> "/tags/#{tag}",
	"name" => "##{tag}",
	"type" => "Hashtag"
	}

	# Do not process tags which are already AS2 tag objects.
	tag when is_map(tag) ->
	tag
	end)

	Map.put(object, "tag", tags)
	end

	def add_mention_tags(object) do
	mentions =
	object
	\|> Utils.get_notified_from_object()
	\|> Enum.map(&build_mention_tag/1)

	tags = object["tag"] \|\| []

	Map.put(object, "tag", tags ++ mentions)
	end

	defp build_mention_tag(%{ap_id: ap_id, nickname: nickname} = _) do
	%{"type" => "Mention", "href" => ap_id, "name" => "@#{nickname}"}
	end

	def take_emoji_tags(%User{emoji: emoji}) do
	emoji
	\|> Enum.flat_map(&Map.to_list/1)
	\|> Enum.map(&build_emoji_tag/1)
	end

	# TODO: we should probably send mtime instead of unix epoch time for updated
	def add_emoji_tags(%{"emoji" => emoji} = object) do
	tags = object["tag"] \|\| []

	out = Enum.map(emoji, &build_emoji_tag/1)

	Map.put(object, "tag", tags ++ out)
	end

	def add_emoji_tags(object), do: object

	defp build_emoji_tag({name, url}) do
	%{
	"icon" => %{"url" => url, "type" => "Image"},
	"name" => ":" <> name <> ":",
	"type" => "Emoji",
	"updated" => "1970-01-01T00:00:00Z",
	"id" => url
	}
	end

	def set_conversation(object) do
	Map.put(object, "conversation", object["context"])
	end

	def set_sensitive(object) do
	tags = object["tag"] \|\| []
	Map.put(object, "sensitive", "nsfw" in tags)
	end

	def set_type(%{"type" => "Answer"} = object) do
	Map.put(object, "type", "Note")
	end

	def set_type(object), do: object

	def add_attributed_to(object) do
	attributed_to = object["attributedTo"] \|\| object["actor"]
	Map.put(object, "attributedTo", attributed_to)
	end

	def prepare_attachments(object) do
	attachments =
	(object["attachment"] \|\| [])
	\|> Enum.map(fn data ->
	[%{"mediaType" => media_type, "href" => href} \| _] = data["url"]
	%{"url" => href, "mediaType" => media_type, "name" => data["name"], "type" => "Document"}
	end)

	Map.put(object, "attachment", attachments)
	end

	def strip_internal_fields(object) do
	object
	\|> Map.drop(Pleroma.Constants.object_internal_fields())
	end

	defp strip_internal_tags(%{"tag" => tags} = object) do
	tags = Enum.filter(tags, fn x -> is_map(x) end)

	Map.put(object, "tag", tags)
	end

	defp strip_internal_tags(object), do: object

	def perform(:user_upgrade, user) do
	# we pass a fake user so that the followers collection is stripped away
	old_follower_address = User.ap_followers(%User{nickname: user.nickname})

	from(
	a in Activity,
	where: ^old_follower_address in a.recipients,
	update: [
	set: [
	recipients:
	fragment(
	"array_replace(?,?,?)",
	a.recipients,
	^old_follower_address,
	^user.follower_address
	)
	]
	]
	)
	\|> Repo.update_all([])
	end

	def upgrade_user_from_ap_id(ap_id) do
	with %User{local: false} = user <- User.get_cached_by_ap_id(ap_id),
	- {:ok, data} <- ActivityPub.fetch_and_prepare_user_from_ap_id(ap_id),
	+ {:ok, data} <- ActivityPub.fetch_and_prepare_user_from_ap_id(ap_id, force_http: true),
	already_ap <- User.ap_enabled?(user),
	{:ok, user} <- upgrade_user(user, data) do
	if not already_ap do
	TransmogrifierWorker.enqueue("user_upgrade", %{"user_id" => user.id})
	end

	{:ok, user}
	else
	%User{} = user -> {:ok, user}
	e -> e
	end
	end

	defp upgrade_user(user, data) do
	user
	\|> User.upgrade_changeset(data, true)
	\|> User.update_and_set_cache()
	end

	def maybe_fix_user_url(%{"url" => url} = data) when is_map(url) do
	Map.put(data, "url", url["href"])
	end

	def maybe_fix_user_url(data), do: data

	def maybe_fix_user_object(data), do: maybe_fix_user_url(data)
	end
	diff --git a/lib/pleroma/web/fed_sockets/fed_socket.ex b/lib/pleroma/web/fed_sockets/fed_socket.ex
	index 44eaa19a8..e0f8355df 100644
	--- a/lib/pleroma/web/fed_sockets/fed_socket.ex
	+++ b/lib/pleroma/web/fed_sockets/fed_socket.ex
	@@ -1,152 +1,153 @@
	defmodule Pleroma.Web.FedSockets.FedSocket do
	@moduledoc """
	The FedSocket module encapsulates the structure of the data packages for both the sending and receiving connections.


	Normally outside modules will have no need to call the FedSocket module directly.
	"""

	defstruct origin: nil,
	pid: nil,
	type: nil

	alias Pleroma.Object
	alias Pleroma.User
	alias Pleroma.Web.ActivityPub.ObjectView
	alias Pleroma.Web.ActivityPub.UserView
	alias Pleroma.Web.ActivityPub.Visibility
	alias Pleroma.Web.FedSockets.FedRegistry
	alias Pleroma.Web.FedSockets.FedSocket
	alias Pleroma.Web.FedSockets.FetchRegistry
	alias Pleroma.Web.FedSockets.IngesterWorker
	alias Pleroma.Web.FedSockets.OutgoingHandler
	alias Pleroma.Web.FedSockets.SocketInfo

	require Logger

	@shake "61dd18f7-f1e6-49a4-939a-a749fcdc1103"

	def connect_to_host(uri) do
	case OutgoingHandler.start_link(uri) do
	{:ok, %SocketInfo{} = socket_info} ->
	{:ok, build_fed_socket(socket_info)}

	{:error, %{original: error}} ->
	{:error, inspect(error)}

	{:error, %{message: message}} ->
	{:error, message}
	end
	end

	def connection_from_host(%SocketInfo{} = socket_info),
	do: {:ok, build_fed_socket(socket_info)}

	def close(%FedSocket{pid: socket_pid}),
	do: Process.send(socket_pid, :close, [])

	def ping(%FedSocket{pid: socket_pid}),
	do: Process.send(socket_pid, :ping, [])

	def publish(%FedSocket{pid: socket_pid}, json) do
	%{action: :publish, data: json}
	\|> Jason.encode!()
	\|> send_packet(socket_pid)
	end

	def fetch(%FedSocket{pid: socket_pid}, id) do
	fetch_uuid = FetchRegistry.register_fetch(id)

	%{action: :fetch, data: id, uuid: fetch_uuid}
	\|> Jason.encode!()
	\|> send_packet(socket_pid)

	wait_for_fetch_to_return(fetch_uuid, 0)
	end

	def receive_package(%FedSocket{} = fed_socket, json) do
	FedRegistry.touch(fed_socket)

	json
	\|> Jason.decode!()
	\|> process_package(fed_socket)
	end

	defp wait_for_fetch_to_return(uuid, cntr) do
	case FetchRegistry.check_fetch(uuid) do
	{:error, :waiting} ->
	Process.sleep(:math.pow(cntr, 3) \|> Kernel.trunc())
	wait_for_fetch_to_return(uuid, cntr + 1)

	{:error, :missing} ->
	{:error, :timeout}

	{:ok, _fr} ->
	FetchRegistry.pop_fetch(uuid)
	end
	end

	defp process_package(%{"action" => "publish", "data" => data}, _fed_socket) do
	IngesterWorker.enqueue("ingest", %{"object" => data})
	{:reply, %{"action" => "publish_reply", "status" => "processed"}}
	end

	defp process_package(%{"action" => "fetch_reply", "uuid" => uuid, "data" => data}, _fed_socket) do
	FetchRegistry.register_fetch_received(uuid, data)
	{:noreply, nil}
	end

	defp process_package(%{"action" => "fetch", "uuid" => uuid, "data" => ap_id}, _fed_socket) do
	{:ok, data} = render_fetched_data(ap_id, uuid)
	+ IO.puts("#{inspect(self())} - fetch processed via FedSockets - #{inspect(uuid)}")
	{:reply, data}
	end

	defp process_package(%{"action" => "publish_reply"}, _fed_socket) do
	{:noreply, nil}
	end

	defp process_package(other, _fed_socket) do
	Logger.warn("unknown json packages received #{inspect(other)}")
	{:noreply, nil}
	end

	defp render_fetched_data(ap_id, uuid) do
	{:ok,
	%{
	"action" => "fetch_reply",
	"status" => "processed",
	"uuid" => uuid,
	"data" => represent_item(ap_id)
	}}
	end

	defp represent_item(ap_id) do
	case User.get_by_ap_id(ap_id) do
	nil ->
	object = Object.get_cached_by_ap_id(ap_id)

	if Visibility.is_public?(object) do
	Phoenix.View.render_to_string(ObjectView, "object.json", object: object)
	else
	nil
	end

	user ->
	Phoenix.View.render_to_string(UserView, "user.json", user: user)
	end
	end

	defp build_fed_socket(%SocketInfo{origin: origin, pid: pid, type: type}) do
	%FedSocket{
	origin: origin,
	pid: pid,
	type: type
	}
	end

	defp send_packet(data, socket_pid) do
	Process.send(socket_pid, {:send, data}, [])
	end

	def shake, do: @shake
	end
	diff --git a/lib/pleroma/web/fed_sockets/fed_sockets.ex b/lib/pleroma/web/fed_sockets/fed_sockets.ex
	index 4aa02f91f..0c207feac 100644
	--- a/lib/pleroma/web/fed_sockets/fed_sockets.ex
	+++ b/lib/pleroma/web/fed_sockets/fed_sockets.ex
	@@ -1,183 +1,183 @@
	defmodule Pleroma.Web.FedSockets do
	@moduledoc """
	This documents the FedSockets framework. A framework for transferring
	ActivityPub objects between servers via persistant WebSocket connections.

	FedSockets allow servers to authenticate on first contact and maintain that
	connection, eliminating the need to authenticate every time data needs to move.

	## Protocol
	FedSockets currently support 2 types of data transfer:
	* `publish` method which doesn't require a response
	* `fetch` method requires a response be sent

	### Publish
	The publish operation sends a json encoded map of the shape:
	%{action: :publish, data: json}
	and accepts (but does not require) a reply of form:
	%{"action" => "publish_reply"}

	The outgoing params represent
	* data: ActivityPub object encoded into json


	### Fetch
	The fetch operation sends a json encoded map of the shape:
	%{action: :fetch, data: id, uuid: fetch_uuid}
	and requires a reply of form:
	%{"action" => "fetch_reply", "uuid" => uuid, "data" => data}

	The outgoing params represent
	* id: an ActivityPub object URI
	* uuid: a unique uuid generated by the sender

	The reply params represent
	* data: an ActivityPub object encoded into json
	* uuid: the uuid sent along with the fetch request

	## Examples
	Clients of FedSocket transfers shouldn't need to use any of the functions outside of this module.

	A typical publish operation can be performed through the following code, and a fetch operation in a similar manner.

	case FedSockets.get_or_create_fed_socket(inbox) do
	{:ok, fedsocket} ->
	FedSockets.publish(fedsocket, json)

	_ ->
	alternative_publish(inbox, actor, json, params)
	end

	## Configuration
	FedSockets have the following config settings

	config :pleroma, :fed_sockets,
	enabled: true,
	ping_interval: :timer.seconds(15),
	connection_duration: :timer.hours(1),
	rejection_duration: :timer.hours(1),
	fed_socket_fetches: [
	default: 12_000,
	interval: 3_000,
	lazy: false
	]
	* enabled - turn FedSockets on or off with this flag. Can be toggled at runtime.
	* ping_interval - How often messages are exchanged by FedSocket connections to avoid timeouts
	* connection_duration - How long a FedSocket can sit idle before it's culled.
	* rejection_duration - After failing to make a FedSocket connection a host will be excluded
	from further connections for this amount of time
	* fed_socket_fetches - Use this paramteres to pass options to the Caches queue backing the FetchRegistry
	* default: the minimum amount of time a fetch can wait before it times out.
	* interval: the interval between checks for timed out fetches. This plus the default represent the maximum time allowed
	* lazy: leave at false for consistant and fast lookups, set to true for stricter timeout enforcement

	"""
	require Logger

	alias Pleroma.Web.FedSockets.FedRegistry
	alias Pleroma.Web.FedSockets.FedSocket
	alias Pleroma.Web.FedSockets.SocketInfo

	@doc """
	returns a FedSocket for the given origin. Will reuse an existing one or create a new one.

	address is expected to be a fully formed URL such as:
	"http://www.example.com" or "http://www.example.com:8080"

	It can and usually does include additional path parameters,
	but these are ignored as the FedSockets are organized by host and port info alone.
	"""
	def get_or_create_fed_socket(address) do
	origin = SocketInfo.origin(address)

	with {:cache, {:error, :missing}} <- {:cache, get_fed_socket(address)},
	{:connect, {:ok, fed_socket}} <- {:connect, FedSocket.connect_to_host(origin)},
	{:register, {:ok, fed_socket}} <- {:register, FedRegistry.add_fed_socket(fed_socket)} do
	- IO.puts("FedSocket created for - #{address}")
	+ IO.puts("#{inspect(self())} - FedSocket created for - #{address}")
	{:ok, fed_socket}
	else
	{:cache, {:ok, socket}} ->
	{:ok, socket}

	{:connect, {:error, _host}} ->
	IO.puts("set host rejected #{inspect(origin)}")
	FedRegistry.set_host_rejected(origin)
	{:error, :rejected}

	{_, {:error, reason}} ->
	{:error, reason}
	end
	end

	@doc """
	returns a FedSocket for the given origin. Will not create a new FedSocket if one does not exist.

	address is expected to be a fully formed URL such as:
	"http://www.example.com" or "http://www.example.com:8080"
	"""
	def get_fed_socket(address) do
	origin = SocketInfo.origin(address)

	with {:config, true} <- {:config, Pleroma.Config.get([:fed_sockets, :enabled], false)},
	{:ok, socket} <- FedRegistry.get_fed_socket(origin) do
	- IO.puts("FedSocket retrieved for - #{origin}")
	+ IO.puts("#{inspect(self())} - FedSocket retrieved for - #{origin}")
	{:ok, socket}
	else
	{:config, _} ->
	{:error, :disabled}

	{:error, :rejected} ->
	IO.puts("FedSocket previously rejected - #{origin}")
	{:error, :rejected}

	{:error, reason} ->
	{:error, reason}
	end
	end

	@doc """
	Sends the supplied data via the publish protocol.
	It will not block waiting for a reply.
	Returns :ok but this is not an indication of a successful transfer.

	the data is expected to be JSON encoded binary data.
	"""
	def publish(%FedSocket{} = fed_socket, json) do
	FedRegistry.touch(fed_socket)
	FedSocket.publish(fed_socket, json)
	end

	@doc """
	Sends the supplied data via the fetch protocol.
	It will block waiting for a reply or timeout.

	Returns {:ok, object} where object is the requested object (or nil)
	{:error, :timeout} in the event the message was not responded to

	the id is expected to be the URI of an ActivityPub object.
	"""
	def fetch(%FedSocket{} = fed_socket, id) do
	FedRegistry.touch(fed_socket)
	FedSocket.fetch(fed_socket, id)
	end

	@doc """
	Closes the given FedSocket on both sides of the transfer.
	Since idle transfer timeout and active transfers will just reconnect
	it should not be necessary to call this under normal circumstances.
	"""
	def close(%FedSocket{} = fed_socket) do
	Logger.debug("closing socket")
	FedSocket.close(fed_socket)
	end

	@doc """
	Disconnect all and restart FedSockets.
	This is mainly used in development and testing but could be useful in production.
	"""
	def reset do
	FedRegistry
	\|> Process.whereis()
	\|> Process.exit(:testing)
	end
	end
	diff --git a/lib/pleroma/web/fed_sockets/incoming_handler.ex b/lib/pleroma/web/fed_sockets/incoming_handler.ex
	index bf3d8ae6a..535dcbfc5 100644
	--- a/lib/pleroma/web/fed_sockets/incoming_handler.ex
	+++ b/lib/pleroma/web/fed_sockets/incoming_handler.ex
	@@ -1,97 +1,98 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Web.FedSockets.IncomingHandler do
	require Logger

	alias Pleroma.Web.FedSockets.FedRegistry
	alias Pleroma.Web.FedSockets.FedSocket
	alias Pleroma.Web.FedSockets.SocketInfo

	import HTTPSignatures, only: [validate_conn: 1, split_signature: 1]

	@behaviour :cowboy_websocket
	@timeout :infinity

	def init(req, state) do
	shake = FedSocket.shake()

	with true <- Pleroma.Config.get([:fed_sockets, :enabled]),
	sec_protocol <- :cowboy_req.header("sec-websocket-protocol", req, nil),
	headers = %{"(request-target)" => ^shake} <- :cowboy_req.headers(req),
	true <- validate_conn(%{req_headers: headers}),
	%{"keyId" => origin} <- split_signature(headers["signature"]) do
	req =
	if is_nil(sec_protocol) do
	req
	else
	:cowboy_req.set_resp_header("sec-websocket-protocol", sec_protocol, req)
	end

	{:cowboy_websocket, req, %{origin: origin}, %{idle_timeout: @timeout}}
	else
	_ ->
	{:ok, req, state}
	end
	end

	def websocket_init(%{origin: origin} = state) do
	sckt = SocketInfo.incoming(self(), origin)

	with {:connect, {:ok, fed_socket}} <- {:connect, FedSocket.connection_from_host(sckt)},
	{:register, {:ok, fed_socket}} <- {:register, FedRegistry.add_fed_socket(fed_socket)} do
	+ IO.puts("#{inspect(self())} - incoming FedSocket created for - #{origin}")
	{:ok, Map.put(state, :fed_socket, fed_socket)}
	else
	{mode, {:error, e}} ->
	Logger.error("FedSocket init failed in #{mode}- #{inspect(e)}")
	{:error, inspect(e)}

	{mode, e} ->
	Logger.error("FedSocket init failed in #{mode} - #{inspect(e)}")
	{:error, inspect(e)}
	end
	end

	# Replying to ping is handled by cowboy
	def websocket_handle(:ping, state) do
	{:ok, state}
	end

	def websocket_handle(:pong, state) do
	{:ok, state}
	end

	def websocket_handle({:text, data}, %{fed_socket: fed_socket} = state) do
	case FedSocket.receive_package(fed_socket, data) do
	{:noreply, _} ->
	{:ok, state}

	{:reply, reply} ->
	{:reply, {:text, Jason.encode!(reply)}, state}

	{:error, reason} ->
	Logger.error("incoming error - receive_package: #{inspect(reason)}")
	{:ok, state}
	end
	end

	def websocket_info({:send, message}, state) do
	{:reply, {:text, message}, state}
	end

	def websocket_info(:close, state) do
	{:stop, state}
	end

	def websocket_info(message, state) do
	Logger.debug("#{__MODULE__} unknown message #{inspect(message)}")
	{:ok, state}
	end

	def terminate(reason, _req, _state) do
	Logger.debug("#{__MODULE__} terminating incoming connection for #{inspect(reason)}")

	:ok
	end
	end
	diff --git a/lib/pleroma/web/fed_sockets/ingester_worker.ex b/lib/pleroma/web/fed_sockets/ingester_worker.ex
	index 3f32bb9b4..0781aa608 100644
	--- a/lib/pleroma/web/fed_sockets/ingester_worker.ex
	+++ b/lib/pleroma/web/fed_sockets/ingester_worker.ex
	@@ -1,50 +1,59 @@
	defmodule Pleroma.Web.FedSockets.IngesterWorker do
	use Pleroma.Workers.WorkerHelper, queue: "ingestion_queue"

	alias Pleroma.Object.Fetcher
	alias Pleroma.User
	alias Pleroma.Web.ActivityPub.Utils
	alias Pleroma.Web.Federator

	@impl Oban.Worker
	def perform(%{"op" => "ingest", "object" => ingestee}, _job) do
	- ingestee
	- \|> Jason.decode!()
	- \|> do_ingestion()
	+ IO.puts("#{inspect(self())} - starting ingestion")
	+
	+ try do
	+ ingestee
	+ \|> Jason.decode!()
	+ \|> do_ingestion()
	+
	+ IO.puts("#{inspect(self())} - finished ingestion")
	+ rescue
	+ e ->
	+ IO.inspect(e, label: "#{inspect(self())} - ingestion error")
	+ end
	end

	defp do_ingestion(%{"nickname" => nickname} = params) do
	with %User{} = recipient <- User.get_cached_by_nickname(nickname),
	{:ok, %User{} = actor} <- User.get_or_fetch_by_ap_id(params["actor"]),
	true <- Utils.recipient_in_message(recipient, actor, params),
	params <- Utils.maybe_splice_recipient(recipient.ap_id, params) do
	case Federator.incoming_ap_doc(params) do
	{:error, reason} ->
	{:error, reason}

	{:ok, object} ->
	{:ok, object}
	end
	end
	end

	defp do_ingestion(%{"type" => "Create", "object" => %{"id" => object_id}}) do
	case Fetcher.fetch_object_from_id(object_id) do
	{:error, reason} ->
	{:error, reason}

	{:ok, object} ->
	{:ok, object}
	end
	end

	defp do_ingestion(params) do
	case Federator.incoming_ap_doc(params) do
	{:error, reason} ->
	{:error, reason}

	{:ok, object} ->
	{:ok, object}
	end
	end
	end
	diff --git a/lib/pleroma/web/push/impl.ex b/lib/pleroma/web/push/impl.ex
	index afa510f08..01b9bf311 100644
	--- a/lib/pleroma/web/push/impl.ex
	+++ b/lib/pleroma/web/push/impl.ex
	@@ -1,160 +1,161 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Web.Push.Impl do
	@moduledoc "The module represents implementation push web notification"

	alias Pleroma.Activity
	alias Pleroma.Notification
	alias Pleroma.Object
	alias Pleroma.Repo
	alias Pleroma.User
	alias Pleroma.Web.Metadata.Utils
	alias Pleroma.Web.Push.Subscription

	require Logger
	import Ecto.Query

	@types ["Create", "Follow", "Announce", "Like", "Move"]

	@doc "Performs sending notifications for user subscriptions"
	@spec perform(Notification.t()) :: list(any) \| :error
	def perform(
	%{
	activity: %{data: %{"type" => activity_type}} = activity,
	user: %User{id: user_id}
	} = notif
	)
	when activity_type in @types do
	actor = User.get_cached_by_ap_id(notif.activity.data["actor"])

	type = Activity.mastodon_notification_type(notif.activity)
	gcm_api_key = Application.get_env(:web_push_encryption, :gcm_api_key)
	avatar_url = User.avatar_url(actor)
	object = Object.normalize(activity)
	user = User.get_cached_by_id(user_id)
	direct_conversation_id = Activity.direct_conversation_id(activity, user)

	for subscription <- fetch_subsriptions(user_id),
	get_in(subscription.data, ["alerts", type]) do
	%{
	access_token: subscription.token.token,
	notification_id: notif.id,
	notification_type: type,
	icon: avatar_url,
	preferred_locale: "en",
	pleroma: %{
	activity_id: notif.activity.id,
	direct_conversation_id: direct_conversation_id
	}
	}
	\|> Map.merge(build_content(notif, actor, object))
	\|> Jason.encode!()
	\|> push_message(build_sub(subscription), gcm_api_key, subscription)
	end
	end

	- def perform(_) do
	+ def perform(obj) do
	+ IO.inspect(obj, label: "unknown message received")
	Logger.warn("Unknown notification type")
	:error
	end

	@doc "Push message to web"
	def push_message(body, sub, api_key, subscription) do
	case WebPushEncryption.send_web_push(body, sub, api_key) do
	{:ok, %{status_code: code}} when 400 <= code and code < 500 ->
	Logger.debug("Removing subscription record")
	Repo.delete!(subscription)
	:ok

	{:ok, %{status_code: code}} when 200 <= code and code < 300 ->
	:ok

	{:ok, %{status_code: code}} ->
	Logger.error("Web Push Notification failed with code: #{code}")
	:error

	_ ->
	Logger.error("Web Push Notification failed with unknown error")
	:error
	end
	end

	@doc "Gets user subscriptions"
	def fetch_subsriptions(user_id) do
	Subscription
	\|> where(user_id: ^user_id)
	\|> preload(:token)
	\|> Repo.all()
	end

	def build_sub(subscription) do
	%{
	keys: %{
	p256dh: subscription.key_p256dh,
	auth: subscription.key_auth
	},
	endpoint: subscription.endpoint
	}
	end

	def build_content(
	%{
	activity: %{data: %{"directMessage" => true}},
	user: %{notification_settings: %{privacy_option: true}}
	},
	actor,
	_
	) do
	%{title: "New Direct Message", body: "@#{actor.nickname}"}
	end

	def build_content(notif, actor, object) do
	%{
	title: format_title(notif),
	body: format_body(notif, actor, object)
	}
	end

	def format_body(
	%{activity: %{data: %{"type" => "Create"}}},
	actor,
	%{data: %{"content" => content}}
	) do
	"@#{actor.nickname}: #{Utils.scrub_html_and_truncate(content, 80)}"
	end

	def format_body(
	%{activity: %{data: %{"type" => "Announce"}}},
	actor,
	%{data: %{"content" => content}}
	) do
	"@#{actor.nickname} repeated: #{Utils.scrub_html_and_truncate(content, 80)}"
	end

	def format_body(
	%{activity: %{data: %{"type" => type}}},
	actor,
	_object
	)
	when type in ["Follow", "Like"] do
	case type do
	"Follow" -> "@#{actor.nickname} has followed you"
	"Like" -> "@#{actor.nickname} has favorited your post"
	end
	end

	def format_title(%{activity: %{data: %{"directMessage" => true}}}) do
	"New Direct Message"
	end

	def format_title(%{activity: %{data: %{"type" => type}}}) do
	case type do
	"Create" -> "New Mention"
	"Follow" -> "New Follower"
	"Announce" -> "New Repeat"
	"Like" -> "New Favorite"
	end
	end
	end

File Metadata

Mime Type: text/x-diff
Expires: Sat, Sep 19, 7:03 AM (1 d, 14 h)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 1769060
Default Alt Text: (190 KB)

No OneTemporaryActions

View Options

File Metadata

Event Timeline

No OneTemporary
Actions