[#570] add user:notification stream

This commit is contained in:
Maksim 2019-06-16 10:33:25 +00:00 committed by rinpatch
parent 57d54a9f09
commit a04bf131e0
6 changed files with 130 additions and 31 deletions

View file

@ -13,6 +13,8 @@ defmodule Pleroma.Notification do
alias Pleroma.User alias Pleroma.User
alias Pleroma.Web.CommonAPI alias Pleroma.Web.CommonAPI
alias Pleroma.Web.CommonAPI.Utils alias Pleroma.Web.CommonAPI.Utils
alias Pleroma.Web.Push
alias Pleroma.Web.Streamer
import Ecto.Query import Ecto.Query
import Ecto.Changeset import Ecto.Changeset
@ -145,8 +147,9 @@ def create_notification(%Activity{} = activity, %User{} = user) do
unless skip?(activity, user) do unless skip?(activity, user) do
notification = %Notification{user_id: user.id, activity: activity} notification = %Notification{user_id: user.id, activity: activity}
{:ok, notification} = Repo.insert(notification) {:ok, notification} = Repo.insert(notification)
Pleroma.Web.Streamer.stream("user", notification) Streamer.stream("user", notification)
Pleroma.Web.Push.send(notification) Streamer.stream("user:notification", notification)
Push.send(notification)
notification notification
end end
end end

View file

@ -17,6 +17,7 @@ defmodule Pleroma.Web.MastodonAPI.WebsocketHandler do
"public:media", "public:media",
"public:local:media", "public:local:media",
"user", "user",
"user:notification",
"direct", "direct",
"list", "list",
"hashtag" "hashtag"

View file

@ -110,23 +110,18 @@ def handle_cast(%{action: :stream, topic: "list", item: item}, topics) do
{:noreply, topics} {:noreply, topics}
end end
def handle_cast(%{action: :stream, topic: "user", item: %Notification{} = item}, topics) do def handle_cast(
topic = "user:#{item.user_id}" %{action: :stream, topic: topic, item: %Notification{} = item},
topics
Enum.each(topics[topic] || [], fn socket -> )
json = when topic in ["user", "user:notification"] do
%{ topics
event: "notification", |> Map.get("#{topic}:#{item.user_id}", [])
payload: |> Enum.each(fn socket ->
NotificationView.render("show.json", %{ send(
notification: item, socket.transport_pid,
for: socket.assigns["user"] {:text, represent_notification(socket.assigns[:user], item)}
}) )
|> Jason.encode!()
}
|> Jason.encode!()
send(socket.transport_pid, {:text, json})
end) end)
{:noreply, topics} {:noreply, topics}
@ -216,6 +211,20 @@ def represent_conversation(%Participation{} = participation) do
|> Jason.encode!() |> Jason.encode!()
end 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 def push_to_socket(topics, topic, %Activity{data: %{"type" => "Announce"}} = item) do
Enum.each(topics[topic] || [], fn socket -> Enum.each(topics[topic] || [], fn socket ->
# Get the current user so we have up-to-date blocks etc. # 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)
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}" "#{topic}:#{socket.assigns[:user].id}"
end end

View file

@ -97,5 +97,15 @@ test "receives well formatted events" do
test "accepts valid tokens", state do test "accepts valid tokens", state do
assert {:ok, _} = start_socket("?stream=user&access_token=#{state.token.token}") assert {:ok, _} = start_socket("?stream=user&access_token=#{state.token.token}")
end 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
end end

View file

@ -8,6 +8,7 @@ defmodule Pleroma.NotificationTest do
alias Pleroma.User alias Pleroma.User
alias Pleroma.Web.ActivityPub.Transmogrifier alias Pleroma.Web.ActivityPub.Transmogrifier
alias Pleroma.Web.CommonAPI alias Pleroma.Web.CommonAPI
alias Pleroma.Web.Streamer
alias Pleroma.Web.TwitterAPI.TwitterAPI alias Pleroma.Web.TwitterAPI.TwitterAPI
import Pleroma.Factory import Pleroma.Factory
@ -44,13 +45,42 @@ test "it creates a notification for subscribed users" do
end end
describe "create_notification" do 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 test "it doesn't create a notification for user if the user blocks the activity author" do
activity = insert(:note_activity) activity = insert(:note_activity)
author = User.get_cached_by_ap_id(activity.data["actor"]) author = User.get_cached_by_ap_id(activity.data["actor"])
user = insert(:user) user = insert(:user)
{:ok, user} = User.block(user, author) {:ok, user} = User.block(user, author)
assert nil == Notification.create_notification(activity, user) refute Notification.create_notification(activity, user)
end end
test "it doesn't create a notificatin for the user if the user mutes the activity author" do 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) muter = Repo.get(User, muter.id)
{:ok, activity} = CommonAPI.post(muted, %{"status" => "Hi @#{muter.nickname}"}) {:ok, activity} = CommonAPI.post(muted, %{"status" => "Hi @#{muter.nickname}"})
assert nil == Notification.create_notification(activity, muter) refute Notification.create_notification(activity, muter)
end end
test "it doesn't create a notification for an activity from a muted thread" do 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 "in_reply_to_status_id" => activity.id
}) })
assert nil == Notification.create_notification(activity, muter) refute Notification.create_notification(activity, muter)
end end
test "it disables notifications from followers" do 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}}) followed = insert(:user, info: %{notification_settings: %{"followers" => false}})
User.follow(follower, followed) User.follow(follower, followed)
{:ok, activity} = CommonAPI.post(follower, %{"status" => "hey @#{followed.nickname}"}) {:ok, activity} = CommonAPI.post(follower, %{"status" => "hey @#{followed.nickname}"})
assert nil == Notification.create_notification(activity, followed) refute Notification.create_notification(activity, followed)
end end
test "it disables notifications from non-followers" do test "it disables notifications from non-followers" do
follower = insert(:user) follower = insert(:user)
followed = insert(:user, info: %{notification_settings: %{"non_followers" => false}}) followed = insert(:user, info: %{notification_settings: %{"non_followers" => false}})
{:ok, activity} = CommonAPI.post(follower, %{"status" => "hey @#{followed.nickname}"}) {:ok, activity} = CommonAPI.post(follower, %{"status" => "hey @#{followed.nickname}"})
assert nil == Notification.create_notification(activity, followed) refute Notification.create_notification(activity, followed)
end end
test "it disables notifications from people the user follows" do 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) User.follow(follower, followed)
follower = Repo.get(User, follower.id) follower = Repo.get(User, follower.id)
{:ok, activity} = CommonAPI.post(followed, %{"status" => "hey @#{follower.nickname}"}) {:ok, activity} = CommonAPI.post(followed, %{"status" => "hey @#{follower.nickname}"})
assert nil == Notification.create_notification(activity, follower) refute Notification.create_notification(activity, follower)
end end
test "it disables notifications from people the user does not follow" do test "it disables notifications from people the user does not follow" do
follower = insert(:user, info: %{notification_settings: %{"non_follows" => false}}) follower = insert(:user, info: %{notification_settings: %{"non_follows" => false}})
followed = insert(:user) followed = insert(:user)
{:ok, activity} = CommonAPI.post(followed, %{"status" => "hey @#{follower.nickname}"}) {:ok, activity} = CommonAPI.post(followed, %{"status" => "hey @#{follower.nickname}"})
assert nil == Notification.create_notification(activity, follower) refute Notification.create_notification(activity, follower)
end end
test "it doesn't create a notification for user if he is the activity author" do test "it doesn't create a notification for user if he is the activity author" do
activity = insert(:note_activity) activity = insert(:note_activity)
author = User.get_cached_by_ap_id(activity.data["actor"]) author = User.get_cached_by_ap_id(activity.data["actor"])
assert nil == Notification.create_notification(activity, author) refute Notification.create_notification(activity, author)
end end
test "it doesn't create a notification for follow-unfollow-follow chains" do 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) Notification.create_notification(activity, followed_user)
TwitterAPI.unfollow(user, %{"user_id" => followed_user.id}) TwitterAPI.unfollow(user, %{"user_id" => followed_user.id})
{:ok, _, _, activity_dupe} = TwitterAPI.follow(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 end
test "it doesn't create a notification for like-unlike-like chains" do 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) Notification.create_notification(fav_status, liked_user)
TwitterAPI.unfav(user, status.id) TwitterAPI.unfav(user, status.id)
{:ok, dupe} = TwitterAPI.fav(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 end
test "it doesn't create a notification for repeat-unrepeat-repeat chains" do 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) Notification.create_notification(retweeted_activity, retweeted_user)
TwitterAPI.unrepeat(user, status.id) TwitterAPI.unrepeat(user, status.id)
{:ok, dupe} = TwitterAPI.repeat(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 end
test "it doesn't create duplicate notifications for follow+subscribed users" do test "it doesn't create duplicate notifications for follow+subscribed users" do

View file

@ -21,6 +21,52 @@ defmodule Pleroma.Web.StreamerTest do
:ok :ok
end 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 test "it sends to public" do
user = insert(:user) user = insert(:user)
other_user = insert(:user) other_user = insert(:user)