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           keep_threads: :boolean,
  72           keep_non_public: :boolean
  73         ]
  74       )
  75
  76     start_pleroma()
  77
  78     deadline = Pleroma.Config.get([:instance, :remote_post_retention_days])
  79     time_deadline = NaiveDateTime.utc_now() |> NaiveDateTime.add(-(deadline * 86_400))
  80
  81     log_message = "Pruning objects older than #{deadline} days"
  82
  83     log_message =
  84       if Keyword.get(options, :keep_non_public) do
  85         log_message <> ", keeping non public posts"
  86       else
  87         log_message
  88       end
  89
  90     log_message =
  91       if Keyword.get(options, :keep_threads) do
  92         log_message <> ", keeping threads intact"
  93       else
  94         log_message
  95       end
  96
  97     Logger.info(log_message)
  98
  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
 105       deletable_context =
 106         if Keyword.get(options, :keep_non_public) do
 107           Pleroma.Activity
 108           |> join(:left, [a], b in Pleroma.Bookmark, on: a.id == b.activity_id)
 109           |> group_by([a], fragment("? ->> 'context'::text", a.data))
 110           |> having(
 111             [a],
 112             not fragment(
 113               # Posts (checked on Create Activity) is non-public
 114               "bool_or((not(?->'to' \\? ? OR ?->'cc' \\? ?)) and ? ->> 'type' = 'Create')",
 115               a.data,
 116               ^Pleroma.Constants.as_public(),
 117               a.data,
 118               ^Pleroma.Constants.as_public(),
 119               a.data
 120             )
 121           )
 122         else
 123           Pleroma.Activity
 124           |> join(:left, [a], b in Pleroma.Bookmark, on: a.id == b.activity_id)
 125           |> group_by([a], fragment("? ->> 'context'::text", a.data))
 126         end
 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))
 131
 132       Pleroma.Object
 133       |> where([o], fragment("? ->> 'context'::text", o.data) in subquery(deletable_context))
 134     else
 135       if Keyword.get(options, :keep_non_public) do
 136         Pleroma.Object
 137         |> where(
 138           [o],
 139           fragment(
 140             "?->'to' \\? ? OR ?->'cc' \\? ?",
 141             o.data,
 142             ^Pleroma.Constants.as_public(),
 143             o.data,
 144             ^Pleroma.Constants.as_public()
 145           )
 146         )
 147       else
 148         Pleroma.Object
 149       end
 150       |> where([o], o.updated_at < ^time_deadline)
 151       |> where(
 152         [o],
 153         fragment("split_part(?->>'actor', '/', 3) != ?", o.data, ^Pleroma.Web.Endpoint.host())
 154       )
 155     end
 156     |> Repo.delete_all(timeout: :infinity)
 157
 158     prune_hashtags_query = """
 159     DELETE FROM hashtags AS ht
 160     WHERE NOT EXISTS (
 161       SELECT 1 FROM hashtags_objects hto
 162       WHERE ht.id = hto.hashtag_id)
 163     """
 164
 165     Repo.query(prune_hashtags_query)
 166
 167     if Keyword.get(options, :vacuum) do
 168       Maintenance.vacuum("full")
 169     end
 170   end
 171
 172   def run(["prune_task"]) do
 173     start_pleroma()
 174
 175     nil
 176     |> Pleroma.Workers.Cron.PruneDatabaseWorker.perform()
 177   end
 178
 179   def run(["fix_likes_collections"]) do
 180     start_pleroma()
 181
 182     from(object in Object,
 183       where: fragment("(?)->>'likes' is not null", object.data),
 184       select: %{id: object.id, likes: fragment("(?)->>'likes'", object.data)}
 185     )
 186     |> Pleroma.Repo.chunk_stream(100, :batches)
 187     |> Stream.each(fn objects ->
 188       ids =
 189         objects
 190         |> Enum.filter(fn object -> object.likes |> Jason.decode!() |> is_map() end)
 191         |> Enum.map(& &1.id)
 192
 193       Object
 194       |> where([object], object.id in ^ids)
 195       |> update([object],
 196         set: [
 197           data:
 198             fragment(
 199               "safe_jsonb_set(?, '{likes}', '[]'::jsonb, true)",
 200               object.data
 201             )
 202         ]
 203       )
 204       |> Repo.update_all([], timeout: :infinity)
 205     end)
 206     |> Stream.run()
 207   end
 208
 209   def run(["vacuum", args]) do
 210     start_pleroma()
 211
 212     Maintenance.vacuum(args)
 213   end
 214
 215   def run(["ensure_expiration"]) do
 216     start_pleroma()
 217     days = Pleroma.Config.get([:mrf_activity_expiration, :days], 365)
 218
 219     Pleroma.Activity
 220     |> join(:inner, [a], o in Object,
 221       on:
 222         fragment(
 223           "(?->>'id') = COALESCE((?)->'object'->> 'id', (?)->>'object')",
 224           o.data,
 225           a.data,
 226           a.data
 227         )
 228     )
 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 ->
 235         expires_at =
 236           activity.inserted_at
 237           |> DateTime.from_naive!("Etc/UTC")
 238           |> Timex.shift(days: days)
 239
 240         Pleroma.Workers.PurgeExpiredActivity.enqueue(%{
 241           activity_id: activity.id,
 242           expires_at: expires_at
 243         })
 244       end)
 245     end)
 246     |> Stream.run()
 247   end
 248
 249   def run(["set_text_search_config", tsconfig]) do
 250     start_pleroma()
 251     %{rows: [[tsc]]} = Ecto.Adapters.SQL.query!(Pleroma.Repo, "SHOW default_text_search_config;")
 252     shell_info("Current default_text_search_config: #{tsc}")
 253
 254     %{rows: [[db]]} = Ecto.Adapters.SQL.query!(Pleroma.Repo, "SELECT current_database();")
 255     shell_info("Update default_text_search_config: #{tsconfig}")
 256
 257     %{messages: msg} =
 258       Ecto.Adapters.SQL.query!(
 259         Pleroma.Repo,
 260         "ALTER DATABASE #{db} SET default_text_search_config = '#{tsconfig}';"
 261       )
 262
 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)}")
 266     else
 267       rum_enabled = Pleroma.Config.get([:database, :rum_enabled])
 268       shell_info("Recreate index, RUM: #{rum_enabled}")
 269
 270       # Note SQL below needs to be kept up-to-date with latest GIN or RUM index definition in future
 271       if rum_enabled do
 272         Ecto.Adapters.SQL.query!(
 273           Pleroma.Repo,
 274           "CREATE OR REPLACE FUNCTION objects_fts_update() RETURNS trigger AS $$ BEGIN
 275           new.fts_content := to_tsvector(new.data->>'content');
 276           RETURN new;
 277           END
 278           $$ LANGUAGE plpgsql",
 279           [],
 280           timeout: :infinity
 281         )
 282
 283         shell_info("Refresh RUM index")
 284         Ecto.Adapters.SQL.query!(Pleroma.Repo, "UPDATE objects SET updated_at = NOW();")
 285       else
 286         Ecto.Adapters.SQL.query!(Pleroma.Repo, "DROP INDEX IF EXISTS objects_fts;")
 287
 288         Ecto.Adapters.SQL.query!(
 289           Pleroma.Repo,
 290           "CREATE INDEX CONCURRENTLY objects_fts ON objects USING gin(to_tsvector('#{tsconfig}', data->>'content')); ",
 291           [],
 292           timeout: :infinity
 293         )
 294       end
 295
 296       shell_info('Done.')
 297     end
 298   end
 299
 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?"
 305
 306     if shell_prompt(prompt, "n") in ~w(Yn Y y) do
 307       {_, result, _} =
 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)
 312
 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}"}
 317           else
 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)}"}
 321           end
 322         end)
 323
 324       shell_info(inspect(result))
 325     end
 326   end
 327 end