open-brainy/tests/integration/fact-log-dual-write.test.ts

/**
 * @module tests/integration/fact-log-dual-write
 * @description The generation fact log end-to-end through real commits: every
 * committed generation (single-op AND transact) appends its AFTER-IMAGE fact
 * at the commit point; removals append body-less tombstones; an aborted
 * transaction leaves no fact; facts survive reopen and continue monotonically;
 * the scan surface (brain.scanFacts) carries the frozen telemetry shape; and
 * the fact-log namespace is protected against prefix-nuking.
 */
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, ProtectedArtifactError, splitNounMetadataRecord, type CommitFact } from '../../src/index.js'

async function allFacts(brain: any): Promise<CommitFact[]> {
  const scan = brain.scanFacts()
  expect(scan).not.toBeNull()
  const facts: CommitFact[] = []
  for await (const batch of scan!.batches()) facts.push(...batch.facts)
  return facts
}

describe('fact log dual-write (memory adapter)', () => {
  let brain: any

  beforeEach(async () => {
    process.env.BRAINY_DETERMINISTIC_EMBEDDINGS = 'true'
    brain = new Brainy({ requireSubtype: false, storage: { type: 'memory' }, silent: true, dimensions: 384 })
    await brain.init()
  })
  afterEach(async () => {
    await brain.close?.().catch(() => {})
  })

  it('every single-op write appends its after-image fact; a remove appends a tombstone', async () => {
    const id = await brain.add({ data: 'first', type: 'document', metadata: { rev: 1 } })
    await brain.update({ id, metadata: { rev: 2 } })
    await brain.remove(id)

    const facts = await allFacts(brain)
    // add + update + remove each committed a generation (the remove may span
    // cascade ops but is ONE generation). Facts are monotonic.
    const gens = facts.map((f) => f.generation)
    expect([...gens].sort((a, b) => a - b)).toEqual(gens)
    expect(facts.length).toBeGreaterThanOrEqual(3)

    // The add fact carries the after-image of the new entity.
    const addFact = facts.find((f) => f.ops.some((op) => op.id === id && op.record !== null))
    expect(addFact).toBeDefined()

    // The remove fact carries a body-less tombstone for the id.
    const removeFact = facts[facts.length - 1]
    const tombstone = removeFact.ops.find((op) => op.id === id)
    expect(tombstone).toBeDefined()
    expect(tombstone!.record).toBeNull()
    expect(tombstone!.kind).toBe('noun')
  })

  it('the update fact holds the NEW state (after-image, not before)', async () => {
    const id = await brain.add({ data: 'versioned', type: 'document', metadata: { v: 'old' } })
    await brain.update({ id, metadata: { v: 'new' } })

    const facts = await allFacts(brain)
    const updateFact = facts[facts.length - 1]
    const op = updateFact.ops.find((o) => o.id === id)!
    expect(op.record).not.toBeNull()
    // The fact log is byte-faithful: op.record.metadata is the RAW stored
    // record (v2 nested-bag since the field-addressing law) — read the user
    // field through the shape-aware split, like every other reader.
    const { custom } = splitNounMetadataRecord(
      op.record!.metadata as Record<string, unknown>
    )
    expect(custom.v).toBe('new')
  })

  it('a transact commits ONE fact carrying all its ops, with meta', async () => {
    const receipt = await brain.transact(
      [
        { op: 'add', type: 'document', metadata: { part: 1 }, data: 'a' },
        { op: 'add', type: 'document', metadata: { part: 2 }, data: 'b' }
      ],
      { meta: { source: 'batch-import' } }
    )

    const facts = await allFacts(brain)
    const txFact = facts.find((f) => f.generation === receipt.generation)
    expect(txFact).toBeDefined()
    expect(txFact!.ops.filter((op) => op.kind === 'noun').length).toBeGreaterThanOrEqual(2)
    expect(txFact!.meta).toEqual({ source: 'batch-import' })
  })

  it('an aborted transact leaves NO fact (absent = never committed)', async () => {
    const id = await brain.add({ data: 'cas target', type: 'document', metadata: { n: 1 } })
    const before = (await allFacts(brain)).length

    await expect(
      brain.transact([{ op: 'update', id, ifRev: 999, metadata: { n: 2 } }])
    ).rejects.toThrow()

    const after = await allFacts(brain)
    expect(after.length).toBe(before)
  })

  it('fact generations line up with the transaction log', async () => {
    await brain.add({ data: 'x', type: 'document', metadata: {} })
    await brain.add({ data: 'y', type: 'document', metadata: {} })
    await brain.flush()

    const facts = await allFacts(brain)
    const logGens = new Set((await brain.transactionLog()).map((e: any) => e.generation))
    for (const f of facts) {
      expect(logGens.has(f.generation)).toBe(true)
    }
  })

  it('the storage fact-scan capability serves a provider holding only `storage`', async () => {
    // An index provider receives `storage` — never the brain — and reaches the
    // fact log through the host-wired capability (it must never construct its
    // own fact-log reader: the log's open path is writer-side).
    const id = await brain.add({ data: 'via storage', type: 'document', metadata: { s: 1 } })
    await brain.remove(id)

    const storage = brain.storage
    expect(typeof storage.scanFacts).toBe('function')
    expect(storage.factLogHeadGeneration()).toBe(brain.scanFacts()!.headGeneration)

    const viaStorage: CommitFact[] = []
    for await (const b of storage.scanFacts()!.batches()) viaStorage.push(...b.facts)
    const viaBrain: CommitFact[] = []
    for await (const b of brain.scanFacts()!.batches()) viaBrain.push(...b.facts)
    expect(viaStorage.map((f) => f.generation)).toEqual(viaBrain.map((f) => f.generation))
    expect(storage.factSegmentPaths()).toEqual(brain.factSegmentPaths())
  })

  it('scan telemetry carries the frozen shape end-to-end', async () => {
    for (let i = 0; i < 5; i++) await brain.add({ data: `t${i}`, type: 'document', metadata: { i } })

    const scan = brain.scanFacts({ batchSize: 2 })!
    expect(scan.headGeneration).toBeGreaterThanOrEqual(5)
    expect(scan.approxFactCount).toBeGreaterThanOrEqual(5)
    let batches = 0
    for await (const b of scan.batches()) {
      batches++
      expect(b.factCount).toBe(b.facts.length)
      expect(b.firstGeneration).toBe(b.facts[0].generation)
      expect(b.lastGeneration).toBe(b.facts[b.facts.length - 1].generation)
      expect(b.byteSize).toBeGreaterThan(0)
      expect(typeof b.segmentId).toBe('string')
    }
    expect(batches).toBeGreaterThan(1)
    expect(scan.summary().factsYielded).toBe(scan.approxFactCount)
  })
})

describe('fact log dual-write (filesystem adapter — durability + protection)', () => {
  let dir: string
  let brain: any

  const open = async () => {
    const b: any = new Brainy({
      requireSubtype: false,
      storage: { type: 'filesystem', path: dir },
      silent: true,
      dimensions: 384
    })
    await b.init()
    return b
  }

  beforeEach(async () => {
    process.env.BRAINY_DETERMINISTIC_EMBEDDINGS = 'true'
    dir = fs.mkdtempSync(path.join(os.tmpdir(), 'brainy-factlog-'))
    brain = await open()
  })
  afterEach(async () => {
    await brain.close?.().catch(() => {})
    fs.rmSync(dir, { recursive: true, force: true })
  })

  it('facts survive close + reopen and appends continue monotonically', async () => {
    const id = await brain.add({ data: 'persist me', type: 'document', metadata: { k: 1 } })
    await brain.remove(id)
    await brain.close()

    brain = await open()
    const facts = await allFacts(brain)
    expect(facts.length).toBeGreaterThanOrEqual(2)
    const headBefore = facts[facts.length - 1].generation

    await brain.add({ data: 'after reopen', type: 'document', metadata: { k: 2 } })
    const facts2 = await allFacts(brain)
    expect(facts2[facts2.length - 1].generation).toBeGreaterThan(headBefore)
  })

  it('the fact segments exist on disk under _generations/facts/ with zero-padded names', async () => {
    await brain.add({ data: 'on disk', type: 'document', metadata: {} })
    await brain.flush()
    const factsDir = path.join(dir, '_generations', 'facts')
    const files = fs.readdirSync(factsDir)
    // The manifest rides the store's JSON object discipline (gzip on disk).
    expect(files.some((f) => f.startsWith('manifest.json'))).toBe(true)
    const segs = files.filter((f) => /^seg-\d{20}\.bfl$/.test(f))
    expect(segs.length).toBeGreaterThanOrEqual(1)
  })

  it('the fact-log namespace is PROTECTED: a prefix-nuke is refused', async () => {
    await brain.add({ data: 'protected', type: 'document', metadata: {} })
    await expect(brain.storage.removeRawPrefix('_generations/facts')).rejects.toBeInstanceOf(
      ProtectedArtifactError
    )
    // Per-generation history cleanup remains unaffected (no false intersect).
    await expect(brain.storage.removeRawPrefix('_generations/999999')).resolves.toBeUndefined()
  })

  it('transact facts are durable-on-return (no flush needed before reopen)', async () => {
    const receipt = await brain.transact([
      { op: 'add', type: 'document', metadata: { durable: true }, data: 'tx' }
    ])
    // Simulate an abrupt end: no flush(), no close() — reopen from disk.
    brain = await open()
    const facts = await allFacts(brain)
    expect(facts.some((f) => f.generation === receipt.generation)).toBe(true)
  })
})