@spec count_jobs(
  keyword(),
  atom()
) :: {:ok, non_neg_integer()} | {:error, Zizq.Error.t()}

Count the jobs matching a set of filters.

Zizq.count_jobs([queue: "emails", status: [:ready]], MyApp.Zizq)
#=> {:ok, 1_284}

Counting is a separate endpoint rather than the length of a listing, so a count costs one request whatever the total.

@spec count_jobs!(
  keyword(),
  atom()
) :: non_neg_integer()

Count jobs, raising on failure.

@spec delete_all_crons(atom()) :: {:ok, non_neg_integer()} | {:error, Zizq.Error.t()}

Delete every cron schedule, and return how many were removed.

@spec delete_all_jobs(
  keyword(),
  atom()
) :: {:ok, non_neg_integer()} | {:error, Zizq.Error.t()}

Delete every job matching a set of filters, and return how many.

Zizq.delete_all_jobs([queue: "emails", status: :dead], MyApp.Zizq)
#=> {:ok, 17}

Selection uses the filters in Zizq.Filter.

Filters restrict what is deleted the way a WHERE clause does, and are optional for the same reason: delete_all_jobs([], client) empties the server, deliberately.

Counting first with the same filters is a cheap way to see what would go:

filters = [queue: "emails", status: :dead]
{:ok, 17} = Zizq.count_jobs(filters, MyApp.Zizq)
{:ok, 17} = Zizq.delete_all_jobs(filters, MyApp.Zizq)

@spec delete_all_jobs!(
  keyword(),
  atom()
) :: non_neg_integer()

Delete every matching job, raising on failure.

@spec delete_cron(Zizq.Cron.t() | String.t(), atom()) ::
  :ok | {:error, Zizq.Error.t()}

Delete a cron schedule and everything all its entries.

Jobs it already enqueued are unaffected.

@spec delete_cron!(Zizq.Cron.t() | String.t(), atom()) :: :ok

Delete a cron schedule, raising on failure.

@spec delete_cron_entry(
  keyword(),
  atom()
) :: :ok | {:error, Zizq.Error.t()}

Delete one cron entry, leaving the rest of its schedule alone.

Zizq.delete_cron_entry([cron: "my_app", entry: "digest"], MyApp.Zizq)

The server-side counterpart to Zizq.Cron.delete_entry/2, which changes a schedule in memory you then replace whole on the server.

@spec delete_cron_entry!(
  keyword(),
  atom()
) :: :ok

Delete one cron entry, raising on failure.

@spec delete_job(Zizq.Job.t() | String.t(), atom()) :: :ok | {:error, Zizq.Error.t()}

Delete a job outright.

Zizq.delete_job(job, MyApp.Zizq)
#=> :ok

Unlike report_failure/3 with kill: true, which leaves a dead job behind to be inspected, this removes it. A job the server no longer holds is {:error, %Zizq.Error{reason: :not_found}}.

@spec delete_job!(Zizq.Job.t() | String.t(), atom()) :: :ok

Delete a job outright, raising on failure.

@spec enqueue(Zizq.Enqueue.t() | keyword() | map(), atom()) ::
  {:ok, Zizq.Job.t()} | {:error, Zizq.Error.t()}

Enqueue a job.

Accepts a Zizq.Enqueue struct, or a keyword list or map of the same fields. Only :type is required; see Zizq.Enqueue for the rest.

Zizq.Enqueue.new!(type: "send_email", payload: %{"user_id" => 42})
|> Zizq.enqueue(MyApp.Zizq)
#=> {:ok, %Zizq.Job{id: "03gn…", status: :ready}}

The job comes first and the client second so that enqueues pipe, which is how they are normally written once job modules are building them.

Returns the job the server recorded, so its :id and server-assigned defaults are available immediately. Note that the returned job carries no :payload — the server omits it from enqueue responses.

Raises ArgumentError for an invalid enqueue, since that is a bug in the calling code rather than a runtime condition to handle.

@spec enqueue!(Zizq.Enqueue.t() | keyword() | map(), atom()) :: Zizq.Job.t()

Enqueue a job, raising on failure.

Suits call sites where a failed enqueue should abort the surrounding work, such as inside a transaction.

@spec enqueue_all([Zizq.Enqueue.t() | keyword() | map()], atom()) ::
  {:ok, [Zizq.Job.t()]} | {:error, Zizq.Error.t()}

Enqueue many jobs in a single atomic bulk request.

Each element may be a Zizq.Enqueue struct, a keyword list, or a map, exactly as enqueue/2 accepts.

users
|> Enum.map(&Zizq.Enqueue.new!(type: "send_email", payload: %{"user_id" => &1.id}))
|> Zizq.enqueue_all(MyApp.Zizq)
#=> {:ok, [%Zizq.Job{}, ...]}

Jobs are returned in the order they were sent. An empty list short circuits immediately without contacting the server.

Note that the returned jobs carry no :payload — the server omits it from enqueue responses.

@spec enqueue_all!([Zizq.Enqueue.t() | keyword() | map()], atom()) :: [Zizq.Job.t()]

Bulk enqueue many jobs atomically, raising on failure.

@spec erase_all_data(atom()) :: :ok | {:error, Zizq.Error.t()}

Delete every job and every cron schedule on the server.

Zizq.erase_all_data(MyApp.Zizq)
#=> :ok

Intended as a setup or teardown step in tests and development, where a known-empty server between scenarios is worth more than the data. It is one request rather than delete_all_jobs/2 followed by delete_all_crons/1.

@spec erase_all_data!(atom()) :: :ok

Delete every job and schedule, raising on failure.

@spec get_cron(Zizq.Cron.t() | String.t(), atom()) ::
  {:ok, Zizq.Cron.t()} | {:error, Zizq.Error.t()}

Read a cron schedule and its entries.

@spec get_cron!(Zizq.Cron.t() | String.t(), atom()) :: Zizq.Cron.t()

Read a cron schedule, raising on failure.

@spec get_error(Zizq.Job.t() | String.t(), pos_integer(), atom()) ::
  {:ok, Zizq.ErrorRecord.t()} | {:error, Zizq.Error.t()}

Read what went wrong on one attempt.

Zizq.get_error(job, 1, MyApp.Zizq)
#=> {:ok, %Zizq.ErrorRecord{attempt: 1, message: "SMTP timeout"}}

Attempts count from 1. An attempt that never failed — or never happened — is {:error, %Zizq.Error{reason: :not_found}}.

@spec get_error!(Zizq.Job.t() | String.t(), pos_integer(), atom()) ::
  Zizq.ErrorRecord.t()

Read one attempt's error, raising on failure.

@spec get_job(Zizq.Job.t() | String.t(), atom()) ::
  {:ok, Zizq.Job.t()} | {:error, Zizq.Error.t()}

Read one job by id.

Zizq.get_job(job.id, MyApp.Zizq)
#=> {:ok, %Zizq.Job{status: :completed}}

Unlike the job returned by enqueue/2, this one carries its :payload.

A job the server no longer holds is {:error, %Zizq.Error{reason: :not_found}} — which a completed job becomes as soon as its retention expires, immediately by default.

@spec get_job!(Zizq.Job.t() | String.t(), atom()) :: Zizq.Job.t()

Read one job by id, raising on failure.

@spec list_crons(atom()) :: {:ok, [String.t()]} | {:error, Zizq.Error.t()}

List the names of every cron schedule on the server.

@spec list_crons!(atom()) :: [String.t()]

List cron schedule names, raising on failure.

@spec list_errors(Zizq.Job.t() | String.t(), atom(), keyword()) ::
  {:ok, Zizq.ErrorPage.t()} | {:error, Zizq.Error.t()}

List a job's failed attempts, oldest first.

Zizq.list_errors(job, MyApp.Zizq)
#=> {:ok, %Zizq.ErrorPage{errors: [%Zizq.ErrorRecord{attempt: 1}, ...]}}

Options

• :limit — records per page, 1 to 200. The server decides if unset.
• :order — :asc (first attempt first) or :desc (most recent first). The server defaults to :asc.

Pages follow with next_page/2, the same as a job listing.

Errors live as long as the job does, so a completed job whose retention has expired takes its failure history with it.

@spec list_errors!(Zizq.Job.t() | String.t(), atom(), keyword()) :: Zizq.ErrorPage.t()

List a job's failed attempts, raising on failure.

@spec list_jobs(
  keyword(),
  atom()
) :: {:ok, Zizq.JobPage.t()} | {:error, Zizq.Error.t()}

List jobs, oldest first, one page at a time.

Narrow with any of the filters in Zizq.Filter:

Zizq.list_jobs([queue: "emails", status: [:ready]], MyApp.Zizq)
#=> {:ok, %Zizq.JobPage{jobs: [%Zizq.Job{}, ...]}}

Options

Every option Zizq.Filter documents, plus:

• :limit — jobs per page, 1 to 2000. The server decides if unset.
• :order — :asc (oldest first) or :desc (newest first). The server defaults to :asc.

Follow the pages with next_page/2 and prev_page/2.

@spec list_jobs!(
  keyword(),
  atom()
) :: Zizq.JobPage.t()

List jobs, raising on failure.

@spec list_queues(atom()) :: {:ok, [String.t()]} | {:error, Zizq.Error.t()}

List the queues the server currently holds jobs for.

Zizq.list_queues(MyApp.Zizq)
#=> {:ok, ["default", "emails"]}

Queues are not declared — one exists because a job named it — so this reports what is there rather than what was configured.

@spec list_queues!(atom()) :: [String.t()]

List queues, raising on failure.

@spec next_page(page, atom()) :: {:ok, page | nil} | {:error, Zizq.Error.t()}
when page: Zizq.JobPage.t() | Zizq.ErrorPage.t()

Fetch the page after this one, or nil at the end of the listing.

{:ok, page} = Zizq.list_jobs([queue: "emails"], MyApp.Zizq)
{:ok, next} = Zizq.next_page(page, MyApp.Zizq)

The link carries the cursor and the original filters.

@spec pause_cron(Zizq.Cron.t() | String.t(), atom()) ::
  {:ok, Zizq.Cron.t()} | {:error, Zizq.Error.t()}

Suspend a whole cron schedule. Its entries stop firing until resumed.

@spec pause_cron_entry(
  keyword(),
  atom()
) :: {:ok, Zizq.CronEntry.t()} | {:error, Zizq.Error.t()}

Suspend one cron entry, leaving the rest of its schedule running.

Zizq.pause_cron_entry([cron: "my_app", entry: "digest"], MyApp.Zizq)

Changes only that entry, on the server, so it is safe while an application is running, unlike reading a schedule, amending it and replacing it, which is last-write-wins.

Named rather than positional for clarity.

@spec prev_page(page, atom()) :: {:ok, page | nil} | {:error, Zizq.Error.t()}
when page: Zizq.JobPage.t() | Zizq.ErrorPage.t()

Fetch the page before this one, or nil at the start of the listing.

{:ok, prev} = Zizq.prev_page(page, MyApp.Zizq)

@spec query(atom()) :: Zizq.Query.t()

Start a composable query.

Zizq.query(MyApp.Zizq)
|> Zizq.Query.where(queue: "emails", status: [:ready])
|> Enum.take(10)

Builds nothing and sends nothing until the query is run. See Zizq.Query.

@spec replace_cron(Zizq.Cron.t(), atom()) ::
  {:ok, Zizq.Cron.t()} | {:error, Zizq.Error.t()}

Install a cron schedule, replacing whatever was there.

This is the call to make at application startup:

Zizq.Cron.new("my_app",
  entries: [
    [name: "nightly_cleanup",
     expression: "0 3 * * *",
     job: MyApp.Cleanup.new(%{})],
    [name: "digest",
     expression: "*/15 * * * *",
     timezone: "Australia/Melbourne",
     job: [type: "digest", queue: "reports"]]
  ]
)
|> Zizq.replace_cron(MyApp.Zizq)

Or alternatively pipelined:

Zizq.Cron.new("my_app")
|> Zizq.Cron.put_entry(
  name: "nightly_cleanup",
  expression: "0 3 * * *",
  job: MyApp.Cleanup.new(%{})
)
|> Zizq.Cron.put_entry(
  name: "digest",
  expression: "*/15 * * * *",
  timezone: "Australia/Melbourne",
  job: [type: "digest", queue: "reports"]
)
|> Zizq.replace_cron(MyApp.Zizq)

It is atomic and idempotent, so every instance of an application can run it on boot without coordinating — none of them needs to be the one that owns the schedule.

The group is created if it does not exist, and entries left out are removed, so a Zizq.Cron is the whole schedule rather than an addition to it. That is what makes running it on every boot converge rather than accumulate.

Once configured, there is no futher integration required. Your Zizq.Worker process receives jobs enqueued via the schedule just like any other job.

Cron needs a Pro licence; without one the server answers 403, which arrives as %Zizq.Error{reason: :forbidden}.

@spec replace_cron!(Zizq.Cron.t(), atom()) :: Zizq.Cron.t()

Install a cron schedule, raising on failure.

@spec report_failure(Zizq.Job.t() | String.t(), atom(), keyword()) ::
  {:ok, Zizq.Job.t()} | {:error, Zizq.Error.t()}

Report a job as failed.

The server decides what happens next — reschedule with backoff, or declare the job dead once its retry limit is spent — and returns the job as it now stands, so the new :status and :attempts are visible immediately.

Zizq.report_failure(job, MyApp.Zizq, message: "SMTP timeout")

Options

• :message — what went wrong. Required.
• :error_type — an exception or error class name, e.g. "Mint.TransportError".
• :backtrace — a formatted stacktrace.
• :kill — when true, declare the job dead now regardless of how many attempts remain.
• :retry_at — a DateTime.t/0 (or Unix milliseconds) to retry at, bypassing the backoff policy. Reschedules the job even if its retry limit is spent.

:kill and :retry_at are what a handler's {:cancel, reason} and {:snooze, milliseconds} results map onto.

@spec report_success(Zizq.Job.t() | String.t(), atom()) ::
  :ok | {:error, Zizq.Error.t()}

Report a job as completed successfully.

Accepts a Zizq.Job or a job id.

Zizq.report_success(job, MyApp.Zizq)
#=> :ok

A job the server no longer holds in flight — already acknowledged, or redelivered elsewhere after its visibility timeout — answers {:error, %Zizq.Error{reason: :not_found}}. That is usually benign rather than a failure: the work is done either way, and nothing is gained by retrying.

@spec report_success_all([Zizq.Job.t() | String.t()], atom()) ::
  {:ok, [String.t()]} | {:error, Zizq.Error.t()}

Report many jobs as completed in a single request.

Returns the ids the server did not recognise, so {:ok, []} means every one was acknowledged.

Zizq.report_success_all(jobs, MyApp.Zizq)
#=> {:ok, []}

A partial result is a success, not an error: the jobs the server did recognise were completed. Only the unrecognised ids come back, and those are typically jobs already acknowledged or redelivered elsewhere. An empty list short-circuits without contacting the server.

@spec resume_cron(Zizq.Cron.t() | String.t(), atom()) ::
  {:ok, Zizq.Cron.t()} | {:error, Zizq.Error.t()}

Resume a suspended cron schedule.

@spec resume_cron_entry(
  keyword(),
  atom()
) :: {:ok, Zizq.CronEntry.t()} | {:error, Zizq.Error.t()}

Resume one suspended cron entry.

@spec server_version(atom()) :: {:ok, String.t()} | {:error, Zizq.Error.t()}

Ask the server for its version.

Useful as a liveness check — it is the cheapest endpoint the server exposes that proves the connection works end to end.

Zizq.server_version(MyApp.Zizq)
#=> {:ok, "0.6.1"}

@spec start_link(keyword()) :: Supervisor.on_start()

Start a client and add it to your supervision tree.

children = [
  {Zizq, name: MyApp.Zizq, url: "http://localhost:7890"}
]

The :name both names the supervisor and is the handle you pass to every other function in this module. Several clients can run side by side under different names.

Options

• :name (atom/0) - Required. Name for this client instance. Also names the supervisor, and is the handle passed to Zizq.enqueue/2 and friends.

• :url - Required. Base URL of the Zizq server, e.g. "http://localhost:7890", as a string or a URI. A path is allowed and is treated as a prefix, for servers behind a reverse proxy.

• :format (atom/0) - Serialization format: :msgpack (default), :json, or a module implementing the Zizq.Codec behaviour. The default value is :msgpack.

• :pool_count (pos_integer/0) - Number of HTTP/2 connections to the server. Each is fully multiplexed, so one is usually enough; raise it only if a single connection becomes a bottleneck. The default value is 1.

• :connect_timeout (timeout/0) - Milliseconds to wait for a connection to be established. The default value is 5000.

• :receive_timeout (timeout/0) - Milliseconds to wait for a response. Does not apply to streaming endpoints. The default value is 15000.

• :stream_idle_timeout (timeout/0) - Milliseconds a streaming connection may go without any data before it is treated as dead and reconnected.

  The server sends heartbeat frames on an otherwise idle stream specifically so this can be detected, so the timeout only has to exceed that interval. The default is ten times the server's own default of three seconds. Raise it if the server runs with a longer heartbeat interval, since a timeout shorter than the heartbeat would reconnect a perfectly healthy connection on a loop.

  The default value is 30000.

• :tls (keyword/0) - Certificates for connecting over HTTPS.

  • :ca — the certificate authority to verify the server against. Pins verification to this CA instead of the system's trust store.
  • :client_cert and :client_key — the client identity to present for mutual TLS. Both are needed together.

  Each value may be the PEM contents or a path to a PEM file.

  {Zizq,
   name: MyApp.Zizq,
   url: "https://zizq.internal:7890",
   tls: [
     ca: "/etc/zizq/ca.pem",
     client_cert: "/etc/zizq/client.pem",
     client_key: "/etc/zizq/client-key.pem"
   ]}

  HTTPS works without this — connections verify against the system trust store by default. It is only needed to pin a private CA or to present a client certificate.

  Mutual TLS requires a Zizq Pro licence on the server.

  The default value is [].

  • :ca (String.t/0)

  • :client_cert (String.t/0)

  • :client_key (String.t/0)

@spec update_all_jobs(
  keyword(),
  atom()
) :: {:ok, non_neg_integer()} | {:error, Zizq.Error.t()}

Change every job matching a set of filters, and return how many.

Zizq.update_all_jobs(
  [where: [queue: "emails", status: :scheduled], apply: [ready_at: nil]],
  MyApp.Zizq
)
#=> {:ok, 42}

Options

• :where — which jobs to change, using the filters in Zizq.Filter. Restricts the operation the way a WHERE clause does, and is optional for the same reason: left out, every job is changed, deliberately.
• :apply — what to change, using the options update_job/3 takes, with the same merge-patch rules. An omitted option leaves a field alone, nil clears it. Required.

Named rather than positional because both halves are keyword lists of overlapping keys — queue: and priority: mean something on each side — so transposing them could quietly change the wrong jobs rather than fail.

Finished jobs cannot be changed, so asking for one is an error rather than a silent no-op: status: :completed or status: :dead in :where is rejected before the request is sent.

@spec update_all_jobs!(
  keyword(),
  atom()
) :: non_neg_integer()

Change every matching job, raising on failure.

@spec update_job(Zizq.Job.t() | String.t(), atom(), keyword()) ::
  {:ok, Zizq.Job.t()} | {:error, Zizq.Error.t()}

Change a job that has not finished yet, and return it as it now stands.

Zizq.update_job(job, MyApp.Zizq, queue: "urgent", priority: 0)

Options

Only the options given are touched; the rest of the job is left alone. Passing nil clears a field to the server's default, which is why an option must be omitted rather than set to nil to leave it as it is:

# Retry with the server's default limit, whatever it now is.
Zizq.update_job(job, MyApp.Zizq, retry_limit: nil)

• :queue — move the job to another queue. Cannot be nil.
• :priority — lower runs sooner. Cannot be nil.
• :ready_at — a DateTime or Unix milliseconds. nil makes the job ready immediately.
• :retry_limit — nil restores the server default.
• :backoff — a Zizq.Backoff or keyword list. nil restores the server default.
• :retention — a Zizq.Retention or keyword list, merged field by field, so retention: [completed: :timer.hours(1)] leaves :dead alone. nil clears the whole override.

Only jobs that have not finished can be changed: the server rejects a completed or dead job with %Zizq.Error{reason: :invalid_request}.

@spec update_job!(Zizq.Job.t() | String.t(), atom(), keyword()) :: Zizq.Job.t()

Change a job, raising on failure.

@spec version() :: String.t()

Returns the version of this client as a string.

Examples

iex> Zizq.version()
"0.6.1"