Guide

# Real-time streams

A drop-in Turbo Streams transport over Postgres SSE — no Action Cable, no Redis, no lost messages on reconnect.

## Swap one helper

pgbus ships a drop-in replacement for turbo-rails' `turbo_stream_from`. Swap the helper in your view; everything else — the model concern, the broadcast helpers — stays the same:

```erb
<%# Before %>
<%= turbo_stream_from @order %>

<%# After — no Action Cable, no Redis %>
<%= pgbus_stream_from @order %>
```

```ruby
class Order < ApplicationRecord
  broadcasts_to ->(order) { [order.account, :orders] }
end
```

broadcast

→ stream queue

listener

LISTEN/NOTIFY

Tab A

open SSE

Tab B

open SSE

SSE push

archive replay on reconnect  (Last-Event-ID)

no lost messages

A broadcast fans out over SSE to every open tab; a reconnecting tab replays what it missed from the archive via Last-Event-ID.

> **Warning:** Add the Puma plugin so SSE connections drain cleanly on deploy: `plugin :pgbus_streams` in config/puma.rb. Streams need Puma 6.1+ or Falcon (they use rack.hijack), and HTTP/2 in production to lift the 6-connection-per-origin SSE limit.

> **Tip:** The default `broadcasts_to` path enqueues the render+broadcast as a background job on the default queue, so a broadcast can wait behind long-running jobs. To keep broadcasts off the critical path, set `config.streams_broadcast_queue` and back it with a dedicated worker capsule, or pass `durable: true` to broadcast synchronously in the request thread.

```ruby
Pgbus.configure do |c|
  c.streams_broadcast_queue = "realtime"
  c.workers = [
    { queues: ["realtime"], threads: 3 },  # broadcasts get their own pool
    { queues: ["*"],        threads: 10 }   # everything else
  ]
end
```

## What it fixes

Three well-known Action Cable correctness bugs.

Every broadcast gets a monotonic PGMQ `msg_id`. The helper captures the current max at render time and embeds it as a cursor; the SSE client sends it as `Last-Event-ID` on reconnect, and the streamer replays anything newer from the live queue and the archive. That cursor model is what fixes the classic Action Cable gaps:

| Bug | What breaks | How pgbus fixes it |
| --- | --- | --- |
| Page born stale | A broadcast between render and subscribe is lost. | A render-time msg_id watermark replays the gap. |
| Missed on reconnect | A dropped connection misses what aired. | `Last-Event-ID` replays from the PGMQ archive. |
| No disconnect signal | The client can't tell it dropped. | `pgbus:open` / `pgbus:gap-detected` / `pgbus:close` DOM events. |

## Transactional broadcasts

Deferred until commit — no phantom updates.

A broadcast issued inside an open Active Record transaction is deferred until the transaction commits. If it rolls back, the broadcast drops — clients never see a change the database never persisted. No other Rails real-time stack can do this, because Action Cable's path goes through a broker with no idea of your transaction boundary.

```ruby
ActiveRecord::Base.transaction do
  @order.update!(status: "shipped")
  @order.broadcast_replace_to :account    # ← deferred until commit
  RelatedService.update_counters!(@order) # ← if this raises, both roll back
end
# Rolled back? No client ever saw "shipped".
```

## Replaying history on connect

By default a stream shows only broadcasts published after render (the page-born-stale fix). For chat-style backlog, pass `replay:`:

```erb
<%= pgbus_stream_from @room, replay: 50 %>       <%# last 50 on load %>
<%= pgbus_stream_from @room, replay: :all %>     <%# everything in retention %>
<%= pgbus_stream_from @room, replay: :watermark %> <%# default: post-render only %>
```

> **Note:** How far back `replay: :all` reaches depends on the stream's retention (`streams_retention` / `streams_default_retention`, default 5 minutes). Bump it for chat streams that need days of history.

## Broadcast options

Every keyword #broadcast and #broadcast_render accept.

`Pgbus.stream(name).broadcast(html, **options)` (and `broadcast_render`, below) take a common set of keyword options that compose freely:

| Option | Type | Effect |
| --- | --- | --- |
| `exclude:` | connection id | Skip delivery to the named SSE connection — actor-echo suppression. |
| `event:` | String, Symbol | Set the SSE `event:` field so clients can route without sniffing the HTML. |
| `coalesce:` | true, Numeric (ms) | Debounce rapid broadcasts to the same `(stream, target)`; requires `target:`. |
| `durable:` | true, false, nil | Per-broadcast override of the stream's durable mode (PGMQ-backed vs. ephemeral NOTIFY-only). |
| `visible_to:` | Symbol | Restrict delivery to connections whose authorize-hook context passes the named filter. |

`durable:` predates this release and is covered in [Transactional broadcasts](#transactional-broadcasts) above; `visible_to:` restricts delivery to connections whose authorize-hook context passes a filter registered via `Pgbus::Streams.filters.register` (see the [README's audience filtering section](https://github.com/mhenrixon/pgbus#server-side-audience-filtering) for the full registry API). The rest — `exclude:`, `event:`, `coalesce:` — are documented below.

## exclude: — actor-echo suppression

Don't re-deliver a broadcast to the connection that triggered it.

An actor who just triggered a change already applied it via the HTTP response of their own action. If the resulting broadcast reaches their own SSE connection too, it double-applies — re-running animations or clobbering an optimistic edit. Pass `exclude:` with a connection id to skip that one connection; everyone else still gets the broadcast:

```ruby
Pgbus.stream(room).broadcast(html, exclude: connection_id)
```

The connection id comes from the server. Right after the SSE handshake opens, pgbus sends a `pgbus:connected` frame carrying a server-minted connection id; `<pgbus-stream-source>` captures it onto its `connection-id` attribute and re-dispatches it as a `pgbus:connected` event (also present in `pgbus:open`'s detail). The page reads that id and sends it back as the `X-Pgbus-Connection` header on the action request that triggers the broadcast — the server then excludes it:

```javascript
document.addEventListener("pgbus:connected", (event) => {
  document.querySelector("meta[name='pgbus-connection-id']")
    ?.setAttribute("content", event.detail.connectionId)
})

// On the next fetch/form submit:
fetch(url, {
  method: "POST",
  headers: { "X-Pgbus-Connection": connectionIdMetaTag() }
})
```

```ruby
def create
  @message = @room.messages.create!(message_params)
  @room.broadcast_append_to(
    :messages,
    exclude: request.headers["X-Pgbus-Connection"]
  )
end
```

> **Note:** A nil or blank `exclude:` is a no-op — the common path for background jobs and other server-initiated broadcasts with no originating connection.

## broadcast_render — render and broadcast in one call

`Stream#broadcast_render` renders a component and broadcasts it as a complete `<turbo-stream>` tag atomically, removing the off-request render-context boilerplate every call site would otherwise hand-roll:

```ruby
Pgbus.stream("chat", room).broadcast_render(
  renderable: Chat::Message.new(chat_message: msg),
  action: :append,
  target: "chat-messages-#{room}",
  exclude: connection_id   # composes with every option above
)
```

| Renderable | Resolution |
| --- | --- |
| String | Used verbatim — already-rendered markup. |
| responds to `#call` | Phlex component — calls it and stringifies the result. |
| responds to `#render_in` | ViewComponent / phlex-rails — calls `render_in(nil)` (no controller view context off-request). |
| else | falls back to `#to_s`. |

`action` defaults to `:replace`; `target:` is required. Content-less actions (`:remove`) emit no `<template>` wrapper and ignore `renderable:`. `exclude:`, `visible_to:`, `durable:`, `event:`, and `coalesce:` all forward to `#broadcast` unchanged.

> **Tip:** A component that needs URL helpers or a full view context (e.g. `link_to`) should be rendered by the app — which has the request context — with the resulting string passed as `renderable:`.

## Typed SSE event names

Route on a name instead of sniffing the HTML.

A broadcast can set the SSE `event:` field while keeping the payload a Turbo Stream, so clients route on a typed name instead of parsing the markup:

```ruby
Pgbus.stream(name).broadcast(html, event: "presence")
Pgbus.stream(name).broadcast_render(renderable: component, target: "cursor", event: "reactive")
```

The default (`nil` or `"turbo-stream"`) is never written into the JSONB payload — it's implicit — but it's still set on the SSE frame's `event:` line (falling back to `turbo-stream`), so default consumers are unaffected either way.

On the client, `<pgbus-stream-source>` dispatches a typed broadcast two ways:

| Event | Detail | Use for |
| --- | --- | --- |
| `pgbus:event` | `{ event, data, msgId }` | One listener that handles every typed event. |
| `pgbus:<event>` | `{ data, msgId }` | `addEventListener("pgbus:presence", …)` ergonomics. |

```javascript
document.addEventListener("pgbus:presence", (event) => {
  const { data, msgId } = event.detail
  // ...
})
```

> **Warning:** Native `EventSource` (the reconnect path) only invokes listeners registered by name, so declare every typed event name you use on the element's `listen-events` attribute (comma- or space-separated) — otherwise a typed broadcast is silently dropped after a reconnect, even though it worked on the first connection (which uses `fetch()` and routes any event generically):

```erb
<%= pgbus_stream_from @room, "listen-events": "presence reactive" %>
```

## coalesce: — publish-side debounce

Batch high-frequency broadcasts to the latest frame per target.

A chatty component — a live cursor, a typing indicator, a progress bar — can fan out many small broadcasts per second. Pass `coalesce:` (a window in milliseconds, or `true` for the 50ms default) together with `target:` to batch broadcasts per `(stream, target)` and publish only the *latest* frame within the window:

```ruby
Pgbus.stream(name).broadcast_render(
  renderable: CursorPosition.new(x:, y:),
  target: "cursor-#{user_id}",
  coalesce: true          # or coalesce: 100 for a 100ms window
)
```

Superseded frames never hit the bus at all — no PGMQ insert, no NOTIFY, no fan-out. This is last-write-wins, so it's only safe for **idempotent replace/update of a stable target** (exactly the high-frequency case above) — never for actions where every intermediate frame matters (an `append` to a running log, for instance).

Semantics: the first submit for a `(stream, target)` schedules the flush one window later; every subsequent submit within that window only overwrites the buffered payload. Latency is bounded to one window, and a continuous stream of updates can't starve the flush indefinitely (trailing-edge-with-max-wait, not a resettable debounce). The flush re-enters the normal broadcast path, so a coalesced frame still composes with `visible_to:`, `exclude:`, `event:`, and `durable:`.

> **Note:** Coalescing is process-wide and in-memory. Behind multiple Puma workers or Falcon processes, each process debounces its own submissions independently.

## Presence

"X people are in this room."

Track who is subscribed to a stream with a presence table. Two modes are available and can be mixed: the **manual API** below (explicit join/leave, full control over when someone counts as "present") and **connection-driven presence** (automatic, opt-in per stream pattern).

```shell
rails generate pgbus:add_presence && rails db:migrate
```

```ruby
Pgbus.stream(@room).presence.join(
  member_id: current_user.id.to_s,
  metadata: { name: current_user.name }
) { |member| render_to_string(partial: "presence/joined", locals: { member: }) }

Pgbus.stream(@room).presence.members # => [{ "id" => "7", "metadata" => {...} }, …]
Pgbus.stream(@room).presence.count   # => 5
```

## Connection-driven presence (opt-in)

Auto-join on connect, auto-leave on disconnect — no explicit wiring.

Streams matching `config.streams_presence_patterns` (an exact string or a `Regexp`, mirroring `streams_durable_patterns`) automatically join a member when an SSE connection opens, leave when it closes, and refresh `last_seen_at` on every keepalive heartbeat tick — no explicit `join`/`leave`/sweeper calls required:

```ruby
Pgbus.configure do |c|
  c.streams_presence_patterns = [/^room:/, "lobby"]
end
```

Identity comes from the connection's authorize-hook context (the value your `StreamApp` `authorize:` callable returns). The built-in extractor handles the common shapes without any configuration:

| Context shape | Extracted member |
| --- | --- |
| `Hash` with `:member_id` or `:id` | `{ id:, metadata: }` — `:metadata` optional, defaults to `{}` |
| any object responding to `#id` | `{ id: object.id, metadata: {} }` |

For anything else, provide a custom extractor — a `->(context) { { id:, metadata: } }` callable returning `nil` for a context with no derivable identity (anonymous connections are simply skipped, not an error):

```ruby
Pgbus.configure do |c|
  c.streams_presence_patterns = [/^room:/]
  c.streams_presence_member = ->(user) {
    { id: user.id, metadata: { name: user.name, avatar: user.avatar_url } } if user
  }
end
```

> **Note:** Membership work runs on the dispatcher thread and presence failures are logged and swallowed — a presence-table hiccup can never knock a live SSE connection out of the registry.

## stream_key idempotency

Hold one key value and reuse it safely.

`Pgbus.stream_key` treats a single `String` argument as an already-built pgbus stream key and returns it unchanged (after the queue-name budget check), instead of tripping the colon-separator guard. This lets a consumer hold one `stream_key` value and pass it to both `turbo_stream_from`/`pgbus_stream_from` and the broadcaster:

```ruby
key = Pgbus.stream_key(chat, :messages)  # => "ai_chat_a3f8c1e9d2b47610:messages"

# Both calls accept the same pre-built key without raising:
pgbus_stream_from(key)
Pgbus.stream(key).broadcast(html)
```

`Pgbus.stream_key!(key)` accepts a pre-built key explicitly — `String` required, budget still enforced — for call sites that want to be explicit that no re-keying should happen.

The guard is still enforced for the cases that are genuinely ambiguous:

| Call | Behavior |
| --- | --- |
| `stream_key("chat:lobby")` | OK — single pre-built key, idempotent. |
| `stream_key("a:b", :c)` | Raises `ArgumentError` — an ambiguous multi-fragment join (`"a:b"` + `:c` could mean `stream_key("a", "b:c")` too). |
| `stream_key(:'a:b')` | Raises `ArgumentError` — a colon in a `Symbol`/record fragment never came from `stream_key` and is treated as a mistake. |

## msg_id reconciliation for optimistic UI

Reconcile out-of-order or duplicate delivery on the client.

Every delivered frame carries its monotonic PGMQ `msg_id` as the SSE `id:` line — this is the same watermark that powers reconnect replay. `<pgbus-stream-source>` surfaces it to the client two ways:

| Event | Detail | Use for |
| --- | --- | --- |
| `message` | standard `MessageEvent`, `lastEventId` set to the msg_id | Turbo ignores it; a reactive runtime listening for `message` reads the revision with no pgbus-specific API. |
| `pgbus:message` | `{ msgId, data }` | Optimistic-UI reconciliation — msgId is a Number when numeric. |

A negative `msgId` marks an ephemeral frame (one that bypassed PGMQ — an ephemeral-mode broadcast) rather than a durable, archived one.

**The reconciliation recipe:** track the highest applied `msgId` per render target; when a frame arrives, skip the morph if you've already applied a newer revision for that target. This stops a late echo — a broadcast that was in flight when a newer one landed — from clobbering a newer optimistic edit:

```javascript
const appliedRevision = new Map() // target -> highest applied msgId

document.addEventListener("pgbus:message", (event) => {
  const { msgId, data } = event.detail
  const target = extractTargetFrom(data) // however your markup encodes it

  const highest = appliedRevision.get(target) ?? -Infinity
  if (msgId != null && msgId < highest) return // stale — skip the morph

  appliedRevision.set(target, msgId)
  applyMorph(target, data)
})
```

> **Note:** This complements `exclude:`: `exclude:` handles the actor (never receives its own echo at all); msg_id reconciliation handles out-of-order delivery for everyone else.

## On the consuming side

[phlex-reactive](https://phlex-reactive.zoolutions.llc) uses pgbus as its broadcast transport, so its [Transport: pgbus](https://phlex-reactive.zoolutions.llc/docs/transport-pgbus) page is a good companion read — it shows the same primitives from a component author's point of view, including the pgbus-only `exclude:` and `visible_to:` broadcast options.