From 8cbad5500cefbba1e0bb67604960fc331b75b498 Mon Sep 17 00:00:00 2001 From: Maksim Pechnikov Date: Wed, 4 Sep 2019 15:25:12 +0300 Subject: add tests for activity_pub/utils.ex --- lib/pleroma/user.ex | 1 + lib/pleroma/web/activity_pub/activity_pub.ex | 12 +- lib/pleroma/web/activity_pub/utils.ex | 300 +++++++++++++-------------- 3 files changed, 144 insertions(+), 169 deletions(-) (limited to 'lib') diff --git a/lib/pleroma/user.ex b/lib/pleroma/user.ex index 29fd6d2ea..424ed772f 100644 --- a/lib/pleroma/user.ex +++ b/lib/pleroma/user.ex @@ -147,6 +147,7 @@ defmodule Pleroma.User do Cachex.fetch!(:user_cache, key, fn _ -> {:commit, follow_state(user, target)} end) end + @spec set_follow_state_cache(String.t(), String.t(), String.t()) :: {:ok | :error, boolean()} def set_follow_state_cache(user_ap_id, target_ap_id, state) do Cachex.put( :user_cache, diff --git a/lib/pleroma/web/activity_pub/activity_pub.ex b/lib/pleroma/web/activity_pub/activity_pub.ex index eeb826814..39b46a595 100644 --- a/lib/pleroma/web/activity_pub/activity_pub.ex +++ b/lib/pleroma/web/activity_pub/activity_pub.ex @@ -435,6 +435,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do end end + @spec block(User.t(), User.t(), String.t() | nil, boolean) :: {:ok, Activity.t() | nil} def block(blocker, blocked, activity_id \\ nil, local \\ true) do outgoing_blocks = Config.get([:activitypub, :outgoing_blocks]) unfollow_blocked = Config.get([:activitypub, :unfollow_blocked]) @@ -463,10 +464,11 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do end end + @spec flag(map()) :: {:ok, Activity.t()} | any def flag( %{ actor: actor, - context: context, + context: _context, account: account, statuses: statuses, content: content @@ -478,14 +480,6 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do additional = params[:additional] || %{} - params = %{ - actor: actor, - context: context, - account: account, - statuses: statuses, - content: content - } - additional = if forward do Map.merge(additional, %{"to" => [], "cc" => [account.ap_id]}) diff --git a/lib/pleroma/web/activity_pub/utils.ex b/lib/pleroma/web/activity_pub/utils.ex index c9c0c3763..cf82d1a9b 100644 --- a/lib/pleroma/web/activity_pub/utils.ex +++ b/lib/pleroma/web/activity_pub/utils.ex @@ -33,50 +33,40 @@ defmodule Pleroma.Web.ActivityPub.Utils do Map.put(params, "actor", get_ap_id(params["actor"])) end - def determine_explicit_mentions(%{"tag" => tag} = _object) when is_list(tag) do - tag - |> Enum.filter(fn x -> is_map(x) end) - |> Enum.filter(fn x -> x["type"] == "Mention" end) - |> Enum.map(fn x -> x["href"] end) + @spec determine_explicit_mentions(map()) :: map() + def determine_explicit_mentions(%{"tag" => tag} = _) when is_list(tag) do + Enum.flat_map(tag, fn + %{"type" => "Mention", "href" => href} -> [href] + _ -> [] + end) end def determine_explicit_mentions(%{"tag" => tag} = object) when is_map(tag) do - Map.put(object, "tag", [tag]) + object + |> Map.put("tag", [tag]) |> determine_explicit_mentions() end def determine_explicit_mentions(_), do: [] + @spec recipient_in_collection(any(), any()) :: boolean() defp recipient_in_collection(ap_id, coll) when is_binary(coll), do: ap_id == coll defp recipient_in_collection(ap_id, coll) when is_list(coll), do: ap_id in coll defp recipient_in_collection(_, _), do: false + @spec recipient_in_message(User.t(), User.t(), map()) :: boolean() def recipient_in_message(%User{ap_id: ap_id} = recipient, %User{} = actor, params) do - cond do - recipient_in_collection(ap_id, params["to"]) -> - true - - recipient_in_collection(ap_id, params["cc"]) -> - true - - recipient_in_collection(ap_id, params["bto"]) -> - true - - recipient_in_collection(ap_id, params["bcc"]) -> - true + addresses = [params["to"], params["cc"], params["bto"], params["bcc"]] + cond do + Enum.any?(addresses, &recipient_in_collection(ap_id, &1)) -> true # if the message is unaddressed at all, then assume it is directly addressed # to the recipient - !params["to"] && !params["cc"] && !params["bto"] && !params["bcc"] -> - true - + Enum.all?(addresses, &is_nil(&1)) -> true # if the message is sent from somebody the user is following, then assume it # is addressed to the recipient - User.following?(recipient, actor) -> - true - - true -> - false + User.following?(recipient, actor) -> true + true -> false end end @@ -188,54 +178,59 @@ defmodule Pleroma.Web.ActivityPub.Utils do Adds an id and a published data if they aren't there, also adds it to an included object """ - def lazy_put_activity_defaults(map, fake \\ false) do - map = - unless fake do - %{data: %{"id" => context}, id: context_id} = create_context(map["context"]) + @spec lazy_put_activity_defaults(map(), boolean) :: map() + def lazy_put_activity_defaults(map, fake \\ false) - map - |> Map.put_new_lazy("id", &generate_activity_id/0) - |> Map.put_new_lazy("published", &make_date/0) - |> Map.put_new("context", context) - |> Map.put_new("context_id", context_id) - else - map - |> Map.put_new("id", "pleroma:fakeid") - |> Map.put_new_lazy("published", &make_date/0) - |> Map.put_new("context", "pleroma:fakecontext") - |> Map.put_new("context_id", -1) - end - - if is_map(map["object"]) do - object = lazy_put_object_defaults(map["object"], map, fake) - %{map | "object" => object} - else - map - end + def lazy_put_activity_defaults(map, true) do + map + |> Map.put_new("id", "pleroma:fakeid") + |> Map.put_new_lazy("published", &make_date/0) + |> Map.put_new("context", "pleroma:fakecontext") + |> Map.put_new("context_id", -1) + |> lazy_put_object_defaults(true) end - @doc """ - Adds an id and published date if they aren't there. - """ - def lazy_put_object_defaults(map, activity \\ %{}, fake) + def lazy_put_activity_defaults(map, _fake) do + %{data: %{"id" => context}, id: context_id} = create_context(map["context"]) - def lazy_put_object_defaults(map, activity, true = _fake) do map + |> Map.put_new_lazy("id", &generate_activity_id/0) |> Map.put_new_lazy("published", &make_date/0) - |> Map.put_new("id", "pleroma:fake_object_id") - |> Map.put_new("context", activity["context"]) - |> Map.put_new("fake", true) - |> Map.put_new("context_id", activity["context_id"]) + |> Map.put_new("context", context) + |> Map.put_new("context_id", context_id) + |> lazy_put_object_defaults(false) end - def lazy_put_object_defaults(map, activity, _fake) do - map - |> Map.put_new_lazy("id", &generate_object_id/0) - |> Map.put_new_lazy("published", &make_date/0) - |> Map.put_new("context", activity["context"]) - |> Map.put_new("context_id", activity["context_id"]) + # Adds an id and published date if they aren't there. + # + @spec lazy_put_object_defaults(map(), boolean()) :: map() + defp lazy_put_object_defaults(%{"object" => map} = activity, true) + when is_map(map) do + object = + map + |> Map.put_new("id", "pleroma:fake_object_id") + |> Map.put_new_lazy("published", &make_date/0) + |> Map.put_new("context", activity["context"]) + |> Map.put_new("context_id", activity["context_id"]) + |> Map.put_new("fake", true) + + %{activity | "object" => object} + end + + defp lazy_put_object_defaults(%{"object" => map} = activity, _) + when is_map(map) do + object = + map + |> Map.put_new_lazy("id", &generate_object_id/0) + |> Map.put_new_lazy("published", &make_date/0) + |> Map.put_new("context", activity["context"]) + |> Map.put_new("context_id", activity["context_id"]) + + %{activity | "object" => object} end + defp lazy_put_object_defaults(activity, _), do: activity + @doc """ Inserts a full object if it is contained in an activity. """ @@ -356,23 +351,30 @@ defmodule Pleroma.Web.ActivityPub.Utils do @doc """ Updates a follow activity's state (for locked accounts). """ + @spec update_follow_state_for_all(Activity.t(), String.t()) :: {:ok, Activity} | {:error, any()} def update_follow_state_for_all( %Activity{data: %{"actor" => actor, "object" => object}} = activity, state ) do - try do - Ecto.Adapters.SQL.query!( - Repo, - "UPDATE activities SET data = jsonb_set(data, '{state}', $1) WHERE data->>'type' = 'Follow' AND data->>'actor' = $2 AND data->>'object' = $3 AND data->>'state' = 'pending'", - [state, actor, object] + query = + from(activity in Activity, + where: fragment("data->>'type' = 'Follow'"), + where: fragment("data->>'state' = 'pending'"), + where: fragment("data->>'actor' = ?", ^actor), + where: fragment("data->>'object' = ?", ^object), + update: [ + set: [ + data: fragment("jsonb_set(data, '{state}', ?)", ^state) + ] + ] ) - User.set_follow_state_cache(actor, object, state) - activity = Activity.get_by_id(activity.id) + with {_, _} <- Repo.update_all(query, []), + {_, _} <- User.set_follow_state_cache(actor, object, state), + %Activity{} = activity <- Activity.get_by_id(activity.id) do {:ok, activity} - rescue - e -> - {:error, e} + else + e -> {:error, e} end end @@ -380,9 +382,7 @@ defmodule Pleroma.Web.ActivityPub.Utils do %Activity{data: %{"actor" => actor, "object" => object}} = activity, state ) do - with new_data <- - activity.data - |> Map.put("state", state), + with new_data <- Map.put(activity.data, "state", state), changeset <- Changeset.change(activity, data: new_data), {:ok, activity} <- Repo.update(changeset), _ <- User.set_follow_state_cache(actor, object, state) do @@ -411,27 +411,17 @@ defmodule Pleroma.Web.ActivityPub.Utils do def fetch_latest_follow(%User{ap_id: follower_id}, %User{ap_id: followed_id}) do query = - from( - activity in Activity, - where: - fragment( - "? ->> 'type' = 'Follow'", - activity.data - ), - where: activity.actor == ^follower_id, - # this is to use the index - where: - fragment( - "coalesce((?)->'object'->>'id', (?)->>'object') = ?", - activity.data, - activity.data, - ^followed_id - ), - order_by: [fragment("? desc nulls last", activity.id)], - limit: 1 - ) - - Repo.one(query) + follower_id + |> Activity.Queries.by_actor() + |> Activity.Queries.by_type("Follow") + |> Activity.Queries.by_object_id(followed_id) + |> Activity.Queries.limit(1) + + from( + activity in query, + order_by: [fragment("? desc nulls last", activity.id)] + ) + |> Repo.one() end #### Announce-related helpers @@ -439,23 +429,14 @@ defmodule Pleroma.Web.ActivityPub.Utils do @doc """ Retruns an existing announce activity if the notice has already been announced """ + @spec get_existing_announce(String.t(), map()) :: Activity.t() | nil def get_existing_announce(actor, %{data: %{"id" => id}}) do - query = - from( - activity in Activity, - where: activity.actor == ^actor, - # this is to use the index - where: - fragment( - "coalesce((?)->'object'->>'id', (?)->>'object') = ?", - activity.data, - activity.data, - ^id - ), - where: fragment("(?)->>'type' = 'Announce'", activity.data) - ) - - Repo.one(query) + actor + |> Activity.Queries.by_actor() + |> Activity.Queries.by_type("Announce") + |> Activity.Queries.by_object_id(id) + |> Activity.Queries.limit(1) + |> Repo.one() end @doc """ @@ -531,31 +512,35 @@ defmodule Pleroma.Web.ActivityPub.Utils do |> maybe_put("id", activity_id) end + @spec add_announce_to_object(Activity.t(), Object.t()) :: + {:ok, Object.t()} | {:error, Ecto.Changeset.t()} def add_announce_to_object( - %Activity{ - data: %{"actor" => actor, "cc" => [Pleroma.Constants.as_public()]} - }, + %Activity{data: %{"actor" => actor, "cc" => [Pleroma.Constants.as_public()]}}, object ) do - announcements = - if is_list(object.data["announcements"]), do: object.data["announcements"], else: [] + announcements = fetch_announcements(object) - with announcements <- [actor | announcements] |> Enum.uniq() do + with announcements <- Enum.uniq([actor | announcements]) do update_element_in_object("announcement", announcements, object) end end def add_announce_to_object(_, object), do: {:ok, object} + @spec remove_announce_from_object(Activity.t(), Object.t()) :: + {:ok, Object.t()} | {:error, Ecto.Changeset.t()} def remove_announce_from_object(%Activity{data: %{"actor" => actor}}, object) do - announcements = - if is_list(object.data["announcements"]), do: object.data["announcements"], else: [] - - with announcements <- announcements |> List.delete(actor) do + with announcements <- List.delete(fetch_announcements(object), actor) do update_element_in_object("announcement", announcements, object) end end + defp fetch_announcements(%{data: %{"announcements" => announcements}} = _) + when is_list(announcements), + do: announcements + + defp fetch_announcements(_), do: [] + #### Unfollow-related helpers def make_unfollow_data(follower, followed, follow_activity, activity_id) do @@ -569,29 +554,20 @@ defmodule Pleroma.Web.ActivityPub.Utils do end #### Block-related helpers + @spec fetch_latest_block(User.t(), User.t()) :: Activity.t() | nil def fetch_latest_block(%User{ap_id: blocker_id}, %User{ap_id: blocked_id}) do query = - from( - activity in Activity, - where: - fragment( - "? ->> 'type' = 'Block'", - activity.data - ), - where: activity.actor == ^blocker_id, - # this is to use the index - where: - fragment( - "coalesce((?)->'object'->>'id', (?)->>'object') = ?", - activity.data, - activity.data, - ^blocked_id - ), - order_by: [fragment("? desc nulls last", activity.id)], - limit: 1 - ) - - Repo.one(query) + blocker_id + |> Activity.Queries.by_actor() + |> Activity.Queries.by_type("Block") + |> Activity.Queries.by_object_id(blocked_id) + |> Activity.Queries.limit(1) + + from( + activity in query, + order_by: [fragment("? desc nulls last", activity.id)] + ) + |> Repo.one() end def make_block_data(blocker, blocked, activity_id) do @@ -631,28 +607,32 @@ defmodule Pleroma.Web.ActivityPub.Utils do end #### Flag-related helpers - - def make_flag_data(params, additional) do - status_ap_ids = - Enum.map(params.statuses || [], fn - %Activity{} = act -> act.data["id"] - act when is_map(act) -> act["id"] - act when is_binary(act) -> act - end) - - object = [params.account.ap_id] ++ status_ap_ids - + @spec make_flag_data(map(), map()) :: map() + def make_flag_data(%{actor: actor, context: context, content: content} = params, additional) do %{ "type" => "Flag", - "actor" => params.actor.ap_id, - "content" => params.content, - "object" => object, - "context" => params.context, + "actor" => actor.ap_id, + "content" => content, + "object" => build_flag_object(params), + "context" => context, "state" => "open" } |> Map.merge(additional) end + def make_flag_data(_, _), do: %{} + + defp build_flag_object(%{account: account, statuses: statuses} = _) do + [account.ap_id] ++ + Enum.map(statuses || [], fn + %Activity{} = act -> act.data["id"] + act when is_map(act) -> act["id"] + act when is_binary(act) -> act + end) + end + + defp build_flag_object(_), do: [] + @doc """ Fetches the OrderedCollection/OrderedCollectionPage from `from`, limiting the amount of pages fetched after the first one to `pages_left` pages. -- cgit v1.2.3 From a890451187f0b1507be96ccf144b18fdb8294dd8 Mon Sep 17 00:00:00 2001 From: Maksim Pechnikov Date: Wed, 4 Sep 2019 17:42:27 +0300 Subject: fetch_announcements -> take_announcements --- lib/pleroma/web/activity_pub/utils.ex | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) (limited to 'lib') diff --git a/lib/pleroma/web/activity_pub/utils.ex b/lib/pleroma/web/activity_pub/utils.ex index cf82d1a9b..0d87b9220 100644 --- a/lib/pleroma/web/activity_pub/utils.ex +++ b/lib/pleroma/web/activity_pub/utils.ex @@ -518,7 +518,7 @@ defmodule Pleroma.Web.ActivityPub.Utils do %Activity{data: %{"actor" => actor, "cc" => [Pleroma.Constants.as_public()]}}, object ) do - announcements = fetch_announcements(object) + announcements = take_announcements(object) with announcements <- Enum.uniq([actor | announcements]) do update_element_in_object("announcement", announcements, object) @@ -530,16 +530,16 @@ defmodule Pleroma.Web.ActivityPub.Utils do @spec remove_announce_from_object(Activity.t(), Object.t()) :: {:ok, Object.t()} | {:error, Ecto.Changeset.t()} def remove_announce_from_object(%Activity{data: %{"actor" => actor}}, object) do - with announcements <- List.delete(fetch_announcements(object), actor) do + with announcements <- List.delete(take_announcements(object), actor) do update_element_in_object("announcement", announcements, object) end end - defp fetch_announcements(%{data: %{"announcements" => announcements}} = _) + defp take_announcements(%{data: %{"announcements" => announcements}} = _) when is_list(announcements), do: announcements - defp fetch_announcements(_), do: [] + defp take_announcements(_), do: [] #### Unfollow-related helpers -- cgit v1.2.3 From 8306078de1abade082f932cda5b8d9297bdcdc80 Mon Sep 17 00:00:00 2001 From: Maksim Date: Wed, 4 Sep 2019 17:31:14 +0000 Subject: Apply suggestion to lib/pleroma/web/activity_pub/utils.ex --- lib/pleroma/web/activity_pub/utils.ex | 33 +++++++++++++-------------------- 1 file changed, 13 insertions(+), 20 deletions(-) (limited to 'lib') diff --git a/lib/pleroma/web/activity_pub/utils.ex b/lib/pleroma/web/activity_pub/utils.ex index 0d87b9220..2de02f607 100644 --- a/lib/pleroma/web/activity_pub/utils.ex +++ b/lib/pleroma/web/activity_pub/utils.ex @@ -356,26 +356,19 @@ defmodule Pleroma.Web.ActivityPub.Utils do %Activity{data: %{"actor" => actor, "object" => object}} = activity, state ) do - query = - from(activity in Activity, - where: fragment("data->>'type' = 'Follow'"), - where: fragment("data->>'state' = 'pending'"), - where: fragment("data->>'actor' = ?", ^actor), - where: fragment("data->>'object' = ?", ^object), - update: [ - set: [ - data: fragment("jsonb_set(data, '{state}', ?)", ^state) - ] - ] - ) - - with {_, _} <- Repo.update_all(query, []), - {_, _} <- User.set_follow_state_cache(actor, object, state), - %Activity{} = activity <- Activity.get_by_id(activity.id) do - {:ok, activity} - else - e -> {:error, e} - end + "Follow" + |> Activity.Queries.by_type() + |> Activity.Queries.by_actor(actor) + |> Activity.Queries.by_object_id(object["id"]) + |> where(fragment("data->>'state' = 'pending'")) + |> update(set: [data: fragment("jsonb_set(data, '{state}', ?)", ^state)]) + |> Repo.update_all([]) + + User.set_follow_state_cache(actor, object, state) + + activity = Activity.get_by_id(activity.id) + + {:ok, activity} end def update_follow_state( -- cgit v1.2.3 From e2011a667cdf5e67f257c9c30a02c206fb4df913 Mon Sep 17 00:00:00 2001 From: Maksim Date: Wed, 4 Sep 2019 18:35:01 +0000 Subject: Apply suggestion to lib/pleroma/web/activity_pub/utils.ex --- lib/pleroma/web/activity_pub/utils.ex | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'lib') diff --git a/lib/pleroma/web/activity_pub/utils.ex b/lib/pleroma/web/activity_pub/utils.ex index 2de02f607..011acd48e 100644 --- a/lib/pleroma/web/activity_pub/utils.ex +++ b/lib/pleroma/web/activity_pub/utils.ex @@ -359,7 +359,7 @@ defmodule Pleroma.Web.ActivityPub.Utils do "Follow" |> Activity.Queries.by_type() |> Activity.Queries.by_actor(actor) - |> Activity.Queries.by_object_id(object["id"]) + |> Activity.Queries.by_object_id(object) |> where(fragment("data->>'state' = 'pending'")) |> update(set: [data: fragment("jsonb_set(data, '{state}', ?)", ^state)]) |> Repo.update_all([]) -- cgit v1.2.3 From ae506ca997619f118d18703a9b0802246eb427d5 Mon Sep 17 00:00:00 2001 From: Maksim Pechnikov Date: Wed, 4 Sep 2019 21:40:53 +0300 Subject: fix formatting --- lib/pleroma/web/activity_pub/utils.ex | 26 +++++++++++++------------- 1 file changed, 13 insertions(+), 13 deletions(-) (limited to 'lib') diff --git a/lib/pleroma/web/activity_pub/utils.ex b/lib/pleroma/web/activity_pub/utils.ex index 011acd48e..72e07b59d 100644 --- a/lib/pleroma/web/activity_pub/utils.ex +++ b/lib/pleroma/web/activity_pub/utils.ex @@ -356,19 +356,19 @@ defmodule Pleroma.Web.ActivityPub.Utils do %Activity{data: %{"actor" => actor, "object" => object}} = activity, state ) do - "Follow" - |> Activity.Queries.by_type() - |> Activity.Queries.by_actor(actor) - |> Activity.Queries.by_object_id(object) - |> where(fragment("data->>'state' = 'pending'")) - |> update(set: [data: fragment("jsonb_set(data, '{state}', ?)", ^state)]) - |> Repo.update_all([]) - - User.set_follow_state_cache(actor, object, state) - - activity = Activity.get_by_id(activity.id) - - {:ok, activity} + "Follow" + |> Activity.Queries.by_type() + |> Activity.Queries.by_actor(actor) + |> Activity.Queries.by_object_id(object) + |> where(fragment("data->>'state' = 'pending'")) + |> update(set: [data: fragment("jsonb_set(data, '{state}', ?)", ^state)]) + |> Repo.update_all([]) + + User.set_follow_state_cache(actor, object, state) + + activity = Activity.get_by_id(activity.id) + + {:ok, activity} end def update_follow_state( -- cgit v1.2.3 From 769fb778d41df77c2514b5e3c663f3f624c0a266 Mon Sep 17 00:00:00 2001 From: rinpatch Date: Thu, 12 Sep 2019 21:37:36 +0300 Subject: Track object/create activity fetches --- lib/pleroma/delivery.ex | 58 ++++++++++++++++++++++ lib/pleroma/plugs/cache.ex | 16 +++++- lib/pleroma/user.ex | 10 ++++ .../web/activity_pub/activity_pub_controller.ex | 29 ++++++++++- 4 files changed, 111 insertions(+), 2 deletions(-) create mode 100644 lib/pleroma/delivery.ex (limited to 'lib') diff --git a/lib/pleroma/delivery.ex b/lib/pleroma/delivery.ex new file mode 100644 index 000000000..f9a9e35cd --- /dev/null +++ b/lib/pleroma/delivery.ex @@ -0,0 +1,58 @@ +# Pleroma: A lightweight social networking server +# Copyright © 2017-2019 Pleroma Authors +# SPDX-License-Identifier: AGPL-3.0-only + +defmodule Pleroma.Delivery do + use Ecto.Schema + + alias Pleroma.Delivery + alias Pleroma.FlakeId + alias Pleroma.User + alias Pleroma.Repo + alias Pleroma.Object + alias Pleroma.User + + import Ecto.Changeset + import Ecto.Query + + schema "deliveries" do + belongs_to(:user, User, type: FlakeId) + belongs_to(:object, Object) + end + + def changeset(delivery, params \\ %{}) do + delivery + |> cast(params, [:user_id, :object_id]) + |> foreign_key_constraint(:object_id) + |> foreign_key_constraint(:user_id) + |> unique_constraint(:user_id, name: :deliveries_user_id_object_id_index) + end + + def create(object_id, user_id) do + %Delivery{} + |> changeset(%{user_id: user_id, object_id: object_id}) + |> Repo.insert() + end + + def get(object_id, user_id) do + from(d in Delivery, where: d.user_id == ^user_id and d.object_id == ^object_id) + |> Repo.one() + end + + def get_or_create(object_id, user_id) do + case get(object_id, user_id) do + %Delivery{} = delivery -> {:ok, delivery} + nil -> create(object_id, user_id) + end + end + + def delete_all_by_object_id(object_id) do + from(d in Delivery, where: d.object_id == ^object_id) + |> Repo.delete_all() + end + + def get_all_by_object_id(object_id) do + from(d in Delivery, where: d.object_id == ^object_id) + |> Repo.all() + end +end diff --git a/lib/pleroma/plugs/cache.ex b/lib/pleroma/plugs/cache.ex index a81a861d0..42d77fc1f 100644 --- a/lib/pleroma/plugs/cache.ex +++ b/lib/pleroma/plugs/cache.ex @@ -20,6 +20,7 @@ defmodule Pleroma.Plugs.Cache do - `ttl`: An expiration time (time-to-live). This value should be in milliseconds or `nil` to disable expiration. Defaults to `nil`. - `query_params`: Take URL query string into account (`true`), ignore it (`false`) or limit to specific params only (list). Defaults to `true`. + - `tracking_fun`: A function that is called on successfull responses, no matter if the request is cached or not. It should accept a conn as the first argument and the value assigned to `tracking_fun_data` as the second. Additionally, you can overwrite the TTL inside a controller action by assigning `cache_ttl` to the connection struct: @@ -56,6 +57,10 @@ defmodule Pleroma.Plugs.Cache do {:ok, nil} -> cache_resp(conn, opts) + {:ok, {content_type, body, tracking_fun_data}} -> + conn = opts.tracking_fun(conn, tracking_fun_data) + send_cached(conn, {content_type, body}) + {:ok, record} -> send_cached(conn, record) @@ -90,7 +95,16 @@ defmodule Pleroma.Plugs.Cache do content_type = content_type(conn) record = {content_type, body} - Cachex.put(:web_resp_cache, key, record, ttl: ttl) + conn = + unless opts[:tracking_fun] do + Cachex.put(:web_resp_cache, key, {content_type, body}, ttl: ttl) + conn + else + tracking_fun_data = Map.get(conn.assigns, :tracking_fun_data, nil) + Cachex.put(:web_resp_cache, {content_type, body, tracking_fun_data}, record, ttl: ttl) + + opts.tracking_fun.(conn, tracking_fun_data) + end put_resp_header(conn, "x-cache", "MISS from Pleroma") diff --git a/lib/pleroma/user.ex b/lib/pleroma/user.ex index 3aa245f2a..9614acdab 100644 --- a/lib/pleroma/user.ex +++ b/lib/pleroma/user.ex @@ -11,6 +11,7 @@ defmodule Pleroma.User do alias Comeonin.Pbkdf2 alias Ecto.Multi alias Pleroma.Activity + alias Pleroma.Delivery alias Pleroma.Keys alias Pleroma.Notification alias Pleroma.Object @@ -61,6 +62,7 @@ defmodule Pleroma.User do field(:last_digest_emailed_at, :naive_datetime) has_many(:notifications, Notification) has_many(:registrations, Registration) + has_many(:deliveries, Delivery) embeds_one(:info, User.Info) timestamps() @@ -1624,4 +1626,12 @@ defmodule Pleroma.User do def is_internal_user?(%User{nickname: nil}), do: true def is_internal_user?(%User{local: true, nickname: "internal." <> _}), do: true def is_internal_user?(_), do: false + + def get_delivered_users_by_object_id(object_id) do + from(u in User, + inner_join: delivery in assoc(u, :deliveries), + where: delivery.object_id == ^object_id + ) + |> Repo.all() + end end diff --git a/lib/pleroma/web/activity_pub/activity_pub_controller.ex b/lib/pleroma/web/activity_pub/activity_pub_controller.ex index 705dbc1c2..009260d3f 100644 --- a/lib/pleroma/web/activity_pub/activity_pub_controller.ex +++ b/lib/pleroma/web/activity_pub/activity_pub_controller.ex @@ -6,6 +6,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPubController do use Pleroma.Web, :controller alias Pleroma.Activity + alias Pleroma.Delivery alias Pleroma.Object alias Pleroma.Object.Fetcher alias Pleroma.User @@ -23,7 +24,15 @@ defmodule Pleroma.Web.ActivityPub.ActivityPubController do action_fallback(:errors) - plug(Pleroma.Plugs.Cache, [query_params: false] when action in [:activity, :object]) + plug( + Pleroma.Plugs.Cache, + [ + query_params: false, + tracking_fun: &Pleroma.Web.ActivityPub.ActivityPubController.track_object_fetch/2 + ] + when action in [:activity, :object] + ) + plug(Pleroma.Web.FederatingPlug when action in [:inbox, :relay]) plug(:set_requester_reachable when action in [:inbox]) plug(:relay_active? when action in [:relay]) @@ -54,6 +63,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPubController do %Object{} = object <- Object.get_cached_by_ap_id(ap_id), {_, true} <- {:public?, Visibility.is_public?(object)} do conn + |> assign(:tracking_fun_data, object.id) |> set_cache_ttl_for(object) |> put_resp_content_type("application/activity+json") |> put_view(ObjectView) @@ -64,6 +74,15 @@ defmodule Pleroma.Web.ActivityPub.ActivityPubController do end end + def track_object_fetch(conn, object_id) do + case conn.assigns[:user] do + %User{id: user_id} -> Delivery.create(object_id, user_id) + _ -> nil + end + + conn + end + def object_likes(conn, %{"uuid" => uuid, "page" => page}) do with ap_id <- o_status_url(conn, :object, uuid), %Object{} = object <- Object.get_cached_by_ap_id(ap_id), @@ -99,6 +118,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPubController do %Activity{} = activity <- Activity.normalize(ap_id), {_, true} <- {:public?, Visibility.is_public?(activity)} do conn + |> maybe_set_tracking_data(activity) |> set_cache_ttl_for(activity) |> put_resp_content_type("application/activity+json") |> put_view(ObjectView) @@ -109,6 +129,13 @@ defmodule Pleroma.Web.ActivityPub.ActivityPubController do end end + defp maybe_set_tracking_data(conn, %Activity{data: %{"type" => "Create"}} = activity) do + object_id = Object.normalize(activity).id + assign(conn, :tracking_fun_data, object_id) + end + + defp maybe_set_tracking_data(conn, _activity), do: assign(conn, :tracking_fun_data, nil) + defp set_cache_ttl_for(conn, %Activity{object: object}) do set_cache_ttl_for(conn, object) end -- cgit v1.2.3 From dabc4a00f5cf08dac75f701457a24fce8735175f Mon Sep 17 00:00:00 2001 From: rinpatch Date: Thu, 12 Sep 2019 22:10:15 +0300 Subject: Put the cache with the right key when using a tracking function --- lib/pleroma/plugs/cache.ex | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) (limited to 'lib') diff --git a/lib/pleroma/plugs/cache.ex b/lib/pleroma/plugs/cache.ex index 42d77fc1f..50b534e7b 100644 --- a/lib/pleroma/plugs/cache.ex +++ b/lib/pleroma/plugs/cache.ex @@ -58,7 +58,8 @@ defmodule Pleroma.Plugs.Cache do cache_resp(conn, opts) {:ok, {content_type, body, tracking_fun_data}} -> - conn = opts.tracking_fun(conn, tracking_fun_data) + conn = opts.tracking_fun.(conn, tracking_fun_data) + send_cached(conn, {content_type, body}) {:ok, record} -> @@ -93,7 +94,6 @@ defmodule Pleroma.Plugs.Cache do ttl = Map.get(conn.assigns, :cache_ttl, opts.ttl) key = cache_key(conn, opts) content_type = content_type(conn) - record = {content_type, body} conn = unless opts[:tracking_fun] do @@ -101,7 +101,7 @@ defmodule Pleroma.Plugs.Cache do conn else tracking_fun_data = Map.get(conn.assigns, :tracking_fun_data, nil) - Cachex.put(:web_resp_cache, {content_type, body, tracking_fun_data}, record, ttl: ttl) + Cachex.put(:web_resp_cache, key, {content_type, body, tracking_fun_data}, ttl: ttl) opts.tracking_fun.(conn, tracking_fun_data) end -- cgit v1.2.3 From b0e60580215e26caae6452099fa1fbace525937c Mon Sep 17 00:00:00 2001 From: rinpatch Date: Thu, 12 Sep 2019 22:40:53 +0300 Subject: Parse http signature for request to objects/activities --- lib/pleroma/plugs/http_signature.ex | 3 ++- lib/pleroma/web/router.ex | 2 ++ 2 files changed, 4 insertions(+), 1 deletion(-) (limited to 'lib') diff --git a/lib/pleroma/plugs/http_signature.ex b/lib/pleroma/plugs/http_signature.ex index d87fa52fa..23d22a712 100644 --- a/lib/pleroma/plugs/http_signature.ex +++ b/lib/pleroma/plugs/http_signature.ex @@ -15,7 +15,8 @@ defmodule Pleroma.Web.Plugs.HTTPSignaturePlug do end def call(conn, _opts) do - [signature | _] = get_req_header(conn, "signature") + headers = get_req_header(conn, "signature") + signature = Enum.at(headers, 0) if signature do # set (request-target) header to the appropriate value diff --git a/lib/pleroma/web/router.ex b/lib/pleroma/web/router.ex index 7cd59acb2..badc7e048 100644 --- a/lib/pleroma/web/router.ex +++ b/lib/pleroma/web/router.ex @@ -135,6 +135,7 @@ defmodule Pleroma.Web.Router do pipeline :http_signature do plug(Pleroma.Web.Plugs.HTTPSignaturePlug) + plug(Pleroma.Web.Plugs.MappedSignatureToIdentityPlug) end scope "/api/pleroma", Pleroma.Web.TwitterAPI do @@ -513,6 +514,7 @@ defmodule Pleroma.Web.Router do scope "/", Pleroma.Web do pipe_through(:ostatus) + pipe_through(:http_signature) get("/objects/:uuid", OStatus.OStatusController, :object) get("/activities/:uuid", OStatus.OStatusController, :activity) -- cgit v1.2.3 From ce23529d917c1830b270a29e774e4ed7768bfeff Mon Sep 17 00:00:00 2001 From: rinpatch Date: Fri, 13 Sep 2019 11:36:49 +0300 Subject: Use delivery info when federating deletes --- lib/pleroma/delivery.ex | 4 ++++ lib/pleroma/user.ex | 4 ++++ lib/pleroma/web/activity_pub/publisher.ex | 15 ++++++++++++++- 3 files changed, 22 insertions(+), 1 deletion(-) (limited to 'lib') diff --git a/lib/pleroma/delivery.ex b/lib/pleroma/delivery.ex index f9a9e35cd..2e7c019fa 100644 --- a/lib/pleroma/delivery.ex +++ b/lib/pleroma/delivery.ex @@ -46,6 +46,10 @@ defmodule Pleroma.Delivery do end end + # A hack because user delete activities have a fake id for whatever reason + # TODO: Get rid of this + def delete_all_by_object_id("pleroma:fake_object_id"), do: {0, []} + def delete_all_by_object_id(object_id) do from(d in Delivery, where: d.object_id == ^object_id) |> Repo.delete_all() diff --git a/lib/pleroma/user.ex b/lib/pleroma/user.ex index 9614acdab..785b22643 100644 --- a/lib/pleroma/user.ex +++ b/lib/pleroma/user.ex @@ -1627,6 +1627,10 @@ defmodule Pleroma.User do def is_internal_user?(%User{local: true, nickname: "internal." <> _}), do: true def is_internal_user?(_), do: false + # A hack because user delete activities have a fake id for whatever reason + # TODO: Get rid of this + def get_delivered_users_by_object_id("pleroma:fake_object_id"), do: [] + def get_delivered_users_by_object_id(object_id) do from(u in User, inner_join: delivery in assoc(u, :deliveries), diff --git a/lib/pleroma/web/activity_pub/publisher.ex b/lib/pleroma/web/activity_pub/publisher.ex index c97405690..db64fd2f6 100644 --- a/lib/pleroma/web/activity_pub/publisher.ex +++ b/lib/pleroma/web/activity_pub/publisher.ex @@ -5,9 +5,11 @@ defmodule Pleroma.Web.ActivityPub.Publisher do alias Pleroma.Activity alias Pleroma.Config + alias Pleroma.Delivery alias Pleroma.HTTP alias Pleroma.Instances alias Pleroma.User + alias Pleroma.Object alias Pleroma.Web.ActivityPub.Relay alias Pleroma.Web.ActivityPub.Transmogrifier @@ -107,7 +109,18 @@ defmodule Pleroma.Web.ActivityPub.Publisher do {:ok, []} end - Pleroma.Web.Salmon.remote_users(actor, activity) ++ followers + fetchers = + with %Activity{data: %{"type" => "Delete"}} <- activity, + %Object{id: object_id} <- Object.normalize(activity), + fetchers <- User.get_delivered_users_by_object_id(object_id), + _ <- Delivery.delete_all_by_object_id(object_id) do + fetchers + else + _ -> + [] + end + + Pleroma.Web.Salmon.remote_users(actor, activity) ++ followers ++ fetchers end defp get_cc_ap_ids(ap_id, recipients) do -- cgit v1.2.3 From fb96facc32fb275efffeefa2892a1098ecd68b77 Mon Sep 17 00:00:00 2001 From: rinpatch Date: Fri, 13 Sep 2019 12:06:31 +0300 Subject: Remove unused functions and fix credo issues --- lib/pleroma/delivery.ex | 16 ++-------------- lib/pleroma/web/activity_pub/publisher.ex | 2 +- 2 files changed, 3 insertions(+), 15 deletions(-) (limited to 'lib') diff --git a/lib/pleroma/delivery.ex b/lib/pleroma/delivery.ex index 2e7c019fa..ce8fb96f4 100644 --- a/lib/pleroma/delivery.ex +++ b/lib/pleroma/delivery.ex @@ -7,9 +7,9 @@ defmodule Pleroma.Delivery do alias Pleroma.Delivery alias Pleroma.FlakeId - alias Pleroma.User - alias Pleroma.Repo alias Pleroma.Object + alias Pleroma.Repo + alias Pleroma.User alias Pleroma.User import Ecto.Changeset @@ -39,13 +39,6 @@ defmodule Pleroma.Delivery do |> Repo.one() end - def get_or_create(object_id, user_id) do - case get(object_id, user_id) do - %Delivery{} = delivery -> {:ok, delivery} - nil -> create(object_id, user_id) - end - end - # A hack because user delete activities have a fake id for whatever reason # TODO: Get rid of this def delete_all_by_object_id("pleroma:fake_object_id"), do: {0, []} @@ -54,9 +47,4 @@ defmodule Pleroma.Delivery do from(d in Delivery, where: d.object_id == ^object_id) |> Repo.delete_all() end - - def get_all_by_object_id(object_id) do - from(d in Delivery, where: d.object_id == ^object_id) - |> Repo.all() - end end diff --git a/lib/pleroma/web/activity_pub/publisher.ex b/lib/pleroma/web/activity_pub/publisher.ex index db64fd2f6..c39e89a6a 100644 --- a/lib/pleroma/web/activity_pub/publisher.ex +++ b/lib/pleroma/web/activity_pub/publisher.ex @@ -8,8 +8,8 @@ defmodule Pleroma.Web.ActivityPub.Publisher do alias Pleroma.Delivery alias Pleroma.HTTP alias Pleroma.Instances - alias Pleroma.User alias Pleroma.Object + alias Pleroma.User alias Pleroma.Web.ActivityPub.Relay alias Pleroma.Web.ActivityPub.Transmogrifier -- cgit v1.2.3 From 517017048316a52172d60d26b03beddb85af7b39 Mon Sep 17 00:00:00 2001 From: rinpatch Date: Fri, 13 Sep 2019 10:09:46 +0000 Subject: Apply suggestion to lib/pleroma/web/activity_pub/activity_pub_controller.ex --- lib/pleroma/web/activity_pub/activity_pub_controller.ex | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) (limited to 'lib') diff --git a/lib/pleroma/web/activity_pub/activity_pub_controller.ex b/lib/pleroma/web/activity_pub/activity_pub_controller.ex index 009260d3f..025641722 100644 --- a/lib/pleroma/web/activity_pub/activity_pub_controller.ex +++ b/lib/pleroma/web/activity_pub/activity_pub_controller.ex @@ -26,10 +26,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPubController do plug( Pleroma.Plugs.Cache, - [ - query_params: false, - tracking_fun: &Pleroma.Web.ActivityPub.ActivityPubController.track_object_fetch/2 - ] + [query_params: false, tracking_fun: &__MODULE__.track_object_fetch/2] when action in [:activity, :object] ) -- cgit v1.2.3 From 3896a51b8aefe6fe54251ffd559c636980faa87e Mon Sep 17 00:00:00 2001 From: rinpatch Date: Fri, 13 Sep 2019 10:09:56 +0000 Subject: Apply suggestion to lib/pleroma/web/activity_pub/activity_pub_controller.ex --- lib/pleroma/web/activity_pub/activity_pub_controller.ex | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) (limited to 'lib') diff --git a/lib/pleroma/web/activity_pub/activity_pub_controller.ex b/lib/pleroma/web/activity_pub/activity_pub_controller.ex index 025641722..4bd13defb 100644 --- a/lib/pleroma/web/activity_pub/activity_pub_controller.ex +++ b/lib/pleroma/web/activity_pub/activity_pub_controller.ex @@ -72,9 +72,8 @@ defmodule Pleroma.Web.ActivityPub.ActivityPubController do end def track_object_fetch(conn, object_id) do - case conn.assigns[:user] do - %User{id: user_id} -> Delivery.create(object_id, user_id) - _ -> nil + with %{assigns: %{user: %User{id: user_id}}} <- conn do + Delivery.create(object_id, user_id) end conn -- cgit v1.2.3 From 2784962dba295ee35677e93996df53d1711e5768 Mon Sep 17 00:00:00 2001 From: rinpatch Date: Fri, 13 Sep 2019 15:23:03 +0000 Subject: Apply suggestion to lib/pleroma/web/activity_pub/activity_pub_controller.ex --- lib/pleroma/web/activity_pub/activity_pub_controller.ex | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'lib') diff --git a/lib/pleroma/web/activity_pub/activity_pub_controller.ex b/lib/pleroma/web/activity_pub/activity_pub_controller.ex index 4bd13defb..70d4a5baf 100644 --- a/lib/pleroma/web/activity_pub/activity_pub_controller.ex +++ b/lib/pleroma/web/activity_pub/activity_pub_controller.ex @@ -130,7 +130,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPubController do assign(conn, :tracking_fun_data, object_id) end - defp maybe_set_tracking_data(conn, _activity), do: assign(conn, :tracking_fun_data, nil) + defp maybe_set_tracking_data(conn, _activity), do: conn defp set_cache_ttl_for(conn, %Activity{object: object}) do set_cache_ttl_for(conn, object) -- cgit v1.2.3 From 05f8a066a107af2f7151aee8d85af97cf6a4835c Mon Sep 17 00:00:00 2001 From: rinpatch Date: Fri, 13 Sep 2019 15:23:26 +0000 Subject: Apply suggestion to lib/pleroma/delivery.ex --- lib/pleroma/delivery.ex | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'lib') diff --git a/lib/pleroma/delivery.ex b/lib/pleroma/delivery.ex index ce8fb96f4..38c148c34 100644 --- a/lib/pleroma/delivery.ex +++ b/lib/pleroma/delivery.ex @@ -31,7 +31,7 @@ defmodule Pleroma.Delivery do def create(object_id, user_id) do %Delivery{} |> changeset(%{user_id: user_id, object_id: object_id}) - |> Repo.insert() + |> Repo.insert(on_conflict: :nothing) end def get(object_id, user_id) do -- cgit v1.2.3 From 8900cb68aef535dbf60de87fce47d85b91909077 Mon Sep 17 00:00:00 2001 From: rinpatch Date: Fri, 13 Sep 2019 15:25:15 +0000 Subject: Apply suggestion to lib/pleroma/web/activity_pub/activity_pub_controller.ex --- lib/pleroma/web/activity_pub/activity_pub_controller.ex | 2 ++ 1 file changed, 2 insertions(+) (limited to 'lib') diff --git a/lib/pleroma/web/activity_pub/activity_pub_controller.ex b/lib/pleroma/web/activity_pub/activity_pub_controller.ex index 70d4a5baf..01b34fb1d 100644 --- a/lib/pleroma/web/activity_pub/activity_pub_controller.ex +++ b/lib/pleroma/web/activity_pub/activity_pub_controller.ex @@ -71,6 +71,8 @@ defmodule Pleroma.Web.ActivityPub.ActivityPubController do end end + def track_object_fetch(conn, nil), do: conn + def track_object_fetch(conn, object_id) do with %{assigns: %{user: %User{id: user_id}}} <- conn do Delivery.create(object_id, user_id) -- cgit v1.2.3 From ac4a748fad34c02647bf72e802cd9d74205681fe Mon Sep 17 00:00:00 2001 From: rinpatch Date: Fri, 13 Sep 2019 19:28:35 +0300 Subject: Disallow NULLs in deliveries --- lib/pleroma/delivery.ex | 1 + 1 file changed, 1 insertion(+) (limited to 'lib') diff --git a/lib/pleroma/delivery.ex b/lib/pleroma/delivery.ex index 38c148c34..29a1e5a77 100644 --- a/lib/pleroma/delivery.ex +++ b/lib/pleroma/delivery.ex @@ -23,6 +23,7 @@ defmodule Pleroma.Delivery do def changeset(delivery, params \\ %{}) do delivery |> cast(params, [:user_id, :object_id]) + |> validate_required([:user_id, :object_id]) |> foreign_key_constraint(:object_id) |> foreign_key_constraint(:user_id) |> unique_constraint(:user_id, name: :deliveries_user_id_object_id_index) -- cgit v1.2.3 From b4cf74c1067b866574a63fbd25ccb12cc1fed619 Mon Sep 17 00:00:00 2001 From: Maksim Pechnikov Date: Sun, 15 Sep 2019 14:53:58 +0300 Subject: added prepare html for RichMedia.Parser --- lib/pleroma/web/rich_media/parser.ex | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) (limited to 'lib') diff --git a/lib/pleroma/web/rich_media/parser.ex b/lib/pleroma/web/rich_media/parser.ex index f5f9e358c..c06b0a0f2 100644 --- a/lib/pleroma/web/rich_media/parser.ex +++ b/lib/pleroma/web/rich_media/parser.ex @@ -81,6 +81,7 @@ defmodule Pleroma.Web.RichMedia.Parser do {:ok, %Tesla.Env{body: html}} = Pleroma.HTTP.get(url, [], adapter: @hackney_options) html + |> parse_html |> maybe_parse() |> Map.put(:url, url) |> clean_parsed_data() @@ -91,6 +92,8 @@ defmodule Pleroma.Web.RichMedia.Parser do end end + defp parse_html(html), do: Floki.parse(html) + defp maybe_parse(html) do Enum.reduce_while(parsers(), %{}, fn parser, acc -> case parser.parse(html, acc) do @@ -100,7 +103,8 @@ defmodule Pleroma.Web.RichMedia.Parser do end) end - defp check_parsed_data(%{title: title} = data) when is_binary(title) and byte_size(title) > 0 do + defp check_parsed_data(%{title: title} = data) + when is_binary(title) and byte_size(title) > 0 do {:ok, data} end -- cgit v1.2.3 From aab264db82054df470075c65ca25c42bbcc5d7a8 Mon Sep 17 00:00:00 2001 From: Steven Fuchs Date: Mon, 16 Sep 2019 07:44:03 +0000 Subject: Streamer refactoring --- lib/pleroma/activity/ir/topics.ex | 63 +++++ lib/pleroma/application.ex | 2 +- lib/pleroma/notification.ex | 6 +- lib/pleroma/web/activity_pub/activity_pub.ex | 50 +--- lib/pleroma/web/mastodon_api/websocket_handler.ex | 7 +- lib/pleroma/web/streamer.ex | 318 ---------------------- lib/pleroma/web/streamer/ping.ex | 33 +++ lib/pleroma/web/streamer/state.ex | 68 +++++ lib/pleroma/web/streamer/streamer.ex | 55 ++++ lib/pleroma/web/streamer/streamer_socket.ex | 31 +++ lib/pleroma/web/streamer/supervisor.ex | 33 +++ lib/pleroma/web/streamer/worker.ex | 220 +++++++++++++++ lib/pleroma/web/views/streamer_view.ex | 66 +++++ 13 files changed, 590 insertions(+), 362 deletions(-) create mode 100644 lib/pleroma/activity/ir/topics.ex delete mode 100644 lib/pleroma/web/streamer.ex create mode 100644 lib/pleroma/web/streamer/ping.ex create mode 100644 lib/pleroma/web/streamer/state.ex create mode 100644 lib/pleroma/web/streamer/streamer.ex create mode 100644 lib/pleroma/web/streamer/streamer_socket.ex create mode 100644 lib/pleroma/web/streamer/supervisor.ex create mode 100644 lib/pleroma/web/streamer/worker.ex create mode 100644 lib/pleroma/web/views/streamer_view.ex (limited to 'lib') diff --git a/lib/pleroma/activity/ir/topics.ex b/lib/pleroma/activity/ir/topics.ex new file mode 100644 index 000000000..010897abc --- /dev/null +++ b/lib/pleroma/activity/ir/topics.ex @@ -0,0 +1,63 @@ +# Pleroma: A lightweight social networking server +# Copyright © 2017-2019 Pleroma Authors +# SPDX-License-Identifier: AGPL-3.0-only + +defmodule Pleroma.Activity.Ir.Topics do + alias Pleroma.Object + alias Pleroma.Web.ActivityPub.Visibility + + def get_activity_topics(activity) do + activity + |> Object.normalize() + |> generate_topics(activity) + |> List.flatten() + end + + defp generate_topics(%{data: %{"type" => "Answer"}}, _) do + [] + end + + defp generate_topics(object, activity) do + ["user", "list"] ++ visibility_tags(object, activity) + end + + defp visibility_tags(object, activity) do + case Visibility.get_visibility(activity) do + "public" -> + if activity.local do + ["public", "public:local"] + else + ["public"] + end + |> item_creation_tags(object, activity) + + "direct" -> + ["direct"] + + _ -> + [] + end + end + + defp item_creation_tags(tags, %{data: %{"type" => "Create"}} = object, activity) do + tags ++ hashtags_to_topics(object) ++ attachment_topics(object, activity) + end + + defp item_creation_tags(tags, _, _) do + tags + end + + defp hashtags_to_topics(%{data: %{"tag" => tags}}) do + tags + |> Enum.filter(&is_bitstring(&1)) + |> Enum.map(fn tag -> "hashtag:" <> tag end) + end + + defp hashtags_to_topics(_), do: [] + + defp attachment_topics(%{data: %{"attachment" => []}}, _act), do: [] + + defp attachment_topics(_object, %{local: true}), do: ["public:media", "public:local:media"] + + defp attachment_topics(_object, _act), do: ["public:media"] +end diff --git a/lib/pleroma/application.ex b/lib/pleroma/application.ex index 49094704b..3b37ce630 100644 --- a/lib/pleroma/application.ex +++ b/lib/pleroma/application.ex @@ -141,7 +141,7 @@ defmodule Pleroma.Application do defp streamer_child(:test), do: [] defp streamer_child(_) do - [Pleroma.Web.Streamer] + [Pleroma.Web.Streamer.supervisor()] end defp oauth_cleanup_child(true), diff --git a/lib/pleroma/notification.ex b/lib/pleroma/notification.ex index b7c880c51..8012389ac 100644 --- a/lib/pleroma/notification.ex +++ b/lib/pleroma/notification.ex @@ -210,8 +210,10 @@ defmodule Pleroma.Notification do unless skip?(activity, user) do notification = %Notification{user_id: user.id, activity: activity} {:ok, notification} = Repo.insert(notification) - Streamer.stream("user", notification) - Streamer.stream("user:notification", notification) + + ["user", "user:notification"] + |> Streamer.stream(notification) + Push.send(notification) notification end diff --git a/lib/pleroma/web/activity_pub/activity_pub.ex b/lib/pleroma/web/activity_pub/activity_pub.ex index 41f6a0f1f..bc5ae7fbf 100644 --- a/lib/pleroma/web/activity_pub/activity_pub.ex +++ b/lib/pleroma/web/activity_pub/activity_pub.ex @@ -4,6 +4,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do alias Pleroma.Activity + alias Pleroma.Activity.Ir.Topics alias Pleroma.Config alias Pleroma.Conversation alias Pleroma.Notification @@ -16,6 +17,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do alias Pleroma.User alias Pleroma.Web.ActivityPub.MRF alias Pleroma.Web.ActivityPub.Transmogrifier + alias Pleroma.Web.Streamer alias Pleroma.Web.WebFinger alias Pleroma.Workers.BackgroundWorker @@ -187,9 +189,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do participations |> Repo.preload(:user) - Enum.each(participations, fn participation -> - Pleroma.Web.Streamer.stream("participation", participation) - end) + Streamer.stream("participation", participations) end def stream_out_participations(%Object{data: %{"context" => context}}, user) do @@ -208,41 +208,15 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do def stream_out_participations(_, _), do: :noop - def stream_out(activity) do - if activity.data["type"] in ["Create", "Announce", "Delete"] do - object = Object.normalize(activity) - # Do not stream out poll replies - unless object.data["type"] == "Answer" do - Pleroma.Web.Streamer.stream("user", activity) - Pleroma.Web.Streamer.stream("list", activity) - - if get_visibility(activity) == "public" do - Pleroma.Web.Streamer.stream("public", activity) - - if activity.local do - Pleroma.Web.Streamer.stream("public:local", activity) - end - - if activity.data["type"] in ["Create"] do - object.data - |> Map.get("tag", []) - |> Enum.filter(fn tag -> is_bitstring(tag) end) - |> Enum.each(fn tag -> Pleroma.Web.Streamer.stream("hashtag:" <> tag, activity) end) - - if object.data["attachment"] != [] do - Pleroma.Web.Streamer.stream("public:media", activity) - - if activity.local do - Pleroma.Web.Streamer.stream("public:local:media", activity) - end - end - end - else - if get_visibility(activity) == "direct", - do: Pleroma.Web.Streamer.stream("direct", activity) - end - end - end + def stream_out(%Activity{data: %{"type" => data_type}} = activity) + when data_type in ["Create", "Announce", "Delete"] do + activity + |> Topics.get_activity_topics() + |> Streamer.stream(activity) + end + + def stream_out(_activity) do + :noop end def create(%{to: to, actor: actor, context: context, object: object} = params, fake \\ false) do diff --git a/lib/pleroma/web/mastodon_api/websocket_handler.ex b/lib/pleroma/web/mastodon_api/websocket_handler.ex index dbd3542ea..3c26eb406 100644 --- a/lib/pleroma/web/mastodon_api/websocket_handler.ex +++ b/lib/pleroma/web/mastodon_api/websocket_handler.ex @@ -8,6 +8,7 @@ defmodule Pleroma.Web.MastodonAPI.WebsocketHandler do alias Pleroma.Repo alias Pleroma.User alias Pleroma.Web.OAuth.Token + alias Pleroma.Web.Streamer @behaviour :cowboy_websocket @@ -24,7 +25,7 @@ defmodule Pleroma.Web.MastodonAPI.WebsocketHandler do ] @anonymous_streams ["public", "public:local", "hashtag"] - # Handled by periodic keepalive in Pleroma.Web.Streamer. + # Handled by periodic keepalive in Pleroma.Web.Streamer.Ping. @timeout :infinity def init(%{qs: qs} = req, state) do @@ -65,7 +66,7 @@ defmodule Pleroma.Web.MastodonAPI.WebsocketHandler do }, topic #{state.topic}" ) - Pleroma.Web.Streamer.add_socket(state.topic, streamer_socket(state)) + Streamer.add_socket(state.topic, streamer_socket(state)) {:ok, state} end @@ -80,7 +81,7 @@ defmodule Pleroma.Web.MastodonAPI.WebsocketHandler do }, topic #{state.topic || "?"}: #{inspect(reason)}" ) - Pleroma.Web.Streamer.remove_socket(state.topic, streamer_socket(state)) + Streamer.remove_socket(state.topic, streamer_socket(state)) :ok end diff --git a/lib/pleroma/web/streamer.ex b/lib/pleroma/web/streamer.ex deleted file mode 100644 index 587c43f40..000000000 --- a/lib/pleroma/web/streamer.ex +++ /dev/null @@ -1,318 +0,0 @@ -# Pleroma: A lightweight social networking server -# Copyright © 2017-2019 Pleroma Authors -# 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.CommonAPI - 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 - Process.send_after(self(), %{action: :ping}, @keepalive_interval) - - {:ok, args} - end - - def handle_info(%{action: :ping}, topics) do - topics - |> Map.values() - |> List.flatten() - |> Enum.each(fn socket -> - Logger.debug("Sending keepalive ping") - send(socket.transport_pid, {:text, ""}) - end) - - Process.send_after(self(), %{action: :ping}, @keepalive_interval) - - {: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: topic, item: %Notification{} = item}, - topics - ) - when topic in ["user", "user:notification"] do - topics - |> Map.get("#{topic}:#{item.user_id}", []) - |> Enum.each(fn socket -> - with %User{} = user <- User.get_cached_by_ap_id(socket.assigns[:user].ap_id), - true <- should_send?(user, item) do - send( - socket.transport_pid, - {:text, represent_notification(socket.assigns[:user], item)} - ) - end - 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, - for: participation.user - }) - |> Jason.encode!() - } - |> 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 - - defp should_send?(%User{} = user, %Activity{} = item) do - blocks = user.info.blocks || [] - mutes = user.info.mutes || [] - reblog_mutes = user.info.muted_reblogs || [] - domain_blocks = Pleroma.Web.ActivityPub.MRF.subdomains_regex(user.info.domain_blocks) - - 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)), - %{host: item_host} <- URI.parse(item.actor), - %{host: parent_host} <- URI.parse(parent.data["actor"]), - false <- Pleroma.Web.ActivityPub.MRF.subdomain_match?(domain_blocks, item_host), - false <- Pleroma.Web.ActivityPub.MRF.subdomain_match?(domain_blocks, parent_host), - true <- thread_containment(item, user), - false <- CommonAPI.thread_muted?(user, item) do - true - else - _ -> false - end - end - - defp should_send?(%User{} = user, %Notification{activity: activity}) do - should_send?(user, activity) - 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) - - if should_send?(user, item) 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 user:notification 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 diff --git a/lib/pleroma/web/streamer/ping.ex b/lib/pleroma/web/streamer/ping.ex new file mode 100644 index 000000000..f77cbb95c --- /dev/null +++ b/lib/pleroma/web/streamer/ping.ex @@ -0,0 +1,33 @@ +defmodule Pleroma.Web.Streamer.Ping do + use GenServer + require Logger + + alias Pleroma.Web.Streamer.State + alias Pleroma.Web.Streamer.StreamerSocket + + @keepalive_interval :timer.seconds(30) + + def start_link(opts) do + ping_interval = Keyword.get(opts, :ping_interval, @keepalive_interval) + GenServer.start_link(__MODULE__, %{ping_interval: ping_interval}, name: __MODULE__) + end + + def init(%{ping_interval: ping_interval} = args) do + Process.send_after(self(), :ping, ping_interval) + {:ok, args} + end + + def handle_info(:ping, %{ping_interval: ping_interval} = state) do + State.get_sockets() + |> Map.values() + |> List.flatten() + |> Enum.each(fn %StreamerSocket{transport_pid: transport_pid} -> + Logger.debug("Sending keepalive ping") + send(transport_pid, {:text, ""}) + end) + + Process.send_after(self(), :ping, ping_interval) + + {:noreply, state} + end +end diff --git a/lib/pleroma/web/streamer/state.ex b/lib/pleroma/web/streamer/state.ex new file mode 100644 index 000000000..7b5199068 --- /dev/null +++ b/lib/pleroma/web/streamer/state.ex @@ -0,0 +1,68 @@ +defmodule Pleroma.Web.Streamer.State do + use GenServer + require Logger + + alias Pleroma.Web.Streamer.StreamerSocket + + def start_link(_) do + GenServer.start_link(__MODULE__, %{sockets: %{}}, name: __MODULE__) + end + + def add_socket(topic, socket) do + GenServer.call(__MODULE__, {:add, socket, topic}) + end + + def remove_socket(topic, socket) do + GenServer.call(__MODULE__, {:remove, socket, topic}) + end + + def get_sockets do + %{sockets: stream_sockets} = GenServer.call(__MODULE__, :get_state) + stream_sockets + end + + def init(init_arg) do + {:ok, init_arg} + end + + def handle_call(:get_state, _from, state) do + {:reply, state, state} + end + + def handle_call({:add, socket, topic}, _from, %{sockets: sockets} = state) do + internal_topic = internal_topic(topic, socket) + stream_socket = StreamerSocket.from_socket(socket) + + sockets_for_topic = + sockets + |> Map.get(internal_topic, []) + |> List.insert_at(0, stream_socket) + |> Enum.uniq() + + state = put_in(state, [:sockets, internal_topic], sockets_for_topic) + Logger.debug("Got new conn for #{topic}") + {:reply, state, state} + end + + def handle_call({:remove, socket, topic}, _from, %{sockets: sockets} = state) do + internal_topic = internal_topic(topic, socket) + stream_socket = StreamerSocket.from_socket(socket) + + sockets_for_topic = + sockets + |> Map.get(internal_topic, []) + |> List.delete(stream_socket) + + state = Kernel.put_in(state, [:sockets, internal_topic], sockets_for_topic) + {:reply, state, state} + end + + defp internal_topic(topic, socket) + when topic in ~w[user user:notification direct] do + "#{topic}:#{socket.assigns[:user].id}" + end + + defp internal_topic(topic, _) do + topic + end +end diff --git a/lib/pleroma/web/streamer/streamer.ex b/lib/pleroma/web/streamer/streamer.ex new file mode 100644 index 000000000..8cf719277 --- /dev/null +++ b/lib/pleroma/web/streamer/streamer.ex @@ -0,0 +1,55 @@ +# Pleroma: A lightweight social networking server +# Copyright © 2017-2019 Pleroma Authors +# SPDX-License-Identifier: AGPL-3.0-only + +defmodule Pleroma.Web.Streamer do + alias Pleroma.Web.Streamer.State + alias Pleroma.Web.Streamer.Worker + + @timeout 60_000 + @mix_env Mix.env() + + def add_socket(topic, socket) do + State.add_socket(topic, socket) + end + + def remove_socket(topic, socket) do + State.remove_socket(topic, socket) + end + + def get_sockets do + State.get_sockets() + end + + def stream(topics, items) do + if should_send?() do + Task.async(fn -> + :poolboy.transaction( + :streamer_worker, + &Worker.stream(&1, topics, items), + @timeout + ) + end) + end + end + + def supervisor, do: Pleroma.Web.Streamer.Supervisor + + defp should_send? do + handle_should_send(@mix_env) + end + + defp handle_should_send(:test) do + case Process.whereis(:streamer_worker) do + nil -> + false + + pid -> + Process.alive?(pid) + end + end + + defp handle_should_send(_) do + true + end +end diff --git a/lib/pleroma/web/streamer/streamer_socket.ex b/lib/pleroma/web/streamer/streamer_socket.ex new file mode 100644 index 000000000..f006c0306 --- /dev/null +++ b/lib/pleroma/web/streamer/streamer_socket.ex @@ -0,0 +1,31 @@ +defmodule Pleroma.Web.Streamer.StreamerSocket do + defstruct transport_pid: nil, user: nil + + alias Pleroma.User + alias Pleroma.Web.Streamer.StreamerSocket + + def from_socket(%{ + transport_pid: transport_pid, + assigns: %{user: nil} + }) do + %StreamerSocket{ + transport_pid: transport_pid + } + end + + def from_socket(%{ + transport_pid: transport_pid, + assigns: %{user: %User{} = user} + }) do + %StreamerSocket{ + transport_pid: transport_pid, + user: user + } + end + + def from_socket(%{transport_pid: transport_pid}) do + %StreamerSocket{ + transport_pid: transport_pid + } + end +end diff --git a/lib/pleroma/web/streamer/supervisor.ex b/lib/pleroma/web/streamer/supervisor.ex new file mode 100644 index 000000000..6afe19323 --- /dev/null +++ b/lib/pleroma/web/streamer/supervisor.ex @@ -0,0 +1,33 @@ +defmodule Pleroma.Web.Streamer.Supervisor do + use Supervisor + + def start_link(opts) do + Supervisor.start_link(__MODULE__, opts, name: __MODULE__) + end + + def init(args) do + children = [ + {Pleroma.Web.Streamer.State, args}, + {Pleroma.Web.Streamer.Ping, args}, + :poolboy.child_spec(:streamer_worker, poolboy_config()) + ] + + opts = [strategy: :one_for_one, name: Pleroma.Web.Streamer.Supervisor] + Supervisor.init(children, opts) + end + + defp poolboy_config do + opts = + Pleroma.Config.get(:streamer, + workers: 3, + overflow_workers: 2 + ) + + [ + {:name, {:local, :streamer_worker}}, + {:worker_module, Pleroma.Web.Streamer.Worker}, + {:size, opts[:workers]}, + {:max_overflow, opts[:overflow_workers]} + ] + end +end diff --git a/lib/pleroma/web/streamer/worker.ex b/lib/pleroma/web/streamer/worker.ex new file mode 100644 index 000000000..5804508eb --- /dev/null +++ b/lib/pleroma/web/streamer/worker.ex @@ -0,0 +1,220 @@ +defmodule Pleroma.Web.Streamer.Worker 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.CommonAPI + alias Pleroma.Web.Streamer.State + alias Pleroma.Web.Streamer.StreamerSocket + alias Pleroma.Web.StreamerView + + def start_link(_) do + GenServer.start_link(__MODULE__, %{}, []) + end + + def init(init_arg) do + {:ok, init_arg} + end + + def stream(pid, topics, items) do + GenServer.call(pid, {:stream, topics, items}) + end + + def handle_call({:stream, topics, item}, _from, state) when is_list(topics) do + Enum.each(topics, fn t -> + do_stream(%{topic: t, item: item}) + end) + + {:reply, state, state} + end + + def handle_call({:stream, topic, items}, _from, state) when is_list(items) do + Enum.each(items, fn i -> + do_stream(%{topic: topic, item: i}) + end) + + {:reply, state, state} + end + + def handle_call({:stream, topic, item}, _from, state) do + do_stream(%{topic: topic, item: item}) + + {:reply, state, state} + end + + defp do_stream(%{topic: "direct", item: item}) 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(State.get_sockets(), user_topic, item) + end) + end + + defp do_stream(%{topic: "participation", item: participation}) do + user_topic = "direct:#{participation.user_id}" + Logger.debug("Trying to push a conversation participation to #{user_topic}\n\n") + + push_to_socket(State.get_sockets(), user_topic, participation) + end + + defp do_stream(%{topic: "list", item: item}) 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(State.get_sockets(), list_topic, item) + end) + end + + defp do_stream(%{topic: topic, item: %Notification{} = item}) + when topic in ["user", "user:notification"] do + State.get_sockets() + |> Map.get("#{topic}:#{item.user_id}", []) + |> Enum.each(fn %StreamerSocket{transport_pid: transport_pid, user: socket_user} -> + with %User{} = user <- User.get_cached_by_ap_id(socket_user.ap_id), + true <- should_send?(user, item) do + send(transport_pid, {:text, StreamerView.render("notification.json", socket_user, item)}) + end + end) + end + + defp do_stream(%{topic: "user", item: item}) 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(State.get_sockets(), topic, item) + end) + end + + defp do_stream(%{topic: topic, item: item}) do + Logger.debug("Trying to push to #{topic}") + Logger.debug("Pushing item to #{topic}") + push_to_socket(State.get_sockets(), topic, item) + end + + defp should_send?(%User{} = user, %Activity{} = item) do + blocks = user.info.blocks || [] + mutes = user.info.mutes || [] + reblog_mutes = user.info.muted_reblogs || [] + domain_blocks = Pleroma.Web.ActivityPub.MRF.subdomains_regex(user.info.domain_blocks) + + 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)), + %{host: item_host} <- URI.parse(item.actor), + %{host: parent_host} <- URI.parse(parent.data["actor"]), + false <- Pleroma.Web.ActivityPub.MRF.subdomain_match?(domain_blocks, item_host), + false <- Pleroma.Web.ActivityPub.MRF.subdomain_match?(domain_blocks, parent_host), + true <- thread_containment(item, user), + false <- CommonAPI.thread_muted?(user, item) do + true + else + _ -> false + end + end + + defp should_send?(%User{} = user, %Notification{activity: activity}) do + should_send?(user, activity) + end + + def push_to_socket(topics, topic, %Activity{data: %{"type" => "Announce"}} = item) do + Enum.each(topics[topic] || [], fn %StreamerSocket{ + transport_pid: transport_pid, + user: socket_user + } -> + # Get the current user so we have up-to-date blocks etc. + if socket_user do + user = User.get_cached_by_ap_id(socket_user.ap_id) + + if should_send?(user, item) do + send(transport_pid, {:text, StreamerView.render("update.json", item, user)}) + end + else + send(transport_pid, {:text, StreamerView.render("update.json", item)}) + end + end) + end + + def push_to_socket(topics, topic, %Participation{} = participation) do + Enum.each(topics[topic] || [], fn %StreamerSocket{transport_pid: transport_pid} -> + send(transport_pid, {:text, StreamerView.render("conversation.json", 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 %StreamerSocket{transport_pid: transport_pid} -> + send( + 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 %StreamerSocket{ + transport_pid: transport_pid, + user: socket_user + } -> + # Get the current user so we have up-to-date blocks etc. + if socket_user do + user = User.get_cached_by_ap_id(socket_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(transport_pid, {:text, StreamerView.render("update.json", item, user)}) + end + else + send(transport_pid, {:text, StreamerView.render("update.json", item)}) + end + end) + end + + @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 diff --git a/lib/pleroma/web/views/streamer_view.ex b/lib/pleroma/web/views/streamer_view.ex new file mode 100644 index 000000000..b13030fa0 --- /dev/null +++ b/lib/pleroma/web/views/streamer_view.ex @@ -0,0 +1,66 @@ +# Pleroma: A lightweight social networking server +# Copyright © 2017-2019 Pleroma Authors +# SPDX-License-Identifier: AGPL-3.0-only + +defmodule Pleroma.Web.StreamerView do + use Pleroma.Web, :view + + alias Pleroma.Activity + alias Pleroma.Conversation.Participation + alias Pleroma.Notification + alias Pleroma.User + alias Pleroma.Web.MastodonAPI.NotificationView + + def render("update.json", %Activity{} = activity, %User{} = user) do + %{ + event: "update", + payload: + Pleroma.Web.MastodonAPI.StatusView.render( + "status.json", + activity: activity, + for: user + ) + |> Jason.encode!() + } + |> Jason.encode!() + end + + def render("notification.json", %User{} = user, %Notification{} = notify) do + %{ + event: "notification", + payload: + NotificationView.render( + "show.json", + %{notification: notify, for: user} + ) + |> Jason.encode!() + } + |> Jason.encode!() + end + + def render("update.json", %Activity{} = activity) do + %{ + event: "update", + payload: + Pleroma.Web.MastodonAPI.StatusView.render( + "status.json", + activity: activity + ) + |> Jason.encode!() + } + |> Jason.encode!() + end + + def render("conversation.json", %Participation{} = participation) do + %{ + event: "conversation", + payload: + Pleroma.Web.MastodonAPI.ConversationView.render("participation.json", %{ + participation: participation, + for: participation.user + }) + |> Jason.encode!() + } + |> Jason.encode!() + end +end -- cgit v1.2.3 From c623b4324deaf236334a0f77a81435b5bffadf3c Mon Sep 17 00:00:00 2001 From: kaniini Date: Mon, 16 Sep 2019 09:09:21 +0000 Subject: Revert "Merge branch 'streamer-refactoring' into 'develop'" This reverts merge request !1653 --- lib/pleroma/activity/ir/topics.ex | 63 ----- lib/pleroma/application.ex | 2 +- lib/pleroma/notification.ex | 6 +- lib/pleroma/web/activity_pub/activity_pub.ex | 50 +++- lib/pleroma/web/mastodon_api/websocket_handler.ex | 7 +- lib/pleroma/web/streamer.ex | 318 ++++++++++++++++++++++ lib/pleroma/web/streamer/ping.ex | 33 --- lib/pleroma/web/streamer/state.ex | 68 ----- lib/pleroma/web/streamer/streamer.ex | 55 ---- lib/pleroma/web/streamer/streamer_socket.ex | 31 --- lib/pleroma/web/streamer/supervisor.ex | 33 --- lib/pleroma/web/streamer/worker.ex | 220 --------------- lib/pleroma/web/views/streamer_view.ex | 66 ----- 13 files changed, 362 insertions(+), 590 deletions(-) delete mode 100644 lib/pleroma/activity/ir/topics.ex create mode 100644 lib/pleroma/web/streamer.ex delete mode 100644 lib/pleroma/web/streamer/ping.ex delete mode 100644 lib/pleroma/web/streamer/state.ex delete mode 100644 lib/pleroma/web/streamer/streamer.ex delete mode 100644 lib/pleroma/web/streamer/streamer_socket.ex delete mode 100644 lib/pleroma/web/streamer/supervisor.ex delete mode 100644 lib/pleroma/web/streamer/worker.ex delete mode 100644 lib/pleroma/web/views/streamer_view.ex (limited to 'lib') diff --git a/lib/pleroma/activity/ir/topics.ex b/lib/pleroma/activity/ir/topics.ex deleted file mode 100644 index 010897abc..000000000 --- a/lib/pleroma/activity/ir/topics.ex +++ /dev/null @@ -1,63 +0,0 @@ -# Pleroma: A lightweight social networking server -# Copyright © 2017-2019 Pleroma Authors -# SPDX-License-Identifier: AGPL-3.0-only - -defmodule Pleroma.Activity.Ir.Topics do - alias Pleroma.Object - alias Pleroma.Web.ActivityPub.Visibility - - def get_activity_topics(activity) do - activity - |> Object.normalize() - |> generate_topics(activity) - |> List.flatten() - end - - defp generate_topics(%{data: %{"type" => "Answer"}}, _) do - [] - end - - defp generate_topics(object, activity) do - ["user", "list"] ++ visibility_tags(object, activity) - end - - defp visibility_tags(object, activity) do - case Visibility.get_visibility(activity) do - "public" -> - if activity.local do - ["public", "public:local"] - else - ["public"] - end - |> item_creation_tags(object, activity) - - "direct" -> - ["direct"] - - _ -> - [] - end - end - - defp item_creation_tags(tags, %{data: %{"type" => "Create"}} = object, activity) do - tags ++ hashtags_to_topics(object) ++ attachment_topics(object, activity) - end - - defp item_creation_tags(tags, _, _) do - tags - end - - defp hashtags_to_topics(%{data: %{"tag" => tags}}) do - tags - |> Enum.filter(&is_bitstring(&1)) - |> Enum.map(fn tag -> "hashtag:" <> tag end) - end - - defp hashtags_to_topics(_), do: [] - - defp attachment_topics(%{data: %{"attachment" => []}}, _act), do: [] - - defp attachment_topics(_object, %{local: true}), do: ["public:media", "public:local:media"] - - defp attachment_topics(_object, _act), do: ["public:media"] -end diff --git a/lib/pleroma/application.ex b/lib/pleroma/application.ex index 3b37ce630..49094704b 100644 --- a/lib/pleroma/application.ex +++ b/lib/pleroma/application.ex @@ -141,7 +141,7 @@ defmodule Pleroma.Application do defp streamer_child(:test), do: [] defp streamer_child(_) do - [Pleroma.Web.Streamer.supervisor()] + [Pleroma.Web.Streamer] end defp oauth_cleanup_child(true), diff --git a/lib/pleroma/notification.ex b/lib/pleroma/notification.ex index 8012389ac..b7c880c51 100644 --- a/lib/pleroma/notification.ex +++ b/lib/pleroma/notification.ex @@ -210,10 +210,8 @@ defmodule Pleroma.Notification do unless skip?(activity, user) do notification = %Notification{user_id: user.id, activity: activity} {:ok, notification} = Repo.insert(notification) - - ["user", "user:notification"] - |> Streamer.stream(notification) - + Streamer.stream("user", notification) + Streamer.stream("user:notification", notification) Push.send(notification) notification end diff --git a/lib/pleroma/web/activity_pub/activity_pub.ex b/lib/pleroma/web/activity_pub/activity_pub.ex index bc5ae7fbf..41f6a0f1f 100644 --- a/lib/pleroma/web/activity_pub/activity_pub.ex +++ b/lib/pleroma/web/activity_pub/activity_pub.ex @@ -4,7 +4,6 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do alias Pleroma.Activity - alias Pleroma.Activity.Ir.Topics alias Pleroma.Config alias Pleroma.Conversation alias Pleroma.Notification @@ -17,7 +16,6 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do alias Pleroma.User alias Pleroma.Web.ActivityPub.MRF alias Pleroma.Web.ActivityPub.Transmogrifier - alias Pleroma.Web.Streamer alias Pleroma.Web.WebFinger alias Pleroma.Workers.BackgroundWorker @@ -189,7 +187,9 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do participations |> Repo.preload(:user) - Streamer.stream("participation", participations) + Enum.each(participations, fn participation -> + Pleroma.Web.Streamer.stream("participation", participation) + end) end def stream_out_participations(%Object{data: %{"context" => context}}, user) do @@ -208,15 +208,41 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do def stream_out_participations(_, _), do: :noop - def stream_out(%Activity{data: %{"type" => data_type}} = activity) - when data_type in ["Create", "Announce", "Delete"] do - activity - |> Topics.get_activity_topics() - |> Streamer.stream(activity) - end - - def stream_out(_activity) do - :noop + def stream_out(activity) do + if activity.data["type"] in ["Create", "Announce", "Delete"] do + object = Object.normalize(activity) + # Do not stream out poll replies + unless object.data["type"] == "Answer" do + Pleroma.Web.Streamer.stream("user", activity) + Pleroma.Web.Streamer.stream("list", activity) + + if get_visibility(activity) == "public" do + Pleroma.Web.Streamer.stream("public", activity) + + if activity.local do + Pleroma.Web.Streamer.stream("public:local", activity) + end + + if activity.data["type"] in ["Create"] do + object.data + |> Map.get("tag", []) + |> Enum.filter(fn tag -> is_bitstring(tag) end) + |> Enum.each(fn tag -> Pleroma.Web.Streamer.stream("hashtag:" <> tag, activity) end) + + if object.data["attachment"] != [] do + Pleroma.Web.Streamer.stream("public:media", activity) + + if activity.local do + Pleroma.Web.Streamer.stream("public:local:media", activity) + end + end + end + else + if get_visibility(activity) == "direct", + do: Pleroma.Web.Streamer.stream("direct", activity) + end + end + end end def create(%{to: to, actor: actor, context: context, object: object} = params, fake \\ false) do diff --git a/lib/pleroma/web/mastodon_api/websocket_handler.ex b/lib/pleroma/web/mastodon_api/websocket_handler.ex index 3c26eb406..dbd3542ea 100644 --- a/lib/pleroma/web/mastodon_api/websocket_handler.ex +++ b/lib/pleroma/web/mastodon_api/websocket_handler.ex @@ -8,7 +8,6 @@ defmodule Pleroma.Web.MastodonAPI.WebsocketHandler do alias Pleroma.Repo alias Pleroma.User alias Pleroma.Web.OAuth.Token - alias Pleroma.Web.Streamer @behaviour :cowboy_websocket @@ -25,7 +24,7 @@ defmodule Pleroma.Web.MastodonAPI.WebsocketHandler do ] @anonymous_streams ["public", "public:local", "hashtag"] - # Handled by periodic keepalive in Pleroma.Web.Streamer.Ping. + # Handled by periodic keepalive in Pleroma.Web.Streamer. @timeout :infinity def init(%{qs: qs} = req, state) do @@ -66,7 +65,7 @@ defmodule Pleroma.Web.MastodonAPI.WebsocketHandler do }, topic #{state.topic}" ) - Streamer.add_socket(state.topic, streamer_socket(state)) + Pleroma.Web.Streamer.add_socket(state.topic, streamer_socket(state)) {:ok, state} end @@ -81,7 +80,7 @@ defmodule Pleroma.Web.MastodonAPI.WebsocketHandler do }, topic #{state.topic || "?"}: #{inspect(reason)}" ) - Streamer.remove_socket(state.topic, streamer_socket(state)) + Pleroma.Web.Streamer.remove_socket(state.topic, streamer_socket(state)) :ok end diff --git a/lib/pleroma/web/streamer.ex b/lib/pleroma/web/streamer.ex new file mode 100644 index 000000000..587c43f40 --- /dev/null +++ b/lib/pleroma/web/streamer.ex @@ -0,0 +1,318 @@ +# Pleroma: A lightweight social networking server +# Copyright © 2017-2019 Pleroma Authors +# 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.CommonAPI + 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 + Process.send_after(self(), %{action: :ping}, @keepalive_interval) + + {:ok, args} + end + + def handle_info(%{action: :ping}, topics) do + topics + |> Map.values() + |> List.flatten() + |> Enum.each(fn socket -> + Logger.debug("Sending keepalive ping") + send(socket.transport_pid, {:text, ""}) + end) + + Process.send_after(self(), %{action: :ping}, @keepalive_interval) + + {: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: topic, item: %Notification{} = item}, + topics + ) + when topic in ["user", "user:notification"] do + topics + |> Map.get("#{topic}:#{item.user_id}", []) + |> Enum.each(fn socket -> + with %User{} = user <- User.get_cached_by_ap_id(socket.assigns[:user].ap_id), + true <- should_send?(user, item) do + send( + socket.transport_pid, + {:text, represent_notification(socket.assigns[:user], item)} + ) + end + 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, + for: participation.user + }) + |> Jason.encode!() + } + |> 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 + + defp should_send?(%User{} = user, %Activity{} = item) do + blocks = user.info.blocks || [] + mutes = user.info.mutes || [] + reblog_mutes = user.info.muted_reblogs || [] + domain_blocks = Pleroma.Web.ActivityPub.MRF.subdomains_regex(user.info.domain_blocks) + + 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)), + %{host: item_host} <- URI.parse(item.actor), + %{host: parent_host} <- URI.parse(parent.data["actor"]), + false <- Pleroma.Web.ActivityPub.MRF.subdomain_match?(domain_blocks, item_host), + false <- Pleroma.Web.ActivityPub.MRF.subdomain_match?(domain_blocks, parent_host), + true <- thread_containment(item, user), + false <- CommonAPI.thread_muted?(user, item) do + true + else + _ -> false + end + end + + defp should_send?(%User{} = user, %Notification{activity: activity}) do + should_send?(user, activity) + 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) + + if should_send?(user, item) 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 user:notification 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 diff --git a/lib/pleroma/web/streamer/ping.ex b/lib/pleroma/web/streamer/ping.ex deleted file mode 100644 index f77cbb95c..000000000 --- a/lib/pleroma/web/streamer/ping.ex +++ /dev/null @@ -1,33 +0,0 @@ -defmodule Pleroma.Web.Streamer.Ping do - use GenServer - require Logger - - alias Pleroma.Web.Streamer.State - alias Pleroma.Web.Streamer.StreamerSocket - - @keepalive_interval :timer.seconds(30) - - def start_link(opts) do - ping_interval = Keyword.get(opts, :ping_interval, @keepalive_interval) - GenServer.start_link(__MODULE__, %{ping_interval: ping_interval}, name: __MODULE__) - end - - def init(%{ping_interval: ping_interval} = args) do - Process.send_after(self(), :ping, ping_interval) - {:ok, args} - end - - def handle_info(:ping, %{ping_interval: ping_interval} = state) do - State.get_sockets() - |> Map.values() - |> List.flatten() - |> Enum.each(fn %StreamerSocket{transport_pid: transport_pid} -> - Logger.debug("Sending keepalive ping") - send(transport_pid, {:text, ""}) - end) - - Process.send_after(self(), :ping, ping_interval) - - {:noreply, state} - end -end diff --git a/lib/pleroma/web/streamer/state.ex b/lib/pleroma/web/streamer/state.ex deleted file mode 100644 index 7b5199068..000000000 --- a/lib/pleroma/web/streamer/state.ex +++ /dev/null @@ -1,68 +0,0 @@ -defmodule Pleroma.Web.Streamer.State do - use GenServer - require Logger - - alias Pleroma.Web.Streamer.StreamerSocket - - def start_link(_) do - GenServer.start_link(__MODULE__, %{sockets: %{}}, name: __MODULE__) - end - - def add_socket(topic, socket) do - GenServer.call(__MODULE__, {:add, socket, topic}) - end - - def remove_socket(topic, socket) do - GenServer.call(__MODULE__, {:remove, socket, topic}) - end - - def get_sockets do - %{sockets: stream_sockets} = GenServer.call(__MODULE__, :get_state) - stream_sockets - end - - def init(init_arg) do - {:ok, init_arg} - end - - def handle_call(:get_state, _from, state) do - {:reply, state, state} - end - - def handle_call({:add, socket, topic}, _from, %{sockets: sockets} = state) do - internal_topic = internal_topic(topic, socket) - stream_socket = StreamerSocket.from_socket(socket) - - sockets_for_topic = - sockets - |> Map.get(internal_topic, []) - |> List.insert_at(0, stream_socket) - |> Enum.uniq() - - state = put_in(state, [:sockets, internal_topic], sockets_for_topic) - Logger.debug("Got new conn for #{topic}") - {:reply, state, state} - end - - def handle_call({:remove, socket, topic}, _from, %{sockets: sockets} = state) do - internal_topic = internal_topic(topic, socket) - stream_socket = StreamerSocket.from_socket(socket) - - sockets_for_topic = - sockets - |> Map.get(internal_topic, []) - |> List.delete(stream_socket) - - state = Kernel.put_in(state, [:sockets, internal_topic], sockets_for_topic) - {:reply, state, state} - end - - defp internal_topic(topic, socket) - when topic in ~w[user user:notification direct] do - "#{topic}:#{socket.assigns[:user].id}" - end - - defp internal_topic(topic, _) do - topic - end -end diff --git a/lib/pleroma/web/streamer/streamer.ex b/lib/pleroma/web/streamer/streamer.ex deleted file mode 100644 index 8cf719277..000000000 --- a/lib/pleroma/web/streamer/streamer.ex +++ /dev/null @@ -1,55 +0,0 @@ -# Pleroma: A lightweight social networking server -# Copyright © 2017-2019 Pleroma Authors -# SPDX-License-Identifier: AGPL-3.0-only - -defmodule Pleroma.Web.Streamer do - alias Pleroma.Web.Streamer.State - alias Pleroma.Web.Streamer.Worker - - @timeout 60_000 - @mix_env Mix.env() - - def add_socket(topic, socket) do - State.add_socket(topic, socket) - end - - def remove_socket(topic, socket) do - State.remove_socket(topic, socket) - end - - def get_sockets do - State.get_sockets() - end - - def stream(topics, items) do - if should_send?() do - Task.async(fn -> - :poolboy.transaction( - :streamer_worker, - &Worker.stream(&1, topics, items), - @timeout - ) - end) - end - end - - def supervisor, do: Pleroma.Web.Streamer.Supervisor - - defp should_send? do - handle_should_send(@mix_env) - end - - defp handle_should_send(:test) do - case Process.whereis(:streamer_worker) do - nil -> - false - - pid -> - Process.alive?(pid) - end - end - - defp handle_should_send(_) do - true - end -end diff --git a/lib/pleroma/web/streamer/streamer_socket.ex b/lib/pleroma/web/streamer/streamer_socket.ex deleted file mode 100644 index f006c0306..000000000 --- a/lib/pleroma/web/streamer/streamer_socket.ex +++ /dev/null @@ -1,31 +0,0 @@ -defmodule Pleroma.Web.Streamer.StreamerSocket do - defstruct transport_pid: nil, user: nil - - alias Pleroma.User - alias Pleroma.Web.Streamer.StreamerSocket - - def from_socket(%{ - transport_pid: transport_pid, - assigns: %{user: nil} - }) do - %StreamerSocket{ - transport_pid: transport_pid - } - end - - def from_socket(%{ - transport_pid: transport_pid, - assigns: %{user: %User{} = user} - }) do - %StreamerSocket{ - transport_pid: transport_pid, - user: user - } - end - - def from_socket(%{transport_pid: transport_pid}) do - %StreamerSocket{ - transport_pid: transport_pid - } - end -end diff --git a/lib/pleroma/web/streamer/supervisor.ex b/lib/pleroma/web/streamer/supervisor.ex deleted file mode 100644 index 6afe19323..000000000 --- a/lib/pleroma/web/streamer/supervisor.ex +++ /dev/null @@ -1,33 +0,0 @@ -defmodule Pleroma.Web.Streamer.Supervisor do - use Supervisor - - def start_link(opts) do - Supervisor.start_link(__MODULE__, opts, name: __MODULE__) - end - - def init(args) do - children = [ - {Pleroma.Web.Streamer.State, args}, - {Pleroma.Web.Streamer.Ping, args}, - :poolboy.child_spec(:streamer_worker, poolboy_config()) - ] - - opts = [strategy: :one_for_one, name: Pleroma.Web.Streamer.Supervisor] - Supervisor.init(children, opts) - end - - defp poolboy_config do - opts = - Pleroma.Config.get(:streamer, - workers: 3, - overflow_workers: 2 - ) - - [ - {:name, {:local, :streamer_worker}}, - {:worker_module, Pleroma.Web.Streamer.Worker}, - {:size, opts[:workers]}, - {:max_overflow, opts[:overflow_workers]} - ] - end -end diff --git a/lib/pleroma/web/streamer/worker.ex b/lib/pleroma/web/streamer/worker.ex deleted file mode 100644 index 5804508eb..000000000 --- a/lib/pleroma/web/streamer/worker.ex +++ /dev/null @@ -1,220 +0,0 @@ -defmodule Pleroma.Web.Streamer.Worker 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.CommonAPI - alias Pleroma.Web.Streamer.State - alias Pleroma.Web.Streamer.StreamerSocket - alias Pleroma.Web.StreamerView - - def start_link(_) do - GenServer.start_link(__MODULE__, %{}, []) - end - - def init(init_arg) do - {:ok, init_arg} - end - - def stream(pid, topics, items) do - GenServer.call(pid, {:stream, topics, items}) - end - - def handle_call({:stream, topics, item}, _from, state) when is_list(topics) do - Enum.each(topics, fn t -> - do_stream(%{topic: t, item: item}) - end) - - {:reply, state, state} - end - - def handle_call({:stream, topic, items}, _from, state) when is_list(items) do - Enum.each(items, fn i -> - do_stream(%{topic: topic, item: i}) - end) - - {:reply, state, state} - end - - def handle_call({:stream, topic, item}, _from, state) do - do_stream(%{topic: topic, item: item}) - - {:reply, state, state} - end - - defp do_stream(%{topic: "direct", item: item}) 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(State.get_sockets(), user_topic, item) - end) - end - - defp do_stream(%{topic: "participation", item: participation}) do - user_topic = "direct:#{participation.user_id}" - Logger.debug("Trying to push a conversation participation to #{user_topic}\n\n") - - push_to_socket(State.get_sockets(), user_topic, participation) - end - - defp do_stream(%{topic: "list", item: item}) 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(State.get_sockets(), list_topic, item) - end) - end - - defp do_stream(%{topic: topic, item: %Notification{} = item}) - when topic in ["user", "user:notification"] do - State.get_sockets() - |> Map.get("#{topic}:#{item.user_id}", []) - |> Enum.each(fn %StreamerSocket{transport_pid: transport_pid, user: socket_user} -> - with %User{} = user <- User.get_cached_by_ap_id(socket_user.ap_id), - true <- should_send?(user, item) do - send(transport_pid, {:text, StreamerView.render("notification.json", socket_user, item)}) - end - end) - end - - defp do_stream(%{topic: "user", item: item}) 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(State.get_sockets(), topic, item) - end) - end - - defp do_stream(%{topic: topic, item: item}) do - Logger.debug("Trying to push to #{topic}") - Logger.debug("Pushing item to #{topic}") - push_to_socket(State.get_sockets(), topic, item) - end - - defp should_send?(%User{} = user, %Activity{} = item) do - blocks = user.info.blocks || [] - mutes = user.info.mutes || [] - reblog_mutes = user.info.muted_reblogs || [] - domain_blocks = Pleroma.Web.ActivityPub.MRF.subdomains_regex(user.info.domain_blocks) - - 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)), - %{host: item_host} <- URI.parse(item.actor), - %{host: parent_host} <- URI.parse(parent.data["actor"]), - false <- Pleroma.Web.ActivityPub.MRF.subdomain_match?(domain_blocks, item_host), - false <- Pleroma.Web.ActivityPub.MRF.subdomain_match?(domain_blocks, parent_host), - true <- thread_containment(item, user), - false <- CommonAPI.thread_muted?(user, item) do - true - else - _ -> false - end - end - - defp should_send?(%User{} = user, %Notification{activity: activity}) do - should_send?(user, activity) - end - - def push_to_socket(topics, topic, %Activity{data: %{"type" => "Announce"}} = item) do - Enum.each(topics[topic] || [], fn %StreamerSocket{ - transport_pid: transport_pid, - user: socket_user - } -> - # Get the current user so we have up-to-date blocks etc. - if socket_user do - user = User.get_cached_by_ap_id(socket_user.ap_id) - - if should_send?(user, item) do - send(transport_pid, {:text, StreamerView.render("update.json", item, user)}) - end - else - send(transport_pid, {:text, StreamerView.render("update.json", item)}) - end - end) - end - - def push_to_socket(topics, topic, %Participation{} = participation) do - Enum.each(topics[topic] || [], fn %StreamerSocket{transport_pid: transport_pid} -> - send(transport_pid, {:text, StreamerView.render("conversation.json", 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 %StreamerSocket{transport_pid: transport_pid} -> - send( - 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 %StreamerSocket{ - transport_pid: transport_pid, - user: socket_user - } -> - # Get the current user so we have up-to-date blocks etc. - if socket_user do - user = User.get_cached_by_ap_id(socket_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(transport_pid, {:text, StreamerView.render("update.json", item, user)}) - end - else - send(transport_pid, {:text, StreamerView.render("update.json", item)}) - end - end) - end - - @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 diff --git a/lib/pleroma/web/views/streamer_view.ex b/lib/pleroma/web/views/streamer_view.ex deleted file mode 100644 index b13030fa0..000000000 --- a/lib/pleroma/web/views/streamer_view.ex +++ /dev/null @@ -1,66 +0,0 @@ -# Pleroma: A lightweight social networking server -# Copyright © 2017-2019 Pleroma Authors -# SPDX-License-Identifier: AGPL-3.0-only - -defmodule Pleroma.Web.StreamerView do - use Pleroma.Web, :view - - alias Pleroma.Activity - alias Pleroma.Conversation.Participation - alias Pleroma.Notification - alias Pleroma.User - alias Pleroma.Web.MastodonAPI.NotificationView - - def render("update.json", %Activity{} = activity, %User{} = user) do - %{ - event: "update", - payload: - Pleroma.Web.MastodonAPI.StatusView.render( - "status.json", - activity: activity, - for: user - ) - |> Jason.encode!() - } - |> Jason.encode!() - end - - def render("notification.json", %User{} = user, %Notification{} = notify) do - %{ - event: "notification", - payload: - NotificationView.render( - "show.json", - %{notification: notify, for: user} - ) - |> Jason.encode!() - } - |> Jason.encode!() - end - - def render("update.json", %Activity{} = activity) do - %{ - event: "update", - payload: - Pleroma.Web.MastodonAPI.StatusView.render( - "status.json", - activity: activity - ) - |> Jason.encode!() - } - |> Jason.encode!() - end - - def render("conversation.json", %Participation{} = participation) do - %{ - event: "conversation", - payload: - Pleroma.Web.MastodonAPI.ConversationView.render("participation.json", %{ - participation: participation, - for: participation.user - }) - |> Jason.encode!() - } - |> Jason.encode!() - end -end -- cgit v1.2.3 From 96816ceaa25c21cec7677e75dcddd7ffb42d83c3 Mon Sep 17 00:00:00 2001 From: Egor Kislitsyn Date: Mon, 16 Sep 2019 17:03:37 +0700 Subject: Revert "Merge branch 'revert-4fabf83a' into 'develop'" This reverts commit fe7fd331263007e0fb2877ef7370a09a9704da36, reversing changes made to 4fabf83ad01352442906d79187aeab4c777f4df8. --- lib/pleroma/activity/ir/topics.ex | 63 +++++ lib/pleroma/application.ex | 2 +- lib/pleroma/notification.ex | 6 +- lib/pleroma/web/activity_pub/activity_pub.ex | 50 +--- lib/pleroma/web/mastodon_api/websocket_handler.ex | 7 +- lib/pleroma/web/streamer.ex | 318 ---------------------- lib/pleroma/web/streamer/ping.ex | 33 +++ lib/pleroma/web/streamer/state.ex | 68 +++++ lib/pleroma/web/streamer/streamer.ex | 55 ++++ lib/pleroma/web/streamer/streamer_socket.ex | 31 +++ lib/pleroma/web/streamer/supervisor.ex | 33 +++ lib/pleroma/web/streamer/worker.ex | 220 +++++++++++++++ lib/pleroma/web/views/streamer_view.ex | 66 +++++ 13 files changed, 590 insertions(+), 362 deletions(-) create mode 100644 lib/pleroma/activity/ir/topics.ex delete mode 100644 lib/pleroma/web/streamer.ex create mode 100644 lib/pleroma/web/streamer/ping.ex create mode 100644 lib/pleroma/web/streamer/state.ex create mode 100644 lib/pleroma/web/streamer/streamer.ex create mode 100644 lib/pleroma/web/streamer/streamer_socket.ex create mode 100644 lib/pleroma/web/streamer/supervisor.ex create mode 100644 lib/pleroma/web/streamer/worker.ex create mode 100644 lib/pleroma/web/views/streamer_view.ex (limited to 'lib') diff --git a/lib/pleroma/activity/ir/topics.ex b/lib/pleroma/activity/ir/topics.ex new file mode 100644 index 000000000..010897abc --- /dev/null +++ b/lib/pleroma/activity/ir/topics.ex @@ -0,0 +1,63 @@ +# Pleroma: A lightweight social networking server +# Copyright © 2017-2019 Pleroma Authors +# SPDX-License-Identifier: AGPL-3.0-only + +defmodule Pleroma.Activity.Ir.Topics do + alias Pleroma.Object + alias Pleroma.Web.ActivityPub.Visibility + + def get_activity_topics(activity) do + activity + |> Object.normalize() + |> generate_topics(activity) + |> List.flatten() + end + + defp generate_topics(%{data: %{"type" => "Answer"}}, _) do + [] + end + + defp generate_topics(object, activity) do + ["user", "list"] ++ visibility_tags(object, activity) + end + + defp visibility_tags(object, activity) do + case Visibility.get_visibility(activity) do + "public" -> + if activity.local do + ["public", "public:local"] + else + ["public"] + end + |> item_creation_tags(object, activity) + + "direct" -> + ["direct"] + + _ -> + [] + end + end + + defp item_creation_tags(tags, %{data: %{"type" => "Create"}} = object, activity) do + tags ++ hashtags_to_topics(object) ++ attachment_topics(object, activity) + end + + defp item_creation_tags(tags, _, _) do + tags + end + + defp hashtags_to_topics(%{data: %{"tag" => tags}}) do + tags + |> Enum.filter(&is_bitstring(&1)) + |> Enum.map(fn tag -> "hashtag:" <> tag end) + end + + defp hashtags_to_topics(_), do: [] + + defp attachment_topics(%{data: %{"attachment" => []}}, _act), do: [] + + defp attachment_topics(_object, %{local: true}), do: ["public:media", "public:local:media"] + + defp attachment_topics(_object, _act), do: ["public:media"] +end diff --git a/lib/pleroma/application.ex b/lib/pleroma/application.ex index 49094704b..3b37ce630 100644 --- a/lib/pleroma/application.ex +++ b/lib/pleroma/application.ex @@ -141,7 +141,7 @@ defmodule Pleroma.Application do defp streamer_child(:test), do: [] defp streamer_child(_) do - [Pleroma.Web.Streamer] + [Pleroma.Web.Streamer.supervisor()] end defp oauth_cleanup_child(true), diff --git a/lib/pleroma/notification.ex b/lib/pleroma/notification.ex index b7c880c51..8012389ac 100644 --- a/lib/pleroma/notification.ex +++ b/lib/pleroma/notification.ex @@ -210,8 +210,10 @@ defmodule Pleroma.Notification do unless skip?(activity, user) do notification = %Notification{user_id: user.id, activity: activity} {:ok, notification} = Repo.insert(notification) - Streamer.stream("user", notification) - Streamer.stream("user:notification", notification) + + ["user", "user:notification"] + |> Streamer.stream(notification) + Push.send(notification) notification end diff --git a/lib/pleroma/web/activity_pub/activity_pub.ex b/lib/pleroma/web/activity_pub/activity_pub.ex index 41f6a0f1f..bc5ae7fbf 100644 --- a/lib/pleroma/web/activity_pub/activity_pub.ex +++ b/lib/pleroma/web/activity_pub/activity_pub.ex @@ -4,6 +4,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do alias Pleroma.Activity + alias Pleroma.Activity.Ir.Topics alias Pleroma.Config alias Pleroma.Conversation alias Pleroma.Notification @@ -16,6 +17,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do alias Pleroma.User alias Pleroma.Web.ActivityPub.MRF alias Pleroma.Web.ActivityPub.Transmogrifier + alias Pleroma.Web.Streamer alias Pleroma.Web.WebFinger alias Pleroma.Workers.BackgroundWorker @@ -187,9 +189,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do participations |> Repo.preload(:user) - Enum.each(participations, fn participation -> - Pleroma.Web.Streamer.stream("participation", participation) - end) + Streamer.stream("participation", participations) end def stream_out_participations(%Object{data: %{"context" => context}}, user) do @@ -208,41 +208,15 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do def stream_out_participations(_, _), do: :noop - def stream_out(activity) do - if activity.data["type"] in ["Create", "Announce", "Delete"] do - object = Object.normalize(activity) - # Do not stream out poll replies - unless object.data["type"] == "Answer" do - Pleroma.Web.Streamer.stream("user", activity) - Pleroma.Web.Streamer.stream("list", activity) - - if get_visibility(activity) == "public" do - Pleroma.Web.Streamer.stream("public", activity) - - if activity.local do - Pleroma.Web.Streamer.stream("public:local", activity) - end - - if activity.data["type"] in ["Create"] do - object.data - |> Map.get("tag", []) - |> Enum.filter(fn tag -> is_bitstring(tag) end) - |> Enum.each(fn tag -> Pleroma.Web.Streamer.stream("hashtag:" <> tag, activity) end) - - if object.data["attachment"] != [] do - Pleroma.Web.Streamer.stream("public:media", activity) - - if activity.local do - Pleroma.Web.Streamer.stream("public:local:media", activity) - end - end - end - else - if get_visibility(activity) == "direct", - do: Pleroma.Web.Streamer.stream("direct", activity) - end - end - end + def stream_out(%Activity{data: %{"type" => data_type}} = activity) + when data_type in ["Create", "Announce", "Delete"] do + activity + |> Topics.get_activity_topics() + |> Streamer.stream(activity) + end + + def stream_out(_activity) do + :noop end def create(%{to: to, actor: actor, context: context, object: object} = params, fake \\ false) do diff --git a/lib/pleroma/web/mastodon_api/websocket_handler.ex b/lib/pleroma/web/mastodon_api/websocket_handler.ex index dbd3542ea..3c26eb406 100644 --- a/lib/pleroma/web/mastodon_api/websocket_handler.ex +++ b/lib/pleroma/web/mastodon_api/websocket_handler.ex @@ -8,6 +8,7 @@ defmodule Pleroma.Web.MastodonAPI.WebsocketHandler do alias Pleroma.Repo alias Pleroma.User alias Pleroma.Web.OAuth.Token + alias Pleroma.Web.Streamer @behaviour :cowboy_websocket @@ -24,7 +25,7 @@ defmodule Pleroma.Web.MastodonAPI.WebsocketHandler do ] @anonymous_streams ["public", "public:local", "hashtag"] - # Handled by periodic keepalive in Pleroma.Web.Streamer. + # Handled by periodic keepalive in Pleroma.Web.Streamer.Ping. @timeout :infinity def init(%{qs: qs} = req, state) do @@ -65,7 +66,7 @@ defmodule Pleroma.Web.MastodonAPI.WebsocketHandler do }, topic #{state.topic}" ) - Pleroma.Web.Streamer.add_socket(state.topic, streamer_socket(state)) + Streamer.add_socket(state.topic, streamer_socket(state)) {:ok, state} end @@ -80,7 +81,7 @@ defmodule Pleroma.Web.MastodonAPI.WebsocketHandler do }, topic #{state.topic || "?"}: #{inspect(reason)}" ) - Pleroma.Web.Streamer.remove_socket(state.topic, streamer_socket(state)) + Streamer.remove_socket(state.topic, streamer_socket(state)) :ok end diff --git a/lib/pleroma/web/streamer.ex b/lib/pleroma/web/streamer.ex deleted file mode 100644 index 587c43f40..000000000 --- a/lib/pleroma/web/streamer.ex +++ /dev/null @@ -1,318 +0,0 @@ -# Pleroma: A lightweight social networking server -# Copyright © 2017-2019 Pleroma Authors -# 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.CommonAPI - 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 - Process.send_after(self(), %{action: :ping}, @keepalive_interval) - - {:ok, args} - end - - def handle_info(%{action: :ping}, topics) do - topics - |> Map.values() - |> List.flatten() - |> Enum.each(fn socket -> - Logger.debug("Sending keepalive ping") - send(socket.transport_pid, {:text, ""}) - end) - - Process.send_after(self(), %{action: :ping}, @keepalive_interval) - - {: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: topic, item: %Notification{} = item}, - topics - ) - when topic in ["user", "user:notification"] do - topics - |> Map.get("#{topic}:#{item.user_id}", []) - |> Enum.each(fn socket -> - with %User{} = user <- User.get_cached_by_ap_id(socket.assigns[:user].ap_id), - true <- should_send?(user, item) do - send( - socket.transport_pid, - {:text, represent_notification(socket.assigns[:user], item)} - ) - end - 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, - for: participation.user - }) - |> Jason.encode!() - } - |> 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 - - defp should_send?(%User{} = user, %Activity{} = item) do - blocks = user.info.blocks || [] - mutes = user.info.mutes || [] - reblog_mutes = user.info.muted_reblogs || [] - domain_blocks = Pleroma.Web.ActivityPub.MRF.subdomains_regex(user.info.domain_blocks) - - 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)), - %{host: item_host} <- URI.parse(item.actor), - %{host: parent_host} <- URI.parse(parent.data["actor"]), - false <- Pleroma.Web.ActivityPub.MRF.subdomain_match?(domain_blocks, item_host), - false <- Pleroma.Web.ActivityPub.MRF.subdomain_match?(domain_blocks, parent_host), - true <- thread_containment(item, user), - false <- CommonAPI.thread_muted?(user, item) do - true - else - _ -> false - end - end - - defp should_send?(%User{} = user, %Notification{activity: activity}) do - should_send?(user, activity) - 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) - - if should_send?(user, item) 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 user:notification 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 diff --git a/lib/pleroma/web/streamer/ping.ex b/lib/pleroma/web/streamer/ping.ex new file mode 100644 index 000000000..f77cbb95c --- /dev/null +++ b/lib/pleroma/web/streamer/ping.ex @@ -0,0 +1,33 @@ +defmodule Pleroma.Web.Streamer.Ping do + use GenServer + require Logger + + alias Pleroma.Web.Streamer.State + alias Pleroma.Web.Streamer.StreamerSocket + + @keepalive_interval :timer.seconds(30) + + def start_link(opts) do + ping_interval = Keyword.get(opts, :ping_interval, @keepalive_interval) + GenServer.start_link(__MODULE__, %{ping_interval: ping_interval}, name: __MODULE__) + end + + def init(%{ping_interval: ping_interval} = args) do + Process.send_after(self(), :ping, ping_interval) + {:ok, args} + end + + def handle_info(:ping, %{ping_interval: ping_interval} = state) do + State.get_sockets() + |> Map.values() + |> List.flatten() + |> Enum.each(fn %StreamerSocket{transport_pid: transport_pid} -> + Logger.debug("Sending keepalive ping") + send(transport_pid, {:text, ""}) + end) + + Process.send_after(self(), :ping, ping_interval) + + {:noreply, state} + end +end diff --git a/lib/pleroma/web/streamer/state.ex b/lib/pleroma/web/streamer/state.ex new file mode 100644 index 000000000..7b5199068 --- /dev/null +++ b/lib/pleroma/web/streamer/state.ex @@ -0,0 +1,68 @@ +defmodule Pleroma.Web.Streamer.State do + use GenServer + require Logger + + alias Pleroma.Web.Streamer.StreamerSocket + + def start_link(_) do + GenServer.start_link(__MODULE__, %{sockets: %{}}, name: __MODULE__) + end + + def add_socket(topic, socket) do + GenServer.call(__MODULE__, {:add, socket, topic}) + end + + def remove_socket(topic, socket) do + GenServer.call(__MODULE__, {:remove, socket, topic}) + end + + def get_sockets do + %{sockets: stream_sockets} = GenServer.call(__MODULE__, :get_state) + stream_sockets + end + + def init(init_arg) do + {:ok, init_arg} + end + + def handle_call(:get_state, _from, state) do + {:reply, state, state} + end + + def handle_call({:add, socket, topic}, _from, %{sockets: sockets} = state) do + internal_topic = internal_topic(topic, socket) + stream_socket = StreamerSocket.from_socket(socket) + + sockets_for_topic = + sockets + |> Map.get(internal_topic, []) + |> List.insert_at(0, stream_socket) + |> Enum.uniq() + + state = put_in(state, [:sockets, internal_topic], sockets_for_topic) + Logger.debug("Got new conn for #{topic}") + {:reply, state, state} + end + + def handle_call({:remove, socket, topic}, _from, %{sockets: sockets} = state) do + internal_topic = internal_topic(topic, socket) + stream_socket = StreamerSocket.from_socket(socket) + + sockets_for_topic = + sockets + |> Map.get(internal_topic, []) + |> List.delete(stream_socket) + + state = Kernel.put_in(state, [:sockets, internal_topic], sockets_for_topic) + {:reply, state, state} + end + + defp internal_topic(topic, socket) + when topic in ~w[user user:notification direct] do + "#{topic}:#{socket.assigns[:user].id}" + end + + defp internal_topic(topic, _) do + topic + end +end diff --git a/lib/pleroma/web/streamer/streamer.ex b/lib/pleroma/web/streamer/streamer.ex new file mode 100644 index 000000000..8cf719277 --- /dev/null +++ b/lib/pleroma/web/streamer/streamer.ex @@ -0,0 +1,55 @@ +# Pleroma: A lightweight social networking server +# Copyright © 2017-2019 Pleroma Authors +# SPDX-License-Identifier: AGPL-3.0-only + +defmodule Pleroma.Web.Streamer do + alias Pleroma.Web.Streamer.State + alias Pleroma.Web.Streamer.Worker + + @timeout 60_000 + @mix_env Mix.env() + + def add_socket(topic, socket) do + State.add_socket(topic, socket) + end + + def remove_socket(topic, socket) do + State.remove_socket(topic, socket) + end + + def get_sockets do + State.get_sockets() + end + + def stream(topics, items) do + if should_send?() do + Task.async(fn -> + :poolboy.transaction( + :streamer_worker, + &Worker.stream(&1, topics, items), + @timeout + ) + end) + end + end + + def supervisor, do: Pleroma.Web.Streamer.Supervisor + + defp should_send? do + handle_should_send(@mix_env) + end + + defp handle_should_send(:test) do + case Process.whereis(:streamer_worker) do + nil -> + false + + pid -> + Process.alive?(pid) + end + end + + defp handle_should_send(_) do + true + end +end diff --git a/lib/pleroma/web/streamer/streamer_socket.ex b/lib/pleroma/web/streamer/streamer_socket.ex new file mode 100644 index 000000000..f006c0306 --- /dev/null +++ b/lib/pleroma/web/streamer/streamer_socket.ex @@ -0,0 +1,31 @@ +defmodule Pleroma.Web.Streamer.StreamerSocket do + defstruct transport_pid: nil, user: nil + + alias Pleroma.User + alias Pleroma.Web.Streamer.StreamerSocket + + def from_socket(%{ + transport_pid: transport_pid, + assigns: %{user: nil} + }) do + %StreamerSocket{ + transport_pid: transport_pid + } + end + + def from_socket(%{ + transport_pid: transport_pid, + assigns: %{user: %User{} = user} + }) do + %StreamerSocket{ + transport_pid: transport_pid, + user: user + } + end + + def from_socket(%{transport_pid: transport_pid}) do + %StreamerSocket{ + transport_pid: transport_pid + } + end +end diff --git a/lib/pleroma/web/streamer/supervisor.ex b/lib/pleroma/web/streamer/supervisor.ex new file mode 100644 index 000000000..6afe19323 --- /dev/null +++ b/lib/pleroma/web/streamer/supervisor.ex @@ -0,0 +1,33 @@ +defmodule Pleroma.Web.Streamer.Supervisor do + use Supervisor + + def start_link(opts) do + Supervisor.start_link(__MODULE__, opts, name: __MODULE__) + end + + def init(args) do + children = [ + {Pleroma.Web.Streamer.State, args}, + {Pleroma.Web.Streamer.Ping, args}, + :poolboy.child_spec(:streamer_worker, poolboy_config()) + ] + + opts = [strategy: :one_for_one, name: Pleroma.Web.Streamer.Supervisor] + Supervisor.init(children, opts) + end + + defp poolboy_config do + opts = + Pleroma.Config.get(:streamer, + workers: 3, + overflow_workers: 2 + ) + + [ + {:name, {:local, :streamer_worker}}, + {:worker_module, Pleroma.Web.Streamer.Worker}, + {:size, opts[:workers]}, + {:max_overflow, opts[:overflow_workers]} + ] + end +end diff --git a/lib/pleroma/web/streamer/worker.ex b/lib/pleroma/web/streamer/worker.ex new file mode 100644 index 000000000..5804508eb --- /dev/null +++ b/lib/pleroma/web/streamer/worker.ex @@ -0,0 +1,220 @@ +defmodule Pleroma.Web.Streamer.Worker 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.CommonAPI + alias Pleroma.Web.Streamer.State + alias Pleroma.Web.Streamer.StreamerSocket + alias Pleroma.Web.StreamerView + + def start_link(_) do + GenServer.start_link(__MODULE__, %{}, []) + end + + def init(init_arg) do + {:ok, init_arg} + end + + def stream(pid, topics, items) do + GenServer.call(pid, {:stream, topics, items}) + end + + def handle_call({:stream, topics, item}, _from, state) when is_list(topics) do + Enum.each(topics, fn t -> + do_stream(%{topic: t, item: item}) + end) + + {:reply, state, state} + end + + def handle_call({:stream, topic, items}, _from, state) when is_list(items) do + Enum.each(items, fn i -> + do_stream(%{topic: topic, item: i}) + end) + + {:reply, state, state} + end + + def handle_call({:stream, topic, item}, _from, state) do + do_stream(%{topic: topic, item: item}) + + {:reply, state, state} + end + + defp do_stream(%{topic: "direct", item: item}) 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(State.get_sockets(), user_topic, item) + end) + end + + defp do_stream(%{topic: "participation", item: participation}) do + user_topic = "direct:#{participation.user_id}" + Logger.debug("Trying to push a conversation participation to #{user_topic}\n\n") + + push_to_socket(State.get_sockets(), user_topic, participation) + end + + defp do_stream(%{topic: "list", item: item}) 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(State.get_sockets(), list_topic, item) + end) + end + + defp do_stream(%{topic: topic, item: %Notification{} = item}) + when topic in ["user", "user:notification"] do + State.get_sockets() + |> Map.get("#{topic}:#{item.user_id}", []) + |> Enum.each(fn %StreamerSocket{transport_pid: transport_pid, user: socket_user} -> + with %User{} = user <- User.get_cached_by_ap_id(socket_user.ap_id), + true <- should_send?(user, item) do + send(transport_pid, {:text, StreamerView.render("notification.json", socket_user, item)}) + end + end) + end + + defp do_stream(%{topic: "user", item: item}) 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(State.get_sockets(), topic, item) + end) + end + + defp do_stream(%{topic: topic, item: item}) do + Logger.debug("Trying to push to #{topic}") + Logger.debug("Pushing item to #{topic}") + push_to_socket(State.get_sockets(), topic, item) + end + + defp should_send?(%User{} = user, %Activity{} = item) do + blocks = user.info.blocks || [] + mutes = user.info.mutes || [] + reblog_mutes = user.info.muted_reblogs || [] + domain_blocks = Pleroma.Web.ActivityPub.MRF.subdomains_regex(user.info.domain_blocks) + + 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)), + %{host: item_host} <- URI.parse(item.actor), + %{host: parent_host} <- URI.parse(parent.data["actor"]), + false <- Pleroma.Web.ActivityPub.MRF.subdomain_match?(domain_blocks, item_host), + false <- Pleroma.Web.ActivityPub.MRF.subdomain_match?(domain_blocks, parent_host), + true <- thread_containment(item, user), + false <- CommonAPI.thread_muted?(user, item) do + true + else + _ -> false + end + end + + defp should_send?(%User{} = user, %Notification{activity: activity}) do + should_send?(user, activity) + end + + def push_to_socket(topics, topic, %Activity{data: %{"type" => "Announce"}} = item) do + Enum.each(topics[topic] || [], fn %StreamerSocket{ + transport_pid: transport_pid, + user: socket_user + } -> + # Get the current user so we have up-to-date blocks etc. + if socket_user do + user = User.get_cached_by_ap_id(socket_user.ap_id) + + if should_send?(user, item) do + send(transport_pid, {:text, StreamerView.render("update.json", item, user)}) + end + else + send(transport_pid, {:text, StreamerView.render("update.json", item)}) + end + end) + end + + def push_to_socket(topics, topic, %Participation{} = participation) do + Enum.each(topics[topic] || [], fn %StreamerSocket{transport_pid: transport_pid} -> + send(transport_pid, {:text, StreamerView.render("conversation.json", 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 %StreamerSocket{transport_pid: transport_pid} -> + send( + 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 %StreamerSocket{ + transport_pid: transport_pid, + user: socket_user + } -> + # Get the current user so we have up-to-date blocks etc. + if socket_user do + user = User.get_cached_by_ap_id(socket_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(transport_pid, {:text, StreamerView.render("update.json", item, user)}) + end + else + send(transport_pid, {:text, StreamerView.render("update.json", item)}) + end + end) + end + + @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 diff --git a/lib/pleroma/web/views/streamer_view.ex b/lib/pleroma/web/views/streamer_view.ex new file mode 100644 index 000000000..b13030fa0 --- /dev/null +++ b/lib/pleroma/web/views/streamer_view.ex @@ -0,0 +1,66 @@ +# Pleroma: A lightweight social networking server +# Copyright © 2017-2019 Pleroma Authors +# SPDX-License-Identifier: AGPL-3.0-only + +defmodule Pleroma.Web.StreamerView do + use Pleroma.Web, :view + + alias Pleroma.Activity + alias Pleroma.Conversation.Participation + alias Pleroma.Notification + alias Pleroma.User + alias Pleroma.Web.MastodonAPI.NotificationView + + def render("update.json", %Activity{} = activity, %User{} = user) do + %{ + event: "update", + payload: + Pleroma.Web.MastodonAPI.StatusView.render( + "status.json", + activity: activity, + for: user + ) + |> Jason.encode!() + } + |> Jason.encode!() + end + + def render("notification.json", %User{} = user, %Notification{} = notify) do + %{ + event: "notification", + payload: + NotificationView.render( + "show.json", + %{notification: notify, for: user} + ) + |> Jason.encode!() + } + |> Jason.encode!() + end + + def render("update.json", %Activity{} = activity) do + %{ + event: "update", + payload: + Pleroma.Web.MastodonAPI.StatusView.render( + "status.json", + activity: activity + ) + |> Jason.encode!() + } + |> Jason.encode!() + end + + def render("conversation.json", %Participation{} = participation) do + %{ + event: "conversation", + payload: + Pleroma.Web.MastodonAPI.ConversationView.render("participation.json", %{ + participation: participation, + for: participation.user + }) + |> Jason.encode!() + } + |> Jason.encode!() + end +end -- cgit v1.2.3 From 085d014f0859b3b3e5023c423ae0361ec6ed6c67 Mon Sep 17 00:00:00 2001 From: Egor Kislitsyn Date: Mon, 16 Sep 2019 19:26:00 +0700 Subject: Fix `Transmogrifier.upgrade_user_from_ap_id/1` --- lib/pleroma/web/activity_pub/transmogrifier.ex | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'lib') diff --git a/lib/pleroma/web/activity_pub/transmogrifier.ex b/lib/pleroma/web/activity_pub/transmogrifier.ex index acb3087d0..8461b666e 100644 --- a/lib/pleroma/web/activity_pub/transmogrifier.ex +++ b/lib/pleroma/web/activity_pub/transmogrifier.ex @@ -1050,7 +1050,7 @@ defmodule Pleroma.Web.ActivityPub.Transmogrifier do with %User{local: false} = user <- User.get_cached_by_ap_id(ap_id), {:ok, data} <- ActivityPub.fetch_and_prepare_user_from_ap_id(ap_id), already_ap <- User.ap_enabled?(user), - {:ok, user} <- user |> User.upgrade_changeset(data) |> User.update_and_set_cache() do + {:ok, user} <- user |> User.upgrade_changeset(data, true) |> User.update_and_set_cache() do unless already_ap do TransmogrifierWorker.enqueue("user_upgrade", %{"user_id" => user.id}) end -- cgit v1.2.3 From a21584556f2c3edb90db3c58ba2a4829a7e220c1 Mon Sep 17 00:00:00 2001 From: rinpatch Date: Tue, 17 Sep 2019 13:04:43 +0300 Subject: Update oban to 0.8.1 This version uses a different locking mechanism, which gets rid of `WARNING: you don't own a lock of type ShareLock` log spam --- lib/pleroma/flake_id.ex | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'lib') diff --git a/lib/pleroma/flake_id.ex b/lib/pleroma/flake_id.ex index 47d61ca5f..042cf8659 100644 --- a/lib/pleroma/flake_id.ex +++ b/lib/pleroma/flake_id.ex @@ -14,7 +14,7 @@ defmodule Pleroma.FlakeId do @type t :: binary - @behaviour Ecto.Type + use Ecto.Type use GenServer require Logger alias __MODULE__ -- cgit v1.2.3 From 80c5c3495bdd7939e576c8746a959f3f89f44042 Mon Sep 17 00:00:00 2001 From: Steven Fuchs Date: Tue, 17 Sep 2019 14:44:52 +0000 Subject: remove remaining errors from tests --- lib/pleroma/application.ex | 53 +++++++++++++++++++++++++++------------ lib/pleroma/web/streamer/state.ex | 18 ++++++++++--- 2 files changed, 51 insertions(+), 20 deletions(-) (limited to 'lib') diff --git a/lib/pleroma/application.ex b/lib/pleroma/application.ex index 3b37ce630..dabce771d 100644 --- a/lib/pleroma/application.ex +++ b/lib/pleroma/application.ex @@ -43,23 +43,9 @@ defmodule Pleroma.Application do hackney_pool_children() ++ [ Pleroma.Stats, - {Oban, Pleroma.Config.get(Oban)}, - %{ - id: :web_push_init, - start: {Task, :start_link, [&Pleroma.Web.Push.init/0]}, - restart: :temporary - }, - %{ - id: :federator_init, - start: {Task, :start_link, [&Pleroma.Web.Federator.init/0]}, - restart: :temporary - }, - %{ - id: :internal_fetch_init, - start: {Task, :start_link, [&Pleroma.Web.ActivityPub.InternalFetchActor.init/0]}, - restart: :temporary - } + {Oban, Pleroma.Config.get(Oban)} ] ++ + task_children(@env) ++ oauth_cleanup_child(oauth_cleanup_enabled?()) ++ streamer_child(@env) ++ chat_child(@env, chat_enabled?()) ++ @@ -163,4 +149,39 @@ defmodule Pleroma.Application do :hackney_pool.child_spec(pool, options) end end + + defp task_children(:test) do + [ + %{ + id: :web_push_init, + start: {Task, :start_link, [&Pleroma.Web.Push.init/0]}, + restart: :temporary + }, + %{ + id: :federator_init, + start: {Task, :start_link, [&Pleroma.Web.Federator.init/0]}, + restart: :temporary + } + ] + end + + defp task_children(_) do + [ + %{ + id: :web_push_init, + start: {Task, :start_link, [&Pleroma.Web.Push.init/0]}, + restart: :temporary + }, + %{ + id: :federator_init, + start: {Task, :start_link, [&Pleroma.Web.Federator.init/0]}, + restart: :temporary + }, + %{ + id: :internal_fetch_init, + start: {Task, :start_link, [&Pleroma.Web.ActivityPub.InternalFetchActor.init/0]}, + restart: :temporary + } + ] + end end diff --git a/lib/pleroma/web/streamer/state.ex b/lib/pleroma/web/streamer/state.ex index 7b5199068..c48752d95 100644 --- a/lib/pleroma/web/streamer/state.ex +++ b/lib/pleroma/web/streamer/state.ex @@ -4,16 +4,18 @@ defmodule Pleroma.Web.Streamer.State do alias Pleroma.Web.Streamer.StreamerSocket + @env Mix.env() + def start_link(_) do GenServer.start_link(__MODULE__, %{sockets: %{}}, name: __MODULE__) end def add_socket(topic, socket) do - GenServer.call(__MODULE__, {:add, socket, topic}) + GenServer.call(__MODULE__, {:add, topic, socket}) end def remove_socket(topic, socket) do - GenServer.call(__MODULE__, {:remove, socket, topic}) + do_remove_socket(@env, topic, socket) end def get_sockets do @@ -29,7 +31,7 @@ defmodule Pleroma.Web.Streamer.State do {:reply, state, state} end - def handle_call({:add, socket, topic}, _from, %{sockets: sockets} = state) do + def handle_call({:add, topic, socket}, _from, %{sockets: sockets} = state) do internal_topic = internal_topic(topic, socket) stream_socket = StreamerSocket.from_socket(socket) @@ -44,7 +46,7 @@ defmodule Pleroma.Web.Streamer.State do {:reply, state, state} end - def handle_call({:remove, socket, topic}, _from, %{sockets: sockets} = state) do + def handle_call({:remove, topic, socket}, _from, %{sockets: sockets} = state) do internal_topic = internal_topic(topic, socket) stream_socket = StreamerSocket.from_socket(socket) @@ -57,6 +59,14 @@ defmodule Pleroma.Web.Streamer.State do {:reply, state, state} end + defp do_remove_socket(:test, _, _) do + :ok + end + + defp do_remove_socket(_env, topic, socket) do + GenServer.call(__MODULE__, {:remove, topic, socket}) + end + defp internal_topic(topic, socket) when topic in ~w[user user:notification direct] do "#{topic}:#{socket.assigns[:user].id}" -- cgit v1.2.3 From 6193157f1998b10ac6cb9f4d36dd863eced81b37 Mon Sep 17 00:00:00 2001 From: Steven Fuchs Date: Tue, 17 Sep 2019 18:12:27 +0000 Subject: Fix notification warnings --- lib/pleroma/workers/web_pusher_worker.ex | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) (limited to 'lib') diff --git a/lib/pleroma/workers/web_pusher_worker.ex b/lib/pleroma/workers/web_pusher_worker.ex index bea2baffb..61b451e3e 100644 --- a/lib/pleroma/workers/web_pusher_worker.ex +++ b/lib/pleroma/workers/web_pusher_worker.ex @@ -10,7 +10,11 @@ defmodule Pleroma.Workers.WebPusherWorker do @impl Oban.Worker def perform(%{"op" => "web_push", "notification_id" => notification_id}, _job) do - notification = Repo.get(Notification, notification_id) + notification = + Notification + |> Repo.get(notification_id) + |> Repo.preload([:activity]) + Pleroma.Web.Push.Impl.perform(notification) end end -- cgit v1.2.3