1 # Pleroma: A lightweight social networking server
2 # Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
3 # SPDX-License-Identifier: AGPL-3.0-only
5 defmodule Pleroma.Web.ActivityPub.ActivityPub do
7 alias Pleroma.Activity.Ir.Topics
9 alias Pleroma.Constants
10 alias Pleroma.Conversation
11 alias Pleroma.Conversation.Participation
12 alias Pleroma.Notification
14 alias Pleroma.Object.Containment
15 alias Pleroma.Object.Fetcher
16 alias Pleroma.Pagination
20 alias Pleroma.Web.ActivityPub.MRF
21 alias Pleroma.Web.ActivityPub.Transmogrifier
22 alias Pleroma.Web.ActivityPub.Utils
23 alias Pleroma.Web.Streamer
24 alias Pleroma.Web.WebFinger
25 alias Pleroma.Workers.BackgroundWorker
28 import Pleroma.Web.ActivityPub.Utils
29 import Pleroma.Web.ActivityPub.Visibility
32 require Pleroma.Constants
34 # For Announce activities, we filter the recipients based on following status for any actors
35 # that match actual users. See issue #164 for more information about why this is necessary.
36 defp get_recipients(%{"type" => "Announce"} = data) do
37 to = Map.get(data, "to", [])
38 cc = Map.get(data, "cc", [])
39 bcc = Map.get(data, "bcc", [])
40 actor = User.get_cached_by_ap_id(data["actor"])
43 Enum.filter(Enum.concat([to, cc, bcc]), fn recipient ->
44 case User.get_cached_by_ap_id(recipient) do
46 user -> User.following?(user, actor)
53 defp get_recipients(%{"type" => "Create"} = data) do
54 to = Map.get(data, "to", [])
55 cc = Map.get(data, "cc", [])
56 bcc = Map.get(data, "bcc", [])
57 actor = Map.get(data, "actor", [])
58 recipients = [to, cc, bcc, [actor]] |> Enum.concat() |> Enum.uniq()
62 defp get_recipients(data) do
63 to = Map.get(data, "to", [])
64 cc = Map.get(data, "cc", [])
65 bcc = Map.get(data, "bcc", [])
66 recipients = Enum.concat([to, cc, bcc])
70 defp check_actor_is_active(actor) do
71 if not is_nil(actor) do
72 with user <- User.get_cached_by_ap_id(actor),
73 false <- user.deactivated do
83 defp check_remote_limit(%{"object" => %{"content" => content}}) when not is_nil(content) do
84 limit = Config.get([:instance, :remote_limit])
85 String.length(content) <= limit
88 defp check_remote_limit(_), do: true
90 def increase_note_count_if_public(actor, object) do
91 if is_public?(object), do: User.increase_note_count(actor), else: {:ok, actor}
94 def decrease_note_count_if_public(actor, object) do
95 if is_public?(object), do: User.decrease_note_count(actor), else: {:ok, actor}
98 def increase_replies_count_if_reply(%{
99 "object" => %{"inReplyTo" => reply_ap_id} = object,
102 if is_public?(object) do
103 Object.increase_replies_count(reply_ap_id)
107 def increase_replies_count_if_reply(_create_data), do: :noop
109 def decrease_replies_count_if_reply(%Object{
110 data: %{"inReplyTo" => reply_ap_id} = object
112 if is_public?(object) do
113 Object.decrease_replies_count(reply_ap_id)
117 def decrease_replies_count_if_reply(_object), do: :noop
119 def increase_poll_votes_if_vote(%{
120 "object" => %{"inReplyTo" => reply_ap_id, "name" => name},
123 Object.increase_vote_count(reply_ap_id, name)
126 def increase_poll_votes_if_vote(_create_data), do: :noop
128 @spec insert(map(), boolean(), boolean(), boolean()) :: {:ok, Activity.t()} | {:error, any()}
129 def insert(map, local \\ true, fake \\ false, bypass_actor_check \\ false) when is_map(map) do
130 with nil <- Activity.normalize(map),
131 map <- lazy_put_activity_defaults(map, fake),
132 true <- bypass_actor_check || check_actor_is_active(map["actor"]),
133 {_, true} <- {:remote_limit_error, check_remote_limit(map)},
134 {:ok, map} <- MRF.filter(map),
135 {recipients, _, _} = get_recipients(map),
136 {:fake, false, map, recipients} <- {:fake, fake, map, recipients},
137 {:containment, :ok} <- {:containment, Containment.contain_child(map)},
138 {:ok, map, object} <- insert_full_object(map) do
140 Repo.insert(%Activity{
144 recipients: recipients
147 # Splice in the child object if we have one.
149 if not is_nil(object) do
150 Map.put(activity, :object, object)
155 BackgroundWorker.enqueue("fetch_data_for_activity", %{"activity_id" => activity.id})
157 Notification.create_notifications(activity)
159 conversation = create_or_bump_conversation(activity, map["actor"])
160 participations = get_participations(conversation)
162 stream_out_participations(participations)
165 %Activity{} = activity ->
168 {:fake, true, map, recipients} ->
169 activity = %Activity{
173 recipients: recipients,
177 Pleroma.Web.RichMedia.Helpers.fetch_data_for_activity(activity)
185 defp create_or_bump_conversation(activity, actor) do
186 with {:ok, conversation} <- Conversation.create_or_bump_for(activity),
187 %User{} = user <- User.get_cached_by_ap_id(actor),
188 Participation.mark_as_read(user, conversation) do
193 defp get_participations({:ok, conversation}) do
195 |> Repo.preload(:participations, force: true)
196 |> Map.get(:participations)
199 defp get_participations(_), do: []
201 def stream_out_participations(participations) do
204 |> Repo.preload(:user)
206 Streamer.stream("participation", participations)
209 def stream_out_participations(%Object{data: %{"context" => context}}, user) do
210 with %Conversation{} = conversation <- Conversation.get_for_ap_id(context),
211 conversation = Repo.preload(conversation, :participations),
213 fetch_latest_activity_id_for_context(conversation.ap_id, %{
215 "blocking_user" => user
217 if last_activity_id do
218 stream_out_participations(conversation.participations)
223 def stream_out_participations(_, _), do: :noop
225 def stream_out(%Activity{data: %{"type" => data_type}} = activity)
226 when data_type in ["Create", "Announce", "Delete"] do
228 |> Topics.get_activity_topics()
229 |> Streamer.stream(activity)
232 def stream_out(_activity) do
236 @spec create(map(), boolean()) :: {:ok, Activity.t()} | {:error, any()}
237 def create(params, fake \\ false) do
238 with {:ok, result} <- Repo.transaction(fn -> do_create(params, fake) end) do
243 defp do_create(%{to: to, actor: actor, context: context, object: object} = params, fake) do
244 additional = params[:additional] || %{}
245 # only accept false as false value
246 local = !(params[:local] == false)
247 published = params[:published]
248 quick_insert? = Config.get([:env]) == :benchmark
252 %{to: to, actor: actor, published: published, context: context, object: object},
255 {:ok, activity} <- insert(create_data, local, fake),
256 {:fake, false, activity} <- {:fake, fake, activity},
257 _ <- increase_replies_count_if_reply(create_data),
258 _ <- increase_poll_votes_if_vote(create_data),
259 {:quick_insert, false, activity} <- {:quick_insert, quick_insert?, activity},
260 {:ok, _actor} <- increase_note_count_if_public(actor, activity),
261 :ok <- maybe_federate(activity) do
264 {:quick_insert, true, activity} ->
267 {:fake, true, activity} ->
271 Repo.rollback(message)
275 @spec listen(map()) :: {:ok, Activity.t()} | {:error, any()}
276 def listen(%{to: to, actor: actor, context: context, object: object} = params) do
277 additional = params[:additional] || %{}
278 # only accept false as false value
279 local = !(params[:local] == false)
280 published = params[:published]
284 %{to: to, actor: actor, published: published, context: context, object: object},
287 {:ok, activity} <- insert(listen_data, local),
288 :ok <- maybe_federate(activity) do
293 @spec accept(map()) :: {:ok, Activity.t()} | {:error, any()}
294 def accept(params) do
295 accept_or_reject("Accept", params)
298 @spec reject(map()) :: {:ok, Activity.t()} | {:error, any()}
299 def reject(params) do
300 accept_or_reject("Reject", params)
303 @spec accept_or_reject(String.t(), map()) :: {:ok, Activity.t()} | {:error, any()}
304 def accept_or_reject(type, %{to: to, actor: actor, object: object} = params) do
305 local = Map.get(params, :local, true)
306 activity_id = Map.get(params, :activity_id, nil)
309 %{"to" => to, "type" => type, "actor" => actor.ap_id, "object" => object}
310 |> Utils.maybe_put("id", activity_id),
311 {:ok, activity} <- insert(data, local),
312 :ok <- maybe_federate(activity) do
317 @spec update(map()) :: {:ok, Activity.t()} | {:error, any()}
318 def update(%{to: to, cc: cc, actor: actor, object: object} = params) do
319 local = !(params[:local] == false)
320 activity_id = params[:activity_id]
329 data <- Utils.maybe_put(data, "id", activity_id),
330 {:ok, activity} <- insert(data, local),
331 :ok <- maybe_federate(activity) do
336 @spec react_with_emoji(User.t(), Object.t(), String.t(), keyword()) ::
337 {:ok, Activity.t(), Object.t()} | {:error, any()}
338 def react_with_emoji(user, object, emoji, options \\ []) do
339 with {:ok, result} <-
340 Repo.transaction(fn -> do_react_with_emoji(user, object, emoji, options) end) do
345 defp do_react_with_emoji(user, object, emoji, options) do
346 with local <- Keyword.get(options, :local, true),
347 activity_id <- Keyword.get(options, :activity_id, nil),
348 true <- Pleroma.Emoji.is_unicode_emoji?(emoji),
349 reaction_data <- make_emoji_reaction_data(user, object, emoji, activity_id),
350 {:ok, activity} <- insert(reaction_data, local),
351 {:ok, object} <- add_emoji_reaction_to_object(activity, object),
352 :ok <- maybe_federate(activity) do
353 {:ok, activity, object}
355 false -> {:error, false}
356 {:error, error} -> Repo.rollback(error)
360 @spec unreact_with_emoji(User.t(), String.t(), keyword()) ::
361 {:ok, Activity.t(), Object.t()} | {:error, any()}
362 def unreact_with_emoji(user, reaction_id, options \\ []) do
363 with {:ok, result} <-
364 Repo.transaction(fn -> do_unreact_with_emoji(user, reaction_id, options) end) do
369 defp do_unreact_with_emoji(user, reaction_id, options) do
370 with local <- Keyword.get(options, :local, true),
371 activity_id <- Keyword.get(options, :activity_id, nil),
372 user_ap_id <- user.ap_id,
373 %Activity{actor: ^user_ap_id} = reaction_activity <- Activity.get_by_ap_id(reaction_id),
374 object <- Object.normalize(reaction_activity),
375 unreact_data <- make_undo_data(user, reaction_activity, activity_id),
376 {:ok, activity} <- insert(unreact_data, local),
377 {:ok, object} <- remove_emoji_reaction_from_object(reaction_activity, object),
378 :ok <- maybe_federate(activity) do
379 {:ok, activity, object}
381 {:error, error} -> Repo.rollback(error)
385 # TODO: This is weird, maybe we shouldn't check here if we can make the activity.
386 @spec like(User.t(), Object.t(), String.t() | nil, boolean()) ::
387 {:ok, Activity.t(), Object.t()} | {:error, any()}
388 def like(user, object, activity_id \\ nil, local \\ true) do
389 with {:ok, result} <- Repo.transaction(fn -> do_like(user, object, activity_id, local) end) do
395 %User{ap_id: ap_id} = user,
396 %Object{data: %{"id" => _}} = object,
400 with nil <- get_existing_like(ap_id, object),
401 like_data <- make_like_data(user, object, activity_id),
402 {:ok, activity} <- insert(like_data, local),
403 {:ok, object} <- add_like_to_object(activity, object),
404 :ok <- maybe_federate(activity) do
405 {:ok, activity, object}
407 %Activity{} = activity ->
408 {:ok, activity, object}
415 @spec unlike(User.t(), Object.t(), String.t() | nil, boolean()) ::
416 {:ok, Activity.t(), Activity.t(), Object.t()} | {:ok, Object.t()} | {:error, any()}
417 def unlike(%User{} = actor, %Object{} = object, activity_id \\ nil, local \\ true) do
418 with {:ok, result} <-
419 Repo.transaction(fn -> do_unlike(actor, object, activity_id, local) end) do
424 defp do_unlike(actor, object, activity_id, local) do
425 with %Activity{} = like_activity <- get_existing_like(actor.ap_id, object),
426 unlike_data <- make_unlike_data(actor, like_activity, activity_id),
427 {:ok, unlike_activity} <- insert(unlike_data, local),
428 {:ok, _activity} <- Repo.delete(like_activity),
429 {:ok, object} <- remove_like_from_object(like_activity, object),
430 :ok <- maybe_federate(unlike_activity) do
431 {:ok, unlike_activity, like_activity, object}
434 {:error, error} -> Repo.rollback(error)
438 @spec announce(User.t(), Object.t(), String.t() | nil, boolean(), boolean()) ::
439 {:ok, Activity.t(), Object.t()} | {:error, any()}
441 %User{ap_id: _} = user,
442 %Object{data: %{"id" => _}} = object,
447 with {:ok, result} <-
448 Repo.transaction(fn -> do_announce(user, object, activity_id, local, public) end) do
453 defp do_announce(user, object, activity_id, local, public) do
454 with true <- is_announceable?(object, user, public),
455 announce_data <- make_announce_data(user, object, activity_id, public),
456 {:ok, activity} <- insert(announce_data, local),
457 {:ok, object} <- add_announce_to_object(activity, object),
458 :ok <- maybe_federate(activity) do
459 {:ok, activity, object}
461 false -> {:error, false}
462 {:error, error} -> Repo.rollback(error)
466 @spec unannounce(User.t(), Object.t(), String.t() | nil, boolean()) ::
467 {:ok, Activity.t(), Object.t()} | {:ok, Object.t()} | {:error, any()}
474 with {:ok, result} <-
475 Repo.transaction(fn -> do_unannounce(actor, object, activity_id, local) end) do
480 defp do_unannounce(actor, object, activity_id, local) do
481 with %Activity{} = announce_activity <- get_existing_announce(actor.ap_id, object),
482 unannounce_data <- make_unannounce_data(actor, announce_activity, activity_id),
483 {:ok, unannounce_activity} <- insert(unannounce_data, local),
484 :ok <- maybe_federate(unannounce_activity),
485 {:ok, _activity} <- Repo.delete(announce_activity),
486 {:ok, object} <- remove_announce_from_object(announce_activity, object) do
487 {:ok, unannounce_activity, object}
490 {:error, error} -> Repo.rollback(error)
494 @spec follow(User.t(), User.t(), String.t() | nil, boolean()) ::
495 {:ok, Activity.t()} | {:error, any()}
496 def follow(follower, followed, activity_id \\ nil, local \\ true) do
497 with {:ok, result} <-
498 Repo.transaction(fn -> do_follow(follower, followed, activity_id, local) end) do
503 defp do_follow(follower, followed, activity_id, local) do
504 with data <- make_follow_data(follower, followed, activity_id),
505 {:ok, activity} <- insert(data, local),
506 :ok <- maybe_federate(activity) do
509 {:error, error} -> Repo.rollback(error)
513 @spec unfollow(User.t(), User.t(), String.t() | nil, boolean()) ::
514 {:ok, Activity.t()} | nil | {:error, any()}
515 def unfollow(follower, followed, activity_id \\ nil, local \\ true) do
516 with {:ok, result} <-
517 Repo.transaction(fn -> do_unfollow(follower, followed, activity_id, local) end) do
522 defp do_unfollow(follower, followed, activity_id, local) do
523 with %Activity{} = follow_activity <- fetch_latest_follow(follower, followed),
524 {:ok, follow_activity} <- update_follow_state(follow_activity, "cancelled"),
525 unfollow_data <- make_unfollow_data(follower, followed, follow_activity, activity_id),
526 {:ok, activity} <- insert(unfollow_data, local),
527 :ok <- maybe_federate(activity) do
531 {:error, error} -> Repo.rollback(error)
535 @spec delete(User.t() | Object.t(), keyword()) :: {:ok, User.t() | Object.t()} | {:error, any()}
536 def delete(entity, options \\ []) do
537 with {:ok, result} <- Repo.transaction(fn -> do_delete(entity, options) end) do
542 defp do_delete(%User{ap_id: ap_id, follower_address: follower_address} = user, _) do
544 "to" => [follower_address],
547 "object" => %{"type" => "Person", "id" => ap_id}
549 {:ok, activity} <- insert(data, true, true, true),
550 :ok <- maybe_federate(activity) do
555 defp do_delete(%Object{data: %{"id" => id, "actor" => actor}} = object, options) do
556 local = Keyword.get(options, :local, true)
557 activity_id = Keyword.get(options, :activity_id, nil)
558 actor = Keyword.get(options, :actor, actor)
560 user = User.get_cached_by_ap_id(actor)
561 to = (object.data["to"] || []) ++ (object.data["cc"] || [])
563 with create_activity <- Activity.get_create_by_object_ap_id(id),
570 "deleted_activity_id" => create_activity && create_activity.id
572 |> maybe_put("id", activity_id),
573 {:ok, activity} <- insert(data, local, false),
574 {:ok, object, _create_activity} <- Object.delete(object),
575 stream_out_participations(object, user),
576 _ <- decrease_replies_count_if_reply(object),
577 {:ok, _actor} <- decrease_note_count_if_public(user, object),
578 :ok <- maybe_federate(activity) do
586 defp do_delete(%Object{data: %{"type" => "Tombstone", "id" => ap_id}}, _) do
589 |> Activity.Queries.by_object_id()
590 |> Activity.Queries.by_type("Delete")
596 @spec block(User.t(), User.t(), String.t() | nil, boolean()) ::
597 {:ok, Activity.t()} | {:error, any()}
598 def block(blocker, blocked, activity_id \\ nil, local \\ true) do
599 with {:ok, result} <-
600 Repo.transaction(fn -> do_block(blocker, blocked, activity_id, local) end) do
605 defp do_block(blocker, blocked, activity_id, local) do
606 outgoing_blocks = Config.get([:activitypub, :outgoing_blocks])
607 unfollow_blocked = Config.get([:activitypub, :unfollow_blocked])
609 if unfollow_blocked do
610 follow_activity = fetch_latest_follow(blocker, blocked)
611 if follow_activity, do: unfollow(blocker, blocked, nil, local)
614 with true <- outgoing_blocks,
615 block_data <- make_block_data(blocker, blocked, activity_id),
616 {:ok, activity} <- insert(block_data, local),
617 :ok <- maybe_federate(activity) do
620 {:error, error} -> Repo.rollback(error)
624 @spec unblock(User.t(), User.t(), String.t() | nil, boolean()) ::
625 {:ok, Activity.t()} | {:error, any()} | nil
626 def unblock(blocker, blocked, activity_id \\ nil, local \\ true) do
627 with {:ok, result} <-
628 Repo.transaction(fn -> do_unblock(blocker, blocked, activity_id, local) end) do
633 defp do_unblock(blocker, blocked, activity_id, local) do
634 with %Activity{} = block_activity <- fetch_latest_block(blocker, blocked),
635 unblock_data <- make_unblock_data(blocker, blocked, block_activity, activity_id),
636 {:ok, activity} <- insert(unblock_data, local),
637 :ok <- maybe_federate(activity) do
641 {:error, error} -> Repo.rollback(error)
645 @spec flag(map()) :: {:ok, Activity.t()} | {:error, any()}
655 # only accept false as false value
656 local = !(params[:local] == false)
657 forward = !(params[:forward] == false)
659 additional = params[:additional] || %{}
663 Map.merge(additional, %{"to" => [], "cc" => [account.ap_id]})
665 Map.merge(additional, %{"to" => [], "cc" => []})
668 with flag_data <- make_flag_data(params, additional),
669 {:ok, activity} <- insert(flag_data, local),
670 {:ok, stripped_activity} <- strip_report_status_data(activity),
671 :ok <- maybe_federate(stripped_activity) do
672 User.all_superusers()
673 |> Enum.filter(fn user -> not is_nil(user.email) end)
674 |> Enum.each(fn superuser ->
676 |> Pleroma.Emails.AdminEmail.report(actor, account, statuses, content)
677 |> Pleroma.Emails.Mailer.deliver_async()
684 @spec move(User.t(), User.t(), boolean()) :: {:ok, Activity.t()} | {:error, any()}
685 def move(%User{} = origin, %User{} = target, local \\ true) do
688 "actor" => origin.ap_id,
689 "object" => origin.ap_id,
690 "target" => target.ap_id
693 with true <- origin.ap_id in target.also_known_as,
694 {:ok, activity} <- insert(params, local) do
695 maybe_federate(activity)
697 BackgroundWorker.enqueue("move_following", %{
698 "origin_id" => origin.id,
699 "target_id" => target.id
704 false -> {:error, "Target account must have the origin in `alsoKnownAs`"}
709 defp fetch_activities_for_context_query(context, opts) do
710 public = [Constants.as_public()]
714 do: [opts["user"].ap_id | User.following(opts["user"])] ++ public,
717 from(activity in Activity)
718 |> maybe_preload_objects(opts)
719 |> maybe_preload_bookmarks(opts)
720 |> maybe_set_thread_muted_field(opts)
721 |> restrict_blocked(opts)
722 |> restrict_recipients(recipients, opts["user"])
726 "?->>'type' = ? and ?->>'context' = ?",
733 |> exclude_poll_votes(opts)
735 |> order_by([activity], desc: activity.id)
738 @spec fetch_activities_for_context(String.t(), keyword() | map()) :: [Activity.t()]
739 def fetch_activities_for_context(context, opts \\ %{}) do
741 |> fetch_activities_for_context_query(opts)
745 @spec fetch_latest_activity_id_for_context(String.t(), keyword() | map()) ::
746 FlakeId.Ecto.CompatType.t() | nil
747 def fetch_latest_activity_id_for_context(context, opts \\ %{}) do
749 |> fetch_activities_for_context_query(Map.merge(%{"skip_preload" => true}, opts))
755 @spec fetch_public_activities(map(), Pagination.type()) :: [Activity.t()]
756 def fetch_public_activities(opts \\ %{}, pagination \\ :keyset) do
757 opts = Map.drop(opts, ["user"])
759 [Constants.as_public()]
760 |> fetch_activities_query(opts)
761 |> restrict_unlisted()
762 |> Pagination.fetch_paginated(opts, pagination)
765 @valid_visibilities ~w[direct unlisted public private]
767 defp restrict_visibility(query, %{visibility: visibility})
768 when is_list(visibility) do
769 if Enum.all?(visibility, &(&1 in @valid_visibilities)) do
775 "activity_visibility(?, ?, ?) = ANY (?)",
785 Logger.error("Could not restrict visibility to #{visibility}")
789 defp restrict_visibility(query, %{visibility: visibility})
790 when visibility in @valid_visibilities do
794 fragment("activity_visibility(?, ?, ?) = ?", a.actor, a.recipients, a.data, ^visibility)
798 defp restrict_visibility(_query, %{visibility: visibility})
799 when visibility not in @valid_visibilities do
800 Logger.error("Could not restrict visibility to #{visibility}")
803 defp restrict_visibility(query, _visibility), do: query
805 defp exclude_visibility(query, %{"exclude_visibilities" => visibility})
806 when is_list(visibility) do
807 if Enum.all?(visibility, &(&1 in @valid_visibilities)) do
812 "activity_visibility(?, ?, ?) = ANY (?)",
820 Logger.error("Could not exclude visibility to #{visibility}")
825 defp exclude_visibility(query, %{"exclude_visibilities" => visibility})
826 when visibility in @valid_visibilities do
831 "activity_visibility(?, ?, ?) = ?",
840 defp exclude_visibility(query, %{"exclude_visibilities" => visibility})
841 when visibility not in @valid_visibilities do
842 Logger.error("Could not exclude visibility to #{visibility}")
846 defp exclude_visibility(query, _visibility), do: query
848 defp restrict_thread_visibility(query, _, %{skip_thread_containment: true} = _),
851 defp restrict_thread_visibility(
853 %{"user" => %User{skip_thread_containment: true}},
858 defp restrict_thread_visibility(query, %{"user" => %User{ap_id: ap_id}}, _) do
861 where: fragment("thread_visibility(?, (?)->>'id') = true", ^ap_id, a.data)
865 defp restrict_thread_visibility(query, _, _), do: query
867 def fetch_user_abstract_activities(user, reading_user, params \\ %{}) do
870 |> Map.put("user", reading_user)
871 |> Map.put("actor_id", user.ap_id)
874 user_activities_recipients(%{
875 "godmode" => params["godmode"],
876 "reading_user" => reading_user
879 fetch_activities(recipients, params)
883 def fetch_user_activities(user, reading_user, params \\ %{}) do
886 |> Map.put("type", ["Create", "Announce"])
887 |> Map.put("user", reading_user)
888 |> Map.put("actor_id", user.ap_id)
889 |> Map.put("pinned_activity_ids", user.pinned_activities)
892 if User.blocks?(reading_user, user) do
896 |> Map.put("blocking_user", reading_user)
897 |> Map.put("muting_user", reading_user)
901 user_activities_recipients(%{
902 "godmode" => params["godmode"],
903 "reading_user" => reading_user
906 fetch_activities(recipients, params)
910 def fetch_statuses(reading_user, params) do
913 |> Map.put("type", ["Create", "Announce"])
916 user_activities_recipients(%{
917 "godmode" => params["godmode"],
918 "reading_user" => reading_user
921 fetch_activities(recipients, params, :offset)
925 defp user_activities_recipients(%{"godmode" => true}) do
929 defp user_activities_recipients(%{"reading_user" => reading_user}) do
931 [Constants.as_public()] ++ [reading_user.ap_id | User.following(reading_user)]
933 [Constants.as_public()]
937 defp restrict_since(query, %{"since_id" => ""}), do: query
939 defp restrict_since(query, %{"since_id" => since_id}) do
940 from(activity in query, where: activity.id > ^since_id)
943 defp restrict_since(query, _), do: query
945 defp restrict_tag_reject(_query, %{"tag_reject" => _tag_reject, "skip_preload" => true}) do
946 raise "Can't use the child object without preloading!"
949 defp restrict_tag_reject(query, %{"tag_reject" => tag_reject})
950 when is_list(tag_reject) and tag_reject != [] do
952 [_activity, object] in query,
953 where: fragment("not (?)->'tag' \\?| (?)", object.data, ^tag_reject)
957 defp restrict_tag_reject(query, _), do: query
959 defp restrict_tag_all(_query, %{"tag_all" => _tag_all, "skip_preload" => true}) do
960 raise "Can't use the child object without preloading!"
963 defp restrict_tag_all(query, %{"tag_all" => tag_all})
964 when is_list(tag_all) and tag_all != [] do
966 [_activity, object] in query,
967 where: fragment("(?)->'tag' \\?& (?)", object.data, ^tag_all)
971 defp restrict_tag_all(query, _), do: query
973 defp restrict_tag(_query, %{"tag" => _tag, "skip_preload" => true}) do
974 raise "Can't use the child object without preloading!"
977 defp restrict_tag(query, %{"tag" => tag}) when is_list(tag) do
979 [_activity, object] in query,
980 where: fragment("(?)->'tag' \\?| (?)", object.data, ^tag)
984 defp restrict_tag(query, %{"tag" => tag}) when is_binary(tag) do
986 [_activity, object] in query,
987 where: fragment("(?)->'tag' \\? (?)", object.data, ^tag)
991 defp restrict_tag(query, _), do: query
993 defp restrict_recipients(query, [], _user), do: query
995 defp restrict_recipients(query, recipients, nil) do
996 from(activity in query, where: fragment("? && ?", ^recipients, activity.recipients))
999 defp restrict_recipients(query, recipients, user) do
1002 where: fragment("? && ?", ^recipients, activity.recipients),
1003 or_where: activity.actor == ^user.ap_id
1007 defp restrict_local(query, %{"local_only" => true}) do
1008 from(activity in query, where: activity.local == true)
1011 defp restrict_local(query, _), do: query
1013 defp restrict_actor(query, %{"actor_id" => actor_id}) do
1014 from(activity in query, where: activity.actor == ^actor_id)
1017 defp restrict_actor(query, _), do: query
1019 defp restrict_type(query, %{"type" => type}) when is_binary(type) do
1020 from(activity in query, where: fragment("?->>'type' = ?", activity.data, ^type))
1023 defp restrict_type(query, %{"type" => type}) do
1024 from(activity in query, where: fragment("?->>'type' = ANY(?)", activity.data, ^type))
1027 defp restrict_type(query, _), do: query
1029 defp restrict_state(query, %{"state" => state}) do
1030 from(activity in query, where: fragment("?->>'state' = ?", activity.data, ^state))
1033 defp restrict_state(query, _), do: query
1035 defp restrict_favorited_by(query, %{"favorited_by" => ap_id}) do
1037 [_activity, object] in query,
1038 where: fragment("(?)->'likes' \\? (?)", object.data, ^ap_id)
1042 defp restrict_favorited_by(query, _), do: query
1044 defp restrict_media(_query, %{"only_media" => _val, "skip_preload" => true}) do
1045 raise "Can't use the child object without preloading!"
1048 defp restrict_media(query, %{"only_media" => val}) when val == "true" or val == "1" do
1050 [_activity, object] in query,
1051 where: fragment("not (?)->'attachment' = (?)", object.data, ^[])
1055 defp restrict_media(query, _), do: query
1057 defp restrict_replies(query, %{"exclude_replies" => val}) when val == "true" or val == "1" do
1059 [_activity, object] in query,
1060 where: fragment("?->>'inReplyTo' is null", object.data)
1064 defp restrict_replies(query, _), do: query
1066 defp restrict_reblogs(query, %{"exclude_reblogs" => val}) when val == "true" or val == "1" do
1067 from(activity in query, where: fragment("?->>'type' != 'Announce'", activity.data))
1070 defp restrict_reblogs(query, _), do: query
1072 defp restrict_muted(query, %{"with_muted" => val}) when val in [true, "true", "1"], do: query
1074 defp restrict_muted(query, %{"muting_user" => %User{} = user} = opts) do
1075 mutes = opts["muted_users_ap_ids"] || User.muted_users_ap_ids(user)
1078 from([activity] in query,
1079 where: fragment("not (? = ANY(?))", activity.actor, ^mutes),
1080 where: fragment("not (?->'to' \\?| ?)", activity.data, ^mutes)
1083 unless opts["skip_preload"] do
1084 from([thread_mute: tm] in query, where: is_nil(tm.user_id))
1090 defp restrict_muted(query, _), do: query
1092 defp restrict_blocked(query, %{"blocking_user" => %User{} = user} = opts) do
1093 blocked_ap_ids = opts["blocked_users_ap_ids"] || User.blocked_users_ap_ids(user)
1094 domain_blocks = user.domain_blocks || []
1096 following_ap_ids = User.get_friends_ap_ids(user)
1099 if has_named_binding?(query, :object), do: query, else: Activity.with_joined_object(query)
1102 [activity, object: o] in query,
1103 where: fragment("not (? = ANY(?))", activity.actor, ^blocked_ap_ids),
1104 where: fragment("not (? && ?)", activity.recipients, ^blocked_ap_ids),
1107 "not (?->>'type' = 'Announce' and ?->'to' \\?| ?)",
1114 "(not (split_part(?, '/', 3) = ANY(?))) or ? = ANY(?)",
1122 "(not (split_part(?->>'actor', '/', 3) = ANY(?))) or (?->>'actor') = ANY(?)",
1131 defp restrict_blocked(query, _), do: query
1133 defp restrict_unlisted(query) do
1138 "not (coalesce(?->'cc', '{}'::jsonb) \\?| ?)",
1140 ^[Constants.as_public()]
1145 defp restrict_pinned(query, %{"pinned" => "true", "pinned_activity_ids" => ids}) do
1146 from(activity in query, where: activity.id in ^ids)
1149 defp restrict_pinned(query, _), do: query
1151 defp restrict_muted_reblogs(query, %{"muting_user" => %User{} = user} = opts) do
1152 muted_reblogs = opts["reblog_muted_users_ap_ids"] || User.reblog_muted_users_ap_ids(user)
1158 "not ( ?->>'type' = 'Announce' and ? = ANY(?))",
1166 defp restrict_muted_reblogs(query, _), do: query
1168 defp restrict_instance(query, %{"instance" => instance}) do
1173 where: fragment("? LIKE ?", u.nickname, ^"%@#{instance}")
1177 from(activity in query, where: activity.actor in ^users)
1180 defp restrict_instance(query, _), do: query
1182 defp exclude_poll_votes(query, %{"include_poll_votes" => true}), do: query
1184 defp exclude_poll_votes(query, _) do
1185 if has_named_binding?(query, :object) do
1186 from([activity, object: o] in query,
1187 where: fragment("not(?->>'type' = ?)", o.data, "Answer")
1194 defp exclude_id(query, %{"exclude_id" => id}) when is_binary(id) do
1195 from(activity in query, where: activity.id != ^id)
1198 defp exclude_id(query, _), do: query
1200 defp maybe_preload_objects(query, %{"skip_preload" => true}), do: query
1202 defp maybe_preload_objects(query, _) do
1204 |> Activity.with_preloaded_object()
1207 defp maybe_preload_bookmarks(query, %{"skip_preload" => true}), do: query
1209 defp maybe_preload_bookmarks(query, opts) do
1211 |> Activity.with_preloaded_bookmark(opts["user"])
1214 defp maybe_preload_report_notes(query, %{"preload_report_notes" => true}) do
1216 |> Activity.with_preloaded_report_notes()
1219 defp maybe_preload_report_notes(query, _), do: query
1221 defp maybe_set_thread_muted_field(query, %{"skip_preload" => true}), do: query
1223 defp maybe_set_thread_muted_field(query, opts) do
1225 |> Activity.with_set_thread_muted_field(opts["muting_user"] || opts["user"])
1228 defp maybe_order(query, %{order: :desc}) do
1230 |> order_by(desc: :id)
1233 defp maybe_order(query, %{order: :asc}) do
1235 |> order_by(asc: :id)
1238 defp maybe_order(query, _), do: query
1240 defp fetch_activities_query_ap_ids_ops(opts) do
1241 source_user = opts["muting_user"]
1242 ap_id_relationships = if source_user, do: [:mute, :reblog_mute], else: []
1244 ap_id_relationships =
1245 ap_id_relationships ++
1246 if opts["blocking_user"] && opts["blocking_user"] == source_user do
1252 preloaded_ap_ids = User.outgoing_relationships_ap_ids(source_user, ap_id_relationships)
1254 restrict_blocked_opts = Map.merge(%{"blocked_users_ap_ids" => preloaded_ap_ids[:block]}, opts)
1255 restrict_muted_opts = Map.merge(%{"muted_users_ap_ids" => preloaded_ap_ids[:mute]}, opts)
1257 restrict_muted_reblogs_opts =
1258 Map.merge(%{"reblog_muted_users_ap_ids" => preloaded_ap_ids[:reblog_mute]}, opts)
1260 {restrict_blocked_opts, restrict_muted_opts, restrict_muted_reblogs_opts}
1263 def fetch_activities_query(recipients, opts \\ %{}) do
1264 {restrict_blocked_opts, restrict_muted_opts, restrict_muted_reblogs_opts} =
1265 fetch_activities_query_ap_ids_ops(opts)
1268 skip_thread_containment: Config.get([:instance, :skip_thread_containment])
1272 |> maybe_preload_objects(opts)
1273 |> maybe_preload_bookmarks(opts)
1274 |> maybe_preload_report_notes(opts)
1275 |> maybe_set_thread_muted_field(opts)
1276 |> maybe_order(opts)
1277 |> restrict_recipients(recipients, opts["user"])
1278 |> restrict_tag(opts)
1279 |> restrict_tag_reject(opts)
1280 |> restrict_tag_all(opts)
1281 |> restrict_since(opts)
1282 |> restrict_local(opts)
1283 |> restrict_actor(opts)
1284 |> restrict_type(opts)
1285 |> restrict_state(opts)
1286 |> restrict_favorited_by(opts)
1287 |> restrict_blocked(restrict_blocked_opts)
1288 |> restrict_muted(restrict_muted_opts)
1289 |> restrict_media(opts)
1290 |> restrict_visibility(opts)
1291 |> restrict_thread_visibility(opts, config)
1292 |> restrict_replies(opts)
1293 |> restrict_reblogs(opts)
1294 |> restrict_pinned(opts)
1295 |> restrict_muted_reblogs(restrict_muted_reblogs_opts)
1296 |> restrict_instance(opts)
1297 |> Activity.restrict_deactivated_users()
1298 |> exclude_poll_votes(opts)
1299 |> exclude_visibility(opts)
1302 def fetch_activities(recipients, opts \\ %{}, pagination \\ :keyset) do
1303 list_memberships = Pleroma.List.memberships(opts["user"])
1305 fetch_activities_query(recipients ++ list_memberships, opts)
1306 |> Pagination.fetch_paginated(opts, pagination)
1308 |> maybe_update_cc(list_memberships, opts["user"])
1312 Fetch favorites activities of user with order by sort adds to favorites
1314 @spec fetch_favourites(User.t(), map(), Pagination.type()) :: list(Activity.t())
1315 def fetch_favourites(user, params \\ %{}, pagination \\ :keyset) do
1317 |> Activity.Queries.by_actor()
1318 |> Activity.Queries.by_type("Like")
1319 |> Activity.with_joined_object()
1320 |> Object.with_joined_activity()
1321 |> select([_like, object, activity], %{activity | object: object})
1322 |> order_by([like, _, _], desc: like.id)
1323 |> Pagination.fetch_paginated(
1324 Map.merge(params, %{"skip_order" => true}),
1330 defp maybe_update_cc(activities, list_memberships, %User{ap_id: user_ap_id})
1331 when is_list(list_memberships) and length(list_memberships) > 0 do
1332 Enum.map(activities, fn
1333 %{data: %{"bcc" => bcc}} = activity when is_list(bcc) and length(bcc) > 0 ->
1334 if Enum.any?(bcc, &(&1 in list_memberships)) do
1335 update_in(activity.data["cc"], &[user_ap_id | &1])
1345 defp maybe_update_cc(activities, _, _), do: activities
1347 def fetch_activities_bounded_query(query, recipients, recipients_with_public) do
1348 from(activity in query,
1350 fragment("? && ?", activity.recipients, ^recipients) or
1351 (fragment("? && ?", activity.recipients, ^recipients_with_public) and
1352 ^Constants.as_public() in activity.recipients)
1356 def fetch_activities_bounded(
1358 recipients_with_public,
1360 pagination \\ :keyset
1362 fetch_activities_query([], opts)
1363 |> fetch_activities_bounded_query(recipients, recipients_with_public)
1364 |> Pagination.fetch_paginated(opts, pagination)
1368 @spec upload(Upload.source(), keyword()) :: {:ok, Object.t()} | {:error, any()}
1369 def upload(file, opts \\ []) do
1370 with {:ok, data} <- Upload.store(file, opts) do
1373 Map.put(data, "actor", opts[:actor])
1378 Repo.insert(%Object{data: obj_data})
1382 @spec get_actor_url(any()) :: binary() | nil
1383 defp get_actor_url(url) when is_binary(url), do: url
1384 defp get_actor_url(%{"href" => href}) when is_binary(href), do: href
1386 defp get_actor_url(url) when is_list(url) do
1392 defp get_actor_url(_url), do: nil
1394 defp object_to_user_data(data) do
1396 data["icon"]["url"] &&
1399 "url" => [%{"href" => data["icon"]["url"]}]
1403 data["image"]["url"] &&
1406 "url" => [%{"href" => data["image"]["url"]}]
1411 |> Map.get("attachment", [])
1412 |> Enum.filter(fn %{"type" => t} -> t == "PropertyValue" end)
1413 |> Enum.map(fn fields -> Map.take(fields, ["name", "value"]) end)
1415 locked = data["manuallyApprovesFollowers"] || false
1416 data = Transmogrifier.maybe_fix_user_object(data)
1417 discoverable = data["discoverable"] || false
1418 invisible = data["invisible"] || false
1419 actor_type = data["type"] || "Person"
1423 uri: get_actor_url(data["url"]),
1429 discoverable: discoverable,
1430 invisible: invisible,
1433 follower_address: data["followers"],
1434 following_address: data["following"],
1435 bio: data["summary"],
1436 actor_type: actor_type,
1437 also_known_as: Map.get(data, "alsoKnownAs", [])
1440 # nickname can be nil because of virtual actors
1442 if data["preferredUsername"] do
1446 "#{data["preferredUsername"]}@#{URI.parse(data["id"]).host}"
1449 Map.put(user_data, :nickname, nil)
1455 def fetch_follow_information_for_user(user) do
1456 with {:ok, following_data} <-
1457 Fetcher.fetch_and_contain_remote_object_from_id(user.following_address),
1458 {:ok, hide_follows} <- collection_private(following_data),
1459 {:ok, followers_data} <-
1460 Fetcher.fetch_and_contain_remote_object_from_id(user.follower_address),
1461 {:ok, hide_followers} <- collection_private(followers_data) do
1464 hide_follows: hide_follows,
1465 follower_count: normalize_counter(followers_data["totalItems"]),
1466 following_count: normalize_counter(following_data["totalItems"]),
1467 hide_followers: hide_followers
1470 {:error, _} = e -> e
1475 defp normalize_counter(counter) when is_integer(counter), do: counter
1476 defp normalize_counter(_), do: 0
1478 defp maybe_update_follow_information(data) do
1479 with {:enabled, true} <- {:enabled, Config.get([:instance, :external_user_synchronization])},
1480 {:ok, info} <- fetch_follow_information_for_user(data) do
1481 info = Map.merge(data[:info] || %{}, info)
1482 Map.put(data, :info, info)
1484 {:enabled, false} ->
1489 "Follower/Following counter update for #{data.ap_id} failed.\n" <> inspect(e)
1496 defp collection_private(%{"first" => %{"type" => type}})
1497 when type in ["CollectionPage", "OrderedCollectionPage"],
1500 defp collection_private(%{"first" => first}) do
1501 with {:ok, %{"type" => type}} when type in ["CollectionPage", "OrderedCollectionPage"] <-
1502 Fetcher.fetch_and_contain_remote_object_from_id(first) do
1505 {:error, {:ok, %{status: code}}} when code in [401, 403] -> {:ok, true}
1506 {:error, _} = e -> e
1511 defp collection_private(_data), do: {:ok, true}
1513 def user_data_from_user_object(data) do
1514 with {:ok, data} <- MRF.filter(data),
1515 {:ok, data} <- object_to_user_data(data) do
1522 def fetch_and_prepare_user_from_ap_id(ap_id) do
1523 with {:ok, data} <- Fetcher.fetch_and_contain_remote_object_from_id(ap_id),
1524 {:ok, data} <- user_data_from_user_object(data),
1525 data <- maybe_update_follow_information(data) do
1528 {:error, "Object has been deleted"} = e ->
1529 Logger.debug("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
1533 Logger.error("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
1538 def make_user_from_ap_id(ap_id) do
1539 if _user = User.get_cached_by_ap_id(ap_id) do
1540 Transmogrifier.upgrade_user_from_ap_id(ap_id)
1542 with {:ok, data} <- fetch_and_prepare_user_from_ap_id(ap_id) do
1543 User.insert_or_update_user(data)
1550 def make_user_from_nickname(nickname) do
1551 with {:ok, %{"ap_id" => ap_id}} when not is_nil(ap_id) <- WebFinger.finger(nickname) do
1552 make_user_from_ap_id(ap_id)
1554 _e -> {:error, "No AP id in WebFinger"}
1558 # filter out broken threads
1559 def contain_broken_threads(%Activity{} = activity, %User{} = user) do
1560 entire_thread_visible_for_user?(activity, user)
1563 # do post-processing on a specific activity
1564 def contain_activity(%Activity{} = activity, %User{} = user) do
1565 contain_broken_threads(activity, user)
1568 def fetch_direct_messages_query do
1570 |> restrict_type(%{"type" => "Create"})
1571 |> restrict_visibility(%{visibility: "direct"})
1572 |> order_by([activity], asc: activity.id)