open-brainy/tests/integration/fact-log-v2-cutover.test.ts

/**
 * @module tests/integration/fact-log-v2-cutover
 * @description The fact log's LIVE WRITE FORMAT cutover to v2, end-to-end
 * through real brains: (a) a NEW brain's tail segment carries a v2 header
 * (formatVersion 2, sealSize 4096), opens with the log.genesis record
 * (id-space width 64 + the manifest-persisted brainId), and scanFacts yields
 * the same CommitFact shape a v1 brain would — reconstruction included,
 * proven by digest-equality against canonical after a reopen; (b) MIXED
 * logs: an existing v1 segment stays readable forever beside a v2 tail
 * (cutover-by-rotation; the v1 segment is never rewritten); (c) MINT:
 * after-image records carry the metadata index id mapper's exact int
 * assignments (white-box compare); (d) SEALS: every flush leaves the tail
 * sector-aligned, and pads are invisible to scans; (e) REPLAY: the
 * log-authority recovery path resurrects an acked write from a v2 tail
 * after a crash-style abandon.
 */
import { describe, it, expect, afterEach } from 'vitest'
import * as fs from 'node:fs'
import * as path from 'node:path'
import { mkdtempSync, rmSync } from 'node:fs'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { Brainy } from '../../src/brainy.js'
import { NounType } from '../../src/types/graphTypes.js'
import {
  parseSegmentHeader,
  decodeGroupV2,
  SEGMENT_HEADER_BYTES,
  FACT_LOG_FORMAT_V1,
  FACT_LOG_FORMAT_V2,
  type LogGenesisRecord,
  type NounAfterImageRecord
} from '../../src/db/factLogFormat.js'
import type { CommitFact, FactIntMinter, FactLog } from '../../src/db/factLog.js'
import {
  makeTempDir,
  openBrain,
  storeOf,
  abandonAsCrashed,
  factGenerations,
  vec,
  uid
} from '../helpers/durabilityKillMatrix.js'

/** The VFS root — created at init by a baseline (generation-less) write. */
const VFS_ROOT = '00000000-0000-0000-0000-000000000000'
const FACTS_DIR = ['_generations', 'facts'] as const
const MANIFEST_PATH = '_generations/facts/manifest.json'

/** White-box internals this suite instruments. */
type BrainInternals = {
  storage: {
    readRawObject(p: string): Promise<unknown | null>
    readNounRaw(id: string): Promise<{ metadata: unknown | null; vector: unknown | null }>
  }
  metadataIndex: {
    getIdMapper(): { getInt(uuid: string): number | undefined }
  }
}
const internals = (brain: Brainy): BrainInternals => brain as unknown as BrainInternals

/** The facts manifest as stored (additive brainId included). */
interface StoredFactsManifest {
  segments: Array<{ file: string }>
  tailSegment: string | null
  brainId?: string
}

async function readManifest(brain: Brainy): Promise<StoredFactsManifest> {
  const manifest = (await internals(brain).storage.readRawObject(
    MANIFEST_PATH
  )) as StoredFactsManifest | null
  expect(manifest, 'the facts manifest exists').toBeTruthy()
  return manifest!
}

/** Raw on-disk bytes of one fact segment file. */
function segmentBytes(dir: string, file: string): Uint8Array {
  return new Uint8Array(fs.readFileSync(path.join(dir, ...FACTS_DIR, file)))
}

async function allFacts(brain: Brainy): Promise<CommitFact[]> {
  const scan = (brain as unknown as { scanFacts(): { batches(): AsyncGenerator<{ facts: CommitFact[] }> } | null }).scanFacts()
  expect(scan, 'this storage hosts a fact log').not.toBeNull()
  const facts: CommitFact[] = []
  for await (const batch of scan!.batches()) facts.push(...batch.facts)
  return facts
}

/** The live FactLog instance (white-box: the minter strip in scenario b). */
function factLogOf(brain: Brainy): FactLog & { intMinter: FactIntMinter | null } {
  const log = storeOf(brain).getFactLog()
  expect(log, 'filesystem storage hosts a fact log').not.toBeNull()
  return log as FactLog & { intMinter: FactIntMinter | null }
}

describe('fact log v2 cutover — live writes land in the v2 segment format', () => {
  const dirs: string[] = []
  const brains: Brainy[] = []

  const trackDir = (): string => {
    const dir = makeTempDir()
    dirs.push(dir)
    return dir
  }
  const track = (brain: Brainy): Brainy => {
    brains.push(brain)
    return brain
  }

  afterEach(async () => {
    for (const b of brains.splice(0)) {
      await (b as unknown as { close?: () => Promise<void> }).close?.().catch(() => {})
    }
    for (const d of dirs.splice(0)) rmSync(d, { recursive: true, force: true })
  })

  it('(a) NEW BRAIN: v2 tail header, genesis-first, and scanFacts parity with canonical across a reopen', async () => {
    const dir = trackDir()
    const brain = track(await openBrain(dir))
    const idA = uid('v2-new-a')
    const idB = uid('v2-new-b')
    await brain.add({ id: idA, data: 'alpha', type: NounType.Document, vector: vec(1), metadata: { n: 1 } })
    await brain.add({ id: idB, data: 'beta', type: NounType.Document, vector: vec(2), metadata: { n: 2 } })
    await brain.flush()

    // The tail segment's raw header bytes: formatVersion 2, sealSize 4096.
    const manifest = await readManifest(brain)
    expect(manifest.tailSegment).toBeTruthy()
    expect(manifest.brainId, 'the brain id was minted into the manifest').toBeTruthy()
    const bytes = segmentBytes(dir, manifest.tailSegment!)
    const header = parseSegmentHeader(bytes.subarray(0, SEGMENT_HEADER_BYTES))
    expect(header.formatVersion).toBe(FACT_LOG_FORMAT_V2)
    expect(header.sealSize).toBe(4096)

    // Genesis is the FIRST record of the FIRST fact — and appears exactly once.
    const group = decodeGroupV2(bytes.subarray(SEGMENT_HEADER_BYTES), { expectedIdSpaceWidth: 64 })
    expect(group.facts.length).toBeGreaterThanOrEqual(2)
    const firstRecord = group.facts[0].records[0]
    expect(firstRecord.type).toBe('log.genesis')
    const genesis = firstRecord as LogGenesisRecord
    expect(genesis.idSpaceWidth).toBe(64)
    expect(genesis.brainId).toBe(manifest.brainId)
    const genesisCount = group.facts
      .flatMap((f) => f.records)
      .filter((r) => r.type === 'log.genesis').length
    expect(genesisCount).toBe(1)

    // Shape parity + reconstruction fidelity: REOPEN (so the tail decodes
    // from disk, not from the in-session originals) and compare each add's
    // CommitFact op against canonical byte truth — metadata leg (bigint
    // timestamps normalized back to numbers) AND the reconstructed vector
    // wrapper must equal what readNounRaw returns, exactly as a v1 log's
    // byte-faithful capture would.
    await (brain as unknown as { close: () => Promise<void> }).close()
    brains.splice(brains.indexOf(brain), 1)
    const reopened = track(await openBrain(dir))
    const facts = await allFacts(reopened)
    const gens = facts.map((f) => f.generation)
    expect([...gens].sort((a, b) => a - b)).toEqual(gens)
    expect(new Set(gens).size).toBe(gens.length)

    const logGens = new Set(
      ((await (reopened as unknown as { transactionLog(): Promise<Array<{ generation: number }>> }).transactionLog()) ?? []).map(
        (e) => e.generation
      )
    )
    for (const g of gens) expect(logGens.has(g), `generation ${g} is a real commit`).toBe(true)

    for (const id of [idA, idB]) {
      const fact = facts.find((f) => f.ops.some((op) => op.id === id && op.record !== null))
      expect(fact, `the add fact for ${id} survives the reopen`).toBeDefined()
      const op = fact!.ops.find((o) => o.id === id)!
      expect(op.kind).toBe('noun')
      const canonical = await internals(reopened).storage.readNounRaw(id)
      expect(op.record!.metadata).toStrictEqual(canonical.metadata)
      // ENTITY TRUTH comparison: canonical wrappers denormalize HNSW residue
      // (connections + the randomly-assigned level) that the log record
      // deliberately reconstructs empty — strip both sides (the oracle's
      // normalizer law) so a nonzero random level can't fake a divergence.
      const strip = (w: unknown) => {
        const { connections: _c, level: _l, ...rest } = w as Record<string, unknown>
        return rest
      }
      expect(strip(op.record!.vector)).toStrictEqual(strip(canonical.vector))
    }
  })

  it('(b) MIXED LOG: an existing v1 segment stays readable forever beside the v2 tail (cutover by rotation, v1 bytes untouched)', async () => {
    // ROUTE: a REAL v1 segment is written by the v1 writer itself — the live
    // FactLog with its minter stripped (the exact pre-cutover code path,
    // still shipped for minter-less configurations) — then the minter is
    // restored mid-session and the next append performs the cutover
    // rotation. Stronger than hand-crafted bytes: both formats come from
    // their real writers, on one log.
    const dir = trackDir()
    const brain = track(await openBrain(dir))
    const log = factLogOf(brain)
    const minter = log.intMinter
    expect(minter, 'the brain wired the int minter at init').toBeTruthy()

    log.intMinter = null // the pre-cutover writer
    const idOld1 = uid('v1-old-1')
    const idOld2 = uid('v1-old-2')
    await brain.add({ id: idOld1, data: 'old one', type: NounType.Document, vector: vec(3), metadata: { era: 'v1' } })
    await brain.add({ id: idOld2, data: 'old two', type: NounType.Document, vector: vec(4), metadata: { era: 'v1' } })
    await brain.flush()

    const before = await readManifest(brain)
    expect(before.segments).toHaveLength(0)
    const v1TailFile = before.tailSegment!
    const v1Bytes = segmentBytes(dir, v1TailFile)
    expect(parseSegmentHeader(v1Bytes.subarray(0, SEGMENT_HEADER_BYTES)).formatVersion).toBe(
      FACT_LOG_FORMAT_V1
    )

    log.intMinter = minter // the cutover lands mid-session
    const idNew = uid('v2-new')
    await brain.add({ id: idNew, data: 'new era', type: NounType.Document, vector: vec(5), metadata: { era: 'v2' } })
    await brain.flush()

    // The v1 tail was SEALED (bytes untouched), the new tail is v2.
    const after = await readManifest(brain)
    expect(after.segments.map((s) => s.file)).toContain(v1TailFile)
    expect(after.tailSegment).not.toBe(v1TailFile)
    const sealedBytes = segmentBytes(dir, v1TailFile)
    expect(parseSegmentHeader(sealedBytes.subarray(0, SEGMENT_HEADER_BYTES)).formatVersion).toBe(
      FACT_LOG_FORMAT_V1
    )
    expect(
      Buffer.compare(Buffer.from(sealedBytes), Buffer.from(v1Bytes)),
      'the sealed v1 segment is byte-identical — never rewritten'
    ).toBe(0)
    const tailBytes = segmentBytes(dir, after.tailSegment!)
    expect(parseSegmentHeader(tailBytes.subarray(0, SEGMENT_HEADER_BYTES)).formatVersion).toBe(
      FACT_LOG_FORMAT_V2
    )
    // NOT a brand-new log: no genesis on a rotated-in v2 tail.
    const tailGroup = decodeGroupV2(tailBytes.subarray(SEGMENT_HEADER_BYTES), {
      expectedIdSpaceWidth: 64
    })
    expect(
      tailGroup.facts.flatMap((f) => f.records).some((r) => r.type === 'log.genesis')
    ).toBe(false)

    // One scan spans both formats, shape-identically, in generation order.
    const liveFacts = await allFacts(brain)
    const liveGens = liveFacts.map((f) => f.generation)
    expect([...liveGens].sort((a, b) => a - b)).toEqual(liveGens)
    for (const id of [idOld1, idOld2, idNew]) {
      const fact = liveFacts.find((f) => f.ops.some((op) => op.id === id))
      expect(fact, `fact for ${id} is scannable`).toBeDefined()
      const op = fact!.ops.find((o) => o.id === id)!
      expect(op.kind).toBe('noun')
      expect(op.record).not.toBeNull()
    }

    // The MIXED log survives a reopen and keeps appending (v2 tail).
    await (brain as unknown as { close: () => Promise<void> }).close()
    brains.splice(brains.indexOf(brain), 1)
    const reopened = track(await openBrain(dir))
    const reFacts = await allFacts(reopened)
    expect(reFacts.map((f) => f.generation)).toEqual(liveGens)
    // The v1 fact still reads exactly as the v1 decoder always read it.
    // (Not compared byte-strict against canonical: the v1 CAPTURE has a
    // known pre-existing wart — write-cache-warm objects carry
    // undefined-valued engine keys that msgpack preserves as nil while the
    // durable JSON drops them. v1 bytes are frozen; the v2 encoder
    // sanitizes to durable truth instead — pinned in scenario (a).)
    const oldOp = reFacts
      .find((f) => f.ops.some((op) => op.id === idOld1))!
      .ops.find((o) => o.id === idOld1)!
    const canonicalOld = await internals(reopened).storage.readNounRaw(idOld1)
    const oldMeta = oldOp.record!.metadata as Record<string, unknown>
    expect(oldMeta.noun).toBe('document')
    expect((oldMeta.metadata as Record<string, unknown>).era).toBe('v1')
    const oldWrapper = oldOp.record!.vector as { id: string; vector: number[] }
    const canonicalWrapper = canonicalOld.vector as { id: string; vector: number[] }
    expect(oldWrapper.id).toBe(idOld1)
    expect(oldWrapper.vector).toStrictEqual(canonicalWrapper.vector)
    await reopened.add({ id: uid('post-reopen'), data: 'still writing', type: NounType.Document, vector: vec(6), metadata: {} })
    expect((await factGenerations(reopened)).length).toBe(liveGens.length + 1)
  })

  it('(c) MINT-AT-APPEND: after-image records carry the id mapper\'s EXACT int assignments — distinct, nonzero, reproducible', async () => {
    const dir = trackDir()
    const brain = track(await openBrain(dir))
    const idA = uid('mint-a')
    const idB = uid('mint-b')
    await brain.add({ id: idA, data: 'mint one', type: NounType.Document, vector: vec(7), metadata: { m: 1 } })
    await brain.add({ id: idB, data: 'mint two', type: NounType.Document, vector: vec(8), metadata: { m: 2 } })
    await brain.flush()

    const manifest = await readManifest(brain)
    const bytes = segmentBytes(dir, manifest.tailSegment!)
    const group = decodeGroupV2(bytes.subarray(SEGMENT_HEADER_BYTES), { expectedIdSpaceWidth: 64 })
    const afterImages = new Map<string, NounAfterImageRecord>()
    for (const fact of group.facts) {
      for (const record of fact.records) {
        if (record.type === 'noun.afterImage') afterImages.set(record.id, record)
      }
    }
    const recA = afterImages.get(idA)
    const recB = afterImages.get(idB)
    expect(recA, 'idA has a decoded after-image').toBeDefined()
    expect(recB, 'idB has a decoded after-image').toBeDefined()
    expect(recA!.entityInt).toBeGreaterThan(0n)
    expect(recB!.entityInt).toBeGreaterThan(0n)
    expect(recA!.entityInt).not.toBe(recB!.entityInt)

    // White-box: the ints on the wire ARE the metadata index mapper's
    // assignments — the exact ints a mapper rebuild must reproduce.
    const mapper = internals(brain).metadataIndex.getIdMapper()
    expect(recA!.entityInt).toBe(BigInt(mapper.getInt(idA)!))
    expect(recB!.entityInt).toBe(BigInt(mapper.getInt(idB)!))
  })

  it('(d) SEALS AT SYNC: every flush leaves the tail sector-aligned; pads are invisible to scans', async () => {
    const dir = trackDir()
    const brain = track(await openBrain(dir))
    await brain.add({ id: uid('seal-1'), data: 'one', type: NounType.Document, vector: vec(10), metadata: {} })
    await brain.flush()

    const manifest = await readManifest(brain)
    const tailPath = path.join(dir, ...FACTS_DIR, manifest.tailSegment!)
    const sizeAfterFirstFlush = fs.statSync(tailPath).size
    expect(sizeAfterFirstFlush).toBeGreaterThan(0)
    expect(sizeAfterFirstFlush % 4096, 'tail is sector-aligned after flush').toBe(0)
    const countAfterFirstFlush = (await factGenerations(brain)).length

    for (let i = 0; i < 3; i++) {
      await brain.add({ id: uid(`seal-more-${i}`), data: `more ${i}`, type: NounType.Document, vector: vec(11 + i), metadata: { i } })
    }
    await brain.flush()
    const sizeAfterSecondFlush = fs.statSync(tailPath).size
    expect(sizeAfterSecondFlush).toBeGreaterThan(sizeAfterFirstFlush)
    expect(sizeAfterSecondFlush % 4096, 'still aligned after more writes + flush').toBe(0)

    // Pads count toward bytes, never toward facts.
    expect((await factGenerations(brain)).length).toBe(countAfterFirstFlush + 3)
  })

  it('(e) REPLAY COMPAT: the log-authority recovery path resurrects an acked write from a v2 tail after a crash-style abandon', async () => {
    // The flip idiom from the log-authority suite: seed writes, baseline
    // backfill LAST (the init-time VFS root never got a fact), flush, then
    // the sanctioned guarded flip — the oracle goes green over an ALL-V2
    // log, which is itself the reproduction proof for the v2 record path.
    const dir = mkdtempSync(join(tmpdir(), 'brainy-v2-cutover-'))
    dirs.push(dir)
    process.env.BRAINY_DETERMINISTIC_EMBEDDINGS = 'true'
    const open = async (): Promise<Brainy> => {
      const b = new Brainy({
        storage: { type: 'filesystem', path: dir },
        requireSubtype: false,
        silent: true,
        dimensions: 384
      })
      await b.init()
      return track(b)
    }

    const brain = await open()
    const kept = await brain.add({ data: 'alpha document', type: 'document', metadata: { n: 1 } })
    const removed = await brain.add({ data: 'beta document', type: 'document', metadata: { n: 2 } })
    await brain.update({ id: kept, metadata: { n: 10 } })
    await brain.remove(removed)
    const root = await brain.get(VFS_ROOT)
    expect(root, 'the VFS root exists').toBeTruthy()
    await brain.update({ id: VFS_ROOT, metadata: root!.metadata }) // baseline backfill — final write
    await brain.flush()

    const report = await (brain as unknown as { adoptLogAuthority(): Promise<{ verdict: string }> }).adoptLogAuthority()
    expect(report.verdict, 'the oracle is green over a pure-v2 log').toBe('green')

    // An at-ack write: its v2 fact is fsynced (sector-sealed) at ack.
    const survivor = await brain.add({
      data: 'survives power loss',
      type: 'document',
      metadata: { s: 1 }
    })

    // Crash-style abandon: RAM state gone, no flush, no close.
    await abandonAsCrashed(brain)

    // Reopen: open() finds the acked fact ABOVE the manifest watermark in
    // the v2 tail (peekFactsAbove → v2 decode) and REPLAYS it into
    // canonical — an acked write is never lost.
    const reopened = await open()
    expect(
      (reopened as unknown as { logAuthority(): { authority: string } }).logAuthority().authority
    ).toBe('log')
    const resurrected = await reopened.get(survivor)
    expect(resurrected, 'the acked write survived the crash').toBeTruthy()
    expect((resurrected as { metadata?: { s?: number } }).metadata?.s).toBe(1)
    expect((await factGenerations(reopened)).length).toBeGreaterThan(0)
  })
})