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