David Snelling fd5edb5a13 feat: brain.onChange — the in-process change feed for every committed mutation

Subscribe once and receive one post-commit event per affected record for
EVERY canonical write, regardless of origin: direct calls, batch methods,
transact(), imports, and Virtual Filesystem writes all funnel through the
same commit points the feed is emitted from. This is the authoritative
in-process signal for live UIs, cache invalidation, and realtime sync layers
that forward it over their own transports.

Architecture — the post-commit dual of the ifRev precommit hook: mutation
methods hand lightweight event descriptors to the commit seam
(persistSingleOp / transact's plan), which stamps the committed
{generation, timestamp}, enriches entity deletes with the record's LAST
committed state from the commit's own before-images (free — Model B reads
them anyway; removeMany's id-only deletes gain full payloads this way), and
emits only after the commit succeeds — a losing CAS or rejected batch never
announces anything. Dispatch is a microtask FIFO after the mutex releases:
commit-ordered, a slow listener never delays a write, a throwing listener is
logged and isolated, and with no subscribers the write path constructs no
events at all. Single-point emission was chosen over per-method hooks
because the aggregation-hook pattern demonstrably drifted (relation ops and
removeMany were silently missing from it).

Coverage: add/update/remove (+ cascade unrelate per deleted relationship),
relate (both edges when bidirectional)/unrelate/updateRelation, per-item
events for addMany/updateMany/relateMany/removeMany, per-item events sharing
one generation for transact(), and transitively imports + VFS. clear() and
restore() — wholesale raw-state operations outside the per-record commit
path — emit a single store-level event meaning "refetch everything".
brain.close() drops all listeners.

New module src/events/changeFeed.ts (BrainyChangeEvent + ChangeFeed,
exported from the package root); guide docs/guides/reacting-to-changes.md.
Integration suite pins the contract: per-op payload fidelity, delete
last-state payloads, batch per-item emission, one-generation transact
batches, VFS-origin events, CAS-loser silence, ordering, listener isolation,
unsubscribe, and store-level events.

2026-07-10 11:24:31 -07:00

4.3 KiB

Raw Blame History

title

slug

public

Reacting to Changes

brain.onChange(cb) is Brainy's in-process change feed: subscribe once and receive one event per committed mutation — every mutation, regardless of how it happened. Direct calls, batch methods, transact(), imports, and Virtual Filesystem writes all funnel through the same commit point the feed is emitted from, so nothing slips past it.

const off = brain.onChange((e) => {
  if (e.kind === 'entity') {
    console.log(`${e.op} ${e.entity?.type} ${e.id} @ generation ${e.generation}`)
  }
})

await brain.add({ data: 'Ada Lovelace', type: 'person' })
// → "add person 0198... @ generation 42"

off() // unsubscribe when done

The event

interface BrainyChangeEvent {
  kind: 'entity' | 'relation' | 'store'
  op: 'add' | 'update' | 'remove' | 'relate' | 'unrelate' | 'updateRelation'
    | 'clear' | 'restore'
  id?: string
  entity?:   { id: string; type: string; subtype?: string;
               metadata: Record<string, unknown>; service?: string }
  relation?: { id: string; from: string; to: string; type: string;
               metadata?: Record<string, unknown> }
  generation?: number
  timestamp: number
}

Entity events (add / update / remove) carry the post-commit indexed view — type, subtype, and the full custom metadata, so you can match your own where-style filters against events without a read.
Deletes are fully described. A remove or unrelate event carries the record's last committed state (sourced from the commit's own history record), not just an id.
Batches emit per item. addMany / updateMany / relateMany / removeMany emit one event per affected record; a transact() batch emits one event per item, all sharing the batch's single generation.
Cascades are visible. Removing an entity also emits unrelate for each relationship the delete cascaded to.
Store-level events (kind: 'store') fire for the two wholesale operations — clear() and restore() — and mean "everything may have changed; refetch what you care about."

Delivery guarantees

Post-commit only. An aborted write — a losing ifRev compare-and-swap, a rejected transaction — never emits. If you received the event, the write is durable.
Commit-ordered. Events arrive in the order writes committed; generation is monotonic.
Asynchronous, never blocking. Delivery happens in a microtask after the write completes. A slow listener cannot delay a write; a throwing listener is logged and isolated from other listeners.
Zero overhead when unused. With no subscribers, the write path does no event work at all.
Fire-and-forget. There is no replay or backpressure. For catch-up after a disconnect, use the generation on each event together with asOf() / the transaction log: record the last generation you processed, and on reconnect diff from there.

Patterns

Cache invalidation — drop cached reads for whatever changed:

brain.onChange((e) => {
  if (e.kind === 'store') return cache.clear()
  if (e.id) cache.delete(e.id)
})

Live queries (notify-and-refetch) — re-run a query when a relevant change lands, rather than diffing incrementally:

brain.onChange((e) => {
  if (e.kind === 'entity' && e.entity?.type === 'order') {
    refreshOpenOrdersView() // debounce as needed
  }
})

Forwarding to other processes — the feed is in-process by design. To push changes to browsers or other services, forward events through your own transport (WebSocket, SSE) from the process that owns the brain.

Lifecycle

onChange returns an unsubscribe function — call it when tearing down a subscriber (for example, when evicting a pooled instance). brain.close() drops all listeners; no events are delivered for or after close().

4.3 KiB Raw Blame History