Post-refactoring and improving of the new batch tags update flow

docker-compose.yml

@@ -20,6 +20,8 @@ services:
       # be a bare `level` which will be used as a catch-all for all other log
       # events that do not match any of the previous filters.
       - PHILOMENA_LOG=${PHILOMENA_LOG-Ecto=debug,Exq=none,PhilomenaMedia.Objects=info,debug}
+      # Set this env var to `y` to regenerate the test snapshots
+      - ASSERT_VALUE_ACCEPT_DIFFS
       - MIX_ENV=dev
       - PGPASSWORD=postgres
       - ANONYMOUS_NAME_SALT=2fmJRo0OgMFe65kyAJBxPT0QtkVes/jnKDdtP21fexsRqiw8TlSY7yO+uFyMZycp
@@ -63,6 +65,7 @@ services:
       - postgres
       - opensearch
       - valkey
+      - files
     ports:
       - '5173:5173'
 
@@ -72,6 +75,12 @@ services:
       - POSTGRES_PASSWORD=postgres
     volumes:
       - postgres_data:/var/lib/postgresql/data
+
+    # Expose DB port for connecting from a DB viewer client. `POSTGRES_PORT`
+    # variable can be used to prevent conflicts with other postgres instances
+    # potentially already running on the host.
+    ports: ['${POSTGRES_PORT:-5432}:5432']
+
     attach: false
 
   opensearch:

lib/philomena/images.ex

@@ -37,6 +37,7 @@ defmodule Philomena.Images do
   alias Philomena.Galleries.Gallery
   alias Philomena.Galleries.Interaction
   alias Philomena.Users.User
+  alias Philomena.Users
 
   use Philomena.Subscriptions,
     on_delete: :clear_image_notification,
@@ -69,6 +70,16 @@ defmodule Philomena.Images do
     |> Enum.map_join(", ", & &1.name)
   end
 
+  @typedoc """
+    Result of the `create_image/3` function. The image was created in a DB but
+    an upload process is probably still running in the background with its PID
+    given in the `upload_pid` field.
+  """
+  @type image_upload :: %{
+          image: Image,
+          upload_pid: pid
+        }
+
   @doc """
   Creates a image.
 
@@ -81,6 +92,8 @@ defmodule Philomena.Images do
       {:error, %Ecto.Changeset{}}
 
   """
+  @spec create_image(Users.principal(), %{String.t() => any()}) ::
+          {:ok, image_upload()} | {:error, any()}
   def create_image(attribution, attrs \\ %{}) do
     tags = Tags.get_or_create_tags(attrs["tag_input"])
     sources = attrs["sources"]
@@ -1074,13 +1087,11 @@ defmodule Philomena.Images do
 
     to_insert =
       Enum.flat_map(changes, fn change ->
-        Enum.map(change.added_tags, &%{tag_id: &1.id, image_id: change.image_id})
+        Enum.map(change.added_tag_ids, &%{tag_id: &1, image_id: change.image_id})
       end)
 
     to_delete_ids =
-      Enum.flat_map(changes, fn change ->
-        Enum.map(change.removed_tags, & &1.id)
-      end)
+      Enum.flat_map(changes, & &1.removed_tag_ids)
 
     to_delete =
       Tagging
@@ -1143,7 +1154,10 @@ defmodule Philomena.Images do
       {:ok, _} = result ->
         reindex_images(image_ids)
         Comments.reindex_comments_on_images(image_ids)
-        Tags.reindex_tags(Enum.flat_map(changes, &(&1.added_tags ++ &1.removed_tags)))
+
+        changes
+        |> Enum.flat_map(&(&1.added_tag_ids ++ &1.removed_tag_ids))
+        |> Tags.reindex_tags_by_ids()
 
         result
 
@@ -1153,7 +1167,7 @@ defmodule Philomena.Images do
   end
 
   # Merge any change batches belonging to the same image ID into
-  # one single batch, then deduplicate added_tags by removing any
+  # one single batch, then deduplicate added_tag_ids by removing any
   # which are slated for removal, which is the behavior of the
   # mass tagger anyway (it inserts anything that needs to be inserted
   # into image_taggings, and then deletes anything that needs to be deleted,
@@ -1166,21 +1180,21 @@ defmodule Philomena.Images do
     |> Enum.map(fn {image_id, instances} ->
       added =
         instances
-        |> Enum.flat_map(& &1.added_tags)
-        |> Enum.uniq_by(& &1.id)
+        |> Enum.flat_map(& &1.added_tag_ids)
+        |> Enum.uniq()
 
       removed =
         instances
-        |> Enum.flat_map(& &1.removed_tags)
-        |> Enum.uniq_by(& &1.id)
+        |> Enum.flat_map(& &1.removed_tag_ids)
+        |> Enum.uniq()
 
       %{
         image_id: image_id,
-        added_tags: Enum.reject(added, fn a -> Enum.any?(removed, &(&1.id == a.id)) end),
-        removed_tags: removed
+        added_tag_ids: Enum.reject(added, fn a -> Enum.member?(removed, a) end),
+        removed_tag_ids: removed
       }
     end)
-    |> Enum.reject(&(Enum.empty?(&1.added_tags) && Enum.empty?(&1.removed_tags)))
+    |> Enum.reject(&(Enum.empty?(&1.added_tag_ids) && Enum.empty?(&1.removed_tag_ids)))
   end
 
   # Generate data for TagChanges.Tag struct.

lib/philomena/tag_changes.ex

@@ -12,52 +12,139 @@ defmodule Philomena.TagChanges do
   alias Philomena.Images
   alias Philomena.Images.Image
   alias Philomena.Tags.Tag
+  alias Philomena.Users
 
-  # Accepts a list of TagChanges.TagChange IDs.
-  def mass_revert(ids, attributes) do
-    tag_changes =
+  require Logger
+
+  @typedoc """
+  In the successful case returns the `TagChange`s that were loaded and affected
+  plus a non-negative integer with the number of tags affected by the revert.
+  """
+  @type mass_revert_result ::
+          {:ok, [TagChange.t()], non_neg_integer()}
+          | {:error, any()}
+
+  @type tag_change_id :: integer()
+  @type tag_id :: integer()
+
+  @typedoc """
+  A tuple with a composite identifier for a `TagChange.Tag`.
+  """
+  @type tag_change_tag_id :: {tag_change_id(), tag_id()}
+
+  # Accepts a list of `TagChanges.TagChange` IDs.
+  @spec mass_revert([tag_change_id()], Users.principal()) :: mass_revert_result()
+  def mass_revert(tag_change_ids, principal) do
+    tag_change_ids
+    |> Map.new(&{&1, nil})
+    |> mass_revert_impl(principal)
+  end
+
+  # Accepts a list of `TagChanges.Tag` IDs.
+  @spec mass_revert_tags([tag_change_tag_id()], Users.principal()) :: mass_revert_result()
+  def mass_revert_tags(tag_change_tag_ids, principal) do
+    tag_change_tag_ids
+    |> Enum.group_by(&elem(&1, 0), &elem(&1, 1))
+    |> mass_revert_impl(principal)
+  end
+
+  @spec mass_revert_impl(%{tag_change_id() => [tag_id()] | nil}, Users.principal()) ::
+          mass_revert_result()
+  defp mass_revert_impl(input, principal) do
+    # Load the requested tag changes from the DB. `nil` in the position of the
+    # tag IDs list means we want to revert all tags for that tag change.
+    input =
+      input
+      |> Enum.map(fn {tc_id, tag_ids} ->
+        %{
+          tc_id: tc_id,
+          tag_ids: if(not is_nil(tag_ids), do: tag_ids |> Enum.uniq())
+        }
+      end)
+
+    tags_by_image =
       Repo.all(
         from tc in TagChange,
-          inner_join: i in assoc(tc, :image),
-          where: tc.id in ^ids and i.hidden_from_users == false,
-          order_by: [desc: :created_at],
-          preload: [tags: [:tag, :tag_change]]
+          inner_join:
+            input in fragment(
+              "SELECT * FROM jsonb_to_recordset(?) AS (tc_id int, tag_ids int[])",
+              ^input
+            ),
+          on: tc.id == input.tc_id,
+          inner_join: tct in TagChanges.Tag,
+          on: tc.id == tct.tag_change_id,
+          where: is_nil(input.tag_ids) or tct.tag_id in input.tag_ids,
+          group_by: [tc.image_id, tct.tag_id],
+          # Get min/max rows without subqueries (don't repeat this at home).
+          # Retain only changes that dont cancel themselves out already i.e. we
+          # want to ignore change sequences where the first change adds the tag
+          # and the last change removes it or vice versa, because they are
+          # already self-cancelling.
+          having:
+            fragment("(array_agg(? ORDER BY ? ASC))[1]", tct.added, tc.id) ==
+              fragment("(array_agg(? ORDER BY ? DESC))[1]", tct.added, tc.id),
+          select: %{
+            image_id: tc.image_id,
+            tag_id: tct.tag_id,
+            last_change:
+              fragment("(array_agg(row (?, ?) ORDER BY ? DESC))[1]", tc.id, tct.added, tc.id)
+          }
       )
+      |> Enum.group_by(& &1.image_id)
+
+    # Load latest tag change IDs for each {image, tags} we are interested in.
+    # We'll use this to filter out tag changes that are non-current, meaning
+    # reverting them would not be useful as they were already overwritten by
+    # some other change not in the list of reverted changes.
+    input =
+      tags_by_image
+      |> Enum.map(fn {image_id, tags} ->
+        %{
+          image_id: image_id,
+          tag_ids: tags |> Enum.map(& &1.tag_id)
+        }
+      end)
 
-    case mass_revert_tags(Enum.flat_map(tag_changes, & &1.tags), attributes) do
-      {:ok, _result} ->
-        {:ok, tag_changes}
-
-      error ->
-        error
-    end
-  end
+    latest_tc_ids =
+      Repo.all(
+        from tc in TagChange,
+          inner_join:
+            input in fragment(
+              "SELECT * FROM jsonb_to_recordset(?) AS (image_id int, tag_ids int[])",
+              ^input
+            ),
+          on: tc.image_id == input.image_id,
+          inner_join: tct in TagChanges.Tag,
+          on: tc.id == tct.tag_change_id and tct.tag_id in input.tag_ids,
+          group_by: [tc.image_id, tct.tag_id],
+          select: max(tc.id),
+          # It's possible for the same tag change to cover the latest versions
+          # of several tags, so deduplicate the results.
+          distinct: true
+      )
+      |> MapSet.new()
 
-  # Accepts a list of TagChanges.Tag objects with tag_change and tag relations preloaded.
-  def mass_revert_tags(tags, attributes) do
-    # Sort tags by tag change creation date, then uniq them by tag ID
-    # to keep the first, aka the latest, record. Then prepare the struct
-    # for the batch updater.
-    changes_per_image =
-      tags
-      |> Enum.group_by(& &1.tag_change.image_id)
-      |> Enum.map(fn {image_id, instances} ->
-        changed_tags =
-          instances
-          |> Enum.sort_by(& &1.tag_change.created_at, :desc)
-          |> Enum.uniq_by(& &1.tag_id)
-
-        {added_tags, removed_tags} = Enum.split_with(changed_tags, & &1.added)
+    # Calculate the revert operations for each image.
+    reverts_per_image =
+      tags_by_image
+      |> Enum.map(fn {image_id, tags} ->
+        {added_tags, removed_tags} =
+          tags
+          |> Enum.filter(fn %{last_change: {tc_id, _}} -> tc_id in latest_tc_ids end)
+          |> Enum.split_with(fn %{last_change: {_, added}} -> added end)
 
         # We send removed tags to be added, and added to be removed. That's how reverting works!
         %{
           image_id: image_id,
-          added_tags: Enum.map(removed_tags, & &1.tag),
-          removed_tags: Enum.map(added_tags, & &1.tag)
+          added_tag_ids: Enum.map(removed_tags, & &1.tag_id),
+          removed_tag_ids: Enum.map(added_tags, & &1.tag_id)
         }
       end)
+      |> Enum.reject(&(&1.added_tag_ids == [] and &1.removed_tag_ids == []))
 
-    Images.batch_update(changes_per_image, attributes)
+    with {:ok, {total_tags_affected, _}} <- Images.batch_update(reverts_per_image, principal) do
+      {:ok, total_tags_affected}
+    end
   end
 
   def full_revert(%{user_id: _user_id, attributes: _attributes} = params),
@@ -149,7 +236,7 @@ defmodule Philomena.TagChanges do
       query
       |> preload([:user, image: [:user, :sources, tags: :aliases], tags: [:tag]])
       |> group_by([tc], tc.id)
-      |> order_by(desc: :created_at)
+      |> order_by(desc: :created_at, desc: :id)
 
     {Repo.paginate(query, pagination), item_count}
   end

lib/philomena/tags.ex

@@ -689,11 +689,17 @@ defmodule Philomena.Tags do
 
   """
   def reindex_tags(tags) do
-    Exq.enqueue(Exq, "indexing", IndexWorker, ["Tags", "id", Enum.map(tags, & &1.id)])
+    tags
+    |> Enum.map(& &1.id)
+    |> reindex_tags_by_ids()
 
     tags
   end
 
+  def reindex_tags_by_ids(tag_ids) do
+    Exq.enqueue(Exq, "indexing", IndexWorker, ["Tags", "id", tag_ids])
+  end
+
   @doc """
   Returns the list of associations to preload for tag indexing.
 

lib/philomena/users.ex

@@ -21,6 +21,16 @@ defmodule Philomena.Users do
   alias Philomena.UserEraseWorker
   alias Philomena.UserRenameWorker
 
+  @typedoc """
+  Describes the entity performing the action.
+  The term `principal` was borrowed from AWS IAM terminology.
+  """
+  @type principal :: [
+          ip: EctoNetwork.INET.t(),
+          fingerprint: String.t(),
+          user: User.t() | nil
+        ]
+
   ## Database getters
 
   @doc """

lib/philomena_web/controllers/tag_change/revert_controller.ex

@@ -17,12 +17,20 @@ defmodule PhilomenaWeb.TagChange.RevertController do
     }
 
     case TagChanges.mass_revert(ids, attributes) do
-      {:ok, tag_changes} ->
+      {:ok, total_tags_affected} ->
         conn
-        |> put_flash(:info, "Successfully reverted #{length(tag_changes)} tag changes.")
+        |> put_flash(
+          :info,
+          if total_tags_affected == 0 do
+            "The revert of #{length(ids)} changes resulted in no effective tag updates."
+          else
+            "Successfully reverted #{length(ids)} tag changes with " <>
+              "#{total_tags_affected} tags effectively updated."
+          end
+        )
         |> moderation_log(
           details: &log_details/2,
-          data: %{user: conn.assigns.current_user, count: length(tag_changes)}
+          data: %{user: conn.assigns.current_user, count: length(ids)}
         )
         |> redirect(external: conn.assigns.referrer)
 
@@ -33,6 +41,13 @@ defmodule PhilomenaWeb.TagChange.RevertController do
     end
   end
 
+  # Handles the case where no tag changes were selected for submission at all.
+  def create(conn, _payload) do
+    conn
+    |> put_flash(:error, "No tag changes selected.")
+    |> redirect(external: conn.assigns.referrer)
+  end
+
   defp verify_authorized(conn, _params) do
     if Canada.Can.can?(conn.assigns.current_user, :revert, TagChange) do
       conn