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

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