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.Conversation
   8   alias Pleroma.Notification
   9   alias Pleroma.Object
  10   alias Pleroma.Object.Fetcher
  11   alias Pleroma.Pagination
  12   alias Pleroma.Repo
  13   alias Pleroma.Upload
  14   alias Pleroma.User
  15   alias Pleroma.Web.ActivityPub.MRF
  16   alias Pleroma.Web.ActivityPub.Transmogrifier
  17   alias Pleroma.Web.WebFinger
  18
  19   import Ecto.Query
  20   import Pleroma.Web.ActivityPub.Utils
  21   import Pleroma.Web.ActivityPub.Visibility
  22
  23   require Logger
  24
  25   # For Announce activities, we filter the recipients based on following status for any actors
  26   # that match actual users.  See issue #164 for more information about why this is necessary.
  27   defp get_recipients(%{"type" => "Announce"} = data) do
  28     to = Map.get(data, "to", [])
  29     cc = Map.get(data, "cc", [])
  30     bcc = Map.get(data, "bcc", [])
  31     actor = User.get_cached_by_ap_id(data["actor"])
  32
  33     recipients =
  34       Enum.filter(Enum.concat([to, cc, bcc]), fn recipient ->
  35         case User.get_cached_by_ap_id(recipient) do
  36           nil -> true
  37           user -> User.following?(user, actor)
  38         end
  39       end)
  40
  41     {recipients, to, cc}
  42   end
  43
  44   defp get_recipients(%{"type" => "Create"} = data) do
  45     to = Map.get(data, "to", [])
  46     cc = Map.get(data, "cc", [])
  47     bcc = Map.get(data, "bcc", [])
  48     actor = Map.get(data, "actor", [])
  49     recipients = [to, cc, bcc, [actor]] |> Enum.concat() |> Enum.uniq()
  50     {recipients, to, cc}
  51   end
  52
  53   defp get_recipients(data) do
  54     to = Map.get(data, "to", [])
  55     cc = Map.get(data, "cc", [])
  56     bcc = Map.get(data, "bcc", [])
  57     recipients = Enum.concat([to, cc, bcc])
  58     {recipients, to, cc}
  59   end
  60
  61   defp check_actor_is_active(actor) do
  62     if not is_nil(actor) do
  63       with user <- User.get_cached_by_ap_id(actor),
  64            false <- user.info.deactivated do
  65         :ok
  66       else
  67         _e -> :reject
  68       end
  69     else
  70       :ok
  71     end
  72   end
  73
  74   defp check_remote_limit(%{"object" => %{"content" => content}}) when not is_nil(content) do
  75     limit = Pleroma.Config.get([:instance, :remote_limit])
  76     String.length(content) <= limit
  77   end
  78
  79   defp check_remote_limit(_), do: true
  80
  81   def increase_note_count_if_public(actor, object) do
  82     if is_public?(object), do: User.increase_note_count(actor), else: {:ok, actor}
  83   end
  84
  85   def decrease_note_count_if_public(actor, object) do
  86     if is_public?(object), do: User.decrease_note_count(actor), else: {:ok, actor}
  87   end
  88
  89   def increase_replies_count_if_reply(%{
  90         "object" => %{"inReplyTo" => reply_ap_id} = object,
  91         "type" => "Create"
  92       }) do
  93     if is_public?(object) do
  94       Object.increase_replies_count(reply_ap_id)
  95     end
  96   end
  97
  98   def increase_replies_count_if_reply(_create_data), do: :noop
  99
 100   def decrease_replies_count_if_reply(%Object{
 101         data: %{"inReplyTo" => reply_ap_id} = object
 102       }) do
 103     if is_public?(object) do
 104       Object.decrease_replies_count(reply_ap_id)
 105     end
 106   end
 107
 108   def decrease_replies_count_if_reply(_object), do: :noop
 109
 110   def insert(map, local \\ true, fake \\ false) when is_map(map) do
 111     with nil <- Activity.normalize(map),
 112          map <- lazy_put_activity_defaults(map, fake),
 113          :ok <- check_actor_is_active(map["actor"]),
 114          {_, true} <- {:remote_limit_error, check_remote_limit(map)},
 115          {:ok, map} <- MRF.filter(map),
 116          {recipients, _, _} = get_recipients(map),
 117          {:fake, false, map, recipients} <- {:fake, fake, map, recipients},
 118          {:ok, map, object} <- insert_full_object(map) do
 119       {:ok, activity} =
 120         Repo.insert(%Activity{
 121           data: map,
 122           local: local,
 123           actor: map["actor"],
 124           recipients: recipients
 125         })
 126
 127       # Splice in the child object if we have one.
 128       activity =
 129         if !is_nil(object) do
 130           Map.put(activity, :object, object)
 131         else
 132           activity
 133         end
 134
 135       PleromaJobQueue.enqueue(:background, Pleroma.Web.RichMedia.Helpers, [:fetch, activity])
 136
 137       Notification.create_notifications(activity)
 138
 139       participations =
 140         activity
 141         |> Conversation.create_or_bump_for()
 142         |> get_participations()
 143
 144       stream_out(activity)
 145       stream_out_participations(participations)
 146       {:ok, activity}
 147     else
 148       %Activity{} = activity ->
 149         {:ok, activity}
 150
 151       {:fake, true, map, recipients} ->
 152         activity = %Activity{
 153           data: map,
 154           local: local,
 155           actor: map["actor"],
 156           recipients: recipients,
 157           id: "pleroma:fakeid"
 158         }
 159
 160         Pleroma.Web.RichMedia.Helpers.fetch_data_for_activity(activity)
 161         {:ok, activity}
 162
 163       error ->
 164         {:error, error}
 165     end
 166   end
 167
 168   defp get_participations({:ok, %{participations: participations}}), do: participations
 169   defp get_participations(_), do: []
 170
 171   def stream_out_participations(participations) do
 172     participations =
 173       participations
 174       |> Repo.preload(:user)
 175
 176     Enum.each(participations, fn participation ->
 177       Pleroma.Web.Streamer.stream("participation", participation)
 178     end)
 179   end
 180
 181   def stream_out(activity) do
 182     public = "https://www.w3.org/ns/activitystreams#Public"
 183
 184     if activity.data["type"] in ["Create", "Announce", "Delete"] do
 185       Pleroma.Web.Streamer.stream("user", activity)
 186       Pleroma.Web.Streamer.stream("list", activity)
 187
 188       if Enum.member?(activity.data["to"], public) do
 189         Pleroma.Web.Streamer.stream("public", activity)
 190
 191         if activity.local do
 192           Pleroma.Web.Streamer.stream("public:local", activity)
 193         end
 194
 195         if activity.data["type"] in ["Create"] do
 196           object = Object.normalize(activity)
 197
 198           object.data
 199           |> Map.get("tag", [])
 200           |> Enum.filter(fn tag -> is_bitstring(tag) end)
 201           |> Enum.each(fn tag -> Pleroma.Web.Streamer.stream("hashtag:" <> tag, activity) end)
 202
 203           if object.data["attachment"] != [] do
 204             Pleroma.Web.Streamer.stream("public:media", activity)
 205
 206             if activity.local do
 207               Pleroma.Web.Streamer.stream("public:local:media", activity)
 208             end
 209           end
 210         end
 211       else
 212         # TODO: Write test, replace with visibility test
 213         if !Enum.member?(activity.data["cc"] || [], public) &&
 214              !Enum.member?(
 215                activity.data["to"],
 216                User.get_cached_by_ap_id(activity.data["actor"]).follower_address
 217              ),
 218            do: Pleroma.Web.Streamer.stream("direct", activity)
 219       end
 220     end
 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          # Changing note count prior to enqueuing federation task in order to avoid
 238          # race conditions on updating user.info
 239          {:ok, _actor} <- increase_note_count_if_public(actor, activity),
 240          :ok <- maybe_federate(activity) do
 241       {:ok, activity}
 242     else
 243       {:fake, true, activity} ->
 244         {:ok, activity}
 245     end
 246   end
 247
 248   def accept(%{to: to, actor: actor, object: object} = params) do
 249     # only accept false as false value
 250     local = !(params[:local] == false)
 251
 252     with data <- %{"to" => to, "type" => "Accept", "actor" => actor.ap_id, "object" => object},
 253          {:ok, activity} <- insert(data, local),
 254          :ok <- maybe_federate(activity) do
 255       {:ok, activity}
 256     end
 257   end
 258
 259   def reject(%{to: to, actor: actor, object: object} = params) do
 260     # only accept false as false value
 261     local = !(params[:local] == false)
 262
 263     with data <- %{"to" => to, "type" => "Reject", "actor" => actor.ap_id, "object" => object},
 264          {:ok, activity} <- insert(data, local),
 265          :ok <- maybe_federate(activity) do
 266       {:ok, activity}
 267     end
 268   end
 269
 270   def update(%{to: to, cc: cc, actor: actor, object: object} = params) do
 271     # only accept false as false value
 272     local = !(params[:local] == false)
 273
 274     with data <- %{
 275            "to" => to,
 276            "cc" => cc,
 277            "type" => "Update",
 278            "actor" => actor,
 279            "object" => object
 280          },
 281          {:ok, activity} <- insert(data, local),
 282          :ok <- maybe_federate(activity) do
 283       {:ok, activity}
 284     end
 285   end
 286
 287   # TODO: This is weird, maybe we shouldn't check here if we can make the activity.
 288   def like(
 289         %User{ap_id: ap_id} = user,
 290         %Object{data: %{"id" => _}} = object,
 291         activity_id \\ nil,
 292         local \\ true
 293       ) do
 294     with nil <- get_existing_like(ap_id, object),
 295          like_data <- make_like_data(user, object, activity_id),
 296          {:ok, activity} <- insert(like_data, local),
 297          {:ok, object} <- add_like_to_object(activity, object),
 298          :ok <- maybe_federate(activity) do
 299       {:ok, activity, object}
 300     else
 301       %Activity{} = activity -> {:ok, activity, object}
 302       error -> {:error, error}
 303     end
 304   end
 305
 306   def unlike(
 307         %User{} = actor,
 308         %Object{} = object,
 309         activity_id \\ nil,
 310         local \\ true
 311       ) 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) do
 364       {:ok, activity}
 365     end
 366   end
 367
 368   def unfollow(follower, followed, activity_id \\ nil, local \\ true) do
 369     with %Activity{} = follow_activity <- fetch_latest_follow(follower, followed),
 370          {:ok, follow_activity} <- update_follow_state(follow_activity, "cancelled"),
 371          unfollow_data <- make_unfollow_data(follower, followed, follow_activity, activity_id),
 372          {:ok, activity} <- insert(unfollow_data, local),
 373          :ok <- maybe_federate(activity) do
 374       {:ok, activity}
 375     end
 376   end
 377
 378   def delete(%Object{data: %{"id" => id, "actor" => actor}} = object, local \\ true) do
 379     user = User.get_cached_by_ap_id(actor)
 380     to = (object.data["to"] || []) ++ (object.data["cc"] || [])
 381
 382     with {:ok, object, activity} <- Object.delete(object),
 383          data <- %{
 384            "type" => "Delete",
 385            "actor" => actor,
 386            "object" => id,
 387            "to" => to,
 388            "deleted_activity_id" => activity && activity.id
 389          },
 390          {:ok, activity} <- insert(data, local),
 391          _ <- decrease_replies_count_if_reply(object),
 392          # Changing note count prior to enqueuing federation task in order to avoid
 393          # race conditions on updating user.info
 394          {:ok, _actor} <- decrease_note_count_if_public(user, object),
 395          :ok <- maybe_federate(activity) do
 396       {:ok, activity}
 397     end
 398   end
 399
 400   def block(blocker, blocked, activity_id \\ nil, local \\ true) do
 401     ap_config = Application.get_env(:pleroma, :activitypub)
 402     unfollow_blocked = Keyword.get(ap_config, :unfollow_blocked)
 403     outgoing_blocks = Keyword.get(ap_config, :outgoing_blocks)
 404
 405     with true <- unfollow_blocked do
 406       follow_activity = fetch_latest_follow(blocker, blocked)
 407
 408       if follow_activity do
 409         unfollow(blocker, blocked, nil, local)
 410       end
 411     end
 412
 413     with true <- outgoing_blocks,
 414          block_data <- make_block_data(blocker, blocked, activity_id),
 415          {:ok, activity} <- insert(block_data, local),
 416          :ok <- maybe_federate(activity) do
 417       {:ok, activity}
 418     else
 419       _e -> {:ok, nil}
 420     end
 421   end
 422
 423   def unblock(blocker, blocked, activity_id \\ nil, local \\ true) do
 424     with %Activity{} = block_activity <- fetch_latest_block(blocker, blocked),
 425          unblock_data <- make_unblock_data(blocker, blocked, block_activity, activity_id),
 426          {:ok, activity} <- insert(unblock_data, local),
 427          :ok <- maybe_federate(activity) do
 428       {:ok, activity}
 429     end
 430   end
 431
 432   def flag(
 433         %{
 434           actor: actor,
 435           context: context,
 436           account: account,
 437           statuses: statuses,
 438           content: content
 439         } = params
 440       ) do
 441     # only accept false as false value
 442     local = !(params[:local] == false)
 443     forward = !(params[:forward] == false)
 444
 445     additional = params[:additional] || %{}
 446
 447     params = %{
 448       actor: actor,
 449       context: context,
 450       account: account,
 451       statuses: statuses,
 452       content: content
 453     }
 454
 455     additional =
 456       if forward do
 457         Map.merge(additional, %{"to" => [], "cc" => [account.ap_id]})
 458       else
 459         Map.merge(additional, %{"to" => [], "cc" => []})
 460       end
 461
 462     with flag_data <- make_flag_data(params, additional),
 463          {:ok, activity} <- insert(flag_data, local),
 464          :ok <- maybe_federate(activity) do
 465       Enum.each(User.all_superusers(), fn superuser ->
 466         superuser
 467         |> Pleroma.Emails.AdminEmail.report(actor, account, statuses, content)
 468         |> Pleroma.Emails.Mailer.deliver_async()
 469       end)
 470
 471       {:ok, activity}
 472     end
 473   end
 474
 475   defp fetch_activities_for_context_query(context, opts) do
 476     public = ["https://www.w3.org/ns/activitystreams#Public"]
 477
 478     recipients =
 479       if opts["user"], do: [opts["user"].ap_id | opts["user"].following] ++ public, else: public
 480
 481     from(activity in Activity)
 482     |> restrict_blocked(opts)
 483     |> restrict_recipients(recipients, opts["user"])
 484     |> where(
 485       [activity],
 486       fragment(
 487         "?->>'type' = ? and ?->>'context' = ?",
 488         activity.data,
 489         "Create",
 490         activity.data,
 491         ^context
 492       )
 493     )
 494     |> order_by([activity], desc: activity.id)
 495   end
 496
 497   @spec fetch_activities_for_context(String.t(), keyword() | map()) :: [Activity.t()]
 498   def fetch_activities_for_context(context, opts \\ %{}) do
 499     context
 500     |> fetch_activities_for_context_query(opts)
 501     |> Activity.with_preloaded_object()
 502     |> Repo.all()
 503   end
 504
 505   @spec fetch_latest_activity_id_for_context(String.t(), keyword() | map()) ::
 506           Pleroma.FlakeId.t() | nil
 507   def fetch_latest_activity_id_for_context(context, opts \\ %{}) do
 508     context
 509     |> fetch_activities_for_context_query(opts)
 510     |> limit(1)
 511     |> select([a], a.id)
 512     |> Repo.one()
 513   end
 514
 515   def fetch_public_activities(opts \\ %{}) do
 516     q = fetch_activities_query(["https://www.w3.org/ns/activitystreams#Public"], opts)
 517
 518     q
 519     |> restrict_unlisted()
 520     |> Pagination.fetch_paginated(opts)
 521     |> Enum.reverse()
 522   end
 523
 524   @valid_visibilities ~w[direct unlisted public private]
 525
 526   defp restrict_visibility(query, %{visibility: visibility})
 527        when is_list(visibility) do
 528     if Enum.all?(visibility, &(&1 in @valid_visibilities)) do
 529       query =
 530         from(
 531           a in query,
 532           where:
 533             fragment(
 534               "activity_visibility(?, ?, ?) = ANY (?)",
 535               a.actor,
 536               a.recipients,
 537               a.data,
 538               ^visibility
 539             )
 540         )
 541
 542       Ecto.Adapters.SQL.to_sql(:all, Repo, query)
 543
 544       query
 545     else
 546       Logger.error("Could not restrict visibility to #{visibility}")
 547     end
 548   end
 549
 550   defp restrict_visibility(query, %{visibility: visibility})
 551        when visibility in @valid_visibilities do
 552     query =
 553       from(
 554         a in query,
 555         where:
 556           fragment("activity_visibility(?, ?, ?) = ?", a.actor, a.recipients, a.data, ^visibility)
 557       )
 558
 559     Ecto.Adapters.SQL.to_sql(:all, Repo, query)
 560
 561     query
 562   end
 563
 564   defp restrict_visibility(_query, %{visibility: visibility})
 565        when visibility not in @valid_visibilities do
 566     Logger.error("Could not restrict visibility to #{visibility}")
 567   end
 568
 569   defp restrict_visibility(query, _visibility), do: query
 570
 571   def fetch_user_activities(user, reading_user, params \\ %{}) do
 572     params =
 573       params
 574       |> Map.put("type", ["Create", "Announce"])
 575       |> Map.put("actor_id", user.ap_id)
 576       |> Map.put("whole_db", true)
 577       |> Map.put("pinned_activity_ids", user.info.pinned_activities)
 578
 579     recipients =
 580       if reading_user do
 581         ["https://www.w3.org/ns/activitystreams#Public"] ++
 582           [reading_user.ap_id | reading_user.following]
 583       else
 584         ["https://www.w3.org/ns/activitystreams#Public"]
 585       end
 586
 587     fetch_activities(recipients, params)
 588     |> Enum.reverse()
 589   end
 590
 591   defp restrict_since(query, %{"since_id" => ""}), do: query
 592
 593   defp restrict_since(query, %{"since_id" => since_id}) do
 594     from(activity in query, where: activity.id > ^since_id)
 595   end
 596
 597   defp restrict_since(query, _), do: query
 598
 599   defp restrict_tag_reject(_query, %{"tag_reject" => _tag_reject, "skip_preload" => true}) do
 600     raise "Can't use the child object without preloading!"
 601   end
 602
 603   defp restrict_tag_reject(query, %{"tag_reject" => tag_reject})
 604        when is_list(tag_reject) and tag_reject != [] do
 605     from(
 606       [_activity, object] in query,
 607       where: fragment("not (?)->'tag' \\?| (?)", object.data, ^tag_reject)
 608     )
 609   end
 610
 611   defp restrict_tag_reject(query, _), do: query
 612
 613   defp restrict_tag_all(_query, %{"tag_all" => _tag_all, "skip_preload" => true}) do
 614     raise "Can't use the child object without preloading!"
 615   end
 616
 617   defp restrict_tag_all(query, %{"tag_all" => tag_all})
 618        when is_list(tag_all) and tag_all != [] do
 619     from(
 620       [_activity, object] in query,
 621       where: fragment("(?)->'tag' \\?& (?)", object.data, ^tag_all)
 622     )
 623   end
 624
 625   defp restrict_tag_all(query, _), do: query
 626
 627   defp restrict_tag(_query, %{"tag" => _tag, "skip_preload" => true}) do
 628     raise "Can't use the child object without preloading!"
 629   end
 630
 631   defp restrict_tag(query, %{"tag" => tag}) when is_list(tag) do
 632     from(
 633       [_activity, object] in query,
 634       where: fragment("(?)->'tag' \\?| (?)", object.data, ^tag)
 635     )
 636   end
 637
 638   defp restrict_tag(query, %{"tag" => tag}) when is_binary(tag) do
 639     from(
 640       [_activity, object] in query,
 641       where: fragment("(?)->'tag' \\? (?)", object.data, ^tag)
 642     )
 643   end
 644
 645   defp restrict_tag(query, _), do: query
 646
 647   defp restrict_to_cc(query, recipients_to, recipients_cc) do
 648     from(
 649       activity in query,
 650       where:
 651         fragment(
 652           "(?->'to' \\?| ?) or (?->'cc' \\?| ?)",
 653           activity.data,
 654           ^recipients_to,
 655           activity.data,
 656           ^recipients_cc
 657         )
 658     )
 659   end
 660
 661   defp restrict_recipients(query, [], _user), do: query
 662
 663   defp restrict_recipients(query, recipients, nil) do
 664     from(activity in query, where: fragment("? && ?", ^recipients, activity.recipients))
 665   end
 666
 667   defp restrict_recipients(query, recipients, user) do
 668     from(
 669       activity in query,
 670       where: fragment("? && ?", ^recipients, activity.recipients),
 671       or_where: activity.actor == ^user.ap_id
 672     )
 673   end
 674
 675   defp restrict_local(query, %{"local_only" => true}) do
 676     from(activity in query, where: activity.local == true)
 677   end
 678
 679   defp restrict_local(query, _), do: query
 680
 681   defp restrict_actor(query, %{"actor_id" => actor_id}) do
 682     from(activity in query, where: activity.actor == ^actor_id)
 683   end
 684
 685   defp restrict_actor(query, _), do: query
 686
 687   defp restrict_type(query, %{"type" => type}) when is_binary(type) do
 688     from(activity in query, where: fragment("?->>'type' = ?", activity.data, ^type))
 689   end
 690
 691   defp restrict_type(query, %{"type" => type}) do
 692     from(activity in query, where: fragment("?->>'type' = ANY(?)", activity.data, ^type))
 693   end
 694
 695   defp restrict_type(query, _), do: query
 696
 697   defp restrict_favorited_by(query, %{"favorited_by" => ap_id}) do
 698     from(
 699       activity in query,
 700       where: fragment(~s(? <@ (? #> '{"object","likes"}'\)), ^ap_id, activity.data)
 701     )
 702   end
 703
 704   defp restrict_favorited_by(query, _), do: query
 705
 706   defp restrict_media(_query, %{"only_media" => _val, "skip_preload" => true}) do
 707     raise "Can't use the child object without preloading!"
 708   end
 709
 710   defp restrict_media(query, %{"only_media" => val}) when val == "true" or val == "1" do
 711     from(
 712       [_activity, object] in query,
 713       where: fragment("not (?)->'attachment' = (?)", object.data, ^[])
 714     )
 715   end
 716
 717   defp restrict_media(query, _), do: query
 718
 719   defp restrict_replies(query, %{"exclude_replies" => val}) when val == "true" or val == "1" do
 720     from(
 721       activity in query,
 722       where: fragment("?->'object'->>'inReplyTo' is null", activity.data)
 723     )
 724   end
 725
 726   defp restrict_replies(query, _), do: query
 727
 728   defp restrict_reblogs(query, %{"exclude_reblogs" => val}) when val == "true" or val == "1" do
 729     from(activity in query, where: fragment("?->>'type' != 'Announce'", activity.data))
 730   end
 731
 732   defp restrict_reblogs(query, _), do: query
 733
 734   defp restrict_muted(query, %{"with_muted" => val}) when val in [true, "true", "1"], do: query
 735
 736   defp restrict_muted(query, %{"muting_user" => %User{info: info}}) do
 737     mutes = info.mutes
 738
 739     from(
 740       activity in query,
 741       where: fragment("not (? = ANY(?))", activity.actor, ^mutes),
 742       where: fragment("not (?->'to' \\?| ?)", activity.data, ^mutes)
 743     )
 744   end
 745
 746   defp restrict_muted(query, _), do: query
 747
 748   defp restrict_blocked(query, %{"blocking_user" => %User{info: info}}) do
 749     blocks = info.blocks || []
 750     domain_blocks = info.domain_blocks || []
 751
 752     from(
 753       activity in query,
 754       where: fragment("not (? = ANY(?))", activity.actor, ^blocks),
 755       where: fragment("not (? && ?)", activity.recipients, ^blocks),
 756       where:
 757         fragment(
 758           "not (?->>'type' = 'Announce' and ?->'to' \\?| ?)",
 759           activity.data,
 760           activity.data,
 761           ^blocks
 762         ),
 763       where: fragment("not (split_part(?, '/', 3) = ANY(?))", activity.actor, ^domain_blocks)
 764     )
 765   end
 766
 767   defp restrict_blocked(query, _), do: query
 768
 769   defp restrict_unlisted(query) do
 770     from(
 771       activity in query,
 772       where:
 773         fragment(
 774           "not (coalesce(?->'cc', '{}'::jsonb) \\?| ?)",
 775           activity.data,
 776           ^["https://www.w3.org/ns/activitystreams#Public"]
 777         )
 778     )
 779   end
 780
 781   defp restrict_pinned(query, %{"pinned" => "true", "pinned_activity_ids" => ids}) do
 782     from(activity in query, where: activity.id in ^ids)
 783   end
 784
 785   defp restrict_pinned(query, _), do: query
 786
 787   defp restrict_muted_reblogs(query, %{"muting_user" => %User{info: info}}) do
 788     muted_reblogs = info.muted_reblogs || []
 789
 790     from(
 791       activity in query,
 792       where:
 793         fragment(
 794           "not ( ?->>'type' = 'Announce' and ? = ANY(?))",
 795           activity.data,
 796           activity.actor,
 797           ^muted_reblogs
 798         )
 799     )
 800   end
 801
 802   defp restrict_muted_reblogs(query, _), do: query
 803
 804   defp maybe_preload_objects(query, %{"skip_preload" => true}), do: query
 805
 806   defp maybe_preload_objects(query, _) do
 807     query
 808     |> Activity.with_preloaded_object()
 809   end
 810
 811   defp maybe_preload_bookmarks(query, %{"skip_preload" => true}), do: query
 812
 813   defp maybe_preload_bookmarks(query, opts) do
 814     query
 815     |> Activity.with_preloaded_bookmark(opts["user"])
 816   end
 817
 818   defp maybe_order(query, %{order: :desc}) do
 819     query
 820     |> order_by(desc: :id)
 821   end
 822
 823   defp maybe_order(query, %{order: :asc}) do
 824     query
 825     |> order_by(asc: :id)
 826   end
 827
 828   defp maybe_order(query, _), do: query
 829
 830   def fetch_activities_query(recipients, opts \\ %{}) do
 831     Activity
 832     |> maybe_preload_objects(opts)
 833     |> maybe_preload_bookmarks(opts)
 834     |> maybe_order(opts)
 835     |> restrict_recipients(recipients, opts["user"])
 836     |> restrict_tag(opts)
 837     |> restrict_tag_reject(opts)
 838     |> restrict_tag_all(opts)
 839     |> restrict_since(opts)
 840     |> restrict_local(opts)
 841     |> restrict_actor(opts)
 842     |> restrict_type(opts)
 843     |> restrict_favorited_by(opts)
 844     |> restrict_blocked(opts)
 845     |> restrict_muted(opts)
 846     |> restrict_media(opts)
 847     |> restrict_visibility(opts)
 848     |> restrict_replies(opts)
 849     |> restrict_reblogs(opts)
 850     |> restrict_pinned(opts)
 851     |> restrict_muted_reblogs(opts)
 852     |> Activity.restrict_deactivated_users()
 853   end
 854
 855   def fetch_activities(recipients, opts \\ %{}) do
 856     list_memberships = Pleroma.List.memberships(opts["user"])
 857
 858     fetch_activities_query(recipients ++ list_memberships, opts)
 859     |> Pagination.fetch_paginated(opts)
 860     |> Enum.reverse()
 861     |> maybe_update_cc(list_memberships, opts["user"])
 862   end
 863
 864   defp maybe_update_cc(activities, [], _), do: activities
 865   defp maybe_update_cc(activities, _, nil), do: activities
 866
 867   defp maybe_update_cc(activities, list_memberships, user) do
 868     Enum.map(activities, fn
 869       %{data: %{"bcc" => bcc}} = activity when is_list(bcc) ->
 870         if Enum.any?(bcc, &(&1 in list_memberships)) do
 871           update_in(activity.data["cc"], &[user.ap_id | &1])
 872         else
 873           activity
 874         end
 875
 876       activity ->
 877         activity
 878     end)
 879   end
 880
 881   def fetch_activities_bounded(recipients_to, recipients_cc, opts \\ %{}) do
 882     fetch_activities_query([], opts)
 883     |> restrict_to_cc(recipients_to, recipients_cc)
 884     |> Pagination.fetch_paginated(opts)
 885     |> Enum.reverse()
 886   end
 887
 888   def upload(file, opts \\ []) do
 889     with {:ok, data} <- Upload.store(file, opts) do
 890       obj_data =
 891         if opts[:actor] do
 892           Map.put(data, "actor", opts[:actor])
 893         else
 894           data
 895         end
 896
 897       Repo.insert(%Object{data: obj_data})
 898     end
 899   end
 900
 901   def user_data_from_user_object(data) do
 902     avatar =
 903       data["icon"]["url"] &&
 904         %{
 905           "type" => "Image",
 906           "url" => [%{"href" => data["icon"]["url"]}]
 907         }
 908
 909     banner =
 910       data["image"]["url"] &&
 911         %{
 912           "type" => "Image",
 913           "url" => [%{"href" => data["image"]["url"]}]
 914         }
 915
 916     locked = data["manuallyApprovesFollowers"] || false
 917     data = Transmogrifier.maybe_fix_user_object(data)
 918
 919     user_data = %{
 920       ap_id: data["id"],
 921       info: %{
 922         "ap_enabled" => true,
 923         "source_data" => data,
 924         "banner" => banner,
 925         "locked" => locked
 926       },
 927       avatar: avatar,
 928       name: data["name"],
 929       follower_address: data["followers"],
 930       bio: data["summary"]
 931     }
 932
 933     # nickname can be nil because of virtual actors
 934     user_data =
 935       if data["preferredUsername"] do
 936         Map.put(
 937           user_data,
 938           :nickname,
 939           "#{data["preferredUsername"]}@#{URI.parse(data["id"]).host}"
 940         )
 941       else
 942         Map.put(user_data, :nickname, nil)
 943       end
 944
 945     {:ok, user_data}
 946   end
 947
 948   def fetch_and_prepare_user_from_ap_id(ap_id) do
 949     with {:ok, data} <- Fetcher.fetch_and_contain_remote_object_from_id(ap_id) do
 950       user_data_from_user_object(data)
 951     else
 952       e -> Logger.error("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
 953     end
 954   end
 955
 956   def make_user_from_ap_id(ap_id) do
 957     if _user = User.get_cached_by_ap_id(ap_id) do
 958       Transmogrifier.upgrade_user_from_ap_id(ap_id)
 959     else
 960       with {:ok, data} <- fetch_and_prepare_user_from_ap_id(ap_id) do
 961         User.insert_or_update_user(data)
 962       else
 963         e -> {:error, e}
 964       end
 965     end
 966   end
 967
 968   def make_user_from_nickname(nickname) do
 969     with {:ok, %{"ap_id" => ap_id}} when not is_nil(ap_id) <- WebFinger.finger(nickname) do
 970       make_user_from_ap_id(ap_id)
 971     else
 972       _e -> {:error, "No AP id in WebFinger"}
 973     end
 974   end
 975
 976   # filter out broken threads
 977   def contain_broken_threads(%Activity{} = activity, %User{} = user) do
 978     entire_thread_visible_for_user?(activity, user)
 979   end
 980
 981   # do post-processing on a specific activity
 982   def contain_activity(%Activity{} = activity, %User{} = user) do
 983     contain_broken_threads(activity, user)
 984   end
 985
 986   # do post-processing on a timeline
 987   def contain_timeline(timeline, user) do
 988     timeline
 989     |> Enum.filter(fn activity ->
 990       contain_activity(activity, user)
 991     end)
 992   end
 993 end