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
78 deadline = Pleroma.Config.get([:instance, :remote_post_retention_days])
79 time_deadline = NaiveDateTime.utc_now() |> NaiveDateTime.add(-(deadline * 86_400))
81 log_message = "Pruning objects older than #{deadline} days"
84 if Keyword.get(options, :keep_non_public) do
85 log_message <> ", keeping non public posts"
91 if Keyword.get(options, :keep_threads) do
92 log_message <> ", keeping threads intact"
97 Logger.info(log_message)
99 if Keyword.get(options, :keep_threads) do
100 # We want to delete objects from threads where
101 # 1. the newest post is still old
102 # 2. none of the activities is local
103 # 3. none of the activities is bookmarked
104 # 4. optionally none of the posts is non-public
106 if Keyword.get(options, :keep_non_public) do
108 |> join(:left, [a], b in Pleroma.Bookmark, on: a.id == b.activity_id)
109 |> group_by([a], fragment("? ->> 'context'::text", a.data))
113 # Posts (checked on Create Activity) is non-public
114 "bool_or((not(?->'to' \\? ? OR ?->'cc' \\? ?)) and ? ->> 'type' = 'Create')",
116 ^Pleroma.Constants.as_public(),
118 ^Pleroma.Constants.as_public(),
124 |> join(:left, [a], b in Pleroma.Bookmark, on: a.id == b.activity_id)
125 |> group_by([a], fragment("? ->> 'context'::text", a.data))
127 |> having([a], max(a.updated_at) < ^time_deadline)
128 |> having([a], not fragment("bool_or(?)", a.local))
129 |> having([_, b], fragment("max(?::text) is null", b.id))
130 |> select([a], fragment("? ->> 'context'::text", a.data))
133 |> where([o], fragment("? ->> 'context'::text", o.data) in subquery(deletable_context))
135 if Keyword.get(options, :keep_non_public) do
140 "?->'to' \\? ? OR ?->'cc' \\? ?",
142 ^Pleroma.Constants.as_public(),
144 ^Pleroma.Constants.as_public()
150 |> where([o], o.updated_at < ^time_deadline)
153 fragment("split_part(?->>'actor', '/', 3) != ?", o.data, ^Pleroma.Web.Endpoint.host())
156 |> Repo.delete_all(timeout: :infinity)
158 prune_hashtags_query = """
159 DELETE FROM hashtags AS ht
161 SELECT 1 FROM hashtags_objects hto
162 WHERE ht.id = hto.hashtag_id)
165 Repo.query(prune_hashtags_query)
167 if Keyword.get(options, :vacuum) do
168 Maintenance.vacuum("full")
172 def run(["prune_task"]) do
176 |> Pleroma.Workers.Cron.PruneDatabaseWorker.perform()
179 def run(["fix_likes_collections"]) do
182 from(object in Object,
183 where: fragment("(?)->>'likes' is not null", object.data),
184 select: %{id: object.id, likes: fragment("(?)->>'likes'", object.data)}
186 |> Pleroma.Repo.chunk_stream(100, :batches)
187 |> Stream.each(fn objects ->
190 |> Enum.filter(fn object -> object.likes |> Jason.decode!() |> is_map() end)
194 |> where([object], object.id in ^ids)
199 "safe_jsonb_set(?, '{likes}', '[]'::jsonb, true)",
204 |> Repo.update_all([], timeout: :infinity)
209 def run(["vacuum", args]) do
212 Maintenance.vacuum(args)
215 def run(["ensure_expiration"]) do
217 days = Pleroma.Config.get([:mrf_activity_expiration, :days], 365)
220 |> join(:inner, [a], o in Object,
223 "(?->>'id') = COALESCE((?)->'object'->> 'id', (?)->>'object')",
229 |> where(local: true)
230 |> where([a], fragment("(? ->> 'type'::text) = 'Create'", a.data))
231 |> where([_a, o], fragment("?->>'type' = 'Note'", o.data))
232 |> Pleroma.Repo.chunk_stream(100, :batches)
233 |> Stream.each(fn activities ->
234 Enum.each(activities, fn activity ->
237 |> DateTime.from_naive!("Etc/UTC")
238 |> Timex.shift(days: days)
240 Pleroma.Workers.PurgeExpiredActivity.enqueue(%{
241 activity_id: activity.id,
242 expires_at: expires_at
249 def run(["set_text_search_config", tsconfig]) do
251 %{rows: [[tsc]]} = Ecto.Adapters.SQL.query!(Pleroma.Repo, "SHOW default_text_search_config;")
252 shell_info("Current default_text_search_config: #{tsc}")
254 %{rows: [[db]]} = Ecto.Adapters.SQL.query!(Pleroma.Repo, "SELECT current_database();")
255 shell_info("Update default_text_search_config: #{tsconfig}")
258 Ecto.Adapters.SQL.query!(
260 "ALTER DATABASE #{db} SET default_text_search_config = '#{tsconfig}';"
263 # non-exist config will not raise excpetion but only give >0 messages
264 if length(msg) > 0 do
265 shell_info("Error: #{inspect(msg, pretty: true)}")
267 rum_enabled = Pleroma.Config.get([:database, :rum_enabled])
268 shell_info("Recreate index, RUM: #{rum_enabled}")
270 # Note SQL below needs to be kept up-to-date with latest GIN or RUM index definition in future
272 Ecto.Adapters.SQL.query!(
274 "CREATE OR REPLACE FUNCTION objects_fts_update() RETURNS trigger AS $$ BEGIN
275 new.fts_content := to_tsvector(new.data->>'content');
278 $$ LANGUAGE plpgsql",
283 shell_info("Refresh RUM index")
284 Ecto.Adapters.SQL.query!(Pleroma.Repo, "UPDATE objects SET updated_at = NOW();")
286 Ecto.Adapters.SQL.query!(Pleroma.Repo, "DROP INDEX IF EXISTS objects_fts;")
288 Ecto.Adapters.SQL.query!(
290 "CREATE INDEX CONCURRENTLY objects_fts ON objects USING gin(to_tsvector('#{tsconfig}', data->>'content')); ",
300 # Rolls back a specific migration (leaving subsequent migrations applied).
301 # WARNING: imposes a risk of unrecoverable data loss — proceed at your own responsibility.
302 # Based on https://stackoverflow.com/a/53825840
303 def run(["rollback", version]) do
304 prompt = "SEVERE WARNING: this operation may result in unrecoverable data loss. Continue?"
306 if shell_prompt(prompt, "n") in ~w(Yn Y y) do
308 Ecto.Migrator.with_repo(Pleroma.Repo, fn repo ->
309 version = String.to_integer(version)
310 re = ~r/^#{version}_.*\.exs/
311 path = Ecto.Migrator.migrations_path(repo)
313 with {_, "" <> file} <- {:find, Enum.find(File.ls!(path), &String.match?(&1, re))},
314 {_, [{mod, _} | _]} <- {:compile, Code.compile_file(Path.join(path, file))},
315 {_, :ok} <- {:rollback, Ecto.Migrator.down(repo, version, mod)} do
316 {:ok, "Reversed migration: #{file}"}
318 {:find, _} -> {:error, "No migration found with version prefix: #{version}"}
319 {:compile, e} -> {:error, "Problem compiling migration module: #{inspect(e)}"}
320 {:rollback, e} -> {:error, "Problem reversing migration: #{inspect(e)}"}
324 shell_info(inspect(result))