git.squeep.com Git - akkoma/blob - lib/pleroma/web/activity_pub/activity_pub.ex

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