# `PgFlow.Queries.Pgmq`
[🔗](https://github.com/agoodway/pgflow/blob/v0.7.0/lib/pgflow/queries/pgmq.ex#L1)

SQL query interface for pgmq-specific operations.

Provides functions for managing pgmq NOTIFY triggers and querying
pgmq extension metadata.

# `enable_option`

```elixir
@type enable_option() :: {:trigger, notify_trigger()} | {:lock_timeout, pos_integer()}
```

Options accepted by `enable_notify_insert/4`.

# `notify_trigger`

```elixir
@type notify_trigger() :: :deferred | :pgmq_default
```

Which INSERT trigger `enable_notify_insert/4` leaves on the queue table.

# `default_lock_timeout_ms`

```elixir
@spec default_lock_timeout_ms() :: pos_integer()
```

Returns the default milliseconds a `:deferred` trigger install waits for the
queue table lock, `2000`.

# `default_throttle_ms`

```elixir
@spec default_throttle_ms() :: non_neg_integer()
```

Returns the default minimum interval, in milliseconds, between notifications
for a queue, `250`.

# `default_trigger`

```elixir
@spec default_trigger() :: notify_trigger()
```

Returns the default notify trigger mode, `:deferred`.

# `disable_notify_insert`

```elixir
@spec disable_notify_insert(module(), String.t()) :: :ok | {:error, term()}
```

Disables NOTIFY triggers for a pgmq queue.

Calls `pgmq.disable_notify_insert/1` to remove the INSERT trigger
that fires NOTIFY events when messages are added to the queue.

# `enable_notify_insert`

```elixir
@spec enable_notify_insert(module(), String.t(), non_neg_integer(), [enable_option()]) ::
  :ok | {:error, :in_transaction | term()}
```

Enables NOTIFY triggers for a pgmq queue.

`throttle_ms` is the minimum interval between notifications for the queue.

## Options

  * `:trigger` - `:deferred` (default) or `:pgmq_default`.
  * `:lock_timeout` - milliseconds the `:deferred` install waits for the
    queue table lock before failing. Default: `2000`.

Raises `ArgumentError` on unknown options or invalid values.

## `:deferred`

pgmq creates its INSERT trigger `INITIALLY IMMEDIATE`, so each insert
updates the queue's `pgmq.notify_insert_throttle` row and holds that row
lock until the inserting transaction ends: one long transaction that
enqueues blocks every other producer on the queue. This mode leaves the
trigger `DEFERRABLE INITIALLY DEFERRED`, so it runs at COMMIT, the row lock
lasts only for the commit itself, and a rolled-back transaction sends no
notification.

When the queue already has the deferred trigger, the call changes nothing but
the queue's throttle row, with no DDL and no queue table lock: it updates the
row's throttle interval when it differs, and inserts the row when it is
missing. pgmq keeps `pgmq.notify_insert_throttle` UNLOGGED, so crash
recovery empties it, and so does Neon whenever a compute restarts; while a
queue's row is missing, its trigger sends no NOTIFY. `PgFlow.FlowStarter`
calls this on every boot, so steady-state boots take this path.

Otherwise it runs one transaction that sets `lock_timeout` to the
`:lock_timeout` option, takes a per-queue advisory lock so installers on
other nodes wait for each other, re-checks the trigger under that lock, and
then calls `pgmq.enable_notify_insert/2` and recreates its trigger under the
same name as `DEFERRABLE INITIALLY DEFERRED`. pgmq's function drops and
recreates the trigger, which takes an ACCESS EXCLUSIVE lock on the queue
table; when an open transaction holds a conflicting lock for longer than
`:lock_timeout`, the call returns the `lock_not_available` `Postgrex.Error`
instead of making every later send and read on the queue wait behind it.

If pgmq's function succeeds but recreating the trigger deferred fails (for
example on missing privileges or a renamed pgmq object), only that swap is
rolled back: pgmq's immediate trigger stays, so notifications still work,
a warning is logged, and the call returns `:ok`.
`PgFlow.SchemaCheck.warnings/2` reports the immediate trigger.

## `:pgmq_default`

Calls only `pgmq.enable_notify_insert/2`, which leaves pgmq's immediate
trigger, in a single statement with no `lock_timeout`.

## Transactions

Returns `{:error, :in_transaction}` when called inside a transaction on
`repo`. Run there, pgmq's DROP TRIGGER would hold ACCESS EXCLUSIVE on the
queue table until the caller commits, and dropping a deferred trigger after
the caller's own send in that transaction silently discards that send's
pending notification. `PgFlow.FlowStarter` never calls it inside a
transaction.

# `flush_notify_events`

```elixir
@spec flush_notify_events(module()) :: :ok | {:error, term()}
```

Fires the current transaction's pending deferred notify trigger events.

A deferred trigger keeps one pending event per inserted row until COMMIT,
and Postgres refuses to DROP, TRUNCATE, or ALTER a table that has pending
trigger events. `pgmq.drop_queue/1` or `pgmq.purge_queue/1` after a send in
the same transaction therefore fails with "cannot DROP TABLE ... because it
has pending trigger events". Call this earlier in that transaction.

It runs `SET CONSTRAINTS pgmq.trigger_notify_queue_insert_listeners
IMMEDIATE` only when a constraint by that name exists, so it does nothing on
pgmq releases without notify triggers. The immediate mode lasts until the
transaction ends: later sends in that transaction notify at insert time and
hold the throttle row lock until commit. Outside a transaction it has no
lasting effect.

# `get_pgmq_version`

```elixir
@spec get_pgmq_version(module()) ::
  {:ok, String.t()} | {:error, :not_installed | term()}
```

Gets the installed pgmq version.

Checks two sources in order:
1. `pg_extension` catalog — works when pgmq is installed via `CREATE EXTENSION pgmq`
2. Feature detection — checks for the `enable_notify_insert` function in the `pgmq` schema,
   which indicates pgmq >= 1.8.0 installed from a SQL dump (e.g., on Neon or other managed Postgres)

Returns `{:ok, version}` if pgmq is detected, or `{:error, :not_installed}` if not.

# `restore_notify_throttle_rows`

```elixir
@spec restore_notify_throttle_rows(module(), [String.t()], non_neg_integer()) ::
  {:ok, [String.t()]} | {:error, term()}
```

Inserts the missing `pgmq.notify_insert_throttle` row of each queue in
`queue_names` that has a notify trigger, with `throttle_ms` as its interval,
and returns the names of the queues it restored.

pgmq keeps the throttle table UNLOGGED, so crash recovery empties it, and so
does Neon whenever a compute restarts; a trigger whose row is missing sends
no NOTIFY. This runs one statement that changes no trigger and locks no
queue table. It skips queues that already have a row, have no notify
trigger, or do not exist. `PgFlow.Signal.Notify` runs it periodically for
the queues its workers listen on.

---

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