@spec ack(repo_or_conn(), integer()) :: {:ok, integer()} | {:error, term()}

Acknowledges and closes a processed batch.

@spec create_queue(repo_or_conn(), queue_name(), keyword()) ::
  {:ok, integer()} | {:error, term()}

Creates a new queue in the PgQue engine.

Options

• :ntables or :history_tables - Number of partition tables in the ring buffer (default 3).
• :rotation_period - Table rotation interval (e.g. "2 hours", "1 day").

@spec dlq_inspect(repo_or_conn(), queue_name(), pos_integer()) ::
  {:ok, [map()]} | {:error, term()}

Inspects dead-lettered events for a given queue.

@spec dlq_purge(repo_or_conn(), queue_name()) :: {:ok, integer()} | {:error, term()}

Purges all dead-lettered events for a given queue.

@spec dlq_replay(repo_or_conn(), integer()) :: {:ok, integer()} | {:error, term()}

Replays a single event from the Dead-Letter Queue back into the active queue.

@spec dlq_replay_all(repo_or_conn(), queue_name()) ::
  {:ok, integer()} | {:error, term()}

Replays all dead-lettered events for a given queue.

@spec drop_queue(repo_or_conn(), queue_name(), boolean()) ::
  {:ok, integer()} | {:error, term()}

Drops a queue and all its data tables.

@spec event_dead(repo_or_conn(), integer(), integer(), String.t()) ::
  {:ok, integer()} | {:error, term()}

Moves an event directly to the Dead-Letter Queue (DLQ).

@spec event_retry(repo_or_conn(), integer(), integer(), integer()) ::
  {:ok, integer()} | {:error, term()}

Moves an individual failed event into the retry queue with a specified retry delay in seconds.

@spec event_snooze(repo_or_conn(), integer(), integer(), integer()) ::
  {:ok, integer()} | {:error, term()}

Snoozes an event into the retry queue for snooze_seconds without incrementing ev_retry (attempt).

@spec force_next_tick(repo_or_conn(), queue_name()) ::
  {:ok, integer()} | {:error, term()}

Forces the next tick threshold to advance immediately for the queue. Useful for tests and immediate visibility.

@spec get_consumer_info(repo_or_conn(), queue_name() | nil, consumer_name() | nil) ::
  {:ok, [map()]} | {:error, term()}

Returns consumer subscription information for all queues or a specific consumer.

@spec get_queue_info(repo_or_conn(), queue_name() | nil) ::
  {:ok, [map()] | map()} | {:error, term()}

Returns metadata for all queues or a specific queue.

@spec pause_inserts(repo_or_conn(), queue_name()) ::
  {:ok, integer()} | {:error, term()}

Disables event insertion into the given queue.

@spec pause_queue(repo_or_conn(), queue_name()) :: {:ok, integer()} | {:error, term()}

Pauses the ticker for the given queue.

@spec query(repo_or_conn(), String.t(), list(), keyword()) ::
  {:ok, Postgrex.Result.t()} | {:error, Exception.t()}

Executes a raw parameterized SQL query against the repository or database connection.

@spec queue_overview(repo_or_conn()) :: {:ok, map()} | {:error, term()}

Returns a consolidated overview of all queues, consumers, DLQs, and retry counts.

@spec receive(repo_or_conn(), queue_name(), consumer_name(), pos_integer()) ::
  {:ok, {integer() | nil, [map()]}} | {:error, term()}

Receives a batch of messages from the queue. Returns {:ok, {batch_id, messages}} where messages is a list of map structures. If no batch is ready, returns {:ok, {nil, []}}.

@spec receive_coop(
  repo_or_conn(),
  queue_name(),
  consumer_name(),
  String.t(),
  pos_integer()
) ::
  {:ok, {integer() | nil, [map()]}} | {:error, term()}

Receives a batch of messages cooperatively as a subconsumer within a consumer group.

@spec resume_inserts(repo_or_conn(), queue_name()) ::
  {:ok, integer()} | {:error, term()}

Re-enables event insertion into the given queue.

@spec resume_queue(repo_or_conn(), queue_name()) ::
  {:ok, integer()} | {:error, term()}

Resumes the ticker for the given queue.

@spec retry_queue_inspect(repo_or_conn(), queue_name(), pos_integer()) ::
  {:ok, [map()]} | {:error, term()}

Inspects scheduled / retry queue entries for a given queue.

@spec send(repo_or_conn(), queue_name(), term(), String.t()) ::
  {:ok, integer()} | {:error, term()}

Sends a message to the specified queue.

@spec send_batch(repo_or_conn(), queue_name(), [term() | {String.t(), term()}]) ::
  {:ok, [integer()]} | {:error, term()}

Sends a batch of messages to a queue in a single SQL operation. items can be a list of maps/strings, or {type, payload} tuples.

@spec set_queue_config(repo_or_conn(), queue_name(), String.t(), String.t()) ::
  {:ok, integer()} | {:error, term()}

Updates a queue parameter (e.g. rotation_period, ticker_max_count, etc.).

@spec subscribe(repo_or_conn(), queue_name(), consumer_name()) ::
  {:ok, integer()} | {:error, term()}

Subscribes a consumer to a queue.

@spec ticker(repo_or_conn(), queue_name() | nil) ::
  {:ok, integer() | nil} | {:error, term()}

Advances queue snapshots by running a ticker tick.

@spec unsubscribe(repo_or_conn(), queue_name(), consumer_name()) ::
  {:ok, integer()} | {:error, term()}

Unsubscribes a consumer from a queue.