/**
 * @module tests/integration/factlog-open-prune
 * @description THE OPEN READS THE TAIL, NOT THE HISTORY.
 *
 * Every log-authority open asks the fact log one question — "is there a fact
 * above the committed pointer?" — and until this lane existed it answered by
 * reading and CRC-decoding EVERY segment file the manifest names. MEASURED in
 * production on a 16k-row brain at generation ~478,819: 34-37 seconds inside
 * the `generation-store-open-fold` phase, on every open, including the clean
 * one where the answer is always "nothing".
 *
 * The manifest already records each sealed segment's `lastGeneration`, written
 * at seal time AFTER the segment's bytes are fsynced and into a manifest that
 * is itself written atomically and fsynced — and a sealed file is never
 * appended to again (the same manifest flip re-points `tailSegment`). So an
 * entry recording `lastGeneration ≤ committed` PROVES its file holds nothing
 * above the bound, and the open can skip it whole.
 *
 * Pinned here, from the log's own counters (the narration line), never a clock:
 *
 *   1. A clean close and reopen on a log with ≥4 sealed segments reads
 *      EXACTLY the tail (1 of 6), prunes the rest, and finds nothing.
 *   2. A real SIGKILLed process that sealed segments holding facts ABOVE the
 *      committed pointer: the reopen READS those sealed segments and recovers
 *      byte-identically to an unpruned open (differential — the same store,
 *      with the provable field stripped from its manifest, takes the full-scan
 *      path and must agree fact for fact, before and after `open()`).
 *   3. A manifest entry with no `lastGeneration` (legacy, or hand-repaired) is
 *      READ. Never prune what the manifest cannot prove.
 */
import { describe, it, expect, afterEach } from 'vitest'
import * as fs from 'node:fs'
import * as os from 'node:os'
import * as path from 'node:path'
import { spawn } from 'node:child_process'
import {
  FactLog,
  FACTS_MANIFEST_PATH,
  type CommitFact,
  type FactLogStorage
} from '../../src/db/factLog.js'
import { FileSystemStorage } from '../../src/storage/adapters/fileSystemStorage.js'

const REPO_ROOT = process.cwd()
const TSX = path.join(REPO_ROOT, 'node_modules', '.bin', 'tsx')
/** ~1KB frames against a 4KB rotation threshold: ~5 facts per segment. */
const ROTATE_BYTES = 4096

const tmpDirs: string[] = []
function makeTempDir(): string {
  const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'brainy-factlog-prune-'))
  tmpDirs.push(dir)
  return dir
}

afterEach(() => {
  for (const dir of tmpDirs.splice(0)) {
    try {
      fs.rmSync(dir, { recursive: true, force: true })
    } catch {
      /* best effort */
    }
    try {
      fs.rmSync(`${dir}.ready.json`, { force: true })
    } catch {
      /* best effort */
    }
  }
})

const UUID = (n: number): string => `00000000-0000-4000-8000-${String(n).padStart(12, '0')}`

/** One ~1KB fact — the padding is what makes rotation cheap to provoke. */
function fact(generation: number): CommitFact {
  return {
    generation,
    timestamp: 1_700_000_000_000 + generation,
    ops: [
      {
        kind: 'noun',
        id: UUID(generation),
        record: {
          metadata: { noun: 'document', pad: 'x'.repeat(900), g: generation },
          vector: null
        }
      }
    ]
  }
}

/**
 * A deterministic int minter so the log writes the V2 format production
 * writes (the prune is a manifest-level decision and never touches segment
 * bytes — but the pins should run against the bytes the fleet actually has).
 */
function makeMinter(): (kind: 'noun' | 'verb', id: string) => bigint {
  const ints = new Map<string, bigint>()
  return (kind, id) => {
    const key = `${kind}:${id}`
    let minted = ints.get(key)
    if (minted === undefined) {
      minted = BigInt(ints.size + 1)
      ints.set(key, minted)
    }
    return minted
  }
}

/** Open a fact log over a store directory (a fresh adapter each time — this is
 *  what a reopen actually does). */
async function openStore(dir: string): Promise<{ storage: any; log: FactLog }> {
  const storage: any = new FileSystemStorage(dir)
  await storage.init()
  const log = new FactLog(storage as FactLogStorage, { rotateBytes: ROTATE_BYTES })
  log.setIntMinter(makeMinter())
  return { storage, log }
}

/** Build a log of `count` facts (rotating every ~5), left durable, not closed. */
async function buildLog(dir: string, count: number): Promise<number> {
  const { log } = await openStore(dir)
  await log.open(0)
  for (let g = 1; g <= count; g++) await log.append(fact(g))
  await log.sync()
  return log.headGeneration()
}

/** Capture the narration channel (`prodLog.narrate` → console.warn). */
async function captureNarration<T>(
  fn: () => Promise<T>
): Promise<{ result: T; lines: string[] }> {
  const lines: string[] = []
  const original = console.warn
  console.warn = ((...args: unknown[]) => {
    lines.push(args.map((a) => String(a)).join(' '))
  }) as typeof console.warn
  try {
    return { result: await fn(), lines }
  } finally {
    console.warn = original
  }
}

/** The counters the open narrated — the pin's only source of truth for what
 *  was read (a wall-clock assertion could pass on a warm page cache). */
function scanCounts(lines: string[]): { read: number; pruned: number; total: number } {
  const line = lines.find((l) => l.includes('[FactLog] above-manifest peek above generation'))
  if (!line) {
    throw new Error(`no peek narration in:\n${lines.join('\n')}`)
  }
  const match = /(\d+) segment\(s\) read, (\d+) pruned of (\d+)/.exec(line)
  if (!match) throw new Error(`unparsable peek narration: ${line}`)
  return { read: Number(match[1]), pruned: Number(match[2]), total: Number(match[3]) }
}

interface SegmentEntryOnDisk {
  file: string
  firstGeneration: number
  lastGeneration?: number
  facts: number
  bytes: number
}

async function readManifest(dir: string): Promise<{
  segments: SegmentEntryOnDisk[]
  tailSegment: string | null
}> {
  const storage: any = new FileSystemStorage(dir)
  await storage.init()
  return (await storage.readRawObject(FACTS_MANIFEST_PATH)) as any
}

async function rewriteManifest(
  dir: string,
  mutate: (manifest: any) => void
): Promise<void> {
  const storage: any = new FileSystemStorage(dir)
  await storage.init()
  const manifest = await storage.readRawObject(FACTS_MANIFEST_PATH)
  mutate(manifest)
  await storage.writeRawObject(FACTS_MANIFEST_PATH, manifest)
  await storage.syncRawObjects([FACTS_MANIFEST_PATH])
}

/** Every fact the log holds, in order — the recovered state, read back. */
async function allFacts(log: FactLog): Promise<CommitFact[]> {
  const out: CommitFact[] = []
  const handle = log.scanFacts()
  for await (const batch of handle.batches()) out.push(...batch.facts)
  return out
}

describe('fact log — the open reads only the segments that can hold facts above the bound', () => {
  it('a clean close + reopen over ≥4 sealed segments reads exactly the tail and finds nothing', async () => {
    const dir = makeTempDir()
    const head = await buildLog(dir, 30)

    const manifest = await readManifest(dir)
    expect(manifest.segments.length).toBeGreaterThanOrEqual(4) // the fixture is real
    expect(manifest.tailSegment).not.toBeNull()

    // The reopen: a clean close means committed === the log's head.
    const { log } = await openStore(dir)
    const { result: orphans, lines } = await captureNarration(() => log.peekFactsAbove(head))

    expect(orphans).toEqual([]) // the fold finds nothing, as it always does after a clean close
    const counts = scanCounts(lines)
    expect(counts.read).toBe(1) // EXACTLY the tail
    expect(counts.total).toBe(manifest.segments.length + 1)
    expect(counts.pruned).toBe(manifest.segments.length)

    // And the reconciling open still lands on the same committed prefix.
    await log.open(head)
    expect(log.headGeneration()).toBe(head)
    expect((await allFacts(log)).map((f) => f.generation)).toEqual(
      Array.from({ length: head }, (_, i) => i + 1)
    )
  })

  it('a manifest entry with no lastGeneration is READ — never prune what you cannot prove', async () => {
    const dir = makeTempDir()
    const head = await buildLog(dir, 30)
    const before = await readManifest(dir)
    expect(before.segments.length).toBeGreaterThanOrEqual(4)

    // A legacy/hand-repaired entry: the field the prune needs is simply absent.
    await rewriteManifest(dir, (m) => {
      delete m.segments[0].lastGeneration
    })

    const { log } = await openStore(dir)
    const { result: orphans, lines } = await captureNarration(() => log.peekFactsAbove(head))

    expect(orphans).toEqual([]) // still nothing above the bound — it was READ to find out
    const counts = scanCounts(lines)
    expect(counts.read).toBe(2) // the unprovable entry + the tail
    expect(counts.pruned).toBe(before.segments.length - 1)
    expect(counts.total).toBe(before.segments.length + 1)
  })

  it(
    'a SIGKILLed writer that sealed segments above the committed pointer recovers identically to an unpruned open',
    async () => {
      const dir = makeTempDir()
      const readyPath = `${dir}.ready.json`
      // A real process death: the child fsyncs its segments, records what it
      // reached, and SIGKILLs ITSELF — no close, no unwind, no chance to tidy.
      const script = `
        import * as fs from 'node:fs'
        import { FactLog } from ${JSON.stringify(path.join(REPO_ROOT, 'src', 'db', 'factLog.ts'))}
        import { FileSystemStorage } from ${JSON.stringify(path.join(REPO_ROOT, 'src', 'storage', 'adapters', 'fileSystemStorage.ts'))}
        const UUID = (n) => '00000000-0000-4000-8000-' + String(n).padStart(12, '0')
        const fact = (g) => ({
          generation: g,
          timestamp: 1700000000000 + g,
          ops: [{ kind: 'noun', id: UUID(g), record: { metadata: { noun: 'document', pad: 'x'.repeat(900), g }, vector: null } }]
        })
        const ints = new Map()
        const storage = new FileSystemStorage(${JSON.stringify(dir)})
        await storage.init()
        const log = new FactLog(storage, { rotateBytes: ${ROTATE_BYTES} })
        log.setIntMinter((kind, id) => {
          const key = kind + ':' + id
          if (!ints.has(key)) ints.set(key, BigInt(ints.size + 1))
          return ints.get(key)
        })
        await log.open(0)
        for (let g = 1; g <= 30; g++) await log.append(fact(g))
        await log.sync()
        fs.writeFileSync(${JSON.stringify(readyPath)}, JSON.stringify({ head: log.headGeneration() }))
        process.kill(process.pid, 'SIGKILL')
      `
      const scriptPath = path.join(dir, 'crash-writer.mts')
      fs.writeFileSync(scriptPath, script)
      const child = spawn(TSX, [scriptPath], { cwd: REPO_ROOT, stdio: ['ignore', 'pipe', 'pipe'] })
      let output = ''
      child.stdout.on('data', (d) => { output += String(d) })
      child.stderr.on('data', (d) => { output += String(d) })
      const exit = await new Promise<{ code: number | null; signal: string | null }>((resolve) =>
        child.on('exit', (code, signal) => resolve({ code, signal }))
      )
      if (!fs.existsSync(readyPath)) {
        throw new Error(`the crash writer never reached its kill point:\n${output}`)
      }
      // Death, not a shutdown: no close(), no unwind, no orderly exit code.
      expect(exit.signal ?? `code ${exit.code}`).not.toBe('code 0')
      const head = JSON.parse(fs.readFileSync(readyPath, 'utf8')).head as number
      expect(head).toBe(30)

      // The committed pointer the survivor comes back on: mid-log, so sealed
      // segments hold facts ABOVE it — the exact shape the prune must not skip.
      const committed = 12
      const manifest = await readManifest(dir)
      const straddling = manifest.segments.filter(
        (s) => s.firstGeneration <= committed && (s.lastGeneration ?? 0) > committed
      )
      const entirelyAbove = manifest.segments.filter((s) => s.firstGeneration > committed)
      expect(straddling.length).toBeGreaterThanOrEqual(1)
      expect(entirelyAbove.length).toBeGreaterThanOrEqual(1)

      // THE DIFFERENTIAL. The unpruned answer, through the SAME code on the
      // SAME bytes: a peek above generation 0 can prune nothing (no sealed
      // segment ends at or below 0), so it reads every segment file and
      // decodes every frame — exactly what this open used to do — and its
      // facts above the pointer are what the fold is entitled to replay.
      const { log } = await openStore(dir)
      const { result: fullScan, lines: fullLines } = await captureNarration(() =>
        log.peekFactsAbove(0)
      )
      expect(scanCounts(fullLines)).toEqual({
        read: manifest.segments.length + 1,
        pruned: 0,
        total: manifest.segments.length + 1
      })
      const unprunedAnswer = fullScan.filter((f) => f.generation > committed)

      const { result: prunedAnswer, lines } = await captureNarration(() =>
        log.peekFactsAbove(committed)
      )

      // The sealed segments above the bound were READ, not skipped.
      const counts = scanCounts(lines)
      expect(counts.read).toBe(straddling.length + entirelyAbove.length + 1)
      expect(counts.pruned).toBe(manifest.segments.length - straddling.length - entirelyAbove.length)
      expect(counts.pruned).toBeGreaterThan(0) // the prune did engage, and was still right
      expect(prunedAnswer.map((f) => f.generation)).toEqual(
        Array.from({ length: head - committed }, (_, i) => committed + 1 + i)
      )
      // Facts that live in a SEALED segment (not the tail) came back.
      expect(prunedAnswer.some((f) => f.generation <= (straddling[0].lastGeneration ?? 0))).toBe(
        true
      )
      // Fact for fact, the pruned answer IS the unpruned answer — so whatever
      // the recovery replays, it replays identically.
      expect(prunedAnswer).toEqual(unprunedAnswer)

      // The fold's streaming twin (the unclean-open path) agrees too.
      const streamed: CommitFact[] = []
      for await (const batch of log.streamFactsAbove(committed)) streamed.push(...batch)
      expect(streamed).toEqual(unprunedAnswer)

      // And the reconciling open rolls back exactly as it always did: the two
      // never-committed sealed segments dropped, the straddling one cut, the
      // tail truncated — the log left as the committed prefix.
      await log.open(committed)
      expect(log.headGeneration()).toBe(committed)
      expect((await allFacts(log)).map((f) => f.generation)).toEqual(
        Array.from({ length: committed }, (_, i) => i + 1)
      )
      const after = await readManifest(dir)
      expect(after.segments.map((s) => s.file)).toEqual(
        manifest.segments
          .filter((s) => s.firstGeneration <= committed)
          .map((s) => s.file)
      )
      expect(after.segments[after.segments.length - 1].lastGeneration).toBe(committed)
    },
    120_000
  )
})