# `PgFlow.Signal.Notify`
[🔗](https://github.com/agoodway/pgflow/blob/v0.7.0/lib/pgflow/signal/notify.ex#L1)

Manages PostgreSQL LISTEN/NOTIFY for pgmq queue notifications.

Uses pgmq's built-in `enable_notify_insert` (pgmq 1.8.0+) to receive instant
wake-up signals when messages are inserted into queue tables. Dispatches
`:poll_now` messages to registered worker processes.

Built on `Postgrex.Notifications` which is the purpose-built solution for
PostgreSQL LISTEN/NOTIFY. Notifications are delivered asynchronously via
messages to the process that called `listen/2`.

pgmq keeps `pgmq.notify_insert_throttle` UNLOGGED, so crash recovery
empties it, and so does Neon whenever a compute restarts. A queue whose
throttle row is missing sends no NOTIFY, and its workers fall back to their
slow poll. Every `:throttle_reconcile_interval_ms` this process runs
`PgFlow.Queries.Pgmq.restore_notify_throttle_rows/3` for the queues its
workers listen on: one statement that restores missing rows, writes nothing
when every row is present, and never changes a trigger or locks a queue
table.

# `state`

```elixir
@type state() :: %{
  repo: module(),
  notify_throttle_ms: non_neg_integer(),
  throttle_reconcile_interval_ms: pos_integer(),
  conn: pid() | nil,
  workers: %{required(String.t()) => worker_entry()},
  channels: %{required(String.t()) => String.t()}
}
```

# `worker_entry`

```elixir
@type worker_entry() :: %{
  worker_pid: pid(),
  monitor_ref: reference(),
  listen_ref: reference() | nil
}
```

# `child_spec`

Returns a specification to start this module under a supervisor.

See `Supervisor`.

# `register_worker`

```elixir
@spec register_worker(GenServer.server(), String.t(), pid()) :: :ok | {:error, term()}
```

Registers a worker for notifications on a queue.

Starts listening on the queue's channel. Note that `pgmq.enable_notify_insert`
must be called separately (done by PgFlow.Supervisor) before registration.

# `register_worker_async`

```elixir
@spec register_worker_async(GenServer.server(), String.t(), pid()) :: :ok
```

Registers a replacement worker without blocking its supervisor restart.

Transient LISTEN failures retry with capped exponential backoff while the
replacement remains alive. Initial registration remains synchronous through
`register_worker/3` so `PgFlow.FlowStarter` can report startup failures.

# `start_link`

```elixir
@spec start_link(keyword()) :: GenServer.on_start()
```

Starts the Signal.Notify process.

## Options

  * `:repo` - (required) The Ecto repository module
  * `:notify_throttle_ms` - (optional) Throttle interval for pgmq notifications (default: 250)
  * `:throttle_reconcile_interval_ms` - (optional) Milliseconds between
    checks that restore missing `pgmq.notify_insert_throttle` rows
    (default: 30000)

# `unregister_worker`

```elixir
@spec unregister_worker(GenServer.server(), String.t()) :: :ok
```

Unregisters a worker from notifications.

---

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