hj/pleroma
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Merge branch 'develop' into issue/733

Browse source
This commit is contained in:
Maksim Pechnikov 2019-09-17 21:20:08 +03:00
commit 7dd51652f9
46 changed files with 1603 additions and 603 deletions
Download patch file Download diff file Expand all files Collapse all files

63
lib/pleroma/activity/ir/topics.ex Normal file
View file

@ -0,0 +1,63 @@
# Pleroma: A lightweight social networking server
# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
# 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

55
lib/pleroma/application.ex
View file

@ -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?()) ++
@ -141,7 +127,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),
@ -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

51
lib/pleroma/delivery.ex Normal file
View file

@ -0,0 +1,51 @@
# Pleroma: A lightweight social networking server
# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
# SPDX-License-Identifier: AGPL-3.0-only
defmodule Pleroma.Delivery do
use Ecto.Schema
alias Pleroma.Delivery
alias Pleroma.FlakeId
alias Pleroma.Object
alias Pleroma.Repo
alias Pleroma.User
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])
|> 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)
end
def create(object_id, user_id) do
%Delivery{}
|> changeset(%{user_id: user_id, object_id: object_id})
|> Repo.insert(on_conflict: :nothing)
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
# 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()
end
end

2
lib/pleroma/flake_id.ex
View file

@ -14,7 +14,7 @@ defmodule Pleroma.FlakeId do
@type t :: binary
@behaviour Ecto.Type
use Ecto.Type
use GenServer
require Logger
alias __MODULE__

6
lib/pleroma/notification.ex
View file

@ -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

18
lib/pleroma/plugs/cache.ex
View file

@ -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,11 @@ 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)
@ -88,9 +94,17 @@ 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}
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, key, {content_type, body, tracking_fun_data}, ttl: ttl)
opts.tracking_fun.(conn, tracking_fun_data)
end
put_resp_header(conn, "x-cache", "MISS from Pleroma")

3
lib/pleroma/plugs/http_signature.ex
View file

@ -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

15
lib/pleroma/user.ex
View file

@ -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
@ -62,6 +63,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()
@ -148,6 +150,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,
@ -1640,6 +1643,18 @@ 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),
where: delivery.object_id == ^object_id
)
|> Repo.all()
end
def change_email(user, email) do
user
|> cast(%{email: email}, [:email])

60
lib/pleroma/web/activity_pub/activity_pub.ex
View file

@ -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)
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
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) do
:noop
end
def create(%{to: to, actor: actor, context: context, object: object} = params, fake \\ false) do
@ -436,6 +410,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])
@ -464,10 +439,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
@ -479,14 +455,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]})

27
lib/pleroma/web/activity_pub/activity_pub_controller.ex
View file

@ -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,12 @@ 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: &__MODULE__.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 +60,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 +71,16 @@ 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)
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 +116,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 +127,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: conn
defp set_cache_ttl_for(conn, %Activity{object: object}) do
set_cache_ttl_for(conn, object)
end

15
lib/pleroma/web/activity_pub/publisher.ex
View file

@ -5,8 +5,10 @@
defmodule Pleroma.Web.ActivityPub.Publisher do
alias Pleroma.Activity
alias Pleroma.Config
alias Pleroma.Delivery
alias Pleroma.HTTP
alias Pleroma.Instances
alias Pleroma.Object
alias Pleroma.User
alias Pleroma.Web.ActivityPub.Relay
alias Pleroma.Web.ActivityPub.Transmogrifier
@ -116,7 +118,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

2
lib/pleroma/web/activity_pub/transmogrifier.ex
View file

@ -1064,7 +1064,7 @@ defmodule Pleroma.Web.ActivityPub.Transmogrifier do
defp upgrade_user(user, data) do
user
|> User.upgrade_changeset(data)
|> User.upgrade_changeset(data, true)
|> User.update_and_set_cache()
end

217
lib/pleroma/web/activity_pub/utils.ex
View file

@ -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
addresses = [params["to"], params["cc"], params["bto"], params["bcc"]]
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
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
@ -179,53 +169,58 @@ 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 =
if not 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
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
if is_map(map["object"]) do
object = lazy_put_object_defaults(map["object"], map, fake?)
%{map | "object" => object}
else
def lazy_put_activity_defaults(map, _fake?) do
%{data: %{"id" => context}, id: context_id} = create_context(map["context"])
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)
|> lazy_put_object_defaults(false)
end
# 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
end
|> 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
@doc """
Adds an id and published date if they aren't there.
"""
def lazy_put_object_defaults(map, activity \\ %{}, fake?)
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"])
def lazy_put_object_defaults(map, activity, true = _fake?) do
map
|> 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"])
%{activity | "object" => object}
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"])
end
defp lazy_put_object_defaults(activity, _), do: activity
@doc """
Inserts a full object if it is contained in an activity.
@ -345,24 +340,24 @@ 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]
)
"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}
rescue
e ->
{:error, e}
end
User.set_follow_state_cache(actor, object, state)
activity = Activity.get_by_id(activity.id)
{:ok, activity}
end
def update_follow_state(
@ -413,6 +408,7 @@ 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" => ap_id}}) do
"Announce"
|> Activity.Queries.by_type()
@ -495,33 +491,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
Enum.uniq([actor | object.data["announcements"]])
else
[actor]
end
announcements = take_announcements(object)
update_element_in_object("announcement", announcements, object)
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(take_announcements(object), actor) do
update_element_in_object("announcement", announcements, object)
end
end
defp take_announcements(%{data: %{"announcements" => announcements}} = _)
when is_list(announcements),
do: announcements
defp take_announcements(_), do: []
#### Unfollow-related helpers
def make_unfollow_data(follower, followed, follow_activity, activity_id) do
@ -535,6 +533,7 @@ 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
"Block"
|> Activity.Queries.by_type()
@ -583,28 +582,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.

7
lib/pleroma/web/mastodon_api/websocket_handler.ex
View file

@ -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

6
lib/pleroma/web/rich_media/parser.ex
View file

@ -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

2
lib/pleroma/web/router.ex
View file

@ -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
@ -514,6 +515,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)

318
lib/pleroma/web/streamer.ex
View file

@ -1,318 +0,0 @@
# Pleroma: A lightweight social networking server
# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
# SPDX-License-Identifier: AGPL-3.0-only
defmodule Pleroma.Web.Streamer do
use GenServer
require Logger
alias Pleroma.Activity
alias Pleroma.Config
alias Pleroma.Conversation.Participation
alias Pleroma.Notification
alias Pleroma.Object
alias Pleroma.User
alias Pleroma.Web.ActivityPub.ActivityPub
alias Pleroma.Web.ActivityPub.Visibility
alias Pleroma.Web.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

33
lib/pleroma/web/streamer/ping.ex Normal file
View file

@ -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

78
lib/pleroma/web/streamer/state.ex Normal file
View file

@ -0,0 +1,78 @@
defmodule Pleroma.Web.Streamer.State do
use GenServer
require Logger
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, topic, socket})
end
def remove_socket(topic, socket) do
do_remove_socket(@env, topic, socket)
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, topic, socket}, _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, topic, socket}, _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 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}"
end
defp internal_topic(topic, _) do
topic
end
end

55
lib/pleroma/web/streamer/streamer.ex Normal file
View file

@ -0,0 +1,55 @@
# Pleroma: A lightweight social networking server
# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
# SPDX-License-Identifier: AGPL-3.0-only
defmodule Pleroma.Web.Streamer do
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

31
lib/pleroma/web/streamer/streamer_socket.ex Normal file
View file

@ -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

33
lib/pleroma/web/streamer/supervisor.ex Normal file
View file

@ -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

220
lib/pleroma/web/streamer/worker.ex Normal file
View file

@ -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

66
lib/pleroma/web/views/streamer_view.ex Normal file
View file

@ -0,0 +1,66 @@
# Pleroma: A lightweight social networking server
# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
# 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

6
lib/pleroma/workers/web_pusher_worker.ex
View file

@ -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