alias Comeonin.Pbkdf2
alias Ecto.Multi
alias Pleroma.Activity
+ alias Pleroma.Conversation.Participation
+ alias Pleroma.Delivery
alias Pleroma.Keys
alias Pleroma.Notification
alias Pleroma.Object
alias Pleroma.Web.OStatus
alias Pleroma.Web.RelMe
alias Pleroma.Web.Websub
+ alias Pleroma.Workers.BackgroundWorker
require Logger
@type t :: %__MODULE__{}
- @primary_key {:id, Pleroma.FlakeId, autogenerate: true}
+ @primary_key {:id, FlakeId.Ecto.CompatType, autogenerate: true}
# credo:disable-for-next-line Credo.Check.Readability.MaxLineLength
@email_regex ~r/^[a-zA-Z0-9.!#$%&'*+\/=?^_`{|}~-]+@[a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?(?:\.[a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?)*$/
field(:password_hash, :string)
field(:password, :string, virtual: true)
field(:password_confirmation, :string, virtual: true)
+ field(:keys, :string)
field(:following, {:array, :string}, default: [])
field(:ap_id, :string)
field(:avatar, :map)
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()
def profile_url(%User{ap_id: ap_id}), do: ap_id
def profile_url(_), do: nil
- def ap_id(%User{nickname: nickname}) do
- "#{Web.base_url()}/users/#{nickname}"
- end
+ def ap_id(%User{nickname: nickname}), do: "#{Web.base_url()}/users/#{nickname}"
def ap_followers(%User{follower_address: fa}) when is_binary(fa), do: fa
def ap_followers(%User{} = user), do: "#{ap_id(user)}/followers"
def user_info(%User{} = user, args \\ %{}) do
following_count =
- if args[:following_count],
- do: args[:following_count],
- else: user.info.following_count || following_count(user)
+ Map.get(args, :following_count, user.info.following_count || following_count(user))
- follower_count =
- if args[:follower_count], do: args[:follower_count], else: user.info.follower_count
+ follower_count = Map.get(args, :follower_count, user.info.follower_count)
%{
note_count: user.info.note_count,
end
def follow_state(%User{} = user, %User{} = target) do
- follow_activity = Utils.fetch_latest_follow(user, target)
-
- if follow_activity,
- do: follow_activity.data["state"],
+ case Utils.fetch_latest_follow(user, target) do
+ %{data: %{"state" => state}} -> state
# Ideally this would be nil, but then Cachex does not commit the value
- else: false
+ _ -> false
+ end
end
def get_cached_follow_state(user, target) 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,
- "follow_state:#{user_ap_id}|#{target_ap_id}",
- state
- )
+ Cachex.put(:user_cache, "follow_state:#{user_ap_id}|#{target_ap_id}", state)
end
def set_info_cache(user, args) do
|> Repo.aggregate(:count, :id)
end
+ defp truncate_if_exists(params, key, max_length) do
+ if Map.has_key?(params, key) and is_binary(params[key]) do
+ {value, _chopped} = String.split_at(params[key], max_length)
+ Map.put(params, key, value)
+ else
+ params
+ end
+ end
+
def remote_user_creation(params) do
bio_limit = Pleroma.Config.get([:instance, :user_bio_length], 5000)
name_limit = Pleroma.Config.get([:instance, :user_name_length], 100)
- params = Map.put(params, :info, params[:info] || %{})
- info_cng = User.Info.remote_user_creation(%User.Info{}, params[:info])
+ params =
+ params
+ |> Map.put(:info, params[:info] || %{})
+ |> truncate_if_exists(:name, name_limit)
+ |> truncate_if_exists(:bio, bio_limit)
- changes =
- %User{}
+ changeset =
+ %User{local: false}
|> cast(params, [:bio, :name, :ap_id, :nickname, :avatar])
|> validate_required([:name, :ap_id])
|> unique_constraint(:nickname)
|> validate_format(:nickname, @email_regex)
|> validate_length(:bio, max: bio_limit)
|> validate_length(:name, max: name_limit)
- |> put_change(:local, false)
- |> put_embed(:info, info_cng)
-
- if changes.valid? do
- case info_cng.changes[:source_data] do
- %{"followers" => followers, "following" => following} ->
- changes
- |> put_change(:follower_address, followers)
- |> put_change(:following_address, following)
+ |> change_info(&User.Info.remote_user_creation(&1, params[:info]))
- _ ->
- followers = User.ap_followers(%User{nickname: changes.changes[:nickname]})
+ case params[:info][:source_data] do
+ %{"followers" => followers, "following" => following} ->
+ changeset
+ |> put_change(:follower_address, followers)
+ |> put_change(:following_address, following)
- changes
- |> put_change(:follower_address, followers)
- end
- else
- changes
+ _ ->
+ followers = ap_followers(%User{nickname: get_field(changeset, :nickname)})
+ put_change(changeset, :follower_address, followers)
end
end
name_limit = Pleroma.Config.get([:instance, :user_name_length], 100)
params = Map.put(params, :last_refreshed_at, NaiveDateTime.utc_now())
- info_cng = User.Info.user_upgrade(struct.info, params[:info], remote?)
struct
|> cast(params, [
|> validate_format(:nickname, local_nickname_regex())
|> validate_length(:bio, max: bio_limit)
|> validate_length(:name, max: name_limit)
- |> put_embed(:info, info_cng)
+ |> change_info(&User.Info.user_upgrade(&1, params[:info], remote?))
end
def password_update_changeset(struct, params) do
|> validate_required([:password, :password_confirmation])
|> validate_confirmation(:password)
|> put_password_hash
+ |> put_embed(:info, User.Info.set_password_reset_pending(struct.info, false))
end
@spec reset_password(User.t(), map) :: {:ok, User.t()} | {:error, Ecto.Changeset.t()}
end
end
+ def force_password_reset_async(user) do
+ BackgroundWorker.enqueue("force_password_reset", %{"user_id" => user.id})
+ end
+
+ @spec force_password_reset(User.t()) :: {:ok, User.t()} | {:error, Ecto.Changeset.t()}
+ def force_password_reset(user) do
+ info_cng = User.Info.set_password_reset_pending(user.info, true)
+
+ user
+ |> change()
+ |> put_embed(:info, info_cng)
+ |> update_and_set_cache()
+ end
+
def register_changeset(struct, params \\ %{}, opts \\ []) do
bio_limit = Pleroma.Config.get([:instance, :user_bio_length], 5000)
name_limit = Pleroma.Config.get([:instance, :user_name_length], 100)
opts[:need_confirmation]
end
- info_change =
- User.Info.confirmation_changeset(%User.Info{}, need_confirmation: need_confirmation?)
+ struct
+ |> cast(params, [:bio, :email, :name, :nickname, :password, :password_confirmation])
+ |> validate_required([:name, :nickname, :password, :password_confirmation])
+ |> validate_confirmation(:password)
+ |> unique_constraint(:email)
+ |> unique_constraint(:nickname)
+ |> validate_exclusion(:nickname, Pleroma.Config.get([User, :restricted_nicknames]))
+ |> validate_format(:nickname, local_nickname_regex())
+ |> validate_format(:email, @email_regex)
+ |> validate_length(:bio, max: bio_limit)
+ |> validate_length(:name, min: 1, max: name_limit)
+ |> change_info(&User.Info.confirmation_changeset(&1, need_confirmation: need_confirmation?))
+ |> maybe_validate_required_email(opts[:external])
+ |> put_password_hash
+ |> put_ap_id()
+ |> unique_constraint(:ap_id)
+ |> put_following_and_follower_address()
+ end
- changeset =
- struct
- |> cast(params, [:bio, :email, :name, :nickname, :password, :password_confirmation])
- |> validate_required([:name, :nickname, :password, :password_confirmation])
- |> validate_confirmation(:password)
- |> unique_constraint(:email)
- |> unique_constraint(:nickname)
- |> validate_exclusion(:nickname, Pleroma.Config.get([User, :restricted_nicknames]))
- |> validate_format(:nickname, local_nickname_regex())
- |> validate_format(:email, @email_regex)
- |> validate_length(:bio, max: bio_limit)
- |> validate_length(:name, min: 1, max: name_limit)
- |> put_change(:info, info_change)
+ def maybe_validate_required_email(changeset, true), do: changeset
+ def maybe_validate_required_email(changeset, _), do: validate_required(changeset, [:email])
- changeset =
- if opts[:external] do
- changeset
- else
- validate_required(changeset, [:email])
- end
+ defp put_ap_id(changeset) do
+ ap_id = ap_id(%User{nickname: get_field(changeset, :nickname)})
+ put_change(changeset, :ap_id, ap_id)
+ end
- if changeset.valid? do
- ap_id = User.ap_id(%User{nickname: changeset.changes[:nickname]})
- followers = User.ap_followers(%User{nickname: changeset.changes[:nickname]})
+ defp put_following_and_follower_address(changeset) do
+ followers = ap_followers(%User{nickname: get_field(changeset, :nickname)})
- changeset
- |> put_password_hash
- |> put_change(:ap_id, ap_id)
- |> unique_constraint(:ap_id)
- |> put_change(:following, [followers])
- |> put_change(:follower_address, followers)
- else
- changeset
- end
+ changeset
+ |> put_change(:following, [followers])
+ |> put_change(:follower_address, followers)
end
defp autofollow_users(user) do
@doc "Inserts provided changeset, performs post-registration actions (confirmation email sending etc.)"
def register(%Ecto.Changeset{} = changeset) do
- with {:ok, user} <- Repo.insert(changeset),
- {:ok, user} <- post_register_action(user) do
- {:ok, user}
+ with {:ok, user} <- Repo.insert(changeset) do
+ post_register_action(user)
end
end
end
def maybe_direct_follow(%User{} = follower, %User{} = followed) do
- if not User.ap_enabled?(followed) do
+ if not ap_enabled?(followed) do
follow(follower, followed)
else
{:ok, follower}
{1, [follower]} = Repo.update_all(q, [])
- Enum.each(followeds, fn followed ->
- update_follower_count(followed)
- end)
+ Enum.each(followeds, &update_follower_count/1)
set_cache(follower)
end
{:error, "Could not follow user: #{followed.nickname} blocked you."}
true ->
- if !followed.local && follower.local && !ap_enabled?(followed) do
+ benchmark? = Pleroma.Config.get([:env]) == :benchmark
+
+ if !followed.local && follower.local && !ap_enabled?(followed) && !benchmark? do
Websub.subscribe(follower, followed)
end
|> Repo.all()
end
+ def get_all_by_ids(ids) do
+ from(u in __MODULE__, where: u.id in ^ids)
+ |> Repo.all()
+ end
+
# This is mostly an SPC migration fix. This guesses the user nickname by taking the last part
# of the ap_id and the domain and tries to get that user
def get_by_guessed_nickname(ap_id) do
def update_and_set_cache(changeset) do
with {:ok, user} <- Repo.update(changeset, stale_error_field: :id) do
set_cache(user)
- else
- e -> e
end
end
key = "nickname:#{nickname}"
Cachex.fetch!(:user_cache, key, fn ->
- user_result = get_or_fetch_by_nickname(nickname)
-
- case user_result do
+ case get_or_fetch_by_nickname(nickname) do
{:ok, user} -> {:commit, user}
{:error, _error} -> {:ignore, nil}
end
restrict_to_local = Pleroma.Config.get([:instance, :limit_to_local_content])
cond do
- is_integer(nickname_or_id) or Pleroma.FlakeId.is_flake_id?(nickname_or_id) ->
+ is_integer(nickname_or_id) or FlakeId.flake_id?(nickname_or_id) ->
get_cached_by_id(nickname_or_id) || get_cached_by_nickname(nickname_or_id)
- restrict_to_local == false ->
+ restrict_to_local == false or not String.contains?(nickname_or_id, "@") ->
get_cached_by_nickname(nickname_or_id)
restrict_to_local == :unauthenticated and match?(%User{}, opts[:for]) ->
def get_cached_user_info(user) do
key = "user_info:#{user.id}"
- Cachex.fetch!(:user_cache, key, fn _ -> user_info(user) end)
+ Cachex.fetch!(:user_cache, key, fn -> user_info(user) end)
end
def fetch_by_nickname(nickname) do
- ap_try = ActivityPub.make_user_from_nickname(nickname)
-
- case ap_try do
+ case ActivityPub.make_user_from_nickname(nickname) do
{:ok, user} -> {:ok, user}
_ -> OStatus.make_user(nickname)
end
end
@doc "Fetch some posts when the user has just been federated with"
- def fetch_initial_posts(user),
- do: PleromaJobQueue.enqueue(:background, __MODULE__, [:fetch_initial_posts, user])
+ def fetch_initial_posts(user) do
+ BackgroundWorker.enqueue("fetch_initial_posts", %{"user_id" => user.id})
+ end
@spec get_followers_query(User.t(), pos_integer() | nil) :: Ecto.Query.t()
def get_followers_query(%User{} = user, nil) do
end
def get_followers_query(user, page) do
- from(u in get_followers_query(user, nil))
+ user
+ |> get_followers_query(nil)
|> User.Query.paginate(page, 20)
end
@spec get_followers(User.t(), pos_integer()) :: {:ok, list(User.t())}
def get_followers(user, page \\ nil) do
- q = get_followers_query(user, page)
-
- {:ok, Repo.all(q)}
+ user
+ |> get_followers_query(page)
+ |> Repo.all()
end
@spec get_external_followers(User.t(), pos_integer()) :: {:ok, list(User.t())}
def get_external_followers(user, page \\ nil) do
- q =
- user
- |> get_followers_query(page)
- |> User.Query.build(%{external: true})
-
- {:ok, Repo.all(q)}
+ user
+ |> get_followers_query(page)
+ |> User.Query.build(%{external: true})
+ |> Repo.all()
end
def get_followers_ids(user, page \\ nil) do
- q = get_followers_query(user, page)
-
- Repo.all(from(u in q, select: u.id))
+ user
+ |> get_followers_query(page)
+ |> select([u], u.id)
+ |> Repo.all()
end
@spec get_friends_query(User.t(), pos_integer() | nil) :: Ecto.Query.t()
end
def get_friends_query(user, page) do
- from(u in get_friends_query(user, nil))
+ user
+ |> get_friends_query(nil)
|> User.Query.paginate(page, 20)
end
def get_friends_query(user), do: get_friends_query(user, nil)
def get_friends(user, page \\ nil) do
- q = get_friends_query(user, page)
-
- {:ok, Repo.all(q)}
+ user
+ |> get_friends_query(page)
+ |> Repo.all()
end
def get_friends_ids(user, page \\ nil) do
- q = get_friends_query(user, page)
-
- Repo.all(from(u in q, select: u.id))
+ user
+ |> get_friends_query(page)
+ |> select([u], u.id)
+ |> Repo.all()
end
@spec get_follow_requests(User.t()) :: {:ok, [User.t()]}
def get_follow_requests(%User{} = user) do
- users =
- Activity.follow_requests_for_actor(user)
- |> join(:inner, [a], u in User, on: a.actor == u.ap_id)
- |> where([a, u], not fragment("? @> ?", u.following, ^[user.follower_address]))
- |> group_by([a, u], u.id)
- |> select([a, u], u)
- |> Repo.all()
-
- {:ok, users}
+ user
+ |> Activity.follow_requests_for_actor()
+ |> join(:inner, [a], u in User, on: a.actor == u.ap_id)
+ |> where([a, u], not fragment("? @> ?", u.following, ^[user.follower_address]))
+ |> group_by([a, u], u.id)
+ |> select([a, u], u)
+ |> Repo.all()
end
def increase_note_count(%User{} = user) do
end
def update_note_count(%User{} = user) do
- note_count_query =
+ note_count =
from(
a in Object,
where: fragment("?->>'actor' = ? and ?->>'type' = 'Note'", a.data, ^user.ap_id, a.data),
select: count(a.id)
)
+ |> Repo.one()
- note_count = Repo.one(note_count_query)
+ update_info(user, &User.Info.set_note_count(&1, note_count))
+ end
- info_cng = User.Info.set_note_count(user.info, note_count)
+ def update_mascot(user, url) do
+ info_changeset =
+ User.Info.mascot_update(
+ user.info,
+ url
+ )
user
|> change()
- |> put_embed(:info, info_cng)
+ |> put_embed(:info, info_changeset)
|> update_and_set_cache()
end
def fetch_follow_information(user) do
with {:ok, info} <- ActivityPub.fetch_follow_information_for_user(user) do
- info_cng = User.Info.follow_information_update(user.info, info)
-
- changeset =
- user
- |> change()
- |> put_embed(:info, info_cng)
-
- update_and_set_cache(changeset)
- else
- {:error, _} = e -> e
- e -> {:error, e}
+ update_info(user, &User.Info.follow_information_update(&1, info))
end
end
def maybe_update_following_count(user), do: user
+ def set_unread_conversation_count(%User{local: true} = user) do
+ unread_query = Participation.unread_conversation_count_for_user(user)
+
+ User
+ |> join(:inner, [u], p in subquery(unread_query))
+ |> update([u, p],
+ set: [
+ info:
+ fragment(
+ "jsonb_set(?, '{unread_conversation_count}', ?::varchar::jsonb, true)",
+ u.info,
+ p.count
+ )
+ ]
+ )
+ |> where([u], u.id == ^user.id)
+ |> select([u], u)
+ |> Repo.update_all([])
+ |> case do
+ {1, [user]} -> set_cache(user)
+ _ -> {:error, user}
+ end
+ end
+
+ def set_unread_conversation_count(_), do: :noop
+
+ def increment_unread_conversation_count(conversation, %User{local: true} = user) do
+ unread_query =
+ Participation.unread_conversation_count_for_user(user)
+ |> where([p], p.conversation_id == ^conversation.id)
+
+ User
+ |> join(:inner, [u], p in subquery(unread_query))
+ |> update([u, p],
+ set: [
+ info:
+ fragment(
+ "jsonb_set(?, '{unread_conversation_count}', (coalesce((?->>'unread_conversation_count')::int, 0) + 1)::varchar::jsonb, true)",
+ u.info,
+ u.info
+ )
+ ]
+ )
+ |> where([u], u.id == ^user.id)
+ |> where([u, p], p.count == 0)
+ |> select([u], u)
+ |> Repo.update_all([])
+ |> case do
+ {1, [user]} -> set_cache(user)
+ _ -> {:error, user}
+ end
+ end
+
+ def increment_unread_conversation_count(_, _), do: :noop
+
def remove_duplicated_following(%User{following: following} = user) do
uniq_following = Enum.uniq(following)
@spec mute(User.t(), User.t(), boolean()) :: {:ok, User.t()} | {:error, String.t()}
def mute(muter, %User{ap_id: ap_id}, notifications? \\ true) do
- info = muter.info
-
- info_cng =
- User.Info.add_to_mutes(info, ap_id)
- |> User.Info.add_to_muted_notifications(info, ap_id, notifications?)
-
- cng =
- change(muter)
- |> put_embed(:info, info_cng)
-
- update_and_set_cache(cng)
+ update_info(muter, &User.Info.add_to_mutes(&1, ap_id, notifications?))
end
def unmute(muter, %{ap_id: ap_id}) do
- info = muter.info
-
- info_cng =
- User.Info.remove_from_mutes(info, ap_id)
- |> User.Info.remove_from_muted_notifications(info, ap_id)
-
- cng =
- change(muter)
- |> put_embed(:info, info_cng)
-
- update_and_set_cache(cng)
+ update_info(muter, &User.Info.remove_from_mutes(&1, ap_id))
end
def subscribe(subscriber, %{ap_id: ap_id}) do
- deny_follow_blocked = Pleroma.Config.get([:user, :deny_follow_blocked])
-
with %User{} = subscribed <- get_cached_by_ap_id(ap_id) do
- blocked = blocks?(subscribed, subscriber) and deny_follow_blocked
+ deny_follow_blocked = Pleroma.Config.get([:user, :deny_follow_blocked])
- if blocked do
+ if blocks?(subscribed, subscriber) and deny_follow_blocked do
{:error, "Could not subscribe: #{subscribed.nickname} is blocking you"}
else
- info_cng =
- subscribed.info
- |> User.Info.add_to_subscribers(subscriber.ap_id)
-
- change(subscribed)
- |> put_embed(:info, info_cng)
- |> update_and_set_cache()
+ update_info(subscribed, &User.Info.add_to_subscribers(&1, subscriber.ap_id))
end
end
end
def unsubscribe(unsubscriber, %{ap_id: ap_id}) do
with %User{} = user <- get_cached_by_ap_id(ap_id) do
- info_cng =
- user.info
- |> User.Info.remove_from_subscribers(unsubscriber.ap_id)
-
- change(user)
- |> put_embed(:info, info_cng)
- |> update_and_set_cache()
+ update_info(user, &User.Info.remove_from_subscribers(&1, unsubscriber.ap_id))
end
end
blocker
end
- if following?(blocked, blocker) do
- unfollow(blocked, blocker)
- end
+ if following?(blocked, blocker), do: unfollow(blocked, blocker)
{:ok, blocker} = update_follower_count(blocker)
- info_cng =
- blocker.info
- |> User.Info.add_to_block(ap_id)
-
- cng =
- change(blocker)
- |> put_embed(:info, info_cng)
-
- update_and_set_cache(cng)
+ update_info(blocker, &User.Info.add_to_block(&1, ap_id))
end
# helper to handle the block given only an actor's AP id
end
def unblock(blocker, %{ap_id: ap_id}) do
- info_cng =
- blocker.info
- |> User.Info.remove_from_block(ap_id)
-
- cng =
- change(blocker)
- |> put_embed(:info, info_cng)
-
- update_and_set_cache(cng)
+ update_info(blocker, &User.Info.remove_from_block(&1, ap_id))
end
def mutes?(nil, _), do: false
end
def block_domain(user, domain) do
- info_cng =
- user.info
- |> User.Info.add_to_domain_block(domain)
-
- cng =
- change(user)
- |> put_embed(:info, info_cng)
-
- update_and_set_cache(cng)
+ update_info(user, &User.Info.add_to_domain_block(&1, domain))
end
def unblock_domain(user, domain) do
- info_cng =
- user.info
- |> User.Info.remove_from_domain_block(domain)
-
- cng =
- change(user)
- |> put_embed(:info, info_cng)
-
- update_and_set_cache(cng)
+ update_info(user, &User.Info.remove_from_domain_block(&1, domain))
end
def deactivate_async(user, status \\ true) do
- PleromaJobQueue.enqueue(:background, __MODULE__, [:deactivate_async, user, status])
+ BackgroundWorker.enqueue("deactivate_user", %{"user_id" => user.id, "status" => status})
end
def deactivate(%User{} = user, status \\ true) do
- info_cng = User.Info.set_activation_status(user.info, status)
-
- with {:ok, friends} <- User.get_friends(user),
- {:ok, followers} <- User.get_followers(user),
- {:ok, user} <-
- user
- |> change()
- |> put_embed(:info, info_cng)
- |> update_and_set_cache() do
- Enum.each(followers, &invalidate_cache(&1))
- Enum.each(friends, &update_follower_count(&1))
+ with {:ok, user} <- update_info(user, &User.Info.set_activation_status(&1, status)) do
+ Enum.each(get_followers(user), &invalidate_cache/1)
+ Enum.each(get_friends(user), &update_follower_count/1)
{:ok, user}
end
end
def update_notification_settings(%User{} = user, settings \\ %{}) do
- info_changeset = User.Info.update_notification_settings(user.info, settings)
+ update_info(user, &User.Info.update_notification_settings(&1, settings))
+ end
- change(user)
- |> put_embed(:info, info_changeset)
- |> update_and_set_cache()
+ def delete(%User{} = user) do
+ BackgroundWorker.enqueue("delete_user", %{"user_id" => user.id})
end
- @spec delete(User.t()) :: :ok
- def delete(%User{} = user),
- do: PleromaJobQueue.enqueue(:background, __MODULE__, [:delete, user])
+ def perform(:force_password_reset, user), do: force_password_reset(user)
@spec perform(atom(), User.t()) :: {:ok, User.t()}
def perform(:delete, %User{} = user) do
{:ok, _user} = ActivityPub.delete(user)
# Remove all relationships
- {:ok, followers} = User.get_followers(user)
-
- Enum.each(followers, fn follower ->
+ user
+ |> get_followers()
+ |> Enum.each(fn follower ->
ActivityPub.unfollow(follower, user)
- User.unfollow(follower, user)
+ unfollow(follower, user)
end)
- {:ok, friends} = User.get_friends(user)
-
- Enum.each(friends, fn followed ->
+ user
+ |> get_friends()
+ |> Enum.each(fn followed ->
ActivityPub.unfollow(user, followed)
- User.unfollow(user, followed)
+ unfollow(user, followed)
end)
delete_user_activities(user)
def perform(:fetch_initial_posts, %User{} = user) do
pages = Pleroma.Config.get!([:fetch_initial_posts, :pages])
- Enum.each(
- # Insert all the posts in reverse order, so they're in the right order on the timeline
- Enum.reverse(Utils.fetch_ordered_collection(user.info.source_data["outbox"], pages)),
- &Pleroma.Web.Federator.incoming_ap_doc/1
- )
-
- {:ok, user}
+ # Insert all the posts in reverse order, so they're in the right order on the timeline
+ user.info.source_data["outbox"]
+ |> Utils.fetch_ordered_collection(pages)
+ |> Enum.reverse()
+ |> Enum.each(&Pleroma.Web.Federator.incoming_ap_doc/1)
end
def perform(:deactivate_async, user, status), do: deactivate(user, status)
Repo.all(query)
end
- def blocks_import(%User{} = blocker, blocked_identifiers) when is_list(blocked_identifiers),
- do:
- PleromaJobQueue.enqueue(:background, __MODULE__, [
- :blocks_import,
- blocker,
- blocked_identifiers
- ])
+ def blocks_import(%User{} = blocker, blocked_identifiers) when is_list(blocked_identifiers) do
+ BackgroundWorker.enqueue("blocks_import", %{
+ "blocker_id" => blocker.id,
+ "blocked_identifiers" => blocked_identifiers
+ })
+ end
- def follow_import(%User{} = follower, followed_identifiers) when is_list(followed_identifiers),
- do:
- PleromaJobQueue.enqueue(:background, __MODULE__, [
- :follow_import,
- follower,
- followed_identifiers
- ])
+ def follow_import(%User{} = follower, followed_identifiers)
+ when is_list(followed_identifiers) do
+ BackgroundWorker.enqueue("follow_import", %{
+ "follower_id" => follower.id,
+ "followed_identifiers" => followed_identifiers
+ })
+ end
- def delete_user_activities(%User{ap_id: ap_id} = user) do
+ def delete_user_activities(%User{ap_id: ap_id}) do
ap_id
- |> Activity.query_by_actor()
+ |> Activity.Queries.by_actor()
|> RepoStreamer.chunk_stream(50)
- |> Stream.each(fn activities ->
- Enum.each(activities, &delete_activity(&1))
- end)
+ |> Stream.each(fn activities -> Enum.each(activities, &delete_activity/1) end)
|> Stream.run()
-
- {:ok, user}
end
defp delete_activity(%{data: %{"type" => "Create"}} = activity) do
end
defp delete_activity(%{data: %{"type" => "Like"}} = activity) do
- user = get_cached_by_ap_id(activity.actor)
object = Object.normalize(activity)
- ActivityPub.unlike(user, object)
+ activity.actor
+ |> get_cached_by_ap_id()
+ |> ActivityPub.unlike(object)
end
defp delete_activity(%{data: %{"type" => "Announce"}} = activity) do
- user = get_cached_by_ap_id(activity.actor)
object = Object.normalize(activity)
- ActivityPub.unannounce(user, object)
+ activity.actor
+ |> get_cached_by_ap_id()
+ |> ActivityPub.unannounce(object)
end
defp delete_activity(_activity), do: "Doing nothing"
def html_filter_policy(_), do: Pleroma.Config.get([:markup, :scrub_policy])
def fetch_by_ap_id(ap_id) do
- ap_try = ActivityPub.make_user_from_ap_id(ap_id)
-
- case ap_try do
+ case ActivityPub.make_user_from_ap_id(ap_id) do
{:ok, user} ->
{:ok, user}
def get_or_fetch_by_ap_id(ap_id) do
user = get_cached_by_ap_id(ap_id)
- if !is_nil(user) and !User.needs_update?(user) do
+ if !is_nil(user) and !needs_update?(user) do
{:ok, user}
else
# Whether to fetch initial posts for the user (if it's a new user & the fetching is enabled)
@doc "Creates an internal service actor by URI if missing. Optionally takes nickname for addressing."
def get_or_create_service_actor_by_ap_id(uri, nickname \\ nil) do
- if user = get_cached_by_ap_id(uri) do
+ with %User{} = user <- get_cached_by_ap_id(uri) do
user
else
- changes =
- %User{info: %User.Info{}}
- |> cast(%{}, [:ap_id, :nickname, :local])
- |> put_change(:ap_id, uri)
- |> put_change(:nickname, nickname)
- |> put_change(:local, true)
- |> put_change(:follower_address, uri <> "/followers")
-
- {:ok, user} = Repo.insert(changes)
- user
+ _ ->
+ {:ok, user} =
+ %User{info: %User.Info{}}
+ |> cast(%{}, [:ap_id, :nickname, :local])
+ |> put_change(:ap_id, uri)
+ |> put_change(:nickname, nickname)
+ |> put_change(:local, true)
+ |> put_change(:follower_address, uri <> "/followers")
+ |> Repo.insert()
+
+ user
end
end
# this is because we have synchronous follow APIs and need to simulate them
# with an async handshake
def wait_and_refresh(_, %User{local: true} = a, %User{local: true} = b) do
- with %User{} = a <- User.get_cached_by_id(a.id),
- %User{} = b <- User.get_cached_by_id(b.id) do
+ with %User{} = a <- get_cached_by_id(a.id),
+ %User{} = b <- get_cached_by_id(b.id) do
{:ok, a, b}
else
- _e ->
- :error
+ nil -> :error
end
end
def wait_and_refresh(timeout, %User{} = a, %User{} = b) do
with :ok <- :timer.sleep(timeout),
- %User{} = a <- User.get_cached_by_id(a.id),
- %User{} = b <- User.get_cached_by_id(b.id) do
+ %User{} = a <- get_cached_by_id(a.id),
+ %User{} = b <- get_cached_by_id(b.id) do
{:ok, a, b}
else
- _e ->
- :error
+ nil -> :error
end
end
defp normalize_tags(tags) do
[tags]
|> List.flatten()
- |> Enum.map(&String.downcase(&1))
+ |> Enum.map(&String.downcase/1)
end
defp local_nickname_regex do
@spec switch_email_notifications(t(), String.t(), boolean()) ::
{:ok, t()} | {:error, Ecto.Changeset.t()}
def switch_email_notifications(user, type, status) do
- info = Pleroma.User.Info.update_email_notifications(user.info, %{type => status})
-
- change(user)
- |> put_embed(:info, info)
- |> update_and_set_cache()
+ update_info(user, &User.Info.update_email_notifications(&1, %{type => status}))
end
@doc """
def toggle_confirmation(%User{} = user) do
need_confirmation? = !user.info.confirmation_pending
- info_changeset =
- User.Info.confirmation_changeset(user.info, need_confirmation: need_confirmation?)
-
user
- |> change()
- |> put_embed(:info, info_changeset)
- |> update_and_set_cache()
+ |> update_info(&User.Info.confirmation_changeset(&1, need_confirmation: need_confirmation?))
end
def get_mascot(%{info: %{mascot: %{} = mascot}}) when not is_nil(mascot) do
}
end
- def ensure_keys_present(%User{info: info} = user) do
- if info.keys do
- {:ok, user}
- else
- {:ok, pem} = Keys.generate_rsa_pem()
+ def ensure_keys_present(%{keys: keys} = user) when not is_nil(keys), do: {:ok, user}
+ def ensure_keys_present(%User{} = user) do
+ with {:ok, pem} <- Keys.generate_rsa_pem() do
user
- |> Ecto.Changeset.change()
- |> Ecto.Changeset.put_embed(:info, User.Info.set_keys(info, pem))
+ |> cast(%{keys: pem}, [:keys])
+ |> validate_required([:keys])
|> update_and_set_cache()
end
end
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
+
+ # 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])
+ |> validate_required([:email])
+ |> unique_constraint(:email)
+ |> validate_format(:email, @email_regex)
+ |> update_and_set_cache()
+ end
+
+ @doc """
+ Changes `user.info` and returns the user changeset.
+
+ `fun` is called with the `user.info`.
+ """
+ def change_info(user, fun) do
+ changeset = change(user)
+ info = get_field(changeset, :info) || %User.Info{}
+ put_embed(changeset, :info, fun.(info))
+ end
+
+ @doc """
+ Updates `user.info` and sets cache.
+
+ `fun` is called with the `user.info`.
+ """
+ def update_info(user, fun) do
+ user
+ |> change_info(fun)
+ |> update_and_set_cache()
+ end
end
defmodule Pleroma.Web.ActivityPub.ActivityPub do
alias Pleroma.Activity
+ alias Pleroma.Activity.Ir.Topics
alias Pleroma.Config
alias Pleroma.Conversation
alias Pleroma.Notification
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
import Ecto.Query
import Pleroma.Web.ActivityPub.Utils
activity
end
- PleromaJobQueue.enqueue(:background, Pleroma.Web.RichMedia.Helpers, [:fetch, activity])
+ BackgroundWorker.enqueue("fetch_data_for_activity", %{"activity_id" => activity.id})
Notification.create_notifications(activity)
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
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
# 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 listen(%{to: to, actor: actor, context: context, object: object} = params) do
+ additional = params[:additional] || %{}
+ # only accept false as false value
+ local = !(params[:local] == false)
+ published = params[:published]
+
+ with listen_data <-
+ make_listen_data(
+ %{to: to, actor: actor, published: published, context: context, object: object},
+ additional
+ ),
+ {:ok, activity} <- insert(listen_data, local),
+ :ok <- maybe_federate(activity) do
+ {:ok, activity}
+ else
+ {:error, message} ->
+ {:error, message}
+ end
+ end
+
def accept(%{to: to, actor: actor, object: object} = params) do
# only accept false as false value
local = !(params[:local] == false)
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
+ @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])
end
end
+ @spec flag(map()) :: {:ok, Activity.t()} | any
def flag(
%{
actor: actor,
- context: context,
+ context: _context,
account: account,
statuses: statuses,
content: content
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]})
end
@spec fetch_latest_activity_id_for_context(String.t(), keyword() | map()) ::
- Pleroma.FlakeId.t() | nil
+ FlakeId.Ecto.CompatType.t() | nil
def fetch_latest_activity_id_for_context(context, opts \\ %{}) do
context
|> fetch_activities_for_context_query(Map.merge(%{"skip_preload" => true}, opts))
|> Repo.one()
end
- def fetch_public_activities(opts \\ %{}) do
- q = fetch_activities_query([Pleroma.Constants.as_public()], opts)
+ def fetch_public_activities(opts \\ %{}, pagination \\ :keyset) do
+ opts = Map.drop(opts, ["user"])
- q
+ [Pleroma.Constants.as_public()]
+ |> fetch_activities_query(opts)
|> restrict_unlisted()
- |> Pagination.fetch_paginated(opts)
+ |> Pagination.fetch_paginated(opts, pagination)
|> Enum.reverse()
end
defp restrict_thread_visibility(query, _, _), do: query
+ def fetch_user_abstract_activities(user, reading_user, params \\ %{}) do
+ params =
+ params
+ |> Map.put("user", reading_user)
+ |> Map.put("actor_id", user.ap_id)
+ |> Map.put("whole_db", true)
+
+ recipients =
+ user_activities_recipients(%{
+ "godmode" => params["godmode"],
+ "reading_user" => reading_user
+ })
+
+ fetch_activities(recipients, params)
+ |> Enum.reverse()
+ end
+
def fetch_user_activities(user, reading_user, params \\ %{}) do
params =
params
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
)
unless opts["skip_preload"] do
- from([thread_mute: tm] in query, where: is_nil(tm))
+ from([thread_mute: tm] in query, where: is_nil(tm.user_id))
else
query
end
defp restrict_muted_reblogs(query, _), do: query
- defp exclude_poll_votes(query, %{"include_poll_votes" => "true"}), do: query
+ defp exclude_poll_votes(query, %{"include_poll_votes" => true}), do: query
defp exclude_poll_votes(query, _) do
if has_named_binding?(query, :object) do
|> exclude_poll_votes(opts)
end
- def fetch_activities(recipients, opts \\ %{}) do
+ def fetch_activities(recipients, opts \\ %{}, pagination \\ :keyset) do
list_memberships = Pleroma.List.memberships(opts["user"])
fetch_activities_query(recipients ++ list_memberships, opts)
- |> Pagination.fetch_paginated(opts)
+ |> Pagination.fetch_paginated(opts, pagination)
|> Enum.reverse()
|> maybe_update_cc(list_memberships, opts["user"])
end
)
end
- def fetch_activities_bounded(recipients, recipients_with_public, opts \\ %{}) do
+ def fetch_activities_bounded(
+ recipients,
+ recipients_with_public,
+ opts \\ %{},
+ pagination \\ :keyset
+ ) do
fetch_activities_query([], opts)
|> fetch_activities_bounded_query(recipients, recipients_with_public)
- |> Pagination.fetch_paginated(opts)
+ |> Pagination.fetch_paginated(opts, pagination)
|> Enum.reverse()
end
locked = data["manuallyApprovesFollowers"] || false
data = Transmogrifier.maybe_fix_user_object(data)
+ discoverable = data["discoverable"] || false
user_data = %{
ap_id: data["id"],
source_data: data,
banner: banner,
fields: fields,
- locked: locked
+ locked: locked,
+ discoverable: discoverable
},
avatar: avatar,
name: data["name"],
projects / akkoma / commitdiff