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