git.squeep.com Git - akkoma/blob - lib/mix/tasks/pleroma/database.ex

   1 # Pleroma: A lightweight social networking server
   2 # Copyright © 2017-2021 Pleroma Authors <https://pleroma.social/>
   3 # SPDX-License-Identifier: AGPL-3.0-only
   4
   5 defmodule Mix.Tasks.Pleroma.Database do
   6   alias Pleroma.Conversation
   7   alias Pleroma.Maintenance
   8   alias Pleroma.Object
   9   alias Pleroma.Repo
  10   alias Pleroma.User
  11
  12   require Logger
  13   require Pleroma.Constants
  14
  15   import Ecto.Query
  16   import Mix.Pleroma
  17
  18   use Mix.Task
  19
  20   @shortdoc "A collection of database related tasks"
  21   @moduledoc File.read!("docs/docs/administration/CLI_tasks/database.md")
  22
  23   def run(["remove_embedded_objects" | args]) do
  24     {options, [], []} =
  25       OptionParser.parse(
  26         args,
  27         strict: [
  28           vacuum: :boolean
  29         ]
  30       )
  31
  32     start_pleroma()
  33     Logger.info("Removing embedded objects")
  34
  35     Repo.query!(
  36       "update activities set data = safe_jsonb_set(data, '{object}'::text[], data->'object'->'id') where data->'object'->>'id' is not null;",
  37       [],
  38       timeout: :infinity
  39     )
  40
  41     if Keyword.get(options, :vacuum) do
  42       Maintenance.vacuum("full")
  43     end
  44   end
  45
  46   def run(["bump_all_conversations"]) do
  47     start_pleroma()
  48     Conversation.bump_for_all_activities()
  49   end
  50
  51   def run(["update_users_following_followers_counts"]) do
  52     start_pleroma()
  53
  54     Repo.transaction(
  55       fn ->
  56         from(u in User, select: u)
  57         |> Repo.stream()
  58         |> Stream.each(&User.update_follower_count/1)
  59         |> Stream.run()
  60       end,
  61       timeout: :infinity
  62     )
  63   end
  64
  65   def run(["prune_objects" | args]) do
  66     {options, [], []} =
  67       OptionParser.parse(
  68         args,
  69         strict: [
  70           vacuum: :boolean
  71         ]
  72       )
  73
  74     start_pleroma()
  75
  76     deadline = Pleroma.Config.get([:instance, :remote_post_retention_days])
  77
  78     Logger.info("Pruning objects older than #{deadline} days")
  79
  80     time_deadline =
  81       NaiveDateTime.utc_now()
  82       |> NaiveDateTime.add(-(deadline * 86_400))
  83
  84     from(o in Object,
  85       where:
  86         fragment(
  87           "?->'to' \\? ? OR ?->'cc' \\? ?",
  88           o.data,
  89           ^Pleroma.Constants.as_public(),
  90           o.data,
  91           ^Pleroma.Constants.as_public()
  92         ),
  93       where: o.inserted_at < ^time_deadline,
  94       where:
  95         fragment("split_part(?->>'actor', '/', 3) != ?", o.data, ^Pleroma.Web.Endpoint.host())
  96     )
  97     |> Repo.delete_all(timeout: :infinity)
  98
  99     prune_hashtags_query = """
 100     DELETE FROM hashtags AS ht
 101     WHERE NOT EXISTS (
 102       SELECT 1 FROM hashtags_objects hto
 103       WHERE ht.id = hto.hashtag_id)
 104     """
 105
 106     Repo.query(prune_hashtags_query)
 107
 108     if Keyword.get(options, :vacuum) do
 109       Maintenance.vacuum("full")
 110     end
 111   end
 112
 113   def run(["prune_task"]) do
 114     start_pleroma()
 115
 116     nil
 117     |> Pleroma.Workers.Cron.PruneDatabaseWorker.perform()
 118   end
 119
 120   def run(["fix_likes_collections"]) do
 121     start_pleroma()
 122
 123     from(object in Object,
 124       where: fragment("(?)->>'likes' is not null", object.data),
 125       select: %{id: object.id, likes: fragment("(?)->>'likes'", object.data)}
 126     )
 127     |> Pleroma.Repo.chunk_stream(100, :batches)
 128     |> Stream.each(fn objects ->
 129       ids =
 130         objects
 131         |> Enum.filter(fn object -> object.likes |> Jason.decode!() |> is_map() end)
 132         |> Enum.map(& &1.id)
 133
 134       Object
 135       |> where([object], object.id in ^ids)
 136       |> update([object],
 137         set: [
 138           data:
 139             fragment(
 140               "safe_jsonb_set(?, '{likes}', '[]'::jsonb, true)",
 141               object.data
 142             )
 143         ]
 144       )
 145       |> Repo.update_all([], timeout: :infinity)
 146     end)
 147     |> Stream.run()
 148   end
 149
 150   def run(["vacuum", args]) do
 151     start_pleroma()
 152
 153     Maintenance.vacuum(args)
 154   end
 155
 156   def run(["ensure_expiration"]) do
 157     start_pleroma()
 158     days = Pleroma.Config.get([:mrf_activity_expiration, :days], 365)
 159
 160     Pleroma.Activity
 161     |> join(:inner, [a], o in Object,
 162       on:
 163         fragment(
 164           "(?->>'id') = COALESCE((?)->'object'->> 'id', (?)->>'object')",
 165           o.data,
 166           a.data,
 167           a.data
 168         )
 169     )
 170     |> where(local: true)
 171     |> where([a], fragment("(? ->> 'type'::text) = 'Create'", a.data))
 172     |> where([_a, o], fragment("?->>'type' = 'Note'", o.data))
 173     |> Pleroma.Repo.chunk_stream(100, :batches)
 174     |> Stream.each(fn activities ->
 175       Enum.each(activities, fn activity ->
 176         expires_at =
 177           activity.inserted_at
 178           |> DateTime.from_naive!("Etc/UTC")
 179           |> Timex.shift(days: days)
 180
 181         Pleroma.Workers.PurgeExpiredActivity.enqueue(%{
 182           activity_id: activity.id,
 183           expires_at: expires_at
 184         })
 185       end)
 186     end)
 187     |> Stream.run()
 188   end
 189
 190   def run(["set_text_search_config", tsconfig]) do
 191     start_pleroma()
 192     %{rows: [[tsc]]} = Ecto.Adapters.SQL.query!(Pleroma.Repo, "SHOW default_text_search_config;")
 193     shell_info("Current default_text_search_config: #{tsc}")
 194
 195     %{rows: [[db]]} = Ecto.Adapters.SQL.query!(Pleroma.Repo, "SELECT current_database();")
 196     shell_info("Update default_text_search_config: #{tsconfig}")
 197
 198     %{messages: msg} =
 199       Ecto.Adapters.SQL.query!(
 200         Pleroma.Repo,
 201         "ALTER DATABASE #{db} SET default_text_search_config = '#{tsconfig}';"
 202       )
 203
 204     # non-exist config will not raise excpetion but only give >0 messages
 205     if length(msg) > 0 do
 206       shell_info("Error: #{inspect(msg, pretty: true)}")
 207     else
 208       rum_enabled = Pleroma.Config.get([:database, :rum_enabled])
 209       shell_info("Recreate index, RUM: #{rum_enabled}")
 210
 211       # Note SQL below needs to be kept up-to-date with latest GIN or RUM index definition in future
 212       if rum_enabled do
 213         Ecto.Adapters.SQL.query!(
 214           Pleroma.Repo,
 215           "CREATE OR REPLACE FUNCTION objects_fts_update() RETURNS trigger AS $$ BEGIN
 216           new.fts_content := to_tsvector(new.data->>'content');
 217           RETURN new;
 218           END
 219           $$ LANGUAGE plpgsql",
 220           [],
 221           timeout: :infinity
 222         )
 223
 224         shell_info("Refresh RUM index")
 225         Ecto.Adapters.SQL.query!(Pleroma.Repo, "UPDATE objects SET updated_at = NOW();")
 226       else
 227         Ecto.Adapters.SQL.query!(Pleroma.Repo, "DROP INDEX IF EXISTS objects_fts;")
 228
 229         Ecto.Adapters.SQL.query!(
 230           Pleroma.Repo,
 231           "CREATE INDEX CONCURRENTLY objects_fts ON objects USING gin(to_tsvector('#{tsconfig}', data->>'content')); ",
 232           [],
 233           timeout: :infinity
 234         )
 235       end
 236
 237       shell_info('Done.')
 238     end
 239   end
 240
 241   # Rolls back a specific migration (leaving subsequent migrations applied).
 242   # WARNING: imposes a risk of unrecoverable data loss — proceed at your own responsibility.
 243   # Based on https://stackoverflow.com/a/53825840
 244   def run(["rollback", version]) do
 245     prompt = "SEVERE WARNING: this operation may result in unrecoverable data loss. Continue?"
 246
 247     if shell_prompt(prompt, "n") in ~w(Yn Y y) do
 248       {_, result, _} =
 249         Ecto.Migrator.with_repo(Pleroma.Repo, fn repo ->
 250           version = String.to_integer(version)
 251           re = ~r/^#{version}_.*\.exs/
 252           path = Ecto.Migrator.migrations_path(repo)
 253
 254           with {_, "" <> file} <- {:find, Enum.find(File.ls!(path), &String.match?(&1, re))},
 255                {_, [{mod, _} | _]} <- {:compile, Code.compile_file(Path.join(path, file))},
 256                {_, :ok} <- {:rollback, Ecto.Migrator.down(repo, version, mod)} do
 257             {:ok, "Reversed migration: #{file}"}
 258           else
 259             {:find, _} -> {:error, "No migration found with version prefix: #{version}"}
 260             {:compile, e} -> {:error, "Problem compiling migration module: #{inspect(e)}"}
 261             {:rollback, e} -> {:error, "Problem reversing migration: #{inspect(e)}"}
 262           end
 263         end)
 264
 265       shell_info(inspect(result))
 266     end
 267   end
 268 end