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

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

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

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)

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

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

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.

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.

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

@octabits-io/events