# `Fleet.Client`
[🔗](https://github.com/kuku/fleet/blob/main/lib/fleet/client.ex#L1)

Low-level PostgreSQL client wrapper for the `pgque` stored procedures.

Accepts either an `Ecto.Repo` module or a `Postgrex` / `DBConnection` process.

# `consumer_name`

```elixir
@type consumer_name() :: String.t() | atom()
```

# `queue_name`

```elixir
@type queue_name() :: String.t() | atom()
```

# `repo_or_conn`

```elixir
@type repo_or_conn() :: module() | pid() | GenServer.server()
```

# `ack`

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

Acknowledges and closes a processed batch.

# `create_queue`

```elixir
@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"`).

# `dlq_inspect`

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

Inspects dead-lettered events for a given queue.

# `dlq_purge`

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

Purges all dead-lettered events for a given queue.

# `dlq_replay`

```elixir
@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.

# `dlq_replay_all`

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

Replays all dead-lettered events for a given queue.

# `drop_queue`

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

Drops a queue and all its data tables.

# `event_dead`

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

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

# `event_retry`

```elixir
@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.

# `event_snooze`

```elixir
@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`).

# `force_next_tick`

```elixir
@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.

# `get_consumer_info`

```elixir
@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.

# `get_queue_info`

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

Returns metadata for all queues or a specific queue.

# `pause_inserts`

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

Disables event insertion into the given queue.

# `pause_queue`

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

Pauses the ticker for the given queue.

# `query`

```elixir
@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.

# `queue_overview`

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

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

# `receive`

```elixir
@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, []}}`.

# `receive_coop`

```elixir
@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.

# `resume_inserts`

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

Re-enables event insertion into the given queue.

# `resume_queue`

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

Resumes the ticker for the given queue.

# `retry_queue_inspect`

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

Inspects scheduled / retry queue entries for a given queue.

# `send`

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

Sends a message to the specified queue.

# `send_batch`

```elixir
@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.

# `set_queue_config`

```elixir
@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.).

# `subscribe`

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

Subscribes a consumer to a queue.

# `ticker`

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

Advances queue snapshots by running a ticker tick.

# `unsubscribe`

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

Unsubscribes a consumer from a queue.

---

*Consult [api-reference.md](api-reference.md) for complete listing*