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

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