No OneTemporary
Actions

Size

19 KB

Referenced Files

None

Subscribers

None

View Options

	diff --git a/lib/pleroma/notification.ex b/lib/pleroma/notification.ex
	index 46f2107b1..f54495ff8 100644
	--- a/lib/pleroma/notification.ex
	+++ b/lib/pleroma/notification.ex
	@@ -1,249 +1,251 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Notification do
	use Ecto.Schema

	alias Pleroma.Activity
	alias Pleroma.Notification
	alias Pleroma.Object
	alias Pleroma.Pagination
	alias Pleroma.Repo
	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

	schema "notifications" do
	field(:seen, :boolean, default: false)
	belongs_to(:user, User, type: Pleroma.FlakeId)
	belongs_to(:activity, Activity, type: Pleroma.FlakeId)

	timestamps()
	end

	def changeset(%Notification{} = notification, attrs) do
	notification
	\|> cast(attrs, [:seen])
	end

	def for_user_query(user) do
	Notification
	\|> where(user_id: ^user.id)
	\|> where(
	[n, a],
	fragment(
	"? not in (SELECT ap_id FROM users WHERE info->'deactivated' @> 'true')",
	a.actor
	)
	)
	\|> join(:inner, [n], activity in assoc(n, :activity))
	\|> join(:left, [n, a], object in Object,
	on:
	fragment(
	"(?->>'id') = COALESCE((? -> 'object'::text) ->> 'id'::text)",
	object.data,
	a.data
	)
	)
	\|> preload([n, a, o], activity: {a, object: o})
	end

	def for_user(user, opts \\ %{}) do
	user
	\|> for_user_query()
	\|> Pagination.fetch_paginated(opts)
	end

	def set_read_up_to(%{id: user_id} = _user, id) do
	query =
	from(
	n in Notification,
	where: n.user_id == ^user_id,
	where: n.id <= ^id,
	update: [
	set: [seen: true]
	]
	)

	Repo.update_all(query, [])
	end

	def read_one(%User{} = user, notification_id) do
	with {:ok, %Notification{} = notification} <- get(user, notification_id) do
	notification
	\|> changeset(%{seen: true})
	\|> Repo.update()
	end
	end

	def get(%{id: user_id} = _user, id) do
	query =
	from(
	n in Notification,
	where: n.id == ^id,
	join: activity in assoc(n, :activity),
	preload: [activity: activity]
	)

	notification = Repo.one(query)

	case notification do
	%{user_id: ^user_id} ->
	{:ok, notification}

	_ ->
	{:error, "Cannot get notification"}
	end
	end

	def clear(user) do
	from(n in Notification, where: n.user_id == ^user.id)
	\|> Repo.delete_all()
	end

	def destroy_multiple(%{id: user_id} = _user, ids) do
	from(n in Notification,
	where: n.id in ^ids,
	where: n.user_id == ^user_id
	)
	\|> Repo.delete_all()
	end

	def dismiss(%{id: user_id} = _user, id) do
	notification = Repo.get(Notification, id)

	case notification do
	%{user_id: ^user_id} ->
	Repo.delete(notification)

	_ ->
	{:error, "Cannot dismiss notification"}
	end
	end

	def create_notifications(%Activity{data: %{"to" => _, "type" => type}} = activity)
	when type in ["Create", "Like", "Announce", "Follow"] do
	object = Object.normalize(activity)

	unless object && object.data["type"] == "Answer" do
	users = get_notified_from_activity(activity)
	notifications = Enum.map(users, fn user -> create_notification(activity, user) end)
	{:ok, notifications}
	else
	{:ok, []}
	end
	end

	def create_notifications(_), do: {:ok, []}

	# TODO move to sql, too.
	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", notification)
	+ Push.send(notification)
	notification
	end
	end

	def get_notified_from_activity(activity, local_only \\ true)

	def get_notified_from_activity(
	%Activity{data: %{"to" => _, "type" => type} = _data} = activity,
	local_only
	)
	when type in ["Create", "Like", "Announce", "Follow"] do
	recipients =
	[]
	\|> Utils.maybe_notify_to_recipients(activity)
	\|> Utils.maybe_notify_mentioned_recipients(activity)
	\|> Utils.maybe_notify_subscribers(activity)
	\|> Enum.uniq()

	User.get_users_from_set(recipients, local_only)
	end

	def get_notified_from_activity(_, _local_only), do: []

	def skip?(activity, user) do
	[
	:self,
	:blocked,
	:muted,
	:followers,
	:follows,
	:non_followers,
	:non_follows,
	:recently_followed
	]
	\|> Enum.any?(&skip?(&1, activity, user))
	end

	def skip?(:self, activity, user) do
	activity.data["actor"] == user.ap_id
	end

	def skip?(:blocked, activity, user) do
	actor = activity.data["actor"]
	User.blocks?(user, %{ap_id: actor})
	end

	def skip?(:muted, activity, user) do
	actor = activity.data["actor"]

	User.mutes?(user, %{ap_id: actor}) or CommonAPI.thread_muted?(user, activity)
	end

	def skip?(
	:followers,
	activity,
	%{info: %{notification_settings: %{"followers" => false}}} = user
	) do
	actor = activity.data["actor"]
	follower = User.get_cached_by_ap_id(actor)
	User.following?(follower, user)
	end

	def skip?(
	:non_followers,
	activity,
	%{info: %{notification_settings: %{"non_followers" => false}}} = user
	) do
	actor = activity.data["actor"]
	follower = User.get_cached_by_ap_id(actor)
	!User.following?(follower, user)
	end

	def skip?(:follows, activity, %{info: %{notification_settings: %{"follows" => false}}} = user) do
	actor = activity.data["actor"]
	followed = User.get_cached_by_ap_id(actor)
	User.following?(user, followed)
	end

	def skip?(
	:non_follows,
	activity,
	%{info: %{notification_settings: %{"non_follows" => false}}} = user
	) do
	actor = activity.data["actor"]
	followed = User.get_cached_by_ap_id(actor)
	!User.following?(user, followed)
	end

	def skip?(:recently_followed, %{data: %{"type" => "Follow"}} = activity, user) do
	actor = activity.data["actor"]

	Notification.for_user(user)
	\|> Enum.any?(fn
	%{activity: %{data: %{"type" => "Follow", "actor" => ^actor}}} -> true
	_ -> false
	end)
	end

	def skip?(_, _, _), do: false
	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
	@@ -1,125 +1,126 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Web.MastodonAPI.WebsocketHandler do
	require Logger

	alias Pleroma.Repo
	alias Pleroma.User
	alias Pleroma.Web.OAuth.Token

	@behaviour :cowboy_websocket

	@streams [
	"public",
	"public:local",
	"public:media",
	"public:local:media",
	"user",
	+ "user:notification",
	"direct",
	"list",
	"hashtag"
	]
	@anonymous_streams ["public", "public:local", "hashtag"]

	# Handled by periodic keepalive in Pleroma.Web.Streamer.
	@timeout :infinity

	def init(%{qs: qs} = req, state) do
	with params <- :cow_qs.parse_qs(qs),
	access_token <- List.keyfind(params, "access_token", 0),
	{_, stream} <- List.keyfind(params, "stream", 0),
	{:ok, user} <- allow_request(stream, access_token),
	topic when is_binary(topic) <- expand_topic(stream, params) do
	{:cowboy_websocket, req, %{user: user, topic: topic}, %{idle_timeout: @timeout}}
	else
	{:error, code} ->
	Logger.debug("#{__MODULE__} denied connection: #{inspect(code)} - #{inspect(req)}")
	{:ok, req} = :cowboy_req.reply(code, req)
	{:ok, req, state}

	error ->
	Logger.debug("#{__MODULE__} denied connection: #{inspect(error)} - #{inspect(req)}")
	{:ok, req} = :cowboy_req.reply(400, req)
	{:ok, req, state}
	end
	end

	def websocket_init(state) do
	send(self(), :subscribe)
	{:ok, state}
	end

	# We never receive messages.
	def websocket_handle(_frame, state) do
	{:ok, state}
	end

	def websocket_info(:subscribe, state) do
	Logger.debug(
	"#{__MODULE__} accepted websocket connection for user #{
	(state.user \|\| %{id: "anonymous"}).id
	}, topic #{state.topic}"
	)

	Pleroma.Web.Streamer.add_socket(state.topic, streamer_socket(state))
	{:ok, state}
	end

	def websocket_info({:text, message}, state) do
	{:reply, {:text, message}, state}
	end

	def terminate(reason, _req, state) do
	Logger.debug(
	"#{__MODULE__} terminating websocket connection for user #{
	(state.user \|\| %{id: "anonymous"}).id
	}, topic #{state.topic \|\| "?"}: #{inspect(reason)}"
	)

	Pleroma.Web.Streamer.remove_socket(state.topic, streamer_socket(state))
	:ok
	end

	# Public streams without authentication.
	defp allow_request(stream, nil) when stream in @anonymous_streams do
	{:ok, nil}
	end

	# Authenticated streams.
	defp allow_request(stream, {"access_token", access_token}) when stream in @streams do
	with %Token{user_id: user_id} <- Repo.get_by(Token, token: access_token),
	user = %User{} <- User.get_cached_by_id(user_id) do
	{:ok, user}
	else
	_ -> {:error, 403}
	end
	end

	# Not authenticated.
	defp allow_request(stream, _) when stream in @streams, do: {:error, 403}

	# No matching stream.
	defp allow_request(_, _), do: {:error, 404}

	defp expand_topic("hashtag", params) do
	case List.keyfind(params, "tag", 0) do
	{_, tag} -> "hashtag:#{tag}"
	_ -> nil
	end
	end

	defp expand_topic("list", params) do
	case List.keyfind(params, "list", 0) do
	{_, list} -> "list:#{list}"
	_ -> nil
	end
	end

	defp expand_topic(topic, _), do: topic

	defp streamer_socket(state) do
	%{transport_pid: self(), assigns: state}
	end
	end
	diff --git a/lib/pleroma/web/streamer.ex b/lib/pleroma/web/streamer.ex
	index a23f80f26..bf732e078 100644
	--- a/lib/pleroma/web/streamer.ex
	+++ b/lib/pleroma/web/streamer.ex
	@@ -1,293 +1,296 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Web.Streamer do
	use GenServer
	require Logger
	alias Pleroma.Activity
	alias Pleroma.Config
	alias Pleroma.Conversation.Participation
	alias Pleroma.Notification
	alias Pleroma.Object
	alias Pleroma.User
	alias Pleroma.Web.ActivityPub.ActivityPub
	alias Pleroma.Web.ActivityPub.Visibility
	alias Pleroma.Web.MastodonAPI.NotificationView

	@keepalive_interval :timer.seconds(30)

	def start_link do
	GenServer.start_link(__MODULE__, %{}, name: __MODULE__)
	end

	def add_socket(topic, socket) do
	GenServer.cast(__MODULE__, %{action: :add, socket: socket, topic: topic})
	end

	def remove_socket(topic, socket) do
	GenServer.cast(__MODULE__, %{action: :remove, socket: socket, topic: topic})
	end

	def stream(topic, item) do
	GenServer.cast(__MODULE__, %{action: :stream, topic: topic, item: item})
	end

	def init(args) do
	spawn(fn ->
	# 30 seconds
	Process.sleep(@keepalive_interval)
	GenServer.cast(__MODULE__, %{action: :ping})
	end)

	{:ok, args}
	end

	def handle_cast(%{action: :ping}, topics) do
	Map.values(topics)
	\|> List.flatten()
	\|> Enum.each(fn socket ->
	Logger.debug("Sending keepalive ping")
	send(socket.transport_pid, {:text, ""})
	end)

	spawn(fn ->
	# 30 seconds
	Process.sleep(@keepalive_interval)
	GenServer.cast(__MODULE__, %{action: :ping})
	end)

	{:noreply, topics}
	end

	def handle_cast(%{action: :stream, topic: "direct", item: item}, topics) do
	recipient_topics =
	User.get_recipients_from_activity(item)
	\|> Enum.map(fn %{id: id} -> "direct:#{id}" end)

	Enum.each(recipient_topics \|\| [], fn user_topic ->
	Logger.debug("Trying to push direct message to #{user_topic}\n\n")
	push_to_socket(topics, user_topic, item)
	end)

	{:noreply, topics}
	end

	def handle_cast(%{action: :stream, topic: "participation", item: participation}, topics) do
	user_topic = "direct:#{participation.user_id}"
	Logger.debug("Trying to push a conversation participation to #{user_topic}\n\n")

	push_to_socket(topics, user_topic, participation)

	{:noreply, topics}
	end

	def handle_cast(%{action: :stream, topic: "list", item: item}, topics) do
	# filter the recipient list if the activity is not public, see #270.
	recipient_lists =
	case Visibility.is_public?(item) do
	true ->
	Pleroma.List.get_lists_from_activity(item)

	_ ->
	Pleroma.List.get_lists_from_activity(item)
	\|> Enum.filter(fn list ->
	owner = User.get_cached_by_id(list.user_id)

	Visibility.visible_for_user?(item, owner)
	end)
	end

	recipient_topics =
	recipient_lists
	\|> 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
	+ def handle_cast(
	+ %{action: :stream, topic: "user:notification", 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})
	end)

	{:noreply, topics}
	end

	def handle_cast(%{action: :stream, topic: "user", item: item}, topics) do
	Logger.debug("Trying to push to users")

	recipient_topics =
	User.get_recipients_from_activity(item)
	\|> Enum.map(fn %{id: id} -> "user:#{id}" end)

	Enum.each(recipient_topics, fn topic ->
	push_to_socket(topics, topic, item)
	end)

	{:noreply, topics}
	end

	def handle_cast(%{action: :stream, topic: topic, item: item}, topics) do
	Logger.debug("Trying to push to #{topic}")
	Logger.debug("Pushing item to #{topic}")
	push_to_socket(topics, topic, item)
	{:noreply, topics}
	end

	def handle_cast(%{action: :add, topic: topic, socket: socket}, sockets) do
	topic = internal_topic(topic, socket)
	sockets_for_topic = sockets[topic] \|\| []
	sockets_for_topic = Enum.uniq([socket \| sockets_for_topic])
	sockets = Map.put(sockets, topic, sockets_for_topic)
	Logger.debug("Got new conn for #{topic}")
	{:noreply, sockets}
	end

	def handle_cast(%{action: :remove, topic: topic, socket: socket}, sockets) do
	topic = internal_topic(topic, socket)
	sockets_for_topic = sockets[topic] \|\| []
	sockets_for_topic = List.delete(sockets_for_topic, socket)
	sockets = Map.put(sockets, topic, sockets_for_topic)
	Logger.debug("Removed conn for #{topic}")
	{:noreply, sockets}
	end

	def handle_cast(m, state) do
	Logger.info("Unknown: #{inspect(m)}, #{inspect(state)}")
	{:noreply, state}
	end

	defp represent_update(%Activity{} = activity, %User{} = user) do
	%{
	event: "update",
	payload:
	Pleroma.Web.MastodonAPI.StatusView.render(
	"status.json",
	activity: activity,
	for: user
	)
	\|> Jason.encode!()
	}
	\|> Jason.encode!()
	end

	defp represent_update(%Activity{} = activity) do
	%{
	event: "update",
	payload:
	Pleroma.Web.MastodonAPI.StatusView.render(
	"status.json",
	activity: activity
	)
	\|> Jason.encode!()
	}
	\|> Jason.encode!()
	end

	def represent_conversation(%Participation{} = participation) do
	%{
	event: "conversation",
	payload:
	Pleroma.Web.MastodonAPI.ConversationView.render("participation.json", %{
	participation: participation,
	user: participation.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.
	if socket.assigns[:user] do
	user = User.get_cached_by_ap_id(socket.assigns[:user].ap_id)
	blocks = user.info.blocks \|\| []
	mutes = user.info.mutes \|\| []
	reblog_mutes = user.info.muted_reblogs \|\| []

	with parent when not is_nil(parent) <- Object.normalize(item),
	true <- Enum.all?([blocks, mutes, reblog_mutes], &(item.actor not in &1)),
	true <- Enum.all?([blocks, mutes], &(parent.data["actor"] not in &1)),
	true <- thread_containment(item, user) do
	send(socket.transport_pid, {:text, represent_update(item, user)})
	end
	else
	send(socket.transport_pid, {:text, represent_update(item)})
	end
	end)
	end

	def push_to_socket(topics, topic, %Participation{} = participation) do
	Enum.each(topics[topic] \|\| [], fn socket ->
	send(socket.transport_pid, {:text, represent_conversation(participation)})
	end)
	end

	def push_to_socket(topics, topic, %Activity{
	data: %{"type" => "Delete", "deleted_activity_id" => deleted_activity_id}
	}) do
	Enum.each(topics[topic] \|\| [], fn socket ->
	send(
	socket.transport_pid,
	{:text, %{event: "delete", payload: to_string(deleted_activity_id)} \|> Jason.encode!()}
	)
	end)
	end

	def push_to_socket(_topics, _topic, %Activity{data: %{"type" => "Delete"}}), do: :noop

	def push_to_socket(topics, topic, item) do
	Enum.each(topics[topic] \|\| [], fn socket ->
	# Get the current user so we have up-to-date blocks etc.
	if socket.assigns[:user] do
	user = User.get_cached_by_ap_id(socket.assigns[:user].ap_id)
	blocks = user.info.blocks \|\| []
	mutes = user.info.mutes \|\| []

	with true <- Enum.all?([blocks, mutes], &(item.actor not in &1)),
	true <- thread_containment(item, user) do
	send(socket.transport_pid, {:text, represent_update(item, user)})
	end
	else
	send(socket.transport_pid, {:text, represent_update(item)})
	end
	end)
	end

	defp internal_topic(topic, socket) when topic in ~w[user direct] do
	"#{topic}:#{socket.assigns[:user].id}"
	end

	defp internal_topic(topic, _), do: topic

	@spec thread_containment(Activity.t(), User.t()) :: boolean()
	defp thread_containment(_activity, %User{info: %{skip_thread_containment: true}}), do: true

	defp thread_containment(activity, user) do
	if Config.get([:instance, :skip_thread_containment]) do
	true
	else
	ActivityPub.contain_activity(activity, user)
	end
	end
	end

File Metadata

Mime Type: text/x-diff
Expires: Sun, Aug 9, 6:31 AM (1 d, 16 h)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 1724287
Default Alt Text: (19 KB)

No OneTemporaryActions

View Options

File Metadata

Event Timeline

No OneTemporary
Actions