+ )
+ |> where(local: true)
+ |> where([a], fragment("(? ->> 'type'::text) = 'Create'", a.data))
+ |> where([_a, o], fragment("?->>'type' = 'Note'", o.data))
+ |> Pleroma.Repo.chunk_stream(100, :batches)
+ |> Stream.each(fn activities ->
+ Enum.each(activities, fn activity ->
+ expires_at =
+ activity.inserted_at
+ |> DateTime.from_naive!("Etc/UTC")
+ |> Timex.shift(days: days)
+
+ Pleroma.Workers.PurgeExpiredActivity.enqueue(%{
+ activity_id: activity.id,
+ expires_at: expires_at
+ })
+ end)
+ end)
+ |> Stream.run()