From c5fa682c317717c64168bf2d77b28d805ffff450 Mon Sep 17 00:00:00 2001
From: Roger Braun <roger@rogerbraun.net>
Date: Wed, 26 Apr 2017 18:33:10 +0200
Subject: [PATCH] Refactor, add beginnings of websub client subscriptions.

---
 lib/pleroma/web/activity_pub/activity_pub.ex  |  2 +-
 lib/pleroma/web/federator/federator.ex        | 32 +++++++++++++++++++
 lib/pleroma/web/websub/websub.ex              | 26 ++++++++++++---
 .../web/websub/websub_client_subscription.ex  | 13 ++++++++
 ...4155_create_websub_client_subscription.exs | 15 +++++++++
 test/web/websub/websub_test.exs               | 12 ++++++-
 6 files changed, 93 insertions(+), 7 deletions(-)
 create mode 100644 lib/pleroma/web/federator/federator.ex
 create mode 100644 lib/pleroma/web/websub/websub_client_subscription.ex
 create mode 100644 priv/repo/migrations/20170426154155_create_websub_client_subscription.exs

diff --git a/lib/pleroma/web/activity_pub/activity_pub.ex b/lib/pleroma/web/activity_pub/activity_pub.ex
index 7264123d8..82f9fcc1c 100644
--- a/lib/pleroma/web/activity_pub/activity_pub.ex
+++ b/lib/pleroma/web/activity_pub/activity_pub.ex
@@ -36,7 +36,7 @@ def create(to, actor, context, object, additional \\ %{}, published \\ nil) do
       {:ok, activity} = add_conversation_id(activity)
 
       if actor.local do
-        Pleroma.Web.Websub.publish(Pleroma.Web.OStatus.feed_path(actor), actor, activity)
+        Pleroma.Web.Federator.enqueue(:publish, activity)
        end
 
       {:ok, activity}
diff --git a/lib/pleroma/web/federator/federator.ex b/lib/pleroma/web/federator/federator.ex
new file mode 100644
index 000000000..f489ed837
--- /dev/null
+++ b/lib/pleroma/web/federator/federator.ex
@@ -0,0 +1,32 @@
+defmodule Pleroma.Web.Federator do
+  alias Pleroma.User
+  require Logger
+
+  @websub_verifier Application.get_env(:pleroma, :websub_verifier)
+
+  def handle(:publish, activity) do
+    Logger.debug("Running publish for #{activity.data["id"]}")
+    with actor when not is_nil(actor) <- User.get_cached_by_ap_id(activity.data["actor"]) do
+      Pleroma.Web.Websub.publish(Pleroma.Web.OStatus.feed_path(actor), actor, activity)
+    end
+  end
+
+  def handle(:verify_websub, websub) do
+    Logger.debug("Running websub verification for #{websub.id} (#{websub.topic}, #{websub.callback})")
+    @websub_verifier.verify(websub)
+  end
+
+  def handle(type, payload) do
+    Logger.debug("Unknown task: #{type}")
+    {:error, "Don't know what do do with this"}
+  end
+
+  def enqueue(type, payload) do
+    # for now, just run immediately in a new process.
+    if Mix.env == :test do
+      handle(type, payload)
+    else
+      spawn(fn -> handle(type, payload) end)
+    end
+  end
+end
diff --git a/lib/pleroma/web/websub/websub.ex b/lib/pleroma/web/websub/websub.ex
index cc66b52dd..03b0aec8f 100644
--- a/lib/pleroma/web/websub/websub.ex
+++ b/lib/pleroma/web/websub/websub.ex
@@ -1,13 +1,11 @@
 defmodule Pleroma.Web.Websub do
   alias Pleroma.Repo
-  alias Pleroma.Web.Websub.WebsubServerSubscription
+  alias Pleroma.Web.Websub.{WebsubServerSubscription, WebsubClientSubscription}
   alias Pleroma.Web.OStatus.FeedRepresenter
   alias Pleroma.Web.OStatus
 
   import Ecto.Query
 
-  @websub_verifier Application.get_env(:pleroma, :websub_verifier)
-
   def verify(subscription, getter \\ &HTTPoison.get/3 ) do
     challenge = Base.encode16(:crypto.strong_rand_bytes(8))
     lease_seconds = NaiveDateTime.diff(subscription.valid_until, subscription.updated_at) |> to_string
@@ -71,8 +69,7 @@ def incoming_subscription_request(user, %{"hub.mode" => "subscribe"} = params) d
       change = Ecto.Changeset.change(websub, %{valid_until: NaiveDateTime.add(websub.updated_at, lease_time)})
       websub = Repo.update!(change)
 
-      # Just spawn that for now, maybe pool later.
-      spawn(fn -> @websub_verifier.verify(websub) end)
+      Pleroma.Web.Federator.enqueue(:verify_websub, websub)
 
       {:ok, websub}
     else {:error, reason} ->
@@ -99,4 +96,23 @@ defp valid_topic(%{"hub.topic" => topic}, user) do
       {:error, "Wrong topic requested, expected #{OStatus.feed_path(user)}, got #{topic}"}
     end
   end
+
+  def subscribe(user, topic) do
+    # Race condition, use transactions
+    {:ok, subscription} = with subscription when not is_nil(subscription) <- Repo.get_by(WebsubClientSubscription, topic: topic) do
+      subscribers = [user.ap_id, subscription.subcribers] |> Enum.uniq
+      change = Ecto.Changeset.change(subscription, %{subscribers: subscribers})
+      Repo.update(change)
+    else _e ->
+      subscription = %WebsubClientSubscription{
+        topic: topic,
+        subscribers: [user.ap_id],
+        state: "requested",
+        secret: :crypto.strong_rand_bytes(8) |> Base.url_encode64
+      }
+      Repo.insert(subscription)
+    end
+
+    {:ok, subscription}
+  end
 end
diff --git a/lib/pleroma/web/websub/websub_client_subscription.ex b/lib/pleroma/web/websub/websub_client_subscription.ex
new file mode 100644
index 000000000..341e27c51
--- /dev/null
+++ b/lib/pleroma/web/websub/websub_client_subscription.ex
@@ -0,0 +1,13 @@
+defmodule Pleroma.Web.Websub.WebsubClientSubscription do
+  use Ecto.Schema
+
+  schema "websub_client_subscriptions" do
+    field :topic, :string
+    field :secret, :string
+    field :valid_until, :naive_datetime
+    field :state, :string
+    field :subscribers, {:array, :string}, default: []
+
+    timestamps()
+  end
+end
diff --git a/priv/repo/migrations/20170426154155_create_websub_client_subscription.exs b/priv/repo/migrations/20170426154155_create_websub_client_subscription.exs
new file mode 100644
index 000000000..f42782840
--- /dev/null
+++ b/priv/repo/migrations/20170426154155_create_websub_client_subscription.exs
@@ -0,0 +1,15 @@
+defmodule Pleroma.Repo.Migrations.CreateWebsubClientSubscription do
+  use Ecto.Migration
+
+  def change do
+    create table(:websub_client_subscriptions) do
+      add :topic, :string
+      add :secret, :string
+      add :valid_until, :naive_datetime
+      add :state, :string
+      add :subscribers, :map
+
+      timestamps()
+    end
+  end
+end
diff --git a/test/web/websub/websub_test.exs b/test/web/websub/websub_test.exs
index 334ba03fc..7b77e696b 100644
--- a/test/web/websub/websub_test.exs
+++ b/test/web/websub/websub_test.exs
@@ -58,7 +58,6 @@ test "an incoming subscription request" do
       "hub.lease_seconds" => "100"
     }
 
-
     {:ok, subscription } = Websub.incoming_subscription_request(user, data)
     assert subscription.topic == Pleroma.Web.OStatus.feed_path(user)
     assert subscription.state == "requested"
@@ -87,4 +86,15 @@ test "an incoming subscription request for an existing subscription" do
     assert length(Repo.all(WebsubServerSubscription)) == 1
     assert subscription.id == sub.id
   end
+
+  test "initiate a subscription for a given user and topic" do
+    user = insert(:user)
+    topic = "http://example.org/some-topic.atom"
+
+    {:ok, websub} = Websub.subscribe(user, topic)
+    assert websub.subscribers == [user.ap_id]
+    assert websub.topic == topic
+    assert is_binary(websub.secret)
+    assert websub.state == "accepted"
+  end
 end