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       query
 543     else
 544       Logger.error("Could not restrict visibility to #{visibility}")
 545     end
 546   end
 547
 548   defp restrict_visibility(query, %{visibility: visibility})
 549        when visibility in @valid_visibilities do
 550     query =
 551       from(
 552         a in query,
 553         where:
 554           fragment("activity_visibility(?, ?, ?) = ?", a.actor, a.recipients, a.data, ^visibility)
 555       )
 556
 557     query
 558   end
 559
 560   defp restrict_visibility(_query, %{visibility: visibility})
 561        when visibility not in @valid_visibilities do
 562     Logger.error("Could not restrict visibility to #{visibility}")
 563   end
 564
 565   defp restrict_visibility(query, _visibility), do: query
 566
 567   defp restrict_thread_visibility(query, %{"user" => %User{ap_id: ap_id}}) do
 568     query =
 569       from(
 570         a in query,
 571         where: fragment("thread_visibility(?, (?)->>'id') = true", ^ap_id, a.data)
 572       )
 573
 574     query
 575   end
 576
 577   defp restrict_thread_visibility(query, _), do: query
 578
 579   def fetch_user_activities(user, reading_user, params \\ %{}) do
 580     params =
 581       params
 582       |> Map.put("type", ["Create", "Announce"])
 583       |> Map.put("actor_id", user.ap_id)
 584       |> Map.put("whole_db", true)
 585       |> Map.put("pinned_activity_ids", user.info.pinned_activities)
 586
 587     recipients =
 588       if reading_user do
 589         ["https://www.w3.org/ns/activitystreams#Public"] ++
 590           [reading_user.ap_id | reading_user.following]
 591       else
 592         ["https://www.w3.org/ns/activitystreams#Public"]
 593       end
 594
 595     fetch_activities(recipients, params)
 596     |> Enum.reverse()
 597   end
 598
 599   defp restrict_since(query, %{"since_id" => ""}), do: query
 600
 601   defp restrict_since(query, %{"since_id" => since_id}) do
 602     from(activity in query, where: activity.id > ^since_id)
 603   end
 604
 605   defp restrict_since(query, _), do: query
 606
 607   defp restrict_tag_reject(_query, %{"tag_reject" => _tag_reject, "skip_preload" => true}) do
 608     raise "Can't use the child object without preloading!"
 609   end
 610
 611   defp restrict_tag_reject(query, %{"tag_reject" => tag_reject})
 612        when is_list(tag_reject) and tag_reject != [] do
 613     from(
 614       [_activity, object] in query,
 615       where: fragment("not (?)->'tag' \\?| (?)", object.data, ^tag_reject)
 616     )
 617   end
 618
 619   defp restrict_tag_reject(query, _), do: query
 620
 621   defp restrict_tag_all(_query, %{"tag_all" => _tag_all, "skip_preload" => true}) do
 622     raise "Can't use the child object without preloading!"
 623   end
 624
 625   defp restrict_tag_all(query, %{"tag_all" => tag_all})
 626        when is_list(tag_all) and tag_all != [] do
 627     from(
 628       [_activity, object] in query,
 629       where: fragment("(?)->'tag' \\?& (?)", object.data, ^tag_all)
 630     )
 631   end
 632
 633   defp restrict_tag_all(query, _), do: query
 634
 635   defp restrict_tag(_query, %{"tag" => _tag, "skip_preload" => true}) do
 636     raise "Can't use the child object without preloading!"
 637   end
 638
 639   defp restrict_tag(query, %{"tag" => tag}) when is_list(tag) do
 640     from(
 641       [_activity, object] in query,
 642       where: fragment("(?)->'tag' \\?| (?)", object.data, ^tag)
 643     )
 644   end
 645
 646   defp restrict_tag(query, %{"tag" => tag}) when is_binary(tag) do
 647     from(
 648       [_activity, object] in query,
 649       where: fragment("(?)->'tag' \\? (?)", object.data, ^tag)
 650     )
 651   end
 652
 653   defp restrict_tag(query, _), do: query
 654
 655   defp restrict_to_cc(query, recipients_to, recipients_cc) do
 656     from(
 657       activity in query,
 658       where:
 659         fragment(
 660           "(?->'to' \\?| ?) or (?->'cc' \\?| ?)",
 661           activity.data,
 662           ^recipients_to,
 663           activity.data,
 664           ^recipients_cc
 665         )
 666     )
 667   end
 668
 669   defp restrict_recipients(query, [], _user), do: query
 670
 671   defp restrict_recipients(query, recipients, nil) do
 672     from(activity in query, where: fragment("? && ?", ^recipients, activity.recipients))
 673   end
 674
 675   defp restrict_recipients(query, recipients, user) do
 676     from(
 677       activity in query,
 678       where: fragment("? && ?", ^recipients, activity.recipients),
 679       or_where: activity.actor == ^user.ap_id
 680     )
 681   end
 682
 683   defp restrict_local(query, %{"local_only" => true}) do
 684     from(activity in query, where: activity.local == true)
 685   end
 686
 687   defp restrict_local(query, _), do: query
 688
 689   defp restrict_actor(query, %{"actor_id" => actor_id}) do
 690     from(activity in query, where: activity.actor == ^actor_id)
 691   end
 692
 693   defp restrict_actor(query, _), do: query
 694
 695   defp restrict_type(query, %{"type" => type}) when is_binary(type) do
 696     from(activity in query, where: fragment("?->>'type' = ?", activity.data, ^type))
 697   end
 698
 699   defp restrict_type(query, %{"type" => type}) do
 700     from(activity in query, where: fragment("?->>'type' = ANY(?)", activity.data, ^type))
 701   end
 702
 703   defp restrict_type(query, _), do: query
 704
 705   defp restrict_state(query, %{"state" => state}) do
 706     from(activity in query, where: fragment("?->>'state' = ?", activity.data, ^state))
 707   end
 708
 709   defp restrict_state(query, _), do: query
 710
 711   defp restrict_favorited_by(query, %{"favorited_by" => ap_id}) do
 712     from(
 713       activity in query,
 714       where: fragment(~s(? <@ (? #> '{"object","likes"}'\)), ^ap_id, activity.data)
 715     )
 716   end
 717
 718   defp restrict_favorited_by(query, _), do: query
 719
 720   defp restrict_media(_query, %{"only_media" => _val, "skip_preload" => true}) do
 721     raise "Can't use the child object without preloading!"
 722   end
 723
 724   defp restrict_media(query, %{"only_media" => val}) when val == "true" or val == "1" do
 725     from(
 726       [_activity, object] in query,
 727       where: fragment("not (?)->'attachment' = (?)", object.data, ^[])
 728     )
 729   end
 730
 731   defp restrict_media(query, _), do: query
 732
 733   defp restrict_replies(query, %{"exclude_replies" => val}) when val == "true" or val == "1" do
 734     from(
 735       activity in query,
 736       where: fragment("?->'object'->>'inReplyTo' is null", activity.data)
 737     )
 738   end
 739
 740   defp restrict_replies(query, _), do: query
 741
 742   defp restrict_reblogs(query, %{"exclude_reblogs" => val}) when val == "true" or val == "1" do
 743     from(activity in query, where: fragment("?->>'type' != 'Announce'", activity.data))
 744   end
 745
 746   defp restrict_reblogs(query, _), do: query
 747
 748   defp restrict_muted(query, %{"with_muted" => val}) when val in [true, "true", "1"], do: query
 749
 750   defp restrict_muted(query, %{"muting_user" => %User{info: info}}) do
 751     mutes = info.mutes
 752
 753     from(
 754       activity in query,
 755       where: fragment("not (? = ANY(?))", activity.actor, ^mutes),
 756       where: fragment("not (?->'to' \\?| ?)", activity.data, ^mutes)
 757     )
 758   end
 759
 760   defp restrict_muted(query, _), do: query
 761
 762   defp restrict_blocked(query, %{"blocking_user" => %User{info: info}}) do
 763     blocks = info.blocks || []
 764     domain_blocks = info.domain_blocks || []
 765
 766     query =
 767       if has_named_binding?(query, :object), do: query, else: Activity.with_joined_object(query)
 768
 769     from(
 770       [activity, object: o] in query,
 771       where: fragment("not (? = ANY(?))", activity.actor, ^blocks),
 772       where: fragment("not (? && ?)", activity.recipients, ^blocks),
 773       where:
 774         fragment(
 775           "not (?->>'type' = 'Announce' and ?->'to' \\?| ?)",
 776           activity.data,
 777           activity.data,
 778           ^blocks
 779         ),
 780       where: fragment("not (split_part(?, '/', 3) = ANY(?))", activity.actor, ^domain_blocks),
 781       where: fragment("not (split_part(?->>'actor', '/', 3) = ANY(?))", o.data, ^domain_blocks)
 782     )
 783   end
 784
 785   defp restrict_blocked(query, _), do: query
 786
 787   defp restrict_unlisted(query) do
 788     from(
 789       activity in query,
 790       where:
 791         fragment(
 792           "not (coalesce(?->'cc', '{}'::jsonb) \\?| ?)",
 793           activity.data,
 794           ^["https://www.w3.org/ns/activitystreams#Public"]
 795         )
 796     )
 797   end
 798
 799   defp restrict_pinned(query, %{"pinned" => "true", "pinned_activity_ids" => ids}) do
 800     from(activity in query, where: activity.id in ^ids)
 801   end
 802
 803   defp restrict_pinned(query, _), do: query
 804
 805   defp restrict_muted_reblogs(query, %{"muting_user" => %User{info: info}}) do
 806     muted_reblogs = info.muted_reblogs || []
 807
 808     from(
 809       activity in query,
 810       where:
 811         fragment(
 812           "not ( ?->>'type' = 'Announce' and ? = ANY(?))",
 813           activity.data,
 814           activity.actor,
 815           ^muted_reblogs
 816         )
 817     )
 818   end
 819
 820   defp restrict_muted_reblogs(query, _), do: query
 821
 822   defp maybe_preload_objects(query, %{"skip_preload" => true}), do: query
 823
 824   defp maybe_preload_objects(query, _) do
 825     query
 826     |> Activity.with_preloaded_object()
 827   end
 828
 829   defp maybe_preload_bookmarks(query, %{"skip_preload" => true}), do: query
 830
 831   defp maybe_preload_bookmarks(query, opts) do
 832     query
 833     |> Activity.with_preloaded_bookmark(opts["user"])
 834   end
 835
 836   defp maybe_order(query, %{order: :desc}) do
 837     query
 838     |> order_by(desc: :id)
 839   end
 840
 841   defp maybe_order(query, %{order: :asc}) do
 842     query
 843     |> order_by(asc: :id)
 844   end
 845
 846   defp maybe_order(query, _), do: query
 847
 848   def fetch_activities_query(recipients, opts \\ %{}) do
 849     Activity
 850     |> maybe_preload_objects(opts)
 851     |> maybe_preload_bookmarks(opts)
 852     |> maybe_order(opts)
 853     |> restrict_recipients(recipients, opts["user"])
 854     |> restrict_tag(opts)
 855     |> restrict_tag_reject(opts)
 856     |> restrict_tag_all(opts)
 857     |> restrict_since(opts)
 858     |> restrict_local(opts)
 859     |> restrict_actor(opts)
 860     |> restrict_type(opts)
 861     |> restrict_state(opts)
 862     |> restrict_favorited_by(opts)
 863     |> restrict_blocked(opts)
 864     |> restrict_muted(opts)
 865     |> restrict_media(opts)
 866     |> restrict_visibility(opts)
 867     |> restrict_thread_visibility(opts)
 868     |> restrict_replies(opts)
 869     |> restrict_reblogs(opts)
 870     |> restrict_pinned(opts)
 871     |> restrict_muted_reblogs(opts)
 872     |> Activity.restrict_deactivated_users()
 873   end
 874
 875   def fetch_activities(recipients, opts \\ %{}) do
 876     list_memberships = Pleroma.List.memberships(opts["user"])
 877
 878     fetch_activities_query(recipients ++ list_memberships, opts)
 879     |> Pagination.fetch_paginated(opts)
 880     |> Enum.reverse()
 881     |> maybe_update_cc(list_memberships, opts["user"])
 882   end
 883
 884   defp maybe_update_cc(activities, [], _), do: activities
 885   defp maybe_update_cc(activities, _, nil), do: activities
 886
 887   defp maybe_update_cc(activities, list_memberships, user) do
 888     Enum.map(activities, fn
 889       %{data: %{"bcc" => bcc}} = activity when is_list(bcc) ->
 890         if Enum.any?(bcc, &(&1 in list_memberships)) do
 891           update_in(activity.data["cc"], &[user.ap_id | &1])
 892         else
 893           activity
 894         end
 895
 896       activity ->
 897         activity
 898     end)
 899   end
 900
 901   def fetch_activities_bounded(recipients_to, recipients_cc, opts \\ %{}) do
 902     fetch_activities_query([], opts)
 903     |> restrict_to_cc(recipients_to, recipients_cc)
 904     |> Pagination.fetch_paginated(opts)
 905     |> Enum.reverse()
 906   end
 907
 908   def upload(file, opts \\ []) do
 909     with {:ok, data} <- Upload.store(file, opts) do
 910       obj_data =
 911         if opts[:actor] do
 912           Map.put(data, "actor", opts[:actor])
 913         else
 914           data
 915         end
 916
 917       Repo.insert(%Object{data: obj_data})
 918     end
 919   end
 920
 921   def user_data_from_user_object(data) do
 922     avatar =
 923       data["icon"]["url"] &&
 924         %{
 925           "type" => "Image",
 926           "url" => [%{"href" => data["icon"]["url"]}]
 927         }
 928
 929     banner =
 930       data["image"]["url"] &&
 931         %{
 932           "type" => "Image",
 933           "url" => [%{"href" => data["image"]["url"]}]
 934         }
 935
 936     locked = data["manuallyApprovesFollowers"] || false
 937     data = Transmogrifier.maybe_fix_user_object(data)
 938
 939     user_data = %{
 940       ap_id: data["id"],
 941       info: %{
 942         "ap_enabled" => true,
 943         "source_data" => data,
 944         "banner" => banner,
 945         "locked" => locked
 946       },
 947       avatar: avatar,
 948       name: data["name"],
 949       follower_address: data["followers"],
 950       bio: data["summary"]
 951     }
 952
 953     # nickname can be nil because of virtual actors
 954     user_data =
 955       if data["preferredUsername"] do
 956         Map.put(
 957           user_data,
 958           :nickname,
 959           "#{data["preferredUsername"]}@#{URI.parse(data["id"]).host}"
 960         )
 961       else
 962         Map.put(user_data, :nickname, nil)
 963       end
 964
 965     {:ok, user_data}
 966   end
 967
 968   def fetch_and_prepare_user_from_ap_id(ap_id) do
 969     with {:ok, data} <- Fetcher.fetch_and_contain_remote_object_from_id(ap_id) do
 970       user_data_from_user_object(data)
 971     else
 972       e -> Logger.error("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
 973     end
 974   end
 975
 976   def make_user_from_ap_id(ap_id) do
 977     if _user = User.get_cached_by_ap_id(ap_id) do
 978       Transmogrifier.upgrade_user_from_ap_id(ap_id)
 979     else
 980       with {:ok, data} <- fetch_and_prepare_user_from_ap_id(ap_id) do
 981         User.insert_or_update_user(data)
 982       else
 983         e -> {:error, e}
 984       end
 985     end
 986   end
 987
 988   def make_user_from_nickname(nickname) do
 989     with {:ok, %{"ap_id" => ap_id}} when not is_nil(ap_id) <- WebFinger.finger(nickname) do
 990       make_user_from_ap_id(ap_id)
 991     else
 992       _e -> {:error, "No AP id in WebFinger"}
 993     end
 994   end
 995
 996   # filter out broken threads
 997   def contain_broken_threads(%Activity{} = activity, %User{} = user) do
 998     entire_thread_visible_for_user?(activity, user)
 999   end
1000
1001   # do post-processing on a specific activity
1002   def contain_activity(%Activity{} = activity, %User{} = user) do
1003     contain_broken_threads(activity, user)
1004   end
1005
1006   def fetch_direct_messages_query do
1007     Activity
1008     |> restrict_type(%{"type" => "Create"})
1009     |> restrict_visibility(%{visibility: "direct"})
1010     |> order_by([activity], asc: activity.id)
1011   end
1012 end