Como usar Oban para processar consultas de CPF em background no Elixir

# mix.exs
defp deps do
    [
    {:oban, "~> 2.17"},
    {:httpoison, "~> 2.0"},
    {:jason, "~> 1.4"}
    ]
end

# config/config.exs
config :minha_app, Oban,
    repo: MinhaApp.Repo,
    queues: [cpf_consultas: 5, cpf_lote: 2],
    plugins: [
    Oban.Plugins.Pruner,
    {Oban.Plugins.Cron, crontab: []}
    ]

# Terminal
# mix ecto.gen.migration add_oban_jobs_table
# Na migration gerada:
defmodule MinhaApp.Repo.Migrations.AddObanJobsTable do
    use Ecto.Migration

    def up, do: Oban.Migration.up(version: 12)
    def down, do: Oban.Migration.down(version: 1)
end

defmodule MinhaApp.Workers.CpfConsultaWorker do
    use Oban.Worker,
    queue: :cpf_consultas,
    max_attempts: 3,
    priority: 1

    @base_url "https://api.cpfhub.io"

    @impl Oban.Worker
    def perform(%Oban.Job{args: %{"cpf" => cpf, "callback_url" => callback_url} = args}) do
    api_key = Application.get_env(:minha_app, :cpfhub_api_key)
    headers = [{"x-api-key", api_key}]

    case HTTPoison.get("#{@base_url}/cpf/#{cpf}", headers, recv_timeout: 10_000) do
    {:ok, %{status_code: 200, body: body}} ->
    resultado = Jason.decode!(body)
    processar_resultado(cpf, resultado, callback_url)
    :ok

    {:ok, %{status_code: status}} ->
    {:error, "API retornou status #{status}"}

    {:error, %{reason: reason}} ->
    {:error, "Falha na requisição: #{inspect(reason)}"}
    end
    end

    defp processar_resultado(cpf, %{"success" => true, "data" => data}, callback_url) do
    resultado = %{
    cpf: cpf,
    nome: data["name"],
    data_nascimento: data["birthDate"],
    genero: data["gender"],
    status: "encontrado",
    processado_em: DateTime.utc_now()
    }

    if callback_url do
    HTTPoison.post(callback_url, Jason.encode!(resultado),
    [{"Content-Type", "application/json"}])
    end

    MinhaApp.Cpf.salvar_resultado(resultado)
    end

    defp processar_resultado(cpf, _, _callback_url) do
    MinhaApp.Cpf.salvar_resultado(%{cpf: cpf, status: "nao_encontrado"})
    end
end

defmodule MinhaApp.Workers.CpfLoteWorker do
    use Oban.Worker,
    queue: :cpf_lote,
    max_attempts: 1,
    unique: [period: 300]

    alias MinhaApp.Workers.CpfConsultaWorker

    @impl Oban.Worker
    def perform(%Oban.Job{args: %{"cpfs" => cpfs, "lote_id" => lote_id}}) do
    jobs = Enum.map(cpfs, fn cpf ->
    CpfConsultaWorker.new(%{
    cpf: cpf,
    lote_id: lote_id,
    callback_url: nil
    })
    end)

    Oban.insert_all(jobs)
    :ok
    end
end

defmodule MinhaApp.Cpf.Enqueuer do
    alias MinhaApp.Workers.{CpfConsultaWorker, CpfLoteWorker}

    def consultar_async(cpf, opts \\ []) do
    callback_url = Keyword.get(opts, :callback_url)
    prioridade = Keyword.get(opts, :prioridade, 1)
    agendar_em = Keyword.get(opts, :agendar_em)

    args = %{cpf: cpf, callback_url: callback_url}

    worker_opts = [priority: prioridade]
    worker_opts = if agendar_em, do: [{:scheduled_at, agendar_em} | worker_opts], else: worker_opts

    args
    |> CpfConsultaWorker.new(worker_opts)
    |> Oban.insert()
    end

    def processar_lote(cpfs) do
    lote_id = Ecto.UUID.generate()

    %{cpfs: cpfs, lote_id: lote_id}
    |> CpfLoteWorker.new()
    |> Oban.insert()

    {:ok, lote_id}
    end

    def consultar_agendado(cpf, data_hora) do
    consultar_async(cpf, agendar_em: data_hora)
    end
end

# Uso
MinhaApp.Cpf.Enqueuer.consultar_async("12345678900",
    callback_url: "https://meuapp.com/webhook/cpf"
)

MinhaApp.Cpf.Enqueuer.processar_lote([
    "12345678900", "98765432100", "11122233344"
])

# Agendar para daqui a 1 hora
MinhaApp.Cpf.Enqueuer.consultar_agendado(
    "12345678900",
    DateTime.add(DateTime.utc_now(), 3600, :second)
)

defmodule MinhaApp.Cpf.Monitor do
    import Ecto.Query
    alias MinhaApp.Repo

    def estatisticas do
    query = from j in Oban.Job,
    where: j.worker == "MinhaApp.Workers.CpfConsultaWorker",
    group_by: j.state,
    select: {j.state, count(j.id)}

    stats = Repo.all(query) |> Map.new()

    %{
    pendentes: Map.get(stats, "available", 0),
    executando: Map.get(stats, "executing", 0),
    concluidos: Map.get(stats, "completed", 0),
    com_erro: Map.get(stats, "retryable", 0) + Map.get(stats, "discarded", 0),
    agendados: Map.get(stats, "scheduled", 0)
    }
    end

    def jobs_com_erro do
    query = from j in Oban.Job,
    where: j.worker == "MinhaApp.Workers.CpfConsultaWorker",
    where: j.state in ["retryable", "discarded"],
    order_by: [desc: j.inserted_at],
    limit: 20

    Repo.all(query)
    end
end