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