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"]) do
 132     import Ecto.Query
 133
 134     start_pleroma()
 135
 136     from(
 137       o in Object,
 138       where: fragment("(?)->>'hashtags' is null", o.data),
 139       where: fragment("(?)->>'tag' != '[]'", o.data),
 140       select: %{id: o.id, tag: fragment("(?)->>'tag'", o.data)},
 141       order_by: [:desc, o.id]
 142     )
 143     |> Pleroma.Repo.chunk_stream(200, :batches)
 144     |> Stream.each(fn objects ->
 145       Repo.transaction(fn ->
 146         objects_first = objects |> List.first()
 147         objects_last = objects |> List.last()
 148
 149         Logger.info(
 150           "fill_old_hashtags: #{objects_first.id} (#{objects_first.inserted_at}) -- #{
 151             objects_last.id
 152           } (#{objects_last.inserted_at})"
 153         )
 154
 155         objects
 156         |> Enum.map(fn object ->
 157           tags =
 158             object.tag
 159             |> Jason.decode!()
 160             |> Enum.filter(&is_bitstring(&1))
 161
 162           Object
 163           |> where([o], o.id == ^object.id)
 164           |> update([o],
 165             set: [data: fragment("safe_jsonb_set(?, '{hashtags}', ?, true)", o.data, ^tags)]
 166           )
 167           |> Repo.update_all([], timeout: :infinity)
 168         end)
 169       end)
 170     end)
 171     |> Stream.run()
 172   end
 173
 174   def run(["vacuum", args]) do
 175     start_pleroma()
 176
 177     Maintenance.vacuum(args)
 178   end
 179
 180   def run(["ensure_expiration"]) do
 181     start_pleroma()
 182     days = Pleroma.Config.get([:mrf_activity_expiration, :days], 365)
 183
 184     Pleroma.Activity
 185     |> join(:inner, [a], o in Object,
 186       on:
 187         fragment(
 188           "(?->>'id') = COALESCE((?)->'object'->> 'id', (?)->>'object')",
 189           o.data,
 190           a.data,
 191           a.data
 192         )
 193     )
 194     |> where(local: true)
 195     |> where([a], fragment("(? ->> 'type'::text) = 'Create'", a.data))
 196     |> where([_a, o], fragment("?->>'type' = 'Note'", o.data))
 197     |> Pleroma.Repo.chunk_stream(100, :batches)
 198     |> Stream.each(fn activities ->
 199       Enum.each(activities, fn activity ->
 200         expires_at =
 201           activity.inserted_at
 202           |> DateTime.from_naive!("Etc/UTC")
 203           |> Timex.shift(days: days)
 204
 205         Pleroma.Workers.PurgeExpiredActivity.enqueue(%{
 206           activity_id: activity.id,
 207           expires_at: expires_at
 208         })
 209       end)
 210     end)
 211     |> Stream.run()
 212   end
 213 end