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