postgres

Postgres LISTEN/NOTIFY notifier — a faithful port of Elixir Oban’s Oban.Notifiers.Postgres. It is the isolated notifier (the same local subscriber registry and fan-out) plus a Postgres bridge: notify publishes over pg_notify so every node (including this one) hears it back through a dedicated LISTEN connection, decodes the payload into a typed Notification, and fans it out to local subscribers.

Two pieces, exactly like isolated:

new(connection, relay_name, prefix) builds the Notifier record whose three closures talk to the relay actor (resolved by relay_name) and whose notify runs pg_notify on the caller’s pool connection.
child(config, relay_name, prefix) supervises the relay actor as a single worker (registered under relay_name). The relay start_links its pgo_notifications listener in its own initialiser, so the two share a fate.

Robustness: the relay is a direct child of Oban’s root supervisor (the notifier_child), not buried in its own subtree — so a crash on the relay or its linked listener propagates to the root. The root is RestForOne [notifier_child, runtime], so that crash restarts the runtime too and producers re-listen on the fresh relay (the W1 re-subscribe guarantee). A subtree would absorb the crash and leave producers subscribed to a dead relay, which is why we don’t use one.

Types

Message

</>

pub type Message {
  Listen(
    channel: notifier.Channel,
    subscriber: process.Subject(notifier.Notification),
  )
  Unlisten(
    channel: notifier.Channel,
    subscriber: process.Subject(notifier.Notification),
  )
  Notify(notification: notifier.Notification)
  Inbound(notification: notifier.Notification)
  Discard
  SubscriberDown(down: process.Down)
}

Constructors

Listen(
  channel: notifier.Channel,
  subscriber: process.Subject(notifier.Notification),
)

Unlisten(
  channel: notifier.Channel,
  subscriber: process.Subject(notifier.Notification),
)

Notify(notification: notifier.Notification)

Inbound(notification: notifier.Notification)

```
Discard
```
```
SubscriberDown(down: process.Down)
```

State

</>

pub type State {
  State(
    subscribers: dict.Dict(
      notifier.Channel,
      set.Set(process.Subject(notifier.Notification)),
    ),
    monitors: dict.Dict(process.Pid, process.Monitor),
    prefix: String,
  )
}

Constructors

State(
  subscribers: dict.Dict(
    notifier.Channel,
    set.Set(process.Subject(notifier.Notification)),
  ),
  monitors: dict.Dict(process.Pid, process.Monitor),
  prefix: String,
)

Values

channel_name

</>

pub fn channel_name(channel: notifier.Channel) -> String

The bare Mule channel name (without prefix), matching Elixir Oban’s naming.

child

</>

pub fn child(
  config config: pog.Config,
  relay_name relay_name: process.Name(Message),
  prefix prefix: String,
) -> supervision.ChildSpecification(Nil)

Supervise the relay actor as a single worker registered under relay_name. The relay start_links its pgo_notifications listener in its own initialiser, so they share a fate. Wire this in as Mule’s notifier_child (a direct child of the root) so a relay/listener crash cascades to the runtime and producers re-listen — see the module docs.

new

</>

pub fn new(
  connection connection: pog.Connection,
  relay_name relay_name: process.Name(Message),
  prefix prefix: String,
) -> notifier.Notifier

Build the Notifier the runtime consumes. connection is the caller’s pool connection — notify runs pg_notify on it so other nodes hear the broadcast. listen/unlisten talk to the relay via process.named_subject(relay_name). prefix namespaces the Postgres channels (see prefixed_channel).

Pair with child(config, relay_name, prefix) started under the same relay_name.

prefixed_channel

</>

pub fn prefixed_channel(
  prefix: String,
  channel: notifier.Channel,
) -> String

The prefixed Postgres channel string used on both LISTEN and pg_notify, so two Mule instances sharing one database don’t cross-talk: <prefix>_oban_<name>.