Fix buckets for query timing
[akkoma] / lib / pleroma / workers / attachments_cleanup_worker.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 Pleroma.Workers.AttachmentsCleanupWorker do
6 import Ecto.Query
7
8 alias Pleroma.Object
9 alias Pleroma.Repo
10
11 use Pleroma.Workers.WorkerHelper, queue: "attachments_cleanup"
12
13 @impl Oban.Worker
14 def perform(%Job{
15 args: %{
16 "op" => "cleanup_attachments",
17 "object" => %{"data" => %{"attachment" => [_ | _] = attachments, "actor" => actor}}
18 }
19 }) do
20 if Pleroma.Config.get([:instance, :cleanup_attachments], false) do
21 attachments
22 |> Enum.flat_map(fn item -> Enum.map(item["url"], & &1["href"]) end)
23 |> fetch_objects
24 |> prepare_objects(actor, Enum.map(attachments, & &1["name"]))
25 |> filter_objects
26 |> do_clean
27 end
28
29 {:ok, :success}
30 end
31
32 def perform(%Job{args: %{"op" => "cleanup_attachments", "object" => _object}}), do: {:ok, :skip}
33
34 defp do_clean({object_ids, attachment_urls}) do
35 uploader = Pleroma.Config.get([Pleroma.Upload, :uploader])
36
37 base_url =
38 String.trim_trailing(
39 Pleroma.Upload.base_url(),
40 "/"
41 )
42
43 Enum.each(attachment_urls, fn href ->
44 href
45 |> String.trim_leading("#{base_url}")
46 |> uploader.delete_file()
47 end)
48
49 delete_objects(object_ids)
50 end
51
52 defp delete_objects([_ | _] = object_ids) do
53 Repo.delete_all(from(o in Object, where: o.id in ^object_ids))
54 end
55
56 defp delete_objects(_), do: :ok
57
58 # we should delete 1 object for any given attachment, but don't delete
59 # files if there are more than 1 object for it
60 defp filter_objects(objects) do
61 Enum.reduce(objects, {[], []}, fn {href, %{id: id, count: count}}, {ids, hrefs} ->
62 with 1 <- count do
63 {ids ++ [id], hrefs ++ [href]}
64 else
65 _ -> {ids ++ [id], hrefs}
66 end
67 end)
68 end
69
70 defp prepare_objects(objects, actor, names) do
71 objects
72 |> Enum.reduce(%{}, fn %{
73 id: id,
74 data: %{
75 "url" => [%{"href" => href}],
76 "actor" => obj_actor,
77 "name" => name
78 }
79 },
80 acc ->
81 Map.update(acc, href, %{id: id, count: 1}, fn val ->
82 case obj_actor == actor and name in names do
83 true ->
84 # set id of the actor's object that will be deleted
85 %{val | id: id, count: val.count + 1}
86
87 false ->
88 # another actor's object, just increase count to not delete file
89 %{val | count: val.count + 1}
90 end
91 end)
92 end)
93 end
94
95 defp fetch_objects(hrefs) do
96 from(o in Object,
97 where:
98 fragment(
99 "to_jsonb(array(select jsonb_array_elements((?)#>'{url}') ->> 'href' where jsonb_typeof((?)#>'{url}') = 'array'))::jsonb \\?| (?)",
100 o.data,
101 o.data,
102 ^hrefs
103 )
104 )
105 # The query above can be time consumptive on large instances until we
106 # refactor how uploads are stored
107 |> Repo.all(timeout: :infinity)
108 end
109 end