git.squeep.com Git - akkoma/blob - lib/pleroma/web/federator/federator.ex

   1 defmodule Pleroma.Web.Federator do
   2   use GenServer
   3   alias Pleroma.User
   4   alias Pleroma.Activity
   5   alias Pleroma.Web.{WebFinger, Websub}
   6   alias Pleroma.Web.ActivityPub.ActivityPub
   7   alias Pleroma.Web.ActivityPub.Transmogrifier
   8   require Logger
   9
  10   @websub Application.get_env(:pleroma, :websub)
  11   @ostatus Application.get_env(:pleroma, :ostatus)
  12   @httpoison Application.get_env(:pleroma, :httpoison)
  13   @instance Application.get_env(:pleroma, :instance)
  14   @federating Keyword.get(@instance, :federating)
  15   @max_jobs 20
  16
  17   def start_link do
  18     spawn(fn ->
  19       Process.sleep(1000 * 60 * 1) # 1 minute
  20       enqueue(:refresh_subscriptions, nil)
  21     end)
  22     GenServer.start_link(__MODULE__, %{
  23           in: {:sets.new(), []},
  24           out: {:sets.new(), []}
  25                          }, name: __MODULE__)
  26   end
  27
  28   def handle(:refresh_subscriptions, _) do
  29     Logger.debug("Federator running refresh subscriptions")
  30     Websub.refresh_subscriptions()
  31     spawn(fn ->
  32       Process.sleep(1000 * 60 * 60 * 6) # 6 hours
  33       enqueue(:refresh_subscriptions, nil)
  34     end)
  35   end
  36
  37   def handle(:request_subscription, websub) do
  38     Logger.debug("Refreshing #{websub.topic}")
  39     with {:ok, websub } <- Websub.request_subscription(websub) do
  40       Logger.debug("Successfully refreshed #{websub.topic}")
  41     else
  42       _e -> Logger.debug("Couldn't refresh #{websub.topic}")
  43     end
  44   end
  45
  46   def handle(:publish, activity) do
  47     Logger.debug(fn -> "Running publish for #{activity.data["id"]}" end)
  48     with actor when not is_nil(actor) <- User.get_cached_by_ap_id(activity.data["actor"]) do
  49       {:ok, actor} = WebFinger.ensure_keys_present(actor)
  50       if ActivityPub.is_public?(activity) do
  51         Logger.info(fn -> "Sending #{activity.data["id"]} out via websub" end)
  52         Websub.publish(Pleroma.Web.OStatus.feed_path(actor), actor, activity)
  53
  54         Logger.info(fn -> "Sending #{activity.data["id"]} out via salmon" end)
  55         Pleroma.Web.Salmon.publish(actor, activity)
  56       end
  57
  58       Logger.info(fn -> "Sending #{activity.data["id"]} out via AP" end)
  59       Pleroma.Web.ActivityPub.ActivityPub.publish(actor, activity)
  60     end
  61   end
  62
  63   def handle(:verify_websub, websub) do
  64     Logger.debug(fn -> "Running websub verification for #{websub.id} (#{websub.topic}, #{websub.callback})" end)
  65     @websub.verify(websub)
  66   end
  67
  68   def handle(:incoming_doc, doc) do
  69     Logger.info("Got document, trying to parse")
  70     @ostatus.handle_incoming(doc)
  71   end
  72
  73   def handle(:incoming_ap_doc, params) do
  74     Logger.info("Handling incoming ap activity")
  75     with {:ok, _user} <- ap_enabled_actor(params["actor"]),
  76          nil <- Activity.get_by_ap_id(params["id"]),
  77          {:ok, activity} <- Transmogrifier.handle_incoming(params) do
  78     else
  79       %Activity{} ->
  80         Logger.info("Already had #{params["id"]}")
  81       e ->
  82         # Just drop those for now
  83         Logger.info("Unhandled activity")
  84         Logger.info(Poison.encode!(params, [pretty: 2]))
  85     end
  86   end
  87
  88   def handle(:publish_single_ap, params) do
  89     ActivityPub.publish_one(params)
  90   end
  91
  92   def handle(:publish_single_websub, %{xml: xml, topic: topic, callback: callback, secret: secret}) do
  93     signature = @websub.sign(secret || "", xml)
  94     Logger.debug(fn -> "Pushing #{topic} to #{callback}" end)
  95
  96     with {:ok, %{status_code: code}} <- @httpoison.post(callback, xml, [
  97                   {"Content-Type", "application/atom+xml"},
  98                   {"X-Hub-Signature", "sha1=#{signature}"}
  99                 ], timeout: 10000, recv_timeout: 20000, hackney: [pool: :default]) do
 100       Logger.debug(fn -> "Pushed to #{callback}, code #{code}" end)
 101     else e ->
 102         Logger.debug(fn -> "Couldn't push to #{callback}, #{inspect(e)}" end)
 103     end
 104   end
 105
 106   def handle(type, _) do
 107     Logger.debug(fn -> "Unknown task: #{type}" end)
 108     {:error, "Don't know what do do with this"}
 109   end
 110
 111   def enqueue(type, payload, priority \\ 1) do
 112     if @federating do
 113       if Mix.env == :test do
 114         handle(type, payload)
 115       else
 116         GenServer.cast(__MODULE__, {:enqueue, type, payload, priority})
 117       end
 118     end
 119   end
 120
 121   def maybe_start_job(running_jobs, queue) do
 122     if (:sets.size(running_jobs) < @max_jobs) && queue != [] do
 123       {{type, payload}, queue} = queue_pop(queue)
 124       {:ok, pid} = Task.start(fn -> handle(type, payload) end)
 125       mref = Process.monitor(pid)
 126       {:sets.add_element(mref, running_jobs), queue}
 127     else
 128       {running_jobs, queue}
 129     end
 130   end
 131
 132   def handle_cast({:enqueue, type, payload, priority}, state) when type in [:incoming_doc, :incoming_ap_doc] do
 133     %{in: {i_running_jobs, i_queue}, out: {o_running_jobs, o_queue}} = state
 134     i_queue = enqueue_sorted(i_queue, {type, payload}, 1)
 135     {i_running_jobs, i_queue} = maybe_start_job(i_running_jobs, i_queue)
 136     {:noreply, %{in: {i_running_jobs, i_queue}, out: {o_running_jobs, o_queue}}}
 137   end
 138
 139   def handle_cast({:enqueue, type, payload, priority}, state) do
 140     %{in: {i_running_jobs, i_queue}, out: {o_running_jobs, o_queue}} = state
 141     o_queue = enqueue_sorted(o_queue, {type, payload}, 1)
 142     {o_running_jobs, o_queue} = maybe_start_job(o_running_jobs, o_queue)
 143     {:noreply, %{in: {i_running_jobs, i_queue}, out: {o_running_jobs, o_queue}}}
 144   end
 145
 146   def handle_cast(m, state) do
 147     IO.inspect("Unknown: #{inspect(m)}, #{inspect(state)}")
 148     {:noreply, state}
 149   end
 150
 151   def handle_info({:DOWN, ref, :process, _pid, _reason}, state) do
 152     %{in: {i_running_jobs, i_queue}, out: {o_running_jobs, o_queue}} = state
 153     i_running_jobs = :sets.del_element(ref, i_running_jobs)
 154     o_running_jobs = :sets.del_element(ref, o_running_jobs)
 155     {i_running_jobs, i_queue} = maybe_start_job(i_running_jobs, i_queue)
 156     {o_running_jobs, o_queue} = maybe_start_job(o_running_jobs, o_queue)
 157
 158     {:noreply, %{in: {i_running_jobs, i_queue}, out: {o_running_jobs, o_queue}}}
 159   end
 160
 161   def enqueue_sorted(queue, element, priority) do
 162     [%{item: element, priority: priority} | queue]
 163     |> Enum.sort_by(fn (%{priority: priority}) -> priority end)
 164   end
 165
 166   def queue_pop([%{item: element} | queue]) do
 167     {element, queue}
 168   end
 169
 170   def ap_enabled_actor(id) do
 171     user = User.get_by_ap_id(id)
 172     if User.ap_enabled?(user) do
 173       {:ok, user}
 174     else
 175       ActivityPub.make_user_from_ap_id(id)
 176     end
 177   end
 178 end