automated upload deletion through periodic garbage collection

lib/banchan/accounts/accounts.ex

@@ -20,7 +20,6 @@ defmodule Banchan.Accounts do
 
   alias Banchan.Repo
   alias Banchan.Uploads
-  alias Banchan.Uploads.Upload
   alias Banchan.Workers.{EnableUser, Thumbnailer}
 
   alias BanchanWeb.UserAuth
@@ -825,27 +824,6 @@ defmodule Banchan.Accounts do
       end,
       returning: true
     )
-    |> Ecto.Multi.run(:remove_old_pfp_img, fn _, %{updated_user: updated, user: old} ->
-      if old.pfp_img_id && old.pfp_img_id != updated.pfp_img_id do
-        Uploads.delete_upload(%Upload{id: old.pfp_img_id})
-      else
-        {:ok, nil}
-      end
-    end)
-    |> Ecto.Multi.run(:remove_old_pfp_thumb, fn _, %{updated_user: updated, user: old} ->
-      if old.pfp_thumb_id && old.pfp_thumb_id != updated.pfp_thumb_id do
-        Uploads.delete_upload(%Upload{id: old.pfp_thumb_id})
-      else
-        {:ok, nil}
-      end
-    end)
-    |> Ecto.Multi.run(:remove_old_header_img, fn _, %{updated_user: updated, user: old} ->
-      if old.header_img_id && old.header_img_id != updated.header_img_id do
-        Uploads.delete_upload(%Upload{id: old.header_img_id})
-      else
-        {:ok, nil}
-      end
-    end)
     |> Repo.transaction()
     |> case do
       {:ok, %{updated_user: user}} ->
@@ -1788,38 +1766,17 @@ defmodule Banchan.Accounts do
   def prune_users do
     now = NaiveDateTime.utc_now()
 
-    Repo.transaction(fn ->
-      from(
-        u in User,
-        where: not is_nil(u.deactivated_at),
-        where: u.deactivated_at < datetime_add(^now, -30, "day")
-      )
-      |> Repo.stream()
-      |> Enum.reduce(0, fn user, acc ->
-        # NB(@zkat): We hard match on `{:ok, _}` here because scheduling
-        # deletions should really never fail.
-        if user.pfp_img_id do
-          {:ok, _} = Uploads.delete_upload(%Upload{id: user.pfp_img_id})
-        end
-
-        if user.pfp_thumb_id do
-          {:ok, _} = Uploads.delete_upload(%Upload{id: user.pfp_thumb_id})
-        end
-
-        if user.header_img_id do
-          {:ok, _} = Uploads.delete_upload(%Upload{id: user.header_img_id})
-        end
-
-        case Repo.delete(user) do
-          {:ok, _} ->
-            acc + 1
-
-          {:error, error} ->
-            Logger.error("Failed to prune user #{user.handle}: #{inspect(error)}")
-            acc
-        end
+    {:ok, {count, _}} =
+      Repo.transaction(fn ->
+        from(
+          u in User,
+          where: not is_nil(u.deactivated_at),
+          where: u.deactivated_at < datetime_add(^now, -30, "day")
+        )
+        |> Repo.delete_all()
       end)
-    end)
+
+    {:ok, count}
   end
 
   @doc """

lib/banchan/offerings/offerings.ex

@@ -92,16 +92,7 @@ defmodule Banchan.Offerings do
             changeset
           end
 
-        drop_old_card_img =
-          if offering.card_img_id &&
-               offering.card_img_id != Ecto.Changeset.get_field(changeset, :card_img_id) do
-            Uploads.delete_upload(%Upload{id: offering.card_img_id})
-          else
-            {:ok, nil}
-          end
-
-        with {:ok, changed} <- changeset |> Repo.update(returning: true),
-             {:ok, _} <- drop_old_card_img do
+        with {:ok, changed} <- changeset |> Repo.update(returning: true) do
           if !open_before? && changed.open do
             Notifications.offering_opened(changed)
           end
@@ -810,23 +801,15 @@ defmodule Banchan.Offerings do
   def prune_offerings do
     now = NaiveDateTime.utc_now()
 
-    Repo.transaction(fn ->
+    {:ok, {count, _}} = Repo.transaction(fn ->
       from(
         o in Offering,
         where: not is_nil(o.deleted_at),
         where: o.deleted_at < datetime_add(^now, -30, "day")
       )
-      |> Repo.stream()
-      |> Enum.reduce(0, fn off, acc ->
-        case Repo.delete(off) do
-          {:ok, _} ->
-            acc + 1
-
-          {:error, error} ->
-            Logger.error("Failed to prune offering #{off.id}: #{inspect(error)}")
-            acc
-        end
-      end)
+      |> Repo.delete_all()
     end)
+
+    {:ok, count}
   end
 end

lib/banchan/studios/studios.ex

@@ -682,20 +682,6 @@ defmodule Banchan.Studios do
       end,
       returning: true
     )
-    |> Ecto.Multi.run(:remove_old_card_img, fn _, %{updated_studio: updated, studio: old} ->
-      if old.card_img_id && old.card_img_id != updated.card_img_id do
-        Uploads.delete_upload(%Upload{id: old.card_img_id})
-      else
-        {:ok, nil}
-      end
-    end)
-    |> Ecto.Multi.run(:remove_old_header_img, fn _, %{updated_studio: updated, studio: old} ->
-      if old.header_img_id && old.header_img_id != updated.header_img_id do
-        Uploads.delete_upload(%Upload{id: old.header_img_id})
-      else
-        {:ok, nil}
-      end
-    end)
     |> Repo.transaction()
     |> case do
       {:ok, %{updated_studio: studio}} ->
@@ -1155,24 +1141,17 @@ defmodule Banchan.Studios do
   def prune_studios do
     now = NaiveDateTime.utc_now()
 
-    Repo.transaction(fn ->
-      from(
-        s in Studio,
-        where: not is_nil(s.deleted_at),
-        where: s.deleted_at < datetime_add(^now, -30, "day")
-      )
-      |> Repo.stream()
-      |> Enum.reduce(0, fn studio, acc ->
-        case Repo.delete(studio) do
-          {:ok, _} ->
-            acc + 1
-
-          {:error, error} ->
-            Logger.error("Failed to prune studio #{studio.handle}: #{inspect(error)}")
-            acc
-        end
+    {:ok, {count, _}} =
+      Repo.transaction(fn ->
+        from(
+          s in Studio,
+          where: not is_nil(s.deleted_at),
+          where: s.deleted_at < datetime_add(^now, -30, "day")
+        )
+        |> Repo.delete_all()
       end)
-    end)
+
+    {:ok, count}
   end
 
   ## Misc utilities

lib/banchan/uploads/uploads.ex

@@ -160,6 +160,53 @@ defmodule Banchan.Uploads do
     end
   end
 
+  def prune_uploads() do
+    {:ok, {count, _}} =
+      Repo.transaction(fn ->
+        columns =
+          from(
+            tc in "table_constraints",
+            prefix: "information_schema",
+            join: kcu in "key_column_usage",
+            prefix: "information_schema",
+            on: tc.constraint_name == kcu.constraint_name,
+            join: ccu in "constraint_column_usage",
+            prefix: "information_schema",
+            on: ccu.constraint_name == tc.constraint_name,
+            where: tc.constraint_type == "FOREIGN KEY",
+            where: ccu.table_name == "uploads",
+            select: %{column: kcu.column_name, table: tc.table_name}
+          )
+          |> Repo.all()
+
+        sq =
+          Enum.reduce(columns, nil, fn %{column: column, table: table}, acc ->
+            q =
+              from(
+                u in Upload,
+                join: t in ^table,
+                on: field(t, ^String.to_atom(column)) == u.id,
+                select: u.id
+              )
+
+            if is_nil(acc) do
+              q
+            else
+              acc
+              |> union(^q)
+            end
+          end)
+
+          from(
+            u in Upload,
+            where: u.id not in subquery(sq)
+          )
+        |> Repo.delete_all()
+      end)
+
+    {:ok, count}
+  end
+
   ## Creation
 
   @doc """

lib/banchan/workers/pruner.ex

@@ -8,16 +8,21 @@ defmodule Banchan.Workers.Pruner do
     unique: [period: 60],
     tags: ["deletion", "pruning"]
 
-  alias Banchan.{Accounts, Offerings, Repo, Studios}
+  require Logger
+
+  alias Banchan.{Accounts, Offerings, Repo, Studios, Uploads}
 
   def perform(_) do
     Ecto.Multi.new()
-    |> Ecto.Multi.run(:prune_users, fn _, _ -> Accounts.prune_users() end)
-    |> Ecto.Multi.run(:prune_studios, fn _, _ -> Studios.prune_studios() end)
-    |> Ecto.Multi.run(:prune_offerings, fn _, _ -> Offerings.prune_offerings() end)
+    |> Ecto.Multi.run(:pruned_users, fn _, _ -> Accounts.prune_users() end)
+    |> Ecto.Multi.run(:pruned_studios, fn _, _ -> Studios.prune_studios() end)
+    |> Ecto.Multi.run(:pruned_offerings, fn _, _ -> Offerings.prune_offerings() end)
+    |> Ecto.Multi.run(:pruned_uploads, fn _, _ -> Uploads.prune_uploads() end)
     |> Repo.transaction()
     |> case do
-      {:ok, _} -> :ok
+      {:ok, stats} ->
+        Logger.info("Prune job completed successfully: #{inspect(stats)}")
+        :ok
       {:error, _, error, _} -> {:error, error}
     end
   end

lib/banchan/workers/upload_deleter.ex

@@ -3,24 +3,24 @@ defmodule Banchan.Workers.UploadDeleter do
   Worker responsible for deleting uploads and cleaning up their associated backing data.
   """
   use Oban.Worker,
-    queue: :upload_cleanup,
+    queue: :pruning,
     unique: [period: 60],
-    tags: ["media", "cleanup", "uploads"]
+    max_attempts: 3,
+    tags: ["media", "pruning", "uploads"]
 
-  alias Banchan.Repo
   alias Banchan.Uploads
   alias Banchan.Uploads.Upload
 
   @impl Oban.Worker
   def perform(%_{args: %{"id" => id, "bucket" => bucket, "key" => key}}) do
-    Uploads.delete_data!(%Upload{id: id, bucket: bucket, key: key})
+    if is_nil(Uploads.get_by_id(id)) do
+      Uploads.delete_data!(%Upload{id: id, bucket: bucket, key: key})
+    end
 
     :ok
   end
 
   def queue_data_deletion(%Upload{} = upload) do
-    upload = Repo.reload(upload)
-
     __MODULE__.new(%{
       "id" => upload.id,
       "bucket" => upload.bucket,