1 # Pleroma: A lightweight social networking server
2 # Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
3 # SPDX-License-Identifier: AGPL-3.0-only
5 defmodule Pleroma.Web.FedSockets.FedRegistry do
7 The FedRegistry stores the active FedSockets for quick retrieval.
9 The storage and retrieval portion of the FedRegistry is done in process through
10 elixir's `Registry` module for speed and its ability to monitor for terminated processes.
12 Dropped connections will be caught by `Registry` and deleted. Since the next
13 message will initiate a new connection there is no reason to try and reconnect at that point.
15 Normally outside modules should have no need to call or use the FedRegistry themselves.
18 alias Pleroma.Web.FedSockets.FedSocket
19 alias Pleroma.Web.FedSockets.SocketInfo
23 @default_rejection_duration 15 * 60 * 1000
24 @rejections :fed_socket_rejections
27 Retrieves a FedSocket from the Registry given it's origin.
29 The origin is expected to be a string identifying the endpoint "example.com" or "example2.com:8080"
32 * {:ok, fed_socket} for working FedSockets
33 * {:error, :rejected} for origins that have been tried and refused within the rejection duration interval
34 * {:error, some_reason} usually :missing for unknown origins
36 def get_fed_socket(origin) do
37 case get_registry_data(origin) do
41 {:ok, %{state: :connected} = socket_info} ->
47 Adds a connected FedSocket to the Registry.
49 Always returns {:ok, fed_socket}
51 def add_fed_socket(origin, pid \\ nil) do
53 |> SocketInfo.build(pid)
54 |> SocketInfo.connect()
58 defp add_socket_info(%{origin: origin, state: :connected} = socket_info) do
59 case Registry.register(FedSockets.Registry, origin, socket_info) do
61 clear_prior_rejection(origin)
62 Logger.debug("fedsocket added: #{inspect(origin)}")
66 {:error, {:already_registered, _pid}} ->
67 FedSocket.close(socket_info)
68 existing_socket_info = Registry.lookup(FedSockets.Registry, origin)
70 {:ok, existing_socket_info}
73 {:error, :error_adding_socket}
78 Mark this origin as having rejected a connection attempt.
79 This will keep it from getting additional connection attempts
80 for a period of time specified in the config.
82 Always returns {:ok, new_reg_data}
84 def set_host_rejected(uri) do
87 |> SocketInfo.origin()
88 |> get_or_create_registry_data()
90 |> save_registry_data()
96 Retrieves the FedRegistryData from the Registry given it's origin.
98 The origin is expected to be a string identifying the endpoint "example.com" or "example2.com:8080"
101 * {:ok, fed_registry_data} for known origins
102 * {:error, :missing} for uniknown origins
103 * {:error, :cache_error} indicating some low level runtime issues
105 def get_registry_data(origin) do
106 case Registry.lookup(FedSockets.Registry, origin) do
108 if is_rejected?(origin) do
109 Logger.debug("previously rejected fedsocket requested")
115 [{_pid, %{state: :connected} = socket_info}] ->
119 {:error, :cache_error}
124 Retrieves a map of all sockets from the Registry. The keys are the origins and the values are the corresponding SocketInfo
127 (list_all_connected() ++ list_all_rejected())
131 defp list_all_connected do
133 |> Registry.select([{{:"$1", :_, :"$3"}, [], [{{:"$1", :"$3"}}]}])
136 defp list_all_rejected do
137 {:ok, keys} = Cachex.keys(@rejections)
139 {:ok, registry_data} =
140 Cachex.execute(@rejections, fn worker ->
141 Enum.map(keys, fn k -> {k, Cachex.get!(worker, k)} end)
147 defp clear_prior_rejection(origin),
148 do: Cachex.del(@rejections, origin)
150 defp is_rejected?(origin) do
151 case Cachex.get(@rejections, origin) do
160 defp get_or_create_registry_data(origin) do
161 case get_registry_data(origin) do
162 {:error, :missing} ->
163 %SocketInfo{origin: origin}
165 {:ok, socket_info} ->
170 defp save_registry_data(%SocketInfo{origin: origin, state: :connected} = socket_info) do
171 {:ok, true} = Registry.update_value(FedSockets.Registry, origin, fn _ -> socket_info end)
175 defp save_registry_data(%SocketInfo{origin: origin, state: :rejected} = socket_info) do
176 rejection_expiration =
177 Pleroma.Config.get([:fed_sockets, :rejection_duration], @default_rejection_duration)
179 {:ok, true} = Cachex.put(@rejections, origin, socket_info, ttl: rejection_expiration)
183 defp set_to_rejected(%SocketInfo{} = socket_info),
184 do: %SocketInfo{socket_info | state: :rejected}