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 \\ %{}) do
 523     q = fetch_activities_query([Pleroma.Constants.as_public()], opts)
 524
 525     q
 526     |> restrict_unlisted()
 527     |> Pagination.fetch_paginated(opts)
 528     |> Enum.reverse()
 529   end
 530
 531   @valid_visibilities ~w[direct unlisted public private]
 532
 533   defp restrict_visibility(query, %{visibility: visibility})
 534        when is_list(visibility) do
 535     if Enum.all?(visibility, &(&1 in @valid_visibilities)) do
 536       query =
 537         from(
 538           a in query,
 539           where:
 540             fragment(
 541               "activity_visibility(?, ?, ?) = ANY (?)",
 542               a.actor,
 543               a.recipients,
 544               a.data,
 545               ^visibility
 546             )
 547         )
 548
 549       query
 550     else
 551       Logger.error("Could not restrict visibility to #{visibility}")
 552     end
 553   end
 554
 555   defp restrict_visibility(query, %{visibility: visibility})
 556        when visibility in @valid_visibilities do
 557     from(
 558       a in query,
 559       where:
 560         fragment("activity_visibility(?, ?, ?) = ?", a.actor, a.recipients, a.data, ^visibility)
 561     )
 562   end
 563
 564   defp restrict_visibility(_query, %{visibility: visibility})
 565        when visibility not in @valid_visibilities do
 566     Logger.error("Could not restrict visibility to #{visibility}")
 567   end
 568
 569   defp restrict_visibility(query, _visibility), do: query
 570
 571   defp restrict_thread_visibility(query, _, %{skip_thread_containment: true} = _),
 572     do: query
 573
 574   defp restrict_thread_visibility(
 575          query,
 576          %{"user" => %User{info: %{skip_thread_containment: true}}},
 577          _
 578        ),
 579        do: query
 580
 581   defp restrict_thread_visibility(query, %{"user" => %User{ap_id: ap_id}}, _) do
 582     from(
 583       a in query,
 584       where: fragment("thread_visibility(?, (?)->>'id') = true", ^ap_id, a.data)
 585     )
 586   end
 587
 588   defp restrict_thread_visibility(query, _, _), do: query
 589
 590   def fetch_user_activities(user, reading_user, params \\ %{}) do
 591     params =
 592       params
 593       |> Map.put("type", ["Create", "Announce"])
 594       |> Map.put("user", reading_user)
 595       |> Map.put("actor_id", user.ap_id)
 596       |> Map.put("whole_db", true)
 597       |> Map.put("pinned_activity_ids", user.info.pinned_activities)
 598
 599     recipients =
 600       user_activities_recipients(%{
 601         "godmode" => params["godmode"],
 602         "reading_user" => reading_user
 603       })
 604
 605     fetch_activities(recipients, params)
 606     |> Enum.reverse()
 607   end
 608
 609   defp user_activities_recipients(%{"godmode" => true}) do
 610     []
 611   end
 612
 613   defp user_activities_recipients(%{"reading_user" => reading_user}) do
 614     if reading_user do
 615       [Pleroma.Constants.as_public()] ++ [reading_user.ap_id | reading_user.following]
 616     else
 617       [Pleroma.Constants.as_public()]
 618     end
 619   end
 620
 621   defp restrict_since(query, %{"since_id" => ""}), do: query
 622
 623   defp restrict_since(query, %{"since_id" => since_id}) do
 624     from(activity in query, where: activity.id > ^since_id)
 625   end
 626
 627   defp restrict_since(query, _), do: query
 628
 629   defp restrict_tag_reject(_query, %{"tag_reject" => _tag_reject, "skip_preload" => true}) do
 630     raise "Can't use the child object without preloading!"
 631   end
 632
 633   defp restrict_tag_reject(query, %{"tag_reject" => tag_reject})
 634        when is_list(tag_reject) and tag_reject != [] do
 635     from(
 636       [_activity, object] in query,
 637       where: fragment("not (?)->'tag' \\?| (?)", object.data, ^tag_reject)
 638     )
 639   end
 640
 641   defp restrict_tag_reject(query, _), do: query
 642
 643   defp restrict_tag_all(_query, %{"tag_all" => _tag_all, "skip_preload" => true}) do
 644     raise "Can't use the child object without preloading!"
 645   end
 646
 647   defp restrict_tag_all(query, %{"tag_all" => tag_all})
 648        when is_list(tag_all) and tag_all != [] do
 649     from(
 650       [_activity, object] in query,
 651       where: fragment("(?)->'tag' \\?& (?)", object.data, ^tag_all)
 652     )
 653   end
 654
 655   defp restrict_tag_all(query, _), do: query
 656
 657   defp restrict_tag(_query, %{"tag" => _tag, "skip_preload" => true}) do
 658     raise "Can't use the child object without preloading!"
 659   end
 660
 661   defp restrict_tag(query, %{"tag" => tag}) when is_list(tag) do
 662     from(
 663       [_activity, object] in query,
 664       where: fragment("(?)->'tag' \\?| (?)", object.data, ^tag)
 665     )
 666   end
 667
 668   defp restrict_tag(query, %{"tag" => tag}) when is_binary(tag) do
 669     from(
 670       [_activity, object] in query,
 671       where: fragment("(?)->'tag' \\? (?)", object.data, ^tag)
 672     )
 673   end
 674
 675   defp restrict_tag(query, _), do: query
 676
 677   defp restrict_recipients(query, [], _user), do: query
 678
 679   defp restrict_recipients(query, recipients, nil) do
 680     from(activity in query, where: fragment("? && ?", ^recipients, activity.recipients))
 681   end
 682
 683   defp restrict_recipients(query, recipients, user) do
 684     from(
 685       activity in query,
 686       where: fragment("? && ?", ^recipients, activity.recipients),
 687       or_where: activity.actor == ^user.ap_id
 688     )
 689   end
 690
 691   defp restrict_local(query, %{"local_only" => true}) do
 692     from(activity in query, where: activity.local == true)
 693   end
 694
 695   defp restrict_local(query, _), do: query
 696
 697   defp restrict_actor(query, %{"actor_id" => actor_id}) do
 698     from(activity in query, where: activity.actor == ^actor_id)
 699   end
 700
 701   defp restrict_actor(query, _), do: query
 702
 703   defp restrict_type(query, %{"type" => type}) when is_binary(type) do
 704     from(activity in query, where: fragment("?->>'type' = ?", activity.data, ^type))
 705   end
 706
 707   defp restrict_type(query, %{"type" => type}) do
 708     from(activity in query, where: fragment("?->>'type' = ANY(?)", activity.data, ^type))
 709   end
 710
 711   defp restrict_type(query, _), do: query
 712
 713   defp restrict_state(query, %{"state" => state}) do
 714     from(activity in query, where: fragment("?->>'state' = ?", activity.data, ^state))
 715   end
 716
 717   defp restrict_state(query, _), do: query
 718
 719   defp restrict_favorited_by(query, %{"favorited_by" => ap_id}) do
 720     from(
 721       [_activity, object] in query,
 722       where: fragment("(?)->'likes' \\? (?)", object.data, ^ap_id)
 723     )
 724   end
 725
 726   defp restrict_favorited_by(query, _), do: query
 727
 728   defp restrict_media(_query, %{"only_media" => _val, "skip_preload" => true}) do
 729     raise "Can't use the child object without preloading!"
 730   end
 731
 732   defp restrict_media(query, %{"only_media" => val}) when val == "true" or val == "1" do
 733     from(
 734       [_activity, object] in query,
 735       where: fragment("not (?)->'attachment' = (?)", object.data, ^[])
 736     )
 737   end
 738
 739   defp restrict_media(query, _), do: query
 740
 741   defp restrict_replies(query, %{"exclude_replies" => val}) when val == "true" or val == "1" do
 742     from(
 743       activity in query,
 744       where: fragment("?->'object'->>'inReplyTo' is null", activity.data)
 745     )
 746   end
 747
 748   defp restrict_replies(query, _), do: query
 749
 750   defp restrict_reblogs(query, %{"exclude_reblogs" => val}) when val == "true" or val == "1" do
 751     from(activity in query, where: fragment("?->>'type' != 'Announce'", activity.data))
 752   end
 753
 754   defp restrict_reblogs(query, _), do: query
 755
 756   defp restrict_muted(query, %{"with_muted" => val}) when val in [true, "true", "1"], do: query
 757
 758   defp restrict_muted(query, %{"muting_user" => %User{info: info}} = opts) do
 759     mutes = info.mutes
 760
 761     query =
 762       from([activity] in query,
 763         where: fragment("not (? = ANY(?))", activity.actor, ^mutes),
 764         where: fragment("not (?->'to' \\?| ?)", activity.data, ^mutes)
 765       )
 766
 767     unless opts["skip_preload"] do
 768       from([thread_mute: tm] in query, where: is_nil(tm.user_id))
 769     else
 770       query
 771     end
 772   end
 773
 774   defp restrict_muted(query, _), do: query
 775
 776   defp restrict_blocked(query, %{"blocking_user" => %User{info: info}}) do
 777     blocks = info.blocks || []
 778     domain_blocks = info.domain_blocks || []
 779
 780     query =
 781       if has_named_binding?(query, :object), do: query, else: Activity.with_joined_object(query)
 782
 783     from(
 784       [activity, object: o] in query,
 785       where: fragment("not (? = ANY(?))", activity.actor, ^blocks),
 786       where: fragment("not (? && ?)", activity.recipients, ^blocks),
 787       where:
 788         fragment(
 789           "not (?->>'type' = 'Announce' and ?->'to' \\?| ?)",
 790           activity.data,
 791           activity.data,
 792           ^blocks
 793         ),
 794       where: fragment("not (split_part(?, '/', 3) = ANY(?))", activity.actor, ^domain_blocks),
 795       where: fragment("not (split_part(?->>'actor', '/', 3) = ANY(?))", o.data, ^domain_blocks)
 796     )
 797   end
 798
 799   defp restrict_blocked(query, _), do: query
 800
 801   defp restrict_unlisted(query) do
 802     from(
 803       activity in query,
 804       where:
 805         fragment(
 806           "not (coalesce(?->'cc', '{}'::jsonb) \\?| ?)",
 807           activity.data,
 808           ^[Pleroma.Constants.as_public()]
 809         )
 810     )
 811   end
 812
 813   defp restrict_pinned(query, %{"pinned" => "true", "pinned_activity_ids" => ids}) do
 814     from(activity in query, where: activity.id in ^ids)
 815   end
 816
 817   defp restrict_pinned(query, _), do: query
 818
 819   defp restrict_muted_reblogs(query, %{"muting_user" => %User{info: info}}) do
 820     muted_reblogs = info.muted_reblogs || []
 821
 822     from(
 823       activity in query,
 824       where:
 825         fragment(
 826           "not ( ?->>'type' = 'Announce' and ? = ANY(?))",
 827           activity.data,
 828           activity.actor,
 829           ^muted_reblogs
 830         )
 831     )
 832   end
 833
 834   defp restrict_muted_reblogs(query, _), do: query
 835
 836   defp exclude_poll_votes(query, %{"include_poll_votes" => "true"}), do: query
 837
 838   defp exclude_poll_votes(query, _) do
 839     if has_named_binding?(query, :object) do
 840       from([activity, object: o] in query,
 841         where: fragment("not(?->>'type' = ?)", o.data, "Answer")
 842       )
 843     else
 844       query
 845     end
 846   end
 847
 848   defp exclude_id(query, %{"exclude_id" => id}) when is_binary(id) do
 849     from(activity in query, where: activity.id != ^id)
 850   end
 851
 852   defp exclude_id(query, _), do: query
 853
 854   defp maybe_preload_objects(query, %{"skip_preload" => true}), do: query
 855
 856   defp maybe_preload_objects(query, _) do
 857     query
 858     |> Activity.with_preloaded_object()
 859   end
 860
 861   defp maybe_preload_bookmarks(query, %{"skip_preload" => true}), do: query
 862
 863   defp maybe_preload_bookmarks(query, opts) do
 864     query
 865     |> Activity.with_preloaded_bookmark(opts["user"])
 866   end
 867
 868   defp maybe_set_thread_muted_field(query, %{"skip_preload" => true}), do: query
 869
 870   defp maybe_set_thread_muted_field(query, opts) do
 871     query
 872     |> Activity.with_set_thread_muted_field(opts["muting_user"] || opts["user"])
 873   end
 874
 875   defp maybe_order(query, %{order: :desc}) do
 876     query
 877     |> order_by(desc: :id)
 878   end
 879
 880   defp maybe_order(query, %{order: :asc}) do
 881     query
 882     |> order_by(asc: :id)
 883   end
 884
 885   defp maybe_order(query, _), do: query
 886
 887   def fetch_activities_query(recipients, opts \\ %{}) do
 888     config = %{
 889       skip_thread_containment: Config.get([:instance, :skip_thread_containment])
 890     }
 891
 892     Activity
 893     |> maybe_preload_objects(opts)
 894     |> maybe_preload_bookmarks(opts)
 895     |> maybe_set_thread_muted_field(opts)
 896     |> maybe_order(opts)
 897     |> restrict_recipients(recipients, opts["user"])
 898     |> restrict_tag(opts)
 899     |> restrict_tag_reject(opts)
 900     |> restrict_tag_all(opts)
 901     |> restrict_since(opts)
 902     |> restrict_local(opts)
 903     |> restrict_actor(opts)
 904     |> restrict_type(opts)
 905     |> restrict_state(opts)
 906     |> restrict_favorited_by(opts)
 907     |> restrict_blocked(opts)
 908     |> restrict_muted(opts)
 909     |> restrict_media(opts)
 910     |> restrict_visibility(opts)
 911     |> restrict_thread_visibility(opts, config)
 912     |> restrict_replies(opts)
 913     |> restrict_reblogs(opts)
 914     |> restrict_pinned(opts)
 915     |> restrict_muted_reblogs(opts)
 916     |> Activity.restrict_deactivated_users()
 917     |> exclude_poll_votes(opts)
 918   end
 919
 920   def fetch_activities(recipients, opts \\ %{}) do
 921     list_memberships = Pleroma.List.memberships(opts["user"])
 922
 923     fetch_activities_query(recipients ++ list_memberships, opts)
 924     |> Pagination.fetch_paginated(opts)
 925     |> Enum.reverse()
 926     |> maybe_update_cc(list_memberships, opts["user"])
 927   end
 928
 929   defp maybe_update_cc(activities, list_memberships, %User{ap_id: user_ap_id})
 930        when is_list(list_memberships) and length(list_memberships) > 0 do
 931     Enum.map(activities, fn
 932       %{data: %{"bcc" => bcc}} = activity when is_list(bcc) and length(bcc) > 0 ->
 933         if Enum.any?(bcc, &(&1 in list_memberships)) do
 934           update_in(activity.data["cc"], &[user_ap_id | &1])
 935         else
 936           activity
 937         end
 938
 939       activity ->
 940         activity
 941     end)
 942   end
 943
 944   defp maybe_update_cc(activities, _, _), do: activities
 945
 946   def fetch_activities_bounded_query(query, recipients, recipients_with_public) do
 947     from(activity in query,
 948       where:
 949         fragment("? && ?", activity.recipients, ^recipients) or
 950           (fragment("? && ?", activity.recipients, ^recipients_with_public) and
 951              ^Pleroma.Constants.as_public() in activity.recipients)
 952     )
 953   end
 954
 955   def fetch_activities_bounded(recipients, recipients_with_public, opts \\ %{}) do
 956     fetch_activities_query([], opts)
 957     |> fetch_activities_bounded_query(recipients, recipients_with_public)
 958     |> Pagination.fetch_paginated(opts)
 959     |> Enum.reverse()
 960   end
 961
 962   def upload(file, opts \\ []) do
 963     with {:ok, data} <- Upload.store(file, opts) do
 964       obj_data =
 965         if opts[:actor] do
 966           Map.put(data, "actor", opts[:actor])
 967         else
 968           data
 969         end
 970
 971       Repo.insert(%Object{data: obj_data})
 972     end
 973   end
 974
 975   defp object_to_user_data(data) do
 976     avatar =
 977       data["icon"]["url"] &&
 978         %{
 979           "type" => "Image",
 980           "url" => [%{"href" => data["icon"]["url"]}]
 981         }
 982
 983     banner =
 984       data["image"]["url"] &&
 985         %{
 986           "type" => "Image",
 987           "url" => [%{"href" => data["image"]["url"]}]
 988         }
 989
 990     fields =
 991       data
 992       |> Map.get("attachment", [])
 993       |> Enum.filter(fn %{"type" => t} -> t == "PropertyValue" end)
 994       |> Enum.map(fn fields -> Map.take(fields, ["name", "value"]) end)
 995
 996     locked = data["manuallyApprovesFollowers"] || false
 997     data = Transmogrifier.maybe_fix_user_object(data)
 998
 999     user_data = %{
1000       ap_id: data["id"],
1001       info: %{
1002         ap_enabled: true,
1003         source_data: data,
1004         banner: banner,
1005         fields: fields,
1006         locked: locked
1007       },
1008       avatar: avatar,
1009       name: data["name"],
1010       follower_address: data["followers"],
1011       following_address: data["following"],
1012       bio: data["summary"]
1013     }
1014
1015     # nickname can be nil because of virtual actors
1016     user_data =
1017       if data["preferredUsername"] do
1018         Map.put(
1019           user_data,
1020           :nickname,
1021           "#{data["preferredUsername"]}@#{URI.parse(data["id"]).host}"
1022         )
1023       else
1024         Map.put(user_data, :nickname, nil)
1025       end
1026
1027     {:ok, user_data}
1028   end
1029
1030   def fetch_follow_information_for_user(user) do
1031     with {:ok, following_data} <-
1032            Fetcher.fetch_and_contain_remote_object_from_id(user.following_address),
1033          following_count when is_integer(following_count) <- following_data["totalItems"],
1034          {:ok, hide_follows} <- collection_private(following_data),
1035          {:ok, followers_data} <-
1036            Fetcher.fetch_and_contain_remote_object_from_id(user.follower_address),
1037          followers_count when is_integer(followers_count) <- followers_data["totalItems"],
1038          {:ok, hide_followers} <- collection_private(followers_data) do
1039       {:ok,
1040        %{
1041          hide_follows: hide_follows,
1042          follower_count: followers_count,
1043          following_count: following_count,
1044          hide_followers: hide_followers
1045        }}
1046     else
1047       {:error, _} = e ->
1048         e
1049
1050       e ->
1051         {:error, e}
1052     end
1053   end
1054
1055   defp maybe_update_follow_information(data) do
1056     with {:enabled, true} <-
1057            {:enabled, Pleroma.Config.get([:instance, :external_user_synchronization])},
1058          {:ok, info} <- fetch_follow_information_for_user(data) do
1059       info = Map.merge(data.info, info)
1060       Map.put(data, :info, info)
1061     else
1062       {:enabled, false} ->
1063         data
1064
1065       e ->
1066         Logger.error(
1067           "Follower/Following counter update for #{data.ap_id} failed.\n" <> inspect(e)
1068         )
1069
1070         data
1071     end
1072   end
1073
1074   defp collection_private(data) do
1075     if is_map(data["first"]) and
1076          data["first"]["type"] in ["CollectionPage", "OrderedCollectionPage"] do
1077       {:ok, false}
1078     else
1079       with {:ok, %{"type" => type}} when type in ["CollectionPage", "OrderedCollectionPage"] <-
1080              Fetcher.fetch_and_contain_remote_object_from_id(data["first"]) do
1081         {:ok, false}
1082       else
1083         {:error, {:ok, %{status: code}}} when code in [401, 403] ->
1084           {:ok, true}
1085
1086         {:error, _} = e ->
1087           e
1088
1089         e ->
1090           {:error, e}
1091       end
1092     end
1093   end
1094
1095   def user_data_from_user_object(data) do
1096     with {:ok, data} <- MRF.filter(data),
1097          {:ok, data} <- object_to_user_data(data) do
1098       {:ok, data}
1099     else
1100       e -> {:error, e}
1101     end
1102   end
1103
1104   def fetch_and_prepare_user_from_ap_id(ap_id) do
1105     with {:ok, data} <- Fetcher.fetch_and_contain_remote_object_from_id(ap_id),
1106          {:ok, data} <- user_data_from_user_object(data),
1107          data <- maybe_update_follow_information(data) do
1108       {:ok, data}
1109     else
1110       e -> Logger.error("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
1111     end
1112   end
1113
1114   def make_user_from_ap_id(ap_id) do
1115     if _user = User.get_cached_by_ap_id(ap_id) do
1116       Transmogrifier.upgrade_user_from_ap_id(ap_id)
1117     else
1118       with {:ok, data} <- fetch_and_prepare_user_from_ap_id(ap_id) do
1119         User.insert_or_update_user(data)
1120       else
1121         e -> {:error, e}
1122       end
1123     end
1124   end
1125
1126   def make_user_from_nickname(nickname) do
1127     with {:ok, %{"ap_id" => ap_id}} when not is_nil(ap_id) <- WebFinger.finger(nickname) do
1128       make_user_from_ap_id(ap_id)
1129     else
1130       _e -> {:error, "No AP id in WebFinger"}
1131     end
1132   end
1133
1134   # filter out broken threads
1135   def contain_broken_threads(%Activity{} = activity, %User{} = user) do
1136     entire_thread_visible_for_user?(activity, user)
1137   end
1138
1139   # do post-processing on a specific activity
1140   def contain_activity(%Activity{} = activity, %User{} = user) do
1141     contain_broken_threads(activity, user)
1142   end
1143
1144   def fetch_direct_messages_query do
1145     Activity
1146     |> restrict_type(%{"type" => "Create"})
1147     |> restrict_visibility(%{visibility: "direct"})
1148     |> order_by([activity], asc: activity.id)
1149   end
1150 end