Graham McIntire 6c05d45dd6 deps: upgrade oban_pro to 1.7.0

Bumps vendored Oban Pro from 1.6.13 to 1.7.0, which transitively bumps
oban core to 2.22.1 (requires migrations v14) and db_connection to 2.10.0.

- mix.exs constraint: ~> 1.6 → ~> 1.7
- vendor/oban_pro/ replaced with v1.7.0 hex package
- migration 20260430142200: Oban core v12 → v14 (must run first)
- migration 20260430142241: Oban Pro 1.7.0 schema additions

Also corrects a pre-existing constraint name in AgentAssignment that was
left stale by the equipment→devices rename migration (20260117190134).
The unique index is named agent_assignments_device_id_index, but the
schema's unique_constraint/3 still pointed at the old equipment_id name,
causing the test suite to surface Ecto.ConstraintError instead of a
changeset error. Required to get the suite green for this upgrade.

2026-04-30 10:26:18 -05:00

1.9 KiB

Raw Blame History

Workflows

Workflows compose jobs with dependency relationships: sequential, fan-out, and fan-in patterns.

alias Oban.Pro.Workflow

Workflow.new()
|> Workflow.add(:extract, ExtractWorker.new(%{source: "db"}))
|> Workflow.add(:transform, TransformWorker.new(%{}), deps: :extract)
|> Workflow.add(:notify, NotifyWorker.new(%{}), deps: :extract)  # fan-out
|> Workflow.add(:load, LoadWorker.new(%{}), deps: [:transform, :notify])  # fan-in
|> Oban.insert_all()

Patterns

Sequential: deps: :previous_job
Fan-out: Multiple jobs depend on the same parent
Fan-in: One job depends on multiple parents with deps: [:a, :b, :c]

Accessing upstream results

Use recorded: true workers:

def process(job) do
  {:ok, data} = Workflow.fetch_recorded(job, :extract)
end

Cascading Functions

Function captures that automatically receive context and upstream results:

Workflow.new()
|> Workflow.put_context(%{source: source})
|> Workflow.add_cascade(:extract, &extract/1)
|> Workflow.add_cascade(:transform, &transform/1, deps: :extract)
|> Workflow.add_cascade(:load, &load/1, deps: :transform)
|> Oban.insert_all()

def extract(%{source: source}), do: %{records: fetch_records(source)}
def transform(%{extract: %{records: records}}), do: %{data: process(records)}
def load(%{transform: %{data: data}}), do: insert_all(data)

Fan-out with {enumerable, function/2}:

Workflow.add_cascade(:accounts, {account_ids, &sync_account/2})

Sub-workflows

Compose from reusable pieces:

extract_flow =
  Workflow.new()
  |> Workflow.add(:fetch, FetchWorker.new(%{}))
  |> Workflow.add(:parse, ParseWorker.new(%{}), deps: :fetch)

Workflow.new()
|> Workflow.add(:setup, SetupWorker.new(%{}))
|> Workflow.add_workflow(:extract, extract_flow, deps: :setup)
|> Workflow.add(:cleanup, CleanupWorker.new(%{}), deps: :extract)
|> Oban.insert_all()

1.9 KiB Raw Blame History