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

   1 # Pleroma: A lightweight social networking server
   2 # Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
   3 # SPDX-License-Identifier: AGPL-3.0-only
   4
   5 defmodule Pleroma.Web.ActivityPub.ActivityPub do
   6   alias Pleroma.Activity
   7   alias Pleroma.Activity.Ir.Topics
   8   alias Pleroma.Config
   9   alias Pleroma.Conversation
  10   alias Pleroma.Notification
  11   alias Pleroma.Object
  12   alias Pleroma.Object.Containment
  13   alias Pleroma.Object.Fetcher
  14   alias Pleroma.Pagination
  15   alias Pleroma.Repo
  16   alias Pleroma.Upload
  17   alias Pleroma.User
  18   alias Pleroma.Web.ActivityPub.MRF
  19   alias Pleroma.Web.ActivityPub.Transmogrifier
  20   alias Pleroma.Web.ActivityPub.Utils
  21   alias Pleroma.Web.Streamer
  22   alias Pleroma.Web.WebFinger
  23
  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       PleromaJobQueue.enqueue(:background, Pleroma.Web.RichMedia.Helpers, [:fetch, activity])
 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     local = !(params[:local] == false)
 275     activity_id = params[:activity_id]
 276
 277     with data <- %{
 278            "to" => to,
 279            "cc" => cc,
 280            "type" => "Update",
 281            "actor" => actor,
 282            "object" => object
 283          },
 284          data <- Utils.maybe_put(data, "id", activity_id),
 285          {:ok, activity} <- insert(data, local),
 286          :ok <- maybe_federate(activity) do
 287       {:ok, activity}
 288     end
 289   end
 290
 291   # TODO: This is weird, maybe we shouldn't check here if we can make the activity.
 292   def like(
 293         %User{ap_id: ap_id} = user,
 294         %Object{data: %{"id" => _}} = object,
 295         activity_id \\ nil,
 296         local \\ true
 297       ) do
 298     with nil <- get_existing_like(ap_id, object),
 299          like_data <- make_like_data(user, object, activity_id),
 300          {:ok, activity} <- insert(like_data, local),
 301          {:ok, object} <- add_like_to_object(activity, object),
 302          :ok <- maybe_federate(activity) do
 303       {:ok, activity, object}
 304     else
 305       %Activity{} = activity -> {:ok, activity, object}
 306       error -> {:error, error}
 307     end
 308   end
 309
 310   def unlike(%User{} = actor, %Object{} = object, activity_id \\ nil, local \\ true) do
 311     with %Activity{} = like_activity <- get_existing_like(actor.ap_id, object),
 312          unlike_data <- make_unlike_data(actor, like_activity, activity_id),
 313          {:ok, unlike_activity} <- insert(unlike_data, local),
 314          {:ok, _activity} <- Repo.delete(like_activity),
 315          {:ok, object} <- remove_like_from_object(like_activity, object),
 316          :ok <- maybe_federate(unlike_activity) do
 317       {:ok, unlike_activity, like_activity, object}
 318     else
 319       _e -> {:ok, object}
 320     end
 321   end
 322
 323   def announce(
 324         %User{ap_id: _} = user,
 325         %Object{data: %{"id" => _}} = object,
 326         activity_id \\ nil,
 327         local \\ true,
 328         public \\ true
 329       ) do
 330     with true <- is_public?(object),
 331          announce_data <- make_announce_data(user, object, activity_id, public),
 332          {:ok, activity} <- insert(announce_data, local),
 333          {:ok, object} <- add_announce_to_object(activity, object),
 334          :ok <- maybe_federate(activity) do
 335       {:ok, activity, object}
 336     else
 337       error -> {:error, error}
 338     end
 339   end
 340
 341   def unannounce(
 342         %User{} = actor,
 343         %Object{} = object,
 344         activity_id \\ nil,
 345         local \\ true
 346       ) do
 347     with %Activity{} = announce_activity <- get_existing_announce(actor.ap_id, object),
 348          unannounce_data <- make_unannounce_data(actor, announce_activity, activity_id),
 349          {:ok, unannounce_activity} <- insert(unannounce_data, local),
 350          :ok <- maybe_federate(unannounce_activity),
 351          {:ok, _activity} <- Repo.delete(announce_activity),
 352          {:ok, object} <- remove_announce_from_object(announce_activity, object) do
 353       {:ok, unannounce_activity, object}
 354     else
 355       _e -> {:ok, object}
 356     end
 357   end
 358
 359   def follow(follower, followed, activity_id \\ nil, local \\ true) do
 360     with data <- make_follow_data(follower, followed, activity_id),
 361          {:ok, activity} <- insert(data, local),
 362          :ok <- maybe_federate(activity),
 363          _ <- User.set_follow_state_cache(follower.ap_id, followed.ap_id, activity.data["state"]) 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(%User{ap_id: ap_id, follower_address: follower_address} = user) do
 379     with data <- %{
 380            "to" => [follower_address],
 381            "type" => "Delete",
 382            "actor" => ap_id,
 383            "object" => %{"type" => "Person", "id" => ap_id}
 384          },
 385          {:ok, activity} <- insert(data, true, true, true),
 386          :ok <- maybe_federate(activity) do
 387       {:ok, user}
 388     end
 389   end
 390
 391   def delete(%Object{data: %{"id" => id, "actor" => actor}} = object, local \\ true) do
 392     user = User.get_cached_by_ap_id(actor)
 393     to = (object.data["to"] || []) ++ (object.data["cc"] || [])
 394
 395     with {:ok, object, activity} <- Object.delete(object),
 396          data <- %{
 397            "type" => "Delete",
 398            "actor" => actor,
 399            "object" => id,
 400            "to" => to,
 401            "deleted_activity_id" => activity && activity.id
 402          },
 403          {:ok, activity} <- insert(data, local, false),
 404          stream_out_participations(object, user),
 405          _ <- decrease_replies_count_if_reply(object),
 406          # Changing note count prior to enqueuing federation task in order to avoid
 407          # race conditions on updating user.info
 408          {:ok, _actor} <- decrease_note_count_if_public(user, object),
 409          :ok <- maybe_federate(activity) do
 410       {:ok, activity}
 411     end
 412   end
 413
 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   def flag(
 443         %{
 444           actor: actor,
 445           context: context,
 446           account: account,
 447           statuses: statuses,
 448           content: content
 449         } = params
 450       ) do
 451     # only accept false as false value
 452     local = !(params[:local] == false)
 453     forward = !(params[:forward] == false)
 454
 455     additional = params[:additional] || %{}
 456
 457     params = %{
 458       actor: actor,
 459       context: context,
 460       account: account,
 461       statuses: statuses,
 462       content: content
 463     }
 464
 465     additional =
 466       if forward do
 467         Map.merge(additional, %{"to" => [], "cc" => [account.ap_id]})
 468       else
 469         Map.merge(additional, %{"to" => [], "cc" => []})
 470       end
 471
 472     with flag_data <- make_flag_data(params, additional),
 473          {:ok, activity} <- insert(flag_data, local),
 474          :ok <- maybe_federate(activity) do
 475       Enum.each(User.all_superusers(), fn superuser ->
 476         superuser
 477         |> Pleroma.Emails.AdminEmail.report(actor, account, statuses, content)
 478         |> Pleroma.Emails.Mailer.deliver_async()
 479       end)
 480
 481       {:ok, activity}
 482     end
 483   end
 484
 485   defp fetch_activities_for_context_query(context, opts) do
 486     public = [Pleroma.Constants.as_public()]
 487
 488     recipients =
 489       if opts["user"], do: [opts["user"].ap_id | opts["user"].following] ++ public, else: public
 490
 491     from(activity in Activity)
 492     |> maybe_preload_objects(opts)
 493     |> maybe_preload_bookmarks(opts)
 494     |> maybe_set_thread_muted_field(opts)
 495     |> restrict_blocked(opts)
 496     |> restrict_recipients(recipients, opts["user"])
 497     |> where(
 498       [activity],
 499       fragment(
 500         "?->>'type' = ? and ?->>'context' = ?",
 501         activity.data,
 502         "Create",
 503         activity.data,
 504         ^context
 505       )
 506     )
 507     |> exclude_poll_votes(opts)
 508     |> exclude_id(opts)
 509     |> order_by([activity], desc: activity.id)
 510   end
 511
 512   @spec fetch_activities_for_context(String.t(), keyword() | map()) :: [Activity.t()]
 513   def fetch_activities_for_context(context, opts \\ %{}) do
 514     context
 515     |> fetch_activities_for_context_query(opts)
 516     |> Repo.all()
 517   end
 518
 519   @spec fetch_latest_activity_id_for_context(String.t(), keyword() | map()) ::
 520           FlakeId.Ecto.CompatType.t() | nil
 521   def fetch_latest_activity_id_for_context(context, opts \\ %{}) do
 522     context
 523     |> fetch_activities_for_context_query(Map.merge(%{"skip_preload" => true}, opts))
 524     |> limit(1)
 525     |> select([a], a.id)
 526     |> Repo.one()
 527   end
 528
 529   def fetch_public_activities(opts \\ %{}) do
 530     opts = Map.drop(opts, ["user"])
 531
 532     [Pleroma.Constants.as_public()]
 533     |> fetch_activities_query(opts)
 534     |> restrict_unlisted()
 535     |> Pagination.fetch_paginated(opts)
 536     |> Enum.reverse()
 537   end
 538
 539   @valid_visibilities ~w[direct unlisted public private]
 540
 541   defp restrict_visibility(query, %{visibility: visibility})
 542        when is_list(visibility) do
 543     if Enum.all?(visibility, &(&1 in @valid_visibilities)) do
 544       query =
 545         from(
 546           a in query,
 547           where:
 548             fragment(
 549               "activity_visibility(?, ?, ?) = ANY (?)",
 550               a.actor,
 551               a.recipients,
 552               a.data,
 553               ^visibility
 554             )
 555         )
 556
 557       query
 558     else
 559       Logger.error("Could not restrict visibility to #{visibility}")
 560     end
 561   end
 562
 563   defp restrict_visibility(query, %{visibility: visibility})
 564        when visibility in @valid_visibilities do
 565     from(
 566       a in query,
 567       where:
 568         fragment("activity_visibility(?, ?, ?) = ?", a.actor, a.recipients, a.data, ^visibility)
 569     )
 570   end
 571
 572   defp restrict_visibility(_query, %{visibility: visibility})
 573        when visibility not in @valid_visibilities do
 574     Logger.error("Could not restrict visibility to #{visibility}")
 575   end
 576
 577   defp restrict_visibility(query, _visibility), do: query
 578
 579   defp restrict_thread_visibility(query, _, %{skip_thread_containment: true} = _),
 580     do: query
 581
 582   defp restrict_thread_visibility(
 583          query,
 584          %{"user" => %User{info: %{skip_thread_containment: true}}},
 585          _
 586        ),
 587        do: query
 588
 589   defp restrict_thread_visibility(query, %{"user" => %User{ap_id: ap_id}}, _) do
 590     from(
 591       a in query,
 592       where: fragment("thread_visibility(?, (?)->>'id') = true", ^ap_id, a.data)
 593     )
 594   end
 595
 596   defp restrict_thread_visibility(query, _, _), do: query
 597
 598   def fetch_user_activities(user, reading_user, params \\ %{}) do
 599     params =
 600       params
 601       |> Map.put("type", ["Create", "Announce"])
 602       |> Map.put("user", reading_user)
 603       |> Map.put("actor_id", user.ap_id)
 604       |> Map.put("whole_db", true)
 605       |> Map.put("pinned_activity_ids", user.info.pinned_activities)
 606
 607     recipients =
 608       user_activities_recipients(%{
 609         "godmode" => params["godmode"],
 610         "reading_user" => reading_user
 611       })
 612
 613     fetch_activities(recipients, params)
 614     |> Enum.reverse()
 615   end
 616
 617   defp user_activities_recipients(%{"godmode" => true}) do
 618     []
 619   end
 620
 621   defp user_activities_recipients(%{"reading_user" => reading_user}) do
 622     if reading_user do
 623       [Pleroma.Constants.as_public()] ++ [reading_user.ap_id | reading_user.following]
 624     else
 625       [Pleroma.Constants.as_public()]
 626     end
 627   end
 628
 629   defp restrict_since(query, %{"since_id" => ""}), do: query
 630
 631   defp restrict_since(query, %{"since_id" => since_id}) do
 632     from(activity in query, where: activity.id > ^since_id)
 633   end
 634
 635   defp restrict_since(query, _), do: query
 636
 637   defp restrict_tag_reject(_query, %{"tag_reject" => _tag_reject, "skip_preload" => true}) do
 638     raise "Can't use the child object without preloading!"
 639   end
 640
 641   defp restrict_tag_reject(query, %{"tag_reject" => tag_reject})
 642        when is_list(tag_reject) and tag_reject != [] do
 643     from(
 644       [_activity, object] in query,
 645       where: fragment("not (?)->'tag' \\?| (?)", object.data, ^tag_reject)
 646     )
 647   end
 648
 649   defp restrict_tag_reject(query, _), do: query
 650
 651   defp restrict_tag_all(_query, %{"tag_all" => _tag_all, "skip_preload" => true}) do
 652     raise "Can't use the child object without preloading!"
 653   end
 654
 655   defp restrict_tag_all(query, %{"tag_all" => tag_all})
 656        when is_list(tag_all) and tag_all != [] do
 657     from(
 658       [_activity, object] in query,
 659       where: fragment("(?)->'tag' \\?& (?)", object.data, ^tag_all)
 660     )
 661   end
 662
 663   defp restrict_tag_all(query, _), do: query
 664
 665   defp restrict_tag(_query, %{"tag" => _tag, "skip_preload" => true}) do
 666     raise "Can't use the child object without preloading!"
 667   end
 668
 669   defp restrict_tag(query, %{"tag" => tag}) when is_list(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, %{"tag" => tag}) when is_binary(tag) do
 677     from(
 678       [_activity, object] in query,
 679       where: fragment("(?)->'tag' \\? (?)", object.data, ^tag)
 680     )
 681   end
 682
 683   defp restrict_tag(query, _), do: query
 684
 685   defp restrict_recipients(query, [], _user), do: query
 686
 687   defp restrict_recipients(query, recipients, nil) do
 688     from(activity in query, where: fragment("? && ?", ^recipients, activity.recipients))
 689   end
 690
 691   defp restrict_recipients(query, recipients, user) do
 692     from(
 693       activity in query,
 694       where: fragment("? && ?", ^recipients, activity.recipients),
 695       or_where: activity.actor == ^user.ap_id
 696     )
 697   end
 698
 699   defp restrict_local(query, %{"local_only" => true}) do
 700     from(activity in query, where: activity.local == true)
 701   end
 702
 703   defp restrict_local(query, _), do: query
 704
 705   defp restrict_actor(query, %{"actor_id" => actor_id}) do
 706     from(activity in query, where: activity.actor == ^actor_id)
 707   end
 708
 709   defp restrict_actor(query, _), do: query
 710
 711   defp restrict_type(query, %{"type" => type}) when is_binary(type) do
 712     from(activity in query, where: fragment("?->>'type' = ?", activity.data, ^type))
 713   end
 714
 715   defp restrict_type(query, %{"type" => type}) do
 716     from(activity in query, where: fragment("?->>'type' = ANY(?)", activity.data, ^type))
 717   end
 718
 719   defp restrict_type(query, _), do: query
 720
 721   defp restrict_state(query, %{"state" => state}) do
 722     from(activity in query, where: fragment("?->>'state' = ?", activity.data, ^state))
 723   end
 724
 725   defp restrict_state(query, _), do: query
 726
 727   defp restrict_favorited_by(query, %{"favorited_by" => ap_id}) do
 728     from(
 729       [_activity, object] in query,
 730       where: fragment("(?)->'likes' \\? (?)", object.data, ^ap_id)
 731     )
 732   end
 733
 734   defp restrict_favorited_by(query, _), do: query
 735
 736   defp restrict_media(_query, %{"only_media" => _val, "skip_preload" => true}) do
 737     raise "Can't use the child object without preloading!"
 738   end
 739
 740   defp restrict_media(query, %{"only_media" => val}) when val == "true" or val == "1" do
 741     from(
 742       [_activity, object] in query,
 743       where: fragment("not (?)->'attachment' = (?)", object.data, ^[])
 744     )
 745   end
 746
 747   defp restrict_media(query, _), do: query
 748
 749   defp restrict_replies(query, %{"exclude_replies" => val}) when val == "true" or val == "1" do
 750     from(
 751       activity in query,
 752       where: fragment("?->'object'->>'inReplyTo' is null", activity.data)
 753     )
 754   end
 755
 756   defp restrict_replies(query, _), do: query
 757
 758   defp restrict_reblogs(query, %{"exclude_reblogs" => val}) when val == "true" or val == "1" do
 759     from(activity in query, where: fragment("?->>'type' != 'Announce'", activity.data))
 760   end
 761
 762   defp restrict_reblogs(query, _), do: query
 763
 764   defp restrict_muted(query, %{"with_muted" => val}) when val in [true, "true", "1"], do: query
 765
 766   defp restrict_muted(query, %{"muting_user" => %User{info: info}} = opts) do
 767     mutes = info.mutes
 768
 769     query =
 770       from([activity] in query,
 771         where: fragment("not (? = ANY(?))", activity.actor, ^mutes),
 772         where: fragment("not (?->'to' \\?| ?)", activity.data, ^mutes)
 773       )
 774
 775     unless opts["skip_preload"] do
 776       from([thread_mute: tm] in query, where: is_nil(tm.user_id))
 777     else
 778       query
 779     end
 780   end
 781
 782   defp restrict_muted(query, _), do: query
 783
 784   defp restrict_blocked(query, %{"blocking_user" => %User{info: info}}) do
 785     blocks = info.blocks || []
 786     domain_blocks = info.domain_blocks || []
 787
 788     query =
 789       if has_named_binding?(query, :object), do: query, else: Activity.with_joined_object(query)
 790
 791     from(
 792       [activity, object: o] in query,
 793       where: fragment("not (? = ANY(?))", activity.actor, ^blocks),
 794       where: fragment("not (? && ?)", activity.recipients, ^blocks),
 795       where:
 796         fragment(
 797           "not (?->>'type' = 'Announce' and ?->'to' \\?| ?)",
 798           activity.data,
 799           activity.data,
 800           ^blocks
 801         ),
 802       where: fragment("not (split_part(?, '/', 3) = ANY(?))", activity.actor, ^domain_blocks),
 803       where: fragment("not (split_part(?->>'actor', '/', 3) = ANY(?))", o.data, ^domain_blocks)
 804     )
 805   end
 806
 807   defp restrict_blocked(query, _), do: query
 808
 809   defp restrict_unlisted(query) do
 810     from(
 811       activity in query,
 812       where:
 813         fragment(
 814           "not (coalesce(?->'cc', '{}'::jsonb) \\?| ?)",
 815           activity.data,
 816           ^[Pleroma.Constants.as_public()]
 817         )
 818     )
 819   end
 820
 821   defp restrict_pinned(query, %{"pinned" => "true", "pinned_activity_ids" => ids}) do
 822     from(activity in query, where: activity.id in ^ids)
 823   end
 824
 825   defp restrict_pinned(query, _), do: query
 826
 827   defp restrict_muted_reblogs(query, %{"muting_user" => %User{info: info}}) do
 828     muted_reblogs = info.muted_reblogs || []
 829
 830     from(
 831       activity in query,
 832       where:
 833         fragment(
 834           "not ( ?->>'type' = 'Announce' and ? = ANY(?))",
 835           activity.data,
 836           activity.actor,
 837           ^muted_reblogs
 838         )
 839     )
 840   end
 841
 842   defp restrict_muted_reblogs(query, _), do: query
 843
 844   defp exclude_poll_votes(query, %{"include_poll_votes" => true}), do: query
 845
 846   defp exclude_poll_votes(query, _) do
 847     if has_named_binding?(query, :object) do
 848       from([activity, object: o] in query,
 849         where: fragment("not(?->>'type' = ?)", o.data, "Answer")
 850       )
 851     else
 852       query
 853     end
 854   end
 855
 856   defp exclude_id(query, %{"exclude_id" => id}) when is_binary(id) do
 857     from(activity in query, where: activity.id != ^id)
 858   end
 859
 860   defp exclude_id(query, _), do: query
 861
 862   defp maybe_preload_objects(query, %{"skip_preload" => true}), do: query
 863
 864   defp maybe_preload_objects(query, _) do
 865     query
 866     |> Activity.with_preloaded_object()
 867   end
 868
 869   defp maybe_preload_bookmarks(query, %{"skip_preload" => true}), do: query
 870
 871   defp maybe_preload_bookmarks(query, opts) do
 872     query
 873     |> Activity.with_preloaded_bookmark(opts["user"])
 874   end
 875
 876   defp maybe_set_thread_muted_field(query, %{"skip_preload" => true}), do: query
 877
 878   defp maybe_set_thread_muted_field(query, opts) do
 879     query
 880     |> Activity.with_set_thread_muted_field(opts["muting_user"] || opts["user"])
 881   end
 882
 883   defp maybe_order(query, %{order: :desc}) do
 884     query
 885     |> order_by(desc: :id)
 886   end
 887
 888   defp maybe_order(query, %{order: :asc}) do
 889     query
 890     |> order_by(asc: :id)
 891   end
 892
 893   defp maybe_order(query, _), do: query
 894
 895   def fetch_activities_query(recipients, opts \\ %{}) do
 896     config = %{
 897       skip_thread_containment: Config.get([:instance, :skip_thread_containment])
 898     }
 899
 900     Activity
 901     |> maybe_preload_objects(opts)
 902     |> maybe_preload_bookmarks(opts)
 903     |> maybe_set_thread_muted_field(opts)
 904     |> maybe_order(opts)
 905     |> restrict_recipients(recipients, opts["user"])
 906     |> restrict_tag(opts)
 907     |> restrict_tag_reject(opts)
 908     |> restrict_tag_all(opts)
 909     |> restrict_since(opts)
 910     |> restrict_local(opts)
 911     |> restrict_actor(opts)
 912     |> restrict_type(opts)
 913     |> restrict_state(opts)
 914     |> restrict_favorited_by(opts)
 915     |> restrict_blocked(opts)
 916     |> restrict_muted(opts)
 917     |> restrict_media(opts)
 918     |> restrict_visibility(opts)
 919     |> restrict_thread_visibility(opts, config)
 920     |> restrict_replies(opts)
 921     |> restrict_reblogs(opts)
 922     |> restrict_pinned(opts)
 923     |> restrict_muted_reblogs(opts)
 924     |> Activity.restrict_deactivated_users()
 925     |> exclude_poll_votes(opts)
 926   end
 927
 928   def fetch_activities(recipients, opts \\ %{}) do
 929     list_memberships = Pleroma.List.memberships(opts["user"])
 930
 931     fetch_activities_query(recipients ++ list_memberships, opts)
 932     |> Pagination.fetch_paginated(opts)
 933     |> Enum.reverse()
 934     |> maybe_update_cc(list_memberships, opts["user"])
 935   end
 936
 937   defp maybe_update_cc(activities, list_memberships, %User{ap_id: user_ap_id})
 938        when is_list(list_memberships) and length(list_memberships) > 0 do
 939     Enum.map(activities, fn
 940       %{data: %{"bcc" => bcc}} = activity when is_list(bcc) and length(bcc) > 0 ->
 941         if Enum.any?(bcc, &(&1 in list_memberships)) do
 942           update_in(activity.data["cc"], &[user_ap_id | &1])
 943         else
 944           activity
 945         end
 946
 947       activity ->
 948         activity
 949     end)
 950   end
 951
 952   defp maybe_update_cc(activities, _, _), do: activities
 953
 954   def fetch_activities_bounded_query(query, recipients, recipients_with_public) do
 955     from(activity in query,
 956       where:
 957         fragment("? && ?", activity.recipients, ^recipients) or
 958           (fragment("? && ?", activity.recipients, ^recipients_with_public) and
 959              ^Pleroma.Constants.as_public() in activity.recipients)
 960     )
 961   end
 962
 963   def fetch_activities_bounded(recipients, recipients_with_public, opts \\ %{}) do
 964     fetch_activities_query([], opts)
 965     |> fetch_activities_bounded_query(recipients, recipients_with_public)
 966     |> Pagination.fetch_paginated(opts)
 967     |> Enum.reverse()
 968   end
 969
 970   def upload(file, opts \\ []) do
 971     with {:ok, data} <- Upload.store(file, opts) do
 972       obj_data =
 973         if opts[:actor] do
 974           Map.put(data, "actor", opts[:actor])
 975         else
 976           data
 977         end
 978
 979       Repo.insert(%Object{data: obj_data})
 980     end
 981   end
 982
 983   defp object_to_user_data(data) do
 984     avatar =
 985       data["icon"]["url"] &&
 986         %{
 987           "type" => "Image",
 988           "url" => [%{"href" => data["icon"]["url"]}]
 989         }
 990
 991     banner =
 992       data["image"]["url"] &&
 993         %{
 994           "type" => "Image",
 995           "url" => [%{"href" => data["image"]["url"]}]
 996         }
 997
 998     fields =
 999       data
1000       |> Map.get("attachment", [])
1001       |> Enum.filter(fn %{"type" => t} -> t == "PropertyValue" end)
1002       |> Enum.map(fn fields -> Map.take(fields, ["name", "value"]) end)
1003
1004     locked = data["manuallyApprovesFollowers"] || false
1005     data = Transmogrifier.maybe_fix_user_object(data)
1006
1007     user_data = %{
1008       ap_id: data["id"],
1009       info: %{
1010         ap_enabled: true,
1011         source_data: data,
1012         banner: banner,
1013         fields: fields,
1014         locked: locked
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