brainy/src/db/generationStore.ts

/**
 * @module db/generationStore
 * @description The generational record layer behind Brainy 8.0's MVCC Db API.
 *
 * One `GenerationStore` is attached to every writable `Brainy` instance. It
 * owns:
 *
 * - the **monotonic generation counter** (`_system/generation.json`),
 *   reserved per `transact()` commit and bumped per single-operation write
 *   batch via the storage-layer hook;
 * - the **commit protocol** for `brain.transact()`: capture before-images →
 *   write the generation delta → fsync → execute the operation batch →
 *   atomically rename `_system/manifest.json` (the rename IS the commit
 *   point) → append the tx-log line;
 * - **crash recovery**: on open, any `_generations/<N>` directory with
 *   `N > manifest.generation` is an uncommitted transaction — its
 *   before-images are restored to the canonical entity paths and the
 *   directory is removed, so a crash anywhere before the manifest rename
 *   leaves the store byte-identical to its pre-transaction state;
 * - **pin/release refcounting** for live `Db` values, which is what makes
 *   `compactHistory()` safe: a generation record-set is reclaimed only when
 *   no live pin could ever need it;
 * - **point-in-time resolution**: the state of id X at pinned generation G
 *   is the before-image stored by the *first* committed generation after G
 *   that touched X — or the live storage state when nothing after G touched
 *   it (so current-generation reads stay on the existing fast paths,
 *   untouched).
 *
 * Durability and atomicity guarantees, failure modes, and the intellectual
 * lineage (Datomic database-as-value, LMDB reader pins, LSM immutable
 * segments) are specified in `docs/ADR-001-generational-mvcc.md`.
 */

import { prodLog } from '../utils/logger.js'
import { GenerationCompactedError, GenerationConflictError } from './errors.js'
import type {
  ChangedIds,
  CompactHistoryOptions,
  CompactHistoryResult,
  GenerationDelta,
  GenerationManifest,
  GenerationRecord,
  GenerationStorage,
  TxLogEntry
} from './types.js'

/** Storage-root-relative path of the persisted generation counter. */
export const GENERATION_COUNTER_PATH = '_system/generation.json'
/** Storage-root-relative path of the commit manifest. */
export const MANIFEST_PATH = '_system/manifest.json'
/** Storage-root-relative prefix of the per-generation record directories. */
export const GENERATIONS_PREFIX = '_generations'

/**
 * @description Phases of the {@link GenerationStore.commitTransaction} commit
 * protocol at which a test-only fault injector can simulate a process crash.
 * The phases map 1:1 onto the protocol steps documented on
 * `commitTransaction`:
 *
 * - `'after-staging'` — before-images + the `tx.json` delta are written and
 *   fsynced; the operation batch has NOT executed yet.
 * - `'after-execute'` — the operation batch has been applied to canonical
 *   storage and indexes; the manifest still points at the prior generation.
 * - `'before-manifest-rename'` — the batch has executed and the counter is
 *   persisted; the atomic manifest rename — the commit point — has NOT
 *   happened. A crash here is the canonical "fully staged, never committed"
 *   case that recovery must roll back byte-identically.
 * - `'after-manifest-rename'` — the manifest rename landed (the transaction
 *   IS committed); the tx-log append has NOT happened yet. A crash here must
 *   keep the transaction (the tx-log is advisory metadata, not the source of
 *   commit truth).
 */
export type CommitFaultPhase =
  | 'after-staging'
  | 'after-execute'
  | 'before-manifest-rename'
  | 'after-manifest-rename'

/**
 * @description Identifies which ids a transaction touches, split by kind.
 */
export interface TouchedIds {
  /** Entity ids the transaction writes or deletes. */
  nouns: string[]
  /** Relationship ids the transaction writes or deletes. */
  verbs: string[]
}

/**
 * @description Result of {@link GenerationStore.open} — how many uncommitted
 * generations crash recovery rolled back (a non-zero value tells `Brainy` to
 * force an index reconciliation, since derived index state may reference the
 * rolled-back writes).
 */
export interface GenerationStoreOpenResult {
  /** Count of uncommitted generation directories rolled back and removed. */
  rolledBackGenerations: number
}

/**
 * @description The generational MVCC record layer. See the module
 * documentation for the full protocol; every public method documents its
 * own contract.
 */
export class GenerationStore {
  private readonly storage: GenerationStorage

  /** Latest reserved/observed generation (≥ {@link committed}). */
  private counter = 0
  /** Committed-transaction watermark (manifest generation). */
  private committed = 0
  /** Compaction horizon — record-sets ≤ this are reclaimed. */
  private horizonGen = 0

  /** Sorted list of committed generations whose record dirs exist. */
  private committedGens: number[] = []
  /** Delta cache, keyed by generation (lazily loaded from `tx.json`). */
  private readonly deltaCache = new Map<
    number,
    { nouns: Set<string>; verbs: Set<string>; timestamp: number }
  >()

  /** Live pin refcounts, keyed by pinned generation. */
  private readonly pins = new Map<number, number>()

  /** Serializes transact commits, compaction, and counter persistence. */
  private mutexTail: Promise<unknown> = Promise.resolve()

  /** True while a transact batch is executing (suppresses the bump hook). */
  private inTransact = false

  /** Coalescing state for lazy persistence of single-op counter bumps. */
  private counterPersistScheduled = false

  /**
   * Test-only fault injector for the commit protocol (see
   * {@link GenerationStore.setCommitFaultInjector}). `undefined` in production.
   */
  private commitFaultInjector?: (phase: CommitFaultPhase) => void

  private opened = false

  /**
   * @param storage - The storage adapter, narrowed to the
   *   {@link GenerationStorage} contract (`BaseStorage` implements it).
   */
  constructor(storage: GenerationStorage) {
    this.storage = storage
  }

  // ==========================================================================
  // Lifecycle
  // ==========================================================================

  /**
   * @description Load the persisted counter + manifest and run crash
   * recovery. Must be called once, before indexes are built, because
   * recovery may rewrite canonical entity files (restoring before-images of
   * an uncommitted transaction).
   *
   * @param options.readOnly - When true (reader-mode brains), recovery is
   *   skipped: readers never write. An un-repaired crashed directory is
   *   repaired by the next *writer* open.
   * @returns How many uncommitted generations were rolled back.
   */
  async open(options?: { readOnly?: boolean }): Promise<GenerationStoreOpenResult> {
    const counterFile = (await this.storage.readRawObject(GENERATION_COUNTER_PATH)) as
      | { generation?: number }
      | null
    const manifest = (await this.storage.readRawObject(MANIFEST_PATH)) as GenerationManifest | null

    this.committed = manifest?.generation ?? 0
    this.horizonGen = manifest?.horizon ?? 0
    this.counter = Math.max(counterFile?.generation ?? 0, this.committed)

    // Discover existing generation record directories.
    const recordPaths = await this.storage.listRawObjects(GENERATIONS_PREFIX)
    const seenGens = new Set<number>()
    for (const p of recordPaths) {
      const gen = parseGenerationFromPath(p)
      if (gen !== null) seenGens.add(gen)
    }

    let rolledBack = 0
    const committedGens: number[] = []
    for (const gen of [...seenGens].sort((a, b) => a - b)) {
      if (gen <= this.committed) {
        committedGens.push(gen)
      } else if (!options?.readOnly) {
        // Uncommitted (crashed) transaction: restore before-images, drop the dir.
        await this.rollBackUncommittedGeneration(gen)
        rolledBack++
      }
      // Reader mode: leave orphan dirs alone; resolution ignores them because
      // committedGens only includes generations ≤ the manifest watermark.
      this.counter = Math.max(this.counter, gen)
    }
    this.committedGens = committedGens

    if (rolledBack > 0) {
      prodLog.warn(
        `[GenerationStore] Crash recovery rolled back ${rolledBack} uncommitted ` +
          `transaction(s); store restored to generation ${this.committed}.`
      )
    }

    this.opened = true

    // Hook single-op write batches so generation() is always meaningful.
    // Suppressed while a transact batch executes (the batch is ONE generation).
    if (!options?.readOnly) {
      this.storage.setGenerationBumpHook(() => this.noteSingleOpWrite())
    }

    return { rolledBackGenerations: rolledBack }
  }

  /**
   * @description Detach the storage hook and persist the counter. Called
   * from `brain.close()`.
   */
  async close(): Promise<void> {
    this.storage.setGenerationBumpHook(undefined)
    await this.persistCounterNow()
  }

  /**
   * @description TEST-ONLY: install (or clear, with `undefined`) a fault
   * injector that is invoked at each {@link CommitFaultPhase} of the commit
   * protocol. A **throwing** injector simulates a process crash at that
   * phase: the abort cleanup (staging-directory removal, generation-counter
   * reservation return) is deliberately skipped — exactly as if the process
   * had died — so crash-consistency tests exercise the REAL recovery path
   * ({@link GenerationStore.open} rolling back the uncommitted generation on
   * the next open). Never installed in production code; the injector exists
   * solely so the durability protocol can be proven, not trusted.
   *
   * @param injector - Callback invoked synchronously at each phase, or
   *   `undefined` to detach.
   */
  setCommitFaultInjector(injector: ((phase: CommitFaultPhase) => void) | undefined): void {
    this.commitFaultInjector = injector
  }

  // ==========================================================================
  // Generation counter
  // ==========================================================================

  /** @returns The store's current generation (latest write watermark). */
  generation(): number {
    return this.counter
  }

  /** @returns The committed-transaction watermark (manifest generation). */
  committedGeneration(): number {
    return this.committed
  }

  /** @returns The compaction horizon (generations ≤ horizon are reclaimed). */
  horizon(): number {
    return this.horizonGen
  }

  /**
   * @description Single-operation write hook (registered with the storage
   * layer in {@link open}). Bumps the in-memory counter and schedules a
   * coalesced persist of `_system/generation.json` — durable artifacts
   * (records, manifests, snapshots) always persist the counter
   * synchronously at their own commit points, so a crash inside the
   * coalescing window can never move the counter backwards relative to
   * anything durable.
   */
  private noteSingleOpWrite(): void {
    if (this.inTransact) return
    this.counter++
    this.schedulePersistCounter()
  }

  /** Schedule one coalesced counter persist for the current write burst. */
  private schedulePersistCounter(): void {
    if (this.counterPersistScheduled) return
    this.counterPersistScheduled = true
    setImmediate(() => {
      this.counterPersistScheduled = false
      void this.withMutex(() => this.persistCounterUnlocked()).catch((err) => {
        prodLog.warn(`[GenerationStore] generation counter persist failed: ${(err as Error).message}`)
      })
    })
  }

  /**
   * @description Persist the generation counter now (atomic tmp+rename).
   * Called from `flush()`, `close()`, snapshotting, and the commit protocol.
   */
  async persistCounterNow(): Promise<void> {
    if (!this.opened) return
    await this.withMutex(() => this.persistCounterUnlocked())
  }

  private async persistCounterUnlocked(): Promise<void> {
    await this.storage.writeRawObject(GENERATION_COUNTER_PATH, {
      generation: this.counter,
      updatedAt: new Date().toISOString()
    })
  }

  // ==========================================================================
  // Pins
  // ==========================================================================

  /**
   * @description Pin a generation (refcounted). Compaction never reclaims a
   * record-set a live pin could need (i.e. any generation directory `N`
   * with `N > G` for some pinned `G`).
   * @param gen - The generation to pin.
   */
  pin(gen: number): void {
    this.pins.set(gen, (this.pins.get(gen) ?? 0) + 1)
  }

  /**
   * @description Release one pin on a generation. Safe to call for a
   * generation with no live pins (no-op with a warning).
   * @param gen - The generation to release.
   */
  release(gen: number): void {
    const count = this.pins.get(gen)
    if (count === undefined) {
      prodLog.warn(`[GenerationStore] release(${gen}) called with no live pin at that generation`)
      return
    }
    if (count <= 1) this.pins.delete(gen)
    else this.pins.set(gen, count - 1)
  }

  /** @returns Total number of live pins across all generations. */
  activePinCount(): number {
    let total = 0
    for (const count of this.pins.values()) total += count
    return total
  }

  /** @returns The smallest pinned generation, or `Infinity` with no pins. */
  private minPinnedGeneration(): number {
    let min = Infinity
    for (const gen of this.pins.keys()) min = Math.min(min, gen)
    return min
  }

  // ==========================================================================
  // Commit protocol
  // ==========================================================================

  /**
   * @description Commit one transaction. The caller (Brainy.transact) plans
   * the batch up front — resolving every touched id, including generated
   * entity/relationship ids and delete cascades — and hands this method the
   * touched-id sets plus an `execute` closure that runs the already-built
   * operation batch through the TransactionManager.
   *
   * Protocol (under the commit mutex):
   * 1. `ifAtGeneration` CAS check (rejects before anything is staged).
   * 2. Reserve generation `N` (counter increment).
   * 3. Capture before-images of every touched id and write them, plus the
   *    `tx.json` delta, under `_generations/N/`; fsync. From this point a
   *    crash anywhere is recoverable to the exact pre-transaction bytes.
   * 4. Run `execute()`. On failure the TransactionManager has already
   *    rolled back its applied operations; the staging directory is removed
   *    and the reservation is returned (when no concurrent bump consumed a
   *    later number), so a failed transaction leaves the generation
   *    unchanged.
   * 5. Persist the counter, then atomically rename the manifest to
   *    generation `N` — **the rename is the commit point** — and fsync it.
   * 6. Append the tx-log line and update in-memory bookkeeping.
   *
   * @param args.touched - Every id the batch writes or deletes.
   * @param args.meta - Transaction metadata for the tx-log.
   * @param args.ifAtGeneration - Whole-store CAS expectation.
   * @param args.execute - Runs the planned operation batch atomically.
   * @returns The committed generation and its commit timestamp.
   * @throws GenerationConflictError when the CAS expectation fails.
   */
  async commitTransaction(args: {
    touched: TouchedIds
    meta?: Record<string, unknown>
    ifAtGeneration?: number
    execute: () => Promise<void>
  }): Promise<{ generation: number; timestamp: number }> {
    return this.withMutex(async () => {
      if (args.ifAtGeneration !== undefined && args.ifAtGeneration !== this.counter) {
        throw new GenerationConflictError(args.ifAtGeneration, this.counter)
      }

      const nouns = [...new Set(args.touched.nouns)]
      const verbs = [...new Set(args.touched.verbs)]
      const gen = ++this.counter
      const dir = `${GENERATIONS_PREFIX}/${gen}`
      const timestamp = Date.now()

      // Test-only crash simulation (see setCommitFaultInjector): a throwing
      // injector must bypass the abort cleanup below, exactly as a real
      // process death would, so recovery-on-open is what restores the store.
      let crashSimulated = false
      const faultPoint = (phase: CommitFaultPhase): void => {
        if (!this.commitFaultInjector) return
        try {
          this.commitFaultInjector(phase)
        } catch (err) {
          crashSimulated = true
          throw err
        }
      }

      try {
        // -- 3. Before-images + delta (the durable undo log) ------------------
        const stagedPaths: string[] = []
        for (const id of nouns) {
          const prev = await this.storage.readNounRaw(id)
          const recordPath = `${dir}/prev/${id}.json`
          const record: GenerationRecord = { kind: 'noun', metadata: prev.metadata, vector: prev.vector }
          await this.storage.writeRawObject(recordPath, record)
          stagedPaths.push(recordPath)
        }
        for (const id of verbs) {
          const prev = await this.storage.readVerbRaw(id)
          const recordPath = `${dir}/prev/${id}.json`
          const record: GenerationRecord = { kind: 'verb', metadata: prev.metadata, vector: prev.vector }
          await this.storage.writeRawObject(recordPath, record)
          stagedPaths.push(recordPath)
        }
        const delta: GenerationDelta = {
          generation: gen,
          timestamp,
          ...(args.meta && { meta: args.meta }),
          nouns,
          verbs
        }
        const deltaPath = `${dir}/tx.json`
        await this.storage.writeRawObject(deltaPath, delta)
        stagedPaths.push(deltaPath)
        await this.storage.syncRawObjects(stagedPaths)
        faultPoint('after-staging')

        // -- 4. Execute the planned batch -------------------------------------
        this.inTransact = true
        try {
          await args.execute()
        } finally {
          this.inTransact = false
        }
        faultPoint('after-execute')

        // -- 5. Counter + manifest rename (COMMIT POINT) ----------------------
        await this.persistCounterUnlocked()
        faultPoint('before-manifest-rename')
        const manifest: GenerationManifest = {
          version: 1,
          generation: gen,
          committedAt: new Date(timestamp).toISOString(),
          horizon: this.horizonGen
        }
        await this.storage.writeRawObject(MANIFEST_PATH, manifest)
        await this.storage.syncRawObjects([MANIFEST_PATH])
        faultPoint('after-manifest-rename')

        // -- 6. Post-commit bookkeeping ---------------------------------------
        this.committed = gen
        this.committedGens.push(gen)
        this.deltaCache.set(gen, { nouns: new Set(nouns), verbs: new Set(verbs), timestamp })
        const logEntry: TxLogEntry = { generation: gen, timestamp, ...(args.meta && { meta: args.meta }) }
        await this.storage.appendTxLogLine(JSON.stringify(logEntry))

        return { generation: gen, timestamp }
      } catch (err) {
        // Simulated crash (test-only fault injector): skip the abort cleanup
        // entirely — a dead process cannot clean up. Recovery on the next
        // open() is what must (and does) restore the store.
        if (crashSimulated) {
          throw err
        }
        // Failed before the manifest rename: nothing is committed. Remove the
        // staging directory; the TransactionManager already restored any
        // applied operation byte-identically.
        try {
          await this.storage.removeRawPrefix(dir)
        } catch (cleanupErr) {
          prodLog.warn(
            `[GenerationStore] failed to remove staging dir ${dir} after aborted transaction: ` +
              `${(cleanupErr as Error).message} (recovery will remove it on next open)`
          )
        }
        // Return the reservation when no concurrent bump consumed a later
        // number, so a failed transaction leaves generation() unchanged.
        if (this.counter === gen) this.counter = gen - 1
        throw err
      }
    })
  }

  // ==========================================================================
  // Point-in-time resolution
  // ==========================================================================

  /**
   * @description Whether any transaction committed after generation `gen`.
   * O(1) — this is the fast-path check that lets a `Db` pinned at the
   * current generation delegate every read to the live brain untouched.
   * @param gen - The pinned generation.
   */
  hasCommittedAfter(gen: number): boolean {
    const last = this.committedGens[this.committedGens.length - 1]
    return last !== undefined && last > gen
  }

  /**
   * @description Resolve the state of an entity or relationship at pinned
   * generation `gen`: the before-image stored by the first committed
   * generation after `gen` that touched the id — or `{ source: 'current' }`
   * when nothing after `gen` touched it (the live storage state *is* the
   * state at `gen`).
   *
   * @param kind - `'noun'` for entities, `'verb'` for relationships.
   * @param id - The id to resolve.
   * @param gen - The pinned generation.
   * @returns `{ source: 'current' }`, `{ source: 'absent' }` (did not exist
   *   at `gen`), or `{ source: 'record', metadata, vector }` with the raw
   *   stored objects as of `gen`.
   */
  async resolveAt(
    kind: 'noun' | 'verb',
    id: string,
    gen: number
  ): Promise<
    | { source: 'current' }
    | { source: 'absent' }
    | { source: 'record'; metadata: any; vector: any | null }
  > {
    for (const candidate of this.committedGens) {
      if (candidate <= gen) continue
      const delta = await this.getDelta(candidate)
      const touched = kind === 'noun' ? delta.nouns : delta.verbs
      if (!touched.has(id)) continue
      const record = (await this.storage.readRawObject(
        `${GENERATIONS_PREFIX}/${candidate}/prev/${id}.json`
      )) as GenerationRecord | null
      if (record === null) {
        // The record-set exists (candidate is committed) but the id's
        // before-image is missing — only possible through external tampering.
        throw new Error(
          `Generation record missing: ${GENERATIONS_PREFIX}/${candidate}/prev/${id}.json ` +
            `(store corrupted or records removed outside compactHistory())`
        )
      }
      if (record.metadata === null && record.vector === null) {
        return { source: 'absent' }
      }
      return { source: 'record', metadata: record.metadata, vector: record.vector }
    }
    return { source: 'current' }
  }

  /**
   * @description The ids touched by committed transactions in the interval
   * `(fromGen, toGen]`. Used by `db.since()` and by historical `find()` to
   * build its overlay of changed entities.
   * @param fromGen - Exclusive lower bound.
   * @param toGen - Inclusive upper bound.
   */
  async changedBetween(fromGen: number, toGen: number): Promise<ChangedIds> {
    const nouns = new Set<string>()
    const verbs = new Set<string>()
    for (const gen of this.committedGens) {
      if (gen <= fromGen || gen > toGen) continue
      const delta = await this.getDelta(gen)
      for (const id of delta.nouns) nouns.add(id)
      for (const id of delta.verbs) verbs.add(id)
    }
    return {
      nouns: [...nouns].sort(),
      verbs: [...verbs].sort(),
      fromGeneration: fromGen,
      toGeneration: toGen
    }
  }

  /**
   * @description Assert that generation `gen` is reachable (not compacted
   * away) and within the committed range, then return it. Used by `asOf()`.
   * Reachability boundary: reclaiming record-set `N` removes the
   * before-images that reads at generations BELOW `N` depend on, so
   * generations `< horizon` are unreachable while the horizon itself remains
   * servable from the record-sets above it.
   * @param gen - The requested generation.
   * @throws GenerationCompactedError when `gen` is below the horizon.
   * @throws RangeError when `gen` is negative or beyond the current counter.
   */
  assertReachable(gen: number): number {
    if (!Number.isInteger(gen) || gen < 0) {
      throw new RangeError(`asOf(): generation must be a non-negative integer (got ${gen})`)
    }
    if (gen > this.counter) {
      throw new RangeError(
        `asOf(): generation ${gen} is in the future (store is at generation ${this.counter})`
      )
    }
    if (gen < this.horizonGen) {
      throw new GenerationCompactedError(gen, this.horizonGen)
    }
    return gen
  }

  /**
   * @description Resolve a wall-clock timestamp to the generation that was
   * committed at-or-before it, via the tx-log. Resolution granularity is
   * transact commits — single-operation writes between commits are not
   * individually addressable (documented 8.0 history granularity).
   * @param timestampMs - Milliseconds since epoch.
   * @returns The resolved generation (`0` when `timestampMs` predates the
   *   first commit) and the matched tx-log entry when one exists.
   */
  async resolveTimestamp(
    timestampMs: number
  ): Promise<{ generation: number; entry: TxLogEntry | null }> {
    const lines = await this.storage.readTxLogLines()
    let best: TxLogEntry | null = null
    for (const line of lines) {
      let entry: TxLogEntry
      try {
        entry = JSON.parse(line) as TxLogEntry
      } catch {
        continue // tolerate a torn trailing line from a crashed append
      }
      if (entry.timestamp <= timestampMs && entry.generation <= this.committed) {
        if (best === null || entry.generation > best.generation) best = entry
      }
    }
    return { generation: best?.generation ?? 0, entry: best }
  }

  /**
   * @description The commit timestamp of the newest committed generation at
   * or below `gen`, when its record-set is still resident. Used to populate
   * `Db.timestamp` for `asOf()` values.
   * @param gen - The pinned generation.
   */
  async commitTimestampAtOrBefore(gen: number): Promise<number | null> {
    for (let i = this.committedGens.length - 1; i >= 0; i--) {
      const candidate = this.committedGens[i]
      if (candidate > gen) continue
      const delta = await this.getDelta(candidate)
      return delta.timestamp
    }
    return null
  }

  private async getDelta(
    gen: number
  ): Promise<{ nouns: Set<string>; verbs: Set<string>; timestamp: number }> {
    const cached = this.deltaCache.get(gen)
    if (cached) return cached
    const delta = (await this.storage.readRawObject(
      `${GENERATIONS_PREFIX}/${gen}/tx.json`
    )) as GenerationDelta | null
    if (delta === null) {
      throw new Error(
        `Generation delta missing: ${GENERATIONS_PREFIX}/${gen}/tx.json ` +
          `(store corrupted or records removed outside compactHistory())`
      )
    }
    const entry = {
      nouns: new Set(delta.nouns),
      verbs: new Set(delta.verbs),
      timestamp: delta.timestamp
    }
    this.deltaCache.set(gen, entry)
    return entry
  }

  // ==========================================================================
  // Compaction
  // ==========================================================================

  /**
   * @description Reclaim generation record-sets past the retention policy.
   * A generation directory `N` may be removed only when `N` is at or below
   * **every** live pin (deleting `N` can only break readers pinned *below*
   * `N`, because resolution reads before-images from generations strictly
   * greater than the pin). With both retention options supplied, a
   * generation must satisfy both to be eligible.
   *
   * @param options - Retention policy (see {@link CompactHistoryOptions}).
   * @returns Count of removed record-sets and the new horizon.
   */
  async compact(options?: CompactHistoryOptions): Promise<CompactHistoryResult> {
    return this.withMutex(async () => {
      const minPinned = this.minPinnedGeneration()
      const retainGenerations = options?.retainGenerations ?? 0
      const countWatermark = this.committed - retainGenerations
      const timeCutoff =
        options?.retainMs !== undefined ? Date.now() - options.retainMs : Infinity

      const removed: number[] = []
      for (const gen of [...this.committedGens]) {
        if (gen > countWatermark) continue
        if (gen > minPinned) continue
        if (timeCutoff !== Infinity) {
          const delta = await this.getDelta(gen)
          if (delta.timestamp >= timeCutoff) continue
        }
        await this.storage.removeRawPrefix(`${GENERATIONS_PREFIX}/${gen}`)
        this.deltaCache.delete(gen)
        removed.push(gen)
      }

      if (removed.length > 0) {
        const removedSet = new Set(removed)
        this.committedGens = this.committedGens.filter((gen) => !removedSet.has(gen))
        this.horizonGen = Math.max(this.horizonGen, ...removed)
        const manifest: GenerationManifest = {
          version: 1,
          generation: this.committed,
          committedAt: new Date().toISOString(),
          horizon: this.horizonGen
        }
        await this.storage.writeRawObject(MANIFEST_PATH, manifest)
        await this.storage.syncRawObjects([MANIFEST_PATH])
      }

      return { removedGenerations: removed.length, horizon: this.horizonGen }
    })
  }

  // ==========================================================================
  // Restore support
  // ==========================================================================

  /**
   * @description Re-read all persisted state after `brain.restore()`
   * replaced the store's contents from a snapshot, enforcing counter
   * monotonicity: the counter never moves below `floorGeneration` (the
   * pre-restore counter), so generation numbers observed before a restore
   * are never reissued.
   * @param floorGeneration - The counter value before the restore.
   */
  async reopenAfterRestore(floorGeneration: number): Promise<void> {
    await this.withMutex(async () => {
      this.deltaCache.clear()
      this.opened = false
      // open() re-reads counter/manifest and re-registers the bump hook.
      await this.open()
      if (this.counter < floorGeneration) {
        this.counter = floorGeneration
        await this.persistCounterUnlocked()
      }
    })
  }

  // ==========================================================================
  // Internals
  // ==========================================================================

  /**
   * Restore the before-images of an uncommitted generation to the canonical
   * entity paths, then remove its record directory. Restores are idempotent
   * (writing identical bytes / deleting already-absent files), so recovery
   * is safe at every crash point of the commit protocol.
   */
  private async rollBackUncommittedGeneration(gen: number): Promise<void> {
    const dir = `${GENERATIONS_PREFIX}/${gen}`
    const prevPaths = await this.storage.listRawObjects(`${dir}/prev`)
    for (const recordPath of prevPaths) {
      const id = recordIdFromPath(recordPath)
      if (id === null) continue
      const record = (await this.storage.readRawObject(recordPath)) as GenerationRecord | null
      if (record === null) continue
      const image = { metadata: record.metadata, vector: record.vector }
      if (record.kind === 'verb') {
        await this.storage.writeVerbRaw(id, image)
      } else {
        await this.storage.writeNounRaw(id, image)
      }
    }
    await this.storage.removeRawPrefix(dir)
    prodLog.warn(
      `[GenerationStore] rolled back uncommitted generation ${gen} ` +
        `(${prevPaths.length} record(s) restored)`
    )
  }

  /**
   * @description Run a snapshot section serialized against transact commits,
   * compaction, and counter persistence (the store's single commit mutex),
   * with the generation counter durably persisted first. Used by
   * `db.persist()`: the lock guarantees a snapshot can never interleave with
   * a commit's record-write/manifest-rename sequence, and the up-front
   * counter persist guarantees the snapshot's `_system/generation.json`
   * reflects every single-operation bump (whose persistence is otherwise
   * coalesced and may still be scheduled).
   * @param fn - The exclusive snapshot section.
   * @returns `fn`'s result.
   */
  snapshotWith<R>(fn: () => Promise<R>): Promise<R> {
    return this.withMutex(async () => {
      await this.persistCounterUnlocked()
      return fn()
    })
  }

  /** Run `fn` exclusively, chained behind every prior exclusive section. */
  private withMutex<R>(fn: () => Promise<R>): Promise<R> {
    const run = this.mutexTail.then(fn, fn)
    // Keep the chain alive regardless of fn's outcome.
    this.mutexTail = run.catch(() => {})
    return run
  }
}

/**
 * @description Extract the generation number from a record path of the form
 * `_generations/<N>/...`. Returns `null` for paths that don't match.
 * @param path - A storage-root-relative object path.
 */
function parseGenerationFromPath(path: string): number | null {
  const match = /^_generations[/\\](\d+)[/\\]/.exec(path)
  if (!match) return null
  const gen = Number(match[1])
  return Number.isSafeInteger(gen) ? gen : null
}

/**
 * @description Extract the record id (file basename without `.json`) from a
 * `prev/` before-image record path. Returns `null` for non-record paths.
 * @param path - A storage-root-relative object path.
 */
function recordIdFromPath(path: string): string | null {
  const match = /[/\\]prev[/\\]([^/\\]+)\.json$/.exec(path)
  return match ? match[1] : null
}