soulcraft/brainy
1
0
Fork 0
Code Issues 1 Pull requests Projects Releases 335 Packages Wiki Activity Actions

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

Browse source 
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.
This commit is contained in:
David Snelling 2026-07-10 11:24:31 -07:00
parent ee3db2aae4
commit fd5edb5a13
5 changed files with 975 additions and 15 deletions
Download patch file Download diff file Expand all files Collapse all files

114
docs/guides/reacting-to-changes.md Normal file
View file

@ -0,0 +1,114 @@
---
title: Reacting to Changes
slug: guides/reacting-to-changes
public: true
category: guides
template: guide
order: 11
description: Subscribe to every committed mutation with `brain.onChange` — the in-process change feed behind live UIs, cache invalidation, and realtime sync. Covers the event shape, delivery guarantees, and catch-up patterns.
next:
- guides/optimistic-concurrency
- guides/snapshots-and-time-travel
---
# 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.
```ts
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
```ts
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](optimistic-concurrency.md), 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](snapshots-and-time-travel.md): record the
last generation you processed, and on reconnect diff from there.
## Patterns
**Cache invalidation** — drop cached reads for whatever changed:
```ts
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:
```ts
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()`.

469
src/brainy.ts
View file

@ -167,6 +167,12 @@ import {
type ImportResult
} from './db/portableGraph.js'
import { GenerationStore, type CommitBeforeImages } from './db/generationStore.js'
import {
ChangeFeed,
type BrainyChangeEvent,
type ChangeListener,
type PendingChangeEvent
} from './events/changeFeed.js'
import { isDeterministicEmbedMode } from './embeddings/deterministicEmbedMode.js'
import { GenerationConflictError } from './db/errors.js'
import { BrainyError, GraphIndexNotReadyError, MetadataIndexNotReadyError, MigrationInProgressError } from './errors/brainyError.js'
@ -335,6 +341,13 @@ interface PlannedTransact {
* commit precondition leaves them untouched.
*/
createdNouns: Set<string>
/**
* Change-feed events, one per affected record in op order, populated by the
* planners ONLY when a listener is subscribed. Stamped with the batch's
* committed generation and emitted after `commitTransaction` returns — a
* rejected batch (CAS conflict, failed apply) emits nothing.
*/
changeEvents: PendingChangeEvent[]
}
/**
@ -491,6 +504,15 @@ export class Brainy<T = any> implements BrainyInterface<T> {
* upgrade verifies + stamps; retained on failure. See {@link createMigrationBackupIfNeeded}. */
private _migrationBackupPath: string | null = null
/**
* The in-process change feed behind {@link onChange}. Emitted from the
* commit seam ({@link persistSingleOp} / {@link transact}), so every
* canonical mutation — any origin — produces exactly one post-commit event
* per affected record. See src/events/changeFeed.ts for the delivery
* contract.
*/
private readonly _changeFeed = new ChangeFeed()
/** Set when the on-open VFS-blob adoption left one or more `_cow/` blobs it
* could not fully adopt (bytes or metadata missing). While true, the
* pre-upgrade backup is NOT auto-removed — the upgrade is not verifiably
@ -1553,8 +1575,21 @@ export class Brainy<T = any> implements BrainyInterface<T> {
private async persistSingleOp(
touched: { nouns?: string[]; verbs?: string[] },
run: TransactionFunction<void>,
precommit?: (before: CommitBeforeImages) => void
): Promise<void> {
precommit?: (before: CommitBeforeImages) => void,
pendingEvents?: PendingChangeEvent[]
): Promise<{ generation?: number; timestamp: number }> {
// Change-feed capture: when this write will emit, hold a reference to the
// commit's before-images so `remove` events can carry the record's last
// committed state (free — the commit reads them anyway for Model B).
let capturedBefore: CommitBeforeImages | undefined
const captureAndCheck =
pendingEvents && pendingEvents.length > 0
? (before: CommitBeforeImages): void => {
capturedBefore = before
precommit?.(before)
}
: precommit
if (!this._generationStampingActive) {
// Init-time / infrastructure baseline write (e.g. the VFS root): apply
// WITHOUT creating a generation. Generation 0 is the freshly-materialized
@ -1562,7 +1597,7 @@ export class Brainy<T = any> implements BrainyInterface<T> {
// Bootstrap writes are single-threaded, so a supplied precondition is
// honored against directly-read before-images (no commit mutex exists
// on this path — nothing to race).
if (precommit) {
if (captureAndCheck) {
const nouns = new Map<string, { kind: 'noun'; metadata: unknown; vector: unknown }>()
for (const id of touched.nouns ?? []) {
const prev = await this.storage.readNounRaw(id)
@ -1573,18 +1608,88 @@ export class Brainy<T = any> implements BrainyInterface<T> {
const prev = await this.storage.readVerbRaw(id)
verbs.set(id, { kind: 'verb', metadata: prev.metadata, vector: prev.vector })
}
precommit({ nouns, verbs } as CommitBeforeImages)
captureAndCheck({ nouns, verbs } as CommitBeforeImages)
}
await this.generationStore.runWithoutGeneration(() =>
this.transactionManager.executeTransaction(run)
)
return
const timestamp = Date.now()
// Bootstrap writes are not generation-stamped; emit without one.
this.emitCommitted(pendingEvents, capturedBefore, undefined, timestamp)
return { timestamp }
}
await this.generationStore.commitSingleOp({
const receipt = await this.generationStore.commitSingleOp({
touched,
precommit,
precommit: captureAndCheck,
execute: () => this.transactionManager.executeTransaction(run)
})
// POST-COMMIT ONLY: an aborted commit (CAS conflict, failed apply) throws
// above and never reaches this line — the feed cannot announce a write
// that did not become durable.
this.emitCommitted(pendingEvents, capturedBefore, receipt.generation, receipt.timestamp)
return receipt
}
/**
* @description Stamp pending change events with their commit receipt,
* enrich entity `remove` events with the record's last committed state
* (from the commit's before-images), and hand them to the change feed for
* post-mutex dispatch. No-op when the write produced no events.
* @param pending - Events the mutation method constructed pre-commit
* (only when a listener is subscribed — see {@link ChangeFeed.hasListeners}).
* @param before - The commit's before-images (delete-payload source).
* @param generation - The committed generation (absent for bootstrap writes).
* @param timestamp - The commit timestamp.
*/
private emitCommitted(
pending: PendingChangeEvent[] | undefined,
before: CommitBeforeImages | undefined,
generation: number | undefined,
timestamp: number
): void {
if (!pending || pending.length === 0) return
const events: BrainyChangeEvent[] = pending.map((p) => {
let entity = p.entity
if (!entity && p.kind === 'entity' && p.op === 'remove' && p.id && before) {
const record = before.nouns.get(p.id)?.metadata as
| Record<string, unknown>
| null
| undefined
if (record) {
entity = this.entityViewFromRawRecord(p.id, record)
}
}
return {
...p,
...(entity && { entity }),
...(generation !== undefined && { generation }),
timestamp
}
})
this._changeFeed.emit(events)
}
/**
* @description Build the change-event entity view from a stored flat
* metadata record (the shape before-images and pre-delete reads carry),
* using THE canonical reserved/custom split so the view can never drift
* from the live read paths.
* @param id - The entity's canonical UUID.
* @param record - The stored flat metadata record.
* @returns The `ChangeEventEntity` view (type, subtype, service, custom metadata).
*/
private entityViewFromRawRecord(
id: string,
record: Record<string, unknown>
): NonNullable<BrainyChangeEvent['entity']> {
const { reserved, custom } = splitNounMetadataRecord(record)
return {
id,
type: String(reserved.noun ?? 'unknown'),
...(reserved.subtype !== undefined && { subtype: String(reserved.subtype) }),
metadata: custom,
...(reserved.service !== undefined && { service: String(reserved.service) })
}
}
/**
@ -1795,10 +1900,28 @@ export class Brainy<T = any> implements BrainyInterface<T> {
// or merge (upsert); a merge that then hits a concurrent delete retries
// the insert. Each persistSingleOp call constructs fresh operations, so
// re-running the callback is safe.
// Change feed: built only when someone is listening (zero-cost gate).
const addEvents: PendingChangeEvent[] | undefined = this._changeFeed.hasListeners
? [
{
kind: 'entity',
op: 'add',
id,
entity: {
id,
type: String(params.type),
...(params.subtype !== undefined && { subtype: String(params.subtype) }),
metadata: (params.metadata as Record<string, unknown>) ?? {},
...(params.service !== undefined && { service: String(params.service) })
}
}
]
: undefined
const MAX_UPSERT_ATTEMPTS = 10
for (let attempt = 0; ; attempt++) {
try {
await this.persistSingleOp({ nouns: [id] }, runInsert, insertPrecommit)
await this.persistSingleOp({ nouns: [id] }, runInsert, insertPrecommit, addEvents)
break
} catch (err) {
if (!(err instanceof InsertPreconditionExistsSignal)) {
@ -2786,7 +2909,26 @@ export class Brainy<T = any> implements BrainyInterface<T> {
tx.addOperation(
new AddToMetadataIndexOperation(this.metadataIndex, params.id, entityForIndexing)
)
}, casPrecommit)
}, casPrecommit, this._changeFeed.hasListeners
? [
{
kind: 'entity',
op: 'update',
id: params.id,
entity: {
id: params.id,
type: String(entityForIndexing.type),
...(entityForIndexing.subtype !== undefined && {
subtype: String(entityForIndexing.subtype)
}),
metadata: (newMetadata as Record<string, unknown>) ?? {},
...(entityForIndexing.service !== undefined && {
service: String(entityForIndexing.service)
})
}
}
]
: undefined)
// Aggregation hook (outside transaction — derived data)
if (this._aggregationIndex) {
@ -2876,7 +3018,35 @@ export class Brainy<T = any> implements BrainyInterface<T> {
new DeleteVerbMetadataOperation(this.storage, verb.id)
)
}
})
},
undefined,
this._changeFeed.hasListeners
? [
// The entity delete (payload = last state, from the pre-delete read)
// plus one unrelate per cascade-deleted relationship.
{
kind: 'entity',
op: 'remove',
id,
...(metadata && {
entity: this.entityViewFromRawRecord(id, metadata as Record<string, unknown>)
})
},
...allVerbs.map(
(v): PendingChangeEvent => ({
kind: 'relation',
op: 'unrelate',
id: v.id,
relation: {
id: v.id,
from: v.sourceId,
to: v.targetId,
type: String(v.verb)
}
})
)
]
: undefined)
// Aggregation hook (outside transaction — derived data)
if (this._aggregationIndex && metadata) {
@ -3595,7 +3765,44 @@ export class Brainy<T = any> implements BrainyInterface<T> {
)
)
}
})
},
undefined,
this._changeFeed.hasListeners
? [
{
kind: 'relation',
op: 'relate',
id,
relation: {
id,
from: params.from,
to: params.to,
type: String(params.type),
...(params.metadata && {
metadata: params.metadata as Record<string, unknown>
})
}
},
...(reverseId
? [
{
kind: 'relation',
op: 'relate',
id: reverseId,
relation: {
id: reverseId,
from: params.to,
to: params.from,
type: String(params.type),
...(params.metadata && {
metadata: params.metadata as Record<string, unknown>
})
}
} as PendingChangeEvent
]
: [])
]
: undefined)
return id
}
@ -3643,7 +3850,26 @@ export class Brainy<T = any> implements BrainyInterface<T> {
tx.addOperation(
new DeleteVerbMetadataOperation(this.storage, id)
)
})
},
undefined,
this._changeFeed.hasListeners && verb
? [
{
kind: 'relation',
op: 'unrelate',
id,
relation: {
id,
from: verb.sourceId,
to: verb.targetId,
type: String(verb.verb),
...(verb.metadata && {
metadata: verb.metadata as Record<string, unknown>
})
}
}
]
: undefined)
}
/**
@ -3778,7 +4004,26 @@ export class Brainy<T = any> implements BrainyInterface<T> {
)
)
}
})
},
undefined,
this._changeFeed.hasListeners
? [
{
kind: 'relation',
op: 'updateRelation',
id: params.id,
relation: {
id: params.id,
from: verbForIndex.sourceId,
to: verbForIndex.targetId,
type: String(verbForIndex.verb ?? verbForIndex.type),
...(verbForIndex.metadata && {
metadata: verbForIndex.metadata as Record<string, unknown>
})
}
}
]
: undefined)
}
/**
@ -6335,6 +6580,16 @@ export class Brainy<T = any> implements BrainyInterface<T> {
}
try {
// Change feed: the BUILDER populates this (per id actually staged), so
// an id whose builder failed never emits — the array is read only
// after the chunk's commit succeeds. Verb events dedupe within the
// chunk (two related chunk-members see the same cascade verb twice).
const chunkEvents: PendingChangeEvent[] | undefined = this._changeFeed
.hasListeners
? []
: undefined
const seenVerbEvents = new Set<string>()
// Process chunk as ONE atomic Model-B generation (entities + cascade verbs).
await this.persistSingleOp({ nouns: touchedNouns, verbs: touchedVerbs }, async (tx) => {
for (const id of chunk) {
@ -6373,6 +6628,35 @@ export class Brainy<T = any> implements BrainyInterface<T> {
)
}
if (chunkEvents) {
chunkEvents.push({
kind: 'entity',
op: 'remove',
id,
...(metadata && {
entity: this.entityViewFromRawRecord(
id,
metadata as Record<string, unknown>
)
})
})
for (const verb of allVerbs) {
if (seenVerbEvents.has(verb.id)) continue
seenVerbEvents.add(verb.id)
chunkEvents.push({
kind: 'relation',
op: 'unrelate',
id: verb.id,
relation: {
id: verb.id,
from: verb.sourceId,
to: verb.targetId,
type: String(verb.verb)
}
})
}
}
chunkQueued.push(id)
} catch (error) {
chunkBuilderFailed.push({
@ -6384,7 +6668,7 @@ export class Brainy<T = any> implements BrainyInterface<T> {
}
}
}
})
}, undefined, chunkEvents)
// Transaction committed — queued IDs were actually deleted
result.successful.push(...chunkQueued)
@ -6690,6 +6974,12 @@ export class Brainy<T = any> implements BrainyInterface<T> {
// VFS was never used, reset flag for clean state
this._vfsInitialized = false
}
// Change feed: clear() wipes the store wholesale (raw, not per-record) —
// one store-level event tells subscribers everything they held is gone.
this._changeFeed.emit([
{ kind: 'store', op: 'clear', timestamp: Date.now() }
])
}
// ─── Migration API ───────────────────────────────────────────────
@ -7124,6 +7414,10 @@ export class Brainy<T = any> implements BrainyInterface<T> {
hook()
}
// Change feed: the batch's events share its single committed generation.
// A rejected batch throws at commitTransaction and never reaches here.
this.emitCommitted(plan.changeEvents, undefined, generation, timestamp)
const receipt: TransactReceipt = { generation, timestamp, ids: plan.ids }
return this.createPinnedDb({ generation, timestamp, receipt })
}
@ -7667,6 +7961,12 @@ export class Brainy<T = any> implements BrainyInterface<T> {
this.index.rebuild(),
this.graphIndex.rebuild()
])
// Change feed: a restore is a wholesale raw-state replacement, not a
// per-record commit — one store-level event tells subscribers to refetch.
this._changeFeed.emit([
{ kind: 'store', op: 'restore', timestamp: Date.now() }
])
}
/**
@ -8251,7 +8551,8 @@ export class Brainy<T = any> implements BrainyInterface<T> {
touchedVerbs: [],
postCommit: [],
casUpdates: [],
createdNouns: new Set()
createdNouns: new Set(),
changeEvents: []
}
for (const op of ops) {
@ -8445,6 +8746,20 @@ export class Brainy<T = any> implements BrainyInterface<T> {
this._aggregationIndex.onEntityAdded(id, entityForIndexing)
}
})
if (this._changeFeed.hasListeners) {
plan.changeEvents.push({
kind: 'entity',
op: 'add',
id,
entity: {
id,
type: String(params.type),
...(params.subtype !== undefined && { subtype: String(params.subtype) }),
metadata: (params.metadata as Record<string, unknown>) ?? {},
...(params.service !== undefined && { service: String(params.service) })
}
})
}
state.nouns.set(id, { metadata: storageMetadata, vector })
state.removedNouns.delete(id)
@ -8611,6 +8926,24 @@ export class Brainy<T = any> implements BrainyInterface<T> {
this._aggregationIndex.onEntityUpdated(params.id, entityForIndexing, oldEntityForAgg)
}
})
if (this._changeFeed.hasListeners) {
plan.changeEvents.push({
kind: 'entity',
op: 'update',
id: params.id,
entity: {
id: params.id,
type: String(entityForIndexing.type),
...(entityForIndexing.subtype !== undefined && {
subtype: String(entityForIndexing.subtype)
}),
metadata: (newMetadata as Record<string, unknown>) ?? {},
...(entityForIndexing.service !== undefined && {
service: String(entityForIndexing.service)
})
}
})
}
state.nouns.set(params.id, { metadata: updatedMetadata, vector })
return params.id
@ -8676,8 +9009,31 @@ export class Brainy<T = any> implements BrainyInterface<T> {
plan.touchedVerbs.push(verb.id)
state.verbs.delete(verb.id)
state.removedVerbs.add(verb.id)
if (this._changeFeed.hasListeners) {
plan.changeEvents.push({
kind: 'relation',
op: 'unrelate',
id: verb.id,
relation: {
id: verb.id,
from: verb.sourceId,
to: verb.targetId,
type: String(verb.verb)
}
})
}
}
plan.touchedNouns.push(id)
if (this._changeFeed.hasListeners) {
plan.changeEvents.push({
kind: 'entity',
op: 'remove',
id,
...(metadata && {
entity: this.entityViewFromRawRecord(id, metadata as Record<string, unknown>)
})
})
}
if (metadata) {
// Canonical reserved/custom split — mirror of remove()'s aggregation hook.
@ -8813,6 +9169,20 @@ export class Brainy<T = any> implements BrainyInterface<T> {
plan.touchedVerbs.push(id)
state.verbs.set(id, verb)
state.removedVerbs.delete(id)
if (this._changeFeed.hasListeners) {
plan.changeEvents.push({
kind: 'relation',
op: 'relate',
id,
relation: {
id,
from: params.from,
to: params.to,
type: String(params.type),
...(params.metadata && { metadata: params.metadata as Record<string, unknown> })
}
})
}
if (params.bidirectional) {
const reverseId = uuidv4()
@ -8841,6 +9211,20 @@ export class Brainy<T = any> implements BrainyInterface<T> {
plan.touchedVerbs.push(reverseId)
state.verbs.set(reverseId, reverseVerb)
state.removedVerbs.delete(reverseId)
if (this._changeFeed.hasListeners) {
plan.changeEvents.push({
kind: 'relation',
op: 'relate',
id: reverseId,
relation: {
id: reverseId,
from: params.to,
to: params.from,
type: String(params.type),
...(params.metadata && { metadata: params.metadata as Record<string, unknown> })
}
})
}
}
return id
@ -8869,6 +9253,21 @@ export class Brainy<T = any> implements BrainyInterface<T> {
}
plan.operations.push(new DeleteVerbMetadataOperation(this.storage, id))
plan.touchedVerbs.push(id)
if (this._changeFeed.hasListeners) {
plan.changeEvents.push({
kind: 'relation',
op: 'unrelate',
id,
...(verb && {
relation: {
id,
from: verb.sourceId,
to: verb.targetId,
type: String(verb.verb)
}
})
})
}
state.verbs.delete(id)
state.removedVerbs.add(id)
@ -9358,6 +9757,43 @@ export class Brainy<T = any> implements BrainyInterface<T> {
return this._hub
}
/**
* @description Subscribe to this brain's committed mutations — the
* authoritative in-process change feed. Fires exactly once per affected
* record for EVERY canonical write, regardless of origin: direct API calls,
* batch methods, `transact()`, imports, the Virtual Filesystem, and
* native-accelerated deployments all funnel through the same commit point
* this feed is emitted from.
*
* Delivery contract (see {@link BrainyChangeEvent}):
* - **Post-commit only** — an aborted write (e.g. a losing `ifRev`
* compare-and-swap) never emits.
* - **Commit-ordered, asynchronous** — events arrive in the order writes
* became durable, dispatched in a microtask so a slow listener never
* delays a write. A throwing listener is isolated (logged, others
* unaffected).
* - **Fully described** — `remove`/`unrelate` events carry the record's
* LAST committed state, and batch operations emit one event per item.
* - **Fire-and-forget** — no replay or backpressure. Each event carries its
* committed `generation`, so catch-up after a gap can be built on
* {@link transactionLog} / {@link asOf}.
* - **Zero overhead when unused** — with no subscribers the write path does
* no event work at all.
*
* @param listener - Called once per committed mutation.
* @returns An unsubscribe function — call it when done (e.g. on teardown of
* a pooled instance) to stop delivery.
* @example
* const off = brain.onChange((e) => {
* if (e.kind === 'entity') console.log(e.op, e.entity?.type, e.id)
* })
* await brain.add({ data: 'hello', type: 'document' }) // → "add document <id>"
* off()
*/
onChange(listener: ChangeListener): () => void {
return this._changeFeed.subscribe(listener)
}
/**
* Get Triple Intelligence System
* Advanced pattern recognition and relationship analysis
@ -14587,6 +15023,9 @@ export class Brainy<T = any> implements BrainyInterface<T> {
* This ensures deferred persistence mode data is saved
*/
async close(): Promise<void> {
// Change-feed teardown: no events are delivered for or after close().
this._changeFeed.close()
// Phase 0a: Persist buffered single-op generation history (async
// group-commit) before anything else, so a clean close never drops history
// the caller already observed. No-op when nothing is pending or read-only.

171
src/events/changeFeed.ts Normal file
View file

@ -0,0 +1,171 @@
/**
* @module events/changeFeed
* @description The in-process change feed behind {@link Brainy.onChange} — the
* authoritative "something committed" signal for every canonical mutation,
* regardless of origin (direct API calls, batches, transactions, imports, the
* VFS, or an accelerated native deployment: all of them funnel through the
* same generation-store commit points this feed is emitted from).
*
* Design properties:
* - **Post-commit only.** Events are enqueued after the commit succeeds, so an
* aborted commit (a losing `ifRev` CAS, a failed transaction) never emits.
* - **Commit-ordered.** Events are enqueued in commit order and dispatched
* FIFO, so a subscriber observes mutations in the order they became durable.
* - **Never blocks the write path.** Dispatch happens in a microtask after the
* committing call returns; a slow or throwing listener cannot delay or fail
* a write. Listener errors are isolated per listener and per event.
* - **Zero cost when unused.** Callers consult {@link ChangeFeed.hasListeners}
* before constructing event payloads; with no subscribers the write path
* does no event work at all.
* - **Fire-and-forget.** No acknowledgement, backpressure, or replay. Events
* carry the committed `generation`, so a consumer that needs catch-up
* semantics can pair the live feed with `transactionLog()` / `asOf()`.
*/
/** The post-commit view of an entity carried by entity change events. */
export interface ChangeEventEntity {
/** The entity's canonical UUID. */
id: string
/** The entity's NounType string (e.g. `'person'`, `'document'`). */
type: string
/** The entity's subtype, when set. */
subtype?: string
/** The entity's custom (indexed) metadata fields. */
metadata: Record<string, unknown>
/** The writing service, when set. */
service?: string
}
/** The post-commit view of a relationship carried by relation change events. */
export interface ChangeEventRelation {
/** The relationship's canonical UUID. */
id: string
/** Source entity UUID. */
from: string
/** Target entity UUID. */
to: string
/** The relationship's VerbType string (e.g. `'contains'`). */
type: string
/** The relationship's custom metadata fields, when present. */
metadata?: Record<string, unknown>
}
/**
* One committed mutation, as delivered to {@link Brainy.onChange} listeners.
*
* Exactly one of `entity` / `relation` is populated for `kind: 'entity'` /
* `kind: 'relation'` events. `kind: 'store'` events (`clear` / `restore`)
* carry neither — they mean "the whole store changed; refetch what you care
* about".
*
* For `op: 'remove'` / `op: 'unrelate'` the payload is the record's LAST
* committed state (sourced from the commit's own before-image), so deletes are
* fully described rather than id-only.
*/
export interface BrainyChangeEvent {
/** What changed: one record, one relationship, or the whole store. */
kind: 'entity' | 'relation' | 'store'
/** The mutation, in Brainy's own API vocabulary. */
op:
| 'add'
| 'update'
| 'remove'
| 'relate'
| 'unrelate'
| 'updateRelation'
| 'clear'
| 'restore'
/** The mutated record's id (absent for store-level events). */
id?: string
/** Post-commit entity view (entity events only). */
entity?: ChangeEventEntity
/** Post-commit relation view (relation events only). */
relation?: ChangeEventRelation
/**
* The committed generation this mutation belongs to (Model B: every write
* is a generation; a transaction's items share one). Absent for store-level
* events and init-time bootstrap writes, which are not generation-stamped.
*/
generation?: number
/** Commit timestamp (ms since epoch). */
timestamp: number
}
/** Listener signature for {@link Brainy.onChange}. */
export type ChangeListener = (event: BrainyChangeEvent) => void
/**
* A change event as constructed by a mutation method BEFORE its commit: the
* commit seam stamps `generation`/`timestamp` after the write becomes
* durable. An entity `remove` descriptor may omit `entity` — the seam fills
* it from the commit's own before-image (the record's last committed state).
*/
export type PendingChangeEvent = Omit<BrainyChangeEvent, 'generation' | 'timestamp'>
/**
* @description Listener registry + commit-ordered async dispatcher for
* {@link BrainyChangeEvent}s. One instance per {@link Brainy}.
*/
export class ChangeFeed {
private listeners = new Set<ChangeListener>()
private queue: BrainyChangeEvent[] = []
private draining = false
/** Whether any listener is subscribed — the write path's zero-cost gate. */
get hasListeners(): boolean {
return this.listeners.size > 0
}
/**
* @description Subscribe to committed mutations.
* @param listener - Called once per committed mutation, in commit order.
* @returns An unsubscribe function.
*/
subscribe(listener: ChangeListener): () => void {
this.listeners.add(listener)
return () => {
this.listeners.delete(listener)
}
}
/**
* @description Enqueue committed events and schedule dispatch. Call ONLY
* after the commit has succeeded — this feed must never announce a write
* that did not become durable. Safe to call with an empty array.
* @param events - The committed mutations, in commit order.
*/
emit(events: BrainyChangeEvent[]): void {
if (events.length === 0 || this.listeners.size === 0) return
this.queue.push(...events)
if (!this.draining) {
this.draining = true
queueMicrotask(() => this.drain())
}
}
/** Deliver everything queued, FIFO, isolating listener errors. */
private drain(): void {
try {
while (this.queue.length > 0) {
const event = this.queue.shift()!
for (const listener of this.listeners) {
try {
listener(event)
} catch (err) {
// A subscriber's bug must never affect the write path or its
// sibling subscribers.
console.error('[Brainy] onChange listener threw:', err)
}
}
}
} finally {
this.draining = false
}
}
/** Drop all listeners (brain close/teardown). Queued events are discarded. */
close(): void {
this.listeners.clear()
this.queue.length = 0
}
}

8
src/index.ts
View file

@ -15,6 +15,14 @@ import { Brainy } from './brainy.js'
export { Brainy }
// The in-process change feed (brain.onChange) — event + listener types.
export type {
BrainyChangeEvent,
ChangeEventEntity,
ChangeEventRelation,
ChangeListener
} from './events/changeFeed.js'
// Export diagnostics result type
export type { DiagnosticsResult } from './brainy.js'

228
tests/integration/onchange-feed.test.ts Normal file
View file

@ -0,0 +1,228 @@
/**
* @module tests/integration/onchange-feed
* @description The `brain.onChange` in-process change feed — the authoritative
* post-commit signal for every canonical mutation, regardless of origin.
* Pins the delivery contract:
* - every operation fires exactly once per affected record with the
* post-commit payload (deletes carry the record's LAST state);
* - batch operations emit one event per item; transact items share one
* generation;
* - VFS writes (a Tier-1 blind spot for router-synthesized events) emit;
* - events arrive in commit order with monotonic generations;
* - a losing `ifRev` CAS emits NOTHING (aborted commits are invisible);
* - a throwing listener is isolated; unsubscribe stops delivery;
* - clear()/restore() emit one store-level "refetch everything" event.
*/
import { describe, it, expect, beforeEach, afterEach } from 'vitest'
import * as fs from 'node:fs'
import * as os from 'node:os'
import * as path from 'node:path'
import { Brainy } from '../../src/brainy.js'
import type { BrainyChangeEvent } from '../../src/events/changeFeed.js'
import { NounType, VerbType } from '../../src/types/graphTypes.js'
let seq = 0
const freshId = (): string =>
`00000000-0000-4000-8000-${(++seq).toString(16).padStart(12, '0')}`
/** Let the microtask-dispatched feed drain. */
const drained = (): Promise<void> => new Promise((r) => setTimeout(r, 0))
describe('brain.onChange — the in-process change feed', () => {
let dir: string
let brain: any
let events: BrainyChangeEvent[]
let off: () => void
beforeEach(async () => {
process.env.BRAINY_DETERMINISTIC_EMBEDDINGS = 'true'
dir = fs.mkdtempSync(path.join(os.tmpdir(), 'brainy-onchange-'))
brain = new Brainy({
requireSubtype: false,
storage: { type: 'filesystem', path: dir },
dimensions: 384,
silent: true
})
await brain.init()
events = []
off = brain.onChange((e: BrainyChangeEvent) => events.push(e))
})
afterEach(async () => {
off()
await brain.close()
fs.rmSync(dir, { recursive: true, force: true })
})
it('add → update → remove: one fully-described event each; remove carries the last state', async () => {
const id = await brain.add({
id: freshId(),
data: 'a document',
type: NounType.Document,
metadata: { status: 'draft' }
})
await brain.update({ id, metadata: { status: 'published' } })
await brain.remove(id)
await drained()
expect(events.map((e) => e.op)).toEqual(['add', 'update', 'remove'])
expect(events.every((e) => e.kind === 'entity' && e.id === id)).toBe(true)
expect(events[0].entity).toMatchObject({ id, type: 'document', metadata: { status: 'draft' } })
expect(events[1].entity).toMatchObject({ id, metadata: { status: 'published' } })
// The delete payload is the record's LAST committed state.
expect(events[2].entity).toMatchObject({ id, type: 'document', metadata: { status: 'published' } })
// Post-commit ordering: generations are monotonic.
const gens = events.map((e) => e.generation!)
expect([...gens].sort((a, b) => a - b)).toEqual(gens)
expect(new Set(gens).size).toBe(3)
})
it('relate → updateRelation → unrelate: relation events with endpoints + type', async () => {
const a = await brain.add({ id: freshId(), data: 'a', type: NounType.Thing })
const b = await brain.add({ id: freshId(), data: 'b', type: NounType.Thing })
events.length = 0
const rel = await brain.relate({ from: a, to: b, type: VerbType.RelatedTo, metadata: { w: 1 } })
await brain.updateRelation({ id: rel, metadata: { w: 2 } })
await brain.unrelate(rel)
await drained()
expect(events.map((e) => e.op)).toEqual(['relate', 'updateRelation', 'unrelate'])
expect(events.every((e) => e.kind === 'relation' && e.id === rel)).toBe(true)
for (const e of events) {
expect(e.relation).toMatchObject({ id: rel, from: a, to: b })
}
})
it('remove() cascades: deleting an entity emits unrelate for its relationships too', async () => {
const a = await brain.add({ id: freshId(), data: 'src', type: NounType.Thing })
const b = await brain.add({ id: freshId(), data: 'tgt', type: NounType.Thing })
const rel = await brain.relate({ from: a, to: b, type: VerbType.Contains })
events.length = 0
await brain.remove(a)
await drained()
const removeEvent = events.find((e) => e.op === 'remove')
const unrelateEvent = events.find((e) => e.op === 'unrelate')
expect(removeEvent).toMatchObject({ kind: 'entity', id: a })
expect(unrelateEvent).toMatchObject({
kind: 'relation',
id: rel,
relation: { from: a, to: b }
})
// Entity + cascaded verb share the one remove() generation.
expect(removeEvent!.generation).toBe(unrelateEvent!.generation)
})
it('batches emit one event per item; removeMany deletes carry payloads', async () => {
const ids = [freshId(), freshId(), freshId()]
await brain.addMany({
items: ids.map((id, i) => ({
id,
data: `item ${i}`,
type: NounType.Thing,
metadata: { n: i }
}))
})
await drained()
expect(events.filter((e) => e.op === 'add').length).toBe(3)
events.length = 0
await brain.removeMany({ ids })
await drained()
const removes = events.filter((e) => e.op === 'remove')
expect(removes.length).toBe(3)
// Every delete is fully described (enriched, not id-only).
for (const e of removes) {
expect(e.entity).toBeDefined()
expect(e.entity!.type).toBe('thing')
expect(typeof e.entity!.metadata.n).toBe('number')
}
})
it('transact(): per-item events sharing ONE generation; a rejected batch emits nothing', async () => {
const x = freshId()
const y = freshId()
await brain.transact([
{ op: 'add', id: x, data: 'x', type: NounType.Thing, metadata: { k: 1 } },
{ op: 'add', id: y, data: 'y', type: NounType.Thing },
{ op: 'relate', from: x, to: y, type: VerbType.RelatedTo }
])
await drained()
expect(events.map((e) => e.op)).toEqual(['add', 'add', 'relate'])
expect(new Set(events.map((e) => e.generation)).size).toBe(1)
// Rejected batch (stale per-op CAS) → zero events.
events.length = 0
await expect(
brain.transact([{ op: 'update', id: x, metadata: { k: 2 }, ifRev: 999 }])
).rejects.toMatchObject({ name: 'RevisionConflictError' })
await drained()
expect(events.length).toBe(0)
})
it('a losing ifRev CAS update emits NOTHING; the winner emits once', async () => {
const id = await brain.add({ id: freshId(), data: 'contended', type: NounType.Thing })
const rev = (await brain.get(id))._rev
events.length = 0
const results = await Promise.allSettled(
Array.from({ length: 4 }, (_, i) =>
brain.update({ id, metadata: { w: i }, merge: false, ifRev: rev })
)
)
await drained()
expect(results.filter((r) => r.status === 'fulfilled').length).toBe(1)
expect(events.filter((e) => e.op === 'update').length).toBe(1) // exactly the winner
})
it('VFS writes emit (the router-synthesis blind spot)', async () => {
await brain.vfs.writeFile('/notes/hello.txt', 'hello feed')
await drained()
// A VFS write is entities + containment relations under the hood — the
// feed sees it because VFS delegates to canonical brain methods.
expect(events.length).toBeGreaterThan(0)
expect(events.some((e) => e.kind === 'entity' && e.op === 'add')).toBe(true)
})
it('listener errors are isolated; unsubscribe stops delivery', async () => {
const good: BrainyChangeEvent[] = []
const offBad = brain.onChange(() => {
throw new Error('subscriber bug')
})
const offGood = brain.onChange((e: BrainyChangeEvent) => good.push(e))
const id = await brain.add({ id: freshId(), data: 'p', type: NounType.Thing })
await drained()
expect(good.length).toBeGreaterThan(0) // sibling unaffected by the throwing listener
offBad()
offGood()
good.length = 0
events.length = 0
off() // unsubscribe the outer listener too
await brain.update({ id, metadata: { after: true } })
await drained()
expect(events.length).toBe(0)
expect(good.length).toBe(0)
// Re-subscribe for afterEach symmetry.
off = brain.onChange((e: BrainyChangeEvent) => events.push(e))
})
it('clear() emits one store-level event meaning "refetch everything"', async () => {
await brain.add({ id: freshId(), data: 'doomed', type: NounType.Thing })
events.length = 0
await brain.clear()
await drained()
const store = events.filter((e) => e.kind === 'store')
expect(store).toHaveLength(1)
expect(store[0].op).toBe('clear')
expect(store[0].generation).toBeUndefined()
})
})