# Oban Pro Reference

Oban Pro extends Oban with advanced worker types, job composition, and operational plugins.

> **Official docs**: <https://oban.pro/docs/pro/overview.html>
> Always check for the latest API — this reference covers stable core features.

## Migration: OSS to Pro

### Worker Migration

| OSS Oban | Oban Pro |
|----------|----------|
| `use Oban.Worker` | `use Oban.Pro.Worker` |
| `@impl Oban.Worker` | `@impl Oban.Pro.Worker` |
| `def perform(%Job{})` | `def process(%Job{})` |
| `Oban.Testing` | `Oban.Pro.Testing` |
| Advisory lock engine | `Oban.Pro.Engines.Smart` |

### Plugin Migration

Pro plugins **enhance** OSS equivalents (same base features + extras).
Swap the module name — do NOT run both simultaneously:

| OSS Plugin | Pro Enhancement | Key Addition |
|------------|----------------|-------------|
| `Plugins.Cron` | `DynamicCron` | Runtime CRUD, missed job guarantees, per-entry timezone |
| `Plugins.Lifeline` | `DynamicLifeline` | Auto-repairs stuck workflows/chains, producer-based rescue |
| `Plugins.Pruner` | `DynamicPruner` | Per-queue/worker/state retention policies, before_delete hook |

### Engine Migration

```elixir
# config/config.exs — switch to Smart Engine
config :my_app, Oban,
  engine: Oban.Pro.Engines.Smart,
  repo: MyApp.Repo,
  queues: [default: 10]
```

Smart Engine enables: global concurrency, distributed rate limiting, async tracking.

---

## Pro.Worker Features

Pro.Worker replaces `perform/1` with `process/1` and adds optional features:
structured args, hooks, encryption, deadlines, chaining, and recorded output.

Without `args_schema`, Pro.Worker works identically to OSS — just use `process/1`
with string-key pattern matching instead of `perform/1`.

### Structured Jobs (`args_schema`) — Optional

Opt-in type-safe args with compile-time validation and casting.
When used, `process/1` receives a struct instead of a raw map:

```elixir
defmodule MyApp.Workers.SendEmail do
  use Oban.Pro.Worker, queue: :mailers

  args_schema do
    field :email, :string, required: true
    field :user_id, :id, required: true
    field :priority, :enum, values: ~w(low normal high)a, default: :normal

    embeds_one :config do
      field :subject, :string
      field :template, :string
    end
  end

  @impl Oban.Pro.Worker
  def process(%Job{args: %__MODULE__{email: email, config: config}}) do
    MyApp.Mailer.send(email, config.subject, config.template)
  end
end
```

Supported types: `:id`, `:integer`, `:string`, `:float`, `:boolean`,
`:binary`, `:map`, `:enum`, `:uuid`, `:datetime_utc`.

### Recorded Jobs

Store job output for retrieval by downstream jobs or dashboards:

```elixir
use Oban.Pro.Worker, recorded: true

@impl Oban.Pro.Worker
def process(%Job{} = job) do
  result = MyApp.expensive_computation(job.args)
  {:ok, result}  # Automatically recorded, compressed
end
```

### Encrypted Jobs

AES-256-CTR encryption for args at rest:

```elixir
use Oban.Pro.Worker,
  encryption: {MyApp.Vault, :fetch_key, []}
```

**Iron Law**: Encryption breaks uniqueness on `args` (encrypted args differ
each time). Use `meta` for unique constraints with encrypted workers.

### Deadlines

Preemptively cancel jobs exceeding time limits:

```elixir
use Oban.Pro.Worker, deadline: {1, :hour}

# Or per-job:
MyApp.Worker.new(%{data: "..."}, deadline: {30, :minutes})
```

### Chaining

Enforce sequential execution per partition key:

```elixir
use Oban.Pro.Worker, chain: [by: [args: :account_id]]

@impl Oban.Pro.Worker
def process(%Job{args: %{"account_id" => _aid}}) do
  # Only one job with this account_id runs at a time
  :ok
end
```

Partition options: `:worker`, `[args: :field]`, `[meta: :key]`.

### Worker Hooks

Lifecycle callbacks: `before_new/1`, `before_process/1`, `after_process/2`,
`on_cancelled/1`, `on_discarded/1`. Use for logging, metrics, Sentry integration.

```elixir
def after_process(%Job{} = job, _result), do: Metrics.track(:job_done, %{worker: job.worker})
def on_discarded(%Job{} = job), do: alert_team(job)
```

### Worker Aliases

Rename workers without breaking queued jobs: `aliases: [MyApp.OldWorkerName]`

---

## Job Composition Patterns

### When to Use What

| Pattern | Use When |
|---------|----------|
| **Workflow** | Multi-step with dependencies (ETL, pipelines) |
| **Batch** | Many parallel jobs, need aggregate callbacks |
| **Chunk** | Bulk processing for efficiency (SMS, notifications) |
| **Relay** | Need synchronous job result |
| **Chain** | Sequential per partition key (per-account ordering) |

### Workflows

Compose jobs with arbitrary dependencies (sequential, fan-out, fan-in):

```elixir
alias Oban.Pro.Workflow

Workflow.new()
|> Workflow.add(:extract, ExtractWorker.new(%{source: "api"}))
|> Workflow.add(:transform, TransformWorker.new(%{}), deps: [:extract])
|> Workflow.add(:validate, ValidateWorker.new(%{}), deps: [:transform])
|> Workflow.add(:load, LoadWorker.new(%{}), deps: [:validate])
|> Oban.insert_all()
```

Access upstream results with recorded jobs:

```elixir
# In LoadWorker:
def process(%Job{} = job) do
  {:ok, data} = Oban.Pro.Workflow.get_recorded(job, :transform)
  load_data(data)
end
```

### Batches

Group parallel jobs with aggregate lifecycle callbacks.

> **Note**: Batch API varies between Oban Pro versions. Check your installed
> version's docs with `mix hex.docs online oban_pro` for exact callback pattern.

```elixir
defmodule MyApp.EmailBatch do
  use Oban.Pro.Worker, queue: :mailers

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

  # Batch lifecycle callbacks — check Oban.Pro.Batch docs for your version
  def batch_completed(_job), do: Logger.info("All emails sent!")
  def batch_exhausted(_job), do: alert_team("Batch had failures")
end
```

Common callbacks: `batch_attempted/1`, `batch_completed/1`, `batch_cancelled/1`,
`batch_discarded/1`, `batch_exhausted/1`.

### Chunks

Process jobs atomically in groups for efficiency:

```elixir
defmodule MyApp.SmsSender do
  use Oban.Pro.Workers.Chunk,
    queue: :messages,
    size: 100,
    timeout: 5_000

  @impl true
  def process(jobs) do
    jobs
    |> Enum.map(& &1.args)
    |> MyApp.SMS.send_batch()
  end
end
```

Note: `process/1` receives a **list** of jobs, not a single job.

### Relay

Synchronous job execution: `Oban.Pro.Relay.async/1` + `Relay.await/2`.
Useful for: distributed task results, API endpoints needing job output, testing.

---

## Smart Engine

The Smart Engine replaces advisory locks with index-backed operations,
enabling multi-node features:

```elixir
config :my_app, Oban,
  engine: Oban.Pro.Engines.Smart,
  queues: [
    default: [local_limit: 10, global_limit: 50],
    api_calls: [
      local_limit: 5,
      rate_limit: [allowed: 100, period: 60]
    ]
  ]
```

### Key Capabilities

- **Global concurrency**: `global_limit` caps total jobs across all nodes
- **Rate limiting**: `rate_limit` with algorithms: `:sliding_window`, `:fixed_window`, `:token_bucket`
- **Partitioning**: Segment limits by worker, args, or metadata
- **Async tracking**: Batched status updates for throughput

Also provides `Oban.Pro.RateLimit` for programmatic rate limit checks outside jobs.

### Smart Engine Gotchas

**One partition limiter per queue**: Only ONE of `global_limit` or `rate_limit`
can have `partition` on a given queue. If you need both user isolation AND
rate limiting, use `rate_limit` with partition (provides both):

```elixir
# WRONG — two limiters with partition on same queue
my_queue: [
  global_limit: [allowed: 1, partition: [args: :user_id]],
  rate_limit: [allowed: 1, period: 60, partition: [args: :user_id]]
]

# CORRECT — rate_limit provides both isolation and throttling
my_queue: [
  local_limit: 200,
  rate_limit: [allowed: 1, period: 60, partition: [args: :user_id]]
]
```

**Snooze rolls back attempt counter**: With Smart Engine, `{:snooze, seconds}`
does NOT increment `attempt`. Code guarding on `attempt` to limit snoozes
will loop infinitely. Use `meta["snoozed"]` instead:

```elixir
# WRONG — infinite loop! Smart Engine resets attempt on snooze
def process(%Job{attempt: attempt}) when attempt <= 3 do
  {:snooze, 5}
end

# CORRECT — track snooze count in meta
def process(%Job{meta: meta} = job) do
  snoozed = Map.get(meta, "snoozed", 0)
  if snoozed < 3, do: {:snooze, 5}, else: {:cancel, "Max snoozes reached"}
end
```

---

## Pro Plugins

Pro-only plugins (no OSS equivalent):

| Plugin | Purpose |
|--------|---------|
| `DynamicQueues` | Runtime queue CRUD, node-specific routing |
| `DynamicPrioritizer` | Auto-bumps priority for starved jobs |
| `DynamicScaler` | Auto-scale infrastructure by queue depth |

### DynamicCron Example

```elixir
plugins: [
  {Oban.Pro.Plugins.DynamicCron, crontab: [
    {"0 0 * * *", MyApp.DailyReportWorker},
    {"0 */6 * * *", MyApp.SyncWorker, timezone: "America/New_York"}
  ]}
]
```

Runtime management: insert/update/delete cron entries without redeployment.

### DynamicQueues

Runtime queue management: `insert/3`, `update/3`, `delete/2` for CRUD without redeployment.
Supports node-specific routing via `only:` option.

---

## Anti-patterns

```elixir
# --- Wrong callback name (silent no-op!) ---
# BAD: perform/1 in Pro worker
def perform(%Job{} = job), do: process_data(job)
# GOOD: process/1 in Pro worker
def process(%Job{} = job), do: process_data(job)

# --- Encrypted args with unique on args ---
# BAD: uniqueness on encrypted fields (won't match!)
use Oban.Pro.Worker, encryption: [...], unique: [keys: [:user_id]]
# GOOD: use meta for uniqueness with encryption
use Oban.Pro.Worker, encryption: [...], unique: [keys: [], meta: [:user_id]]

# --- Chunk process/1 expects list ---
# BAD: pattern matching single job in chunk worker
def process(%Job{args: args}), do: ...
# GOOD: pattern matching list of jobs
def process(jobs) when is_list(jobs), do: ...
```

# Queue Configuration Reference

## Basic Configuration

```elixir
# config/config.exs
config :my_app, Oban,
  repo: MyApp.Repo,
  queues: [
    critical: 20,           # High priority, fast
    mailers: 50,            # I/O-bound
    webhooks: 30,           # External API calls
    media_processing: 3,    # CPU-intensive
    external_api: [limit: 5, dispatch_cooldown: 100],
    imports: 10             # Bulk processing
  ],
  plugins: [
    {Oban.Plugins.Pruner, max_age: 60 * 60 * 24 * 7},  # 7 days
    {Oban.Plugins.Lifeline, rescue_after: :timer.minutes(30)}
  ]
```

## Queue Design Principles

- **I/O vs CPU bound** — Separate queues prevent CPU work from blocking I/O
- **External dependencies** — Different APIs get isolated queues
- **Priority separation** — Critical jobs in dedicated high-concurrency queue
- **Rate limiting** — Queue-level `dispatch_cooldown` for API respect

## Connection Pool Sizing

```elixir
# Rule: pool_size >= num_queues + sum(queue_limits) + buffer
config :my_app, MyApp.Repo, pool_size: 25
```

## Cron Scheduling

```elixir
plugins: [
  {Oban.Plugins.Cron,
    timezone: "America/New_York",
    crontab: [
      {"* * * * *", MyApp.MinuteWorker},
      {"0 * * * *", MyApp.HourlyWorker},
      {"0 0 * * *", MyApp.DailyWorker},
      {"0 12 * * MON", MyApp.MondayNoonWorker},
      {"@daily", MyApp.MidnightWorker},
      {"@reboot", MyApp.StartupWorker}
    ]}
]
```

## Smart Engine (Oban Pro)

If using Oban Pro, switch to Smart Engine for multi-node features:

```elixir
config :my_app, Oban,
  engine: Oban.Pro.Engines.Smart,
  queues: [
    default: [local_limit: 10, global_limit: 50],
    api_calls: [
      local_limit: 5,
      rate_limit: [allowed: 100, period: 60]
    ],
    media: [local_limit: 3, global_limit: 10]
  ]
```

- `local_limit` — per-node concurrency (replaces plain integer)
- `global_limit` — cluster-wide concurrency cap
- `rate_limit` — distributed rate limiting (check docs for algorithm/partition options)

### Pro Plugin Config

Pro plugins **enhance** OSS equivalents — swap the module name, don't run both:

```elixir
plugins: [
  {Oban.Pro.Plugins.DynamicCron, crontab: [{"0 0 * * *", DailyWorker}]},
  {Oban.Pro.Plugins.DynamicLifeline, rescue_interval: 60_000},
  {Oban.Pro.Plugins.DynamicPruner, mode: {:max_age, {7, :days}}}
  # Optional: DynamicQueues for runtime queue management
  # Optional: DynamicPrioritizer for starvation prevention
]
```

## Production Checklist

- [ ] Connection pool sized: `>= num_queues + sum(limits) + buffer`
- [ ] Pruner configured with `max_age`
- [ ] Lifeline plugin enabled for stuck jobs
- [ ] Telemetry attached for error tracking
- [ ] Graceful shutdown period set
- [ ] Unique constraints on user-triggered jobs
- [ ] All workers handle return values explicitly
- [ ] Idempotency for critical operations (payments, emails)

# Oban Testing Patterns Reference

> **Official docs**: <https://hexdocs.pm/oban/testing.html>
> **Oban Pro testing**: <https://hexdocs.pm/oban_pro/testing.html>

## Configuration

```elixir
# config/test.exs
config :my_app, Oban, testing: :manual  # Or :inline
```

## Test Helper Setup

```elixir
defmodule MyApp.DataCase do
  using do
    quote do
      use Oban.Testing, repo: MyApp.Repo
    end
  end
end
```

## Assert Enqueued

```elixir
test "enqueues welcome email on signup" do
  {:ok, user} = Accounts.create_user(%{email: "test@example.com"})

  assert_enqueued worker: MyApp.WelcomeWorker,
                  args: %{user_id: user.id},
                  queue: :mailers
end

# Scheduled with tolerance
assert_enqueued worker: MyApp.ReminderWorker,
                scheduled_at: {tomorrow, delta: 60}

# Assert NOT enqueued
refute_enqueued worker: MyApp.WelcomeWorker
```

## Execute Jobs

```elixir
test "processes valid order" do
  assert :ok = perform_job(MyApp.OrderWorker, %{order_id: 1})
end

test "cancels for missing order" do
  assert {:cancel, _} = perform_job(MyApp.OrderWorker, %{order_id: -1})
end
```

## Drain Queues

```elixir
test "full workflow processes correctly" do
  {:ok, _} = MyApp.start_import(file_path: "data.csv")

  assert %{success: 3, failure: 0} = Oban.drain_queue(
    queue: :imports,
    with_scheduled: true,
    with_recursion: true
  )
end
```

## Oban Pro Testing

> If project uses Oban Pro, use `Oban.Pro.Testing` instead of `Oban.Testing`.
> Pro.Testing API may vary between versions — check `mix hex.docs online oban_pro`.

### Setup

```elixir
# In test helper or DataCase
use Oban.Pro.Testing, repo: MyApp.Repo
```

### Drain Jobs

```elixir
# Pro uses drain_jobs/1 instead of drain_queue/2
assert %{success: 3, failure: 0} = drain_jobs()
drain_jobs(with_scheduled: true, with_recursion: true)
```

### Test Workflows and Batches

```elixir
# Workflow test — insert and drain
test "ETL workflow completes" do
  alias Oban.Pro.Workflow

  Workflow.new()
  |> Workflow.add(:extract, ExtractWorker.new(%{source: "test"}))
  |> Workflow.add(:load, LoadWorker.new(%{}), deps: [:extract])
  |> Oban.insert_all()

  assert %{success: 2} = drain_jobs()
end
```

---

## Anti-patterns

```elixir
# ❌ No idempotency for payments
def perform(%Job{args: %{"amount" => amount}}) do
  PaymentGateway.charge(amount)  # Will double-charge on retry!
end

# ✅ Idempotency key
def perform(%Job{args: %{"amount" => amount, "idempotency_key" => key}}) do
  case Payments.find_by_key(key) do
    {:ok, existing} -> {:ok, existing}
    :not_found -> PaymentGateway.charge(amount, idempotency_key: key)
  end
end

# ❌ Large data in args
%{file_content: large_binary}

# ✅ Store reference
%{file_path: "/uploads/abc123.csv"}

# ❌ No unique constraint for user actions
# Double-click creates duplicate jobs!

# ✅ Add unique constraint
unique: [period: {5, :minutes}, keys: [:user_id, :action]]
```

# Worker Patterns Reference

## Worker Options

```elixir
use Oban.Worker,
  queue: :mailers,           # Queue name
  max_attempts: 5,           # Retries before discarded
  priority: 1,               # 0-9, lower = higher priority
  tags: ["email"],           # For filtering/monitoring
  unique: [                  # Deduplication
    period: {5, :minutes},
    keys: [:user_id],
    states: [:available, :scheduled, :executing],
    fields: [:worker, :queue, :args]
  ]
```

## Unique Jobs (Deduplication)

```elixir
use Oban.Worker,
  unique: [
    period: {2, :minutes},     # Uniqueness window
    keys: [:user_id],          # Only compare these arg keys
    states: [:available, :scheduled, :executing],
    fields: [:worker, :queue, :args]
  ]
```

## Custom Backoff

```elixir
@impl Oban.Worker
def backoff(%Job{attempt: attempt}) do
  # Exponential with jitter
  trunc(:math.pow(attempt, 4) + 15 + :rand.uniform(30) * attempt)
end
```

## Custom Timeout

```elixir
@impl Oban.Worker
def timeout(_job), do: :timer.minutes(5)
```

## Idempotency Pattern

```elixir
defmodule MyApp.Workers.ChargeWorker do
  use Oban.Worker,
    queue: :payments,
    max_attempts: 3,
    unique: [period: {24, :hours}, keys: [:idempotency_key]]

  @impl Oban.Worker
  def perform(%Job{args: %{"idempotency_key" => key, "user_id" => user_id, "amount" => amount}}) do
    case Payments.find_by_idempotency_key(key) do
      {:ok, existing} ->
        {:ok, existing}

      :not_found ->
        Payments.charge(user_id, amount, idempotency_key: key)
    end
  end
end
```

## Error Handling & Telemetry

```elixir
# In application.ex or telemetry.ex
:telemetry.attach(
  "oban-errors",
  [:oban, :job, :exception],
  &MyApp.ObanErrorReporter.handle_event/4,
  []
)

defmodule MyApp.ObanErrorReporter do
  def handle_event([:oban, :job, :exception], _measure, %{job: job}, _config) do
    %{reason: exception, stacktrace: stacktrace} = job.unsaved_error

    Sentry.capture_exception(exception,
      stacktrace: stacktrace,
      extra: Map.take(job, [:id, :args, :queue, :worker]),
      tags: %{oban_worker: job.worker}
    )
  end
end
```

## Runtime Queue Control

```elixir
# Pause/resume
Oban.pause_queue(queue: :mailers)
Oban.resume_queue(queue: :mailers)

# Scale
Oban.scale_queue(queue: :mailers, limit: 50)

# Start new queue at runtime
Oban.start_queue(queue: :new_queue, limit: 10)
```

## Anti-patterns

```elixir
# ❌ Atom keys in args (JSON roundtrip converts to strings)
def perform(%Job{args: %{user_id: id}})  # Won't match!

# ✅ String keys
def perform(%Job{args: %{"user_id" => id}})

# ❌ Struct in args
%{user: %User{id: 1, name: "Jane"}}  # Can't serialize!

# ✅ Just the ID
%{user_id: 1}

# ❌ Silent failures
def perform(%Job{args: args}) do
  Mailer.send(args["email"])  # Ignores return value!
end

# ✅ Handle all outcomes
def perform(%Job{args: %{"email" => email}}) do
  case Mailer.send(email) do
    {:ok, _} -> :ok
    {:error, :invalid_email} -> {:cancel, "Invalid email"}
    {:error, reason} -> {:error, reason}
  end
end
```