open-brainy/tests/integration/aggregation-lifecycle-catchup.test.ts

/**
 * @module tests/integration/aggregation-lifecycle-catchup
 * @description THE AGGREGATION LIFECYCLE PINS (SELF-ENGINE-LIFECYCLE-SPRINT /
 * BRAINY-PROD-LATENCY-TRIAD asks (a)+(b)). The production disease: the
 * aggregation stamp persisted ONLY at close(), so a long-lived writer that
 * flushes but never closes left its stamp behind after every write window —
 * and the exact-match adoption rule then forced a WHOLE-STORE backfill walk
 * (per-entity work, measured >60s and door-starving on a 9k-row production
 * brain) on the first stats call after any unclean exit.
 *
 * The cures pinned here:
 * (a) `brain.flush()` persists aggregation state, stamped at the committed
 *     generation — the stamp tracks every flush, not just close().
 * (b) BEHIND-stamp state is ADOPTED and reconciled INCREMENTALLY over its
 *     exact missing window (fact-log affected ids + time-travel before/after
 *     reads) — the full walk never runs for an unclean exit. Pinned by call
 *     shape (the walk spy), not by latency.
 */
import { describe, it, expect, afterEach, vi } from 'vitest'
import { mkdtempSync, rmSync } from 'node:fs'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { Brainy } from '../../src/index.js'
import { NounType } from '../../src/types/graphTypes.js'

const AGG = {
  name: 'by_subtype',
  source: { type: NounType.Document },
  groupBy: ['system.subtype'] as string[],
  metrics: { count: { op: 'count' as const } }
}

const dirs: string[] = []
const brains: Brainy[] = []

async function open(dir: string): Promise<Brainy> {
  const b = new Brainy({ storage: { type: 'filesystem', path: dir }, requireSubtype: false })
  await b.init()
  brains.push(b)
  return b
}

function countFor(results: Array<{ groupKey: Record<string, unknown>; metrics: Record<string, unknown> }>, subtype: string): number {
  const row = results.find(r => r.groupKey['system.subtype'] === subtype)
  return row ? Number(row.metrics.count) : 0
}

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

describe('aggregation lifecycle — flush stamps, behind-stamp catches up incrementally', () => {
  it('(a) brain.flush() persists aggregation state stamped at the committed generation', async () => {
    const dir = mkdtempSync(join(tmpdir(), 'brainy-agg-flush-'))
    dirs.push(dir)
    const brain = await open(dir)
    brain.defineAggregate(AGG)
    await brain.add({ data: 'a', type: NounType.Document, subtype: 'invoice', metadata: {} })
    await brain.add({ data: 'b', type: NounType.Document, subtype: 'invoice', metadata: {} })
    await brain.queryAggregate(AGG.name) // settle backfill-on-define

    await brain.flush()

    const internals = brain as unknown as {
      storage: {
        getMetadata(k: string): Promise<{ sourceGeneration?: number } | null>
        committedGeneration?(): number
      }
    }
    const persisted = await internals.storage.getMetadata('__aggregation_state_by_subtype__')
    expect(persisted, 'state persisted by flush(), not only close()').toBeTruthy()
    expect(
      persisted!.sourceGeneration,
      'stamp equals the committed generation at flush time'
    ).toBe(internals.storage.committedGeneration?.())
  })

  it('(b) an unclean exit reconciles incrementally — exact counts, ZERO full-store walks', async () => {
    const dir = mkdtempSync(join(tmpdir(), 'brainy-agg-catchup-'))
    dirs.push(dir)

    // Session 1: define + write + flush (stamps at G), then MORE writes of
    // every kind (add / update-that-moves-groups / delete) and a clean close
    // — but we then REWIND the persisted aggregation artifact to its at-G
    // bytes, which is byte-for-byte the unclean-exit state: stamp G, store
    // committed at G+k.
    let brain = await open(dir)
    brain.defineAggregate(AGG)
    await brain.add({ data: 'a', type: NounType.Document, subtype: 'invoice', metadata: {} })
    await brain.add({ data: 'b', type: NounType.Document, subtype: 'invoice', metadata: {} })
    const moving = await brain.add({ data: 'c', type: NounType.Document, subtype: 'draft', metadata: {} })
    const doomed = await brain.add({ data: 'd', type: NounType.Document, subtype: 'draft', metadata: {} })
    await brain.queryAggregate(AGG.name)
    await brain.flush()

    const internals = brain as unknown as {
      storage: {
        getMetadata(k: string): Promise<Record<string, unknown> | null>
        saveMetadata(k: string, v: Record<string, unknown>): Promise<void>
      }
    }
    const stateAtG = JSON.parse(
      JSON.stringify(await internals.storage.getMetadata('__aggregation_state_by_subtype__'))
    )

    // The missing window: one add, one group-moving update, one delete.
    await brain.add({ data: 'e', type: NounType.Document, subtype: 'invoice', metadata: {} })
    await brain.update({ id: moving, subtype: 'invoice' })
    await brain.remove(doomed)
    await brain.close()
    brains.pop()

    // Rewind the aggregation artifact to the at-G bytes (the unclean exit).
    {
      const reopenForRewind = await open(dir)
      const rw = reopenForRewind as unknown as typeof internals
      await rw.storage.saveMetadata('__aggregation_state_by_subtype__', stateAtG)
      await reopenForRewind.close()
      brains.pop()
    }

    // Session 2: reopen — adoption must see BEHIND and reconcile, never walk.
    brain = await open(dir)
    brain.defineAggregate(AGG)
    const walkSpy = vi.spyOn(
      brain as unknown as { runAggregationBackfillWalk(): Promise<void> },
      'runAggregationBackfillWalk'
    )

    const results = await brain.queryAggregate(AGG.name)

    // Ground truth after the window: invoice = a,b,e + moved c = 4; draft = 0
    // (c moved out, d deleted).
    expect(countFor(results as never, 'invoice'), 'invoice count exact after catch-up').toBe(4)
    expect(countFor(results as never, 'draft'), 'draft count exact after catch-up').toBe(0)

    // THE CALL-SHAPE PIN: the whole-store walk never ran.
    expect(walkSpy, 'full backfill walk must not run for a behind-stamp reopen').not.toHaveBeenCalled()

    vi.restoreAllMocks()
  }, 120000)
})