brainy/tests/integration/onchange-feed.test.ts

/**
 * @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()
  })
})