git.squeep.com Git - akkoma/blob - priv/repo/migrations/20200825061316_move_activity_expirations_to_oban.exs

   1 defmodule Pleroma.Repo.Migrations.MoveActivityExpirationsToOban do
   2   use Ecto.Migration
   3
   4   import Ecto.Query, only: [from: 2]
   5
   6   def change do
   7     Pleroma.Config.Oban.warn()
   8
   9     Application.ensure_all_started(:oban)
  10
  11     Supervisor.start_link([{Oban, Pleroma.Config.get(Oban)}],
  12       strategy: :one_for_one,
  13       name: Pleroma.Supervisor
  14     )
  15
  16     from(e in "activity_expirations",
  17       select: %{id: e.id, activity_id: e.activity_id, scheduled_at: e.scheduled_at}
  18     )
  19     |> Pleroma.Repo.stream()
  20     |> Stream.each(fn expiration ->
  21       with {:ok, expires_at} <- DateTime.from_naive(expiration.scheduled_at, "Etc/UTC") do
  22         Pleroma.Workers.PurgeExpiredActivity.enqueue(%{
  23           activity_id: FlakeId.to_string(expiration.activity_id),
  24           expires_at: expires_at
  25         })
  26       end
  27     end)
  28     |> Stream.run()
  29   end
  30 end