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