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