alias Pleroma.Activity.Ir.Topics
alias Pleroma.Config
alias Pleroma.Conversation
+ alias Pleroma.Conversation.Participation
alias Pleroma.Notification
alias Pleroma.Object
alias Pleroma.Object.Containment
alias Pleroma.User
alias Pleroma.Web.ActivityPub.MRF
alias Pleroma.Web.ActivityPub.Transmogrifier
+ alias Pleroma.Web.ActivityPub.Utils
alias Pleroma.Web.Streamer
alias Pleroma.Web.WebFinger
alias Pleroma.Workers.BackgroundWorker
{:ok, map} <- MRF.filter(map),
{recipients, _, _} = get_recipients(map),
{:fake, false, map, recipients} <- {:fake, fake, map, recipients},
- :ok <- Containment.contain_child(map),
+ {:containment, :ok} <- {:containment, Containment.contain_child(map)},
{:ok, map, object} <- insert_full_object(map) do
{:ok, activity} =
Repo.insert(%Activity{
Notification.create_notifications(activity)
- participations =
- activity
- |> Conversation.create_or_bump_for()
- |> get_participations()
-
+ conversation = create_or_bump_conversation(activity, map["actor"])
+ participations = get_participations(conversation)
stream_out(activity)
stream_out_participations(participations)
{:ok, activity}
end
end
- defp get_participations({:ok, %{participations: participations}}), do: participations
+ defp create_or_bump_conversation(activity, actor) do
+ with {:ok, conversation} <- Conversation.create_or_bump_for(activity),
+ %User{} = user <- User.get_cached_by_ap_id(actor),
+ Participation.mark_as_read(user, conversation) do
+ {:ok, conversation}
+ end
+ end
+
+ defp get_participations({:ok, conversation}) do
+ conversation
+ |> Repo.preload(:participations, force: true)
+ |> Map.get(:participations)
+ end
+
defp get_participations(_), do: []
def stream_out_participations(participations) do
# only accept false as false value
local = !(params[:local] == false)
published = params[:published]
+ quick_insert? = Pleroma.Config.get([:env]) == :benchmark
with create_data <-
make_create_data(
{:fake, false, activity} <- {:fake, fake, activity},
_ <- increase_replies_count_if_reply(create_data),
_ <- increase_poll_votes_if_vote(create_data),
+ {:quick_insert, false, activity} <- {:quick_insert, quick_insert?, activity},
# Changing note count prior to enqueuing federation task in order to avoid
# race conditions on updating user.info
{:ok, _actor} <- increase_note_count_if_public(actor, activity),
:ok <- maybe_federate(activity) do
{:ok, activity}
else
+ {:quick_insert, true, activity} ->
+ {:ok, activity}
+
{:fake, true, activity} ->
{:ok, activity}
end
end
- def accept(%{to: to, actor: actor, object: object} = params) do
- # only accept false as false value
- local = !(params[:local] == false)
+ def accept(params) do
+ accept_or_reject("Accept", params)
+ end
- with data <- %{"to" => to, "type" => "Accept", "actor" => actor.ap_id, "object" => object},
- {:ok, activity} <- insert(data, local),
- :ok <- maybe_federate(activity) do
- {:ok, activity}
- end
+ def reject(params) do
+ accept_or_reject("Reject", params)
end
- def reject(%{to: to, actor: actor, object: object} = params) do
- # only accept false as false value
- local = !(params[:local] == false)
+ def accept_or_reject(type, %{to: to, actor: actor, object: object} = params) do
+ local = Map.get(params, :local, true)
+ activity_id = Map.get(params, :activity_id, nil)
- with data <- %{"to" => to, "type" => "Reject", "actor" => actor.ap_id, "object" => object},
+ with data <-
+ %{"to" => to, "type" => type, "actor" => actor.ap_id, "object" => object}
+ |> Utils.maybe_put("id", activity_id),
{:ok, activity} <- insert(data, local),
:ok <- maybe_federate(activity) do
{:ok, activity}
end
def update(%{to: to, cc: cc, actor: actor, object: object} = params) do
- # only accept false as false value
local = !(params[:local] == false)
+ activity_id = params[:activity_id]
with data <- %{
"to" => to,
"actor" => actor,
"object" => object
},
+ data <- Utils.maybe_put(data, "id", activity_id),
{:ok, activity} <- insert(data, local),
:ok <- maybe_federate(activity) do
{:ok, activity}
local \\ true,
public \\ true
) do
- with true <- is_public?(object),
+ with true <- is_announceable?(object, user, public),
announce_data <- make_announce_data(user, object, activity_id, public),
{:ok, activity} <- insert(announce_data, local),
{:ok, object} <- add_announce_to_object(activity, object),
end
end
- def delete(%Object{data: %{"id" => id, "actor" => actor}} = object, local \\ true) do
+ def delete(%Object{data: %{"id" => id, "actor" => actor}} = object, options \\ []) do
+ local = Keyword.get(options, :local, true)
+ activity_id = Keyword.get(options, :activity_id, nil)
+ actor = Keyword.get(options, :actor, actor)
+
user = User.get_cached_by_ap_id(actor)
to = (object.data["to"] || []) ++ (object.data["cc"] || [])
with {:ok, object, activity} <- Object.delete(object),
- data <- %{
- "type" => "Delete",
- "actor" => actor,
- "object" => id,
- "to" => to,
- "deleted_activity_id" => activity && activity.id
- },
+ data <-
+ %{
+ "type" => "Delete",
+ "actor" => actor,
+ "object" => id,
+ "to" => to,
+ "deleted_activity_id" => activity && activity.id
+ }
+ |> maybe_put("id", activity_id),
{:ok, activity} <- insert(data, local, false),
stream_out_participations(object, user),
_ <- decrease_replies_count_if_reply(object),
defp restrict_visibility(query, _visibility), do: query
+ defp exclude_visibility(query, %{"exclude_visibilities" => visibility})
+ when is_list(visibility) do
+ if Enum.all?(visibility, &(&1 in @valid_visibilities)) do
+ from(
+ a in query,
+ where:
+ not fragment(
+ "activity_visibility(?, ?, ?) = ANY (?)",
+ a.actor,
+ a.recipients,
+ a.data,
+ ^visibility
+ )
+ )
+ else
+ Logger.error("Could not exclude visibility to #{visibility}")
+ query
+ end
+ end
+
+ defp exclude_visibility(query, %{"exclude_visibilities" => visibility})
+ when visibility in @valid_visibilities do
+ from(
+ a in query,
+ where:
+ not fragment(
+ "activity_visibility(?, ?, ?) = ?",
+ a.actor,
+ a.recipients,
+ a.data,
+ ^visibility
+ )
+ )
+ end
+
+ defp exclude_visibility(query, %{"exclude_visibilities" => visibility})
+ when visibility not in @valid_visibilities do
+ Logger.error("Could not exclude visibility to #{visibility}")
+ query
+ end
+
+ defp exclude_visibility(query, _visibility), do: query
+
defp restrict_thread_visibility(query, _, %{skip_thread_containment: true} = _),
do: query
defp restrict_replies(query, %{"exclude_replies" => val}) when val == "true" or val == "1" do
from(
- activity in query,
- where: fragment("?->'object'->>'inReplyTo' is null", activity.data)
+ [_activity, object] in query,
+ where: fragment("?->>'inReplyTo' is null", object.data)
)
end
|> restrict_muted_reblogs(opts)
|> Activity.restrict_deactivated_users()
|> exclude_poll_votes(opts)
+ |> exclude_visibility(opts)
end
def fetch_activities(recipients, opts \\ %{}, pagination \\ :keyset) do
data <- maybe_update_follow_information(data) do
{:ok, data}
else
- e -> Logger.error("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
+ e ->
+ Logger.error("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
+ {:error, e}
end
end
projects / akkoma / blobdiff