1 # Pleroma: A lightweight social networking server
2 # Copyright © 2017-2021 Pleroma Authors <https://pleroma.social/>
3 # SPDX-License-Identifier: AGPL-3.0-only
5 defmodule Mix.Tasks.Pleroma.Database do
6 alias Pleroma.Conversation
7 alias Pleroma.Maintenance
13 require Pleroma.Constants
20 @shortdoc "A collection of database related tasks"
21 @moduledoc File.read!("docs/docs/administration/CLI_tasks/database.md")
23 def run(["remove_embedded_objects" | args]) do
33 Logger.info("Removing embedded objects")
36 "update activities set data = safe_jsonb_set(data, '{object}'::text[], data->'object'->'id') where data->'object'->>'id' is not null;",
41 if Keyword.get(options, :vacuum) do
42 Maintenance.vacuum("full")
46 def run(["bump_all_conversations"]) do
48 Conversation.bump_for_all_activities()
51 def run(["update_users_following_followers_counts"]) do
56 from(u in User, select: u)
58 |> Stream.each(&User.update_follower_count/1)
65 def run(["prune_objects" | args]) do
71 keep_threads: :boolean,
72 keep_non_public: :boolean,
73 prune_orphaned_activities: :boolean
79 deadline = Pleroma.Config.get([:instance, :remote_post_retention_days])
80 time_deadline = NaiveDateTime.utc_now() |> NaiveDateTime.add(-(deadline * 86_400))
82 log_message = "Pruning objects older than #{deadline} days"
85 if Keyword.get(options, :keep_non_public) do
86 log_message <> ", keeping non public posts"
92 if Keyword.get(options, :keep_threads) do
93 log_message <> ", keeping threads intact"
99 if Keyword.get(options, :prune_orphaned_activities) do
100 log_message <> ", pruning orphaned activities"
106 if Keyword.get(options, :vacuum) do
108 ", doing a full vacuum (you shouldn't do this as a recurring maintanance task)"
113 Logger.info(log_message)
115 if Keyword.get(options, :keep_threads) do
116 # We want to delete objects from threads where
117 # 1. the newest post is still old
118 # 2. none of the activities is local
119 # 3. none of the activities is bookmarked
120 # 4. optionally none of the posts is non-public
122 if Keyword.get(options, :keep_non_public) do
124 |> join(:left, [a], b in Pleroma.Bookmark, on: a.id == b.activity_id)
125 |> group_by([a], fragment("? ->> 'context'::text", a.data))
129 # Posts (checked on Create Activity) is non-public
130 "bool_or((not(?->'to' \\? ? OR ?->'cc' \\? ?)) and ? ->> 'type' = 'Create')",
132 ^Pleroma.Constants.as_public(),
134 ^Pleroma.Constants.as_public(),
140 |> join(:left, [a], b in Pleroma.Bookmark, on: a.id == b.activity_id)
141 |> group_by([a], fragment("? ->> 'context'::text", a.data))
143 |> having([a], max(a.updated_at) < ^time_deadline)
144 |> having([a], not fragment("bool_or(?)", a.local))
145 |> having([_, b], fragment("max(?::text) is null", b.id))
146 |> select([a], fragment("? ->> 'context'::text", a.data))
149 |> where([o], fragment("? ->> 'context'::text", o.data) in subquery(deletable_context))
151 if Keyword.get(options, :keep_non_public) do
156 "?->'to' \\? ? OR ?->'cc' \\? ?",
158 ^Pleroma.Constants.as_public(),
160 ^Pleroma.Constants.as_public()
166 |> where([o], o.updated_at < ^time_deadline)
169 fragment("split_part(?->>'actor', '/', 3) != ?", o.data, ^Pleroma.Web.Endpoint.host())
172 |> Repo.delete_all(timeout: :infinity)
174 if Keyword.get(options, :prune_orphaned_activities) do
176 delete from public.activities
178 select a.id from public.activities a
179 left join public.objects o on a.data ->> 'object' = o.data ->> 'id'
180 left join public.activities a2 on a.data ->> 'object' = a2.data ->> 'id'
181 left join public.users u on a.data ->> 'object' = u.ap_id
182 -- Only clean up remote activities
184 -- For now we only focus on activities with direct links to objects
185 -- e.g. not json objects (in case of embedded objects) or json arrays (in case of multiple objects)
186 and jsonb_typeof(a."data" -> 'object') = 'string'
187 -- Find Activities that don't have existing objects
196 prune_hashtags_query = """
197 DELETE FROM hashtags AS ht
199 SELECT 1 FROM hashtags_objects hto
200 WHERE ht.id = hto.hashtag_id)
203 Repo.query(prune_hashtags_query)
205 if Keyword.get(options, :vacuum) do
206 Maintenance.vacuum("full")
210 def run(["prune_task"]) do
214 |> Pleroma.Workers.Cron.PruneDatabaseWorker.perform()
217 def run(["fix_likes_collections"]) do
220 from(object in Object,
221 where: fragment("(?)->>'likes' is not null", object.data),
222 select: %{id: object.id, likes: fragment("(?)->>'likes'", object.data)}
224 |> Pleroma.Repo.chunk_stream(100, :batches)
225 |> Stream.each(fn objects ->
228 |> Enum.filter(fn object -> object.likes |> Jason.decode!() |> is_map() end)
232 |> where([object], object.id in ^ids)
237 "safe_jsonb_set(?, '{likes}', '[]'::jsonb, true)",
242 |> Repo.update_all([], timeout: :infinity)
247 def run(["vacuum", args]) do
250 Maintenance.vacuum(args)
253 def run(["ensure_expiration"]) do
255 days = Pleroma.Config.get([:mrf_activity_expiration, :days], 365)
258 |> join(:inner, [a], o in Object,
261 "(?->>'id') = COALESCE((?)->'object'->> 'id', (?)->>'object')",
267 |> where(local: true)
268 |> where([a], fragment("(? ->> 'type'::text) = 'Create'", a.data))
269 |> where([_a, o], fragment("?->>'type' = 'Note'", o.data))
270 |> Pleroma.Repo.chunk_stream(100, :batches)
271 |> Stream.each(fn activities ->
272 Enum.each(activities, fn activity ->
275 |> DateTime.from_naive!("Etc/UTC")
276 |> Timex.shift(days: days)
278 Pleroma.Workers.PurgeExpiredActivity.enqueue(%{
279 activity_id: activity.id,
280 expires_at: expires_at
287 def run(["set_text_search_config", tsconfig]) do
289 %{rows: [[tsc]]} = Ecto.Adapters.SQL.query!(Pleroma.Repo, "SHOW default_text_search_config;")
290 shell_info("Current default_text_search_config: #{tsc}")
292 %{rows: [[db]]} = Ecto.Adapters.SQL.query!(Pleroma.Repo, "SELECT current_database();")
293 shell_info("Update default_text_search_config: #{tsconfig}")
296 Ecto.Adapters.SQL.query!(
298 "ALTER DATABASE #{db} SET default_text_search_config = '#{tsconfig}';"
301 # non-exist config will not raise excpetion but only give >0 messages
302 if length(msg) > 0 do
303 shell_info("Error: #{inspect(msg, pretty: true)}")
305 rum_enabled = Pleroma.Config.get([:database, :rum_enabled])
306 shell_info("Recreate index, RUM: #{rum_enabled}")
308 # Note SQL below needs to be kept up-to-date with latest GIN or RUM index definition in future
310 Ecto.Adapters.SQL.query!(
312 "CREATE OR REPLACE FUNCTION objects_fts_update() RETURNS trigger AS $$ BEGIN
313 new.fts_content := to_tsvector(new.data->>'content');
316 $$ LANGUAGE plpgsql",
321 shell_info("Refresh RUM index")
322 Ecto.Adapters.SQL.query!(Pleroma.Repo, "UPDATE objects SET updated_at = NOW();")
324 Ecto.Adapters.SQL.query!(Pleroma.Repo, "DROP INDEX IF EXISTS objects_fts;")
326 Ecto.Adapters.SQL.query!(
328 "CREATE INDEX CONCURRENTLY objects_fts ON objects USING gin(to_tsvector('#{tsconfig}', data->>'content')); ",
338 # Rolls back a specific migration (leaving subsequent migrations applied).
339 # WARNING: imposes a risk of unrecoverable data loss — proceed at your own responsibility.
340 # Based on https://stackoverflow.com/a/53825840
341 def run(["rollback", version]) do
342 prompt = "SEVERE WARNING: this operation may result in unrecoverable data loss. Continue?"
344 if shell_prompt(prompt, "n") in ~w(Yn Y y) do
346 Ecto.Migrator.with_repo(Pleroma.Repo, fn repo ->
347 version = String.to_integer(version)
348 re = ~r/^#{version}_.*\.exs/
349 path = Ecto.Migrator.migrations_path(repo)
351 with {_, "" <> file} <- {:find, Enum.find(File.ls!(path), &String.match?(&1, re))},
352 {_, [{mod, _} | _]} <- {:compile, Code.compile_file(Path.join(path, file))},
353 {_, :ok} <- {:rollback, Ecto.Migrator.down(repo, version, mod)} do
354 {:ok, "Reversed migration: #{file}"}
356 {:find, _} -> {:error, "No migration found with version prefix: #{version}"}
357 {:compile, e} -> {:error, "Problem compiling migration module: #{inspect(e)}"}
358 {:rollback, e} -> {:error, "Problem reversing migration: #{inspect(e)}"}
362 shell_info(inspect(result))