alias Pleroma.User
require Logger
require Pleroma.Constants
+ import Ecto.Query
import Mix.Pleroma
use Mix.Task
def run(["update_users_following_followers_counts"]) do
start_pleroma()
- User
- |> Repo.all()
- |> Enum.each(&User.update_follower_count/1)
+ Repo.transaction(
+ fn ->
+ from(u in User, select: u)
+ |> Repo.stream()
+ |> Stream.each(&User.update_follower_count/1)
+ |> Stream.run()
+ end,
+ timeout: :infinity
+ )
end
def run(["prune_objects" | args]) do
- import Ecto.Query
-
{options, [], []} =
OptionParser.parse(
args,
end
def run(["fix_likes_collections"]) do
- import Ecto.Query
-
start_pleroma()
from(object in Object,
where: fragment("(?)->>'likes' is not null", object.data),
select: %{id: object.id, likes: fragment("(?)->>'likes'", object.data)}
)
- |> Pleroma.RepoStreamer.chunk_stream(100)
+ |> Pleroma.Repo.chunk_stream(100, :batches)
|> Stream.each(fn objects ->
ids =
objects
|> Stream.run()
end
+ def run(["fill_old_hashtags", month_limit]) do
+ import Ecto.Query
+
+ start_pleroma()
+
+ month_limit = String.to_integer(month_limit)
+
+ if month_limit < 1 do
+ shell_error("Invalid `month_limit` argument, needs to be greater than 1")
+ else
+ time_limit = DateTime.utc_now() |> Timex.shift(months: -month_limit)
+
+ from(
+ o in Object,
+ where: fragment("(?)->>'hashtags' is null", o.data),
+ where: fragment("(?)->>'tag' != '[]'", o.data),
+ where: o.inserted_at < ^time_limit,
+ select: %{id: o.id, tag: fragment("(?)->>'tag'", o.data)}
+ )
+ |> Pleroma.Repo.chunk_stream(200, :batches)
+ |> Stream.each(fn objects ->
+ Repo.transaction(fn ->
+ objects_first = objects |> List.first()
+ objects_last = objects |> List.last()
+
+ Logger.info(
+ "fill_old_hashtags: #{objects_first.id} (#{objects_first.inserted_at}) -- #{
+ objects_last.id
+ } (#{objects_last.inserted_at})"
+ )
+
+ objects
+ |> Enum.map(fn object ->
+ tags =
+ object.tag
+ |> Jason.decode!()
+ |> Enum.filter(&is_bitstring(&1))
+
+ Object
+ |> where([o], o.id == ^object.id)
+ |> update([o],
+ set: [data: fragment("safe_jsonb_set(?, '{hashtags}', ?, true)", o.data, ^tags)]
+ )
+ |> Repo.update_all([], timeout: :infinity)
+ end)
+ end)
+ end)
+ |> Stream.run()
+ end
+ end
+
def run(["vacuum", args]) do
start_pleroma()
Maintenance.vacuum(args)
end
+
+ def run(["ensure_expiration"]) do
+ start_pleroma()
+ days = Pleroma.Config.get([:mrf_activity_expiration, :days], 365)
+
+ Pleroma.Activity
+ |> join(:inner, [a], o in Object,
+ on:
+ fragment(
+ "(?->>'id') = COALESCE((?)->'object'->> 'id', (?)->>'object')",
+ o.data,
+ a.data,
+ a.data
+ )
+ )
+ |> where(local: true)
+ |> where([a], fragment("(? ->> 'type'::text) = 'Create'", a.data))
+ |> where([_a, o], fragment("?->>'type' = 'Note'", o.data))
+ |> Pleroma.Repo.chunk_stream(100, :batches)
+ |> Stream.each(fn activities ->
+ Enum.each(activities, fn activity ->
+ expires_at =
+ activity.inserted_at
+ |> DateTime.from_naive!("Etc/UTC")
+ |> Timex.shift(days: days)
+
+ Pleroma.Workers.PurgeExpiredActivity.enqueue(%{
+ activity_id: activity.id,
+ expires_at: expires_at
+ })
+ end)
+ end)
+ |> Stream.run()
+ end
end