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