open-brainy/tests/unit/utils/metadataIndex-watermark.test.ts

/**
 * @module tests/unit/utils/metadataIndex-watermark
 * @description Watermark-stamp pins for the metadata projection.
 *
 * THE LAW under test: every persisted projection artifact carries a stamp
 * asserting "this state reflects every committed generation ≤ W and nothing
 * above W, atomically" — written AFTER every byte it certifies is durable —
 * and at load the owner computes the three-way verdict:
 * stamped==committed → 'adopt' · stamped<committed → 'catchup' (gap
 * reported) · stamped>committed OR unstamped → 'rescan', LOUDLY.
 * Same rule, same verdict names as the shipped aggregation machinery
 * (AggregationIndex.stateAdoptionVerdict).
 *
 * The verdict is computed at init and consumed via
 * {@link MetadataIndexManager.applyWatermarkCatchup} — the coordinator
 * (`Brainy.performInit`) calls it right after `init()`, with an open fact
 * scan when the verdict is `'catchup'`. This file pins both halves: the
 * verdict computation (above) and the fold/no-op/demotion behavior below.
 */
import { describe, it, expect, vi, afterEach } from 'vitest'
import { v4 as uuidv4 } from 'uuid'
import {
  MetadataIndexManager,
  METADATA_INDEX_STAMP_KEY
} from '../../../src/utils/metadataIndex.js'
import { MemoryStorage } from '../../../src/storage/adapters/memoryStorage.js'
import { prodLog } from '../../../src/utils/logger.js'
import type { CommitFact, FactScanBatch, FactScanHandle } from '../../../src/db/factLog.js'

/** A fact scan handle over an in-memory list of facts — batches them one
 *  fact at a time (batch size is irrelevant to the fold, which reads
 *  `batch.facts` only). */
function fakeScan(facts: CommitFact[]): FactScanHandle {
  return {
    headGeneration: facts.length > 0 ? facts[facts.length - 1].generation : 0,
    segmentCount: 1,
    approxFactCount: facts.length,
    async *batches(): AsyncGenerator<FactScanBatch> {
      for (const fact of facts) {
        yield {
          facts: [fact],
          firstGeneration: fact.generation,
          lastGeneration: fact.generation,
          factCount: 1,
          byteSize: 0,
          segmentId: 'fake'
        }
      }
    },
    summary: () => ({ factsYielded: facts.length, segmentsRead: 1 })
  }
}

/** One noun after-image fact — the flat-record shape (no nested `metadata`
 *  key), matching this file's existing `writeArtifact` convention. */
function nounAdd(generation: number, id: string, metadata: Record<string, unknown>): CommitFact {
  return {
    generation,
    timestamp: Date.now(),
    ops: [{ kind: 'noun', id, record: { metadata, vector: null } }]
  }
}

/** One noun tombstone fact. */
function nounDelete(generation: number, id: string): CommitFact {
  return { generation, timestamp: Date.now(), ops: [{ kind: 'noun', id, record: null }] }
}

/** Fresh storage with a controllable committed generation. */
async function makeStorage(committed: number | null): Promise<MemoryStorage> {
  const storage = new MemoryStorage()
  await storage.init()
  if (committed !== null) {
    vi.spyOn(storage, 'committedGeneration').mockReturnValue(committed)
  }
  return storage
}

/** Set (or reset) the mocked committed generation on an existing storage. */
function setCommitted(storage: MemoryStorage, committed: number): void {
  vi.spyOn(storage, 'committedGeneration').mockReturnValue(committed)
}

/** Session 1: index a field, optionally stamp, flush — the durable artifact. */
async function writeArtifact(
  storage: MemoryStorage,
  stamp: number | null
): Promise<void> {
  const index = new MetadataIndexManager(storage)
  await index.init()
  await index.addToIndex(uuidv4(), { status: 'active', role: 'admin' })
  if (stamp !== null) index.stampWatermark(stamp)
  await index.flush()
}

/** Session 2: reopen on the same storage and return the loaded manager. */
async function reopen(storage: MemoryStorage): Promise<MetadataIndexManager> {
  const index = new MetadataIndexManager(storage)
  await index.init()
  return index
}

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

describe('metadata index — watermark stamp + three-way load verdict', () => {
  it("save-with-stamp then reopen at the same committed generation → 'adopt', zero-work verdict", async () => {
    const storage = await makeStorage(5)
    await writeArtifact(storage, 5)

    const index = await reopen(storage)
    expect(index.watermarkVerdict()).toBe('adopt')
    expect(index.watermark()).toBe(5)
    expect(index.watermarkGap()).toBeNull()
  })

  it("stamp BEHIND the committed generation → 'catchup' with the exact gap reported", async () => {
    const storage = await makeStorage(5)
    await writeArtifact(storage, 5)

    // Later commits landed after the last stamped flush (unclean exit shape).
    setCommitted(storage, 8)

    const index = await reopen(storage)
    expect(index.watermarkVerdict()).toBe('catchup')
    expect(index.watermark()).toBe(5)
    expect(index.watermarkGap()).toEqual({ from: 5, to: 8 })
  })

  it("stamp ABOVE the committed generation → 'rescan', said out loud", async () => {
    const storage = await makeStorage(9)
    await writeArtifact(storage, 9)

    // A truncated log on a copied store pulled the watermark back.
    setCommitted(storage, 4)

    const warnSpy = vi.spyOn(prodLog, 'warn')
    const index = await reopen(storage)
    expect(index.watermarkVerdict()).toBe('rescan')
    expect(index.watermarkGap()).toBeNull()
    const said = warnSpy.mock.calls.map(c => String(c[0])).join('\n')
    expect(said).toContain('RESCAN')
    expect(said).toContain('ABOVE')
  })

  it("legacy unstamped artifact on a stamped store → 'rescan', LOUD — never a silent adopt", async () => {
    const storage = await makeStorage(3)
    await writeArtifact(storage, null) // pre-stamp brain: data flushed, no stamp

    expect(await storage.getMetadata(METADATA_INDEX_STAMP_KEY)).toBeNull()

    const warnSpy = vi.spyOn(prodLog, 'warn')
    const index = await reopen(storage)
    expect(index.watermarkVerdict()).toBe('rescan')
    expect(index.watermark()).toBeNull()
    const said = warnSpy.mock.calls.map(c => String(c[0])).join('\n')
    expect(said).toContain('RESCAN')
    expect(said).toContain('unstamped')
  })

  it("a store with no committed-generation capability keeps pre-stamp behavior → 'adopt'", async () => {
    const storage = await makeStorage(null) // committedGeneration() → null
    await writeArtifact(storage, null)

    const index = await reopen(storage)
    expect(index.watermarkVerdict()).toBe('adopt')
    expect(index.watermark()).toBeNull()
  })

  it('STAMP-AFTER-DATA: the stamp is the last saveMetadata of the flush, after registry and field indexes', async () => {
    const storage = await makeStorage(2)
    const index = new MetadataIndexManager(storage)
    await index.init()
    await index.addToIndex(uuidv4(), { status: 'active' })

    const keys: string[] = []
    const originalSave = storage.saveMetadata.bind(storage)
    vi.spyOn(storage, 'saveMetadata').mockImplementation(async (id, metadata) => {
      keys.push(id)
      return originalSave(id, metadata)
    })

    index.stampWatermark(2)
    await index.flush()

    const stampAt = keys.indexOf(METADATA_INDEX_STAMP_KEY)
    expect(stampAt, 'stamp record was written').toBeGreaterThanOrEqual(0)
    expect(stampAt, 'stamp is the FINAL metadata write of the flush').toBe(keys.length - 1)
    const registryAt = keys.indexOf('__metadata_field_registry__')
    expect(registryAt, 'field registry written during this flush').toBeGreaterThanOrEqual(0)
    expect(registryAt).toBeLessThan(stampAt)

    // The persisted stamp record carries the required shape.
    const record = (await storage.getMetadata(METADATA_INDEX_STAMP_KEY)) as {
      watermark: number
      formatVersion: number
      stampedAt: number
    }
    expect(record.watermark).toBe(2)
    expect(record.formatVersion).toBe(1)
    expect(typeof record.stampedAt).toBe('number')
  })

  it('a flush WITHOUT a pending stamp writes no stamp record (no phantom certification)', async () => {
    const storage = await makeStorage(2)
    const index = new MetadataIndexManager(storage)
    await index.init()
    await index.addToIndex(uuidv4(), { status: 'active' })
    await index.flush()

    expect(await storage.getMetadata(METADATA_INDEX_STAMP_KEY)).toBeNull()
  })
})

describe('metadata index — applyWatermarkCatchup (the coordinator door)', () => {
  it("an 'adopt' verdict performs zero index writes", async () => {
    const storage = await makeStorage(5)
    await writeArtifact(storage, 5)
    const index = await reopen(storage)
    expect(index.watermarkVerdict()).toBe('adopt')

    const addSpy = vi.spyOn(index, 'addToIndex')
    const removeSpy = vi.spyOn(index, 'removeFromIndex')

    const result = await index.applyWatermarkCatchup(null)

    expect(result).toEqual({ action: 'noop' })
    expect(addSpy).not.toHaveBeenCalled()
    expect(removeSpy).not.toHaveBeenCalled()
  })

  it('a catchup window folding an add, an update (same id twice), and a delete → the index serves exactly the final state', async () => {
    const storage = await makeStorage(5)

    // Session 1: two pre-existing entities, stamped at generation 5.
    const survivorId = uuidv4()
    const deletedId = uuidv4()
    {
      const index = new MetadataIndexManager(storage)
      await index.init()
      await index.addToIndex(survivorId, { status: 'active' })
      await index.addToIndex(deletedId, { status: 'active' })
      index.stampWatermark(5)
      await index.flush()
    }

    // The store advanced to generation 8 without another metadata flush —
    // the exact shape a crash-then-adopt-reopen leaves behind.
    setCommitted(storage, 8)

    const index = await reopen(storage)
    expect(index.watermarkVerdict()).toBe('catchup')
    expect(index.watermarkGap()).toEqual({ from: 5, to: 8 })

    const addedId = uuidv4()
    const scan = fakeScan([
      nounAdd(6, addedId, { status: 'new' }),       // add
      nounAdd(7, addedId, { status: 'updated' }),   // update — same id twice
      nounDelete(8, deletedId)                       // delete
    ])

    const result = await index.applyWatermarkCatchup(scan)

    expect(result.action).toBe('caught-up')
    expect(result.window).toEqual({ from: 5, to: 8 })
    expect(result.factsApplied).toBe(3)
    expect(result.nounsApplied).toBe(3)
    expect(result.verbsApplied).toBe(0)

    // Final state: the added/updated id serves ONLY its final value...
    expect(await index.getIds('status', 'updated')).toEqual([addedId])
    expect(await index.getIds('status', 'new')).toEqual([]) // stale value gone
    // ...the deleted id is gone...
    expect(await index.getIds('status', 'active')).toEqual([survivorId])
    // ...and the untouched survivor is unaffected.
    expect(await index.getIds('status', 'active')).toContain(survivorId)

    // The window is certified: watermark stamped at `to`, and a fresh
    // reopen now verdicts 'adopt'.
    expect(index.watermark()).toBe(8)
    const reopened = await reopen(storage)
    expect(reopened.watermarkVerdict()).toBe('adopt')
  })

  it("a 'rescan' verdict runs the existing rebuild path instead of folding", async () => {
    const storage = await makeStorage(9)
    await writeArtifact(storage, 9)
    setCommitted(storage, 4) // a truncated log pulled the watermark back — stamp ABOVE committed → rescan

    const index = await reopen(storage)
    expect(index.watermarkVerdict()).toBe('rescan')

    const rebuildSpy = vi.spyOn(index, 'rebuild')
    const result = await index.applyWatermarkCatchup(null)

    expect(result.action).toBe('rescan')
    expect(result.reason).toBeTruthy()
    expect(rebuildSpy).toHaveBeenCalledTimes(1)
  })

  it("a 'catchup' verdict with no fact log available demotes to rebuild, narrated", async () => {
    const storage = await makeStorage(5)
    await writeArtifact(storage, 5)
    setCommitted(storage, 8)

    const index = await reopen(storage)
    expect(index.watermarkVerdict()).toBe('catchup')

    const rebuildSpy = vi.spyOn(index, 'rebuild')
    const result = await index.applyWatermarkCatchup(null) // no scan — no fact log

    expect(result.action).toBe('rescan')
    expect(result.reason).toContain('no fact log')
    expect(rebuildSpy).toHaveBeenCalledTimes(1)
  })
})