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