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