Event bus · pgbus

Publish an event#

An event is a routing key plus a payload. Publish it now, or schedule it with a delay:

Pgbus.publish(
  "orders.created",
  { order_id: order.id, total: order.total }
)

Pgbus.publish_later(
  "invoices.due",
  { invoice_id: invoice.id },
  delay: 30.days
)

The payload is JSON. Keep it to data a subscriber can act on — ids, amounts, timestamps — not whole objects.

Pgbus.publish / Pgbus.publish_later are top-level shortcuts for Pgbus::EventBus::Publisher.publish / .publish_later (symmetric with Pgbus.stream). The long form still works if you prefer it.

Subscribe with a handler#

A handler subclasses Pgbus::EventBus::Handler and implements #handle. Register it against a pattern in an initializer:

app/handlers/order_created_handler.rb

class OrderCreatedHandler < Pgbus::EventBus::Handler
  idempotent! # deduplicate by (event_id, handler_class)

  def handle(event)
    order_id = event.payload["order_id"]
    Analytics.track_order(order_id)
    InventoryService.reserve(order_id)
  end
end

config/initializers/pgbus.rb

Pgbus::EventBus::Registry.instance.subscribe("orders.created", OrderCreatedHandler)

Topic routing#

AMQP-style patterns: * for one segment, # for many.

Patterns match dotted routing keys the way AMQP topic exchanges do: * matches exactly one segment, # matches zero or more. One publish fans out to every subscriber whose pattern matches.

A published topic is matched against subscription patterns, fanned to each subscriber's queue, and deduplicated per handler.

Pattern	Matches	Doesn't match
`orders.created`	`orders.created`	`orders.updated`
`orders.*`	`orders.created`, `orders.updated`	`orders.line.added`
`orders.#`	`orders.created`, `orders.line.added`	`payments.captured`

# Audit everything under the orders.* namespace, at any depth:
Pgbus::EventBus::Registry.instance.subscribe("orders.#", OrderAuditHandler)

Idempotent handlers#

Safe to re-run — deduplicated by (event_id, handler).

At-least-once delivery means a handler can be invoked more than once for the same event (a retry after a crash mid-handle). Declaring idempotent! records each (event_id, handler_class) in pgbus_processed_events with a unique index — a second delivery of the same event to the same handler is skipped, backed by an in-memory cache to avoid the round trip when it can.

How long processed-event records are kept is idempotency_ttl (default 7 days). The dispatcher purges older rows. Set it long enough to cover your worst-case retry window.