X-Git-Url: http://git.squeep.com/?a=blobdiff_plain;f=lib%2Fpleroma%2Fweb%2Fwebsub%2Fwebsub.ex;h=c00ec08582aa328598d14fa02019d14833ecea1f;hb=305d2194136d4560e02c110d528164034d3727b2;hp=396dcf045265cde85d8340f4a22c5e5a188b526f;hpb=c7d08bc1cf37417f8275839f493d09981f2f7e09;p=akkoma

diff --git a/lib/pleroma/web/websub/websub.ex b/lib/pleroma/web/websub/websub.ex
index 396dcf045..c00ec0858 100644
--- a/lib/pleroma/web/websub/websub.ex
+++ b/lib/pleroma/web/websub/websub.ex
@@ -1,10 +1,19 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
 defmodule Pleroma.Web.Websub do
   alias Ecto.Changeset
+  alias Pleroma.Instances
   alias Pleroma.Repo
-  alias Pleroma.Web.Websub.{WebsubServerSubscription, WebsubClientSubscription}
+  alias Pleroma.Web.Websub.WebsubServerSubscription
+  alias Pleroma.Web.Websub.WebsubClientSubscription
   alias Pleroma.Web.OStatus.FeedRepresenter
-  alias Pleroma.Web.{XML, Endpoint, OStatus}
+  alias Pleroma.Web.XML
+  alias Pleroma.Web.Endpoint
+  alias Pleroma.Web.OStatus
   alias Pleroma.Web.Router.Helpers
+  alias Pleroma.Web.Federator
   require Logger
 
   import Ecto.Query
@@ -49,31 +58,37 @@ defmodule Pleroma.Web.Websub do
   ]
   def publish(topic, user, %{data: %{"type" => type}} = activity)
       when type in @supported_activities do
-    # TODO: Only send to still valid subscriptions.
+    response =
+      user
+      |> FeedRepresenter.to_simple_form([activity], [user])
+      |> :xmerl.export_simple(:xmerl_xml)
+      |> to_string
+
     query =
       from(
         sub in WebsubServerSubscription,
         where: sub.topic == ^topic and sub.state == "active",
-        where: fragment("? > NOW()", sub.valid_until)
+        where: fragment("? > (NOW() at time zone 'UTC')", sub.valid_until)
       )
 
     subscriptions = Repo.all(query)
 
-    Enum.each(subscriptions, fn sub ->
-      response =
-        user
-        |> FeedRepresenter.to_simple_form([activity], [user])
-        |> :xmerl.export_simple(:xmerl_xml)
-        |> to_string
+    callbacks = Enum.map(subscriptions, & &1.callback)
+    reachable_callbacks_metadata = Instances.filter_reachable(callbacks)
+    reachable_callbacks = Map.keys(reachable_callbacks_metadata)
 
+    subscriptions
+    |> Enum.filter(&(&1.callback in reachable_callbacks))
+    |> Enum.each(fn sub ->
       data = %{
         xml: response,
         topic: topic,
         callback: sub.callback,
-        secret: sub.secret
+        secret: sub.secret,
+        unreachable_since: reachable_callbacks_metadata[sub.callback]
       }
 
-      Pleroma.Web.Federator.enqueue(:publish_single_websub, data)
+      Federator.publish_single_websub(data)
     end)
   end
 
@@ -105,7 +120,7 @@ defmodule Pleroma.Web.Websub do
 
       websub = Repo.update!(change)
 
-      Pleroma.Web.Federator.enqueue(:verify_websub, websub)
+      Federator.verify_websub(websub)
 
       {:ok, websub}
     else
@@ -117,6 +132,12 @@ defmodule Pleroma.Web.Websub do
     end
   end
 
+  def incoming_subscription_request(user, params) do
+    Logger.info("Unhandled WebSub request for #{user.nickname}: #{inspect(params)}")
+
+    {:error, "Invalid WebSub request"}
+  end
+
   defp get_subscription(topic, callback) do
     Repo.get_by(WebsubServerSubscription, topic: topic, callback: callback) ||
       %WebsubServerSubscription{}
@@ -146,7 +167,7 @@ defmodule Pleroma.Web.Websub do
   end
 
   def subscribe(subscriber, subscribed, requester \\ &request_subscription/1) do
-    topic = subscribed.info["topic"]
+    topic = subscribed.info.topic
     # FIXME: Race condition, use transactions
     {:ok, subscription} =
       with subscription when not is_nil(subscription) <-
@@ -158,7 +179,7 @@ defmodule Pleroma.Web.Websub do
         _e ->
           subscription = %WebsubClientSubscription{
             topic: topic,
-            hub: subscribed.info["hub"],
+            hub: subscribed.info.hub,
             subscribers: [subscriber.ap_id],
             state: "requested",
             secret: :crypto.strong_rand_bytes(8) |> Base.url_encode64(),
@@ -173,7 +194,7 @@ defmodule Pleroma.Web.Websub do
 
   def gather_feed_data(topic, getter \\ &@httpoison.get/1) do
     with {:ok, response} <- getter.(topic),
-         status_code when status_code in 200..299 <- response.status_code,
+         status when status in 200..299 <- response.status,
          body <- response.body,
          doc <- XML.parse_document(body),
          uri when not is_nil(uri) <- XML.string_from_xpath("/feed/author[1]/uri", doc),
@@ -221,7 +242,7 @@ defmodule Pleroma.Web.Websub do
 
     task = Task.async(websub_checker)
 
-    with {:ok, %{status_code: 202}} <-
+    with {:ok, %{status: 202}} <-
            poster.(websub.hub, {:form, data}, "Content-type": "application/x-www-form-urlencoded"),
          {:ok, websub} <- Task.yield(task, timeout) do
       {:ok, websub}
@@ -249,32 +270,33 @@ defmodule Pleroma.Web.Websub do
     subs = Repo.all(query)
 
     Enum.each(subs, fn sub ->
-      Pleroma.Web.Federator.enqueue(:request_subscription, sub)
+      Federator.request_subscription(sub)
     end)
   end
 
-  def publish_one(%{xml: xml, topic: topic, callback: callback, secret: secret}) do
+  def publish_one(%{xml: xml, topic: topic, callback: callback, secret: secret} = params) do
     signature = sign(secret || "", xml)
     Logger.info(fn -> "Pushing #{topic} to #{callback}" end)
 
-    with {:ok, %{status_code: code}} <-
+    with {:ok, %{status: code}} when code in 200..299 <-
            @httpoison.post(
              callback,
              xml,
              [
                {"Content-Type", "application/atom+xml"},
                {"X-Hub-Signature", "sha1=#{signature}"}
-             ],
-             timeout: 10000,
-             recv_timeout: 20000,
-             hackney: [pool: :default]
+             ]
            ) do
+      if !Map.has_key?(params, :unreachable_since) || params[:unreachable_since],
+        do: Instances.set_reachable(callback)
+
       Logger.info(fn -> "Pushed to #{callback}, code #{code}" end)
       {:ok, code}
     else
-      e ->
-        Logger.debug(fn -> "Couldn't push to #{callback}, #{inspect(e)}" end)
-        {:error, e}
+      {_post_result, response} ->
+        unless params[:unreachable_since], do: Instances.set_reachable(callback)
+        Logger.debug(fn -> "Couldn't push to #{callback}, #{inspect(response)}" end)
+        {:error, response}
     end
   end
 end