git.squeep.com Git - akkoma/blob - lib/pleroma/web/activity_pub/activity_pub.ex

   1 # Pleroma: A lightweight social networking server
   2 # Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
   3 # SPDX-License-Identifier: AGPL-3.0-only
   4
   5 defmodule Pleroma.Web.ActivityPub.ActivityPub do
   6   alias Pleroma.Activity
   7   alias Pleroma.Activity.Ir.Topics
   8   alias Pleroma.Config
   9   alias Pleroma.Constants
  10   alias Pleroma.Conversation
  11   alias Pleroma.Conversation.Participation
  12   alias Pleroma.Notification
  13   alias Pleroma.Object
  14   alias Pleroma.Object.Containment
  15   alias Pleroma.Object.Fetcher
  16   alias Pleroma.Pagination
  17   alias Pleroma.Repo
  18   alias Pleroma.Upload
  19   alias Pleroma.User
  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
  26
  27   import Ecto.Query
  28   import Pleroma.Web.ActivityPub.Utils
  29   import Pleroma.Web.ActivityPub.Visibility
  30
  31   require Logger
  32   require Pleroma.Constants
  33
  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"])
  41
  42     recipients =
  43       Enum.filter(Enum.concat([to, cc, bcc]), fn recipient ->
  44         case User.get_cached_by_ap_id(recipient) do
  45           nil -> true
  46           user -> User.following?(user, actor)
  47         end
  48       end)
  49
  50     {recipients, to, cc}
  51   end
  52
  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()
  59     {recipients, to, cc}
  60   end
  61
  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])
  67     {recipients, to, cc}
  68   end
  69
  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
  74         true
  75       else
  76         _e -> false
  77       end
  78     else
  79       true
  80     end
  81   end
  82
  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
  86   end
  87
  88   defp check_remote_limit(_), do: true
  89
  90   def increase_note_count_if_public(actor, object) do
  91     if is_public?(object), do: User.increase_note_count(actor), else: {:ok, actor}
  92   end
  93
  94   def decrease_note_count_if_public(actor, object) do
  95     if is_public?(object), do: User.decrease_note_count(actor), else: {:ok, actor}
  96   end
  97
  98   def increase_replies_count_if_reply(%{
  99         "object" => %{"inReplyTo" => reply_ap_id} = object,
 100         "type" => "Create"
 101       }) do
 102     if is_public?(object) do
 103       Object.increase_replies_count(reply_ap_id)
 104     end
 105   end
 106
 107   def increase_replies_count_if_reply(_create_data), do: :noop
 108
 109   def decrease_replies_count_if_reply(%Object{
 110         data: %{"inReplyTo" => reply_ap_id} = object
 111       }) do
 112     if is_public?(object) do
 113       Object.decrease_replies_count(reply_ap_id)
 114     end
 115   end
 116
 117   def decrease_replies_count_if_reply(_object), do: :noop
 118
 119   def increase_poll_votes_if_vote(%{
 120         "object" => %{"inReplyTo" => reply_ap_id, "name" => name},
 121         "type" => "Create",
 122         "actor" => actor
 123       }) do
 124     Object.increase_vote_count(reply_ap_id, name, actor)
 125   end
 126
 127   def increase_poll_votes_if_vote(_create_data), do: :noop
 128
 129   @spec persist(map(), keyword()) :: {:ok, Activity.t() | Object.t()}
 130   def persist(object, meta) do
 131     with local <- Keyword.fetch!(meta, :local),
 132          {recipients, _, _} <- get_recipients(object),
 133          {:ok, activity} <-
 134            Repo.insert(%Activity{
 135              data: object,
 136              local: local,
 137              recipients: recipients,
 138              actor: object["actor"]
 139            }) do
 140       {:ok, activity, meta}
 141     end
 142   end
 143
 144   @spec insert(map(), boolean(), boolean(), boolean()) :: {:ok, Activity.t()} | {:error, any()}
 145   def insert(map, local \\ true, fake \\ false, bypass_actor_check \\ false) when is_map(map) do
 146     with nil <- Activity.normalize(map),
 147          map <- lazy_put_activity_defaults(map, fake),
 148          true <- bypass_actor_check || check_actor_is_active(map["actor"]),
 149          {_, true} <- {:remote_limit_error, check_remote_limit(map)},
 150          {:ok, map} <- MRF.filter(map),
 151          {recipients, _, _} = get_recipients(map),
 152          {:fake, false, map, recipients} <- {:fake, fake, map, recipients},
 153          {:containment, :ok} <- {:containment, Containment.contain_child(map)},
 154          {:ok, map, object} <- insert_full_object(map) do
 155       {:ok, activity} =
 156         Repo.insert(%Activity{
 157           data: map,
 158           local: local,
 159           actor: map["actor"],
 160           recipients: recipients
 161         })
 162
 163       # Splice in the child object if we have one.
 164       activity =
 165         if not is_nil(object) do
 166           Map.put(activity, :object, object)
 167         else
 168           activity
 169         end
 170
 171       BackgroundWorker.enqueue("fetch_data_for_activity", %{"activity_id" => activity.id})
 172
 173       Notification.create_notifications(activity)
 174
 175       conversation = create_or_bump_conversation(activity, map["actor"])
 176       participations = get_participations(conversation)
 177       stream_out(activity)
 178       stream_out_participations(participations)
 179       {:ok, activity}
 180     else
 181       %Activity{} = activity ->
 182         {:ok, activity}
 183
 184       {:fake, true, map, recipients} ->
 185         activity = %Activity{
 186           data: map,
 187           local: local,
 188           actor: map["actor"],
 189           recipients: recipients,
 190           id: "pleroma:fakeid"
 191         }
 192
 193         Pleroma.Web.RichMedia.Helpers.fetch_data_for_activity(activity)
 194         {:ok, activity}
 195
 196       error ->
 197         {:error, error}
 198     end
 199   end
 200
 201   defp create_or_bump_conversation(activity, actor) do
 202     with {:ok, conversation} <- Conversation.create_or_bump_for(activity),
 203          %User{} = user <- User.get_cached_by_ap_id(actor),
 204          Participation.mark_as_read(user, conversation) do
 205       {:ok, conversation}
 206     end
 207   end
 208
 209   defp get_participations({:ok, conversation}) do
 210     conversation
 211     |> Repo.preload(:participations, force: true)
 212     |> Map.get(:participations)
 213   end
 214
 215   defp get_participations(_), do: []
 216
 217   def stream_out_participations(participations) do
 218     participations =
 219       participations
 220       |> Repo.preload(:user)
 221
 222     Streamer.stream("participation", participations)
 223   end
 224
 225   def stream_out_participations(%Object{data: %{"context" => context}}, user) do
 226     with %Conversation{} = conversation <- Conversation.get_for_ap_id(context),
 227          conversation = Repo.preload(conversation, :participations),
 228          last_activity_id =
 229            fetch_latest_activity_id_for_context(conversation.ap_id, %{
 230              "user" => user,
 231              "blocking_user" => user
 232            }) do
 233       if last_activity_id do
 234         stream_out_participations(conversation.participations)
 235       end
 236     end
 237   end
 238
 239   def stream_out_participations(_, _), do: :noop
 240
 241   def stream_out(%Activity{data: %{"type" => data_type}} = activity)
 242       when data_type in ["Create", "Announce", "Delete"] do
 243     activity
 244     |> Topics.get_activity_topics()
 245     |> Streamer.stream(activity)
 246   end
 247
 248   def stream_out(_activity) do
 249     :noop
 250   end
 251
 252   @spec create(map(), boolean()) :: {:ok, Activity.t()} | {:error, any()}
 253   def create(params, fake \\ false) do
 254     with {:ok, result} <- Repo.transaction(fn -> do_create(params, fake) end) do
 255       result
 256     end
 257   end
 258
 259   defp do_create(%{to: to, actor: actor, context: context, object: object} = params, fake) do
 260     additional = params[:additional] || %{}
 261     # only accept false as false value
 262     local = !(params[:local] == false)
 263     published = params[:published]
 264     quick_insert? = Config.get([:env]) == :benchmark
 265
 266     with create_data <-
 267            make_create_data(
 268              %{to: to, actor: actor, published: published, context: context, object: object},
 269              additional
 270            ),
 271          {:ok, activity} <- insert(create_data, local, fake),
 272          {:fake, false, activity} <- {:fake, fake, activity},
 273          _ <- increase_replies_count_if_reply(create_data),
 274          _ <- increase_poll_votes_if_vote(create_data),
 275          {:quick_insert, false, activity} <- {:quick_insert, quick_insert?, activity},
 276          {:ok, _actor} <- increase_note_count_if_public(actor, activity),
 277          :ok <- maybe_federate(activity) do
 278       {:ok, activity}
 279     else
 280       {:quick_insert, true, activity} ->
 281         {:ok, activity}
 282
 283       {:fake, true, activity} ->
 284         {:ok, activity}
 285
 286       {:error, message} ->
 287         Repo.rollback(message)
 288     end
 289   end
 290
 291   @spec listen(map()) :: {:ok, Activity.t()} | {:error, any()}
 292   def listen(%{to: to, actor: actor, context: context, object: object} = params) do
 293     additional = params[:additional] || %{}
 294     # only accept false as false value
 295     local = !(params[:local] == false)
 296     published = params[:published]
 297
 298     with listen_data <-
 299            make_listen_data(
 300              %{to: to, actor: actor, published: published, context: context, object: object},
 301              additional
 302            ),
 303          {:ok, activity} <- insert(listen_data, local),
 304          :ok <- maybe_federate(activity) do
 305       {:ok, activity}
 306     end
 307   end
 308
 309   @spec accept(map()) :: {:ok, Activity.t()} | {:error, any()}
 310   def accept(params) do
 311     accept_or_reject("Accept", params)
 312   end
 313
 314   @spec reject(map()) :: {:ok, Activity.t()} | {:error, any()}
 315   def reject(params) do
 316     accept_or_reject("Reject", params)
 317   end
 318
 319   @spec accept_or_reject(String.t(), map()) :: {:ok, Activity.t()} | {:error, any()}
 320   def accept_or_reject(type, %{to: to, actor: actor, object: object} = params) do
 321     local = Map.get(params, :local, true)
 322     activity_id = Map.get(params, :activity_id, nil)
 323
 324     with data <-
 325            %{"to" => to, "type" => type, "actor" => actor.ap_id, "object" => object}
 326            |> Utils.maybe_put("id", activity_id),
 327          {:ok, activity} <- insert(data, local),
 328          :ok <- maybe_federate(activity) do
 329       {:ok, activity}
 330     end
 331   end
 332
 333   @spec update(map()) :: {:ok, Activity.t()} | {:error, any()}
 334   def update(%{to: to, cc: cc, actor: actor, object: object} = params) do
 335     local = !(params[:local] == false)
 336     activity_id = params[:activity_id]
 337
 338     with data <- %{
 339            "to" => to,
 340            "cc" => cc,
 341            "type" => "Update",
 342            "actor" => actor,
 343            "object" => object
 344          },
 345          data <- Utils.maybe_put(data, "id", activity_id),
 346          {:ok, activity} <- insert(data, local),
 347          :ok <- maybe_federate(activity) do
 348       {:ok, activity}
 349     end
 350   end
 351
 352   @spec react_with_emoji(User.t(), Object.t(), String.t(), keyword()) ::
 353           {:ok, Activity.t(), Object.t()} | {:error, any()}
 354   def react_with_emoji(user, object, emoji, options \\ []) do
 355     with {:ok, result} <-
 356            Repo.transaction(fn -> do_react_with_emoji(user, object, emoji, options) end) do
 357       result
 358     end
 359   end
 360
 361   defp do_react_with_emoji(user, object, emoji, options) do
 362     with local <- Keyword.get(options, :local, true),
 363          activity_id <- Keyword.get(options, :activity_id, nil),
 364          true <- Pleroma.Emoji.is_unicode_emoji?(emoji),
 365          reaction_data <- make_emoji_reaction_data(user, object, emoji, activity_id),
 366          {:ok, activity} <- insert(reaction_data, local),
 367          {:ok, object} <- add_emoji_reaction_to_object(activity, object),
 368          :ok <- maybe_federate(activity) do
 369       {:ok, activity, object}
 370     else
 371       false -> {:error, false}
 372       {:error, error} -> Repo.rollback(error)
 373     end
 374   end
 375
 376   @spec unreact_with_emoji(User.t(), String.t(), keyword()) ::
 377           {:ok, Activity.t(), Object.t()} | {:error, any()}
 378   def unreact_with_emoji(user, reaction_id, options \\ []) do
 379     with {:ok, result} <-
 380            Repo.transaction(fn -> do_unreact_with_emoji(user, reaction_id, options) end) do
 381       result
 382     end
 383   end
 384
 385   defp do_unreact_with_emoji(user, reaction_id, options) do
 386     with local <- Keyword.get(options, :local, true),
 387          activity_id <- Keyword.get(options, :activity_id, nil),
 388          user_ap_id <- user.ap_id,
 389          %Activity{actor: ^user_ap_id} = reaction_activity <- Activity.get_by_ap_id(reaction_id),
 390          object <- Object.normalize(reaction_activity),
 391          unreact_data <- make_undo_data(user, reaction_activity, activity_id),
 392          {:ok, activity} <- insert(unreact_data, local),
 393          {:ok, object} <- remove_emoji_reaction_from_object(reaction_activity, object),
 394          :ok <- maybe_federate(activity) do
 395       {:ok, activity, object}
 396     else
 397       {:error, error} -> Repo.rollback(error)
 398     end
 399   end
 400
 401   # TODO: This is weird, maybe we shouldn't check here if we can make the activity.
 402   @spec like(User.t(), Object.t(), String.t() | nil, boolean()) ::
 403           {:ok, Activity.t(), Object.t()} | {:error, any()}
 404   def like(user, object, activity_id \\ nil, local \\ true) do
 405     with {:ok, result} <- Repo.transaction(fn -> do_like(user, object, activity_id, local) end) do
 406       result
 407     end
 408   end
 409
 410   defp do_like(
 411          %User{ap_id: ap_id} = user,
 412          %Object{data: %{"id" => _}} = object,
 413          activity_id,
 414          local
 415        ) do
 416     with nil <- get_existing_like(ap_id, object),
 417          like_data <- make_like_data(user, object, activity_id),
 418          {:ok, activity} <- insert(like_data, local),
 419          {:ok, object} <- add_like_to_object(activity, object),
 420          :ok <- maybe_federate(activity) do
 421       {:ok, activity, object}
 422     else
 423       %Activity{} = activity ->
 424         {:ok, activity, object}
 425
 426       {:error, error} ->
 427         Repo.rollback(error)
 428     end
 429   end
 430
 431   @spec unlike(User.t(), Object.t(), String.t() | nil, boolean()) ::
 432           {:ok, Activity.t(), Activity.t(), Object.t()} | {:ok, Object.t()} | {:error, any()}
 433   def unlike(%User{} = actor, %Object{} = object, activity_id \\ nil, local \\ true) do
 434     with {:ok, result} <-
 435            Repo.transaction(fn -> do_unlike(actor, object, activity_id, local) end) do
 436       result
 437     end
 438   end
 439
 440   defp do_unlike(actor, object, activity_id, local) do
 441     with %Activity{} = like_activity <- get_existing_like(actor.ap_id, object),
 442          unlike_data <- make_unlike_data(actor, like_activity, activity_id),
 443          {:ok, unlike_activity} <- insert(unlike_data, local),
 444          {:ok, _activity} <- Repo.delete(like_activity),
 445          {:ok, object} <- remove_like_from_object(like_activity, object),
 446          :ok <- maybe_federate(unlike_activity) do
 447       {:ok, unlike_activity, like_activity, object}
 448     else
 449       nil -> {:ok, object}
 450       {:error, error} -> Repo.rollback(error)
 451     end
 452   end
 453
 454   @spec announce(User.t(), Object.t(), String.t() | nil, boolean(), boolean()) ::
 455           {:ok, Activity.t(), Object.t()} | {:error, any()}
 456   def announce(
 457         %User{ap_id: _} = user,
 458         %Object{data: %{"id" => _}} = object,
 459         activity_id \\ nil,
 460         local \\ true,
 461         public \\ true
 462       ) do
 463     with {:ok, result} <-
 464            Repo.transaction(fn -> do_announce(user, object, activity_id, local, public) end) do
 465       result
 466     end
 467   end
 468
 469   defp do_announce(user, object, activity_id, local, public) do
 470     with true <- is_announceable?(object, user, public),
 471          announce_data <- make_announce_data(user, object, activity_id, public),
 472          {:ok, activity} <- insert(announce_data, local),
 473          {:ok, object} <- add_announce_to_object(activity, object),
 474          :ok <- maybe_federate(activity) do
 475       {:ok, activity, object}
 476     else
 477       false -> {:error, false}
 478       {:error, error} -> Repo.rollback(error)
 479     end
 480   end
 481
 482   @spec unannounce(User.t(), Object.t(), String.t() | nil, boolean()) ::
 483           {:ok, Activity.t(), Object.t()} | {:ok, Object.t()} | {:error, any()}
 484   def unannounce(
 485         %User{} = actor,
 486         %Object{} = object,
 487         activity_id \\ nil,
 488         local \\ true
 489       ) do
 490     with {:ok, result} <-
 491            Repo.transaction(fn -> do_unannounce(actor, object, activity_id, local) end) do
 492       result
 493     end
 494   end
 495
 496   defp do_unannounce(actor, object, activity_id, local) do
 497     with %Activity{} = announce_activity <- get_existing_announce(actor.ap_id, object),
 498          unannounce_data <- make_unannounce_data(actor, announce_activity, activity_id),
 499          {:ok, unannounce_activity} <- insert(unannounce_data, local),
 500          :ok <- maybe_federate(unannounce_activity),
 501          {:ok, _activity} <- Repo.delete(announce_activity),
 502          {:ok, object} <- remove_announce_from_object(announce_activity, object) do
 503       {:ok, unannounce_activity, object}
 504     else
 505       nil -> {:ok, object}
 506       {:error, error} -> Repo.rollback(error)
 507     end
 508   end
 509
 510   @spec follow(User.t(), User.t(), String.t() | nil, boolean()) ::
 511           {:ok, Activity.t()} | {:error, any()}
 512   def follow(follower, followed, activity_id \\ nil, local \\ true) do
 513     with {:ok, result} <-
 514            Repo.transaction(fn -> do_follow(follower, followed, activity_id, local) end) do
 515       result
 516     end
 517   end
 518
 519   defp do_follow(follower, followed, activity_id, local) do
 520     with data <- make_follow_data(follower, followed, activity_id),
 521          {:ok, activity} <- insert(data, local),
 522          :ok <- maybe_federate(activity) do
 523       {:ok, activity}
 524     else
 525       {:error, error} -> Repo.rollback(error)
 526     end
 527   end
 528
 529   @spec unfollow(User.t(), User.t(), String.t() | nil, boolean()) ::
 530           {:ok, Activity.t()} | nil | {:error, any()}
 531   def unfollow(follower, followed, activity_id \\ nil, local \\ true) do
 532     with {:ok, result} <-
 533            Repo.transaction(fn -> do_unfollow(follower, followed, activity_id, local) end) do
 534       result
 535     end
 536   end
 537
 538   defp do_unfollow(follower, followed, activity_id, local) do
 539     with %Activity{} = follow_activity <- fetch_latest_follow(follower, followed),
 540          {:ok, follow_activity} <- update_follow_state(follow_activity, "cancelled"),
 541          unfollow_data <- make_unfollow_data(follower, followed, follow_activity, activity_id),
 542          {:ok, activity} <- insert(unfollow_data, local),
 543          :ok <- maybe_federate(activity) do
 544       {:ok, activity}
 545     else
 546       nil -> nil
 547       {:error, error} -> Repo.rollback(error)
 548     end
 549   end
 550
 551   @spec delete(User.t() | Object.t(), keyword()) :: {:ok, User.t() | Object.t()} | {:error, any()}
 552   def delete(entity, options \\ []) do
 553     with {:ok, result} <- Repo.transaction(fn -> do_delete(entity, options) end) do
 554       result
 555     end
 556   end
 557
 558   defp do_delete(%User{ap_id: ap_id, follower_address: follower_address} = user, _) do
 559     with data <- %{
 560            "to" => [follower_address],
 561            "type" => "Delete",
 562            "actor" => ap_id,
 563            "object" => %{"type" => "Person", "id" => ap_id}
 564          },
 565          {:ok, activity} <- insert(data, true, true, true),
 566          :ok <- maybe_federate(activity) do
 567       {:ok, user}
 568     end
 569   end
 570
 571   defp do_delete(%Object{data: %{"id" => id, "actor" => actor}} = object, options) do
 572     local = Keyword.get(options, :local, true)
 573     activity_id = Keyword.get(options, :activity_id, nil)
 574     actor = Keyword.get(options, :actor, actor)
 575
 576     user = User.get_cached_by_ap_id(actor)
 577     to = (object.data["to"] || []) ++ (object.data["cc"] || [])
 578
 579     with create_activity <- Activity.get_create_by_object_ap_id(id),
 580          data <-
 581            %{
 582              "type" => "Delete",
 583              "actor" => actor,
 584              "object" => id,
 585              "to" => to,
 586              "deleted_activity_id" => create_activity && create_activity.id
 587            }
 588            |> maybe_put("id", activity_id),
 589          {:ok, activity} <- insert(data, local, false),
 590          {:ok, object, _create_activity} <- Object.delete(object),
 591          stream_out_participations(object, user),
 592          _ <- decrease_replies_count_if_reply(object),
 593          {:ok, _actor} <- decrease_note_count_if_public(user, object),
 594          :ok <- maybe_federate(activity) do
 595       {:ok, activity}
 596     else
 597       {:error, error} ->
 598         Repo.rollback(error)
 599     end
 600   end
 601
 602   defp do_delete(%Object{data: %{"type" => "Tombstone", "id" => ap_id}}, _) do
 603     activity =
 604       ap_id
 605       |> Activity.Queries.by_object_id()
 606       |> Activity.Queries.by_type("Delete")
 607       |> Repo.one()
 608
 609     {:ok, activity}
 610   end
 611
 612   @spec block(User.t(), User.t(), String.t() | nil, boolean()) ::
 613           {:ok, Activity.t()} | {:error, any()}
 614   def block(blocker, blocked, activity_id \\ nil, local \\ true) do
 615     with {:ok, result} <-
 616            Repo.transaction(fn -> do_block(blocker, blocked, activity_id, local) end) do
 617       result
 618     end
 619   end
 620
 621   defp do_block(blocker, blocked, activity_id, local) do
 622     outgoing_blocks = Config.get([:activitypub, :outgoing_blocks])
 623     unfollow_blocked = Config.get([:activitypub, :unfollow_blocked])
 624
 625     if unfollow_blocked do
 626       follow_activity = fetch_latest_follow(blocker, blocked)
 627       if follow_activity, do: unfollow(blocker, blocked, nil, local)
 628     end
 629
 630     with true <- outgoing_blocks,
 631          block_data <- make_block_data(blocker, blocked, activity_id),
 632          {:ok, activity} <- insert(block_data, local),
 633          :ok <- maybe_federate(activity) do
 634       {:ok, activity}
 635     else
 636       {:error, error} -> Repo.rollback(error)
 637     end
 638   end
 639
 640   @spec unblock(User.t(), User.t(), String.t() | nil, boolean()) ::
 641           {:ok, Activity.t()} | {:error, any()} | nil
 642   def unblock(blocker, blocked, activity_id \\ nil, local \\ true) do
 643     with {:ok, result} <-
 644            Repo.transaction(fn -> do_unblock(blocker, blocked, activity_id, local) end) do
 645       result
 646     end
 647   end
 648
 649   defp do_unblock(blocker, blocked, activity_id, local) do
 650     with %Activity{} = block_activity <- fetch_latest_block(blocker, blocked),
 651          unblock_data <- make_unblock_data(blocker, blocked, block_activity, activity_id),
 652          {:ok, activity} <- insert(unblock_data, local),
 653          :ok <- maybe_federate(activity) do
 654       {:ok, activity}
 655     else
 656       nil -> nil
 657       {:error, error} -> Repo.rollback(error)
 658     end
 659   end
 660
 661   @spec flag(map()) :: {:ok, Activity.t()} | {:error, any()}
 662   def flag(
 663         %{
 664           actor: actor,
 665           context: _context,
 666           account: account,
 667           statuses: statuses,
 668           content: content
 669         } = params
 670       ) do
 671     # only accept false as false value
 672     local = !(params[:local] == false)
 673     forward = !(params[:forward] == false)
 674
 675     additional = params[:additional] || %{}
 676
 677     additional =
 678       if forward do
 679         Map.merge(additional, %{"to" => [], "cc" => [account.ap_id]})
 680       else
 681         Map.merge(additional, %{"to" => [], "cc" => []})
 682       end
 683
 684     with flag_data <- make_flag_data(params, additional),
 685          {:ok, activity} <- insert(flag_data, local),
 686          {:ok, stripped_activity} <- strip_report_status_data(activity),
 687          :ok <- maybe_federate(stripped_activity) do
 688       User.all_superusers()
 689       |> Enum.filter(fn user -> not is_nil(user.email) end)
 690       |> Enum.each(fn superuser ->
 691         superuser
 692         |> Pleroma.Emails.AdminEmail.report(actor, account, statuses, content)
 693         |> Pleroma.Emails.Mailer.deliver_async()
 694       end)
 695
 696       {:ok, activity}
 697     end
 698   end
 699
 700   @spec move(User.t(), User.t(), boolean()) :: {:ok, Activity.t()} | {:error, any()}
 701   def move(%User{} = origin, %User{} = target, local \\ true) do
 702     params = %{
 703       "type" => "Move",
 704       "actor" => origin.ap_id,
 705       "object" => origin.ap_id,
 706       "target" => target.ap_id
 707     }
 708
 709     with true <- origin.ap_id in target.also_known_as,
 710          {:ok, activity} <- insert(params, local) do
 711       maybe_federate(activity)
 712
 713       BackgroundWorker.enqueue("move_following", %{
 714         "origin_id" => origin.id,
 715         "target_id" => target.id
 716       })
 717
 718       {:ok, activity}
 719     else
 720       false -> {:error, "Target account must have the origin in `alsoKnownAs`"}
 721       err -> err
 722     end
 723   end
 724
 725   def fetch_activities_for_context_query(context, opts) do
 726     public = [Constants.as_public()]
 727
 728     recipients =
 729       if opts["user"],
 730         do: [opts["user"].ap_id | User.following(opts["user"])] ++ public,
 731         else: public
 732
 733     from(activity in Activity)
 734     |> maybe_preload_objects(opts)
 735     |> maybe_preload_bookmarks(opts)
 736     |> maybe_set_thread_muted_field(opts)
 737     |> restrict_blocked(opts)
 738     |> restrict_recipients(recipients, opts["user"])
 739     |> where(
 740       [activity],
 741       fragment(
 742         "?->>'type' = ? and ?->>'context' = ?",
 743         activity.data,
 744         "Create",
 745         activity.data,
 746         ^context
 747       )
 748     )
 749     |> exclude_poll_votes(opts)
 750     |> exclude_id(opts)
 751     |> order_by([activity], desc: activity.id)
 752   end
 753
 754   @spec fetch_activities_for_context(String.t(), keyword() | map()) :: [Activity.t()]
 755   def fetch_activities_for_context(context, opts \\ %{}) do
 756     context
 757     |> fetch_activities_for_context_query(opts)
 758     |> Repo.all()
 759   end
 760
 761   @spec fetch_latest_activity_id_for_context(String.t(), keyword() | map()) ::
 762           FlakeId.Ecto.CompatType.t() | nil
 763   def fetch_latest_activity_id_for_context(context, opts \\ %{}) do
 764     context
 765     |> fetch_activities_for_context_query(Map.merge(%{"skip_preload" => true}, opts))
 766     |> limit(1)
 767     |> select([a], a.id)
 768     |> Repo.one()
 769   end
 770
 771   @spec fetch_public_activities(map(), Pagination.type()) :: [Activity.t()]
 772   def fetch_public_activities(opts \\ %{}, pagination \\ :keyset) do
 773     opts = Map.drop(opts, ["user"])
 774
 775     [Constants.as_public()]
 776     |> fetch_activities_query(opts)
 777     |> restrict_unlisted()
 778     |> Pagination.fetch_paginated(opts, pagination)
 779   end
 780
 781   @valid_visibilities ~w[direct unlisted public private]
 782
 783   defp restrict_visibility(query, %{visibility: visibility})
 784        when is_list(visibility) do
 785     if Enum.all?(visibility, &(&1 in @valid_visibilities)) do
 786       query =
 787         from(
 788           a in query,
 789           where:
 790             fragment(
 791               "activity_visibility(?, ?, ?) = ANY (?)",
 792               a.actor,
 793               a.recipients,
 794               a.data,
 795               ^visibility
 796             )
 797         )
 798
 799       query
 800     else
 801       Logger.error("Could not restrict visibility to #{visibility}")
 802     end
 803   end
 804
 805   defp restrict_visibility(query, %{visibility: visibility})
 806        when visibility in @valid_visibilities do
 807     from(
 808       a in query,
 809       where:
 810         fragment("activity_visibility(?, ?, ?) = ?", a.actor, a.recipients, a.data, ^visibility)
 811     )
 812   end
 813
 814   defp restrict_visibility(_query, %{visibility: visibility})
 815        when visibility not in @valid_visibilities do
 816     Logger.error("Could not restrict visibility to #{visibility}")
 817   end
 818
 819   defp restrict_visibility(query, _visibility), do: query
 820
 821   defp exclude_visibility(query, %{"exclude_visibilities" => visibility})
 822        when is_list(visibility) do
 823     if Enum.all?(visibility, &(&1 in @valid_visibilities)) do
 824       from(
 825         a in query,
 826         where:
 827           not fragment(
 828             "activity_visibility(?, ?, ?) = ANY (?)",
 829             a.actor,
 830             a.recipients,
 831             a.data,
 832             ^visibility
 833           )
 834       )
 835     else
 836       Logger.error("Could not exclude visibility to #{visibility}")
 837       query
 838     end
 839   end
 840
 841   defp exclude_visibility(query, %{"exclude_visibilities" => visibility})
 842        when visibility in @valid_visibilities do
 843     from(
 844       a in query,
 845       where:
 846         not fragment(
 847           "activity_visibility(?, ?, ?) = ?",
 848           a.actor,
 849           a.recipients,
 850           a.data,
 851           ^visibility
 852         )
 853     )
 854   end
 855
 856   defp exclude_visibility(query, %{"exclude_visibilities" => visibility})
 857        when visibility not in @valid_visibilities do
 858     Logger.error("Could not exclude visibility to #{visibility}")
 859     query
 860   end
 861
 862   defp exclude_visibility(query, _visibility), do: query
 863
 864   defp restrict_thread_visibility(query, _, %{skip_thread_containment: true} = _),
 865     do: query
 866
 867   defp restrict_thread_visibility(
 868          query,
 869          %{"user" => %User{skip_thread_containment: true}},
 870          _
 871        ),
 872        do: query
 873
 874   defp restrict_thread_visibility(query, %{"user" => %User{ap_id: ap_id}}, _) do
 875     from(
 876       a in query,
 877       where: fragment("thread_visibility(?, (?)->>'id') = true", ^ap_id, a.data)
 878     )
 879   end
 880
 881   defp restrict_thread_visibility(query, _, _), do: query
 882
 883   def fetch_user_abstract_activities(user, reading_user, params \\ %{}) do
 884     params =
 885       params
 886       |> Map.put("user", reading_user)
 887       |> Map.put("actor_id", user.ap_id)
 888
 889     recipients =
 890       user_activities_recipients(%{
 891         "godmode" => params["godmode"],
 892         "reading_user" => reading_user
 893       })
 894
 895     fetch_activities(recipients, params)
 896     |> Enum.reverse()
 897   end
 898
 899   def fetch_user_activities(user, reading_user, params \\ %{}) do
 900     params =
 901       params
 902       |> Map.put("type", ["Create", "Announce"])
 903       |> Map.put("user", reading_user)
 904       |> Map.put("actor_id", user.ap_id)
 905       |> Map.put("pinned_activity_ids", user.pinned_activities)
 906
 907     params =
 908       if User.blocks?(reading_user, user) do
 909         params
 910       else
 911         params
 912         |> Map.put("blocking_user", reading_user)
 913         |> Map.put("muting_user", reading_user)
 914       end
 915
 916     recipients =
 917       user_activities_recipients(%{
 918         "godmode" => params["godmode"],
 919         "reading_user" => reading_user
 920       })
 921
 922     fetch_activities(recipients, params)
 923     |> Enum.reverse()
 924   end
 925
 926   def fetch_statuses(reading_user, params) do
 927     params =
 928       params
 929       |> Map.put("type", ["Create", "Announce"])
 930
 931     recipients =
 932       user_activities_recipients(%{
 933         "godmode" => params["godmode"],
 934         "reading_user" => reading_user
 935       })
 936
 937     fetch_activities(recipients, params, :offset)
 938     |> Enum.reverse()
 939   end
 940
 941   defp user_activities_recipients(%{"godmode" => true}) do
 942     []
 943   end
 944
 945   defp user_activities_recipients(%{"reading_user" => reading_user}) do
 946     if reading_user do
 947       [Constants.as_public()] ++ [reading_user.ap_id | User.following(reading_user)]
 948     else
 949       [Constants.as_public()]
 950     end
 951   end
 952
 953   defp restrict_since(query, %{"since_id" => ""}), do: query
 954
 955   defp restrict_since(query, %{"since_id" => since_id}) do
 956     from(activity in query, where: activity.id > ^since_id)
 957   end
 958
 959   defp restrict_since(query, _), do: query
 960
 961   defp restrict_tag_reject(_query, %{"tag_reject" => _tag_reject, "skip_preload" => true}) do
 962     raise "Can't use the child object without preloading!"
 963   end
 964
 965   defp restrict_tag_reject(query, %{"tag_reject" => tag_reject})
 966        when is_list(tag_reject) and tag_reject != [] do
 967     from(
 968       [_activity, object] in query,
 969       where: fragment("not (?)->'tag' \\?| (?)", object.data, ^tag_reject)
 970     )
 971   end
 972
 973   defp restrict_tag_reject(query, _), do: query
 974
 975   defp restrict_tag_all(_query, %{"tag_all" => _tag_all, "skip_preload" => true}) do
 976     raise "Can't use the child object without preloading!"
 977   end
 978
 979   defp restrict_tag_all(query, %{"tag_all" => tag_all})
 980        when is_list(tag_all) and tag_all != [] do
 981     from(
 982       [_activity, object] in query,
 983       where: fragment("(?)->'tag' \\?& (?)", object.data, ^tag_all)
 984     )
 985   end
 986
 987   defp restrict_tag_all(query, _), do: query
 988
 989   defp restrict_tag(_query, %{"tag" => _tag, "skip_preload" => true}) do
 990     raise "Can't use the child object without preloading!"
 991   end
 992
 993   defp restrict_tag(query, %{"tag" => tag}) when is_list(tag) do
 994     from(
 995       [_activity, object] in query,
 996       where: fragment("(?)->'tag' \\?| (?)", object.data, ^tag)
 997     )
 998   end
 999
1000   defp restrict_tag(query, %{"tag" => tag}) when is_binary(tag) do
1001     from(
1002       [_activity, object] in query,
1003       where: fragment("(?)->'tag' \\? (?)", object.data, ^tag)
1004     )
1005   end
1006
1007   defp restrict_tag(query, _), do: query
1008
1009   defp restrict_recipients(query, [], _user), do: query
1010
1011   defp restrict_recipients(query, recipients, nil) do
1012     from(activity in query, where: fragment("? && ?", ^recipients, activity.recipients))
1013   end
1014
1015   defp restrict_recipients(query, recipients, user) do
1016     from(
1017       activity in query,
1018       where: fragment("? && ?", ^recipients, activity.recipients),
1019       or_where: activity.actor == ^user.ap_id
1020     )
1021   end
1022
1023   defp restrict_local(query, %{"local_only" => true}) do
1024     from(activity in query, where: activity.local == true)
1025   end
1026
1027   defp restrict_local(query, _), do: query
1028
1029   defp restrict_actor(query, %{"actor_id" => actor_id}) do
1030     from(activity in query, where: activity.actor == ^actor_id)
1031   end
1032
1033   defp restrict_actor(query, _), do: query
1034
1035   defp restrict_type(query, %{"type" => type}) when is_binary(type) do
1036     from(activity in query, where: fragment("?->>'type' = ?", activity.data, ^type))
1037   end
1038
1039   defp restrict_type(query, %{"type" => type}) do
1040     from(activity in query, where: fragment("?->>'type' = ANY(?)", activity.data, ^type))
1041   end
1042
1043   defp restrict_type(query, _), do: query
1044
1045   defp restrict_state(query, %{"state" => state}) do
1046     from(activity in query, where: fragment("?->>'state' = ?", activity.data, ^state))
1047   end
1048
1049   defp restrict_state(query, _), do: query
1050
1051   defp restrict_favorited_by(query, %{"favorited_by" => ap_id}) do
1052     from(
1053       [_activity, object] in query,
1054       where: fragment("(?)->'likes' \\? (?)", object.data, ^ap_id)
1055     )
1056   end
1057
1058   defp restrict_favorited_by(query, _), do: query
1059
1060   defp restrict_media(_query, %{"only_media" => _val, "skip_preload" => true}) do
1061     raise "Can't use the child object without preloading!"
1062   end
1063
1064   defp restrict_media(query, %{"only_media" => val}) when val == "true" or val == "1" do
1065     from(
1066       [_activity, object] in query,
1067       where: fragment("not (?)->'attachment' = (?)", object.data, ^[])
1068     )
1069   end
1070
1071   defp restrict_media(query, _), do: query
1072
1073   defp restrict_replies(query, %{"exclude_replies" => val}) when val == "true" or val == "1" do
1074     from(
1075       [_activity, object] in query,
1076       where: fragment("?->>'inReplyTo' is null", object.data)
1077     )
1078   end
1079
1080   defp restrict_replies(query, _), do: query
1081
1082   defp restrict_reblogs(query, %{"exclude_reblogs" => val}) when val == "true" or val == "1" do
1083     from(activity in query, where: fragment("?->>'type' != 'Announce'", activity.data))
1084   end
1085
1086   defp restrict_reblogs(query, _), do: query
1087
1088   defp restrict_muted(query, %{"with_muted" => val}) when val in [true, "true", "1"], do: query
1089
1090   defp restrict_muted(query, %{"muting_user" => %User{} = user} = opts) do
1091     mutes = opts["muted_users_ap_ids"] || User.muted_users_ap_ids(user)
1092
1093     query =
1094       from([activity] in query,
1095         where: fragment("not (? = ANY(?))", activity.actor, ^mutes),
1096         where: fragment("not (?->'to' \\?| ?)", activity.data, ^mutes)
1097       )
1098
1099     unless opts["skip_preload"] do
1100       from([thread_mute: tm] in query, where: is_nil(tm.user_id))
1101     else
1102       query
1103     end
1104   end
1105
1106   defp restrict_muted(query, _), do: query
1107
1108   defp restrict_blocked(query, %{"blocking_user" => %User{} = user} = opts) do
1109     blocked_ap_ids = opts["blocked_users_ap_ids"] || User.blocked_users_ap_ids(user)
1110     domain_blocks = user.domain_blocks || []
1111
1112     following_ap_ids = User.get_friends_ap_ids(user)
1113
1114     query =
1115       if has_named_binding?(query, :object), do: query, else: Activity.with_joined_object(query)
1116
1117     from(
1118       [activity, object: o] in query,
1119       where: fragment("not (? = ANY(?))", activity.actor, ^blocked_ap_ids),
1120       where: fragment("not (? && ?)", activity.recipients, ^blocked_ap_ids),
1121       where:
1122         fragment(
1123           "not (?->>'type' = 'Announce' and ?->'to' \\?| ?)",
1124           activity.data,
1125           activity.data,
1126           ^blocked_ap_ids
1127         ),
1128       where:
1129         fragment(
1130           "(not (split_part(?, '/', 3) = ANY(?))) or ? = ANY(?)",
1131           activity.actor,
1132           ^domain_blocks,
1133           activity.actor,
1134           ^following_ap_ids
1135         ),
1136       where:
1137         fragment(
1138           "(not (split_part(?->>'actor', '/', 3) = ANY(?))) or (?->>'actor') = ANY(?)",
1139           o.data,
1140           ^domain_blocks,
1141           o.data,
1142           ^following_ap_ids
1143         )
1144     )
1145   end
1146
1147   defp restrict_blocked(query, _), do: query
1148
1149   defp restrict_unlisted(query) do
1150     from(
1151       activity in query,
1152       where:
1153         fragment(
1154           "not (coalesce(?->'cc', '{}'::jsonb) \\?| ?)",
1155           activity.data,
1156           ^[Constants.as_public()]
1157         )
1158     )
1159   end
1160
1161   defp restrict_pinned(query, %{"pinned" => "true", "pinned_activity_ids" => ids}) do
1162     from(activity in query, where: activity.id in ^ids)
1163   end
1164
1165   defp restrict_pinned(query, _), do: query
1166
1167   defp restrict_muted_reblogs(query, %{"muting_user" => %User{} = user} = opts) do
1168     muted_reblogs = opts["reblog_muted_users_ap_ids"] || User.reblog_muted_users_ap_ids(user)
1169
1170     from(
1171       activity in query,
1172       where:
1173         fragment(
1174           "not ( ?->>'type' = 'Announce' and ? = ANY(?))",
1175           activity.data,
1176           activity.actor,
1177           ^muted_reblogs
1178         )
1179     )
1180   end
1181
1182   defp restrict_muted_reblogs(query, _), do: query
1183
1184   defp restrict_instance(query, %{"instance" => instance}) do
1185     users =
1186       from(
1187         u in User,
1188         select: u.ap_id,
1189         where: fragment("? LIKE ?", u.nickname, ^"%@#{instance}")
1190       )
1191       |> Repo.all()
1192
1193     from(activity in query, where: activity.actor in ^users)
1194   end
1195
1196   defp restrict_instance(query, _), do: query
1197
1198   defp exclude_poll_votes(query, %{"include_poll_votes" => true}), do: query
1199
1200   defp exclude_poll_votes(query, _) do
1201     if has_named_binding?(query, :object) do
1202       from([activity, object: o] in query,
1203         where: fragment("not(?->>'type' = ?)", o.data, "Answer")
1204       )
1205     else
1206       query
1207     end
1208   end
1209
1210   defp exclude_id(query, %{"exclude_id" => id}) when is_binary(id) do
1211     from(activity in query, where: activity.id != ^id)
1212   end
1213
1214   defp exclude_id(query, _), do: query
1215
1216   defp maybe_preload_objects(query, %{"skip_preload" => true}), do: query
1217
1218   defp maybe_preload_objects(query, _) do
1219     query
1220     |> Activity.with_preloaded_object()
1221   end
1222
1223   defp maybe_preload_bookmarks(query, %{"skip_preload" => true}), do: query
1224
1225   defp maybe_preload_bookmarks(query, opts) do
1226     query
1227     |> Activity.with_preloaded_bookmark(opts["user"])
1228   end
1229
1230   defp maybe_preload_report_notes(query, %{"preload_report_notes" => true}) do
1231     query
1232     |> Activity.with_preloaded_report_notes()
1233   end
1234
1235   defp maybe_preload_report_notes(query, _), do: query
1236
1237   defp maybe_set_thread_muted_field(query, %{"skip_preload" => true}), do: query
1238
1239   defp maybe_set_thread_muted_field(query, opts) do
1240     query
1241     |> Activity.with_set_thread_muted_field(opts["muting_user"] || opts["user"])
1242   end
1243
1244   defp maybe_order(query, %{order: :desc}) do
1245     query
1246     |> order_by(desc: :id)
1247   end
1248
1249   defp maybe_order(query, %{order: :asc}) do
1250     query
1251     |> order_by(asc: :id)
1252   end
1253
1254   defp maybe_order(query, _), do: query
1255
1256   defp fetch_activities_query_ap_ids_ops(opts) do
1257     source_user = opts["muting_user"]
1258     ap_id_relationships = if source_user, do: [:mute, :reblog_mute], else: []
1259
1260     ap_id_relationships =
1261       ap_id_relationships ++
1262         if opts["blocking_user"] && opts["blocking_user"] == source_user do
1263           [:block]
1264         else
1265           []
1266         end
1267
1268     preloaded_ap_ids = User.outgoing_relationships_ap_ids(source_user, ap_id_relationships)
1269
1270     restrict_blocked_opts = Map.merge(%{"blocked_users_ap_ids" => preloaded_ap_ids[:block]}, opts)
1271     restrict_muted_opts = Map.merge(%{"muted_users_ap_ids" => preloaded_ap_ids[:mute]}, opts)
1272
1273     restrict_muted_reblogs_opts =
1274       Map.merge(%{"reblog_muted_users_ap_ids" => preloaded_ap_ids[:reblog_mute]}, opts)
1275
1276     {restrict_blocked_opts, restrict_muted_opts, restrict_muted_reblogs_opts}
1277   end
1278
1279   def fetch_activities_query(recipients, opts \\ %{}) do
1280     {restrict_blocked_opts, restrict_muted_opts, restrict_muted_reblogs_opts} =
1281       fetch_activities_query_ap_ids_ops(opts)
1282
1283     config = %{
1284       skip_thread_containment: Config.get([:instance, :skip_thread_containment])
1285     }
1286
1287     Activity
1288     |> maybe_preload_objects(opts)
1289     |> maybe_preload_bookmarks(opts)
1290     |> maybe_preload_report_notes(opts)
1291     |> maybe_set_thread_muted_field(opts)
1292     |> maybe_order(opts)
1293     |> restrict_recipients(recipients, opts["user"])
1294     |> restrict_tag(opts)
1295     |> restrict_tag_reject(opts)
1296     |> restrict_tag_all(opts)
1297     |> restrict_since(opts)
1298     |> restrict_local(opts)
1299     |> restrict_actor(opts)
1300     |> restrict_type(opts)
1301     |> restrict_state(opts)
1302     |> restrict_favorited_by(opts)
1303     |> restrict_blocked(restrict_blocked_opts)
1304     |> restrict_muted(restrict_muted_opts)
1305     |> restrict_media(opts)
1306     |> restrict_visibility(opts)
1307     |> restrict_thread_visibility(opts, config)
1308     |> restrict_replies(opts)
1309     |> restrict_reblogs(opts)
1310     |> restrict_pinned(opts)
1311     |> restrict_muted_reblogs(restrict_muted_reblogs_opts)
1312     |> restrict_instance(opts)
1313     |> Activity.restrict_deactivated_users()
1314     |> exclude_poll_votes(opts)
1315     |> exclude_visibility(opts)
1316   end
1317
1318   def fetch_activities(recipients, opts \\ %{}, pagination \\ :keyset) do
1319     list_memberships = Pleroma.List.memberships(opts["user"])
1320
1321     fetch_activities_query(recipients ++ list_memberships, opts)
1322     |> Pagination.fetch_paginated(opts, pagination)
1323     |> Enum.reverse()
1324     |> maybe_update_cc(list_memberships, opts["user"])
1325   end
1326
1327   @doc """
1328   Fetch favorites activities of user with order by sort adds to favorites
1329   """
1330   @spec fetch_favourites(User.t(), map(), Pagination.type()) :: list(Activity.t())
1331   def fetch_favourites(user, params \\ %{}, pagination \\ :keyset) do
1332     user.ap_id
1333     |> Activity.Queries.by_actor()
1334     |> Activity.Queries.by_type("Like")
1335     |> Activity.with_joined_object()
1336     |> Object.with_joined_activity()
1337     |> select([_like, object, activity], %{activity | object: object})
1338     |> order_by([like, _, _], desc: like.id)
1339     |> Pagination.fetch_paginated(
1340       Map.merge(params, %{"skip_order" => true}),
1341       pagination,
1342       :object_activity
1343     )
1344   end
1345
1346   defp maybe_update_cc(activities, list_memberships, %User{ap_id: user_ap_id})
1347        when is_list(list_memberships) and length(list_memberships) > 0 do
1348     Enum.map(activities, fn
1349       %{data: %{"bcc" => bcc}} = activity when is_list(bcc) and length(bcc) > 0 ->
1350         if Enum.any?(bcc, &(&1 in list_memberships)) do
1351           update_in(activity.data["cc"], &[user_ap_id | &1])
1352         else
1353           activity
1354         end
1355
1356       activity ->
1357         activity
1358     end)
1359   end
1360
1361   defp maybe_update_cc(activities, _, _), do: activities
1362
1363   def fetch_activities_bounded_query(query, recipients, recipients_with_public) do
1364     from(activity in query,
1365       where:
1366         fragment("? && ?", activity.recipients, ^recipients) or
1367           (fragment("? && ?", activity.recipients, ^recipients_with_public) and
1368              ^Constants.as_public() in activity.recipients)
1369     )
1370   end
1371
1372   def fetch_activities_bounded(
1373         recipients,
1374         recipients_with_public,
1375         opts \\ %{},
1376         pagination \\ :keyset
1377       ) do
1378     fetch_activities_query([], opts)
1379     |> fetch_activities_bounded_query(recipients, recipients_with_public)
1380     |> Pagination.fetch_paginated(opts, pagination)
1381     |> Enum.reverse()
1382   end
1383
1384   @spec upload(Upload.source(), keyword()) :: {:ok, Object.t()} | {:error, any()}
1385   def upload(file, opts \\ []) do
1386     with {:ok, data} <- Upload.store(file, opts) do
1387       obj_data =
1388         if opts[:actor] do
1389           Map.put(data, "actor", opts[:actor])
1390         else
1391           data
1392         end
1393
1394       Repo.insert(%Object{data: obj_data})
1395     end
1396   end
1397
1398   @spec get_actor_url(any()) :: binary() | nil
1399   defp get_actor_url(url) when is_binary(url), do: url
1400   defp get_actor_url(%{"href" => href}) when is_binary(href), do: href
1401
1402   defp get_actor_url(url) when is_list(url) do
1403     url
1404     |> List.first()
1405     |> get_actor_url()
1406   end
1407
1408   defp get_actor_url(_url), do: nil
1409
1410   defp object_to_user_data(data) do
1411     avatar =
1412       data["icon"]["url"] &&
1413         %{
1414           "type" => "Image",
1415           "url" => [%{"href" => data["icon"]["url"]}]
1416         }
1417
1418     banner =
1419       data["image"]["url"] &&
1420         %{
1421           "type" => "Image",
1422           "url" => [%{"href" => data["image"]["url"]}]
1423         }
1424
1425     fields =
1426       data
1427       |> Map.get("attachment", [])
1428       |> Enum.filter(fn %{"type" => t} -> t == "PropertyValue" end)
1429       |> Enum.map(fn fields -> Map.take(fields, ["name", "value"]) end)
1430
1431     emojis =
1432       data
1433       |> Map.get("tag", [])
1434       |> Enum.filter(fn
1435         %{"type" => "Emoji"} -> true
1436         _ -> false
1437       end)
1438       |> Enum.reduce(%{}, fn %{"icon" => %{"url" => url}, "name" => name}, acc ->
1439         Map.put(acc, String.trim(name, ":"), url)
1440       end)
1441
1442     locked = data["manuallyApprovesFollowers"] || false
1443     data = Transmogrifier.maybe_fix_user_object(data)
1444     discoverable = data["discoverable"] || false
1445     invisible = data["invisible"] || false
1446     actor_type = data["type"] || "Person"
1447
1448     public_key =
1449       if is_map(data["publicKey"]) && is_binary(data["publicKey"]["publicKeyPem"]) do
1450         data["publicKey"]["publicKeyPem"]
1451       else
1452         nil
1453       end
1454
1455     shared_inbox =
1456       if is_map(data["endpoints"]) && is_binary(data["endpoints"]["sharedInbox"]) do
1457         data["endpoints"]["sharedInbox"]
1458       else
1459         nil
1460       end
1461
1462     user_data = %{
1463       ap_id: data["id"],
1464       uri: get_actor_url(data["url"]),
1465       ap_enabled: true,
1466       banner: banner,
1467       fields: fields,
1468       emoji: emojis,
1469       locked: locked,
1470       discoverable: discoverable,
1471       invisible: invisible,
1472       avatar: avatar,
1473       name: data["name"],
1474       follower_address: data["followers"],
1475       following_address: data["following"],
1476       bio: data["summary"],
1477       actor_type: actor_type,
1478       also_known_as: Map.get(data, "alsoKnownAs", []),
1479       public_key: public_key,
1480       inbox: data["inbox"],
1481       shared_inbox: shared_inbox
1482     }
1483
1484     # nickname can be nil because of virtual actors
1485     user_data =
1486       if data["preferredUsername"] do
1487         Map.put(
1488           user_data,
1489           :nickname,
1490           "#{data["preferredUsername"]}@#{URI.parse(data["id"]).host}"
1491         )
1492       else
1493         Map.put(user_data, :nickname, nil)
1494       end
1495
1496     {:ok, user_data}
1497   end
1498
1499   def fetch_follow_information_for_user(user) do
1500     with {:ok, following_data} <-
1501            Fetcher.fetch_and_contain_remote_object_from_id(user.following_address),
1502          {:ok, hide_follows} <- collection_private(following_data),
1503          {:ok, followers_data} <-
1504            Fetcher.fetch_and_contain_remote_object_from_id(user.follower_address),
1505          {:ok, hide_followers} <- collection_private(followers_data) do
1506       {:ok,
1507        %{
1508          hide_follows: hide_follows,
1509          follower_count: normalize_counter(followers_data["totalItems"]),
1510          following_count: normalize_counter(following_data["totalItems"]),
1511          hide_followers: hide_followers
1512        }}
1513     else
1514       {:error, _} = e -> e
1515       e -> {:error, e}
1516     end
1517   end
1518
1519   defp normalize_counter(counter) when is_integer(counter), do: counter
1520   defp normalize_counter(_), do: 0
1521
1522   defp maybe_update_follow_information(data) do
1523     with {:enabled, true} <- {:enabled, Config.get([:instance, :external_user_synchronization])},
1524          {:ok, info} <- fetch_follow_information_for_user(data) do
1525       info = Map.merge(data[:info] || %{}, info)
1526       Map.put(data, :info, info)
1527     else
1528       {:enabled, false} ->
1529         data
1530
1531       e ->
1532         Logger.error(
1533           "Follower/Following counter update for #{data.ap_id} failed.\n" <> inspect(e)
1534         )
1535
1536         data
1537     end
1538   end
1539
1540   defp collection_private(%{"first" => %{"type" => type}})
1541        when type in ["CollectionPage", "OrderedCollectionPage"],
1542        do: {:ok, false}
1543
1544   defp collection_private(%{"first" => first}) do
1545     with {:ok, %{"type" => type}} when type in ["CollectionPage", "OrderedCollectionPage"] <-
1546            Fetcher.fetch_and_contain_remote_object_from_id(first) do
1547       {:ok, false}
1548     else
1549       {:error, {:ok, %{status: code}}} when code in [401, 403] -> {:ok, true}
1550       {:error, _} = e -> e
1551       e -> {:error, e}
1552     end
1553   end
1554
1555   defp collection_private(_data), do: {:ok, true}
1556
1557   def user_data_from_user_object(data) do
1558     with {:ok, data} <- MRF.filter(data),
1559          {:ok, data} <- object_to_user_data(data) do
1560       {:ok, data}
1561     else
1562       e -> {:error, e}
1563     end
1564   end
1565
1566   def fetch_and_prepare_user_from_ap_id(ap_id) do
1567     with {:ok, data} <- Fetcher.fetch_and_contain_remote_object_from_id(ap_id),
1568          {:ok, data} <- user_data_from_user_object(data),
1569          data <- maybe_update_follow_information(data) do
1570       {:ok, data}
1571     else
1572       {:error, "Object has been deleted"} = e ->
1573         Logger.debug("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
1574         {:error, e}
1575
1576       e ->
1577         Logger.error("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
1578         {:error, e}
1579     end
1580   end
1581
1582   def make_user_from_ap_id(ap_id) do
1583     user = User.get_cached_by_ap_id(ap_id)
1584
1585     if user && !User.ap_enabled?(user) do
1586       Transmogrifier.upgrade_user_from_ap_id(ap_id)
1587     else
1588       with {:ok, data} <- fetch_and_prepare_user_from_ap_id(ap_id) do
1589         if user do
1590           user
1591           |> User.remote_user_changeset(data)
1592           |> User.update_and_set_cache()
1593         else
1594           data
1595           |> User.remote_user_changeset()
1596           |> Repo.insert()
1597           |> User.set_cache()
1598         end
1599       else
1600         e -> {:error, e}
1601       end
1602     end
1603   end
1604
1605   def make_user_from_nickname(nickname) do
1606     with {:ok, %{"ap_id" => ap_id}} when not is_nil(ap_id) <- WebFinger.finger(nickname) do
1607       make_user_from_ap_id(ap_id)
1608     else
1609       _e -> {:error, "No AP id in WebFinger"}
1610     end
1611   end
1612
1613   # filter out broken threads
1614   def contain_broken_threads(%Activity{} = activity, %User{} = user) do
1615     entire_thread_visible_for_user?(activity, user)
1616   end
1617
1618   # do post-processing on a specific activity
1619   def contain_activity(%Activity{} = activity, %User{} = user) do
1620     contain_broken_threads(activity, user)
1621   end
1622
1623   def fetch_direct_messages_query do
1624     Activity
1625     |> restrict_type(%{"type" => "Create"})
1626     |> restrict_visibility(%{visibility: "direct"})
1627     |> order_by([activity], asc: activity.id)
1628   end
1629 end