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