mule - v0.1.0

mule/jobs/job

Types

Changes

</>

An updates list folded to one optional value per field — the analog of the changeset’s changes map, duplicates already collapsed to the last occurrence. Engines that render SQL consume this instead of the raw list so a repeated field never yields two SET clauses on one column.

pub type Changes {
  Changes(
    args: option.Option(json.Json),
    max_attempts: option.Option(Int),
    meta: option.Option(json.Json),
    priority: option.Option(Int),
    queue: option.Option(String),
    scheduled_at: option.Option(timestamp.Timestamp),
    tags: option.Option(List(String)),
  )
}

Constructors

Job

</> 
pub type Job {
  Job(
    id: Int,
    state: states.State,
    queue: String,
    worker: String,
    args: dynamic.Dynamic,
    meta: dynamic.Dynamic,
    tags: List(String),
    errors: List(error.Error),
    attempt: Int,
    attempted_by: List(String),
    max_attempts: Int,
    priority: Int,
    inserted_at: timestamp.Timestamp,
    scheduled_at: timestamp.Timestamp,
    attempted_at: option.Option(timestamp.Timestamp),
    cancelled_at: option.Option(timestamp.Timestamp),
    completed_at: option.Option(timestamp.Timestamp),
    discarded_at: option.Option(timestamp.Timestamp),
    unsaved_error: option.Option(exceptions.UnsavedError),
  )
}

Constructors

NewJob

</> 
pub type NewJob {
  NewJob(
    worker: String,
    args: json.Json,
    queue: String,
    meta: json.Json,
    tags: List(String),
    max_attempts: Int,
    priority: Int,
    scheduled_at: option.Option(timestamp.Timestamp),
    unique: option.Option(UniqueOptions),
    replace: List(#(states.State, List(ReplaceField))),
  )
}

Constructors

Option

</>

Mirrors Oban.Job.option() (job.ex): the keyword options accepted by Job.new/2, minus worker/args (positional here).

pub type Option {
  MaxAttempts(max_attempts: Int)
  Meta(meta: json.Json)
  Priority(priority: Int)
  Queue(queue: String)
  Replace(replace: List(#(states.State, List(ReplaceField))))
  ScheduleIn(schedule_in: duration.Duration)
  ScheduledAt(scheduled_at: timestamp.Timestamp)
  Tags(tags: List(String))
  Unique(unique: List(UniqueOption))
}

Constructors 

  • MaxAttempts(max_attempts: Int)
  • Meta(meta: json.Json)

    meta must be a JSON object: the uniqueness keys path runs jsonb_each over it, and non-object jsonb breaks containment matching.

  • Priority(priority: Int)
  • Queue(queue: String)
  • Replace(replace: List(#(states.State, List(ReplaceField))))

    Per-state conflict resolution for a unique insert (job.ex’s keyword replace form): on Conflict, overwrite the listed fields of the existing row — scoped by the state that row is in — with this insert’s values. replace_in_all_states / replace_args build the flat-list and replace_args: true shorthands. A duplicate state keeps the FIRST entry (Elixir’s Keyword.get).

  • ScheduleIn(schedule_in: duration.Duration)

    Relative scheduling: resolved to scheduled_at = now + schedule_in at construction time, and overriding any ScheduledAt regardless of list order (job.ex applies put_scheduling after cast).

  • ScheduledAt(scheduled_at: timestamp.Timestamp)
  • Tags(tags: List(String))
  • Unique(unique: List(UniqueOption))

    Unique sub-options folded over unique_defaults() (resolve_unique) — the typed analog of job.ex’s unique: [keyword] merged into @unique_defaults. Unique([]) is therefore the bare defaults; Elixir’s unique: true additionally sets an infinite period, so its analog is Unique([UniquePeriod(Infinity)]).

Query

</>

The typed analog of Oban.Job.query/1’s keyword filters. Each set field is a conjunct (all must match); None means unfiltered. List fields are IN matches, tags compares the complete array, args/meta are one-way jsonb containment (@>). Built pipeline-style from query() through the filter_* combinators; a repeated combinator call replaces the earlier value for that field. limit caps how many rows an operation touches (the analog of composing Ecto’s limit onto the queryable), applied after the deterministic order-by-id.

pub type Query {
  Query(
    ids: option.Option(List(Int)),
    states: option.Option(List(states.State)),
    queues: option.Option(List(String)),
    workers: option.Option(List(String)),
    priorities: option.Option(List(Int)),
    tags: option.Option(List(String)),
    args: option.Option(json.Json),
    meta: option.Option(json.Json),
    limit: option.Option(Int),
  )
}

Constructors

ReplaceField

</>

Mirrors job.ex’s @replace_options: the existing row’s fields a unique Conflict may overwrite with the conflicting insert’s values.

pub type ReplaceField {
  ReplaceArgs
  ReplaceMaxAttempts
  ReplaceMeta
  ReplacePriority
  ReplaceQueue
  ReplaceScheduledAt
  ReplaceTags
  ReplaceWorker
}

Constructors 

  • ReplaceArgs
  • ReplaceMaxAttempts
  • ReplaceMeta
  • ReplacePriority
  • ReplaceQueue
  • ReplaceScheduledAt
  • ReplaceTags
  • ReplaceWorker

UniqueField

</> 
pub type UniqueField {
  ByWorker
  ByQueue
  ByArgs
  ByMeta
}

Constructors 

  • ByWorker
  • ByQueue
  • ByArgs
  • ByMeta

UniqueOption

</>

One unique: [...] keyword — the partial form the Unique option and worker defaults carry, so worker.merge_options can deep-merge a worker’s unique sub-options with a caller’s (Worker.merge_opts/2’s :unique clause) instead of replacing the whole record.

pub type UniqueOption {
  UniqueFields(fields: List(UniqueField))
  UniqueKeys(keys: List(String))
  UniquePeriod(period: UniquePeriod)
  UniqueStates(states: set.Set(states.State))
  UniqueTimestamp(timestamp: UniqueTimestamp)
}

Constructors

UniqueOptions

</>

Mirrors job.ex’s @unique_defaults shape. keys narrows args/meta matching to the named top-level JSON keys (Elixir’s atom keys are strings here); period bounds how far back a match may lie, measured against timestamp.

pub type UniqueOptions {
  UniqueOptions(
    fields: List(UniqueField),
    keys: List(String),
    period: UniquePeriod,
    states: set.Set(states.State),
    timestamp: UniqueTimestamp,
  )
}

Constructors

UniquePeriod

</>

Mirrors unique_period(): a finite window or :infinity.

pub type UniquePeriod {
  Period(window: duration.Duration)
  Infinity
}

Constructors

UniqueStateGroup

</>

Mirrors Oban.Job.unique_states/1’s named groups.

pub type UniqueStateGroup {
  All
  Incomplete
  Scheduled
  Successful
}

Constructors 

  • All
  • Incomplete
  • Scheduled
  • Successful

UniqueTimestamp

</>

Which timestamp the period window is measured against (mirrors unique_timestamp(): :inserted_at | :scheduled_at). By* prefix because a bare ScheduledAt constructor would collide with the Option variant in this module.

pub type UniqueTimestamp {
  ByInsertedAt
  ByScheduledAt
}

Constructors 

  • ByInsertedAt
  • ByScheduledAt

Update

</>

Mirrors job.ex’s @updatable_params (minus worker — the port pairs a worker’s name with its typed args codec, so retargeting the name alone is deliberately unrepresentable): the fields mule.update_job may change. A duplicate keeps the last occurrence, like new_with options.

pub type Update {
  UpdateArgs(args: json.Json)
  UpdateMaxAttempts(max_attempts: Int)
  UpdateMeta(meta: json.Json)
  UpdatePriority(priority: Int)
  UpdateQueue(queue: String)
  UpdateScheduledAt(scheduled_at: timestamp.Timestamp)
  UpdateTags(tags: List(String))
}

Constructors 

  • UpdateArgs(args: json.Json)
  • UpdateMaxAttempts(max_attempts: Int)
  • UpdateMeta(meta: json.Json)
  • UpdatePriority(priority: Int)
  • UpdateQueue(queue: String)
  • UpdateScheduledAt(scheduled_at: timestamp.Timestamp)
  • UpdateTags(tags: List(String))

ValidationError

</>

One failed new_with validation: the field that failed and why (the analog of an Ecto changeset error from Job.new/2).

pub type ValidationError {
  ValidationError(field: String, message: String)
}

Constructors 

  • ValidationError(field: String, message: String)

Values

filter_args

</> 
pub fn filter_args(
  query query: Query,
  args args: json.Json,
) -> Query

One-way jsonb containment (args @> filter) — NOT the mutual containment the unique-insert path uses. An empty object {} therefore matches every job (true @> semantics; Elixir raises on an empty filter instead).

filter_id

</> 
pub fn filter_id(query query: Query, ids ids: List(Int)) -> Query

An empty list matches nothing — the typed, total-function analog of Elixir raising ArgumentError on empty filter lists. The same holds for the other list combinators below (except filter_tags).

filter_meta

</> 
pub fn filter_meta(
  query query: Query,
  meta meta: json.Json,
) -> Query

One-way jsonb containment (meta @> filter); see filter_args.

filter_priority

</> 
pub fn filter_priority(
  query query: Query,
  priorities priorities: List(Int),
) -> Query

filter_queue

</> 
pub fn filter_queue(
  query query: Query,
  queues queues: List(String),
) -> Query

filter_state

</> 
pub fn filter_state(
  query query: Query,
  states states: List(states.State),
) -> Query

filter_tags

</> 
pub fn filter_tags(
  query query: Query,
  tags tags: List(String),
) -> Query

Compares the complete tags array, element-order-sensitive, exactly like Elixir’s job.tags == ^value — so filter_tags(query, []) legitimately matches jobs with no tags.

filter_worker

</> 
pub fn filter_worker(
  query query: Query,
  workers workers: List(String),
) -> Query

format_attempt

</> 
pub fn format_attempt(job job: Job) -> option.Option(error.Error)

Flatten unsaved_error into the persisted errors-column shape (job.ex format_attempt/1). None when no unsaved error is recorded.

json_recode

</> 
pub fn json_recode(value value: json.Json) -> dynamic.Dynamic

Mirror the SQL wire path (Elixir’s json_recode in inline.ex/testing.ex): encode the JSON, parse it back to the opaque Dynamic the read side carries — an unpersisted job’s args/meta end up shaped exactly like a stored row’s. Safe assert: the JSON was just produced by the caller’s encoder. Shared by this module’s update path, the in-memory and inline engines, and mule/testing.

limit

</> 
pub fn limit(query query: Query, limit limit: Int) -> Query

Cap how many rows the operation touches, in id order (the analog of composing Ecto’s limit(n) onto Job.query/1).

merge_updates

</> 
pub fn merge_updates(updates updates: List(Update)) -> Changes

new

</> 
pub fn new(worker: String, args: json.Json) -> NewJob

new_with

</> 
pub fn new_with(
  worker worker: String,
  args args: json.Json,
  options options: List(Option),
) -> Result(NewJob, ValidationError)

Job.new/2 with options: fold each option over the defaults (a duplicate option keeps the last occurrence, like Map.new(opts)), then validate the same bounds job.ex does — worker/queue length 1–128, max_attempts > 0, priority 0–9.

query

</> 
pub fn query() -> Query

Matches every job — the Oban.Job bare queryable.

replace_args

</> 
pub fn replace_args() -> Option

The replace_args: true shorthand: replace args in every state.

replace_in_all_states

</> 
pub fn replace_in_all_states(
  fields fields: List(ReplaceField),
) -> Option

The flat-list replace: [fields] form: the same fields in every state (job.ex put_replace’s with_states).

resolve_unique

</> 
pub fn resolve_unique(
  options options: List(UniqueOption),
) -> UniqueOptions

Fold unique sub-options over unique_defaults(), a duplicate keeping the last — job.ex put_unique’s @unique_defaults |> Map.merge(Map.new(opts)).

unique_defaults

</> 
pub fn unique_defaults() -> UniqueOptions

Mirrors job.ex’s @unique_defaults. Note the states are NOT unique_states(Successful) — Elixir’s default set lacks suspended.

unique_states

</> 
pub fn unique_states(
  group: UniqueStateGroup,
) -> set.Set(states.State)

update

</> 
pub fn update(
  job job: Job,
  updates updates: List(Update),
) -> Result(Job, ValidationError)

Job.update/2: fold validated changes over the job. UpdateScheduledAt also moves the job to scheduled — job.ex’s update never casts state, so normalize_state fires on ANY scheduled_at change.

validate_changes

</> 
pub fn validate_changes(
  changes changes: Changes,
) -> Result(Changes, ValidationError)

Job.update/2’s validations — the same bounds new_with checks, applied to only the changed fields: queue length 1–128, max_attempts > 0, priority 0–9. Tags are NOT normalized (job.ex’s update casts them without normalize_tags).

✨ Search Document