@spec default_lock_timeout_ms() :: pos_integer()

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

@spec default_throttle_ms() :: non_neg_integer()

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

@spec default_trigger() :: notify_trigger()

Returns the default notify trigger mode, :deferred.

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

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

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

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

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