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