open-brainy/src/reprojection/factLogSource.ts

/**
 * @module reprojection/factLogSource
 * @description The production {@link FactSource}: adapts the database's
 * committed-fact scan to the reprojection engine's `scan(from, limit)`
 * window contract.
 *
 * DEPENDENCY-CLEAN BY DESIGN: this module never imports the database class.
 * It wraps a host-owned scan callback `(from, limit) => Promise<CommitFact[]>`
 * injected at construction, so the host wires itself in one line — either by
 * handing {@link FactLogSource} a callback built on its own scan API, or via
 * {@link factSourceFromHost}, which builds that callback from any object
 * structurally exposing `scanFacts` (the batch-handle shape the fact log
 * serves).
 *
 * CONTRACT ENFORCEMENT — loud, never quiet: every `scan` return is checked
 * (≤ limit facts, strictly ascending generations, all strictly above `from`);
 * a violating callback throws instead of silently corrupting a fold. A host
 * with NO fact log throws too — reporting "caught up" against an unscannable
 * store would be a silent lie.
 */

import type { CommitFact } from '../db/factLog.js'
import type { FactSource } from './reprojectionEngine.js'

/**
 * The host-owned scan callback: return up to `limit` committed facts with
 * generation strictly greater than `from`, in ascending generation order;
 * empty means caught up to the head as of the call.
 */
export type FactScanCallback = (from: number, limit: number) => Promise<CommitFact[]>

/**
 * The minimal structural surface of a fact-scanning host — matches the
 * database's `scanFacts` shape without importing it. `scanFacts` returns a
 * handle whose `batches()` yields ordered, non-empty fact batches, or `null`
 * when the store hosts no fact log.
 */
export interface FactScanHost {
  scanFacts(options?: { fromGeneration?: number; batchSize?: number }): {
    batches: () => AsyncGenerator<{ facts: CommitFact[] }>
  } | null
}

/**
 * The production {@link FactSource}: wraps an injected scan callback and
 * enforces the window contract on every return.
 *
 * COST NOTE: each `scan` call is stateless (a fresh window above the caller's
 * watermark), which is exactly what resumable, crash-tolerant folds need —
 * at the price of the host re-opening its scan per call. Fine for
 * budget-capped maintenance; not a hot-path read primitive.
 */
export class FactLogSource implements FactSource {
  private readonly scanCallback: FactScanCallback

  /** @param scanCallback - The host-owned scan (see {@link FactScanCallback}). */
  constructor(scanCallback: FactScanCallback) {
    if (typeof scanCallback !== 'function') {
      throw new Error('FactLogSource: a scan callback (from, limit) => Promise<CommitFact[]> is required')
    }
    this.scanCallback = scanCallback
  }

  /**
   * Fetch up to `limit` committed facts strictly above generation `from`,
   * verifying the callback honored the window contract.
   * @param from - Exclusive lower bound generation (≥ 0 integer).
   * @param limit - Maximum facts to return (≥ 1 integer).
   */
  async scan(from: number, limit: number): Promise<CommitFact[]> {
    if (!Number.isInteger(from) || from < 0) {
      throw new Error(`FactLogSource.scan: 'from' must be a non-negative integer (got ${from})`)
    }
    if (!Number.isInteger(limit) || limit < 1) {
      throw new Error(`FactLogSource.scan: 'limit' must be a positive integer (got ${limit})`)
    }
    const facts = await this.scanCallback(from, limit)
    if (!Array.isArray(facts)) {
      throw new Error('FactLogSource.scan: the scan callback must resolve to an array of facts')
    }
    if (facts.length > limit) {
      throw new Error(
        `FactLogSource.scan: the scan callback returned ${facts.length} facts for limit ${limit} — ` +
          `contract violation; refusing to fold an oversized window`
      )
    }
    let prev = from
    for (const fact of facts) {
      const g = fact?.generation
      if (typeof g !== 'number' || !Number.isFinite(g) || g <= prev) {
        throw new Error(
          `FactLogSource.scan: the scan callback violated the window contract — generation ` +
            `${String(g)} is not strictly ascending above ${prev} (from=${from}); refusing to fold`
        )
      }
      prev = g
    }
    return facts
  }
}

/**
 * Build the production source from any host structurally exposing
 * `scanFacts` — the one-line wiring for the database side:
 *
 * ```ts
 * const source = factSourceFromHost(brain)
 * ```
 *
 * Each `scan(from, limit)` opens `scanFacts({ fromGeneration: from + 1,
 * batchSize: limit })` (the engine's `from` is exclusive; `scanFacts` bounds
 * are inclusive) and returns the FIRST batch, closing the handle — short
 * batches at segment boundaries are legal under the source contract (only
 * EMPTY means caught up). A host with no fact log throws loudly.
 *
 * @param host - Any object with the `scanFacts` batch-handle shape.
 */
export function factSourceFromHost(host: FactScanHost): FactLogSource {
  if (!host || typeof host.scanFacts !== 'function') {
    throw new Error('factSourceFromHost: the host must expose scanFacts(options)')
  }
  return new FactLogSource(async (from, limit) => {
    const scan = host.scanFacts({ fromGeneration: from + 1, batchSize: limit })
    if (scan === null) {
      throw new Error(
        'reprojection: this store hosts no fact log — reprojection folds committed facts, ' +
          'and reporting a caught-up fold against an unscannable store would be a silent lie'
      )
    }
    const iterator = scan.batches()
    try {
      const first = await iterator.next()
      return first.done ? [] : first.value.facts
    } finally {
      // Close the abandoned generator so its cleanup (timers) runs.
      if (typeof iterator.return === 'function') {
        await iterator.return(undefined)
      }
    }
  })
}