`Oban` Tutorial

[ndrean]

https://github.com/oban-bg/oban
https://github.com/oban-bg/oban_training

[ndrean]

A quick video to start

[nelsonic]

Stoked that Oban exists in the Elixir ecosystem; it's a perfect fit for the underlying tech. 🎉
Totally understand/respect that the makers of Oban have to make money to keep building their project. 👍 🫶
But not a fan of their pricing model: https://oban.pro/pricing 🙅

I don't think I need these features that are only available in the Pro plan ...

• Batches (can code this myself)
• Chains (|> ...?)
• Global and Rate Limiting
• Decorators
• Dynamic Plugins

But the Workflows are kinda the whole point of having all this overhead. 🙃
A "job processing" framework without a way to define workflows is deliberately handicapped. 👎

And the $135/month ($1650/year) for that single feature is silly expensive. 💸 🔥

It's fine if the Pro features are core to the product you are building and you need/want their Dedicated Support. 📲
But means the architecture of your app is tightly coupled to a proprietary system. 🔒
Yes, the code may be Open Source but it will have some sort of API Key to unlock the Pro features. 🔑
Which you then have to share with all the contributors ... 😬
And it all stops working if you don't continue paying rent into infinity. 🤦‍♂️

[ndrean]

@nelsonic you can do a lot already with the free plan. For Workflows, I use https://reactor.hexdocs.pm/readme.html for synchronous request/response pipelines. You would use Oban for async background work when fan-out execution (call multiple parallel tasks) is important. But if you are for example doing multiple independent inserts, then you don't want a rollback for an isolated failure. IT all depends if you need safety at the price of complexity.

[ndrean]

@nelsonic Let me push my AI-assisted workflow runner. I cannot detail/explain everything in this repo but it gives you ideas. I use a few Elixir libs:

It uses Reactor: given a user's input (via mobile/Telegram text or audio, via webapp text), I have a workflow classify → dispatch → execute → reply where I reach a cloud LLM to interpret the input and select which workflow I want to use, and again to extract the action with the params within a workflow (the input "show me my contacts" is sent to the LLM which returns the workflow "contacts", and then "search, *").

It uses Oban, for reminders for appointments, for cron file cleaning, for log retention cleaning...

It uses Hammer o rate limit (firstly the LLM of course, but also logins of course...).

[nelsonic]

Yeah, there's always a tradeoff. ⚖️ 👍
Love the features of Oban. ❤️
Just not their pricing/security model ... 😬
Having worked in high security environments (Banks, Insurance Companies, Stock Exchanges, Gov Agencies) 🏦
Requiring an API Key to use Pro features and thus having their script "phoning home" on my server is a no-go. 🙅
Which is the reason I haven't dived into using it ... 🙃

IfOban had a way of running the Pro version without phoning home I would use it and pay for it ...
But without it, I'm just going to have to let Ai recreate the features we need from the many available examples. #90 🙈

[ndrean]

recreate the features we need <= I might be wrong but it seems like a big challenge with very uncertain results

I already made quite a bit of efforts just using parts of existing libraries (Oban and Reactor). An example:

I had to use Telegram. I used https://hex.pm/packages/telegex . A user sends some input and the app reacts on it and sends back a response to the user.

I have to use some persistence on the mobile channel. Why? Because on mobile, a user has no idea if the backend fails, or he can send a message and his device loses his connection.

1. I implement the on_update callback (that handles the Telegram input).

@impl true
  def on_update(%{message: %{text: text, chat: %{id: chat_id}}} = _update) when is_binary(text) do
    %{
      "user_id" => to_string(chat_id),
      "text" => text,
      "channel" => "telegram",
      "chat_id" => to_string(chat_id),
      "is_audio" => false
    }
    |> IngestWorker.new()
    |> Oban.insert()
    |> log_insert_error(chat_id)

    :ok
  end

It looks simple but isn't really.

2. I introduced Oban for this persistence. I define an Oban changeset and persist it. The Oban pattern is:
   %{...} |> IngestWorker.new() |> Oban.insert()

It inserts the arguments in the database.
The Oban polling picks a job and calls perform. The IngestWorker contains the implementation of the perform/1 callback:

defmodule CrmReactor.Workers.IngestWorker do
  @moduledoc "Oban worker: runs the MasterIngest Reactor pipeline asynchronously."
  use Oban.Worker, queue: :ingest, max_attempts: 3

  import Ecto.Query

  @impl true
  def perform(%Oban.Job{id: job_id, args: args}) do
    input = %{
      user_id: args["user_id"],
      raw_input: args["text"],
      is_audio: args["is_audio"] || false,
      channel: to_channel(args["channel"]),
      job_id: "oban-#{job_id}",
      attachment: nil
    }

    case Reactor.run(MasterIngest, input) do
      {:ok, result} ->
        case maybe_send_reply(args["channel"], args["chat_id"], result) do
           ....

This run the Reactor MasterIngest workflow with the following steps: tenant lookup, call the LLM, save execution logs, classify, dispatch to workflow, finalize (see below).

Oban has a coarse grained retry: the whole Reactor will rerun, and since I have to save logs, I have to ensure of idempotence in case of failure and retry.

With Reactor, you can opt-in for a Reactor native compensate for undoing side-effects. Or if not, simply application level retries.

All this needs quite a bit of design and thinking. Claude can help to you use the library (but I believe you need to understand the library because there are a few design points) but designing all this from ground seems a heavy task with uncertain results.

The MasterIngest module that defines the Reactor steps, but without the run/3 implementation

defmodule CrmReactor.Reactors.MasterIngest do
  @moduledoc "Main pipeline: tenant resolution, transcription, classification, dispatch."
  use Reactor

  middlewares do
    middleware(Reactor.Middleware.Telemetry)
  end

  input(:user_id)
  input(:raw_input)
  input(:is_audio)
  input(:channel)
  input(:job_id)
  input(:attachment)

  # :tenant (DB lookup) and :text (optional Whisper HTTP call) have no inter-dependency
  # and are run concurrently by Reactor before either feeds into :classification.
  step :tenant, CrmReactor.Reactors.Steps.ResolveTenant do
    argument(:user_id, input(:user_id))
  end

  step :text, CrmReactor.Reactors.Steps.Transcribe do
    argument(:raw_input, input(:raw_input))
    argument(:is_audio, input(:is_audio))
  end

  step :log, CrmReactor.Reactors.Steps.LogExecution do
    argument(:tenant, result(:tenant))
    argument(:raw_input, input(:raw_input))
    argument(:channel, input(:channel))
    argument(:user_id, input(:user_id))
    argument(:job_id, input(:job_id))
  end

  step :classification, CrmReactor.Reactors.Steps.ClassifyIntent do
    argument(:text, result(:text))
    argument(:attachment, input(:attachment))
    argument(:tenant, result(:tenant))
    argument(:user_id, input(:user_id))
  end

  step :result, CrmReactor.Reactors.Steps.DispatchModule do
    argument(:classification, result(:classification))
    argument(:tenant, result(:tenant))
    argument(:channel, input(:channel))
    argument(:user_id, input(:user_id))
    argument(:log, result(:log))
    argument(:text, result(:text))
  end

  step :finalize, CrmReactor.Reactors.Steps.FinalizeReply do
    argument(:result, result(:result))
    argument(:log, result(:log))
    argument(:tenant, result(:tenant))
    argument(:classification, result(:classification))
    argument(:attachment, input(:attachment))
    argument(:user_id, input(:user_id))
    argument(:text, result(:text))
  end

  return(:finalize)
end