---
title: "@octabits-io/events"
summary: "Live events from your database to the browser: a transactional outbox (emit in the write's transaction, notify at COMMIT, replay by watermark), an in-process hub with fail-closed audience filtering, an SSE endpoint as a plain fetch handler, and the matching browser client."
install: "pnpm add @octabits-io/events"
availability: "Available from November 2026"
version: "0.9.0"
license: "MIT"
related:
  - name: "@octabits-io/postgres"
    url: "https://octabits.io/platform/postgres.md"
canonical_url: "https://octabits.io/platform/events"
---

# `@octabits-io/events`

Live events from your database to the browser, without a message broker.
Emit in the same transaction as the write, so a rolled-back write never
announces itself. Fan out per scope with fail-closed audience filtering. Serve
it over SSE from a plain fetch handler. Replay from a watermark when a client
comes back. Postgres `LISTEN/NOTIFY` does the wakeups; the outbox is a seam.

```bash
npm i @octabits-io/events zod          # + pg for the Postgres listener
```

| Subpath | Runs on | Deps | Contents |
|---|---|---|---|
| `.` | any runtime | `zod` | envelope + schema, notify codec, hub, publisher, relay, SSE fetch handler |
| `./client` | browsers, workers, any runtime | none | fetch-based SSE reader: header auth, `Last-Event-ID` replay, full-jitter backoff, durable-only watermark, bounded dedupe |
| `./postgres` | Node, Bun | `pg` (optional peer) | `createPgNotifyListener` — the LISTEN side, one dedicated connection |
| `./pglite` | any runtime | none (structural) | `createPgliteNotifyListener` — the LISTEN side for an embedded PGlite instance |
| `./outbox` | any runtime | none (structural) | `createPostgresEventOutboxStore` — the outbox store over plain SQL (any `query(text, params)` executor: `pg`, PGlite, postgres.js), plus `eventOutboxDdl()` |
| `./drizzle` | Node, Bun | `drizzle-orm` (optional peer), `@octabits-io/postgres` (peer) | `eventOutboxColumns` (timestamps read back as ISO-8601 in UTC) + `createDrizzleEventOutboxStore` — the same outbox for a host that already runs Drizzle |
| `./broadcast` | Node, Bun | `drizzle-orm` (+ `pg` for the default listener) | `createBroadcastChannel` — fire-and-forget hints on the same wire: no envelope, no outbox, no replay |
| `./webhooks` | any runtime | none | durable events delivered to another system's URL: `withWebhookDeliveries` (deliveries written in the event's transaction), `createWebhookDispatcher`, `signWebhook` / `verifyWebhook` (Standard Webhooks), `staticWebhookSubscriptions`, the delivery-store contract and its in-memory reference |
| `./webhooks/postgres` | any runtime | none (structural) | `createPostgresWebhookDeliveryStore` + `webhookDeliveryDdl()` — the delivery table, which is also the queue |
| `./webhooks/drizzle` | Node, Bun | `drizzle-orm` (optional peer) | `webhookDeliveryColumns` — the table for drizzle-kit; `createDrizzleWebhookDeliveryStore`, the SQL store on a Drizzle database, whose `enqueue` takes the Drizzle transaction |

The outbox store (`EventOutboxStore`) and the notification listener
(`EventNotificationListener`) are structural seams. A driver-free Postgres
outbox ships here at `./outbox` and the Drizzle one at `./drizzle`; the Vue
binding for the client (`useEventStream`) in `@octabits-io/agent-ui/vue`. A
boundary lint keeps this package at `zod`, with `pg` confined to `./postgres`
and `drizzle-orm` to `./drizzle`, `./broadcast` and `./webhooks/drizzle`, and a Node-typings-free
type check keeps everything else runtime-agnostic.

## When not to use it

- **Background work.** This is not a message queue and not a replacement for
  pg-boss: no job claiming, no retry, no acknowledgement, no competing
  consumers. An event reaches the subscribers connected right now, plus whoever
  replays from a watermark. Work that must run once, later, with retries, is a
  queue ([`@octabits-io/queue`](https://octabits.io/platform/queue.md)) or a workflow (`octaflow`).
- **A durable history.** The outbox is a delivery log to prune within hours,
  not a store to query. The user-facing feed with read state is
  [`@octabits-io/activity`](https://octabits.io/platform/activity.md), written by the same emit; an
  audit trail is a table of its own.
- **Exactly-once or strict ordering.** Delivery is at-least-once everywhere and
  `seq` becomes visible in commit order, so an envelope can arrive twice or out
  of turn; consumers dedupe by `id`. A design that can tolerate neither needs a
  broker with a stronger contract.
- **Carrying data.** Payloads are identifiers and hints; the ephemeral lane is
  capped under `pg_notify`'s limit and the durable one is meant to say
  "re-fetch". Replicating rows, syncing state or streaming files is a different
  tool.
- **More than one database, or a runtime that cannot LISTEN.** Fan-out follows
  the Postgres that holds the outbox; subscribers on another database need a
  broker. The listener needs a dedicated connection that bypasses
  transaction-mode pooling — a runtime that reaches Postgres only through such a
  pooler has no LISTEN side.
- **Two-way traffic.** SSE is one direction, server to browser, with header auth
  on a fetch. What the browser sends goes through your API; a bidirectional
  channel is a WebSocket.

## The shape

```
 producer process                          subscriber process
 ─────────────────                         ──────────────────
 db.transaction(tx =>                      listener (LISTEN) ──▶ createEventRelay
   write state                                                    │ pointer → store.readSince
   publisher.emit(event, tx) ──┐                                  ▼
 )                             │           createEventHub (per-scope fan-out, audience filter)
   │ durable: outbox row       │                                  │
   ▼ + notify(pointer)         │ ephemeral:                       ▼
 outbox table                  └─ notify(envelope inline)   createEventStreamHandler ──▶ SSE ──▶ ./client
```

Two lanes. **Durable** events are rows in a consumer-owned outbox, delivered
at-least-once and replayable by watermark; the notification is only a
`(scopeKey, seq)` pointer. **Ephemeral** events (progress ticks, invalidation
hints) ride inline on the notification channel, capped under `pg_notify`'s
~8000-byte limit, and are never stored. `scopeKey` is whatever partitions your
subscribers — a tenant, a workspace, a constant in a single-scope app.

## Why the same transaction

`emit(event, tx)` inserts the outbox row **and** issues the notify on the
caller's transaction. Postgres delivers NOTIFY at COMMIT, so the state change,
the row and the wakeup are one atomic unit: you can never notify about a write
that rolled back, and never commit a write whose event was lost. This is also
why the store and publisher **throw** instead of returning a `Result` — inside
the transaction, the throw is the rollback. Do not catch-and-continue around
`emit`.

## Setting it up

### 1. Store + publisher (every process that writes)

```ts
import { createEventPublisher } from '@octabits-io/events';
import { createDrizzleEventOutboxStore } from '@octabits-io/events/drizzle';

const store = createDrizzleEventOutboxStore({ db, table: eventOutbox, channel: 'app_events', scope: { column: 'orgId' } });
const publisher = createEventPublisher({ store });

await db.transaction(async (tx) => {
  const order = await createOrder(params, tx);
  await publisher.emit({
    type: 'order.created',                       // your taxonomy
    scopeKey, lane: 'durable',
    data: { orderId: order.id },                 // identifiers, never entities
    audience: { permission: { orders: ['read'] } }, // opaque — evaluated by YOUR predicate
    resources: [`order:${order.id}`],
  }, tx);
});

await publisher.emit({ type: 'sync.progress', scopeKey, lane: 'ephemeral', data: { done: 3, total: 9 } });
```

Any store that can append in the caller's transaction and read by watermark
implements `EventOutboxStore` — the seam is four methods over the envelope
codec exported here (`encodeEventPointer`, `encodeInlineEvent`,
`decodeEventNotification`), which a store must use rather than re-derive: a
wire format is shared by import, not by copy.

**Typed emit** is opt-in. Declare the vocabulary as Zod schemas, derive the
`EventDataMap`, and both layers hold you to it — `emit` correlates `type` with
its payload at compile time, `payloadSchemas` enforces it at runtime and is
authoritative when present (unregistered type or failing payload → throw):

```ts
const EVENT_SCHEMAS = { 'order.created': z.object({ orderId: z.number() }) } as const;
type AppEvents = { [K in keyof typeof EVENT_SCHEMAS]: z.infer<(typeof EVENT_SCHEMAS)[K]> };
const publisher = createEventPublisher<AppEvents>({ store, payloadSchemas: EVENT_SCHEMAS });
```

### 2. Listener + relay + hub (the process serving subscribers)

```ts
import { createEventHub, createEventRelay } from '@octabits-io/events';
import { createPgNotifyListener } from '@octabits-io/events/postgres';

const hub = createEventHub({ logger });
const listener = createPgNotifyListener({
  connectionString: databaseDirectUrl,   // MUST bypass transaction-mode poolers
  channel: 'app_events',
  logger,
});
const relay = createEventRelay({ hub, store, listener, logger });
await relay.start();                     // relay.stop() in graceful shutdown
```

The listener must be a **dedicated connection** (a pooled checkout silently
drops the LISTEN registration on reset) and must **bypass PgBouncer-style
transaction pooling**. Reconnect is built in (full jitter); after a reconnect
the relay re-reads every active scope from its watermark, because notifications
sent while it was down are gone. On an embedded database none of that applies —
`./pglite` is the same contract with no connection and no reconnect, taking the
instance structurally.

`logger` is structural (`EventsLogger`: `info`, `warn`, `error(message, error?, attributes?)`).

### 3. The SSE endpoint

```ts
import { createEventStreamHandler } from '@octabits-io/events';

const { handler, metrics } = createEventStreamHandler({
  hub,
  store,                                   // enables Last-Event-ID replay; omit for live-only
  resolveSubscriber: async (request) => {
    const auth = await validateBearer(request.headers.get('authorization'));
    if (!auth) return null;                // → 401
    return { scopeKey: auth.scopeKey, subscriberId: auth.userId, can: (permission) => auth.allows(permission) };
  },
});

app.mount('/events', handler);            // Hono, Elysia, Bun.serve, a Worker — it is (Request) => Response
```

Baked in: `x-accel-buffering: no` (nginx would otherwise hold the stream
forever; Traefik's `buffering` middleware kills SSE regardless), comment-frame
heartbeats (20 s), a **capped connection age** (5 min — auth is re-resolved on
reconnect, so reconnect is routine and replay is exercised constantly),
per-subscriber connection caps (5 → 429), and **the watermark rule**: only
durable events carry an SSE `id:` line, because an ephemeral id would point at
nothing in the outbox.

### 4. The browser

```ts
import { createEventStreamClient } from '@octabits-io/events/client';

const client = createEventStreamClient({
  buildRequest: () => ({ url: '/events', headers: { authorization: `Bearer ${token}` } }),
  onEvent: (event) => { /* event.type, event.data, event.lane, event.id */ },
  onStateChange: (state) => { /* connecting | connected | reconnecting | degraded | stopped */ },
  onResync: () => { /* read again what the page shows */ },
});
client.start();
```

Header auth (native `EventSource` cannot send one), hand-rolled
`Last-Event-ID` and reconnect with full-jitter backoff, a `degraded` state
after repeated failures to drive fallback polling, a durable-only watermark,
and a bounded seen-id set for the at-least-once dedupe.

**`onResync` is not optional in practice.** A connection opened without a
`Last-Event-ID` — the first one of a page, and any before a durable event has
arrived — replays nothing, and the relay does not read the outbox for a scope
with no stream open. So a change committed between the page's first read and
its subscription (a stop and start on an organisation switch, a slow
connect) never reaches it. `onResync` fires on every such connection once the
server has subscribed it; reload there — `defaultInvalidationRegistry.refreshAll()`
re-runs every watcher — and nothing is missed.

## Delivery filtering

Per subscriber, at delivery time, in the hub: scope equality always;
`audience.users` for personal targeting; `audience.permission` as an **opaque
value** handed to the subscriber's `can(...)` — the package never interprets
it, and a permission-carrying event with no evaluator is withheld
(**fail closed**). The rule lives in one exported predicate,
`isEnvelopePermitted`, and replay calls the same one rather than re-deriving
it. If you build your own catch-up source, run it on every envelope.

## Ordering, replay, retention

`seq` is assigned at INSERT but becomes visible in COMMIT order, so a lower
`seq` can commit after a higher one was delivered. Both read paths compensate
by over-reading and deduplicating: the relay reads from
`min(watermark, pointer.seq − 1)`, the SSE handler replays from
`lastEventId − replayLookback` (default 100). Delivery is therefore
**at-least-once, everywhere**; consumers dedupe by envelope `id` (the client
does). The outbox is a delivery log, not a product surface — prune it short
(~24 h); a user-facing activity feed is a separate table written by the same emit.

## Why this shape

- **No broker.** The database you already run is the bus; a second system
  would be a second thing to keep consistent with it.
- **Outbox in the transaction, not after it.** Every "publish after commit"
  design eventually publishes for a rollback or loses an event on a crash
  between the two. Here neither is possible.
- **Seams, not adapters.** Store, listener, logger and permission evaluation
  are structural types; nothing here knows your ORM, your auth, or your tenancy.
- **A fetch handler, not a route.** The SSE endpoint spends no route-type
  budget and emits no client types; it mounts anywhere a `Request` arrives.

Grew up inside `@octabits-io/framework` (`./events`) with the client in
`@octabits-io/agent-ui`.

## Broadcast channels (`./broadcast`)

Sometimes you do not want an event. You want to tell the other processes to
drop a cache entry, and you do not care whether one that was restarting hears
it. That is `createBroadcastChannel`: the same NOTIFY wire, none of the
taxonomy — no envelope, no outbox, no audience filtering, no SSE. **A
broadcast message is a hint, not a fact.** Delivery is at-most-once with no
replay, so every consumer needs its own correctness backstop, normally a TTL
on whatever the hint invalidates; the channel only shortens the staleness
window.

```ts
import { createBroadcastChannel } from '@octabits-io/events/broadcast';

const invalidations = createBroadcastChannel({
  channel: 'app_cache_invalidation',
  schema: z.object({ namespace: z.string(), scopeKey: z.string() }),
  logger,
});

// Best-effort hint — any process, regular (pooled) connection. Database
// failures are logged, never thrown; safe to void-discard.
await invalidations.publish(db, { namespace: 'config', scopeKey });

// Inside a transaction — Postgres delivers at COMMIT and drops on ROLLBACK,
// so the hint can never announce a write that rolled back. Failures throw.
await db.transaction(async (tx) => {
  await writeTheThing(tx);
  await invalidations.publishInTx(tx, { namespace: 'config', scopeKey });
});

// Subscribe — one listener per process, DIRECT connection string (the same
// LISTEN constraint as the relay), or `listener: createPgliteNotifyListener(…)`
// on an embedded database. onReconnect fires after a gap in which messages
// were lost: flush whatever the channel invalidates.
const sub = await invalidations.subscribe({
  connectionString: directDatabaseUrl,
  onMessage: ({ namespace, scopeKey }) => caches[namespace]?.invalidate(scopeKey),
  onReconnect: () => Object.values(caches).forEach((c) => c.clear()),
});
```

Both publish methods throw on schema or size violations — those are
programming errors at the publish site. On a database failure they diverge,
which is why they are two named methods rather than one flag: `publish` logs
and swallows (the hint is lost, the TTL covers it), `publishInTx` rethrows (a
failed statement has aborted the transaction, and the caller's rollback
handling must see it). Never hand a transaction context to `publish`. Both
take a `BroadcastDatabase` — `execute(query)`, satisfied by a Drizzle db
instance and by a transaction context — because the statement is built with
drizzle's `sql` template. `subscribe` throws on a first-connect failure
(boot-time misconfiguration fails loudly) and reconnects afterwards;
schema-invalid payloads are dropped silently, and a throwing `onMessage` is
logged without taking the listener down.

Lived in `@octabits-io/framework/drizzle/broadcast` until 2026-09-16.

## Webhooks (`./webhooks`)

The durable lane, delivered to another system: a ticket system, a chat tool,
a customer's own integration. No broker and no job runner — the delivery
table is the queue.

```ts
import { createWebhookDispatcher, staticWebhookSubscriptions, withWebhookDeliveries } from '@octabits-io/events/webhooks';
import { createPostgresWebhookDeliveryStore, webhookDeliveryDdl } from '@octabits-io/events/webhooks/postgres';

await pool.query(webhookDeliveryDdl({ scope: { column: 'org_id' } }));   // your scope column, as everywhere
const deliveries = createPostgresWebhookDeliveryStore({ db: pool, scope: { column: 'org_id' } });
const subscriptions = staticWebhookSubscriptions([
  { id: 'tickets', url: 'https://tickets.example/hooks/approvals', secret: process.env.TICKETS_WEBHOOK_SECRET!, types: ['request.*'] },
]);

// Every durable emit through this store also writes its deliveries, in the same transaction.
const store = withWebhookDeliveries(createPostgresEventOutboxStore({ db: pool, channel: 'app_events' }), { deliveries, subscriptions });

// One per replica is fine: a claim takes rows FOR UPDATE SKIP LOCKED.
const dispatcher = createWebhookDispatcher({ store: deliveries, subscriptions });
dispatcher.start();
```

- **Same transaction.** A delivery exists exactly when its event does: a
  rolled-back write leaves neither. Ephemeral events have no deliveries.
  The plain-SQL stores write in `tx` only when it is an executor
  (`query(text, params)`); any other object passed as `tx` — a Drizzle or
  Kysely transaction — is refused with a `TypeError`, never taken for "no
  transaction". On Drizzle, pair `createDrizzleEventOutboxStore` with
  `createDrizzleWebhookDeliveryStore` from `./webhooks/drizzle` and pass the
  Drizzle `tx` to both, or wrap it with `drizzleExecutor(tx)` from
  `@octabits-io/postgres/drizzle` for a plain-SQL store.
- **Signed after [Standard Webhooks](https://www.standardwebhooks.com/).**
  `webhook-id` (the event id, the same on every attempt, so a receiver
  dedupes on it), `webhook-timestamp`, `webhook-signature: v1,<HMAC-SHA256>`.
  A `whsec_…` secret is used as its decoded bytes, any other string as UTF-8.
  `verifyWebhook` is the receiving side, with a five-minute timestamp
  tolerance and every `v1,` signature tried, so a secret can be rotated.
- **Retries.** A 2xx is delivered. Anything else — another status, a
  redirect (never followed), a timeout, a network error — is retried after
  5 s, 5 min, 30 min, 2 h, 5 h, 10 h and 10 h; after that the delivery is
  `dead`. A 410 Gone, or a subscription that no longer exists, gives it up at
  once. `retryScheduleMs` changes the schedule.
- **At-least-once.** A claim leases the row (`leaseMs`, default 60 s); a worker
  that dies mid-request leaves the lease to run out and the delivery is sent
  again.
- **The log.** `store.list({ scopeKey, status, eventId, subscriptionId })`
  shows every delivery with its attempts, last status and last error;
  `store.redeliver(id)` puts one back with a fresh schedule; `store.prune(before)`
  removes finished ones.
- **What leaves.** The body is the envelope without `audience`. Keep event
  data to identifiers. The subscription's secret is never stored with a
  delivery; each attempt goes to the subscription's URL at that moment.

The subscriptions are a seam: `staticWebhookSubscriptions` takes a list from
configuration, and a host that keeps them in a table implements `for(envelope)`
— a lookup inside the event's transaction, never a call out — and `get(id)`.
The scope column is yours (`scope: { column }` on the DDL and the store): every
row is stamped with its event's `scopeKey`, for a scope's log and its
row-level security policy. Without one, listing by scope is refused.

## Wire format

Three things cross a boundary, and all three are specified:

| Document | What |
|---|---|
| `schema/event-envelope.schema.json` | `EventEnvelope` — the outbox row's columns, the inline form in a notification, the `data:` line of an SSE frame, the browser client's input |
| `schema/event-notification.schema.json` | What travels on the notification channel, discriminated on `k`: `{ k: 'ptr', s: scopeKey, q: seq }` for a durable row, `{ k: 'evt', e: envelope }` for an ephemeral event inline |
| the SSE frame | text, not JSON: `id: <seq>` on durable events only (the replay watermark the client sends back as `Last-Event-ID`), `event: <type>`, `data: <envelope as JSON>`, a blank line; `retry: <ms>` once on connect; `: hb` as the heartbeat comment |

The two JSON Schemas (draft 2020-12) are generated from `src/schema.ts` and
the codec's two message kinds and checked in `pnpm lint`; they are in the
package (`@octabits-io/events/schema/event-envelope.schema.json`) and at their
`$id`, the raw GitHub URL. A producer or a reader in another language builds
against them; the tests validate what the codec encodes against them with a
JSON Schema validator, not with Zod.

## License

MIT.