Git of dcc - anni/blob - lib/mix/tasks/pleroma/database.ex

   1 # Pleroma: A lightweight social networking server
   2 # Copyright © 2017-2022 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/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(["fix_likes_collections"]) do
 114     start_pleroma()
 115
 116     from(object in Object,
 117       where: fragment("(?)->>'likes' is not null", object.data),
 118       select: %{id: object.id, likes: fragment("(?)->>'likes'", object.data)}
 119     )
 120     |> Pleroma.Repo.chunk_stream(100, :batches)
 121     |> Stream.each(fn objects ->
 122       ids =
 123         objects
 124         |> Enum.filter(fn object -> object.likes |> Jason.decode!() |> is_map() end)
 125         |> Enum.map(& &1.id)
 126
 127       Object
 128       |> where([object], object.id in ^ids)
 129       |> update([object],
 130         set: [
 131           data:
 132             fragment(
 133               "safe_jsonb_set(?, '{likes}', '[]'::jsonb, true)",
 134               object.data
 135             )
 136         ]
 137       )
 138       |> Repo.update_all([], timeout: :infinity)
 139     end)
 140     |> Stream.run()
 141   end
 142
 143   def run(["vacuum", args]) do
 144     start_pleroma()
 145
 146     Maintenance.vacuum(args)
 147   end
 148
 149   def run(["ensure_expiration"]) do
 150     start_pleroma()
 151     days = Pleroma.Config.get([:mrf_activity_expiration, :days], 365)
 152
 153     Pleroma.Activity
 154     |> join(:inner, [a], o in Object,
 155       on:
 156         fragment(
 157           "(?->>'id') = associated_object_id((?))",
 158           o.data,
 159           a.data
 160         )
 161     )
 162     |> where(local: true)
 163     |> where([a], fragment("(? ->> 'type'::text) = 'Create'", a.data))
 164     |> where([_a, o], fragment("?->>'type' = 'Note'", o.data))
 165     |> Pleroma.Repo.chunk_stream(100, :batches)
 166     |> Stream.each(fn activities ->
 167       Enum.each(activities, fn activity ->
 168         expires_at =
 169           activity.inserted_at
 170           |> DateTime.from_naive!("Etc/UTC")
 171           |> Timex.shift(days: days)
 172
 173         Pleroma.Workers.PurgeExpiredActivity.enqueue(%{
 174           activity_id: activity.id,
 175           expires_at: expires_at
 176         })
 177       end)
 178     end)
 179     |> Stream.run()
 180   end
 181
 182   def run(["set_text_search_config", tsconfig]) do
 183     start_pleroma()
 184     %{rows: [[tsc]]} = Ecto.Adapters.SQL.query!(Pleroma.Repo, "SHOW default_text_search_config;")
 185     shell_info("Current default_text_search_config: #{tsc}")
 186
 187     %{rows: [[db]]} = Ecto.Adapters.SQL.query!(Pleroma.Repo, "SELECT current_database();")
 188     shell_info("Update default_text_search_config: #{tsconfig}")
 189
 190     %{messages: msg} =
 191       Ecto.Adapters.SQL.query!(
 192         Pleroma.Repo,
 193         "ALTER DATABASE #{db} SET default_text_search_config = '#{tsconfig}';"
 194       )
 195
 196     # non-exist config will not raise exception but only give >0 messages
 197     if length(msg) > 0 do
 198       shell_info("Error: #{inspect(msg, pretty: true)}")
 199     else
 200       rum_enabled = Pleroma.Config.get([:database, :rum_enabled])
 201       shell_info("Recreate index, RUM: #{rum_enabled}")
 202
 203       # Note SQL below needs to be kept up-to-date with latest GIN or RUM index definition in future
 204       if rum_enabled do
 205         Ecto.Adapters.SQL.query!(
 206           Pleroma.Repo,
 207           "CREATE OR REPLACE FUNCTION objects_fts_update() RETURNS trigger AS $$ BEGIN
 208           new.fts_content := to_tsvector(new.data->>'content');
 209           RETURN new;
 210           END
 211           $$ LANGUAGE plpgsql",
 212           [],
 213           timeout: :infinity
 214         )
 215
 216         shell_info("Refresh RUM index")
 217         Ecto.Adapters.SQL.query!(Pleroma.Repo, "UPDATE objects SET updated_at = NOW();")
 218       else
 219         Ecto.Adapters.SQL.query!(Pleroma.Repo, "DROP INDEX IF EXISTS objects_fts;")
 220
 221         Ecto.Adapters.SQL.query!(
 222           Pleroma.Repo,
 223           "CREATE INDEX CONCURRENTLY objects_fts ON objects USING gin(to_tsvector('#{tsconfig}', data->>'content')); ",
 224           [],
 225           timeout: :infinity
 226         )
 227       end
 228
 229       shell_info('Done.')
 230     end
 231   end
 232
 233   # Rolls back a specific migration (leaving subsequent migrations applied).
 234   # WARNING: imposes a risk of unrecoverable data loss — proceed at your own responsibility.
 235   # Based on https://stackoverflow.com/a/53825840
 236   def run(["rollback", version]) do
 237     prompt = "SEVERE WARNING: this operation may result in unrecoverable data loss. Continue?"
 238
 239     if shell_prompt(prompt, "n") in ~w(Yn Y y) do
 240       {_, result, _} =
 241         Ecto.Migrator.with_repo(Pleroma.Repo, fn repo ->
 242           version = String.to_integer(version)
 243           re = ~r/^#{version}_.*\.exs/
 244           path = Ecto.Migrator.migrations_path(repo)
 245
 246           with {_, "" <> file} <- {:find, Enum.find(File.ls!(path), &String.match?(&1, re))},
 247                {_, [{mod, _} | _]} <- {:compile, Code.compile_file(Path.join(path, file))},
 248                {_, :ok} <- {:rollback, Ecto.Migrator.down(repo, version, mod)} do
 249             {:ok, "Reversed migration: #{file}"}
 250           else
 251             {:find, _} -> {:error, "No migration found with version prefix: #{version}"}
 252             {:compile, e} -> {:error, "Problem compiling migration module: #{inspect(e)}"}
 253             {:rollback, e} -> {:error, "Problem reversing migration: #{inspect(e)}"}
 254           end
 255         end)
 256
 257       shell_info(inspect(result))
 258     end
 259   end
 260 end