open-brainy/tests/unit/reprojection/reprojection-engine.test.ts

/**
 * @module tests/unit/reprojection/reprojection-engine
 * @description Spec-by-example for the pure-TS reprojection engine — the
 * frozen contract mirrored from the native twin (a shared conformance suite
 * runs against both, so the shapes pinned here are load-bearing):
 *
 * (a) register + advance folds a scripted source to caught-up with exact
 *     watermark/applied counts and adapter-owned stamping;
 * (b) budget exhaustion answers mid-stream and a second advance RESUMES from
 *     the watermark — never a refold;
 * (c) a door bump mid-advance preempts within one installment — pinned by
 *     MECHANISM (no further applyBatch after the bumping step), with only a
 *     generous wall-clock sanity bound;
 * (d) advanceAll round-robins families at batch granularity — no starvation;
 * (e) swap builds beside (the old adapter serves throughout), flips
 *     atomically at parity, refuses a concurrent swap with a typed error;
 * (f) quarantine: a typed poison fact is skipped + ledgered, narration
 *     doubles, a NON-typed throw aborts loudly;
 * (g) discard() lands on the LOSING adapter after a swap.
 */
import { describe, it, expect, vi, afterEach } from 'vitest'
import {
  ReprojectionEngine,
  DoorSignal,
  ProjectionApplyError,
  SwapInFlightError,
  MAX_INSTALLMENT_MS,
  type ProjectionAdapter,
  type FactSource
} from '../../../src/reprojection/reprojectionEngine.js'
import { FactLogSource } from '../../../src/reprojection/factLogSource.js'
import type { CommitFact } from '../../../src/db/factLog.js'
import { prodLog } from '../../../src/utils/logger.js'

/** Build one committed fact for a generation. */
function fact(generation: number): CommitFact {
  return {
    generation,
    timestamp: 1_700_000_000_000 + generation,
    ops: [
      {
        kind: 'noun',
        id: `id-${generation}`,
        record: { metadata: { n: generation }, vector: null }
      }
    ]
  }
}

/** A scripted FactSource over a (possibly mutable) list of generations. */
function scriptedSource(gens: () => number[]): FactSource {
  return {
    async scan(from: number, limit: number): Promise<CommitFact[]> {
      return gens()
        .filter((g) => g > from)
        .sort((x, y) => x - y)
        .slice(0, limit)
        .map(fact)
    }
  }
}

/**
 * A recording in-memory adapter: stamps after data (the watermark advances
 * only after a successful apply), applies idempotently (a Map keyed by
 * generation), and can be scripted to poison (typed) or hard-fail (untyped)
 * specific generations, or to run a hook inside applyBatch.
 */
class RecordingAdapter implements ProjectionAdapter {
  readonly family: string
  /** Generations per applyBatch call, in call order (empty arrays included). */
  readonly batches: number[][] = []
  /** The upTo passed to each applyBatch call, in call order. */
  readonly upTos: number[] = []
  /** Latest state per generation — idempotent under at-least-once delivery. */
  readonly state = new Map<number, unknown>()
  /** Generations that throw a typed ProjectionApplyError. */
  readonly poison = new Set<number>()
  /** Generations that throw a plain (untyped) Error. */
  readonly hardFail = new Set<number>()
  /** Runs inside applyBatch after validation, before the stamp. */
  onApply?: (gens: number[]) => void | Promise<void>
  discarded = 0
  private wm: number | null

  constructor(family: string, watermark: number | null = null) {
    this.family = family
    this.wm = watermark
  }

  watermark(): number | null {
    return this.wm
  }

  async applyBatch(facts: CommitFact[], upTo: number): Promise<void> {
    for (const [i, f] of facts.entries()) {
      if (this.hardFail.has(f.generation)) {
        throw new Error(`disk exploded at generation ${f.generation}`)
      }
      if (this.poison.has(f.generation)) {
        throw new ProjectionApplyError({
          generation: f.generation,
          recordIndex: i,
          cause: new Error(`unfoldable payload at ${f.generation}`)
        })
      }
    }
    for (const f of facts) this.state.set(f.generation, f.ops)
    const gens = facts.map((f) => f.generation)
    this.batches.push(gens)
    this.upTos.push(upTo)
    if (this.onApply) await this.onApply(gens)
    this.wm = upTo // stamp-after-data
  }

  async discard(): Promise<void> {
    this.discarded++
  }
}

const range = (from: number, to: number): number[] =>
  Array.from({ length: to - from + 1 }, (_, i) => from + i)

afterEach(() => {
  vi.restoreAllMocks()
})

describe('reprojection engine — (a) register + advance to caught-up', () => {
  it('folds a scripted source in order, adapter-stamped, with exact counts', async () => {
    const source = scriptedSource(() => range(1, 7))
    const engine = new ReprojectionEngine({ source, batchSize: 3 })
    const adapter = new RecordingAdapter('a')
    engine.register(adapter)

    const result = await engine.advance('a', { budgetMs: 10_000 })

    expect(result.status).toBe('caught-up')
    expect(result.watermark).toBe(7)
    expect(result.applied).toBe(7)
    // Batch shape and the upTo handed to the adapter's own stamp.
    expect(adapter.batches).toEqual([[1, 2, 3], [4, 5, 6], [7]])
    expect(adapter.upTos).toEqual([3, 6, 7])
    // The watermark is the ADAPTER's stamp — the engine never wrote one.
    expect(adapter.watermark()).toBe(7)
    expect(engine.getAdapter('a')).toBe(adapter)
  })

  it('honors upTo as an inclusive cap and answers caught-up at the cap', async () => {
    const source = scriptedSource(() => range(1, 9))
    const engine = new ReprojectionEngine({ source, batchSize: 3 })
    const adapter = new RecordingAdapter('a')
    engine.register(adapter)

    const result = await engine.advance('a', { budgetMs: 10_000, upTo: 5 })

    expect(result.status).toBe('caught-up')
    expect(result.watermark).toBe(5)
    expect(result.applied).toBe(5)
    expect(adapter.batches.flat()).toEqual([1, 2, 3, 4, 5])
  })

  it('a caught-up family answers immediately with zero applied', async () => {
    const source = scriptedSource(() => range(1, 4))
    const engine = new ReprojectionEngine({ source, batchSize: 10 })
    const adapter = new RecordingAdapter('a', 4) // already stamped to the head
    engine.register(adapter)

    const result = await engine.advance('a', { budgetMs: 10_000 })

    expect(result).toEqual({ status: 'caught-up', watermark: 4, applied: 0 })
    expect(adapter.batches).toEqual([])
  })

  it('refuses duplicate registration and unregistered families loudly', async () => {
    const engine = new ReprojectionEngine({ source: scriptedSource(() => []) })
    engine.register(new RecordingAdapter('a'))
    expect(() => engine.register(new RecordingAdapter('a'))).toThrow(/already registered/)
    await expect(engine.advance('ghost', { budgetMs: 0 })).rejects.toThrow(/not registered/)
  })
})

describe('reprojection engine — (b) budget exhaustion resumes, never refolds', () => {
  it('returns budget-exhausted mid-stream; the next advance resumes from the watermark', async () => {
    const source = scriptedSource(() => range(1, 10))
    const engine = new ReprojectionEngine({ source, batchSize: 2 })
    const adapter = new RecordingAdapter('b')
    engine.register(adapter)

    // Zero budget: exactly ONE step of guaranteed progress, then the answer.
    const first = await engine.advance('b', { budgetMs: 0 })
    expect(first.status).toBe('budget-exhausted')
    expect(first.watermark).toBe(2)
    expect(first.applied).toBe(2)
    expect(adapter.batches).toEqual([[1, 2]])

    // The second advance RESUMES from the stamp — its first batch starts at 3.
    const second = await engine.advance('b', { budgetMs: 10_000 })
    expect(second.status).toBe('caught-up')
    expect(second.watermark).toBe(10)
    expect(second.applied).toBe(8)
    expect(adapter.batches[1]).toEqual([3, 4])
    // No refold: every generation delivered exactly once across both calls.
    expect(adapter.batches.flat()).toEqual(range(1, 10))
  })
})

describe('reprojection engine — (c) door bump preempts within one installment', () => {
  it('a bump during a step yields preempted at that step boundary — no further applyBatch', async () => {
    const source = scriptedSource(() => range(1, 12))
    const engine = new ReprojectionEngine({ source, batchSize: 2 })
    const adapter = new RecordingAdapter('c')
    adapter.onApply = (gens) => {
      if (gens[0] === 3) engine.doorSignal.bump() // door traffic mid-second-batch
    }
    engine.register(adapter)

    const started = Date.now()
    const result = await engine.advance('c', { budgetMs: 60_000 })
    const elapsed = Date.now() - started

    expect(result.status).toBe('preempted')
    expect(result.watermark).toBe(4)
    expect(result.applied).toBe(4)
    // THE MECHANISM PIN: the batch that observed the bump was the LAST batch —
    // preemption landed at the very next boundary, not after more work.
    expect(adapter.batches).toEqual([[1, 2], [3, 4]])
    // Generous wall-clock sanity only (the pin above carries the contract):
    // two tiny batches plus one installment boundary sit far under 5s.
    expect(elapsed).toBeLessThan(5_000)
    expect(MAX_INSTALLMENT_MS).toBe(50)

    // Resuming folds the rest — preemption lost nothing.
    const resumed = await engine.advance('c', { budgetMs: 60_000 })
    expect(resumed.status).toBe('caught-up')
    expect(resumed.watermark).toBe(12)
    expect(adapter.batches.flat()).toEqual(range(1, 12))
  })

  it('bumps are edge-triggered per advance: a stale bump never preempts', async () => {
    const source = scriptedSource(() => range(1, 4))
    const doorSignal = new DoorSignal()
    const engine = new ReprojectionEngine({ source, doorSignal, batchSize: 2 })
    const adapter = new RecordingAdapter('c2')
    engine.register(adapter)

    doorSignal.bump() // BEFORE the advance — belongs to earlier traffic
    const result = await engine.advance('c2', { budgetMs: 10_000 })
    expect(result.status).toBe('caught-up')
    expect(result.watermark).toBe(4)
  })
})

describe('reprojection engine — (d) advanceAll round-robin fairness', () => {
  it('a one-batch family is served on the first round despite a huge backlog next to it', async () => {
    const source = scriptedSource(() => range(1, 40))
    const engine = new ReprojectionEngine({ source, batchSize: 5 })
    const callOrder: string[] = []
    const big = new RecordingAdapter('big') // 8 batches behind
    const small = new RecordingAdapter('small', 35) // 1 batch behind
    big.onApply = () => {
      callOrder.push('big')
    }
    small.onApply = () => {
      callOrder.push('small')
    }
    engine.register(big)
    engine.register(small)

    const results = await engine.advanceAll({ budgetMs: 10_000 })

    expect(results.big).toEqual({ status: 'caught-up', watermark: 40, applied: 40 })
    expect(results.small).toEqual({ status: 'caught-up', watermark: 40, applied: 5 })
    // Fairness pin: 'small' folded its single batch on round ONE — it never
    // waited behind 'big''s backlog.
    expect(callOrder[1]).toBe('small')
    expect(callOrder.filter((f) => f === 'small')).toHaveLength(1)
  })

  it('two full-backlog families interleave strictly, one batch each per round', async () => {
    const source = scriptedSource(() => range(1, 40))
    const engine = new ReprojectionEngine({ source, batchSize: 5 })
    const callOrder: string[] = []
    const first = new RecordingAdapter('first')
    const second = new RecordingAdapter('second')
    first.onApply = () => {
      callOrder.push('first')
    }
    second.onApply = () => {
      callOrder.push('second')
    }
    engine.register(first)
    engine.register(second)

    const results = await engine.advanceAll({ budgetMs: 10_000 })

    expect(results.first.status).toBe('caught-up')
    expect(results.second.status).toBe('caught-up')
    // 8 rounds × (first, second): strict alternation — neither ever ran twice
    // while the other waited.
    expect(callOrder).toHaveLength(16)
    for (let i = 0; i < callOrder.length; i += 2) {
      expect(callOrder.slice(i, i + 2)).toEqual(['first', 'second'])
    }
  })

  it('budget exhaustion mid-round reports every unfinished family at its own watermark', async () => {
    const source = scriptedSource(() => range(1, 40))
    const engine = new ReprojectionEngine({ source, batchSize: 5 })
    const a = new RecordingAdapter('a')
    const b = new RecordingAdapter('b')
    engine.register(a)
    engine.register(b)

    const results = await engine.advanceAll({ budgetMs: 0 })

    // Zero budget: the leading family gets its one guaranteed step, then the
    // budget answer lands for everyone still mid-stream.
    expect(results.a.status).toBe('budget-exhausted')
    expect(results.b.status).toBe('budget-exhausted')
    expect(results.a.applied + results.b.applied).toBeGreaterThanOrEqual(5)
    // A later advanceAll resumes both to the head.
    const finished = await engine.advanceAll({ budgetMs: 10_000 })
    expect(finished.a.status).toBe('caught-up')
    expect(finished.b.status).toBe('caught-up')
    expect(a.batches.flat()).toEqual(range(1, 40))
    expect(b.batches.flat()).toEqual(range(1, 40))
  })
})

describe('reprojection engine — (e) swap: build-beside, atomic flip, single-flight', () => {
  it('the old adapter serves at its own watermark throughout the build; the flip is atomic at parity', async () => {
    const log = range(1, 20)
    const source = scriptedSource(() => log)
    const engine = new ReprojectionEngine({ source, batchSize: 4 })
    const oldAdapter = new RecordingAdapter('e')
    engine.register(oldAdapter)
    await engine.advance('e', { budgetMs: 10_000 })
    expect(oldAdapter.watermark()).toBe(20)

    // The log grows after the old adapter stamped — the build must reach the
    // HEAD (24), not merely the old watermark (20), before the flip.
    log.push(21, 22, 23, 24)

    const servingDuringBuild: Array<{ adapter: ProjectionAdapter | undefined; watermark: number | null }> = []
    let replacement!: RecordingAdapter
    const result = await engine.swap('e', async () => {
      replacement = new RecordingAdapter('e')
      replacement.onApply = () => {
        servingDuringBuild.push({
          adapter: engine.getAdapter('e'),
          watermark: engine.getAdapter('e')!.watermark()
        })
      }
      return replacement
    })

    // Build-beside pin: EVERY mid-build observation saw the OLD adapter,
    // still serving, still at its own stamp.
    expect(servingDuringBuild.length).toBeGreaterThan(0)
    for (const seen of servingDuringBuild) {
      expect(seen.adapter).toBe(oldAdapter)
      expect(seen.watermark).toBe(20)
    }
    // The flip: the registry now serves the replacement, at parity with head.
    expect(engine.getAdapter('e')).toBe(replacement)
    expect(result.watermark).toBe(24)
    expect(result.applied).toBe(24)
    expect(replacement.batches.flat()).toEqual(range(1, 24))
  })

  it('a second concurrent swap on the same family refuses with the typed single-flight error', async () => {
    const source = scriptedSource(() => range(1, 8))
    const engine = new ReprojectionEngine({ source, batchSize: 4 })
    engine.register(new RecordingAdapter('e2'))

    let release!: () => void
    const gate = new Promise<void>((resolve) => {
      release = resolve
    })
    const inFlight = engine.swap('e2', async () => {
      const building = new RecordingAdapter('e2')
      building.onApply = () => gate // the build parks mid-fold
      return building
    })

    // While the first swap builds, a second one is refused — typed.
    const refusal = await engine.swap('e2', async () => new RecordingAdapter('e2')).catch((e) => e)
    expect(refusal).toBeInstanceOf(SwapInFlightError)
    expect((refusal as SwapInFlightError).family).toBe('e2')

    release()
    const done = await inFlight
    expect(done.watermark).toBe(8)
    // Single-flight released: a follow-up swap is admitted again.
    const again = await engine.swap('e2', async () => new RecordingAdapter('e2'))
    expect(again.watermark).toBe(8)
  })

  it('a failed build discards the partial replacement and leaves the old adapter serving', async () => {
    const source = scriptedSource(() => range(1, 8))
    const engine = new ReprojectionEngine({ source, batchSize: 4 })
    const oldAdapter = new RecordingAdapter('e3')
    engine.register(oldAdapter)
    await engine.advance('e3', { budgetMs: 10_000 })

    let failed!: RecordingAdapter
    await expect(
      engine.swap('e3', async () => {
        failed = new RecordingAdapter('e3')
        failed.hardFail.add(5) // an UNTYPED failure mid-build
        return failed
      })
    ).rejects.toThrow(/disk exploded/)

    expect(failed.discarded).toBe(1) // the partial build was cleaned up
    expect(oldAdapter.discarded).toBe(0)
    expect(engine.getAdapter('e3')).toBe(oldAdapter) // still serving, untouched
    expect(oldAdapter.watermark()).toBe(8)
  })
})

describe('reprojection engine — (f) quarantine: the fourth answer class', () => {
  it('a typed poison fact is skipped, ledgered, and the rest folds to quarantined', async () => {
    const source = scriptedSource(() => range(1, 10))
    const engine = new ReprojectionEngine({ source, batchSize: 4 })
    const adapter = new RecordingAdapter('f')
    adapter.poison.add(6)
    engine.register(adapter)

    const result = await engine.advance('f', { budgetMs: 10_000 })

    expect(result.status).toBe('quarantined')
    expect(result.watermark).toBe(10)
    expect(result.applied).toBe(9) // every generation but the poison
    expect(adapter.batches.flat().sort((x, y) => x - y)).toEqual([1, 2, 3, 4, 5, 7, 8, 9, 10])
    expect(adapter.state.has(6)).toBe(false)

    const ledger = engine.quarantined('f')
    expect(ledger).toHaveLength(1)
    expect(ledger[0].generation).toBe(6)
    expect(ledger[0].error).toBeInstanceOf(ProjectionApplyError)
    expect(ledger[0].error.recordIndex).toBe(1) // 6 sat at index 1 of [5..8]
    expect(typeof ledger[0].at).toBe('number')
  })

  it('narration doubles: warns on the 1st, 2nd, and 4th quarantine — not the 3rd', async () => {
    const warnSpy = vi.spyOn(prodLog, 'warn').mockImplementation(() => {})
    const source = scriptedSource(() => range(1, 10))
    const engine = new ReprojectionEngine({ source, batchSize: 10 })
    const adapter = new RecordingAdapter('f2')
    for (const g of [2, 4, 6, 8]) adapter.poison.add(g)
    engine.register(adapter)

    const result = await engine.advance('f2', { budgetMs: 10_000 })

    expect(result.status).toBe('quarantined')
    expect(result.watermark).toBe(10)
    expect(result.applied).toBe(6)
    expect(engine.quarantined('f2').map((q) => q.generation)).toEqual([2, 4, 6, 8])
    const quarantineWarns = warnSpy.mock.calls.filter((c) => String(c[0]).includes('quarantined generation'))
    // 4 entries, narrated at counts 1, 2, and 4 — the 3rd stayed quiet.
    expect(quarantineWarns).toHaveLength(3)
    expect(quarantineWarns.map((c) => String(c[0]))).toEqual([
      expect.stringContaining('(1 quarantined total)'),
      expect.stringContaining('(2 quarantined total)'),
      expect.stringContaining('(4 quarantined total)')
    ])
  })

  it('an all-poison window still advances the stamp via an empty applyBatch', async () => {
    const source = scriptedSource(() => range(1, 3))
    const engine = new ReprojectionEngine({ source, batchSize: 3 })
    const adapter = new RecordingAdapter('f3')
    for (const g of [1, 2, 3]) adapter.poison.add(g)
    engine.register(adapter)

    const result = await engine.advance('f3', { budgetMs: 10_000 })

    expect(result.status).toBe('quarantined')
    expect(result.watermark).toBe(3)
    expect(result.applied).toBe(0)
    // The final call carried NO facts but a real upTo — the pure watermark
    // advance past poison, stamped by the adapter itself.
    expect(adapter.batches).toEqual([[]])
    expect(adapter.upTos).toEqual([3])
    expect(engine.quarantined('f3').map((q) => q.generation)).toEqual([1, 2, 3])
  })

  it('a NON-typed throw aborts the advance loudly — unknown failure is never poison', async () => {
    const source = scriptedSource(() => range(1, 8))
    const engine = new ReprojectionEngine({ source, batchSize: 4 })
    const adapter = new RecordingAdapter('f4')
    adapter.hardFail.add(5)
    engine.register(adapter)

    await expect(engine.advance('f4', { budgetMs: 10_000 })).rejects.toThrow(/disk exploded at generation 5/)

    expect(adapter.watermark()).toBe(4) // the clean first batch landed; nothing after
    expect(engine.quarantined('f4')).toEqual([]) // no ledger entry for an unknown failure
  })

  it('an adapter re-condemning an already-quarantined generation is refused loudly', async () => {
    const source = scriptedSource(() => range(1, 4))
    const engine = new ReprojectionEngine({ source, batchSize: 4 })
    // A misbehaving adapter: always blames generation 3, even once it is
    // filtered out of its batches.
    const adapter: ProjectionAdapter = {
      family: 'f5',
      watermark: () => null,
      applyBatch: async () => {
        throw new ProjectionApplyError({ generation: 3, cause: new Error('always 3') })
      },
      discard: async () => {}
    }
    engine.register(adapter)

    await expect(engine.advance('f5', { budgetMs: 10_000 })).rejects.toThrow(/ALREADY quarantined/)
    expect(engine.quarantined('f5').map((q) => q.generation)).toEqual([3])
  })

  it('an adapter that never stamps is refused loudly instead of spinning', async () => {
    const source = scriptedSource(() => range(1, 4))
    const engine = new ReprojectionEngine({ source, batchSize: 2 })
    const adapter: ProjectionAdapter = {
      family: 'f6',
      watermark: () => null, // never advances
      applyBatch: async () => {},
      discard: async () => {}
    }
    engine.register(adapter)

    await expect(engine.advance('f6', { budgetMs: 10_000 })).rejects.toThrow(/not stamping/)
  })
})

describe('reprojection engine — (g) discard lands on the losing adapter after a swap', () => {
  it('the OLD adapter is discarded exactly once, after the flip; the winner is never discarded', async () => {
    const source = scriptedSource(() => range(1, 6))
    const engine = new ReprojectionEngine({ source, batchSize: 3 })
    const losing = new RecordingAdapter('g')
    engine.register(losing)
    await engine.advance('g', { budgetMs: 10_000 })
    expect(losing.discarded).toBe(0) // serving adapters are never discarded

    let winner!: RecordingAdapter
    await engine.swap('g', async () => {
      winner = new RecordingAdapter('g')
      winner.onApply = () => {
        // Mid-build the loser still serves and is still intact.
        expect(losing.discarded).toBe(0)
      }
      return winner
    })

    expect(losing.discarded).toBe(1)
    expect(winner.discarded).toBe(0)
    expect(engine.getAdapter('g')).toBe(winner)
  })
})

describe('FactLogSource — the production source enforces the window contract', () => {
  it('delegates to the injected callback and passes clean windows through', async () => {
    const calls: Array<[number, number]> = []
    const source = new FactLogSource(async (from, limit) => {
      calls.push([from, limit])
      return range(from + 1, Math.min(from + limit, 5)).map(fact)
    })
    const facts = await source.scan(2, 2)
    expect(facts.map((f) => f.generation)).toEqual([3, 4])
    expect(calls).toEqual([[2, 2]])
    expect(await source.scan(5, 3)).toEqual([])
  })

  it('refuses out-of-contract callbacks loudly: oversize, non-ascending, at-or-below from', async () => {
    const oversize = new FactLogSource(async () => range(1, 5).map(fact))
    await expect(oversize.scan(0, 2)).rejects.toThrow(/contract violation/)

    const unsorted = new FactLogSource(async () => [fact(3), fact(2)])
    await expect(unsorted.scan(0, 10)).rejects.toThrow(/strictly ascending/)

    const stale = new FactLogSource(async () => [fact(2)])
    await expect(stale.scan(2, 10)).rejects.toThrow(/strictly ascending/)
  })

  it('validates its own window arguments', async () => {
    const source = new FactLogSource(async () => [])
    await expect(source.scan(-1, 5)).rejects.toThrow(/non-negative integer/)
    await expect(source.scan(0, 0)).rejects.toThrow(/positive integer/)
  })
})