From fd5edb5a13e0dd436e6965cca7ad4ddae8af561b Mon Sep 17 00:00:00 2001 From: David Snelling Date: Fri, 10 Jul 2026 11:24:31 -0700 Subject: [PATCH] =?UTF-8?q?feat:=20brain.onChange=20=E2=80=94=20the=20in-p?= =?UTF-8?q?rocess=20change=20feed=20for=20every=20committed=20mutation?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- docs/guides/reacting-to-changes.md | 114 ++++++ src/brainy.ts | 469 +++++++++++++++++++++++- src/events/changeFeed.ts | 171 +++++++++ src/index.ts | 8 + tests/integration/onchange-feed.test.ts | 228 ++++++++++++ 5 files changed, 975 insertions(+), 15 deletions(-) create mode 100644 docs/guides/reacting-to-changes.md create mode 100644 src/events/changeFeed.ts create mode 100644 tests/integration/onchange-feed.test.ts diff --git a/docs/guides/reacting-to-changes.md b/docs/guides/reacting-to-changes.md new file mode 100644 index 00000000..99436ff9 --- /dev/null +++ b/docs/guides/reacting-to-changes.md @@ -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; service?: string } + relation?: { id: string; from: string; to: string; type: string; + metadata?: Record } + 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()`. diff --git a/src/brainy.ts b/src/brainy.ts index 82f95e81..ee91bfed 100644 --- a/src/brainy.ts +++ b/src/brainy.ts @@ -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 + /** + * 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 implements BrainyInterface { * 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 implements BrainyInterface { private async persistSingleOp( touched: { nouns?: string[]; verbs?: string[] }, run: TransactionFunction, - precommit?: (before: CommitBeforeImages) => void - ): Promise { + 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 implements BrainyInterface { // 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() for (const id of touched.nouns ?? []) { const prev = await this.storage.readNounRaw(id) @@ -1573,18 +1608,88 @@ export class Brainy implements BrainyInterface { 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 + | 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 + ): NonNullable { + 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 implements BrainyInterface { // 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) ?? {}, + ...(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 implements BrainyInterface { 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) ?? {}, + ...(entityForIndexing.service !== undefined && { + service: String(entityForIndexing.service) + }) + } + } + ] + : undefined) // Aggregation hook (outside transaction — derived data) if (this._aggregationIndex) { @@ -2876,7 +3018,35 @@ export class Brainy implements BrainyInterface { 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) + }) + }, + ...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 implements BrainyInterface { ) ) } - }) + }, + 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 + }) + } + }, + ...(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 + }) + } + } as PendingChangeEvent + ] + : []) + ] + : undefined) return id } @@ -3643,7 +3850,26 @@ export class Brainy implements BrainyInterface { 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 + }) + } + } + ] + : undefined) } /** @@ -3778,7 +4004,26 @@ export class Brainy implements BrainyInterface { ) ) } - }) + }, + 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 + }) + } + } + ] + : undefined) } /** @@ -6335,6 +6580,16 @@ export class Brainy implements BrainyInterface { } 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() + // 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 implements BrainyInterface { ) } + if (chunkEvents) { + chunkEvents.push({ + kind: 'entity', + op: 'remove', + id, + ...(metadata && { + entity: this.entityViewFromRawRecord( + id, + metadata as Record + ) + }) + }) + 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 implements BrainyInterface { } } } - }) + }, undefined, chunkEvents) // Transaction committed — queued IDs were actually deleted result.successful.push(...chunkQueued) @@ -6690,6 +6974,12 @@ export class Brainy implements BrainyInterface { // 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 implements BrainyInterface { 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 implements BrainyInterface { 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 implements BrainyInterface { touchedVerbs: [], postCommit: [], casUpdates: [], - createdNouns: new Set() + createdNouns: new Set(), + changeEvents: [] } for (const op of ops) { @@ -8445,6 +8746,20 @@ export class Brainy implements BrainyInterface { 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) ?? {}, + ...(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 implements BrainyInterface { 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) ?? {}, + ...(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 implements BrainyInterface { 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) + }) + }) + } if (metadata) { // Canonical reserved/custom split — mirror of remove()'s aggregation hook. @@ -8813,6 +9169,20 @@ export class Brainy implements BrainyInterface { 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 }) + } + }) + } if (params.bidirectional) { const reverseId = uuidv4() @@ -8841,6 +9211,20 @@ export class Brainy implements BrainyInterface { 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 }) + } + }) + } } return id @@ -8869,6 +9253,21 @@ export class Brainy implements BrainyInterface { } 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 implements BrainyInterface { 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 " + * 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 implements BrainyInterface { * This ensures deferred persistence mode data is saved */ async close(): Promise { + // 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. diff --git a/src/events/changeFeed.ts b/src/events/changeFeed.ts new file mode 100644 index 00000000..9b9fda6f --- /dev/null +++ b/src/events/changeFeed.ts @@ -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 + /** 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 +} + +/** + * 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 + +/** + * @description Listener registry + commit-ordered async dispatcher for + * {@link BrainyChangeEvent}s. One instance per {@link Brainy}. + */ +export class ChangeFeed { + private listeners = new Set() + 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 + } +} diff --git a/src/index.ts b/src/index.ts index d9acb524..ed19fdb1 100644 --- a/src/index.ts +++ b/src/index.ts @@ -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' diff --git a/tests/integration/onchange-feed.test.ts b/tests/integration/onchange-feed.test.ts new file mode 100644 index 00000000..8470b59c --- /dev/null +++ b/tests/integration/onchange-feed.test.ts @@ -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 => 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() + }) +})