diff --git a/lib/pleroma/workers/purge_expired_activity.ex b/lib/pleroma/workers/purge_expired_activity.ex index 01256831b..027171c1e 100644 --- a/lib/pleroma/workers/purge_expired_activity.ex +++ b/lib/pleroma/workers/purge_expired_activity.ex @@ -7,7 +7,7 @@ defmodule Pleroma.Workers.PurgeExpiredActivity do Worker which purges expired activity. """ - use Oban.Worker, queue: :activity_expiration, max_attempts: 1 + use Oban.Worker, queue: :activity_expiration, max_attempts: 1, unique: [period: :infinity] import Ecto.Query diff --git a/priv/repo/migrations/20210128092834_remove_duplicates_from_activity_expiration_queue.exs b/priv/repo/migrations/20210128092834_remove_duplicates_from_activity_expiration_queue.exs new file mode 100644 index 000000000..309009205 --- /dev/null +++ b/priv/repo/migrations/20210128092834_remove_duplicates_from_activity_expiration_queue.exs @@ -0,0 +1,29 @@ +defmodule Pleroma.Repo.Migrations.RemoveDuplicatesFromActivityExpirationQueue do + use Ecto.Migration + + import Ecto.Query, only: [from: 2] + + def up do + duplicate_ids = + from(j in Oban.Job, + where: j.queue == "activity_expiration", + where: j.worker == "Pleroma.Workers.PurgeExpiredActivity", + where: j.state == "scheduled", + select: + {fragment("(?)->>'activity_id'", j.args), fragment("array_agg(?)", j.id), count(j.id)}, + group_by: fragment("(?)->>'activity_id'", j.args), + having: count(j.id) > 1 + ) + |> Pleroma.Repo.all() + |> Enum.map(fn {_, ids, _} -> + max_id = Enum.max(ids) + List.delete(ids, max_id) + end) + |> List.flatten() + + from(j in Oban.Job, where: j.id in ^duplicate_ids) + |> Pleroma.Repo.delete_all() + end + + def down, do: :noop +end