git.squeep.com Git - akkoma/blob - lib/pleroma/workers/worker_helper.ex

   1 # Pleroma: A lightweight social networking server
   2 # Copyright © 2017-2021 Pleroma Authors <https://pleroma.social/>
   3 # SPDX-License-Identifier: AGPL-3.0-only
   4
   5 defmodule Pleroma.Workers.WorkerHelper do
   6   alias Pleroma.Config
   7   alias Pleroma.Workers.WorkerHelper
   8
   9   def worker_args(queue) do
  10     case Config.get([:workers, :retries, queue]) do
  11       nil -> []
  12       max_attempts -> [max_attempts: max_attempts]
  13     end
  14   end
  15
  16   def sidekiq_backoff(attempt, pow \\ 4, base_backoff \\ 15) do
  17     backoff =
  18       :math.pow(attempt, pow) +
  19         base_backoff +
  20         :rand.uniform(2 * base_backoff) * attempt
  21
  22     trunc(backoff)
  23   end
  24
  25   defmacro __using__(opts) do
  26     caller_module = __CALLER__.module
  27     queue = Keyword.fetch!(opts, :queue)
  28
  29     quote do
  30       # Note: `max_attempts` is intended to be overridden in `new/2` call
  31       use Oban.Worker,
  32         queue: unquote(queue),
  33         max_attempts: 1
  34
  35       alias Oban.Job
  36
  37       def enqueue(op, params, worker_args \\ []) do
  38         params = Map.merge(%{"op" => op}, params)
  39         queue_atom = String.to_atom(unquote(queue))
  40         worker_args = worker_args ++ WorkerHelper.worker_args(queue_atom)
  41
  42         unquote(caller_module)
  43         |> apply(:new, [params, worker_args])
  44         |> Oban.insert()
  45       end
  46
  47       @impl Oban.Worker
  48       def timeout(_job) do
  49         queue_atom = String.to_atom(unquote(queue))
  50         Config.get([:workers, :timeout, queue_atom], :timer.minutes(1))
  51       end
  52     end
  53   end
  54 end