Merge branch 'fix/2473-purge-expired-activities-duplicates' into 'develop'
[akkoma] / lib / pleroma / web / federator.ex
1 # Pleroma: A lightweight social networking server
2 # Copyright © 2017-2021 Pleroma Authors <https://pleroma.social/>
3 # SPDX-License-Identifier: AGPL-3.0-only
4
5 defmodule Pleroma.Web.Federator do
6 alias Pleroma.Activity
7 alias Pleroma.Object.Containment
8 alias Pleroma.User
9 alias Pleroma.Web.ActivityPub.ActivityPub
10 alias Pleroma.Web.ActivityPub.Transmogrifier
11 alias Pleroma.Web.ActivityPub.Utils
12 alias Pleroma.Web.Federator.Publisher
13 alias Pleroma.Workers.PublisherWorker
14 alias Pleroma.Workers.ReceiverWorker
15
16 require Logger
17
18 @behaviour Pleroma.Web.Federator.Publishing
19
20 @doc """
21 Returns `true` if the distance to target object does not exceed max configured value.
22 Serves to prevent fetching of very long threads, especially useful on smaller instances.
23 Addresses [memory leaks on recursive replies fetching](https://git.pleroma.social/pleroma/pleroma/issues/161).
24 Applies to fetching of both ancestor (reply-to) and child (reply) objects.
25 """
26 # credo:disable-for-previous-line Credo.Check.Readability.MaxLineLength
27 def allowed_thread_distance?(distance) do
28 max_distance = Pleroma.Config.get([:instance, :federation_incoming_replies_max_depth])
29
30 if max_distance && max_distance >= 0 do
31 # Default depth is 0 (an object has zero distance from itself in its thread)
32 (distance || 0) <= max_distance
33 else
34 true
35 end
36 end
37
38 # Client API
39
40 def incoming_ap_doc(params) do
41 ReceiverWorker.enqueue("incoming_ap_doc", %{"params" => params})
42 end
43
44 @impl true
45 def publish(%{id: "pleroma:fakeid"} = activity) do
46 perform(:publish, activity)
47 end
48
49 @impl true
50 def publish(activity) do
51 PublisherWorker.enqueue("publish", %{"activity_id" => activity.id})
52 end
53
54 # Job Worker Callbacks
55
56 @spec perform(atom(), module(), any()) :: {:ok, any()} | {:error, any()}
57 def perform(:publish_one, module, params) do
58 apply(module, :publish_one, [params])
59 end
60
61 def perform(:publish, activity) do
62 Logger.debug(fn -> "Running publish for #{activity.data["id"]}" end)
63
64 with %User{} = actor <- User.get_cached_by_ap_id(activity.data["actor"]),
65 {:ok, actor} <- User.ensure_keys_present(actor) do
66 Publisher.publish(actor, activity)
67 end
68 end
69
70 def perform(:incoming_ap_doc, params) do
71 Logger.debug("Handling incoming AP activity")
72
73 actor =
74 params
75 |> Map.get("actor")
76 |> Utils.get_ap_id()
77
78 # NOTE: we use the actor ID to do the containment, this is fine because an
79 # actor shouldn't be acting on objects outside their own AP server.
80 with {_, {:ok, _user}} <- {:actor, ap_enabled_actor(actor)},
81 nil <- Activity.normalize(params["id"]),
82 {_, :ok} <-
83 {:correct_origin?, Containment.contain_origin_from_id(actor, params)},
84 {:ok, activity} <- Transmogrifier.handle_incoming(params) do
85 {:ok, activity}
86 else
87 {:correct_origin?, _} ->
88 Logger.debug("Origin containment failure for #{params["id"]}")
89 {:error, :origin_containment_failed}
90
91 %Activity{} ->
92 Logger.debug("Already had #{params["id"]}")
93 {:error, :already_present}
94
95 {:actor, e} ->
96 Logger.debug("Unhandled actor #{actor}, #{inspect(e)}")
97 {:error, e}
98
99 e ->
100 # Just drop those for now
101 Logger.debug(fn -> "Unhandled activity\n" <> Jason.encode!(params, pretty: true) end)
102 {:error, e}
103 end
104 end
105
106 def ap_enabled_actor(id) do
107 user = User.get_cached_by_ap_id(id)
108
109 if User.ap_enabled?(user) do
110 {:ok, user}
111 else
112 ActivityPub.make_user_from_ap_id(id)
113 end
114 end
115 end