X-Git-Url: https://git.squeep.com/?a=blobdiff_plain;f=priv%2Frepo%2Fmigrations%2F20200825061316_move_activity_expirations_to_oban.exs;h=a703af83fd270064001df1554bb45153dc35d96b;hb=78cea44553c936813ffc2c4c47984a64255096ef;hp=585d1a600ff5e85d604087f938fe570839e44f21;hpb=5ad0cc4c863f7f8a1e6fdfa40eb884a5c94ebf67;p=akkoma

diff --git a/priv/repo/migrations/20200825061316_move_activity_expirations_to_oban.exs b/priv/repo/migrations/20200825061316_move_activity_expirations_to_oban.exs
index 585d1a600..a703af83f 100644
--- a/priv/repo/migrations/20200825061316_move_activity_expirations_to_oban.exs
+++ b/priv/repo/migrations/20200825061316_move_activity_expirations_to_oban.exs
@@ -4,6 +4,8 @@ defmodule Pleroma.Repo.Migrations.MoveActivityExpirationsToOban do
   import Ecto.Query, only: [from: 2]
 
   def change do
+    Pleroma.Config.Oban.warn()
+
     Supervisor.start_link([{Oban, Pleroma.Config.get(Oban)}],
       strategy: :one_for_one,
       name: Pleroma.Supervisor
@@ -12,17 +14,14 @@ defmodule Pleroma.Repo.Migrations.MoveActivityExpirationsToOban do
     from(e in "activity_expirations",
       select: %{id: e.id, activity_id: e.activity_id, scheduled_at: e.scheduled_at}
     )
-    |> Pleroma.RepoStreamer.chunk_stream(500)
-    |> Stream.each(fn expirations ->
-      Enum.each(expirations, fn expiration ->
-        with {:ok, expires_at} <- DateTime.from_naive(expiration.scheduled_at, "Etc/UTC") do
-          Pleroma.Workers.PurgeExpiredActivity.enqueue(%{
-            activity_id: FlakeId.to_string(expiration.activity_id),
-            expires_at: expires_at,
-            validate: false
-          })
-        end
-      end)
+    |> Pleroma.Repo.stream()
+    |> Stream.each(fn expiration ->
+      with {:ok, expires_at} <- DateTime.from_naive(expiration.scheduled_at, "Etc/UTC") do
+        Pleroma.Workers.PurgeExpiredActivity.enqueue(%{
+          activity_id: FlakeId.to_string(expiration.activity_id),
+          expires_at: expires_at
+        })
+      end
     end)
     |> Stream.run()
   end