X-Git-Url: http://git.squeep.com/?a=blobdiff_plain;f=lib%2Fpleroma%2Fweb%2Ffederator%2Ffederator.ex;h=ac3d7c1323dcc5e3e9469905753891e899405cc3;hb=627e5a0a4992cc19fc65a7e93a09c470c8e2bf33;hp=51d2931960bd3b043f499967aad922371b1fe2c1;hpb=9b97b023c53b88f4c04aed17e36022201016b6c9;p=akkoma diff --git a/lib/pleroma/web/federator/federator.ex b/lib/pleroma/web/federator/federator.ex index 51d293196..a1f6373a4 100644 --- a/lib/pleroma/web/federator/federator.ex +++ b/lib/pleroma/web/federator/federator.ex @@ -1,58 +1,113 @@ +# Pleroma: A lightweight social networking server +# Copyright © 2017-2019 Pleroma Authors +# SPDX-License-Identifier: AGPL-3.0-only + defmodule Pleroma.Web.Federator do - use GenServer - alias Pleroma.User alias Pleroma.Activity - alias Pleroma.Web.{WebFinger, Websub} + alias Pleroma.User alias Pleroma.Web.ActivityPub.ActivityPub + alias Pleroma.Web.ActivityPub.Relay alias Pleroma.Web.ActivityPub.Transmogrifier + alias Pleroma.Web.ActivityPub.Utils + alias Pleroma.Web.ActivityPub.Visibility + alias Pleroma.Web.Federator.RetryQueue + alias Pleroma.Web.OStatus + alias Pleroma.Object.Containment + alias Pleroma.Web.Salmon + alias Pleroma.Web.WebFinger + alias Pleroma.Web.Websub + require Logger @websub Application.get_env(:pleroma, :websub) @ostatus Application.get_env(:pleroma, :ostatus) - @httpoison Application.get_env(:pleroma, :httpoison) - @instance Application.get_env(:pleroma, :instance) - @federating Keyword.get(@instance, :federating) - @max_jobs 20 - def start_link do - spawn(fn -> - Process.sleep(1000 * 60 * 1) # 1 minute - enqueue(:refresh_subscriptions, nil) - end) - GenServer.start_link(__MODULE__, %{ - in: {:sets.new(), []}, - out: {:sets.new(), []} - }, name: __MODULE__) + def init do + # 1 minute + Process.sleep(1000 * 60) + refresh_subscriptions() + end + + # Client API + + def incoming_doc(doc) do + PleromaJobQueue.enqueue(:federator_incoming, __MODULE__, [:incoming_doc, doc]) + end + + def incoming_ap_doc(params) do + PleromaJobQueue.enqueue(:federator_incoming, __MODULE__, [:incoming_ap_doc, params]) + end + + def publish(activity, priority \\ 1) do + PleromaJobQueue.enqueue(:federator_outgoing, __MODULE__, [:publish, activity], priority) + end + + def publish_single_ap(params) do + PleromaJobQueue.enqueue(:federator_outgoing, __MODULE__, [:publish_single_ap, params]) end - def handle(:refresh_subscriptions, _) do + def publish_single_websub(websub) do + PleromaJobQueue.enqueue(:federator_outgoing, __MODULE__, [:publish_single_websub, websub]) + end + + def verify_websub(websub) do + PleromaJobQueue.enqueue(:federator_outgoing, __MODULE__, [:verify_websub, websub]) + end + + def request_subscription(sub) do + PleromaJobQueue.enqueue(:federator_outgoing, __MODULE__, [:request_subscription, sub]) + end + + def refresh_subscriptions do + PleromaJobQueue.enqueue(:federator_outgoing, __MODULE__, [:refresh_subscriptions]) + end + + def publish_single_salmon(params) do + PleromaJobQueue.enqueue(:federator_outgoing, __MODULE__, [:publish_single_salmon, params]) + end + + # Job Worker Callbacks + + def perform(:refresh_subscriptions) do Logger.debug("Federator running refresh subscriptions") Websub.refresh_subscriptions() + spawn(fn -> - Process.sleep(1000 * 60 * 60 * 6) # 6 hours - enqueue(:refresh_subscriptions, nil) + # 6 hours + Process.sleep(1000 * 60 * 60 * 6) + refresh_subscriptions() end) end - def handle(:request_subscription, websub) do + def perform(:request_subscription, websub) do Logger.debug("Refreshing #{websub.topic}") - with {:ok, websub } <- Websub.request_subscription(websub) do + + with {:ok, websub} <- Websub.request_subscription(websub) do Logger.debug("Successfully refreshed #{websub.topic}") else _e -> Logger.debug("Couldn't refresh #{websub.topic}") end end - def handle(:publish, activity) do + def perform(:publish, activity) do Logger.debug(fn -> "Running publish for #{activity.data["id"]}" end) + with actor when not is_nil(actor) <- User.get_cached_by_ap_id(activity.data["actor"]) do {:ok, actor} = WebFinger.ensure_keys_present(actor) - if ActivityPub.is_public?(activity) do - Logger.info(fn -> "Sending #{activity.data["id"]} out via WebSub" end) - Websub.publish(Pleroma.Web.OStatus.feed_path(actor), actor, activity) - Logger.info(fn -> "Sending #{activity.data["id"]} out via Salmon" end) - Pleroma.Web.Salmon.publish(actor, activity) + if Visibility.is_public?(activity) do + if OStatus.is_representable?(activity) do + Logger.info(fn -> "Sending #{activity.data["id"]} out via WebSub" end) + Websub.publish(Pleroma.Web.OStatus.feed_path(actor), actor, activity) + + Logger.info(fn -> "Sending #{activity.data["id"]} out via Salmon" end) + Pleroma.Web.Salmon.publish(actor, activity) + end + + if Keyword.get(Application.get_env(:pleroma, :instance), :allow_relay) do + Logger.info(fn -> "Relaying #{activity.data["id"]} out" end) + Relay.publish(activity) + end end Logger.info(fn -> "Sending #{activity.data["id"]} out via AP" end) @@ -60,115 +115,79 @@ defmodule Pleroma.Web.Federator do end end - def handle(:verify_websub, websub) do - Logger.debug(fn -> "Running WebSub verification for #{websub.id} (#{websub.topic}, #{websub.callback})" end) + def perform(:verify_websub, websub) do + Logger.debug(fn -> + "Running WebSub verification for #{websub.id} (#{websub.topic}, #{websub.callback})" + end) + @websub.verify(websub) end - def handle(:incoming_doc, doc) do + def perform(:incoming_doc, doc) do Logger.info("Got document, trying to parse") @ostatus.handle_incoming(doc) end - def handle(:incoming_ap_doc, params) do + def perform(:incoming_ap_doc, params) do Logger.info("Handling incoming AP activity") + + params = Utils.normalize_params(params) + + # NOTE: we use the actor ID to do the containment, this is fine because an + # actor shouldn't be acting on objects outside their own AP server. with {:ok, _user} <- ap_enabled_actor(params["actor"]), - nil <- Activity.get_by_ap_id(params["id"]), + nil <- Activity.normalize(params["id"]), + :ok <- Containment.contain_origin_from_id(params["actor"], params), {:ok, activity} <- Transmogrifier.handle_incoming(params) do + {:ok, activity} else %Activity{} -> Logger.info("Already had #{params["id"]}") - e -> + :error + + _e -> # Just drop those for now Logger.info("Unhandled activity") - Logger.info(Poison.encode!(params, [pretty: 2])) + Logger.info(Poison.encode!(params, pretty: 2)) + :error end end - def handle(:publish_single_ap, params) do - ActivityPub.publish_one(params) + def perform(:publish_single_salmon, params) do + Salmon.send_to_user(params) end - def handle(:publish_single_websub, %{xml: xml, topic: topic, callback: callback, secret: secret}) do - signature = @websub.sign(secret || "", xml) - Logger.debug(fn -> "Pushing #{topic} to #{callback}" end) + def perform(:publish_single_ap, params) do + case ActivityPub.publish_one(params) do + {:ok, _} -> + :ok - with {:ok, %{status_code: code}} <- @httpoison.post(callback, xml, [ - {"Content-Type", "application/atom+xml"}, - {"X-Hub-Signature", "sha1=#{signature}"} - ], timeout: 10000, recv_timeout: 20000, hackney: [pool: :default]) do - Logger.debug(fn -> "Pushed to #{callback}, code #{code}" end) - else e -> - Logger.debug(fn -> "Couldn't push to #{callback}, #{inspect(e)}" end) + {:error, _} -> + RetryQueue.enqueue(params, ActivityPub) end end - def handle(type, _) do - Logger.debug(fn -> "Unknown task: #{type}" end) - {:error, "Don't know what to do with this"} - end + def perform( + :publish_single_websub, + %{xml: _xml, topic: _topic, callback: _callback, secret: _secret} = params + ) do + case Websub.publish_one(params) do + {:ok, _} -> + :ok - def enqueue(type, payload, priority \\ 1) do - if @federating do - if Mix.env == :test do - handle(type, payload) - else - GenServer.cast(__MODULE__, {:enqueue, type, payload, priority}) - end - end - end - - def maybe_start_job(running_jobs, queue) do - if (:sets.size(running_jobs) < @max_jobs) && queue != [] do - {{type, payload}, queue} = queue_pop(queue) - {:ok, pid} = Task.start(fn -> handle(type, payload) end) - mref = Process.monitor(pid) - {:sets.add_element(mref, running_jobs), queue} - else - {running_jobs, queue} + {:error, _} -> + RetryQueue.enqueue(params, Websub) end end - def handle_cast({:enqueue, type, payload, priority}, state) when type in [:incoming_doc, :incoming_ap_doc] do - %{in: {i_running_jobs, i_queue}, out: {o_running_jobs, o_queue}} = state - i_queue = enqueue_sorted(i_queue, {type, payload}, 1) - {i_running_jobs, i_queue} = maybe_start_job(i_running_jobs, i_queue) - {:noreply, %{in: {i_running_jobs, i_queue}, out: {o_running_jobs, o_queue}}} - end - - def handle_cast({:enqueue, type, payload, priority}, state) do - %{in: {i_running_jobs, i_queue}, out: {o_running_jobs, o_queue}} = state - o_queue = enqueue_sorted(o_queue, {type, payload}, 1) - {o_running_jobs, o_queue} = maybe_start_job(o_running_jobs, o_queue) - {:noreply, %{in: {i_running_jobs, i_queue}, out: {o_running_jobs, o_queue}}} - end - - def handle_cast(m, state) do - IO.inspect("Unknown: #{inspect(m)}, #{inspect(state)}") - {:noreply, state} - end - - def handle_info({:DOWN, ref, :process, _pid, _reason}, state) do - %{in: {i_running_jobs, i_queue}, out: {o_running_jobs, o_queue}} = state - i_running_jobs = :sets.del_element(ref, i_running_jobs) - o_running_jobs = :sets.del_element(ref, o_running_jobs) - {i_running_jobs, i_queue} = maybe_start_job(i_running_jobs, i_queue) - {o_running_jobs, o_queue} = maybe_start_job(o_running_jobs, o_queue) - - {:noreply, %{in: {i_running_jobs, i_queue}, out: {o_running_jobs, o_queue}}} - end - - def enqueue_sorted(queue, element, priority) do - [%{item: element, priority: priority} | queue] - |> Enum.sort_by(fn (%{priority: priority}) -> priority end) - end - - def queue_pop([%{item: element} | queue]) do - {element, queue} + def perform(type, _) do + Logger.debug(fn -> "Unknown task: #{type}" end) + {:error, "Don't know what to do with this"} end def ap_enabled_actor(id) do user = User.get_by_ap_id(id) + if User.ap_enabled?(user) do {:ok, user} else