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