Abstracting generic Oban jobs

Special Sections Oban
,
1

Perhaps I’m missing a general understanding of “generic” jobs and how to handle them, one example would be sending out emails:

Lets say I have 30 different email templates to send, OnboardEmail, SuccessfulPayment, etc.

I would just create a job named Worker.SendEmail and provide necessary arguments to fetch all the data I need, like %{"org_id" => 42, "template" => "org_onboarding"} as job args, instead of actually creating 30 separate workers right? :smile:

What is your approach/design to this problem?

1 Like
2

That’s the approach I would use to generically send emails. Depending on how involved your delivery and template logic is, you could abstract it down to a worker that takes MFA (mod, fun, args) to build an email and deliver it.

defmodule MyApp.Emails.Sender do
  use Oban.Worker, queue: :email

  @impl Oban.Worker
  def process(%Job{args: %{"mod" => mod, "fun" => fun, "args" => args}}) do
    mod = String.to_existing_atom(mod)
    fun = String.to_existing_atom(fun)

    mod
    |> apply(fun, args)
    |> MyApp.Mailer.deliver()
  end
end

Then build a new job with %{mod: SomeEmail, fun: :onboard, args: [42]}. For convenience, you can override the worker’s new/2 callback:

def new(mod, fun, args, opts \\ []) when is_atom(mod) and is_atom(fun) and is_list(args) do
  new(%{mod: mod, fun: fun, args: args}, opts)
end

Then you can call Sender.new(SomeEmail, :onboard, [42]) with a light bit of validation. It’s possible to verify that the mod exists and the function is exported, but can be flaky during development.

2 Likes
3

Appreciate the answer! That is indeed even more flexible :smile:

Now that I think of it I would need to stop myself from over using this worker to call arbitrary functions all over the place /s

1 Like
4

Agree that this can get a little hairy. I would personally use a little bit of metaprogramming and mark the modules/functions that are allowed to be used by this worker, throwing a compile-time error when you start using random functions.

2 Likes
5

While @sorentwo already answered your question. Consider that in some cases, the email building process can be costly or done in bulk and you might want a more granular control about the delivery process, which depending on your provider, could take some considerable time. Or you might want to share the same email struct between the delivery process and your email tracker. Anyway, the following worker should implement a general “Deliver Later” strategy:

defmodule MyApp.Email.DeliverLaterWorker do
  @moduledoc """
  Oban Worker that deliver emails
  """
  use Oban.Pro.Worker,
    queue: :email,
    max_attempts: 3,
    encrypted: [key: {System, :fetch_env!, ["ENC_KEY"]}]

  @doc """
  Enqueues an Oban Worker to deliver an email
  """
  def enqueue(%Swoosh.Email{} = email) do
    email
    # NOTE: You might not need the assigns,
    # so, let's reduce the payload right now.
    |> Map.put(:assigns, %{})
    |> then(&%{email: encode(&1)})
    |> new()
    |> Oban.insert()
  end

  @impl Oban.Pro.Worker
  def process(%Job{args: %{"email" => email}}) do
    email = decode(email)

    case MyApp.Mailer.deliver(email) do
      {:ok, email} -> {:ok, email}

      {:error, {:send, {:permanent_failure, _, _}}} ->
        # NOTE: Do something here to avoid wasting resources in the future to these recipients
        {:cancel, :permanent_failure}

      {:error, reason} -> raise Swoosh.DeliveryError, reason: reason
    end
  end

  #...
end

The first public function, enqueue/1 is just a helper to insert an oban job, you probably need to adapt that if you’re inserting jobs in bulk. The main benefit is that you reduce the payload by getting rid of the assigns, which you probably don’t need anyway.

In particular, here I’m using an encrypted job to avoid leaking some Personally Identifiable Information (PII) in case of an exception and that would depend on the way you log your exceptions (e.g. Sentry, DDOG), but that might not be required in your case.

For completeness, your encode/1, and decode/1 could be like this:

  defp encode(email) do
    email
    |> :erlang.term_to_binary([:compressed])
    |> Base.encode64(padding: false)
  end

  defp decode(email) do
    email
    |> Base.decode64!(padding: false)
    |> :erlang.binary_to_term([:safe])
  end

HTH

1 Like
6

How would you do that besides some global registry a’la ETS?

7

One easy way to check at runtime is to introspect on which behaviours the module implements:

:attributes
|> module.__info__()
|> Keyword.get_values(:behaviour)
|> List.flatten()
|> Enum.member?(MyApp.SomeBehaviour)

But, as I mentioned above, that approach is risky in development because modules are lazily loaded.

2 Likes
8

Either that or inject some custom identifier into your modules, for example (note that this macro might not compile as it stand, it’s just for demo porposes):

defmodule Marker do
  defmacro __using__(_env, only: funcs) do
    quote do
      def __these_functions_can_use_mailer_worker__(), do: unquote(funcs) 
    end
  end
end

def MyModule do
  use Marker, only: [send: 0]

  def send() do
...
end

In this way, your Oban worker can access that special name function and check whether the specific function is marked as accepted for job. You can also skip the only option and just say that if the module has the special function generated, then it is good to go.

3 Likes
9

I’m glad I asked for an approach but a little bit ashamed because I’ve used this same technique before… Thank you!

2 Likes