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

   1 # Pleroma: A lightweight social networking server
   2 # Copyright © 2017-2020 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   require Logger
  12   require Pleroma.Constants
  13   import Ecto.Query
  14   import Mix.Pleroma
  15   use Mix.Task
  16
  17   @shortdoc "A collection of database related tasks"
  18   @moduledoc File.read!("docs/administration/CLI_tasks/database.md")
  19
  20   def run(["remove_embedded_objects" | args]) do
  21     {options, [], []} =
  22       OptionParser.parse(
  23         args,
  24         strict: [
  25           vacuum: :boolean
  26         ]
  27       )
  28
  29     start_pleroma()
  30     Logger.info("Removing embedded objects")
  31
  32     Repo.query!(
  33       "update activities set data = safe_jsonb_set(data, '{object}'::text[], data->'object'->'id') where data->'object'->>'id' is not null;",
  34       [],
  35       timeout: :infinity
  36     )
  37
  38     if Keyword.get(options, :vacuum) do
  39       Maintenance.vacuum("full")
  40     end
  41   end
  42
  43   def run(["bump_all_conversations"]) do
  44     start_pleroma()
  45     Conversation.bump_for_all_activities()
  46   end
  47
  48   def run(["update_users_following_followers_counts"]) do
  49     start_pleroma()
  50
  51     Repo.transaction(
  52       fn ->
  53         from(u in User, select: u)
  54         |> Repo.stream()
  55         |> Stream.each(&User.update_follower_count/1)
  56         |> Stream.run()
  57       end,
  58       timeout: :infinity
  59     )
  60   end
  61
  62   def run(["prune_objects" | args]) do
  63     {options, [], []} =
  64       OptionParser.parse(
  65         args,
  66         strict: [
  67           vacuum: :boolean
  68         ]
  69       )
  70
  71     start_pleroma()
  72
  73     deadline = Pleroma.Config.get([:instance, :remote_post_retention_days])
  74
  75     Logger.info("Pruning objects older than #{deadline} days")
  76
  77     time_deadline =
  78       NaiveDateTime.utc_now()
  79       |> NaiveDateTime.add(-(deadline * 86_400))
  80
  81     from(o in Object,
  82       where:
  83         fragment(
  84           "?->'to' \\? ? OR ?->'cc' \\? ?",
  85           o.data,
  86           ^Pleroma.Constants.as_public(),
  87           o.data,
  88           ^Pleroma.Constants.as_public()
  89         ),
  90       where: o.inserted_at < ^time_deadline,
  91       where:
  92         fragment("split_part(?->>'actor', '/', 3) != ?", o.data, ^Pleroma.Web.Endpoint.host())
  93     )
  94     |> Repo.delete_all(timeout: :infinity)
  95
  96     if Keyword.get(options, :vacuum) do
  97       Maintenance.vacuum("full")
  98     end
  99   end
 100
 101   def run(["fix_likes_collections"]) do
 102     start_pleroma()
 103
 104     from(object in Object,
 105       where: fragment("(?)->>'likes' is not null", object.data),
 106       select: %{id: object.id, likes: fragment("(?)->>'likes'", object.data)}
 107     )
 108     |> Pleroma.Repo.chunk_stream(100, :batches)
 109     |> Stream.each(fn objects ->
 110       ids =
 111         objects
 112         |> Enum.filter(fn object -> object.likes |> Jason.decode!() |> is_map() end)
 113         |> Enum.map(& &1.id)
 114
 115       Object
 116       |> where([object], object.id in ^ids)
 117       |> update([object],
 118         set: [
 119           data:
 120             fragment(
 121               "safe_jsonb_set(?, '{likes}', '[]'::jsonb, true)",
 122               object.data
 123             )
 124         ]
 125       )
 126       |> Repo.update_all([], timeout: :infinity)
 127     end)
 128     |> Stream.run()
 129   end
 130
 131   def run(["fill_old_hashtags", month_limit]) do
 132     import Ecto.Query
 133
 134     start_pleroma()
 135
 136     month_limit = String.to_integer(month_limit)
 137
 138     if month_limit < 1 do
 139       shell_error("Invalid `month_limit` argument, needs to be greater than 1")
 140     else
 141       time_limit = DateTime.utc_now() |> Timex.shift(months: -month_limit)
 142
 143       from(
 144         o in Object,
 145         where: fragment("(?)->>'hashtags' is null", o.data),
 146         where: fragment("(?)->>'tag' != '[]'", o.data),
 147         where: o.inserted_at < ^time_limit,
 148         select: %{id: o.id, tag: fragment("(?)->>'tag'", o.data)}
 149       )
 150       |> Pleroma.Repo.chunk_stream(200, :batches)
 151       |> Stream.each(fn objects ->
 152         Repo.transaction(fn ->
 153           objects_first = objects |> List.first()
 154           objects_last = objects |> List.last()
 155
 156           Logger.info(
 157             "fill_old_hashtags: #{objects_first.id} (#{objects_first.inserted_at}) -- #{
 158               objects_last.id
 159             } (#{objects_last.inserted_at})"
 160           )
 161
 162           objects
 163           |> Enum.map(fn object ->
 164             tags =
 165               object.tag
 166               |> Jason.decode!()
 167               |> Enum.filter(&is_bitstring(&1))
 168
 169             Object
 170             |> where([o], o.id == ^object.id)
 171             |> update([o],
 172               set: [data: fragment("safe_jsonb_set(?, '{hashtags}', ?, true)", o.data, ^tags)]
 173             )
 174             |> Repo.update_all([], timeout: :infinity)
 175           end)
 176         end)
 177       end)
 178       |> Stream.run()
 179     end
 180   end
 181
 182   def run(["vacuum", args]) do
 183     start_pleroma()
 184
 185     Maintenance.vacuum(args)
 186   end
 187
 188   def run(["ensure_expiration"]) do
 189     start_pleroma()
 190     days = Pleroma.Config.get([:mrf_activity_expiration, :days], 365)
 191
 192     Pleroma.Activity
 193     |> join(:inner, [a], o in Object,
 194       on:
 195         fragment(
 196           "(?->>'id') = COALESCE((?)->'object'->> 'id', (?)->>'object')",
 197           o.data,
 198           a.data,
 199           a.data
 200         )
 201     )
 202     |> where(local: true)
 203     |> where([a], fragment("(? ->> 'type'::text) = 'Create'", a.data))
 204     |> where([_a, o], fragment("?->>'type' = 'Note'", o.data))
 205     |> Pleroma.Repo.chunk_stream(100, :batches)
 206     |> Stream.each(fn activities ->
 207       Enum.each(activities, fn activity ->
 208         expires_at =
 209           activity.inserted_at
 210           |> DateTime.from_naive!("Etc/UTC")
 211           |> Timex.shift(days: days)
 212
 213         Pleroma.Workers.PurgeExpiredActivity.enqueue(%{
 214           activity_id: activity.id,
 215           expires_at: expires_at
 216         })
 217       end)
 218     end)
 219     |> Stream.run()
 220   end
 221 end