[#3213] transfer_hashtags mix task refactoring.

lib/mix/tasks/pleroma/database.ex

@@ -132,74 +132,6 @@ def run(["fix_likes_collections"]) do
     |> Stream.run()
   end
 
-  def run(["transfer_hashtags"]) do
-    import Ecto.Query
-
-    start_pleroma()
-
-    Logger.info("Starting transferring object embedded hashtags to `hashtags` table...")
-
-    # Note: most objects have Mention-type AS2 tags and no hashtags (but we can't filter them out)
-    from(
-      object in Object,
-      left_join: hashtag in assoc(object, :hashtags),
-      where: is_nil(hashtag.id),
-      where: fragment("(?)->>'tag' != '[]'", object.data),
-      select: %{
-        id: object.id,
-        tag: fragment("(?)->>'tag'", object.data)
-      }
-    )
-    |> Repo.chunk_stream(100, :batches, timeout: :infinity)
-    |> Stream.each(fn objects ->
-      Logger.info("Processing #{length(objects)} objects starting from id #{hd(objects).id}...")
-
-      failed_ids =
-        objects
-        |> Enum.map(fn object ->
-          hashtags = Object.object_data_hashtags(%{"tag" => Jason.decode!(object.tag)})
-
-          Repo.transaction(fn ->
-            with {:ok, hashtag_records} <- Hashtag.get_or_create_by_names(hashtags) do
-              for hashtag_record <- hashtag_records do
-                with {:ok, _} <-
-                       Repo.query(
-                         "insert into hashtags_objects(hashtag_id, object_id) values ($1, $2);",
-                         [hashtag_record.id, object.id]
-                       ) do
-                  nil
-                else
-                  {:error, e} ->
-                    error =
-                      "ERROR: could not link object #{object.id} and hashtag " <>
-                        "#{hashtag_record.id}: #{inspect(e)}"
-
-                    Logger.error(error)
-                    Repo.rollback(object.id)
-                end
-              end
-
-              object.id
-            else
-              e ->
-                error = "ERROR: could not create hashtags for object #{object.id}: #{inspect(e)}"
-                Logger.error(error)
-                Repo.rollback(object.id)
-            end
-          end)
-        end)
-        |> Enum.filter(&(elem(&1, 0) == :error))
-        |> Enum.map(&elem(&1, 1))
-
-      if Enum.any?(failed_ids) do
-        Logger.error("ERROR: transfer_hashtags iteration failed for ids: #{inspect(failed_ids)}")
-      end
-    end)
-    |> Stream.run()
-
-    Logger.info("Done transferring hashtags. Please check logs to ensure no errors.")
-  end
-
   def run(["vacuum", args]) do
     start_pleroma()
 
@@ -239,4 +171,63 @@ def run(["ensure_expiration"]) do
     end)
     |> Stream.run()
   end
+
+  def run(["transfer_hashtags"]) do
+    import Ecto.Query
+
+    start_pleroma()
+
+    Logger.info("Starting transferring object embedded hashtags to `hashtags` table...")
+
+    # Note: most objects have Mention-type AS2 tags and no hashtags (but we can't filter them out)
+    from(
+      object in Object,
+      left_join: hashtag in assoc(object, :hashtags),
+      where: is_nil(hashtag.id),
+      where:
+        fragment("(?)->'tag' IS NOT NULL AND (?)->'tag' != '[]'::jsonb", object.data, object.data),
+      select: %{
+        id: object.id,
+        tag: fragment("(?)->'tag'", object.data)
+      }
+    )
+    |> Repo.chunk_stream(100, :one, timeout: :infinity)
+    |> Stream.each(&transfer_object_hashtags(&1))
+    |> Stream.run()
+
+    Logger.info("Done transferring hashtags. Please check logs to ensure no errors.")
+  end
+
+  defp transfer_object_hashtags(object) do
+    hashtags = Object.object_data_hashtags(%{"tag" => object.tag})
+
+    Repo.transaction(fn ->
+      with {:ok, hashtag_records} <- Hashtag.get_or_create_by_names(hashtags) do
+        for hashtag_record <- hashtag_records do
+          with {:ok, _} <-
+                 Repo.query(
+                   "insert into hashtags_objects(hashtag_id, object_id) values ($1, $2);",
+                   [hashtag_record.id, object.id]
+                 ) do
+            nil
+          else
+            {:error, e} ->
+              error =
+                "ERROR: could not link object #{object.id} and hashtag " <>
+                  "#{hashtag_record.id}: #{inspect(e)}"
+
+              Logger.error(error)
+              Repo.rollback(object.id)
+          end
+        end
+
+        object.id
+      else
+        e ->
+          error = "ERROR: could not create hashtags for object #{object.id}: #{inspect(e)}"
+          Logger.error(error)
+          Repo.rollback(object.id)
+      end
+    end)
+  end
 end