open-brainy/tests/unit/aggregation/aggregation-provider-rebuild.test.ts

/**
 * @module tests/unit/aggregation/aggregation-provider-rebuild
 * @description Pins for SELF-ENGINE-LIFECYCLE-SPRINT asks (c) + (d):
 * (c) the native provider's parallel `rebuildAggregate` — on the provider
 *     contract since 8.x but NEVER invoked (the JS walk streamed per-entity
 *     FFI calls instead) — is now the backfill walk's preferred door;
 * (d) a write-path hook that cannot see its entity (before-image-less
 *     delete) flags an exact rescan LOUDLY instead of silently skipping the
 *     decrement (the skip let counts drift upward forever).
 */
import { describe, it, expect, vi } from 'vitest'
import { AggregationIndex } from '../../../src/aggregation/AggregationIndex.js'
import { NounType } from '../../../src/types/graphTypes.js'
import type { AggregationProvider, AggregateGroupState } from '../../../src/types/brainy.types.js'

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

/** Minimal in-memory storage double for the index's persistence surface. */
function memStorage() {
  const store = new Map<string, unknown>()
  return {
    saveMetadata: async (k: string, v: unknown) => void store.set(k, v),
    getMetadata: async (k: string) => store.get(k) ?? null
  } as never
}

function providerDouble(): AggregationProvider & { rebuildAggregate: ReturnType<typeof vi.fn> } {
  return {
    defineAggregate: vi.fn(),
    removeAggregate: vi.fn(),
    incrementalUpdate: vi.fn(() => []),
    computeGroupKey: vi.fn(() => ({})),
    rebuildAggregate: vi.fn((): Map<string, AggregateGroupState> => {
      return new Map([
        [
          'system.subtype=invoice',
          {
            groupKey: { 'system.subtype': 'invoice' },
            metrics: { count: { sum: 0, count: 2, min: Infinity, max: -Infinity, m2: 0 } }
          } as AggregateGroupState
        ]
      ])
    }),
    queryAggregate: vi.fn(() => [])
  } as never
}

describe('ask (c) — the native parallel rebuild is invoked, never dead code', () => {
  it('rebuildWithProvider hands SOURCE-MATCHED entities to the provider once and swaps state in', () => {
    const provider = providerDouble()
    const index = new AggregationIndex(memStorage(), provider)
    index.defineAggregate(DEF)

    expect(index.hasProviderRebuild()).toBe(true)

    const entities = [
      { type: NounType.Document, subtype: 'invoice', metadata: {} },
      { type: NounType.Document, subtype: 'invoice', metadata: {} },
      // Source-filter mismatch: a different noun type must be filtered OUT
      // before the provider sees the batch.
      { type: NounType.Person, subtype: 'invoice', metadata: {} }
    ]
    const handled = index.rebuildWithProvider(DEF.name, entities)

    expect(handled).toBe(true)
    expect(provider.rebuildAggregate).toHaveBeenCalledTimes(1)
    const [defArg, entArg] = provider.rebuildAggregate.mock.calls[0]
    expect(defArg.name).toBe(DEF.name)
    expect(entArg).toHaveLength(2)

    // The rebuilt state serves — and the aggregate is no longer pending.
    expect(index.getPendingBackfills()).not.toContain(DEF.name)
  })

  it('returns false without a provider rebuild — the caller streams the JS walk', () => {
    const index = new AggregationIndex(memStorage())
    index.defineAggregate(DEF)
    expect(index.hasProviderRebuild()).toBe(false)
    expect(index.rebuildWithProvider(DEF.name, [])).toBe(false)
  })
})

describe('ask (d) — the before-image-less delete is LOUD, never a silent skip', () => {
  it('flagAllForRescan puts every defined aggregate back on the backfill list', () => {
    const index = new AggregationIndex(memStorage())
    index.defineAggregate(DEF)
    index.defineAggregate({ ...DEF, name: 'second' })
    // Simulate settled state: nothing pending.
    for (const n of index.getPendingBackfills()) {
      index.beginBackfill(n)
      index.finishBackfill(n)
    }
    expect(index.getPendingBackfills()).toEqual([])

    index.flagAllForRescan('delete of X carried no before-image metadata')

    expect(index.getPendingBackfills().sort()).toEqual(['by_subtype', 'second'])
  })
})

describe('reconcileEntity — the exact delta algebra at the catch-up boundary', () => {
  it('before-only removes, after-only adds, both reconciles a group move', () => {
    const index = new AggregationIndex(memStorage())
    index.defineAggregate(DEF)
    for (const n of index.getPendingBackfills()) {
      index.beginBackfill(n)
      index.finishBackfill(n)
    }
    const doc = (subtype: string) => ({ type: NounType.Document, subtype, metadata: {} })

    // Pre-window state, applied through the LIVE hooks (as adoption would
    // have counted it): c and seed exist as drafts, x1 as an invoice.
    index.onEntityAdded('c', doc('draft'))
    index.onEntityAdded('seed', doc('draft'))
    index.onEntityAdded('x1', doc('invoice'))

    // The window's reconciliation: two adds, one group move, one delete.
    index.reconcileEntity(DEF.name, 'a', null, doc('invoice'))
    index.reconcileEntity(DEF.name, 'b', null, doc('invoice'))
    index.reconcileEntity(DEF.name, 'c', doc('draft'), doc('invoice'))
    index.reconcileEntity(DEF.name, 'seed', doc('draft'), null)

    const rows = index.queryAggregate({ name: DEF.name })
    const count = (st: string) =>
      Number(rows.find(r => r.groupKey['system.subtype'] === st)?.metrics.count ?? 0)
    expect(count('invoice')).toBe(4) // x1 + a + b + moved c
    expect(count('draft')).toBe(0) // c moved out, seed deleted
  })
})