X-Git-Url: http://git.squeep.com/?a=blobdiff_plain;f=lib%2Fmix%2Ftasks%2Fpleroma%2Fdatabase.ex;h=093c7dd30de565c61fafcdf5839497daafa33820;hb=4134abef63e1165f5701741c1012e64cb908654c;hp=7f1108dcfa87cee488266a2cc1343be7824ece75;hpb=55562ca9362d66553ea3638c91174bbeb6c637f1;p=akkoma

diff --git a/lib/mix/tasks/pleroma/database.ex b/lib/mix/tasks/pleroma/database.ex
index 7f1108dcf..093c7dd30 100644
--- a/lib/mix/tasks/pleroma/database.ex
+++ b/lib/mix/tasks/pleroma/database.ex
@@ -4,14 +4,18 @@
 
 defmodule Mix.Tasks.Pleroma.Database do
   alias Pleroma.Conversation
+  alias Pleroma.Hashtag
   alias Pleroma.Maintenance
   alias Pleroma.Object
   alias Pleroma.Repo
   alias Pleroma.User
+
   require Logger
   require Pleroma.Constants
+
   import Ecto.Query
   import Mix.Pleroma
+
   use Mix.Task
 
   @shortdoc "A collection of database related tasks"
@@ -48,9 +52,15 @@ defmodule Mix.Tasks.Pleroma.Database do
   def run(["update_users_following_followers_counts"]) do
     start_pleroma()
 
-    User
-    |> Repo.all()
-    |> Enum.each(&User.update_follower_count/1)
+    Repo.transaction(
+      fn ->
+        from(u in User, select: u)
+        |> Repo.stream()
+        |> Stream.each(&User.update_follower_count/1)
+        |> Stream.run()
+      end,
+      timeout: :infinity
+    )
   end
 
   def run(["prune_objects" | args]) do
@@ -99,7 +109,7 @@ defmodule Mix.Tasks.Pleroma.Database do
       where: fragment("(?)->>'likes' is not null", object.data),
       select: %{id: object.id, likes: fragment("(?)->>'likes'", object.data)}
     )
-    |> Pleroma.RepoStreamer.chunk_stream(100)
+    |> Pleroma.Repo.chunk_stream(100, :batches)
     |> Stream.each(fn objects ->
       ids =
         objects
@@ -122,6 +132,66 @@ defmodule Mix.Tasks.Pleroma.Database do
     |> Stream.run()
   end
 
+  def run(["transfer_hashtags"]) do
+    import Ecto.Query
+
+    start_pleroma()
+
+    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,
+        inserted_at: object.inserted_at,
+        tag: fragment("(?)->>'tag'", object.data)
+      },
+      order_by: [desc: object.id]
+    )
+    |> Pleroma.Repo.chunk_stream(100, :batches)
+    |> Stream.each(fn objects ->
+      chunk_start = List.first(objects)
+      chunk_end = List.last(objects)
+
+      Logger.info(
+        "transfer_hashtags: " <>
+          "#{chunk_start.id} (#{chunk_start.inserted_at}) -- " <>
+          "#{chunk_end.id} (#{chunk_end.inserted_at})"
+      )
+
+      Enum.map(
+        objects,
+        fn object ->
+          hashtags =
+            object.tag
+            |> Jason.decode!()
+            |> Enum.filter(&is_bitstring(&1))
+
+          with {:ok, hashtag_records} <- Hashtag.get_or_create_by_names(hashtags) do
+            Repo.transaction(fn ->
+              for hashtag_record <- hashtag_records do
+                with {:error, _} <-
+                       Ecto.Adapters.SQL.query(
+                         Repo,
+                         "insert into hashtags_objects(hashtag_id, object_id) values " <>
+                           "(#{hashtag_record.id}, #{object.id});"
+                       ) do
+                  Logger.warn(
+                    "ERROR: could not link object #{object.id} and hashtag #{hashtag_record.id}"
+                  )
+                end
+              end
+            end)
+          else
+            e -> Logger.warn("ERROR: could not process object #{object.id}: #{inspect(e)}")
+          end
+        end
+      )
+    end)
+    |> Stream.run()
+  end
+
   def run(["vacuum", args]) do
     start_pleroma()
 
@@ -145,7 +215,7 @@ defmodule Mix.Tasks.Pleroma.Database do
     |> where(local: true)
     |> where([a], fragment("(? ->> 'type'::text) = 'Create'", a.data))
     |> where([_a, o], fragment("?->>'type' = 'Note'", o.data))
-    |> Pleroma.RepoStreamer.chunk_stream(100)
+    |> Pleroma.Repo.chunk_stream(100, :batches)
     |> Stream.each(fn activities ->
       Enum.each(activities, fn activity ->
         expires_at =