# SPDX-License-Identifier: AGPL-3.0-only
defmodule Pleroma.Web.Federator do
- alias Pleroma.User
alias Pleroma.Activity
- alias Pleroma.Jobs
- alias Pleroma.Web.{WebFinger, Websub}
- alias Pleroma.Web.Federator.RetryQueue
+ alias Pleroma.Object.Containment
+ 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.OStatus
+ alias Pleroma.Web.Federator.Publisher
+ alias Pleroma.Web.Federator.RetryQueue
+ alias Pleroma.Web.Websub
+
require Logger
@websub Application.get_env(:pleroma, :websub)
@ostatus Application.get_env(:pleroma, :ostatus)
- def init() do
+ def init do
# 1 minute
- Process.sleep(1000 * 60 * 1)
+ Process.sleep(1000 * 60)
refresh_subscriptions()
end
# Client API
def incoming_doc(doc) do
- Jobs.enqueue(:federator_incoming, __MODULE__, [:incoming_doc, doc])
+ PleromaJobQueue.enqueue(:federator_incoming, __MODULE__, [:incoming_doc, doc])
end
def incoming_ap_doc(params) do
- Jobs.enqueue(:federator_incoming, __MODULE__, [:incoming_ap_doc, params])
+ PleromaJobQueue.enqueue(:federator_incoming, __MODULE__, [:incoming_ap_doc, params])
end
def publish(activity, priority \\ 1) do
- Jobs.enqueue(:federator_out, __MODULE__, [:publish, activity], priority)
- end
-
- def publish_single_ap(params) do
- Jobs.enqueue(:federator_out, __MODULE__, [:publish_single_ap, params])
- end
-
- def publish_single_websub(websub) do
- Jobs.enqueue(:federator_out, __MODULE__, [:publish_single_websub, websub])
+ PleromaJobQueue.enqueue(:federator_outgoing, __MODULE__, [:publish, activity], priority)
end
def verify_websub(websub) do
- Jobs.enqueue(:federator_out, __MODULE__, [:verify_websub, websub])
+ PleromaJobQueue.enqueue(:federator_outgoing, __MODULE__, [:verify_websub, websub])
end
def request_subscription(sub) do
- Jobs.enqueue(:federator_out, __MODULE__, [:request_subscription, sub])
+ PleromaJobQueue.enqueue(:federator_outgoing, __MODULE__, [:request_subscription, sub])
end
- def refresh_subscriptions() do
- Jobs.enqueue(:federator_out, __MODULE__, [:refresh_subscriptions])
+ def refresh_subscriptions do
+ PleromaJobQueue.enqueue(:federator_outgoing, __MODULE__, [:refresh_subscriptions])
end
# Job Worker Callbacks
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
- 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)
- Pleroma.Web.ActivityPub.ActivityPub.publish(actor, activity)
+ with %User{} = actor <- User.get_cached_by_ap_id(activity.data["actor"]),
+ {:ok, actor} <- User.ensure_keys_present(actor) do
+ Publisher.publish(actor, activity)
end
end
# actor shouldn't be acting on objects outside their own AP server.
with {:ok, _user} <- ap_enabled_actor(params["actor"]),
nil <- Activity.normalize(params["id"]),
- :ok <- Transmogrifier.contain_origin_from_id(params["actor"], params),
+ :ok <- Containment.contain_origin_from_id(params["actor"], params),
{:ok, activity} <- Transmogrifier.handle_incoming(params) do
{:ok, activity}
else
_e ->
# Just drop those for now
Logger.info("Unhandled activity")
- Logger.info(Poison.encode!(params, pretty: 2))
+ Logger.info(Jason.encode!(params, pretty: true))
:error
end
end
- def perform(:publish_single_ap, params) do
- case ActivityPub.publish_one(params) do
- {:ok, _} ->
- :ok
-
- {:error, _} ->
- RetryQueue.enqueue(params, ActivityPub)
- end
- end
-
def perform(
:publish_single_websub,
%{xml: _xml, topic: _topic, callback: _callback, secret: _secret} = params
end
def ap_enabled_actor(id) do
- user = User.get_by_ap_id(id)
+ user = User.get_cached_by_ap_id(id)
if User.ap_enabled?(user) do
{:ok, user}