/**
 * @module tests/integration/pending-embed-checkpoint
 * @description THE PENDING-EMBED CHECKPOINT — the bound that engages on the
 * brains that need it.
 *
 * 10.4.9 bounded the open-path `recover-pending-embeds` fold with a LOW-WATER
 * MARK: the log head at which the pending set last drained to EMPTY. That mark
 * carries no set, so it can only be written when the set is empty — and a brain
 * holding even ONE id that never lands (an embed that keeps failing, a worker
 * that never gets to it, a row reaped in memory only and re-folded every open)
 * never drains, therefore never writes a mark, therefore re-reads its WHOLE
 * fact log on every single open. The bound was absent from exactly the brains
 * whose fold is expensive: a silent scaling defect.
 *
 * The cure is a CHECKPOINT of the pending set —
 * `_system/pending_embeds_checkpoint.json` = `{ generation, pending, writtenAt }`,
 * meaning "as of durable generation G the pending set was exactly this list".
 * Open seeds the set from `pending` and scans only from `G + 1`, so the fold is
 * O(facts since G) whether or not the set ever drains.
 *
 * What this suite pins:
 *   1. A brain with one permanently-stuck pending id, closed cleanly and
 *      reopened, scans ONLY the facts after the checkpoint — asserted from the
 *      fold's own accounting, never a clock. The same fixture pins the DEFECT:
 *      no low-water mark exists on that brain, because it never drained.
 *   2. A crash matrix in a REAL child process (SIGKILL, no close), for kills
 *      before a checkpoint write, after one with embeds landed and flushed
 *      after it, and after one with an UN-FLUSHED tail at the moment of death.
 *      The invariant in every row is differential: the checkpoint-bounded fold
 *      the reopened brain actually ran ≡ a full fold from generation 1 over the
 *      same recovered log.
 *   3. A torn checkpoint falls back — loudly (the adapter's torn-record gauge
 *      plus the fold's own narration of which bound applied) and correctly.
 *   4. The existing low-water pins keep passing unchanged
 *      (`pending-embed-low-water.test.ts`): the mark is still written and is
 *      still read, now as the FALLBACK bound beneath the checkpoint.
 *
 * The crash-recovery contract is untouched: the fold runs on the open's
 * foreground, so a reopened brain has its markers re-armed when open() returns.
 */
import { describe, it, expect, afterEach } from 'vitest'
import { mkdtempSync, rmSync, existsSync, readFileSync, writeFileSync } from 'node:fs'
import { spawn } from 'node:child_process'
import { gunzipSync } from 'node:zlib'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { Brainy } from '../../src/brainy.js'
import { NounType } from '../../src/types/graphTypes.js'
import { getTornRecordGauge } from '../../src/storage/tornRecordError.js'

const CHECKPOINT_PATH = '_system/pending_embeds_checkpoint.json'
const LOWWATER_PATH = '_system/pending_embeds_lowwater.json'
const REPO_ROOT = process.cwd()
const TSX = join(REPO_ROOT, 'node_modules', '.bin', 'tsx')

/** The fold's own accounting for the most recent open. */
interface FoldReport {
  bound: 'checkpoint' | 'low-water' | 'genesis'
  fromGeneration: number
  factsScanned: number
  seeded: number
  pending: number
}

const roots: string[] = []
const liveBrains: Brainy<any>[] = []

function dir(): string {
  const d = mkdtempSync(join(tmpdir(), 'brainy-embed-ckpt-'))
  roots.push(d)
  return d
}

async function open(root: string, opts?: { blockWorker?: boolean }): Promise<Brainy<any>> {
  const brain = new Brainy<any>({
    requireSubtype: false,
    storage: { type: 'filesystem', path: root }
  })
  // Blocking the worker BEFORE init() is how a "permanently stuck" pending id
  // is built deterministically: the state under test is "an id the fold keeps
  // re-arming and nothing ever disarms", and its production causes (a failing
  // embedder, a wedged model, a data-less row) all reduce to exactly that.
  if (opts?.blockWorker) (brain as unknown as { kickEmbedWorker: () => void }).kickEmbedWorker = () => {}
  await brain.init()
  liveBrains.push(brain)
  return brain
}

function foldReport(brain: Brainy<any>): FoldReport {
  const report = (brain as unknown as { _pendingEmbedFoldReport: FoldReport | null })
    ._pendingEmbedFoldReport
  if (report === null) throw new Error('the open ran no pending-embed fold')
  return report
}

function pendingIds(brain: Brainy<any>): string[] {
  return [
    ...(brain as unknown as { _pendingEmbedIds: Set<string> })._pendingEmbedIds
  ].sort()
}

/** Read an artifact straight off disk (the adapter gzips raw objects). */
function readArtifact(root: string, path: string): Record<string, unknown> | null {
  const plain = join(root, ...path.split('/'))
  const gz = `${plain}.gz`
  if (existsSync(gz)) return JSON.parse(gunzipSync(readFileSync(gz)).toString('utf-8'))
  if (existsSync(plain)) return JSON.parse(readFileSync(plain, 'utf-8'))
  return null
}

/** The on-disk path the adapter actually used for an artifact. */
function artifactPath(root: string, path: string): string | null {
  const plain = join(root, ...path.split('/'))
  const gz = `${plain}.gz`
  if (existsSync(gz)) return gz
  if (existsSync(plain)) return plain
  return null
}

/**
 * THE DIFFERENTIAL ORACLE: fold the log from generation 1 with exactly the
 * engine's own rules. This is what the bounded fold must agree with, and its
 * fact count is what the unbounded fold used to read at every open.
 */
async function fullFold(brain: Brainy<any>): Promise<{ ids: string[]; facts: number }> {
  const log = (
    brain as unknown as { generationStore: { getFactLog(): any } }
  ).generationStore.getFactLog()
  const pending = new Set<string>()
  let facts = 0
  const scan = log.scanFacts({ fromGeneration: 1 })
  for await (const batch of scan.batches()) {
    for (const fact of batch.facts) {
      facts++
      for (const record of fact.records ?? []) {
        if (record.type === 'embed.pending') pending.add(record.id)
        else if (record.type === 'embed.landed') pending.delete(record.id)
      }
      for (const op of fact.ops) {
        if (op.kind === 'noun' && op.record === null) pending.delete(op.id)
      }
    }
  }
  return { ids: [...pending].sort(), facts }
}

/** Capture every console.warn/error line emitted while `fn` runs. */
async function captureConsole<T>(fn: () => Promise<T>): Promise<{ result: T; lines: string[] }> {
  const lines: string[] = []
  const origWarn = console.warn
  const origError = console.error
  const sink = (...args: unknown[]) => {
    lines.push(args.map((a) => String(a)).join(' '))
  }
  console.warn = sink as typeof console.warn
  console.error = sink as typeof console.error
  try {
    const result = await fn()
    return { result, lines }
  } finally {
    console.warn = origWarn
    console.error = origError
  }
}

/**
 * Run a child process that arranges a store and then waits forever, so the
 * parent can SIGKILL it. A real process death is the only honest way to pin
 * "no close ran, no shutdown hook ran, RAM is gone".
 *
 * `detached` puts the child in its own process GROUP: tsx runs the script in a
 * grandchild, and only a group-wide signal reaches the process holding the
 * writer lock.
 */
function spawnArranger(root: string, body: string): Promise<{
  child: ReturnType<typeof spawn>
  output: () => string
}> {
  const scriptPath = join(root, 'arrange.mts')
  writeFileSync(scriptPath, body)
  const child = spawn(TSX, [scriptPath], {
    cwd: REPO_ROOT,
    stdio: ['ignore', 'pipe', 'pipe'],
    detached: true
  })
  let out = ''
  child.stdout!.on('data', (d) => { out += String(d) })
  child.stderr!.on('data', (d) => { out += String(d) })
  return new Promise((resolvePromise, rejectPromise) => {
    const timer = setTimeout(
      () => rejectPromise(new Error(`arranger never became READY:\n${out}`)),
      180_000
    )
    child.stdout!.on('data', () => {
      if (out.includes('READY')) {
        clearTimeout(timer)
        resolvePromise({ child, output: () => out })
      }
    })
    child.on('exit', (code) => {
      clearTimeout(timer)
      if (!out.includes('READY')) rejectPromise(new Error(`arranger exited ${code}:\n${out}`))
    })
  })
}

/** Parse the `IDS:{...}` line an arranger prints — supplied ids are normalised
 *  to canonical uuids, and the markers, checkpoint and fold all speak those. */
function childIds(output: string): Record<string, string> {
  const line = output.split('\n').find((l) => l.startsWith('IDS:'))
  if (!line) throw new Error(`arranger printed no IDS line:\n${output}`)
  return JSON.parse(line.slice('IDS:'.length))
}

/** SIGKILL the whole group and wait for the grandchild's death to settle. */
async function sigkill(child: ReturnType<typeof spawn>): Promise<void> {
  process.kill(-(child.pid as number), 'SIGKILL')
  await new Promise<void>((r) => child.on('exit', () => r()))
  await new Promise<void>((r) => setTimeout(r, 500))
}

/** The preamble every arranger child shares. */
function childPreamble(root: string): string {
  return `
    import { Brainy } from ${JSON.stringify(join(REPO_ROOT, 'src', 'brainy.ts'))}
    const ROOT = ${JSON.stringify(root)}
    const brain = new Brainy<any>({ requireSubtype: false, storage: { type: 'filesystem', path: ROOT } })
    const block = () => { (brain as any).kickEmbedWorker = () => {} }
    const settleCheckpoint = async () => {
      // The cadence write is fire-and-forget; wait for the single flight.
      for (let i = 0; i < 200; i++) {
        if (!(brain as any)._pendingEmbedCheckpointFlight) break
        await (brain as any)._pendingEmbedCheckpointFlight.catch(() => {})
      }
    }
  `
}

afterEach(async () => {
  for (const brain of liveBrains.splice(0)) {
    try { await brain.close() } catch { /* already closed / crashed — teardown only */ }
  }
  for (const d of roots.splice(0)) rmSync(d, { recursive: true, force: true })
})

// ===========================================================================
// 1. The stuck-id brain — the defect, and the bound that now engages on it
// ===========================================================================

describe('pending-embed checkpoint — a brain whose pending set never drains', () => {
  it('a permanently-stuck pending id: the reopen scans only the facts after the checkpoint', async () => {
    const root = dir()
    const first = await open(root, { blockWorker: true })
    // add() returns the CANONICAL id (supplied ids are normalised), and that is
    // the id the markers, the checkpoint and the fold all speak.
    const stuck = await first.add({
      id: 'stuck',
      data: 'a deferred row whose embed never lands',
      type: NounType.Thing,
      deferEmbedding: true
    })
    expect(first.pendingEmbedCount()).toBe(1)
    // Ordinary traffic after it — every one of these is a fact the unbounded
    // fold had to re-read at every open, forever, because of that one id.
    for (let i = 0; i < 12; i++) {
      await first.add({ id: `row-${i}`, data: `row ${i}`, type: NounType.Thing })
    }
    await first.close()
    liveBrains.splice(liveBrains.indexOf(first), 1)

    // THE DEFECT, PINNED: the pending set never drained, so the old bound was
    // never written — nothing on this brain could have shortened its fold.
    expect(readArtifact(root, LOWWATER_PATH)).toBeNull()
    // The checkpoint IS written at the clean close, set non-empty and all.
    const checkpoint = readArtifact(root, CHECKPOINT_PATH) as {
      generation: number
      pending: string[]
    } | null
    expect(checkpoint).not.toBeNull()
    expect(checkpoint!.generation).toBeGreaterThan(0)
    expect(checkpoint!.pending).toEqual([stuck])

    const second = await open(root, { blockWorker: true })
    const report = foldReport(second)
    // THE FIX, from the fold's own counter — not the clock.
    expect(report.bound).toBe('checkpoint')
    expect(report.fromGeneration).toBe(checkpoint!.generation + 1)
    expect(report.factsScanned).toBe(0)
    expect(report.seeded).toBe(1)
    // The crash-recovery contract is intact: the marker is re-armed by open().
    expect(pendingIds(second)).toEqual([stuck])
    expect(second.pendingEmbedCount()).toBe(1)

    // The differential: the bounded answer is the full-fold answer, and the
    // full fold is what the previous bound would have had to read.
    const full = await fullFold(second)
    expect(full.ids).toEqual([stuck])
    expect(full.facts).toBeGreaterThanOrEqual(13)
    expect(report.factsScanned).toBeLessThan(full.facts)
  }, 180_000)

  it('the bound stays O(delta) across repeated opens while the id is still stuck', async () => {
    const root = dir()
    const first = await open(root, { blockWorker: true })
    const stuck = await first.add({
      id: 'stuck',
      data: 'never lands',
      type: NounType.Thing,
      deferEmbedding: true
    })
    for (let i = 0; i < 6; i++) {
      await first.add({ id: `a-${i}`, data: `a ${i}`, type: NounType.Thing })
    }
    await first.close()
    liveBrains.splice(liveBrains.indexOf(first), 1)

    const second = await open(root, { blockWorker: true })
    expect(foldReport(second).factsScanned).toBe(0)
    // More history under the same stuck id.
    for (let i = 0; i < 9; i++) {
      await second.add({ id: `b-${i}`, data: `b ${i}`, type: NounType.Thing })
    }
    await second.close()
    liveBrains.splice(liveBrains.indexOf(second), 1)

    const third = await open(root, { blockWorker: true })
    const report = foldReport(third)
    const full = await fullFold(third)
    expect(report.bound).toBe('checkpoint')
    expect(report.factsScanned).toBe(0)
    // The unbounded fold grew with the store; the bounded one did not.
    expect(full.facts).toBeGreaterThanOrEqual(16)
    expect(pendingIds(third)).toEqual([stuck])
    expect(full.ids).toEqual([stuck])
  }, 180_000)
})

// ===========================================================================
// 2. Torn checkpoint — falls back, loudly, correctly
// ===========================================================================

describe('pending-embed checkpoint — a torn checkpoint never shortens the fold', () => {
  it('an undecodable checkpoint file degrades to the next bound, loudly, with the right pending set', async () => {
    const root = dir()
    const first = await open(root, { blockWorker: true })
    const stuck = await first.add({
      id: 'stuck',
      data: 'never lands',
      type: NounType.Thing,
      deferEmbedding: true
    })
    for (let i = 0; i < 5; i++) {
      await first.add({ id: `row-${i}`, data: `row ${i}`, type: NounType.Thing })
    }
    await first.close()
    liveBrains.splice(liveBrains.indexOf(first), 1)

    const onDisk = artifactPath(root, CHECKPOINT_PATH)
    expect(onDisk).not.toBeNull()
    // Tear it: bytes that are neither valid gzip nor valid JSON. A torn file
    // must THROW on read — never parse into a partial `pending` list.
    writeFileSync(onDisk!, 'not a checkpoint at all {{{')

    const before = getTornRecordGauge().count
    const { result: second, lines } = await captureConsole(async () =>
      open(root, { blockWorker: true })
    )
    const report = foldReport(second)
    // Fell back — never to a shorter bound, and never silently.
    expect(report.bound).not.toBe('checkpoint')
    expect(report.seeded).toBe(0)
    expect(report.fromGeneration).toBe(1) // no mark either: this brain never drained
    // LOUD, two ways: the adapter's torn-record gauge and its production error…
    expect(getTornRecordGauge().count).toBeGreaterThan(before)
    expect(getTornRecordGauge().lastPath).toContain('pending_embeds_checkpoint')
    expect(lines.some((l) => /TORN RECORD/.test(l))).toBe(true)
    // …and the fold's own narration of which bound it actually used.
    expect(lines.some((l) => /pending-embed fold: genesis bound/.test(l))).toBe(true)

    // CORRECT: the marker is still recovered, from the log itself.
    expect(pendingIds(second)).toEqual([stuck])
    const full = await fullFold(second)
    expect(full.ids).toEqual([stuck])
    expect(report.factsScanned).toBe(full.facts)
  }, 180_000)

  it('a well-formed but shape-invalid checkpoint is refused whole, never partially trusted', async () => {
    const root = dir()
    const first = await open(root, { blockWorker: true })
    const stuck = await first.add({
      id: 'stuck',
      data: 'never lands',
      type: NounType.Thing,
      deferEmbedding: true
    })
    await first.add({ id: 'other', data: 'ordinary row', type: NounType.Thing })
    await first.close()
    liveBrains.splice(liveBrains.indexOf(first), 1)

    // A checkpoint with a plausible generation but a `pending` that is not a
    // list of ids: trusting the generation alone would bound the scan behind a
    // set that was never recovered — the exact shape that loses a vector.
    const onDisk = artifactPath(root, CHECKPOINT_PATH)!
    const good = readArtifact(root, CHECKPOINT_PATH) as { generation: number }
    rmSync(onDisk)
    writeFileSync(
      join(root, '_system', 'pending_embeds_checkpoint.json'),
      JSON.stringify({ generation: good.generation, pending: { stuck: true }, writtenAt: 1 })
    )

    const { result: second, lines } = await captureConsole(async () =>
      open(root, { blockWorker: true })
    )
    expect(lines.some((l) => /pending-embed checkpoint REFUSED/.test(l))).toBe(true)
    const report = foldReport(second)
    expect(report.bound).not.toBe('checkpoint')
    expect(report.seeded).toBe(0)
    expect(pendingIds(second)).toEqual([stuck])
  }, 180_000)
})

// ===========================================================================
// 3. The crash matrix — real processes, real SIGKILL, differential invariant
// ===========================================================================

describe('pending-embed checkpoint — crash matrix (real child process, SIGKILL)', () => {
  /**
   * The invariant every row shares: whatever the reopened brain's fold did with
   * whatever bound survived the crash, its pending set must equal the truth a
   * full fold from generation 1 derives from the SAME recovered log.
   */
  async function assertDifferentialAfterCrash(root: string): Promise<{
    report: FoldReport
    full: { ids: string[]; facts: number }
    pending: string[]
  }> {
    const reopened = await open(root, { blockWorker: true })
    const report = foldReport(reopened)
    const full = await fullFold(reopened)
    const pending = pendingIds(reopened)
    expect(pending).toEqual(full.ids)
    return { report, full, pending }
  }

  it('killed BEFORE any checkpoint was written — falls back and recovers the marker from the log', async () => {
    const root = dir()
    const { child, output } = await spawnArranger(
      root,
      `${childPreamble(root)}
      block()
      await brain.init()
      await brain.add({ id: 'landed-row', data: 'an ordinary row', type: 'thing' })
      const stuck = await brain.add({ id: 'stuck-1', data: 'deferred, never lands', type: 'thing', deferEmbedding: true })
      await brain.flush()
      console.log('IDS:' + JSON.stringify({ stuck }))
      console.log('READY')
      setInterval(() => {}, 1000)
      `
    )
    const ids = childIds(output())
    // One enqueue is well under the cadence and the set never drained, so no
    // checkpoint exists — this is the pre-checkpoint crash.
    expect(readArtifact(root, CHECKPOINT_PATH)).toBeNull()
    await sigkill(child)

    const { report, pending } = await assertDifferentialAfterCrash(root)
    expect(report.bound).toBe('genesis')
    expect(pending).toEqual([ids.stuck])
  }, 300_000)

  it('killed AFTER a checkpoint, with an embed landed and flushed after it — the post-checkpoint facts carry the disarm', async () => {
    const root = dir()
    const { child, output } = await spawnArranger(
      root,
      `${childPreamble(root)}
      await brain.init()
      // Land one deferred embed: the drain arms the checkpoint debt.
      await brain.add({ id: 'seed', data: 'lands first', type: 'thing', deferEmbedding: true })
      await brain.awaitPendingEmbeds()
      await brain.flush()
      // A second deferred write pays the debt (the head is at the manifest now),
      // then LANDS — its embed.landed rides a fact ABOVE the checkpoint.
      const landsAfter = await brain.add({ id: 'lands-after', data: 'lands after the checkpoint', type: 'thing', deferEmbedding: true })
      await settleCheckpoint()
      await brain.awaitPendingEmbeds()
      // …and one that never will.
      block()
      const stuck = await brain.add({ id: 'stuck-1', data: 'deferred, never lands', type: 'thing', deferEmbedding: true })
      await brain.add({ id: 'plain', data: 'more history', type: 'thing' })
      await brain.flush()
      console.log('IDS:' + JSON.stringify({ stuck, landsAfter }))
      console.log('READY')
      setInterval(() => {}, 1000)
      `
    )
    const ids = childIds(output())
    const checkpoint = readArtifact(root, CHECKPOINT_PATH) as {
      generation: number
      pending: string[]
    } | null
    expect(checkpoint).not.toBeNull()
    await sigkill(child)

    const { report, full, pending } = await assertDifferentialAfterCrash(root)
    expect(report.bound).toBe('checkpoint')
    expect(report.fromGeneration).toBe(checkpoint!.generation + 1)
    // The bound really bounded: fewer facts than the whole log.
    expect(report.factsScanned).toBeLessThan(full.facts)
    // A landed embed above the checkpoint is disarmed by the scan, not lost;
    // the stuck one is re-armed.
    expect(pending).toEqual([ids.stuck])
    expect(pending).not.toContain(ids.landsAfter)
  }, 300_000)

  it('killed AFTER a checkpoint with an UN-FLUSHED tail — truncated facts and the bounded fold still agree', async () => {
    const root = dir()
    const { child } = await spawnArranger(
      root,
      `${childPreamble(root)}
      await brain.init()
      await brain.add({ id: 'seed', data: 'lands first', type: 'thing', deferEmbedding: true })
      await brain.awaitPendingEmbeds()
      await brain.flush()
      await brain.add({ id: 'lands-after', data: 'lands after the checkpoint', type: 'thing', deferEmbedding: true })
      await settleCheckpoint()
      await brain.awaitPendingEmbeds()
      await brain.flush()
      // Now write PAST the manifest and never flush: these facts are the tail a
      // crash truncates. Whatever survives, the two folds must agree on it.
      block()
      await brain.add({ id: 'stuck-tail', data: 'deferred, never lands', type: 'thing', deferEmbedding: true })
      await brain.add({ id: 'plain-tail', data: 'unflushed history', type: 'thing' })
      console.log('READY')
      setInterval(() => {}, 1000)
      `
    )
    const checkpoint = readArtifact(root, CHECKPOINT_PATH) as { generation: number } | null
    expect(checkpoint).not.toBeNull()
    await sigkill(child)

    const { report } = await assertDifferentialAfterCrash(root)
    // The checkpoint's generation is at or below the manifest by construction,
    // so it survived the truncation and still bounds the fold.
    expect(report.bound).toBe('checkpoint')
    expect(report.fromGeneration).toBe(checkpoint!.generation + 1)
  }, 300_000)
})