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