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

## Fair share across tenants

One tenant's event burst must not starve everyone else's handlers.

A subscriber queue is FIFO, so a bulk import emitting `orders.created` 100 000 times for one tenant puts every other tenant's events behind it — in *every* handler subscribed to that topic. `config.event_fair_share` is the event twin of `config.fair_share` (jobs): a callable tags each event with a key (and an optional weight) at publish time, and the consumer's read interleaves across keys inside each subscriber queue — the same weighted, work-conserving round-robin workers use.

```ruby
Pgbus.configure do |config|
  # Receives the Pgbus::Event (routing_key, payload as passed to publish, headers).
  # Return a key (String/Symbol/Integer), [key, weight], or nil to leave the event unkeyed.
  config.event_fair_share = ->(event) { Current.tenant&.id || event.payload["tenant_id"] }
end
```

- The key rides in the **event envelope** (`pgbus_fair_key` next to `event_id` / `published_at`), never inside your payload — handlers see `event.payload` unchanged, and the tag is copied to every subscriber queue the topic fans out to.
- It is resolved where you publish — `Pgbus.publish`, `publish_later`, and `Pgbus::Outbox.publish_event` — so `Current.*` context is available, and the outbox relay carries it to the bus unchanged. A system writing `pgbus_outbox_entries` directly can set `"pgbus_fair_key"` in the envelope JSON itself.
- Independent of `fair_share`: enable either side, or both with the same resolver shape.
- Consumers keep strict list order *across* their subscriber queues and are fair *within* each; the circuit breaker still skips a paused queue.
- The fair index is built on each subscriber queue by `setup_all!` (at creation) and by the consumer at boot (`CONCURRENTLY`). Split rule, weights, cost model and index details: [Routing & ordering](https://pgbus.zoolutions.llc/docs/routing-ordering).

## Current attributes

Request context travels with the event.

The same `config.current_attributes` switch that carries `Current.tenant` / `Current.user` / `Current.request_id` into jobs ([Active Job guide](https://pgbus.zoolutions.llc/docs/active-job)) carries it into event handlers. With it on, `Pgbus.publish` snapshots the assigned attributes of each persisted class into the event envelope under `pgbus_current` — same filters (`only:` / `except:`), same `ActiveJob::Arguments` serialization — and the consumer restores them around `handle`:

```ruby
# In the request: Current.tenant = tenant; then…
Pgbus.publish("orders.created", { order_id: order.id })

# In the handler, on the consumer:
class OrderCreatedHandler < Pgbus::EventBus::Handler
  def handle(event)
    Current.tenant   # => the publisher's tenant
    event.context    # the raw stored form, if you want it explicitly
  end
end
```

- The context lives in the **envelope** (next to `event_id`), never in `event.payload`, and is copied to every subscriber queue the topic fans out to. Previous `Current` values come back after each `handle`.
- `Pgbus::Outbox.publish_event` captures at write time — inside your transaction, where `Current` is set — and the relay carries it to the bus unchanged.
- GlobalIDs inside the context are gated by `allowed_global_id_models` before anything is loaded, exactly like job arguments.
- `Pgbus::Testing` round-trips it: fake-mode events expose `context`, and `drain!` / inline dispatch restore it around handlers.
- The dashboard's pending-events rows show a **Context** card (through the same parameter filter as the payload).

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