brainy/tests/integration/history-repacking.test.ts

/**
 * @module tests/integration/history-repacking
 * @description The D1+D3 two-tier history lifecycle end-to-end on a real
 * brain. Laws: (1) repacking is RE-REPRESENTATION — after folding, every
 * asOf() read below the fold boundary answers exactly as before, across a
 * cold reopen; (2) folded per-generation directories are physically gone
 * (the file-count cure is real, not cosmetic); (3) repack + reclaim compose:
 * bounded retention after repacking drops whole segments and asOf below the
 * horizon throws GenerationCompactedError; (4) repackHistory is explicit
 * API and time-bounded (spent budget = consistent no-op).
 *
 * Uses a tiny REPACK_LIVE_WINDOW override so a small history has a cold
 * tier at all (the production window is 1024).
 */
import { describe, it, expect, afterEach } from 'vitest'
import * as fs from 'node:fs'
import * as path from 'node:path'
import * as os from 'node:os'
import { Brainy } from '../../src/brainy.js'
import { NounType } from '../../src/types/graphTypes.js'
import { GenerationStore } from '../../src/db/generationStore.js'
import { GenerationCompactedError } from '../../src/db/errors.js'
import { SEGMENTS_PREFIX } from '../../src/db/generationSegments.js'

const stub = async (text: string): Promise<number[]> => {
  const h = text.split('').reduce((a, c) => a + c.charCodeAt(0), 0)
  return new Array(384).fill(0).map((_, i) => Math.sin(h + i))
}

const openBrain = async (dir: string): Promise<Brainy> => {
  const brain = new Brainy({
    requireSubtype: false,
    storage: { type: 'filesystem', path: dir },
    embeddingFunction: stub
  })
  await brain.init()
  return brain
}

describe('history repacking — the two-tier lifecycle', () => {
  const dirs: string[] = []
  const tempDir = (): string => {
    const d = fs.mkdtempSync(path.join(os.tmpdir(), 'brainy-repack-'))
    dirs.push(d)
    return d
  }
  const originalWindow = GenerationStore.REPACK_LIVE_WINDOW

  afterEach(() => {
    ;(GenerationStore as any).REPACK_LIVE_WINDOW = originalWindow
    for (const d of dirs.splice(0)) {
      try {
        fs.rmSync(d, { recursive: true, force: true })
      } catch {
        /* best effort */
      }
    }
  })

  it('repack preserves every historical read across cold reopen; folded dirs are gone', async () => {
    ;(GenerationStore as any).REPACK_LIVE_WINDOW = 3
    const dir = tempDir()
    const brain = await openBrain(dir)

    const id = await brain.add({
      data: 'versioned-entity',
      type: NounType.Document,
      metadata: { v: 0 }
    })
    for (let v = 1; v <= 10; v++) await brain.update({ id, metadata: { v } })
    await brain.flush()

    // Ground truth BEFORE repacking: capture asOf views for early generations.
    const before: Record<number, number> = {}
    for (const g of [2, 4, 6]) {
      const db = await brain.asOf(g)
      before[g] = (await db.get(id))?.metadata?.v as number
      await db.release()
    }

    const result = await brain.repackHistory()
    expect(result.foldedGenerations).toBeGreaterThan(0)
    expect(result.segmentsCreated).toBeGreaterThan(0)

    // The folded per-generation directories are PHYSICALLY gone…
    const genDirs = fs
      .readdirSync(path.join(dir, '_generations'), { withFileTypes: true })
      .filter((e) => e.isDirectory() && /^\d+$/.test(e.name)).length
    expect(genDirs).toBeLessThanOrEqual(4) // live window (3) + at most the newest
    // …and the segment tier exists (the filesystem adapter stores objects
    // gzipped, so the manifest may live at either spelling).
    const segDir = path.join(dir, SEGMENTS_PREFIX)
    expect(
      fs.existsSync(path.join(segDir, 'manifest.json')) ||
        fs.existsSync(path.join(segDir, 'manifest.json.gz'))
    ).toBe(true)
    expect(fs.readdirSync(segDir).some((f) => f.endsWith('.bgs'))).toBe(true)

    // Same asOf answers from the packed tier, same process…
    for (const g of [2, 4, 6]) {
      const db = await brain.asOf(g)
      expect((await db.get(id))?.metadata?.v).toBe(before[g])
      await db.release()
    }
    await brain.close()

    // …and across a COLD REOPEN (manifest discovery, no live dirs to list).
    const reopened = await openBrain(dir)
    for (const g of [2, 4, 6]) {
      const db = await reopened.asOf(g)
      expect((await db.get(id))?.metadata?.v).toBe(before[g])
      await db.release()
    }
    expect((await reopened.get(id))?.metadata?.v).toBe(10) // live state untouched
    await reopened.close()
  })

  it('repack + bounded reclaim compose: whole segments drop, horizon is loud', async () => {
    ;(GenerationStore as any).REPACK_LIVE_WINDOW = 2
    const dir = tempDir()
    const brain = await openBrain(dir)
    const id = await brain.add({ data: 'reclaim-probe', type: NounType.Document, metadata: { v: 0 } })
    for (let v = 1; v <= 8; v++) await brain.update({ id, metadata: { v } })
    await brain.flush()
    await brain.repackHistory()

    // Reclaim down to the 3 newest generations — packed segments below the
    // horizon drop whole; asOf below throws loudly.
    const res = await brain.compactHistory({ maxGenerations: 3 })
    expect(res.removedGenerations).toBeGreaterThan(0)
    await expect(brain.asOf(1)).rejects.toBeInstanceOf(GenerationCompactedError)
    expect((await brain.get(id))?.metadata?.v).toBe(8)
    await brain.close()
  })

  it('generationDigest: reopen-stable, divergence-sensitive, loud below the horizon', async () => {
    ;(GenerationStore as any).REPACK_LIVE_WINDOW = 2
    const dir = tempDir()
    const brain = await openBrain(dir)
    const id = await brain.add({ data: 'digest-probe', type: NounType.Document, metadata: { v: 0 } })
    for (let v = 1; v <= 6; v++) await brain.update({ id, metadata: { v } })
    await brain.flush()
    await brain.repackHistory()

    const gen = brain.generation()
    const atHead = await brain.generationDigest(gen)
    const atMid = await brain.generationDigest(3)
    expect(atHead).toMatch(/^[0-9a-f]{8}$/)
    expect(atMid).not.toBe(atHead) // more history ⇒ different digest
    await brain.close()

    // Reopen-stable: same history, same digests (packed prefix stability).
    const reopened = await openBrain(dir)
    expect(await reopened.generationDigest(gen)).toBe(atHead)
    expect(await reopened.generationDigest(3)).toBe(atMid)

    // New history diverges the head digest.
    await reopened.update({ id, metadata: { v: 7 } })
    await reopened.flush()
    expect(await reopened.generationDigest(reopened.generation())).not.toBe(atHead)

    // Below the horizon: LOUD, never a silent pin of reclaimed history.
    await reopened.compactHistory({ maxGenerations: 2 })
    await expect(reopened.generationDigest(1)).rejects.toBeInstanceOf(GenerationCompactedError)
    await reopened.close()
  })

  it('a spent time budget is a consistent no-op; the next pass resumes', async () => {
    ;(GenerationStore as any).REPACK_LIVE_WINDOW = 2
    const dir = tempDir()
    const brain = await openBrain(dir)
    const id = await brain.add({ data: 'budget-probe', type: NounType.Document, metadata: { v: 0 } })
    for (let v = 1; v <= 6; v++) await brain.update({ id, metadata: { v } })
    await brain.flush()

    const bounded = await brain.repackHistory({ timeBudgetMs: 0 })
    expect(bounded).toEqual({ foldedGenerations: 0, segmentsCreated: 0 })

    const resumed = await brain.repackHistory()
    expect(resumed.foldedGenerations).toBeGreaterThan(0)
    const db = await brain.asOf(3)
    expect((await db.get(id))?.metadata?.v).toBeDefined()
    await db.release()
    await brain.close()
  })
})