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