8 files changed, 856 insertions, 0 deletions
diff --git a/lib/pleroma/web/fed_sockets/fed_registry.ex b/lib/pleroma/web/fed_sockets/fed_registry.ex
new file mode 100644
index 000000000..e00ea69c0
--- /dev/null
+++ b/lib/pleroma/web/fed_sockets/fed_registry.ex
@@ -0,0 +1,185 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Web.FedSockets.FedRegistry do
+  @moduledoc """
+  The FedRegistry stores the active FedSockets for quick retrieval.
+
+  The storage and retrieval portion of the FedRegistry is done in process through
+  elixir's `Registry` module for speed and its ability to monitor for terminated processes.
+
+  Dropped connections will be caught by `Registry` and deleted. Since the next
+  message will initiate a new connection there is no reason to try and reconnect at that point.
+
+  Normally outside modules should have no need to call or use the FedRegistry themselves.
+  """
+
+  alias Pleroma.Web.FedSockets.FedSocket
+  alias Pleroma.Web.FedSockets.SocketInfo
+
+  require Logger
+
+  @default_rejection_duration 15 * 60 * 1000
+  @rejections :fed_socket_rejections
+
+  @doc """
+  Retrieves a FedSocket from the Registry given it's origin.
+
+  The origin is expected to be a string identifying the endpoint "example.com" or "example2.com:8080"
+
+  Will return:
+    * {:ok, fed_socket} for working FedSockets
+    * {:error, :rejected} for origins that have been tried and refused within the rejection duration interval
+    * {:error, some_reason} usually :missing for unknown origins
+  """
+  def get_fed_socket(origin) do
+    case get_registry_data(origin) do
+      {:error, reason} ->
+        {:error, reason}
+
+      {:ok, %{state: :connected} = socket_info} ->
+        {:ok, socket_info}
+    end
+  end
+
+  @doc """
+  Adds a connected FedSocket to the Registry.
+
+  Always returns {:ok, fed_socket}
+  """
+  def add_fed_socket(origin, pid \\ nil) do
+    origin
+    |> SocketInfo.build(pid)
+    |> SocketInfo.connect()
+    |> add_socket_info
+  end
+
+  defp add_socket_info(%{origin: origin, state: :connected} = socket_info) do
+    case Registry.register(FedSockets.Registry, origin, socket_info) do
+      {:ok, _owner} ->
+        clear_prior_rejection(origin)
+        Logger.debug("fedsocket added: #{inspect(origin)}")
+
+        {:ok, socket_info}
+
+      {:error, {:already_registered, _pid}} ->
+        FedSocket.close(socket_info)
+        existing_socket_info = Registry.lookup(FedSockets.Registry, origin)
+
+        {:ok, existing_socket_info}
+
+      _ ->
+        {:error, :error_adding_socket}
+    end
+  end
+
+  @doc """
+  Mark this origin as having rejected a connection attempt.
+  This will keep it from getting additional connection attempts
+  for a period of time specified in the config.
+
+  Always returns {:ok, new_reg_data}
+  """
+  def set_host_rejected(uri) do
+    new_reg_data =
+      uri
+      |> SocketInfo.origin()
+      |> get_or_create_registry_data()
+      |> set_to_rejected()
+      |> save_registry_data()
+
+    {:ok, new_reg_data}
+  end
+
+  @doc """
+  Retrieves the FedRegistryData from the Registry given it's origin.
+
+  The origin is expected to be a string identifying the endpoint "example.com" or "example2.com:8080"
+
+  Will return:
+    * {:ok, fed_registry_data} for known origins
+    * {:error, :missing} for uniknown origins
+    * {:error, :cache_error} indicating some low level runtime issues
+  """
+  def get_registry_data(origin) do
+    case Registry.lookup(FedSockets.Registry, origin) do
+      [] ->
+        if is_rejected?(origin) do
+          Logger.debug("previously rejected fedsocket requested")
+          {:error, :rejected}
+        else
+          {:error, :missing}
+        end
+
+      [{_pid, %{state: :connected} = socket_info}] ->
+        {:ok, socket_info}
+
+      _ ->
+        {:error, :cache_error}
+    end
+  end
+
+  @doc """
+  Retrieves a map of all sockets from the Registry. The keys are the origins and the values are the corresponding SocketInfo
+  """
+  def list_all do
+    (list_all_connected() ++ list_all_rejected())
+    |> Enum.into(%{})
+  end
+
+  defp list_all_connected do
+    FedSockets.Registry
+    |> Registry.select([{{:"$1", :_, :"$3"}, [], [{{:"$1", :"$3"}}]}])
+  end
+
+  defp list_all_rejected do
+    {:ok, keys} = Cachex.keys(@rejections)
+
+    {:ok, registry_data} =
+      Cachex.execute(@rejections, fn worker ->
+        Enum.map(keys, fn k -> {k, Cachex.get!(worker, k)} end)
+      end)
+
+    registry_data
+  end
+
+  defp clear_prior_rejection(origin),
+    do: Cachex.del(@rejections, origin)
+
+  defp is_rejected?(origin) do
+    case Cachex.get(@rejections, origin) do
+      {:ok, nil} ->
+        false
+
+      {:ok, _} ->
+        true
+    end
+  end
+
+  defp get_or_create_registry_data(origin) do
+    case get_registry_data(origin) do
+      {:error, :missing} ->
+        %SocketInfo{origin: origin}
+
+      {:ok, socket_info} ->
+        socket_info
+    end
+  end
+
+  defp save_registry_data(%SocketInfo{origin: origin, state: :connected} = socket_info) do
+    {:ok, true} = Registry.update_value(FedSockets.Registry, origin, fn _ -> socket_info end)
+    socket_info
+  end
+
+  defp save_registry_data(%SocketInfo{origin: origin, state: :rejected} = socket_info) do
+    rejection_expiration =
+      Pleroma.Config.get([:fed_sockets, :rejection_duration], @default_rejection_duration)
+
+    {:ok, true} = Cachex.put(@rejections, origin, socket_info, ttl: rejection_expiration)
+    socket_info
+  end
+
+  defp set_to_rejected(%SocketInfo{} = socket_info),
+    do: %SocketInfo{socket_info | state: :rejected}
+end
diff --git a/lib/pleroma/web/fed_sockets/fed_socket.ex b/lib/pleroma/web/fed_sockets/fed_socket.ex
new file mode 100644
index 000000000..98d64e65a
--- /dev/null
+++ b/lib/pleroma/web/fed_sockets/fed_socket.ex
@@ -0,0 +1,137 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Web.FedSockets.FedSocket do
+  @moduledoc """
+  The FedSocket module abstracts the actions to be taken taken on connections regardless of
+  whether the connection started as inbound or outbound.
+
+
+  Normally outside modules will have no need to call the FedSocket module directly.
+  """
+
+  alias Pleroma.Object
+  alias Pleroma.Object.Containment
+  alias Pleroma.User
+  alias Pleroma.Web.ActivityPub.ObjectView
+  alias Pleroma.Web.ActivityPub.UserView
+  alias Pleroma.Web.ActivityPub.Visibility
+  alias Pleroma.Web.FedSockets.FetchRegistry
+  alias Pleroma.Web.FedSockets.IngesterWorker
+  alias Pleroma.Web.FedSockets.OutgoingHandler
+  alias Pleroma.Web.FedSockets.SocketInfo
+
+  require Logger
+
+  @shake "61dd18f7-f1e6-49a4-939a-a749fcdc1103"
+
+  def connect_to_host(uri) do
+    case OutgoingHandler.start_link(uri) do
+      {:ok, pid} ->
+        {:ok, pid}
+
+      error ->
+        {:error, error}
+    end
+  end
+
+  def close(%SocketInfo{pid: socket_pid}),
+    do: Process.send(socket_pid, :close, [])
+
+  def publish(%SocketInfo{pid: socket_pid}, json) do
+    %{action: :publish, data: json}
+    |> Jason.encode!()
+    |> send_packet(socket_pid)
+  end
+
+  def fetch(%SocketInfo{pid: socket_pid}, id) do
+    fetch_uuid = FetchRegistry.register_fetch(id)
+
+    %{action: :fetch, data: id, uuid: fetch_uuid}
+    |> Jason.encode!()
+    |> send_packet(socket_pid)
+
+    wait_for_fetch_to_return(fetch_uuid, 0)
+  end
+
+  def receive_package(%SocketInfo{} = fed_socket, json) do
+    json
+    |> Jason.decode!()
+    |> process_package(fed_socket)
+  end
+
+  defp wait_for_fetch_to_return(uuid, cntr) do
+    case FetchRegistry.check_fetch(uuid) do
+      {:error, :waiting} ->
+        Process.sleep(:math.pow(cntr, 3) |> Kernel.trunc())
+        wait_for_fetch_to_return(uuid, cntr + 1)
+
+      {:error, :missing} ->
+        Logger.error("FedSocket fetch timed out - #{inspect(uuid)}")
+        {:error, :timeout}
+
+      {:ok, _fr} ->
+        FetchRegistry.pop_fetch(uuid)
+    end
+  end
+
+  defp process_package(%{"action" => "publish", "data" => data}, %{origin: origin} = _fed_socket) do
+    if Containment.contain_origin(origin, data) do
+      IngesterWorker.enqueue("ingest", %{"object" => data})
+    end
+
+    {:reply, %{"action" => "publish_reply", "status" => "processed"}}
+  end
+
+  defp process_package(%{"action" => "fetch_reply", "uuid" => uuid, "data" => data}, _fed_socket) do
+    FetchRegistry.register_fetch_received(uuid, data)
+    {:noreply, nil}
+  end
+
+  defp process_package(%{"action" => "fetch", "uuid" => uuid, "data" => ap_id}, _fed_socket) do
+    {:ok, data} = render_fetched_data(ap_id, uuid)
+    {:reply, data}
+  end
+
+  defp process_package(%{"action" => "publish_reply"}, _fed_socket) do
+    {:noreply, nil}
+  end
+
+  defp process_package(other, _fed_socket) do
+    Logger.warn("unknown json packages received #{inspect(other)}")
+    {:noreply, nil}
+  end
+
+  defp render_fetched_data(ap_id, uuid) do
+    {:ok,
+     %{
+       "action" => "fetch_reply",
+       "status" => "processed",
+       "uuid" => uuid,
+       "data" => represent_item(ap_id)
+     }}
+  end
+
+  defp represent_item(ap_id) do
+    case User.get_by_ap_id(ap_id) do
+      nil ->
+        object = Object.get_cached_by_ap_id(ap_id)
+
+        if Visibility.is_public?(object) do
+          Phoenix.View.render_to_string(ObjectView, "object.json", object: object)
+        else
+          nil
+        end
+
+      user ->
+        Phoenix.View.render_to_string(UserView, "user.json", user: user)
+    end
+  end
+
+  defp send_packet(data, socket_pid) do
+    Process.send(socket_pid, {:send, data}, [])
+  end
+
+  def shake, do: @shake
+end
diff --git a/lib/pleroma/web/fed_sockets/fetch_registry.ex b/lib/pleroma/web/fed_sockets/fetch_registry.ex
new file mode 100644
index 000000000..7897f0fc6
--- /dev/null
+++ b/lib/pleroma/web/fed_sockets/fetch_registry.ex
@@ -0,0 +1,151 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Web.FedSockets.FetchRegistry do
+  @moduledoc """
+  The FetchRegistry acts as a broker for fetch requests and return values.
+  This allows calling processes to block while waiting for a reply.
+  It doesn't impose it's own process instead using `Cachex` to handle fetches in process, allowing
+  multi threaded processes to avoid bottlenecking.
+
+  Normally outside modules will have no need to call or use the FetchRegistry themselves.
+
+  The `Cachex` parameters can be controlled from the config. Since exact timeout intervals
+  aren't necessary the following settings are used by default:
+
+  config :pleroma, :fed_sockets,
+    fed_socket_fetches: [
+      default: 12_000,
+      interval: 3_000,
+      lazy: false
+    ]
+
+  """
+
+  defmodule FetchRegistryData do
+    defstruct uuid: nil,
+              sent_json: nil,
+              received_json: nil,
+              sent_at: nil,
+              received_at: nil
+  end
+
+  alias Ecto.UUID
+
+  require Logger
+
+  @fetches :fed_socket_fetches
+
+  @doc """
+  Registers a json request wth the FetchRegistry and returns the identifying UUID.
+  """
+  def register_fetch(json) do
+    %FetchRegistryData{uuid: uuid} =
+      json
+      |> new_registry_data
+      |> save_registry_data
+
+    uuid
+  end
+
+  @doc """
+  Reports on the status of a Fetch given the identifying UUID.
+
+  Will return
+    * {:ok, fetched_object} if a fetch has completed
+    * {:error, :waiting} if a fetch is still pending
+    * {:error, other_error} usually :missing to indicate a fetch that has timed out
+  """
+  def check_fetch(uuid) do
+    case get_registry_data(uuid) do
+      {:ok, %FetchRegistryData{received_at: nil}} ->
+        {:error, :waiting}
+
+      {:ok, %FetchRegistryData{} = reg_data} ->
+        {:ok, reg_data}
+
+      e ->
+        e
+    end
+  end
+
+  @doc """
+  Retrieves the response to a fetch given the identifying UUID.
+  The completed fetch will be deleted from the FetchRegistry
+
+  Will return
+    * {:ok, fetched_object} if a fetch has completed
+    * {:error, :waiting} if a fetch is still pending
+    * {:error, other_error} usually :missing to indicate a fetch that has timed out
+  """
+  def pop_fetch(uuid) do
+    case check_fetch(uuid) do
+      {:ok, %FetchRegistryData{received_json: received_json}} ->
+        delete_registry_data(uuid)
+        {:ok, received_json}
+
+      e ->
+        e
+    end
+  end
+
+  @doc """
+  This is called to register a fetch has returned.
+  It expects the result data along with the UUID that was sent in the request
+
+  Will return the fetched object or :error
+  """
+  def register_fetch_received(uuid, data) do
+    case get_registry_data(uuid) do
+      {:ok, %FetchRegistryData{received_at: nil} = reg_data} ->
+        reg_data
+        |> set_fetch_received(data)
+        |> save_registry_data()
+
+      {:ok, %FetchRegistryData{} = reg_data} ->
+        Logger.warn("tried to add fetched data twice - #{uuid}")
+        reg_data
+
+      {:error, _} ->
+        Logger.warn("Error adding fetch to registry - #{uuid}")
+        :error
+    end
+  end
+
+  defp new_registry_data(json) do
+    %FetchRegistryData{
+      uuid: UUID.generate(),
+      sent_json: json,
+      sent_at: :erlang.monotonic_time(:millisecond)
+    }
+  end
+
+  defp get_registry_data(origin) do
+    case Cachex.get(@fetches, origin) do
+      {:ok, nil} ->
+        {:error, :missing}
+
+      {:ok, reg_data} ->
+        {:ok, reg_data}
+
+      _ ->
+        {:error, :cache_error}
+    end
+  end
+
+  defp set_fetch_received(%FetchRegistryData{} = reg_data, data),
+    do: %FetchRegistryData{
+      reg_data
+      | received_at: :erlang.monotonic_time(:millisecond),
+        received_json: data
+    }
+
+  defp save_registry_data(%FetchRegistryData{uuid: uuid} = reg_data) do
+    {:ok, true} = Cachex.put(@fetches, uuid, reg_data)
+    reg_data
+  end
+
+  defp delete_registry_data(origin),
+    do: {:ok, true} = Cachex.del(@fetches, origin)
+end
diff --git a/lib/pleroma/web/fed_sockets/incoming_handler.ex b/lib/pleroma/web/fed_sockets/incoming_handler.ex
new file mode 100644
index 000000000..49d0d9d84
--- /dev/null
+++ b/lib/pleroma/web/fed_sockets/incoming_handler.ex
@@ -0,0 +1,88 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Web.FedSockets.IncomingHandler do
+  require Logger
+
+  alias Pleroma.Web.FedSockets.FedRegistry
+  alias Pleroma.Web.FedSockets.FedSocket
+  alias Pleroma.Web.FedSockets.SocketInfo
+
+  import HTTPSignatures, only: [validate_conn: 1, split_signature: 1]
+
+  @behaviour :cowboy_websocket
+
+  def init(req, state) do
+    shake = FedSocket.shake()
+
+    with true <- Pleroma.Config.get([:fed_sockets, :enabled]),
+         sec_protocol <- :cowboy_req.header("sec-websocket-protocol", req, nil),
+         headers = %{"(request-target)" => ^shake} <- :cowboy_req.headers(req),
+         true <- validate_conn(%{req_headers: headers}),
+         %{"keyId" => origin} <- split_signature(headers["signature"]) do
+      req =
+        if is_nil(sec_protocol) do
+          req
+        else
+          :cowboy_req.set_resp_header("sec-websocket-protocol", sec_protocol, req)
+        end
+
+      {:cowboy_websocket, req, %{origin: origin}, %{}}
+    else
+      _ ->
+        {:ok, req, state}
+    end
+  end
+
+  def websocket_init(%{origin: origin}) do
+    case FedRegistry.add_fed_socket(origin) do
+      {:ok, socket_info} ->
+        {:ok, socket_info}
+
+      e ->
+        Logger.error("FedSocket websocket_init failed - #{inspect(e)}")
+        {:error, inspect(e)}
+    end
+  end
+
+  # Use the ping to  check if the connection should be expired
+  def websocket_handle(:ping, socket_info) do
+    if SocketInfo.expired?(socket_info) do
+      {:stop, socket_info}
+    else
+      {:ok, socket_info, :hibernate}
+    end
+  end
+
+  def websocket_handle({:text, data}, socket_info) do
+    socket_info = SocketInfo.touch(socket_info)
+
+    case FedSocket.receive_package(socket_info, data) do
+      {:noreply, _} ->
+        {:ok, socket_info}
+
+      {:reply, reply} ->
+        {:reply, {:text, Jason.encode!(reply)}, socket_info}
+
+      {:error, reason} ->
+        Logger.error("incoming error - receive_package: #{inspect(reason)}")
+        {:ok, socket_info}
+    end
+  end
+
+  def websocket_info({:send, message}, socket_info) do
+    socket_info = SocketInfo.touch(socket_info)
+
+    {:reply, {:text, message}, socket_info}
+  end
+
+  def websocket_info(:close, state) do
+    {:stop, state}
+  end
+
+  def websocket_info(message, state) do
+    Logger.debug("#{__MODULE__} unknown message #{inspect(message)}")
+    {:ok, state}
+  end
+end
diff --git a/lib/pleroma/web/fed_sockets/ingester_worker.ex b/lib/pleroma/web/fed_sockets/ingester_worker.ex
new file mode 100644
index 000000000..325f2a4ab
--- /dev/null
+++ b/lib/pleroma/web/fed_sockets/ingester_worker.ex
@@ -0,0 +1,33 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Web.FedSockets.IngesterWorker do
+  use Pleroma.Workers.WorkerHelper, queue: "ingestion_queue"
+  require Logger
+
+  alias Pleroma.Web.Federator
+
+  @impl Oban.Worker
+  def perform(%Job{args: %{"op" => "ingest", "object" => ingestee}}) do
+    try do
+      ingestee
+      |> Jason.decode!()
+      |> do_ingestion()
+    rescue
+      e ->
+        Logger.error("IngesterWorker error - #{inspect(e)}")
+        e
+    end
+  end
+
+  defp do_ingestion(params) do
+    case Federator.incoming_ap_doc(params) do
+      {:error, reason} ->
+        {:error, reason}
+
+      {:ok, object} ->
+        {:ok, object}
+    end
+  end
+end
diff --git a/lib/pleroma/web/fed_sockets/outgoing_handler.ex b/lib/pleroma/web/fed_sockets/outgoing_handler.ex
new file mode 100644
index 000000000..e235a7c43
--- /dev/null
+++ b/lib/pleroma/web/fed_sockets/outgoing_handler.ex
@@ -0,0 +1,151 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Web.FedSockets.OutgoingHandler do
+  use GenServer
+
+  require Logger
+
+  alias Pleroma.Application
+  alias Pleroma.Web.ActivityPub.InternalFetchActor
+  alias Pleroma.Web.FedSockets
+  alias Pleroma.Web.FedSockets.FedRegistry
+  alias Pleroma.Web.FedSockets.FedSocket
+  alias Pleroma.Web.FedSockets.SocketInfo
+
+  def start_link(uri) do
+    GenServer.start_link(__MODULE__, %{uri: uri})
+  end
+
+  def init(%{uri: uri}) do
+    case initiate_connection(uri) do
+      {:ok, ws_origin, conn_pid} ->
+        FedRegistry.add_fed_socket(ws_origin, conn_pid)
+
+      {:error, reason} ->
+        Logger.debug("Outgoing connection failed - #{inspect(reason)}")
+        :ignore
+    end
+  end
+
+  def handle_info({:gun_ws, conn_pid, _ref, {:text, data}}, socket_info) do
+    socket_info = SocketInfo.touch(socket_info)
+
+    case FedSocket.receive_package(socket_info, data) do
+      {:noreply, _} ->
+        {:noreply, socket_info}
+
+      {:reply, reply} ->
+        :gun.ws_send(conn_pid, {:text, Jason.encode!(reply)})
+        {:noreply, socket_info}
+
+      {:error, reason} ->
+        Logger.error("incoming error - receive_package: #{inspect(reason)}")
+        {:noreply, socket_info}
+    end
+  end
+
+  def handle_info(:close, state) do
+    Logger.debug("Sending close frame !!!!!!!")
+    {:close, state}
+  end
+
+  def handle_info({:gun_down, _pid, _prot, :closed, _}, state) do
+    {:stop, :normal, state}
+  end
+
+  def handle_info({:send, data}, %{conn_pid: conn_pid} = socket_info) do
+    socket_info = SocketInfo.touch(socket_info)
+    :gun.ws_send(conn_pid, {:text, data})
+    {:noreply, socket_info}
+  end
+
+  def handle_info({:gun_ws, _, _, :pong}, state) do
+    {:noreply, state, :hibernate}
+  end
+
+  def handle_info(msg, state) do
+    Logger.debug("#{__MODULE__} unhandled event #{inspect(msg)}")
+    {:noreply, state}
+  end
+
+  def terminate(reason, state) do
+    Logger.debug(
+      "#{__MODULE__} terminating outgoing connection for #{inspect(state)} for #{inspect(reason)}"
+    )
+
+    {:ok, state}
+  end
+
+  def initiate_connection(uri) do
+    ws_uri =
+      uri
+      |> SocketInfo.origin()
+      |> FedSockets.uri_for_origin()
+
+    %{host: host, port: port, path: path} = URI.parse(ws_uri)
+
+    with {:ok, conn_pid} <- :gun.open(to_charlist(host), port, %{protocols: [:http]}),
+         {:ok, _} <- :gun.await_up(conn_pid),
+         reference <-
+           :gun.get(conn_pid, to_charlist(path), [
+             {'user-agent', to_charlist(Application.user_agent())}
+           ]),
+         {:response, :fin, 204, _} <- :gun.await(conn_pid, reference),
+         headers <- build_headers(uri),
+         ref <- :gun.ws_upgrade(conn_pid, to_charlist(path), headers, %{silence_pings: false}) do
+      receive do
+        {:gun_upgrade, ^conn_pid, ^ref, [<<"websocket">>], _} ->
+          {:ok, ws_uri, conn_pid}
+      after
+        15_000 ->
+          Logger.debug("Fedsocket timeout connecting to #{inspect(uri)}")
+          {:error, :timeout}
+      end
+    else
+      {:response, :nofin, 404, _} ->
+        {:error, :fedsockets_not_supported}
+
+      e ->
+        Logger.debug("Fedsocket error connecting to #{inspect(uri)}")
+        {:error, e}
+    end
+  end
+
+  defp build_headers(uri) do
+    host_for_sig = uri |> URI.parse() |> host_signature()
+
+    shake = FedSocket.shake()
+    digest = "SHA-256=" <> (:crypto.hash(:sha256, shake) |> Base.encode64())
+    date = Pleroma.Signature.signed_date()
+    shake_size = byte_size(shake)
+
+    signature_opts = %{
+      "(request-target)": shake,
+      "content-length": to_charlist("#{shake_size}"),
+      date: date,
+      digest: digest,
+      host: host_for_sig
+    }
+
+    signature = Pleroma.Signature.sign(InternalFetchActor.get_actor(), signature_opts)
+
+    [
+      {'signature', to_charlist(signature)},
+      {'date', date},
+      {'digest', to_charlist(digest)},
+      {'content-length', to_charlist("#{shake_size}")},
+      {to_charlist("(request-target)"), to_charlist(shake)},
+      {'user-agent', to_charlist(Application.user_agent())}
+    ]
+  end
+
+  defp host_signature(%{host: host, scheme: scheme, port: port}) do
+    if port == URI.default_port(scheme) do
+      host
+    else
+      "#{host}:#{port}"
+    end
+  end
+end
diff --git a/lib/pleroma/web/fed_sockets/socket_info.ex b/lib/pleroma/web/fed_sockets/socket_info.ex
new file mode 100644
index 000000000..d6fdffe1a
--- /dev/null
+++ b/lib/pleroma/web/fed_sockets/socket_info.ex
@@ -0,0 +1,52 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Web.FedSockets.SocketInfo do
+  defstruct origin: nil,
+            pid: nil,
+            conn_pid: nil,
+            state: :default,
+            connected_until: nil
+
+  alias Pleroma.Web.FedSockets.SocketInfo
+  @default_connection_duration 15 * 60 * 1000
+
+  def build(uri, conn_pid \\ nil) do
+    uri
+    |> build_origin()
+    |> build_pids(conn_pid)
+    |> touch()
+  end
+
+  def touch(%SocketInfo{} = socket_info),
+    do: %{socket_info | connected_until: new_ttl()}
+
+  def connect(%SocketInfo{} = socket_info),
+    do: %{socket_info | state: :connected}
+
+  def expired?(%{connected_until: connected_until}),
+    do: connected_until < :erlang.monotonic_time(:millisecond)
+
+  def origin(uri),
+    do: build_origin(uri).origin
+
+  defp build_pids(socket_info, conn_pid),
+    do: struct(socket_info, pid: self(), conn_pid: conn_pid)
+
+  defp build_origin(uri) when is_binary(uri),
+    do: uri |> URI.parse() |> build_origin
+
+  defp build_origin(%{host: host, port: nil, scheme: scheme}),
+    do: build_origin(%{host: host, port: URI.default_port(scheme)})
+
+  defp build_origin(%{host: host, port: port}),
+    do: %SocketInfo{origin: "#{host}:#{port}"}
+
+  defp new_ttl do
+    connection_duration =
+      Pleroma.Config.get([:fed_sockets, :connection_duration], @default_connection_duration)
+
+    :erlang.monotonic_time(:millisecond) + connection_duration
+  end
+end
diff --git a/lib/pleroma/web/fed_sockets/supervisor.ex b/lib/pleroma/web/fed_sockets/supervisor.ex
new file mode 100644
index 000000000..a5f4bebfb
--- /dev/null
+++ b/lib/pleroma/web/fed_sockets/supervisor.ex
@@ -0,0 +1,59 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Web.FedSockets.Supervisor do
+  use Supervisor
+  import Cachex.Spec
+
+  def start_link(opts) do
+    Supervisor.start_link(__MODULE__, opts, name: __MODULE__)
+  end
+
+  def init(args) do
+    children = [
+      build_cache(:fed_socket_fetches, args),
+      build_cache(:fed_socket_rejections, args),
+      {Registry, keys: :unique, name: FedSockets.Registry, meta: [rejected: %{}]}
+    ]
+
+    opts = [strategy: :one_for_all, name: Pleroma.Web.Streamer.Supervisor]
+    Supervisor.init(children, opts)
+  end
+
+  defp build_cache(name, args) do
+    opts = get_opts(name, args)
+
+    %{
+      id: String.to_atom("#{name}_cache"),
+      start: {Cachex, :start_link, [name, opts]},
+      type: :worker
+    }
+  end
+
+  defp get_opts(cache_name, args)
+       when cache_name in [:fed_socket_fetches, :fed_socket_rejections] do
+    default = get_opts_or_config(args, cache_name, :default, 15_000)
+    interval = get_opts_or_config(args, cache_name, :interval, 3_000)
+    lazy = get_opts_or_config(args, cache_name, :lazy, false)
+
+    [expiration: expiration(default: default, interval: interval, lazy: lazy)]
+  end
+
+  defp get_opts(name, args) do
+    Keyword.get(args, name, [])
+  end
+
+  defp get_opts_or_config(args, name, key, default) do
+    args
+    |> Keyword.get(name, [])
+    |> Keyword.get(key)
+    |> case do
+      nil ->
+        Pleroma.Config.get([:fed_sockets, name, key], default)
+
+      value ->
+        value
+    end
+  end
+end