git.squeep.com Git - akkoma/blob - lib/pleroma/web/activity_pub/activity_pub.ex

   1 # Pleroma: A lightweight social networking server
   2 # Copyright © 2017-2019 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.Conversation
  10   alias Pleroma.Notification
  11   alias Pleroma.Object
  12   alias Pleroma.Object.Containment
  13   alias Pleroma.Object.Fetcher
  14   alias Pleroma.Pagination
  15   alias Pleroma.Repo
  16   alias Pleroma.Upload
  17   alias Pleroma.User
  18   alias Pleroma.Web.ActivityPub.MRF
  19   alias Pleroma.Web.ActivityPub.Transmogrifier
  20   alias Pleroma.Web.Streamer
  21   alias Pleroma.Web.WebFinger
  22   alias Pleroma.Workers.BackgroundWorker
  23
  24   import Ecto.Query
  25   import Pleroma.Web.ActivityPub.Utils
  26   import Pleroma.Web.ActivityPub.Visibility
  27
  28   require Logger
  29   require Pleroma.Constants
  30
  31   # For Announce activities, we filter the recipients based on following status for any actors
  32   # that match actual users.  See issue #164 for more information about why this is necessary.
  33   defp get_recipients(%{"type" => "Announce"} = data) do
  34     to = Map.get(data, "to", [])
  35     cc = Map.get(data, "cc", [])
  36     bcc = Map.get(data, "bcc", [])
  37     actor = User.get_cached_by_ap_id(data["actor"])
  38
  39     recipients =
  40       Enum.filter(Enum.concat([to, cc, bcc]), fn recipient ->
  41         case User.get_cached_by_ap_id(recipient) do
  42           nil -> true
  43           user -> User.following?(user, actor)
  44         end
  45       end)
  46
  47     {recipients, to, cc}
  48   end
  49
  50   defp get_recipients(%{"type" => "Create"} = data) do
  51     to = Map.get(data, "to", [])
  52     cc = Map.get(data, "cc", [])
  53     bcc = Map.get(data, "bcc", [])
  54     actor = Map.get(data, "actor", [])
  55     recipients = [to, cc, bcc, [actor]] |> Enum.concat() |> Enum.uniq()
  56     {recipients, to, cc}
  57   end
  58
  59   defp get_recipients(data) do
  60     to = Map.get(data, "to", [])
  61     cc = Map.get(data, "cc", [])
  62     bcc = Map.get(data, "bcc", [])
  63     recipients = Enum.concat([to, cc, bcc])
  64     {recipients, to, cc}
  65   end
  66
  67   defp check_actor_is_active(actor) do
  68     if not is_nil(actor) do
  69       with user <- User.get_cached_by_ap_id(actor),
  70            false <- user.info.deactivated do
  71         true
  72       else
  73         _e -> false
  74       end
  75     else
  76       true
  77     end
  78   end
  79
  80   defp check_remote_limit(%{"object" => %{"content" => content}}) when not is_nil(content) do
  81     limit = Config.get([:instance, :remote_limit])
  82     String.length(content) <= limit
  83   end
  84
  85   defp check_remote_limit(_), do: true
  86
  87   def increase_note_count_if_public(actor, object) do
  88     if is_public?(object), do: User.increase_note_count(actor), else: {:ok, actor}
  89   end
  90
  91   def decrease_note_count_if_public(actor, object) do
  92     if is_public?(object), do: User.decrease_note_count(actor), else: {:ok, actor}
  93   end
  94
  95   def increase_replies_count_if_reply(%{
  96         "object" => %{"inReplyTo" => reply_ap_id} = object,
  97         "type" => "Create"
  98       }) do
  99     if is_public?(object) do
 100       Object.increase_replies_count(reply_ap_id)
 101     end
 102   end
 103
 104   def increase_replies_count_if_reply(_create_data), do: :noop
 105
 106   def decrease_replies_count_if_reply(%Object{
 107         data: %{"inReplyTo" => reply_ap_id} = object
 108       }) do
 109     if is_public?(object) do
 110       Object.decrease_replies_count(reply_ap_id)
 111     end
 112   end
 113
 114   def decrease_replies_count_if_reply(_object), do: :noop
 115
 116   def increase_poll_votes_if_vote(%{
 117         "object" => %{"inReplyTo" => reply_ap_id, "name" => name},
 118         "type" => "Create"
 119       }) do
 120     Object.increase_vote_count(reply_ap_id, name)
 121   end
 122
 123   def increase_poll_votes_if_vote(_create_data), do: :noop
 124
 125   def insert(map, local \\ true, fake \\ false, bypass_actor_check \\ false) when is_map(map) do
 126     with nil <- Activity.normalize(map),
 127          map <- lazy_put_activity_defaults(map, fake),
 128          true <- bypass_actor_check || check_actor_is_active(map["actor"]),
 129          {_, true} <- {:remote_limit_error, check_remote_limit(map)},
 130          {:ok, map} <- MRF.filter(map),
 131          {recipients, _, _} = get_recipients(map),
 132          {:fake, false, map, recipients} <- {:fake, fake, map, recipients},
 133          :ok <- Containment.contain_child(map),
 134          {:ok, map, object} <- insert_full_object(map) do
 135       {:ok, activity} =
 136         Repo.insert(%Activity{
 137           data: map,
 138           local: local,
 139           actor: map["actor"],
 140           recipients: recipients
 141         })
 142
 143       # Splice in the child object if we have one.
 144       activity =
 145         if not is_nil(object) do
 146           Map.put(activity, :object, object)
 147         else
 148           activity
 149         end
 150
 151       BackgroundWorker.enqueue("fetch_data_for_activity", %{"activity_id" => activity.id})
 152
 153       Notification.create_notifications(activity)
 154
 155       participations =
 156         activity
 157         |> Conversation.create_or_bump_for()
 158         |> get_participations()
 159
 160       stream_out(activity)
 161       stream_out_participations(participations)
 162       {:ok, activity}
 163     else
 164       %Activity{} = activity ->
 165         {:ok, activity}
 166
 167       {:fake, true, map, recipients} ->
 168         activity = %Activity{
 169           data: map,
 170           local: local,
 171           actor: map["actor"],
 172           recipients: recipients,
 173           id: "pleroma:fakeid"
 174         }
 175
 176         Pleroma.Web.RichMedia.Helpers.fetch_data_for_activity(activity)
 177         {:ok, activity}
 178
 179       error ->
 180         {:error, error}
 181     end
 182   end
 183
 184   defp get_participations({:ok, %{participations: participations}}), do: participations
 185   defp get_participations(_), do: []
 186
 187   def stream_out_participations(participations) do
 188     participations =
 189       participations
 190       |> Repo.preload(:user)
 191
 192     Streamer.stream("participation", participations)
 193   end
 194
 195   def stream_out_participations(%Object{data: %{"context" => context}}, user) do
 196     with %Conversation{} = conversation <- Conversation.get_for_ap_id(context),
 197          conversation = Repo.preload(conversation, :participations),
 198          last_activity_id =
 199            fetch_latest_activity_id_for_context(conversation.ap_id, %{
 200              "user" => user,
 201              "blocking_user" => user
 202            }) do
 203       if last_activity_id do
 204         stream_out_participations(conversation.participations)
 205       end
 206     end
 207   end
 208
 209   def stream_out_participations(_, _), do: :noop
 210
 211   def stream_out(%Activity{data: %{"type" => data_type}} = activity)
 212       when data_type in ["Create", "Announce", "Delete"] do
 213     activity
 214     |> Topics.get_activity_topics()
 215     |> Streamer.stream(activity)
 216   end
 217
 218   def stream_out(_activity) do
 219     :noop
 220   end
 221
 222   def create(%{to: to, actor: actor, context: context, object: object} = params, fake \\ false) do
 223     additional = params[:additional] || %{}
 224     # only accept false as false value
 225     local = !(params[:local] == false)
 226     published = params[:published]
 227
 228     with create_data <-
 229            make_create_data(
 230              %{to: to, actor: actor, published: published, context: context, object: object},
 231              additional
 232            ),
 233          {:ok, activity} <- insert(create_data, local, fake),
 234          {:fake, false, activity} <- {:fake, fake, activity},
 235          _ <- increase_replies_count_if_reply(create_data),
 236          _ <- increase_poll_votes_if_vote(create_data),
 237          # Changing note count prior to enqueuing federation task in order to avoid
 238          # race conditions on updating user.info
 239          {:ok, _actor} <- increase_note_count_if_public(actor, activity),
 240          :ok <- maybe_federate(activity) do
 241       {:ok, activity}
 242     else
 243       {:fake, true, activity} ->
 244         {:ok, activity}
 245
 246       {:error, message} ->
 247         {:error, message}
 248     end
 249   end
 250
 251   def listen(%{to: to, actor: actor, context: context, object: object} = params) do
 252     additional = params[:additional] || %{}
 253     # only accept false as false value
 254     local = !(params[:local] == false)
 255     published = params[:published]
 256
 257     with listen_data <-
 258            make_listen_data(
 259              %{to: to, actor: actor, published: published, context: context, object: object},
 260              additional
 261            ),
 262          {:ok, activity} <- insert(listen_data, local),
 263          :ok <- maybe_federate(activity) do
 264       {:ok, activity}
 265     else
 266       {:error, message} ->
 267         {:error, message}
 268     end
 269   end
 270
 271   def accept(%{to: to, actor: actor, object: object} = params) do
 272     # only accept false as false value
 273     local = !(params[:local] == false)
 274
 275     with data <- %{"to" => to, "type" => "Accept", "actor" => actor.ap_id, "object" => object},
 276          {:ok, activity} <- insert(data, local),
 277          :ok <- maybe_federate(activity) do
 278       {:ok, activity}
 279     end
 280   end
 281
 282   def reject(%{to: to, actor: actor, object: object} = params) do
 283     # only accept false as false value
 284     local = !(params[:local] == false)
 285
 286     with data <- %{"to" => to, "type" => "Reject", "actor" => actor.ap_id, "object" => object},
 287          {:ok, activity} <- insert(data, local),
 288          :ok <- maybe_federate(activity) do
 289       {:ok, activity}
 290     end
 291   end
 292
 293   def update(%{to: to, cc: cc, actor: actor, object: object} = params) do
 294     # only accept false as false value
 295     local = !(params[:local] == false)
 296
 297     with data <- %{
 298            "to" => to,
 299            "cc" => cc,
 300            "type" => "Update",
 301            "actor" => actor,
 302            "object" => object
 303          },
 304          {:ok, activity} <- insert(data, local),
 305          :ok <- maybe_federate(activity) do
 306       {:ok, activity}
 307     end
 308   end
 309
 310   def react_with_emoji(user, object, emoji, options \\ []) do
 311     with local <- Keyword.get(options, :local, true),
 312          activity_id <- Keyword.get(options, :activity_id, nil),
 313          Pleroma.Emoji.is_unicode_emoji?(emoji),
 314          reaction_data <- make_emoji_reaction_data(user, object, emoji, activity_id),
 315          {:ok, activity} <- insert(reaction_data, local),
 316          {:ok, object} <- add_emoji_reaction_to_object(activity, object) do
 317       {:ok, activity, object}
 318     end
 319   end
 320
 321   # TODO: This is weird, maybe we shouldn't check here if we can make the activity.
 322   def like(
 323         %User{ap_id: ap_id} = user,
 324         %Object{data: %{"id" => _}} = object,
 325         activity_id \\ nil,
 326         local \\ true
 327       ) do
 328     with nil <- get_existing_like(ap_id, object),
 329          like_data <- make_like_data(user, object, activity_id),
 330          {:ok, activity} <- insert(like_data, local),
 331          {:ok, object} <- add_like_to_object(activity, object),
 332          :ok <- maybe_federate(activity) do
 333       {:ok, activity, object}
 334     else
 335       %Activity{} = activity -> {:ok, activity, object}
 336       error -> {:error, error}
 337     end
 338   end
 339
 340   def unlike(%User{} = actor, %Object{} = object, activity_id \\ nil, local \\ true) do
 341     with %Activity{} = like_activity <- get_existing_like(actor.ap_id, object),
 342          unlike_data <- make_unlike_data(actor, like_activity, activity_id),
 343          {:ok, unlike_activity} <- insert(unlike_data, local),
 344          {:ok, _activity} <- Repo.delete(like_activity),
 345          {:ok, object} <- remove_like_from_object(like_activity, object),
 346          :ok <- maybe_federate(unlike_activity) do
 347       {:ok, unlike_activity, like_activity, object}
 348     else
 349       _e -> {:ok, object}
 350     end
 351   end
 352
 353   def announce(
 354         %User{ap_id: _} = user,
 355         %Object{data: %{"id" => _}} = object,
 356         activity_id \\ nil,
 357         local \\ true,
 358         public \\ true
 359       ) do
 360     with true <- is_public?(object),
 361          announce_data <- make_announce_data(user, object, activity_id, public),
 362          {:ok, activity} <- insert(announce_data, local),
 363          {:ok, object} <- add_announce_to_object(activity, object),
 364          :ok <- maybe_federate(activity) do
 365       {:ok, activity, object}
 366     else
 367       error -> {:error, error}
 368     end
 369   end
 370
 371   def unannounce(
 372         %User{} = actor,
 373         %Object{} = object,
 374         activity_id \\ nil,
 375         local \\ true
 376       ) do
 377     with %Activity{} = announce_activity <- get_existing_announce(actor.ap_id, object),
 378          unannounce_data <- make_unannounce_data(actor, announce_activity, activity_id),
 379          {:ok, unannounce_activity} <- insert(unannounce_data, local),
 380          :ok <- maybe_federate(unannounce_activity),
 381          {:ok, _activity} <- Repo.delete(announce_activity),
 382          {:ok, object} <- remove_announce_from_object(announce_activity, object) do
 383       {:ok, unannounce_activity, object}
 384     else
 385       _e -> {:ok, object}
 386     end
 387   end
 388
 389   def follow(follower, followed, activity_id \\ nil, local \\ true) do
 390     with data <- make_follow_data(follower, followed, activity_id),
 391          {:ok, activity} <- insert(data, local),
 392          :ok <- maybe_federate(activity),
 393          _ <- User.set_follow_state_cache(follower.ap_id, followed.ap_id, activity.data["state"]) do
 394       {:ok, activity}
 395     end
 396   end
 397
 398   def unfollow(follower, followed, activity_id \\ nil, local \\ true) do
 399     with %Activity{} = follow_activity <- fetch_latest_follow(follower, followed),
 400          {:ok, follow_activity} <- update_follow_state(follow_activity, "cancelled"),
 401          unfollow_data <- make_unfollow_data(follower, followed, follow_activity, activity_id),
 402          {:ok, activity} <- insert(unfollow_data, local),
 403          :ok <- maybe_federate(activity) do
 404       {:ok, activity}
 405     end
 406   end
 407
 408   def delete(%User{ap_id: ap_id, follower_address: follower_address} = user) do
 409     with data <- %{
 410            "to" => [follower_address],
 411            "type" => "Delete",
 412            "actor" => ap_id,
 413            "object" => %{"type" => "Person", "id" => ap_id}
 414          },
 415          {:ok, activity} <- insert(data, true, true, true),
 416          :ok <- maybe_federate(activity) do
 417       {:ok, user}
 418     end
 419   end
 420
 421   def delete(%Object{data: %{"id" => id, "actor" => actor}} = object, local \\ true) do
 422     user = User.get_cached_by_ap_id(actor)
 423     to = (object.data["to"] || []) ++ (object.data["cc"] || [])
 424
 425     with {:ok, object, activity} <- Object.delete(object),
 426          data <- %{
 427            "type" => "Delete",
 428            "actor" => actor,
 429            "object" => id,
 430            "to" => to,
 431            "deleted_activity_id" => activity && activity.id
 432          },
 433          {:ok, activity} <- insert(data, local, false),
 434          stream_out_participations(object, user),
 435          _ <- decrease_replies_count_if_reply(object),
 436          # Changing note count prior to enqueuing federation task in order to avoid
 437          # race conditions on updating user.info
 438          {:ok, _actor} <- decrease_note_count_if_public(user, object),
 439          :ok <- maybe_federate(activity) do
 440       {:ok, activity}
 441     end
 442   end
 443
 444   @spec block(User.t(), User.t(), String.t() | nil, boolean) :: {:ok, Activity.t() | nil}
 445   def block(blocker, blocked, activity_id \\ nil, local \\ true) do
 446     outgoing_blocks = Config.get([:activitypub, :outgoing_blocks])
 447     unfollow_blocked = Config.get([:activitypub, :unfollow_blocked])
 448
 449     if unfollow_blocked do
 450       follow_activity = fetch_latest_follow(blocker, blocked)
 451       if follow_activity, do: unfollow(blocker, blocked, nil, local)
 452     end
 453
 454     with true <- outgoing_blocks,
 455          block_data <- make_block_data(blocker, blocked, activity_id),
 456          {:ok, activity} <- insert(block_data, local),
 457          :ok <- maybe_federate(activity) do
 458       {:ok, activity}
 459     else
 460       _e -> {:ok, nil}
 461     end
 462   end
 463
 464   def unblock(blocker, blocked, activity_id \\ nil, local \\ true) do
 465     with %Activity{} = block_activity <- fetch_latest_block(blocker, blocked),
 466          unblock_data <- make_unblock_data(blocker, blocked, block_activity, activity_id),
 467          {:ok, activity} <- insert(unblock_data, local),
 468          :ok <- maybe_federate(activity) do
 469       {:ok, activity}
 470     end
 471   end
 472
 473   @spec flag(map()) :: {:ok, Activity.t()} | any
 474   def flag(
 475         %{
 476           actor: actor,
 477           context: _context,
 478           account: account,
 479           statuses: statuses,
 480           content: content
 481         } = params
 482       ) do
 483     # only accept false as false value
 484     local = !(params[:local] == false)
 485     forward = !(params[:forward] == false)
 486
 487     additional = params[:additional] || %{}
 488
 489     additional =
 490       if forward do
 491         Map.merge(additional, %{"to" => [], "cc" => [account.ap_id]})
 492       else
 493         Map.merge(additional, %{"to" => [], "cc" => []})
 494       end
 495
 496     with flag_data <- make_flag_data(params, additional),
 497          {:ok, activity} <- insert(flag_data, local),
 498          :ok <- maybe_federate(activity) do
 499       Enum.each(User.all_superusers(), fn superuser ->
 500         superuser
 501         |> Pleroma.Emails.AdminEmail.report(actor, account, statuses, content)
 502         |> Pleroma.Emails.Mailer.deliver_async()
 503       end)
 504
 505       {:ok, activity}
 506     end
 507   end
 508
 509   defp fetch_activities_for_context_query(context, opts) do
 510     public = [Pleroma.Constants.as_public()]
 511
 512     recipients =
 513       if opts["user"], do: [opts["user"].ap_id | opts["user"].following] ++ public, else: public
 514
 515     from(activity in Activity)
 516     |> maybe_preload_objects(opts)
 517     |> maybe_preload_bookmarks(opts)
 518     |> maybe_set_thread_muted_field(opts)
 519     |> restrict_blocked(opts)
 520     |> restrict_recipients(recipients, opts["user"])
 521     |> where(
 522       [activity],
 523       fragment(
 524         "?->>'type' = ? and ?->>'context' = ?",
 525         activity.data,
 526         "Create",
 527         activity.data,
 528         ^context
 529       )
 530     )
 531     |> exclude_poll_votes(opts)
 532     |> exclude_id(opts)
 533     |> order_by([activity], desc: activity.id)
 534   end
 535
 536   @spec fetch_activities_for_context(String.t(), keyword() | map()) :: [Activity.t()]
 537   def fetch_activities_for_context(context, opts \\ %{}) do
 538     context
 539     |> fetch_activities_for_context_query(opts)
 540     |> Repo.all()
 541   end
 542
 543   @spec fetch_latest_activity_id_for_context(String.t(), keyword() | map()) ::
 544           FlakeId.Ecto.CompatType.t() | nil
 545   def fetch_latest_activity_id_for_context(context, opts \\ %{}) do
 546     context
 547     |> fetch_activities_for_context_query(Map.merge(%{"skip_preload" => true}, opts))
 548     |> limit(1)
 549     |> select([a], a.id)
 550     |> Repo.one()
 551   end
 552
 553   def fetch_public_activities(opts \\ %{}, pagination \\ :keyset) do
 554     opts = Map.drop(opts, ["user"])
 555
 556     [Pleroma.Constants.as_public()]
 557     |> fetch_activities_query(opts)
 558     |> restrict_unlisted()
 559     |> Pagination.fetch_paginated(opts, pagination)
 560     |> Enum.reverse()
 561   end
 562
 563   @valid_visibilities ~w[direct unlisted public private]
 564
 565   defp restrict_visibility(query, %{visibility: visibility})
 566        when is_list(visibility) do
 567     if Enum.all?(visibility, &(&1 in @valid_visibilities)) do
 568       query =
 569         from(
 570           a in query,
 571           where:
 572             fragment(
 573               "activity_visibility(?, ?, ?) = ANY (?)",
 574               a.actor,
 575               a.recipients,
 576               a.data,
 577               ^visibility
 578             )
 579         )
 580
 581       query
 582     else
 583       Logger.error("Could not restrict visibility to #{visibility}")
 584     end
 585   end
 586
 587   defp restrict_visibility(query, %{visibility: visibility})
 588        when visibility in @valid_visibilities do
 589     from(
 590       a in query,
 591       where:
 592         fragment("activity_visibility(?, ?, ?) = ?", a.actor, a.recipients, a.data, ^visibility)
 593     )
 594   end
 595
 596   defp restrict_visibility(_query, %{visibility: visibility})
 597        when visibility not in @valid_visibilities do
 598     Logger.error("Could not restrict visibility to #{visibility}")
 599   end
 600
 601   defp restrict_visibility(query, _visibility), do: query
 602
 603   defp restrict_thread_visibility(query, _, %{skip_thread_containment: true} = _),
 604     do: query
 605
 606   defp restrict_thread_visibility(
 607          query,
 608          %{"user" => %User{info: %{skip_thread_containment: true}}},
 609          _
 610        ),
 611        do: query
 612
 613   defp restrict_thread_visibility(query, %{"user" => %User{ap_id: ap_id}}, _) do
 614     from(
 615       a in query,
 616       where: fragment("thread_visibility(?, (?)->>'id') = true", ^ap_id, a.data)
 617     )
 618   end
 619
 620   defp restrict_thread_visibility(query, _, _), do: query
 621
 622   def fetch_user_abstract_activities(user, reading_user, params \\ %{}) do
 623     params =
 624       params
 625       |> Map.put("user", reading_user)
 626       |> Map.put("actor_id", user.ap_id)
 627       |> Map.put("whole_db", true)
 628
 629     recipients =
 630       user_activities_recipients(%{
 631         "godmode" => params["godmode"],
 632         "reading_user" => reading_user
 633       })
 634
 635     fetch_activities(recipients, params)
 636     |> Enum.reverse()
 637   end
 638
 639   def fetch_user_activities(user, reading_user, params \\ %{}) do
 640     params =
 641       params
 642       |> Map.put("type", ["Create", "Announce"])
 643       |> Map.put("user", reading_user)
 644       |> Map.put("actor_id", user.ap_id)
 645       |> Map.put("whole_db", true)
 646       |> Map.put("pinned_activity_ids", user.info.pinned_activities)
 647
 648     recipients =
 649       user_activities_recipients(%{
 650         "godmode" => params["godmode"],
 651         "reading_user" => reading_user
 652       })
 653
 654     fetch_activities(recipients, params)
 655     |> Enum.reverse()
 656   end
 657
 658   defp user_activities_recipients(%{"godmode" => true}) do
 659     []
 660   end
 661
 662   defp user_activities_recipients(%{"reading_user" => reading_user}) do
 663     if reading_user do
 664       [Pleroma.Constants.as_public()] ++ [reading_user.ap_id | reading_user.following]
 665     else
 666       [Pleroma.Constants.as_public()]
 667     end
 668   end
 669
 670   defp restrict_since(query, %{"since_id" => ""}), do: query
 671
 672   defp restrict_since(query, %{"since_id" => since_id}) do
 673     from(activity in query, where: activity.id > ^since_id)
 674   end
 675
 676   defp restrict_since(query, _), do: query
 677
 678   defp restrict_tag_reject(_query, %{"tag_reject" => _tag_reject, "skip_preload" => true}) do
 679     raise "Can't use the child object without preloading!"
 680   end
 681
 682   defp restrict_tag_reject(query, %{"tag_reject" => tag_reject})
 683        when is_list(tag_reject) and tag_reject != [] do
 684     from(
 685       [_activity, object] in query,
 686       where: fragment("not (?)->'tag' \\?| (?)", object.data, ^tag_reject)
 687     )
 688   end
 689
 690   defp restrict_tag_reject(query, _), do: query
 691
 692   defp restrict_tag_all(_query, %{"tag_all" => _tag_all, "skip_preload" => true}) do
 693     raise "Can't use the child object without preloading!"
 694   end
 695
 696   defp restrict_tag_all(query, %{"tag_all" => tag_all})
 697        when is_list(tag_all) and tag_all != [] do
 698     from(
 699       [_activity, object] in query,
 700       where: fragment("(?)->'tag' \\?& (?)", object.data, ^tag_all)
 701     )
 702   end
 703
 704   defp restrict_tag_all(query, _), do: query
 705
 706   defp restrict_tag(_query, %{"tag" => _tag, "skip_preload" => true}) do
 707     raise "Can't use the child object without preloading!"
 708   end
 709
 710   defp restrict_tag(query, %{"tag" => tag}) when is_list(tag) do
 711     from(
 712       [_activity, object] in query,
 713       where: fragment("(?)->'tag' \\?| (?)", object.data, ^tag)
 714     )
 715   end
 716
 717   defp restrict_tag(query, %{"tag" => tag}) when is_binary(tag) do
 718     from(
 719       [_activity, object] in query,
 720       where: fragment("(?)->'tag' \\? (?)", object.data, ^tag)
 721     )
 722   end
 723
 724   defp restrict_tag(query, _), do: query
 725
 726   defp restrict_recipients(query, [], _user), do: query
 727
 728   defp restrict_recipients(query, recipients, nil) do
 729     from(activity in query, where: fragment("? && ?", ^recipients, activity.recipients))
 730   end
 731
 732   defp restrict_recipients(query, recipients, user) do
 733     from(
 734       activity in query,
 735       where: fragment("? && ?", ^recipients, activity.recipients),
 736       or_where: activity.actor == ^user.ap_id
 737     )
 738   end
 739
 740   defp restrict_local(query, %{"local_only" => true}) do
 741     from(activity in query, where: activity.local == true)
 742   end
 743
 744   defp restrict_local(query, _), do: query
 745
 746   defp restrict_actor(query, %{"actor_id" => actor_id}) do
 747     from(activity in query, where: activity.actor == ^actor_id)
 748   end
 749
 750   defp restrict_actor(query, _), do: query
 751
 752   defp restrict_type(query, %{"type" => type}) when is_binary(type) do
 753     from(activity in query, where: fragment("?->>'type' = ?", activity.data, ^type))
 754   end
 755
 756   defp restrict_type(query, %{"type" => type}) do
 757     from(activity in query, where: fragment("?->>'type' = ANY(?)", activity.data, ^type))
 758   end
 759
 760   defp restrict_type(query, _), do: query
 761
 762   defp restrict_state(query, %{"state" => state}) do
 763     from(activity in query, where: fragment("?->>'state' = ?", activity.data, ^state))
 764   end
 765
 766   defp restrict_state(query, _), do: query
 767
 768   defp restrict_favorited_by(query, %{"favorited_by" => ap_id}) do
 769     from(
 770       [_activity, object] in query,
 771       where: fragment("(?)->'likes' \\? (?)", object.data, ^ap_id)
 772     )
 773   end
 774
 775   defp restrict_favorited_by(query, _), do: query
 776
 777   defp restrict_media(_query, %{"only_media" => _val, "skip_preload" => true}) do
 778     raise "Can't use the child object without preloading!"
 779   end
 780
 781   defp restrict_media(query, %{"only_media" => val}) when val == "true" or val == "1" do
 782     from(
 783       [_activity, object] in query,
 784       where: fragment("not (?)->'attachment' = (?)", object.data, ^[])
 785     )
 786   end
 787
 788   defp restrict_media(query, _), do: query
 789
 790   defp restrict_replies(query, %{"exclude_replies" => val}) when val == "true" or val == "1" do
 791     from(
 792       activity in query,
 793       where: fragment("?->'object'->>'inReplyTo' is null", activity.data)
 794     )
 795   end
 796
 797   defp restrict_replies(query, _), do: query
 798
 799   defp restrict_reblogs(query, %{"exclude_reblogs" => val}) when val == "true" or val == "1" do
 800     from(activity in query, where: fragment("?->>'type' != 'Announce'", activity.data))
 801   end
 802
 803   defp restrict_reblogs(query, _), do: query
 804
 805   defp restrict_muted(query, %{"with_muted" => val}) when val in [true, "true", "1"], do: query
 806
 807   defp restrict_muted(query, %{"muting_user" => %User{info: info}} = opts) do
 808     mutes = info.mutes
 809
 810     query =
 811       from([activity] in query,
 812         where: fragment("not (? = ANY(?))", activity.actor, ^mutes),
 813         where: fragment("not (?->'to' \\?| ?)", activity.data, ^mutes)
 814       )
 815
 816     unless opts["skip_preload"] do
 817       from([thread_mute: tm] in query, where: is_nil(tm.user_id))
 818     else
 819       query
 820     end
 821   end
 822
 823   defp restrict_muted(query, _), do: query
 824
 825   defp restrict_blocked(query, %{"blocking_user" => %User{info: info}}) do
 826     blocks = info.blocks || []
 827     domain_blocks = info.domain_blocks || []
 828
 829     query =
 830       if has_named_binding?(query, :object), do: query, else: Activity.with_joined_object(query)
 831
 832     from(
 833       [activity, object: o] in query,
 834       where: fragment("not (? = ANY(?))", activity.actor, ^blocks),
 835       where: fragment("not (? && ?)", activity.recipients, ^blocks),
 836       where:
 837         fragment(
 838           "not (?->>'type' = 'Announce' and ?->'to' \\?| ?)",
 839           activity.data,
 840           activity.data,
 841           ^blocks
 842         ),
 843       where: fragment("not (split_part(?, '/', 3) = ANY(?))", activity.actor, ^domain_blocks),
 844       where: fragment("not (split_part(?->>'actor', '/', 3) = ANY(?))", o.data, ^domain_blocks)
 845     )
 846   end
 847
 848   defp restrict_blocked(query, _), do: query
 849
 850   defp restrict_unlisted(query) do
 851     from(
 852       activity in query,
 853       where:
 854         fragment(
 855           "not (coalesce(?->'cc', '{}'::jsonb) \\?| ?)",
 856           activity.data,
 857           ^[Pleroma.Constants.as_public()]
 858         )
 859     )
 860   end
 861
 862   defp restrict_pinned(query, %{"pinned" => "true", "pinned_activity_ids" => ids}) do
 863     from(activity in query, where: activity.id in ^ids)
 864   end
 865
 866   defp restrict_pinned(query, _), do: query
 867
 868   defp restrict_muted_reblogs(query, %{"muting_user" => %User{info: info}}) do
 869     muted_reblogs = info.muted_reblogs || []
 870
 871     from(
 872       activity in query,
 873       where:
 874         fragment(
 875           "not ( ?->>'type' = 'Announce' and ? = ANY(?))",
 876           activity.data,
 877           activity.actor,
 878           ^muted_reblogs
 879         )
 880     )
 881   end
 882
 883   defp restrict_muted_reblogs(query, _), do: query
 884
 885   defp exclude_poll_votes(query, %{"include_poll_votes" => true}), do: query
 886
 887   defp exclude_poll_votes(query, _) do
 888     if has_named_binding?(query, :object) do
 889       from([activity, object: o] in query,
 890         where: fragment("not(?->>'type' = ?)", o.data, "Answer")
 891       )
 892     else
 893       query
 894     end
 895   end
 896
 897   defp exclude_id(query, %{"exclude_id" => id}) when is_binary(id) do
 898     from(activity in query, where: activity.id != ^id)
 899   end
 900
 901   defp exclude_id(query, _), do: query
 902
 903   defp maybe_preload_objects(query, %{"skip_preload" => true}), do: query
 904
 905   defp maybe_preload_objects(query, _) do
 906     query
 907     |> Activity.with_preloaded_object()
 908   end
 909
 910   defp maybe_preload_bookmarks(query, %{"skip_preload" => true}), do: query
 911
 912   defp maybe_preload_bookmarks(query, opts) do
 913     query
 914     |> Activity.with_preloaded_bookmark(opts["user"])
 915   end
 916
 917   defp maybe_set_thread_muted_field(query, %{"skip_preload" => true}), do: query
 918
 919   defp maybe_set_thread_muted_field(query, opts) do
 920     query
 921     |> Activity.with_set_thread_muted_field(opts["muting_user"] || opts["user"])
 922   end
 923
 924   defp maybe_order(query, %{order: :desc}) do
 925     query
 926     |> order_by(desc: :id)
 927   end
 928
 929   defp maybe_order(query, %{order: :asc}) do
 930     query
 931     |> order_by(asc: :id)
 932   end
 933
 934   defp maybe_order(query, _), do: query
 935
 936   def fetch_activities_query(recipients, opts \\ %{}) do
 937     config = %{
 938       skip_thread_containment: Config.get([:instance, :skip_thread_containment])
 939     }
 940
 941     Activity
 942     |> maybe_preload_objects(opts)
 943     |> maybe_preload_bookmarks(opts)
 944     |> maybe_set_thread_muted_field(opts)
 945     |> maybe_order(opts)
 946     |> restrict_recipients(recipients, opts["user"])
 947     |> restrict_tag(opts)
 948     |> restrict_tag_reject(opts)
 949     |> restrict_tag_all(opts)
 950     |> restrict_since(opts)
 951     |> restrict_local(opts)
 952     |> restrict_actor(opts)
 953     |> restrict_type(opts)
 954     |> restrict_state(opts)
 955     |> restrict_favorited_by(opts)
 956     |> restrict_blocked(opts)
 957     |> restrict_muted(opts)
 958     |> restrict_media(opts)
 959     |> restrict_visibility(opts)
 960     |> restrict_thread_visibility(opts, config)
 961     |> restrict_replies(opts)
 962     |> restrict_reblogs(opts)
 963     |> restrict_pinned(opts)
 964     |> restrict_muted_reblogs(opts)
 965     |> Activity.restrict_deactivated_users()
 966     |> exclude_poll_votes(opts)
 967   end
 968
 969   def fetch_activities(recipients, opts \\ %{}, pagination \\ :keyset) do
 970     list_memberships = Pleroma.List.memberships(opts["user"])
 971
 972     fetch_activities_query(recipients ++ list_memberships, opts)
 973     |> Pagination.fetch_paginated(opts, pagination)
 974     |> Enum.reverse()
 975     |> maybe_update_cc(list_memberships, opts["user"])
 976   end
 977
 978   defp maybe_update_cc(activities, list_memberships, %User{ap_id: user_ap_id})
 979        when is_list(list_memberships) and length(list_memberships) > 0 do
 980     Enum.map(activities, fn
 981       %{data: %{"bcc" => bcc}} = activity when is_list(bcc) and length(bcc) > 0 ->
 982         if Enum.any?(bcc, &(&1 in list_memberships)) do
 983           update_in(activity.data["cc"], &[user_ap_id | &1])
 984         else
 985           activity
 986         end
 987
 988       activity ->
 989         activity
 990     end)
 991   end
 992
 993   defp maybe_update_cc(activities, _, _), do: activities
 994
 995   def fetch_activities_bounded_query(query, recipients, recipients_with_public) do
 996     from(activity in query,
 997       where:
 998         fragment("? && ?", activity.recipients, ^recipients) or
 999           (fragment("? && ?", activity.recipients, ^recipients_with_public) and
1000              ^Pleroma.Constants.as_public() in activity.recipients)
1001     )
1002   end
1003
1004   def fetch_activities_bounded(
1005         recipients,
1006         recipients_with_public,
1007         opts \\ %{},
1008         pagination \\ :keyset
1009       ) do
1010     fetch_activities_query([], opts)
1011     |> fetch_activities_bounded_query(recipients, recipients_with_public)
1012     |> Pagination.fetch_paginated(opts, pagination)
1013     |> Enum.reverse()
1014   end
1015
1016   def upload(file, opts \\ []) do
1017     with {:ok, data} <- Upload.store(file, opts) do
1018       obj_data =
1019         if opts[:actor] do
1020           Map.put(data, "actor", opts[:actor])
1021         else
1022           data
1023         end
1024
1025       Repo.insert(%Object{data: obj_data})
1026     end
1027   end
1028
1029   defp object_to_user_data(data) do
1030     avatar =
1031       data["icon"]["url"] &&
1032         %{
1033           "type" => "Image",
1034           "url" => [%{"href" => data["icon"]["url"]}]
1035         }
1036
1037     banner =
1038       data["image"]["url"] &&
1039         %{
1040           "type" => "Image",
1041           "url" => [%{"href" => data["image"]["url"]}]
1042         }
1043
1044     fields =
1045       data
1046       |> Map.get("attachment", [])
1047       |> Enum.filter(fn %{"type" => t} -> t == "PropertyValue" end)
1048       |> Enum.map(fn fields -> Map.take(fields, ["name", "value"]) end)
1049
1050     locked = data["manuallyApprovesFollowers"] || false
1051     data = Transmogrifier.maybe_fix_user_object(data)
1052     discoverable = data["discoverable"] || false
1053
1054     user_data = %{
1055       ap_id: data["id"],
1056       info: %{
1057         ap_enabled: true,
1058         source_data: data,
1059         banner: banner,
1060         fields: fields,
1061         locked: locked,
1062         discoverable: discoverable
1063       },
1064       avatar: avatar,
1065       name: data["name"],
1066       follower_address: data["followers"],
1067       following_address: data["following"],
1068       bio: data["summary"]
1069     }
1070
1071     # nickname can be nil because of virtual actors
1072     user_data =
1073       if data["preferredUsername"] do
1074         Map.put(
1075           user_data,
1076           :nickname,
1077           "#{data["preferredUsername"]}@#{URI.parse(data["id"]).host}"
1078         )
1079       else
1080         Map.put(user_data, :nickname, nil)
1081       end
1082
1083     {:ok, user_data}
1084   end
1085
1086   def fetch_follow_information_for_user(user) do
1087     with {:ok, following_data} <-
1088            Fetcher.fetch_and_contain_remote_object_from_id(user.following_address),
1089          following_count when is_integer(following_count) <- following_data["totalItems"],
1090          {:ok, hide_follows} <- collection_private(following_data),
1091          {:ok, followers_data} <-
1092            Fetcher.fetch_and_contain_remote_object_from_id(user.follower_address),
1093          followers_count when is_integer(followers_count) <- followers_data["totalItems"],
1094          {:ok, hide_followers} <- collection_private(followers_data) do
1095       {:ok,
1096        %{
1097          hide_follows: hide_follows,
1098          follower_count: followers_count,
1099          following_count: following_count,
1100          hide_followers: hide_followers
1101        }}
1102     else
1103       {:error, _} = e ->
1104         e
1105
1106       e ->
1107         {:error, e}
1108     end
1109   end
1110
1111   defp maybe_update_follow_information(data) do
1112     with {:enabled, true} <-
1113            {:enabled, Pleroma.Config.get([:instance, :external_user_synchronization])},
1114          {:ok, info} <- fetch_follow_information_for_user(data) do
1115       info = Map.merge(data.info, info)
1116       Map.put(data, :info, info)
1117     else
1118       {:enabled, false} ->
1119         data
1120
1121       e ->
1122         Logger.error(
1123           "Follower/Following counter update for #{data.ap_id} failed.\n" <> inspect(e)
1124         )
1125
1126         data
1127     end
1128   end
1129
1130   defp collection_private(data) do
1131     if is_map(data["first"]) and
1132          data["first"]["type"] in ["CollectionPage", "OrderedCollectionPage"] do
1133       {:ok, false}
1134     else
1135       with {:ok, %{"type" => type}} when type in ["CollectionPage", "OrderedCollectionPage"] <-
1136              Fetcher.fetch_and_contain_remote_object_from_id(data["first"]) do
1137         {:ok, false}
1138       else
1139         {:error, {:ok, %{status: code}}} when code in [401, 403] ->
1140           {:ok, true}
1141
1142         {:error, _} = e ->
1143           e
1144
1145         e ->
1146           {:error, e}
1147       end
1148     end
1149   end
1150
1151   def user_data_from_user_object(data) do
1152     with {:ok, data} <- MRF.filter(data),
1153          {:ok, data} <- object_to_user_data(data) do
1154       {:ok, data}
1155     else
1156       e -> {:error, e}
1157     end
1158   end
1159
1160   def fetch_and_prepare_user_from_ap_id(ap_id) do
1161     with {:ok, data} <- Fetcher.fetch_and_contain_remote_object_from_id(ap_id),
1162          {:ok, data} <- user_data_from_user_object(data),
1163          data <- maybe_update_follow_information(data) do
1164       {:ok, data}
1165     else
1166       e -> Logger.error("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
1167     end
1168   end
1169
1170   def make_user_from_ap_id(ap_id) do
1171     if _user = User.get_cached_by_ap_id(ap_id) do
1172       Transmogrifier.upgrade_user_from_ap_id(ap_id)
1173     else
1174       with {:ok, data} <- fetch_and_prepare_user_from_ap_id(ap_id) do
1175         User.insert_or_update_user(data)
1176       else
1177         e -> {:error, e}
1178       end
1179     end
1180   end
1181
1182   def make_user_from_nickname(nickname) do
1183     with {:ok, %{"ap_id" => ap_id}} when not is_nil(ap_id) <- WebFinger.finger(nickname) do
1184       make_user_from_ap_id(ap_id)
1185     else
1186       _e -> {:error, "No AP id in WebFinger"}
1187     end
1188   end
1189
1190   # filter out broken threads
1191   def contain_broken_threads(%Activity{} = activity, %User{} = user) do
1192     entire_thread_visible_for_user?(activity, user)
1193   end
1194
1195   # do post-processing on a specific activity
1196   def contain_activity(%Activity{} = activity, %User{} = user) do
1197     contain_broken_threads(activity, user)
1198   end
1199
1200   def fetch_direct_messages_query do
1201     Activity
1202     |> restrict_type(%{"type" => "Create"})
1203     |> restrict_visibility(%{visibility: "direct"})
1204     |> order_by([activity], asc: activity.id)
1205   end
1206 end