e7f4b67a463d5b8b1dd540901b0db47e853d0a3b
[akkoma] / 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