From a04bf131e052f12c82e09b22c5e942e99c36d0ee Mon Sep 17 00:00:00 2001 From: Maksim Date: Sun, 16 Jun 2019 10:33:25 +0000 Subject: [PATCH] [#570] add user:notification stream --- lib/pleroma/notification.ex | 7 ++- .../web/mastodon_api/websocket_handler.ex | 1 + lib/pleroma/web/streamer.ex | 45 +++++++++------- test/integration/mastodon_websocket_test.exs | 10 ++++ test/notification_test.exs | 52 +++++++++++++++---- test/web/streamer_test.exs | 46 ++++++++++++++++ 6 files changed, 130 insertions(+), 31 deletions(-) diff --git a/lib/pleroma/notification.ex b/lib/pleroma/notification.ex index 46f2107b1..e25692006 100644 --- a/lib/pleroma/notification.ex +++ b/lib/pleroma/notification.ex @@ -13,6 +13,8 @@ defmodule Pleroma.Notification do alias Pleroma.User alias Pleroma.Web.CommonAPI alias Pleroma.Web.CommonAPI.Utils + alias Pleroma.Web.Push + alias Pleroma.Web.Streamer import Ecto.Query import Ecto.Changeset @@ -145,8 +147,9 @@ def create_notification(%Activity{} = activity, %User{} = user) do unless skip?(activity, user) do notification = %Notification{user_id: user.id, activity: activity} {:ok, notification} = Repo.insert(notification) - Pleroma.Web.Streamer.stream("user", notification) - Pleroma.Web.Push.send(notification) + Streamer.stream("user", notification) + Streamer.stream("user:notification", notification) + Push.send(notification) notification end end diff --git a/lib/pleroma/web/mastodon_api/websocket_handler.ex b/lib/pleroma/web/mastodon_api/websocket_handler.ex index abfa26754..3299e1721 100644 --- a/lib/pleroma/web/mastodon_api/websocket_handler.ex +++ b/lib/pleroma/web/mastodon_api/websocket_handler.ex @@ -17,6 +17,7 @@ defmodule Pleroma.Web.MastodonAPI.WebsocketHandler do "public:media", "public:local:media", "user", + "user:notification", "direct", "list", "hashtag" diff --git a/lib/pleroma/web/streamer.ex b/lib/pleroma/web/streamer.ex index a23f80f26..4f325113a 100644 --- a/lib/pleroma/web/streamer.ex +++ b/lib/pleroma/web/streamer.ex @@ -110,23 +110,18 @@ def handle_cast(%{action: :stream, topic: "list", item: item}, topics) do {:noreply, topics} end - def handle_cast(%{action: :stream, topic: "user", item: %Notification{} = item}, topics) do - topic = "user:#{item.user_id}" - - Enum.each(topics[topic] || [], fn socket -> - json = - %{ - event: "notification", - payload: - NotificationView.render("show.json", %{ - notification: item, - for: socket.assigns["user"] - }) - |> Jason.encode!() - } - |> Jason.encode!() - - send(socket.transport_pid, {:text, json}) + def handle_cast( + %{action: :stream, topic: topic, item: %Notification{} = item}, + topics + ) + when topic in ["user", "user:notification"] do + topics + |> Map.get("#{topic}:#{item.user_id}", []) + |> Enum.each(fn socket -> + send( + socket.transport_pid, + {:text, represent_notification(socket.assigns[:user], item)} + ) end) {:noreply, topics} @@ -216,6 +211,20 @@ def represent_conversation(%Participation{} = participation) do |> Jason.encode!() end + @spec represent_notification(User.t(), Notification.t()) :: binary() + defp represent_notification(%User{} = user, %Notification{} = notify) do + %{ + event: "notification", + payload: + NotificationView.render( + "show.json", + %{notification: notify, for: user} + ) + |> Jason.encode!() + } + |> Jason.encode!() + end + def push_to_socket(topics, topic, %Activity{data: %{"type" => "Announce"}} = item) do Enum.each(topics[topic] || [], fn socket -> # Get the current user so we have up-to-date blocks etc. @@ -274,7 +283,7 @@ def push_to_socket(topics, topic, item) do end) end - defp internal_topic(topic, socket) when topic in ~w[user direct] do + defp internal_topic(topic, socket) when topic in ~w[user user:notification direct] do "#{topic}:#{socket.assigns[:user].id}" end diff --git a/test/integration/mastodon_websocket_test.exs b/test/integration/mastodon_websocket_test.exs index b42c9ef07..a604713d8 100644 --- a/test/integration/mastodon_websocket_test.exs +++ b/test/integration/mastodon_websocket_test.exs @@ -97,5 +97,15 @@ test "receives well formatted events" do test "accepts valid tokens", state do assert {:ok, _} = start_socket("?stream=user&access_token=#{state.token.token}") end + + test "accepts the 'user' stream", %{token: token} = _state do + assert {:ok, _} = start_socket("?stream=user&access_token=#{token.token}") + assert {:error, {403, "Forbidden"}} = start_socket("?stream=user") + end + + test "accepts the 'user:notification' stream", %{token: token} = _state do + assert {:ok, _} = start_socket("?stream=user:notification&access_token=#{token.token}") + assert {:error, {403, "Forbidden"}} = start_socket("?stream=user:notification") + end end end diff --git a/test/notification_test.exs b/test/notification_test.exs index be292abd9..1d36f14bf 100644 --- a/test/notification_test.exs +++ b/test/notification_test.exs @@ -8,6 +8,7 @@ defmodule Pleroma.NotificationTest do alias Pleroma.User alias Pleroma.Web.ActivityPub.Transmogrifier alias Pleroma.Web.CommonAPI + alias Pleroma.Web.Streamer alias Pleroma.Web.TwitterAPI.TwitterAPI import Pleroma.Factory @@ -44,13 +45,42 @@ test "it creates a notification for subscribed users" do end describe "create_notification" do + setup do + GenServer.start(Streamer, %{}, name: Streamer) + + on_exit(fn -> + if pid = Process.whereis(Streamer) do + Process.exit(pid, :kill) + end + end) + end + + test "it creates a notification for user and send to the 'user' and the 'user:notification' stream" do + user = insert(:user) + task = Task.async(fn -> assert_receive {:text, _}, 4_000 end) + task_user_notification = Task.async(fn -> assert_receive {:text, _}, 4_000 end) + Streamer.add_socket("user", %{transport_pid: task.pid, assigns: %{user: user}}) + + Streamer.add_socket( + "user:notification", + %{transport_pid: task_user_notification.pid, assigns: %{user: user}} + ) + + activity = insert(:note_activity) + + notify = Notification.create_notification(activity, user) + assert notify.user_id == user.id + Task.await(task) + Task.await(task_user_notification) + end + test "it doesn't create a notification for user if the user blocks the activity author" do activity = insert(:note_activity) author = User.get_cached_by_ap_id(activity.data["actor"]) user = insert(:user) {:ok, user} = User.block(user, author) - assert nil == Notification.create_notification(activity, user) + refute Notification.create_notification(activity, user) end test "it doesn't create a notificatin for the user if the user mutes the activity author" do @@ -60,7 +90,7 @@ test "it doesn't create a notificatin for the user if the user mutes the activit muter = Repo.get(User, muter.id) {:ok, activity} = CommonAPI.post(muted, %{"status" => "Hi @#{muter.nickname}"}) - assert nil == Notification.create_notification(activity, muter) + refute Notification.create_notification(activity, muter) end test "it doesn't create a notification for an activity from a muted thread" do @@ -75,7 +105,7 @@ test "it doesn't create a notification for an activity from a muted thread" do "in_reply_to_status_id" => activity.id }) - assert nil == Notification.create_notification(activity, muter) + refute Notification.create_notification(activity, muter) end test "it disables notifications from followers" do @@ -83,14 +113,14 @@ test "it disables notifications from followers" do followed = insert(:user, info: %{notification_settings: %{"followers" => false}}) User.follow(follower, followed) {:ok, activity} = CommonAPI.post(follower, %{"status" => "hey @#{followed.nickname}"}) - assert nil == Notification.create_notification(activity, followed) + refute Notification.create_notification(activity, followed) end test "it disables notifications from non-followers" do follower = insert(:user) followed = insert(:user, info: %{notification_settings: %{"non_followers" => false}}) {:ok, activity} = CommonAPI.post(follower, %{"status" => "hey @#{followed.nickname}"}) - assert nil == Notification.create_notification(activity, followed) + refute Notification.create_notification(activity, followed) end test "it disables notifications from people the user follows" do @@ -99,21 +129,21 @@ test "it disables notifications from people the user follows" do User.follow(follower, followed) follower = Repo.get(User, follower.id) {:ok, activity} = CommonAPI.post(followed, %{"status" => "hey @#{follower.nickname}"}) - assert nil == Notification.create_notification(activity, follower) + refute Notification.create_notification(activity, follower) end test "it disables notifications from people the user does not follow" do follower = insert(:user, info: %{notification_settings: %{"non_follows" => false}}) followed = insert(:user) {:ok, activity} = CommonAPI.post(followed, %{"status" => "hey @#{follower.nickname}"}) - assert nil == Notification.create_notification(activity, follower) + refute Notification.create_notification(activity, follower) end test "it doesn't create a notification for user if he is the activity author" do activity = insert(:note_activity) author = User.get_cached_by_ap_id(activity.data["actor"]) - assert nil == Notification.create_notification(activity, author) + refute Notification.create_notification(activity, author) end test "it doesn't create a notification for follow-unfollow-follow chains" do @@ -123,7 +153,7 @@ test "it doesn't create a notification for follow-unfollow-follow chains" do Notification.create_notification(activity, followed_user) TwitterAPI.unfollow(user, %{"user_id" => followed_user.id}) {:ok, _, _, activity_dupe} = TwitterAPI.follow(user, %{"user_id" => followed_user.id}) - assert nil == Notification.create_notification(activity_dupe, followed_user) + refute Notification.create_notification(activity_dupe, followed_user) end test "it doesn't create a notification for like-unlike-like chains" do @@ -134,7 +164,7 @@ test "it doesn't create a notification for like-unlike-like chains" do Notification.create_notification(fav_status, liked_user) TwitterAPI.unfav(user, status.id) {:ok, dupe} = TwitterAPI.fav(user, status.id) - assert nil == Notification.create_notification(dupe, liked_user) + refute Notification.create_notification(dupe, liked_user) end test "it doesn't create a notification for repeat-unrepeat-repeat chains" do @@ -150,7 +180,7 @@ test "it doesn't create a notification for repeat-unrepeat-repeat chains" do Notification.create_notification(retweeted_activity, retweeted_user) TwitterAPI.unrepeat(user, status.id) {:ok, dupe} = TwitterAPI.repeat(user, status.id) - assert nil == Notification.create_notification(dupe, retweeted_user) + refute Notification.create_notification(dupe, retweeted_user) end test "it doesn't create duplicate notifications for follow+subscribed users" do diff --git a/test/web/streamer_test.exs b/test/web/streamer_test.exs index c18b9f9fe..648e28712 100644 --- a/test/web/streamer_test.exs +++ b/test/web/streamer_test.exs @@ -21,6 +21,52 @@ defmodule Pleroma.Web.StreamerTest do :ok end + describe "user streams" do + setup do + GenServer.start(Streamer, %{}, name: Streamer) + + on_exit(fn -> + if pid = Process.whereis(Streamer) do + Process.exit(pid, :kill) + end + end) + + user = insert(:user) + notify = insert(:notification, user: user, activity: build(:note_activity)) + {:ok, %{user: user, notify: notify}} + end + + test "it sends notify to in the 'user' stream", %{user: user, notify: notify} do + task = + Task.async(fn -> + assert_receive {:text, _}, 4_000 + end) + + Streamer.add_socket( + "user", + %{transport_pid: task.pid, assigns: %{user: user}} + ) + + Streamer.stream("user", notify) + Task.await(task) + end + + test "it sends notify to in the 'user:notification' stream", %{user: user, notify: notify} do + task = + Task.async(fn -> + assert_receive {:text, _}, 4_000 + end) + + Streamer.add_socket( + "user:notification", + %{transport_pid: task.pid, assigns: %{user: user}} + ) + + Streamer.stream("user:notification", notify) + Task.await(task) + end + end + test "it sends to public" do user = insert(:user) other_user = insert(:user)