From 5d3fdbc08223a1b8b6f36f2621214285c383b840 Mon Sep 17 00:00:00 2001
From: eal <eal@waifu.club>
Date: Wed, 30 May 2018 16:33:37 +0300
Subject: [PATCH] MastoAPI: Add streaming for lists.

---
 lib/pleroma/list.ex                              | 15 ++++++++++++++-
 lib/pleroma/web/activity_pub/activity_pub.ex     |  1 +
 lib/pleroma/web/mastodon_api/mastodon_socket.ex  |  7 +++++--
 lib/pleroma/web/streamer.ex                      | 13 +++++++++++++
 .../20180530123448_add_list_follow_index.exs     |  7 +++++++
 test/list_test.exs                               | 16 ++++++++++++++++
 6 files changed, 56 insertions(+), 3 deletions(-)
 create mode 100644 priv/repo/migrations/20180530123448_add_list_follow_index.exs

diff --git a/lib/pleroma/list.ex b/lib/pleroma/list.ex
index 9d0b9285b..53d98665b 100644
--- a/lib/pleroma/list.ex
+++ b/lib/pleroma/list.ex
@@ -1,7 +1,7 @@
 defmodule Pleroma.List do
   use Ecto.Schema
   import Ecto.{Changeset, Query}
-  alias Pleroma.{User, Repo}
+  alias Pleroma.{User, Repo, Activity}
 
   schema "lists" do
     belongs_to(:user, Pleroma.User)
@@ -56,6 +56,19 @@ def get_following(%Pleroma.List{following: following} = list) do
     {:ok, Repo.all(q)}
   end
 
+  # Get lists the activity should be streamed to.
+  def get_lists_from_activity(%Activity{actor: ap_id}) do
+    actor = User.get_cached_by_ap_id(ap_id)
+
+    query =
+      from(
+        l in Pleroma.List,
+        where: fragment("? && ?", l.following, ^[actor.follower_address])
+      )
+
+    Repo.all(query)
+  end
+
   def rename(%Pleroma.List{} = list, title) do
     list
     |> title_changeset(%{title: title})
diff --git a/lib/pleroma/web/activity_pub/activity_pub.ex b/lib/pleroma/web/activity_pub/activity_pub.ex
index 4e0be5ba2..91e851d54 100644
--- a/lib/pleroma/web/activity_pub/activity_pub.ex
+++ b/lib/pleroma/web/activity_pub/activity_pub.ex
@@ -57,6 +57,7 @@ def stream_out(activity) do
 
     if activity.data["type"] in ["Create", "Announce"] do
       Pleroma.Web.Streamer.stream("user", activity)
+      Pleroma.Web.Streamer.stream("list", activity)
 
       if Enum.member?(activity.data["to"], public) do
         Pleroma.Web.Streamer.stream("public", activity)
diff --git a/lib/pleroma/web/mastodon_api/mastodon_socket.ex b/lib/pleroma/web/mastodon_api/mastodon_socket.ex
index 080f62b31..46648c366 100644
--- a/lib/pleroma/web/mastodon_api/mastodon_socket.ex
+++ b/lib/pleroma/web/mastodon_api/mastodon_socket.ex
@@ -15,10 +15,13 @@ def connect(params, socket) do
     with token when not is_nil(token) <- params["access_token"],
          %Token{user_id: user_id} <- Repo.get_by(Token, token: token),
          %User{} = user <- Repo.get(User, user_id),
-         stream when stream in ["public", "public:local", "user", "direct"] <- params["stream"] do
+         stream when stream in ["public", "public:local", "user", "direct", "list"] <-
+           params["stream"] do
+      topic = if stream == "list", do: "list:#{params["list"]}", else: stream
+
       socket =
         socket
-        |> assign(:topic, params["stream"])
+        |> assign(:topic, topic)
         |> assign(:user, user)
 
       Pleroma.Web.Streamer.add_socket(params["stream"], socket)
diff --git a/lib/pleroma/web/streamer.ex b/lib/pleroma/web/streamer.ex
index 6ed9035fb..0c9e95485 100644
--- a/lib/pleroma/web/streamer.ex
+++ b/lib/pleroma/web/streamer.ex
@@ -59,6 +59,19 @@ def handle_cast(%{action: :stream, topic: "direct", item: item}, topics) do
     {:noreply, topics}
   end
 
+  def handle_cast(%{action: :stream, topic: "list", item: item}, topics) do
+    recipient_topics =
+      Pleroma.List.get_lists_from_activity(item)
+      |> Enum.map(fn %{id: id} -> "list:#{id}" end)
+
+    Enum.each(recipient_topics || [], fn list_topic ->
+      Logger.debug("Trying to push message to #{list_topic}\n\n")
+      push_to_socket(topics, list_topic, item)
+    end)
+
+    {:noreply, topics}
+  end
+
   def handle_cast(%{action: :stream, topic: "user", item: %Notification{} = item}, topics) do
     topic = "user:#{item.user_id}"
 
diff --git a/priv/repo/migrations/20180530123448_add_list_follow_index.exs b/priv/repo/migrations/20180530123448_add_list_follow_index.exs
new file mode 100644
index 000000000..d6603e916
--- /dev/null
+++ b/priv/repo/migrations/20180530123448_add_list_follow_index.exs
@@ -0,0 +1,7 @@
+defmodule Pleroma.Repo.Migrations.AddListFollowIndex do
+  use Ecto.Migration
+
+  def change do
+    create index(:lists, [:following])
+  end
+end
diff --git a/test/list_test.exs b/test/list_test.exs
index ced012093..da3b88024 100644
--- a/test/list_test.exs
+++ b/test/list_test.exs
@@ -74,4 +74,20 @@ test "getting all lists by an user" do
     assert list_two in lists
     refute list_three in lists
   end
+
+  test "getting all lists the user is a member of" do
+    user = insert(:user)
+    other_user = insert(:user)
+    {:ok, list_one} = Pleroma.List.create("title", user)
+    {:ok, list_two} = Pleroma.List.create("other title", user)
+    {:ok, list_three} = Pleroma.List.create("third title", other_user)
+    {:ok, list_one} = Pleroma.List.follow(list_one, other_user)
+    {:ok, list_two} = Pleroma.List.follow(list_two, other_user)
+    {:ok, list_three} = Pleroma.List.follow(list_three, user)
+
+    lists = Pleroma.List.get_lists_from_activity(%Pleroma.Activity{actor: other_user.ap_id})
+    assert list_one in lists
+    assert list_two in lists
+    refute list_three in lists
+  end
 end