Please pardon the slow reply, I’ve been on vacation for the past few weeks
This is entirely doable with a single worker through the use of :snooze. Here’s a pseudo example:
defmodule MyApp.SequentialWorker do
use Oban.Worker, queue: :some_queue
import Ecto.Query, only: [where: 3]
@impl Worker
def perform(%Job{args: args}) do
if executable_for_user?(args) do
# do normal stuff
else
# snooze for 60 seconds, make it as short as necessary
{:snooze, 60}
end
end
defp executable_for_user?(%{"user_id" => user_id}) do
Job
|> where(j, j.worker == Worker.to_string(__MODULE__))
|> where(j, j.queue == "some_queue")
|> where(j, j.state == "executing")
|> where(j, fragment("? @> ?", j.args, ^%{user_id: user_id})
|> MyApp.Repo.exists?()
end
end
By checking whether there is another job executing you can force global concurrency. In this case the worker checks for other executing workers with the same user_id, though that is arbitrary and you could scope to anything you like.
It isn’t as optimized as running multiple queues with global concurrency, but it will work right now and is accurate.
