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