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