Oban.Pro.Backfill.Cursor (Oban Pro v1.8.0-rc.0)

Process a backfill query in ordered, limited windows.

Each helper runs an operation over the next batch of rows in a query and returns either:

:halt — nothing matched, so the chain stops
{:cont, next, count} — next is the advanced cursor and count is the number of affected rows

Pass a filtered query and the current cursor to a helper. Use update_all/4 or delete_all/3 for bulk database operations, each/4 for per-row work, or fetch/3 when you need the rows for batching or concurrent processing. Each helper limits and orders the query according to the backfill's options, then returns the cursor value for the next window.

See Oban.Pro.Backfill for how to define and run a backfill.

Options

Every helper accepts:

:repo — the Ecto repo to run against. Defaults to the executing backfill job's configured repo. Pass an explicit repo for cross-database backfills.

Summary

Types

advance()

cursor()

Functions

delete_all(query, cursor, opts \\ [])

Delete every row in the cursor's window, then advance.

each(query, cursor, fun, opts \\ [])

Run a function over each row in the cursor's window, then advance.

fetch(query, cursor, opts \\ [])

Fetch the cursor's window of rows and the cursor for the next window.

update_all(query, cursor, updates, opts \\ [])

Update every row in the cursor's window, then advance.

Types

advance()

(since 1.8.0)

@type advance() :: :halt | {:cont, next :: cursor(), count :: non_neg_integer()}

cursor()

(since 1.8.0)

@type cursor() :: Oban.Pro.Backfill.cursor()

Functions

delete_all(query, cursor, opts \\ [])

(since 1.8.0)

@spec delete_all(Ecto.Queryable.t(), cursor(), keyword()) :: advance()

Delete every row in the cursor's window, then advance.

Returns :halt when nothing matched, or {:cont, next, count} with the cursor for the next window and the number of rows deleted.

Example

MyApp.Event
|> where([e], e.expired)
|> Cursor.delete_all(cursor)

each(query, cursor, fun, opts \\ [])

(since 1.8.0)

@spec each(Ecto.Queryable.t(), cursor(), (term() -> term()), keyword()) :: advance()

Run a function over each row in the cursor's window, then advance.

Calls fun once per row and returns :halt when the window is empty, or {:cont, next, count} with the cursor for the next window and the number of rows processed. Use it for per-row work that isn't a single database operation, such as an external request.

Example

Cursor.each(MyApp.User, cursor, fn user ->
  MyApp.Crm.sync(user)
end)

fetch(query, cursor, opts \\ [])

(since 1.8.0)

@spec fetch(Ecto.Queryable.t(), cursor(), keyword()) :: :halt | {[term()], cursor()}

Fetch the cursor's window of rows and the cursor for the next window.

Returns :halt when nothing matched, otherwise {rows, next_cursor}. Reach for this over each/4 when you want the rows in hand—to batch them through a single request, or fan them out with Task.async_stream/3.

Example

with {users, next} <- Cursor.fetch(MyApp.User, cursor) do
  MyApp.Crm.sync_all(users)

  {:cont, next, length(users)}
end

update_all(query, cursor, updates, opts \\ [])

(since 1.8.0)

@spec update_all(Ecto.Queryable.t(), cursor(), keyword(), keyword()) :: advance()

Update every row in the cursor's window, then advance.

Returns :halt when nothing matched, or {:cont, next, count} with the cursor for the next window and the number of rows updated.

Example

MyApp.User
|> where([u], is_nil(u.migrated_at))
|> Cursor.update_all(cursor, set: [migrated_at: ^DateTime.utc_now()])