@spec add_delayed_job(atom(), queue_context(), map() | struct(), map()) ::
  script_result()

Adds a delayed job to the queue.

@spec add_log(
  atom(),
  queue_context(),
  String.t(),
  String.t(),
  non_neg_integer() | nil
) ::
  script_result()

Adds a log entry to a job.

@spec add_parent_job(atom(), queue_context(), map() | struct(), map()) ::
  script_result()

Adds a parent job to the queue (waiting-children state).

Parent jobs are added in waiting-children state until all children complete. Used by FlowProducer to create job hierarchies.

@spec add_prioritized_job(atom(), queue_context(), map() | struct(), map()) ::
  script_result()

Adds a prioritized job to the queue.

@spec add_standard_job(atom(), queue_context(), map() | struct(), map()) ::
  script_result()

Adds a standard job to the queue.

@spec add_standard_jobs_pipelined(atom(), queue_context(), [{map() | struct(), map()}]) ::
  {:ok, [String.t()]} | {:error, term()}

Adds multiple standard jobs atomically in a single transaction (MULTI/EXEC). Much more efficient than calling add_standard_job multiple times.

This operation is atomic - all jobs are added or none are.

Returns {:ok, job_ids} on success or {:error, reason} on failure.

@spec build_add_parent_job_command(queue_context(), map() | struct(), map()) ::
  {:ok, [String.t()]}

Builds a command for adding a parent job without executing it. Used for building flow transactions where all jobs are added atomically.

@spec build_add_standard_job_command(queue_context(), map() | struct(), map()) ::
  {:ok, [String.t()]}

Builds a command for adding a standard job without executing it. Used for pipelining multiple job additions.

@spec build_bulk_add_commands(queue_context(), [{map() | struct(), map()}]) ::
  {:ok, [{map() | struct(), [String.t()]}]} | {:error, term()}

Builds multiple add_standard_job commands efficiently by precomputing shared data. Much faster than calling build_add_standard_job_command multiple times.

Returns a list of {job, command} tuples.

@spec build_command(script_name(), [String.t()], [any()]) ::
  {:ok, [String.t()]} | {:error, term()}

Builds a Redis command for a Lua script without executing it. Useful for pipelining multiple script calls.

Returns {:ok, command} where command is a list that can be passed to Redis pipeline, or {:error, reason} if the script is not found.

@spec drain(atom(), queue_context(), boolean()) :: script_result()

Drains the queue (removes all jobs).

@spec ensure_scripts_loaded(atom(), [script_name()]) :: :ok | {:error, term()}

Ensures scripts are loaded into Redis cache by executing a dummy SCRIPT LOAD. Call this before using pipelined operations to avoid NOSCRIPT errors.

@spec execute(atom(), script_name(), [String.t()], [any()]) :: script_result()

Executes a Lua script against Redis by name.

Uses EVALSHA for efficiency, falling back to EVAL if the script is not yet cached in Redis.

Parameters

• conn - The Redis connection pool name
• script_name - The script name as an atom
• keys - List of Redis keys
• args - List of arguments to pass to the script

Returns

• {:ok, result} on success
• {:error, reason} on failure

@spec execute_pipeline(atom(), [[String.t()]]) :: {:ok, [any()]} | {:error, term()}

Executes multiple script commands in a pipeline. Returns a list of results in the same order as the commands.

Note: If any script is not cached (NOSCRIPT error), this will fail. Use ensure_scripts_loaded/2 first to cache scripts.

@spec execute_raw(atom(), String.t(), [String.t()], [any()]) :: script_result()

Executes a raw Lua script against Redis.

@spec execute_transaction(atom(), [[String.t()]]) :: {:ok, [any()]} | {:error, term()}

Executes multiple script commands in a Redis transaction (MULTI/EXEC). All commands are executed atomically - either all succeed or none do.

Returns {:ok, results} on success or {:error, reason} on failure.

Note: If any script is not cached (NOSCRIPT error), this will fail. Use ensure_scripts_loaded/2 first to cache scripts.

@spec exists?(script_name()) :: boolean()

Checks if a script exists.

@spec extend_lock(atom(), queue_context(), String.t(), String.t(), non_neg_integer()) ::
  script_result()

Extends the lock on a job.

@spec extend_locks(
  atom(),
  queue_context(),
  [String.t()],
  [String.t()],
  non_neg_integer()
) ::
  script_result()

Extends locks for multiple jobs in a single call.

This is more efficient than calling extend_lock multiple times when processing many concurrent jobs, as it uses a single Redis call.

Returns a list of results (1 for success, job_id for failures).

@spec get(script_name()) :: {String.t(), non_neg_integer()} | nil

Returns the script content and number of keys for a given script name.

Parameters

• name - The script name as an atom (e.g., :extend_lock, :move_to_active)

Returns

• {content, key_count} tuple if script exists
• nil if script not found

Examples

iex> {content, keys} = BullMQ.Scripts.get(:extend_lock)
iex> is_binary(content) and is_integer(keys)
true

@spec get_content(script_name()) :: String.t() | nil

Returns the script content for a given script name.

@spec get_counts(atom(), queue_context()) :: script_result()

Gets job counts for the queue.

@spec get_jobs(
  atom(),
  queue_context(),
  [String.t()],
  integer(),
  integer(),
  boolean(),
  integer()
) :: script_result()

Fetches job ids and their job hashes for the provided states in a single script, skipping ids whose job hash is missing (for example the deprecated wait list marker or jobs removed after their id was read).

types are the Lua state type strings (e.g. "wait", "active", "waiting-children"). The result is one array per requested type; each entry is a [job_id, [field, value, ...]] tuple where the field/value list is the flattened job hash. For bounded ranges the script iterates forward using the range offset as a cursor to backfill skipped ids.

@spec get_key_count(script_name()) :: non_neg_integer() | nil

Returns the number of keys for a given script.

@spec get_metrics(atom(), queue_context(), :completed | :failed, integer(), integer()) ::
  script_result()

Gets queue metrics for completed or failed jobs.

Returns metrics data including count, previous timestamp, previous count, data points, and total number of points.

@spec get_rate_limit_ttl(atom(), queue_context(), keyword()) :: script_result()

Gets the rate limit TTL.

Options

• :max_jobs - Maximum jobs for rate limit (default: 0, uses meta key)

@spec get_sha(script_name()) :: String.t() | nil

Gets the precomputed SHA for a script. SHAs are computed at compile time for efficiency.

@spec get_state(atom(), queue_context(), String.t()) :: script_result()

Gets the state of a job.

@spec is_maxed(atom(), queue_context()) :: script_result()

Checks if the queue is at its max limit.

@spec list_scripts() :: [script_name()]

Lists all available script names.

@spec move_job_from_active_to_wait(atom(), queue_context(), String.t(), String.t()) ::
  script_result()

Moves a job from active state back to wait.

This is useful when manually processing jobs and you need to release a job back to the queue (e.g., due to rate limiting).

Returns

• {:ok, pttl} - The rate limit TTL in milliseconds (0 if no rate limit)

@spec move_stalled_jobs_to_wait(atom(), queue_context(), non_neg_integer(), keyword()) ::
  script_result()

Moves stalled jobs back to wait.

@spec move_to_active(atom(), queue_context(), String.t(), keyword()) ::
  script_result()

Moves a job to the active state for processing.

@spec move_to_completed(
  atom(),
  queue_context(),
  String.t(),
  String.t(),
  any(),
  keyword()
) ::
  script_result()

Moves a job to completed state.

@spec move_to_delayed(
  atom(),
  queue_context(),
  String.t(),
  String.t(),
  non_neg_integer(),
  keyword()
) :: script_result()

Moves a job to delayed state (for retry with delay).

@spec move_to_failed(
  atom(),
  queue_context(),
  String.t(),
  String.t(),
  any(),
  keyword()
) ::
  script_result()

Moves a job to failed state.

@spec move_to_finished(
  atom(),
  queue_context(),
  String.t(),
  String.t(),
  any(),
  atom(),
  keyword()
) ::
  script_result()

Moves a job to finished (completed/failed) state.

@spec move_to_waiting_children(
  atom(),
  queue_context(),
  String.t(),
  String.t(),
  keyword()
) ::
  script_result()

Moves a job from active to waiting-children state.

This is used when a job needs to wait for its child jobs to complete before continuing. The job will be automatically moved back to waiting when all children complete.

Returns

• {:ok, 0} - Successfully moved to waiting-children
• {:ok, 1} - No pending dependencies
• {:ok, -1} - Missing job
• {:ok, -2} - Missing lock
• {:ok, -3} - Job not in active set
• {:ok, -9} - Job has failed children

@spec obliterate(atom(), queue_context(), non_neg_integer(), boolean()) ::
  script_result()

Obliterates the queue (removes everything including meta).

@spec pause(atom(), queue_context(), boolean()) :: script_result()

Pauses or resumes the queue.

@spec promote(atom(), queue_context(), String.t()) :: script_result()

Promotes a delayed job to wait.

@spec release_lock(atom(), queue_context(), String.t(), String.t()) :: script_result()

Releases the lock on a job.

@spec remove_job(atom(), queue_context(), String.t(), boolean()) :: script_result()

Removes a job from the queue.

@spec reprocess_job(atom(), queue_context(), String.t(), atom(), keyword()) ::
  script_result()

Reprocesses a job that is in completed or failed state.

Moves a finished job back to the wait queue for reprocessing.

Parameters

• conn - Redis connection
• ctx - Queue context from Keys.new/2
• job_id - The job ID to reprocess
• state - The expected current state (:failed or :completed)
• opts - Options:
  • :lifo - If true, push to front of queue (default: false)
  • :reset_attempts_made - Reset attempts counter (default: false)
  • :reset_attempts_started - Reset attempts started counter (default: false)

Returns

• {:ok, 1} - Job successfully moved to wait
• {:error, reason} - Error with code indicating failure:
  • -1: Job does not exist
  • -3: Job was not found in the expected state

@spec retry_job(atom(), queue_context(), String.t(), boolean(), String.t()) ::
  script_result()

Retries a failed job.

@spec update_data(atom(), queue_context(), String.t(), map()) :: script_result()

Updates the data of a job.

@spec update_job_scheduler(
  atom(),
  queue_context(),
  String.t(),
  non_neg_integer(),
  String.t(),
  binary(),
  String.t()
) :: script_result()

Updates a job scheduler and adds the next delayed job.

Called by the worker after completing a repeatable job to schedule the next iteration.

Parameters

• conn - Redis connection
• ctx - Queue context (keys structure)
• scheduler_id - The job scheduler ID (repeat_job_key)
• next_millis - Next execution time in milliseconds
• template_data - JSON-encoded job data
• job_opts - Msgpacked job options
• producer_id - The ID of the job that produced this iteration

Returns

• {:ok, job_id} - The ID of the next scheduled job
• {:ok, nil} - Scheduler doesn't exist or duplicate
• {:error, reason} - Error

@spec update_progress(atom(), queue_context(), String.t(), any()) :: script_result()

Updates the progress of a job.