Guide

# Event bus

Publish a topic once; AMQP-style patterns fan it out to idempotent subscribers.

## Publish an event

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

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

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

> **Note:** 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:

```ruby
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
```

```ruby
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.

publish

"orders.created"

topic routing

orders.#      ✓

orders.*      ✓

payments.*   ✗

OrderAudit

subscriber queue

Analytics

subscriber queue

pgbus_processed_events  (event_id, handler)

idempotent! → skip if already handled

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` |

```ruby
# 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.

> **Tip:** 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.