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