/**
 * @module db/db
 * @description `Db` — Brainy 8.0's immutable, Datomic-style database value.
 *
 * A `Db` is a **readonly view pinned at one generation** of a Brainy store.
 * It is produced by `brain.now()` (O(1) pin of the current generation),
 * `brain.transact()` (pinned at the freshly committed generation),
 * `brain.asOf()` (a past generation, a timestamp, or a persisted snapshot
 * path), `Brainy.load(path)` (a snapshot opened read-only), and `db.with()`
 * (a speculative in-memory overlay).
 *
 * **Read semantics.** While no transaction has committed past the pinned
 * generation, every read delegates straight to the live brain — the pinned
 * view IS the current state, and the existing fast paths serve it untouched.
 * Once later transactions commit, the `Db` serves the FULL query surface at
 * the pinned generation through two complementary paths:
 *
 * - `get()`, metadata-level `find()`, and filter-based `related()` resolve
 *   through the generational record layer (changed ids from immutable
 *   before-images; unchanged ids still ride the live fast path) — no
 *   materialization cost.
 * - Index-accelerated queries (semantic/vector search, graph traversal,
 *   cursors, aggregation) are served by **at-generation index
 *   materialization**: on first use, the host rebuilds ephemeral in-memory
 *   indexes (the same vector/metadata/graph index classes the live brain
 *   uses) over the exact at-generation record set, caches them on this `Db`,
 *   and frees them on {@link Db.release}. This costs O(n at G) time and
 *   memory ONCE per `Db` — the open-core price of historical index queries;
 *   a native {@link ../plugin.js VersionedIndexProvider} serves the same
 *   reads from its retained segments without any rebuild.
 *
 * Speculative `with()` overlays keep one honest boundary: overlay entities
 * carry no embeddings, so index-accelerated queries on overlays throw
 * {@link SpeculativeOverlayError} instead of returning silently-incomplete
 * results.
 *
 * **History granularity.** Generation records are written per `transact()`
 * batch. Single-operation writes (`add`/`update`/`relate`/… outside
 * `transact()`) advance the generation counter but do not produce historical
 * records, so they remain visible through earlier pins — the documented 8.0
 * granularity (see `docs/ADR-001-generational-mvcc.md`, "History
 * granularity").
 *
 * **Lifecycle.** Each `Db` holds one refcounted pin on its generation (and
 * on every registered {@link ../plugin.js VersionedIndexProvider}). Call
 * {@link Db.release} when done — a `FinalizationRegistry` backstop releases
 * leaked pins on garbage collection, but explicit release is what makes
 * `compactHistory()` deterministic.
 */

import type {
  Entity,
  FindParams,
  GetOptions,
  RelatedParams,
  Relation,
  Result
} from '../types/brainy.types.js'
import type { StorageAdapter } from '../coreTypes.js'
import { exportGraph } from './portableGraph.js'
import type { ExportSelector, ExportOptions, PortableGraph } from './portableGraph.js'
import {
  splitNounMetadataRecord,
  splitVerbMetadataRecord
} from '../types/reservedFields.js'
import { v4 as uuidv4 } from '../universal/uuid.js'
import { coerceNewEntityId, resolveEntityId, ORIGINAL_ID_KEY } from '../utils/idNormalization.js'
import { EntityNotFoundError } from '../errors/notFound.js'
import { SpeculativeOverlayError } from './errors.js'
import type { GenerationStore } from './generationStore.js'
import type { ChangedIds, TransactReceipt, TxOperation } from './types.js'
import { entityMatchesFind, resolveEntityField, UnsupportedWhereOperatorError } from './whereMatcher.js'

/**
 * @description The query surface a historical materialization exposes back
 * to its `Db`: the full live read pipeline of an ephemeral reader brain
 * whose storage holds the exact at-generation record set. Produced by
 * {@link DbHost.materializeAt}; closed (indexes freed) via `close()` when
 * the owning `Db` is released.
 */
export interface HistoricalQueryHandle<T = any> {
  /** Full `find()` surface (vector/semantic/graph/cursor/aggregate) at the pinned generation. */
  find(query: string | FindParams<T>): Promise<Result<T>[]>
  /** Full `related()` surface (including cursor pagination) at the pinned generation. */
  related(paramsOrId?: string | RelatedParams): Promise<Relation<T>[]>
  /** Free the materialized indexes (closes the ephemeral reader). Idempotent. */
  close(): Promise<void>
}

/**
 * @description The speculative overlay carried by a `db.with()` value:
 * per-id replacement entities/relations (`null` = tombstone). Reads on the
 * overlay `Db` consult these maps first, then fall through to the pinned
 * base view. Overlays are pure in-memory values — they never touch disk or
 * index providers.
 */
export interface SpeculativeOverlay<T = any> {
  /** Entity overlays: replacement entity, or `null` for a speculative delete. */
  nouns: Map<string, Entity<T> | null>
  /** Relation overlays: replacement relation, or `null` for a speculative delete. */
  verbs: Map<string, Relation<T> | null>
}

/**
 * @description The host surface a `Db` needs from its `Brainy` instance —
 * constructed once per brain (see `brainy.ts`). Internal: consumers never
 * build one; `Db` values are only created by `Brainy` and `db.with()`.
 */
export interface DbHost<T = any> {
  /** The generational record layer of the host brain. */
  readonly store: GenerationStore
  /** The host brain's storage adapter (portable-export VFS blob bytes). */
  readonly storage: StorageAdapter
  /** Live `brain.get()` (current-generation fast path). */
  get(id: string, options?: GetOptions): Promise<Entity<T> | null>
  /** Live `brain.find()` (current-generation fast path). */
  find(query: string | FindParams<T>): Promise<Result<T>[]>
  /** Live `brain.related()` (current-generation fast path). */
  related(paramsOrId?: string | RelatedParams): Promise<Relation<T>[]>
  /**
   * Resolve a `generation | Date` to a concrete generation (+ timestamp) via the
   * brain's shared `asOf` resolver — so `since(gen|Date)` shares `asOf`'s
   * reachability and `Date` semantics. `exclusive` pins the generation before the
   * resolved one.
   */
  resolveGeneration(
    target: number | Date,
    options?: { exclusive?: boolean }
  ): Promise<{ generation: number; timestamp: number }>
  /** Materialize an entity from a generation record's raw stored objects. */
  entityFromRecord(
    id: string,
    record: { metadata: any; vector: any | null },
    includeVectors: boolean
  ): Promise<Entity<T>>
  /** Materialize a relation from a generation record's raw stored objects. */
  relationFromRecord(id: string, record: { metadata: any; vector: any | null }): Relation<T> | null
  /** Snapshot the store at `generation` into `targetPath` (throws if the pin is stale). */
  persistPinned(targetPath: string, generation: number): Promise<void>
  /**
   * Build the at-`generation` index materialization: an ephemeral in-memory
   * reader over the exact record set at that generation, serving the full
   * query surface. O(n at G) — called at most once per `Db` (cached by the
   * caller) and freed via the returned handle's `close()`.
   */
  materializeAt(generation: number): Promise<HistoricalQueryHandle<T>>
  /** Add one refcounted pin (store + versioned providers) on `generation`. */
  pinGeneration(generation: number): void
  /** Release one refcounted pin (store + versioned providers) on `generation`. */
  releaseGeneration(generation: number): void
  /** Register a `Db` with the GC-backstop finalization registry. */
  registerDbForFinalization(db: Db<T>, generation: number, closeOnRelease?: () => Promise<void>): void
  /** Unregister a `Db` after an explicit release (no double-release on GC). */
  unregisterDbFromFinalization(db: Db<T>): void
}

/**
 * @description Internal constructor arguments for {@link Db}. The pin on
 * `generation` is taken by the creator (`Brainy` or `db.with()`) **before**
 * construction; the constructor only registers the GC backstop.
 */
export interface DbInit<T = any> {
  /** The host surface (one per brain). */
  host: DbHost<T>
  /** The pinned generation. */
  generation: number
  /** The view's timestamp (ms since epoch) — see {@link Db.timestamp}. */
  timestamp: number
  /** Per-operation receipt, present only on `transact()`-produced values. */
  receipt?: TransactReceipt
  /** Speculative overlay, present only on `with()`-produced values. */
  overlay?: SpeculativeOverlay<T>
  /** Extra teardown on release (e.g. closing an owned snapshot brain). */
  closeOnRelease?: () => Promise<void>
}

/**
 * @description An immutable, generation-pinned view of a Brainy store. See
 * the module documentation for the full read-semantics contract.
 *
 * @example
 * const db = brain.now()                       // O(1) pin
 * await brain.transact([{ op: 'update', id, metadata: { v: 2 } }])
 * await db.get(id)                             // still sees v: 1 — pinned
 * await brain.get(id)                          // sees v: 2 — live
 * await db.release()                           // unpin (enables compaction)
 */
export class Db<T = any> {
  private readonly host: DbHost<T>
  private readonly gen: number
  private readonly ts: number
  private readonly overlay?: SpeculativeOverlay<T>
  private readonly closeOnRelease?: () => Promise<void>
  private isReleased = false
  /**
   * Cached at-generation index materialization (built lazily by the first
   * index-accelerated query at a historical pin; freed on release()).
   */
  private materializedHandle?: Promise<HistoricalQueryHandle<T>>

  /**
   * Per-operation receipt (committed generation, timestamp, resolved ids in
   * input order). Present only on values returned by `brain.transact()`.
   */
  public readonly receipt?: TransactReceipt

  /**
   * @param init - Internal construction arguments. `Db` values are created
   *   by `Brainy` (`now`/`transact`/`asOf`/`load`) and `db.with()` — never
   *   directly by consumers.
   */
  constructor(init: DbInit<T>) {
    this.host = init.host
    this.gen = init.generation
    this.ts = init.timestamp
    this.receipt = init.receipt
    this.overlay = init.overlay
    this.closeOnRelease = init.closeOnRelease
    this.host.registerDbForFinalization(this, this.gen, this.closeOnRelease)
  }

  /** The generation this view is pinned at. */
  get generation(): number {
    return this.gen
  }

  /**
   * @internal Same-instance guard for cross-`Db` operations like
   * `Brainy.diff()`: whether this view is backed by `store`. Lets `Brainy`
   * reject a `Db` endpoint from a different brain without exposing the private
   * host.
   */
  belongsToStore(store: GenerationStore): boolean {
    return this.host.store === store
  }

  /**
   * The view's wall-clock timestamp (ms since epoch): pin time for `now()`,
   * commit time for `transact()`, and the resolved commit time for `asOf()`.
   */
  get timestamp(): number {
    return this.ts
  }

  /** Whether {@link Db.release} has been called (released views throw on use). */
  get released(): boolean {
    return this.isReleased
  }

  /** Whether this view carries a speculative `with()` overlay. */
  get speculative(): boolean {
    return this.overlay !== undefined
  }

  // ==========================================================================
  // Reads
  // ==========================================================================

  /**
   * @description Get an entity by id **as of this view's generation**.
   * Overlay entries win (speculative views); ids untouched since the pin ride
   * the live fast path; ids changed by later transactions resolve from
   * immutable generation records. Always correct at the pinned generation.
   *
   * @param id - The entity id.
   * @param options - Same options as `brain.get()` (`includeVectors`).
   * @returns The entity as of this generation, or `null`.
   */
  async get(id: string, options?: GetOptions): Promise<Entity<T> | null> {
    this.assertUsable('get')

    // Id normalization (8.0): resolve a natural key to the canonical UUID the
    // write path stored, so this view reads by natural key consistently. A real
    // UUID passes through. Overlay keys are canonical (see with()).
    id = resolveEntityId(id)

    if (this.overlay && this.overlay.nouns.has(id)) {
      return this.overlay.nouns.get(id) ?? null
    }

    if (!this.isHistorical()) {
      return this.host.get(id, options)
    }

    const resolved = await this.host.store.resolveAt('noun', id, this.gen)
    if (resolved.source === 'current') {
      return this.host.get(id, options)
    }
    if (resolved.source === 'absent' || resolved.metadata === null) {
      // Live semantics: an entity without metadata is not a live entity.
      return null
    }
    return this.host.entityFromRecord(
      id,
      { metadata: resolved.metadata, vector: resolved.vector },
      options?.includeVectors ?? false
    )
  }

  /**
   * @description The full `find()` surface **as of this view's generation**.
   *
   * Routing:
   * - **Current generation, no overlay** — delegates to `brain.find()`
   *   untouched (the existing fast paths).
   * - **Historical generation** — metadata dimensions (`type`, `subtype`,
   *   `where`, `service`, `excludeVFS`, `orderBy`/`order`,
   *   `limit`/`offset`) are computed from the generational record layer
   *   directly (no materialization cost): the live index serves unchanged
   *   ids, and per-entity evaluation of the same filters covers
   *   record-resolved ids — set-correct at the pinned generation.
   *   Index-accelerated dimensions (`query`, `vector`, `near`, `connected`,
   *   `cursor`, `aggregate`, `includeRelations`, non-metadata search modes)
   *   are served by the at-generation index materialization — built lazily
   *   on first use (O(n at G) time and memory, once per `Db`; see the module
   *   doc), then cached until {@link Db.release}.
   * - **Speculative overlay** — metadata dimensions only; index-accelerated
   *   dimensions throw {@link SpeculativeOverlayError} (overlay entities
   *   carry no embeddings — see the error's documentation).
   *
   * Result ordering is deterministic when `orderBy` is supplied; without it,
   * record-path results list live-index matches before record-resolved
   * matches, while materialized results follow the live engine's ordering.
   *
   * @param query - A `FindParams` object, or a string (semantic search).
   * @returns Matching results as of this generation (score `1.0` for
   *   record-resolved metadata matches, mirroring live metadata-only finds).
   */
  async find(query: string | FindParams<T>): Promise<Result<T>[]> {
    this.assertUsable('find')

    const params: FindParams<T> = typeof query === 'string' ? { query } : query
    const historical = this.isHistorical()

    if (!historical && !this.overlay) {
      return this.host.find(query)
    }

    if (this.overlay) {
      this.assertOverlayCompatibleFind(params)
    } else if (findRequiresIndexes(params)) {
      const materialized = await this.materialize()
      return materialized.find(params)
    }

    const limit = params.limit ?? 10
    const offset = params.offset ?? 0

    // Ids whose live-index answer is NOT valid at this generation.
    const changedNouns = historical
      ? (await this.host.store.changedBetween(this.gen, this.host.store.committedGeneration())).nouns
      : []
    const overlayNouns = this.overlay?.nouns ?? new Map<string, Entity<T> | null>()
    const excluded = new Set<string>([...changedNouns, ...overlayNouns.keys()])

    try {
      // Over-fetch from the live index so dropping superseded ids cannot
      // starve the requested page, then merge and re-window.
      const base = await this.host.find({
        ...params,
        limit: limit + offset + excluded.size,
        offset: 0
      })
      const merged = base.filter((result) => !excluded.has(result.id))

      for (const id of changedNouns) {
        if (overlayNouns.has(id)) continue // The overlay supersedes history.
        const resolved = await this.host.store.resolveAt('noun', id, this.gen)
        let entity: Entity<T> | null = null
        if (resolved.source === 'record' && resolved.metadata !== null) {
          entity = await this.host.entityFromRecord(
            id,
            { metadata: resolved.metadata, vector: resolved.vector },
            false
          )
        } else if (resolved.source === 'current') {
          // Defensive: changed ids always resolve to a record or absent, but
          // a 'current' answer is still served correctly from the live path.
          entity = await this.host.get(id)
        }
        if (entity && entityMatchesFind(entity as Entity, params as FindParams)) {
          merged.push(resultFromEntity(entity))
        }
      }

      for (const [, entity] of overlayNouns) {
        if (entity === null) continue // Speculative delete.
        if (entityMatchesFind(entity as Entity, params as FindParams)) {
          merged.push(resultFromEntity(entity))
        }
      }

      if (params.orderBy) {
        sortResultsBy(merged, params.orderBy, params.order ?? 'asc')
      }
      return merged.slice(offset, offset + limit)
    } catch (err) {
      if (err instanceof UnsupportedWhereOperatorError) {
        // The record-path's per-entity matcher doesn't implement this
        // where-operator. On an overlay that is a hard boundary; at a
        // historical generation the materialization serves it via the full
        // live query engine.
        if (this.overlay) {
          throw new SpeculativeOverlayError(
            `metadata find with where-operator '${err.operator}'`,
            this.gen
          )
        }
        const materialized = await this.materialize()
        return materialized.find(params)
      }
      throw err
    }
  }

  /**
   * @description Serialize part or all of this database value into a portable
   * `PortableGraph` document, read **as of this view's generation**. Because `export`
   * lives on the immutable `Db`, it composes with every way of obtaining one:
   * `brain.now().export(sel)` (current), `(await brain.asOf(g)).export(sel)`
   * (time-travel export), and `brain.now().with(ops).export(sel)` (what-if export).
   *
   * The document is portable JSON, versioned (`formatVersion`), and current-state
   * (no generation history) — distinct from `persist()` (native whole-brain snapshot
   * that preserves history). Restore with `brain.import(backup)`.
   *
   * @param selector - WHAT to export (omit for the whole brain). See {@link ExportSelector}.
   * @param options - HOW to export (vectors / VFS bytes / edge policy). See {@link ExportOptions}.
   * @returns A versioned, portable `PortableGraph` document.
   * @example
   * const backup = await brain.now().export({ collection: id }, { includeVectors: true })
   */
  async export(selector: ExportSelector = {}, options: ExportOptions = {}): Promise<PortableGraph> {
    this.assertUsable('export')
    return exportGraph(this, this.host.storage, selector, options)
  }

  /**
   * @description Relationships **as of this view's generation** — the `Db`
   * counterpart of `brain.related()` (same string-id shorthand and
   * filter surface). Relations whose edges changed after the pin are
   * resolved from generation records; overlay relations (speculative
   * `relate`/`unrelate`) and cascade tombstones (relations touching a
   * speculatively deleted entity) are applied on top. Cursor pagination is
   * index-only: at a historical generation it is served by the
   * at-generation index materialization (O(n at G) once per `Db`); on a
   * speculative overlay it throws {@link SpeculativeOverlayError}.
   *
   * @param paramsOrId - An entity id (shorthand for `{ from: id }`) or a
   *   `RelatedParams` filter object.
   * @returns Relations as of this generation.
   */
  async related(paramsOrId?: string | RelatedParams): Promise<Relation<T>[]> {
    this.assertUsable('related')

    const rawParams: RelatedParams =
      typeof paramsOrId === 'string' ? { from: paramsOrId } : (paramsOrId ?? {})
    // Id normalization (8.0): resolve the anchor id(s) so this view traverses by
    // natural key consistently with the live engine. Real UUIDs pass through.
    const params: RelatedParams = {
      ...rawParams,
      ...(rawParams.from !== undefined && { from: resolveEntityId(rawParams.from) }),
      ...(rawParams.to !== undefined && { to: resolveEntityId(rawParams.to) })
    }
    const historical = this.isHistorical()

    if (!historical && !this.overlay) {
      return this.host.related(params)
    }

    if (params.cursor !== undefined) {
      if (this.overlay) {
        throw new SpeculativeOverlayError('cursor pagination on related()', this.gen)
      }
      const materialized = await this.materialize()
      return materialized.related(params)
    }

    const limit = params.limit ?? 100
    const offset = params.offset ?? 0

    const changedVerbs = historical
      ? (await this.host.store.changedBetween(this.gen, this.host.store.committedGeneration())).verbs
      : []
    const overlayVerbs = this.overlay?.verbs ?? new Map<string, Relation<T> | null>()
    const excluded = new Set<string>([...changedVerbs, ...overlayVerbs.keys()])

    const base = await this.host.related({
      ...params,
      limit: limit + offset + excluded.size,
      offset: 0
    })
    let merged = base.filter((relation) => !excluded.has(relation.id))

    for (const id of changedVerbs) {
      if (overlayVerbs.has(id)) continue
      const resolved = await this.host.store.resolveAt('verb', id, this.gen)
      if (resolved.source !== 'record') continue
      const relation = this.host.relationFromRecord(id, {
        metadata: resolved.metadata,
        vector: resolved.vector
      })
      if (relation && relationMatchesParams(relation, params)) {
        merged.push(relation)
      }
    }

    for (const [, relation] of overlayVerbs) {
      if (relation === null) continue // Speculative unrelate.
      if (relationMatchesParams(relation, params)) {
        merged.push(relation)
      }
    }

    // Cascade semantics for speculative deletes: a relation whose endpoint
    // is tombstoned in the overlay is gone in this view, exactly as
    // brain.remove() cascades on the durable path.
    if (this.overlay) {
      const tombstoned = new Set<string>()
      for (const [id, entity] of this.overlay.nouns) {
        if (entity === null) tombstoned.add(id)
      }
      if (tombstoned.size > 0) {
        merged = merged.filter(
          (relation) => !tombstoned.has(relation.from) && !tombstoned.has(relation.to)
        )
      }
    }

    return merged.slice(offset, offset + limit)
  }

  // ==========================================================================
  // Derived views
  // ==========================================================================

  /**
   * @description Speculative write: returns a NEW `Db` whose reads see the
   * given transaction data applied **in memory, on top of this view** —
   * Datomic's `with`. Nothing touches disk, the generation counter, or index
   * providers; the underlying brain and this view are untouched. Use it to
   * answer "what would the store look like if…" questions, then call
   * `brain.transact()` with the same operations to make it real.
   *
   * Speculative semantics (documented deltas from the durable paths):
   * - `add` without an explicit `vector` produces an entity with an empty
   *   stub vector — speculative entities are not semantically searchable
   *   (`with()` never invokes the embedder).
   * - `remove` cascades on the read side: relations touching the removed
   *   entity disappear from `related()` without being enumerated.
   * - `relate` deduplicates against the edges visible in this view (overlay
   *   first, then the record layer), mirroring `brain.relate()`.
   * - Brain-level vocabulary/subtype enforcement does not run — overlays
   *   never persist, so there is nothing to protect.
   *
   * The returned view takes its own pin on this generation; release both
   * views independently.
   *
   * @param ops - The same declarative operations `brain.transact()` accepts.
   * @returns A new speculative `Db` over this view.
   * @throws EntityNotFoundError when an operation references a missing
   *   entity, mirroring the durable path's planning errors.
   */
  async with(ops: TxOperation<T>[]): Promise<Db<T>> {
    this.assertUsable('with')
    if (!Array.isArray(ops) || ops.length === 0) {
      throw new Error('with(): provide a non-empty array of transaction operations')
    }

    // Child overlay starts as a copy of this view's overlay (chained with()).
    const overlay: SpeculativeOverlay<T> = {
      nouns: new Map(this.overlay?.nouns ?? []),
      verbs: new Map(this.overlay?.verbs ?? [])
    }

    // Reads during planning must see "this view + overlay so far".
    const speculativeGet = async (id: string): Promise<Entity<T> | null> => {
      if (overlay.nouns.has(id)) return overlay.nouns.get(id) ?? null
      return this.get(id)
    }

    for (const op of ops) {
      switch (op.op) {
        case 'add': {
          // Reserved-field normalization — mirror of the brain.transact()
          // write path: user-settable fields lift to their dedicated field
          // (top-level wins), system-managed fields drop, and the entity's
          // metadata bag carries ONLY custom fields. Speculative views skip
          // the one-shot warnings — committing the same ops through
          // `brain.transact()` warns on the real write path.
          const { reserved, custom } = splitNounMetadataRecord(
            op.metadata as Record<string, unknown> | undefined
          )
          const confidence =
            op.confidence ?? (typeof reserved.confidence === 'number' ? reserved.confidence : undefined)
          const weight =
            op.weight ?? (typeof reserved.weight === 'number' ? reserved.weight : undefined)
          const subtype =
            op.subtype ?? (typeof reserved.subtype === 'string' ? reserved.subtype : undefined)
          const service =
            op.service ?? (typeof reserved.service === 'string' ? reserved.service : undefined)

          // Id normalization (8.0) — mirror of the committed transact() add
          // path: a natural key coerces to a STABLE UUID (v5), preserving the
          // original under ORIGINAL_ID_KEY; a real UUID passes through; no id
          // mints a fresh id so the overlay keys match the durable path.
          const { id, originalId } = coerceNewEntityId(op.id)
          const now = Date.now()
          overlay.nouns.set(id, {
            id,
            vector: op.vector ?? [],
            type: op.type,
            ...(subtype !== undefined && { subtype }),
            data: op.data,
            metadata: {
              ...(custom as object),
              ...(originalId !== undefined && { [ORIGINAL_ID_KEY]: originalId })
            } as T,
            ...(service !== undefined && { service }),
            createdAt: now,
            updatedAt: now,
            ...(confidence !== undefined && { confidence }),
            ...(weight !== undefined && { weight }),
            _rev: 1
          })
          break
        }
        case 'update': {
          // Id normalization (8.0): resolve a natural key to the canonical UUID
          // the write path stored. A real UUID passes through.
          const updateId = resolveEntityId(op.id)
          const base = await speculativeGet(updateId)
          if (!base) {
            throw new EntityNotFoundError(
              updateId,
              `with(): entity ${updateId} not found at generation ${this.gen}`
            )
          }
          // Same reserved-field normalization as the committed update path.
          const { reserved, custom } = splitNounMetadataRecord(
            op.metadata as Record<string, unknown> | undefined
          )
          const confidence =
            op.confidence ?? (typeof reserved.confidence === 'number' ? reserved.confidence : undefined)
          const weight =
            op.weight ?? (typeof reserved.weight === 'number' ? reserved.weight : undefined)
          const subtype =
            op.subtype ?? (typeof reserved.subtype === 'string' ? reserved.subtype : undefined)
          const mergedMetadata =
            op.merge !== false
              ? ({ ...(base.metadata as object), ...custom } as T)
              : ((op.metadata !== undefined ? custom : base.metadata) as T)
          overlay.nouns.set(updateId, {
            ...base,
            ...(op.type !== undefined && { type: op.type }),
            ...(subtype !== undefined && { subtype }),
            ...(op.data !== undefined && { data: op.data }),
            ...(op.vector !== undefined && { vector: op.vector }),
            ...(confidence !== undefined && { confidence }),
            ...(weight !== undefined && { weight }),
            metadata: mergedMetadata,
            updatedAt: Date.now(),
            _rev: (base._rev ?? 1) + 1
          })
          break
        }
        case 'remove': {
          // Id normalization (8.0): resolve a natural key to the canonical UUID
          // the write path stored. A real UUID passes through.
          const removeId = resolveEntityId(op.id)
          overlay.nouns.set(removeId, null)
          // Tombstone overlay relations touching the removed entity; base
          // relations are cascade-filtered at read time in related().
          for (const [verbId, relation] of overlay.verbs) {
            if (relation && (relation.from === removeId || relation.to === removeId)) {
              overlay.verbs.set(verbId, null)
            }
          }
          break
        }
        case 'relate': {
          // Id normalization (8.0) — mirror of relate(): resolve BOTH endpoints
          // to the canonical UUID the write path stored. Real UUIDs pass through.
          const relFrom = resolveEntityId(op.from)
          const relTo = resolveEntityId(op.to)
          const from = await speculativeGet(relFrom)
          const to = await speculativeGet(relTo)
          if (!from) {
            throw new EntityNotFoundError(relFrom, `with(): source entity ${relFrom} not found`)
          }
          if (!to) {
            throw new EntityNotFoundError(relTo, `with(): target entity ${relTo} not found`)
          }

          // Dedupe against the view (overlay first, then the edges visible
          // at this generation via the record layer) — mirror of relate().
          let duplicate: Relation<T> | undefined
          for (const relation of overlay.verbs.values()) {
            if (relation && relation.from === relFrom && relation.to === relTo && relation.type === op.type) {
              duplicate = relation
              break
            }
          }
          if (!duplicate) {
            const existing = await this.related({ from: relFrom, type: op.type })
            duplicate = existing.find((relation) => relation.to === relTo)
          }
          if (duplicate) break

          // Reserved-field normalization — relationship mirror of the add
          // op above (and of the committed relate() path).
          const { reserved, custom } = splitVerbMetadataRecord(
            op.metadata as Record<string, unknown> | undefined
          )
          const confidence =
            op.confidence ?? (typeof reserved.confidence === 'number' ? reserved.confidence : undefined)
          const weight =
            op.weight ?? (typeof reserved.weight === 'number' ? reserved.weight : undefined)
          const subtype =
            op.subtype ?? (typeof reserved.subtype === 'string' ? reserved.subtype : undefined)
          const service =
            op.service ?? (typeof reserved.service === 'string' ? reserved.service : undefined)

          const id = uuidv4()
          overlay.verbs.set(id, {
            id,
            from: relFrom,
            to: relTo,
            type: op.type,
            ...(subtype !== undefined && { subtype }),
            weight: weight ?? 1.0,
            ...(confidence !== undefined && { confidence }),
            ...(op.data !== undefined && { data: op.data }),
            metadata: custom as T,
            ...(service !== undefined && { service }),
            createdAt: Date.now()
          })
          if (op.bidirectional) {
            const reverseId = uuidv4()
            overlay.verbs.set(reverseId, {
              id: reverseId,
              from: relTo,
              to: relFrom,
              type: op.type,
              ...(subtype !== undefined && { subtype }),
              weight: weight ?? 1.0,
              ...(confidence !== undefined && { confidence }),
              ...(op.data !== undefined && { data: op.data }),
              metadata: custom as T,
              ...(service !== undefined && { service }),
              createdAt: Date.now()
            })
          }
          break
        }
        case 'unrelate': {
          overlay.verbs.set(op.id, null)
          break
        }
        default: {
          const exhaustive: never = op
          throw new Error(`with(): unknown operation ${JSON.stringify(exhaustive)}`)
        }
      }
    }

    this.host.pinGeneration(this.gen)
    return new Db<T>({
      host: this.host,
      generation: this.gen,
      timestamp: Date.now(),
      overlay
    })
  }

  /**
   * @description The ids changed by committed transactions in the interval
   * `(lowerBound, this.generation]` — the diff between an earlier state and this
   * view. The lower bound is **exclusive** (contrast `transactionLog`'s inclusive
   * window). Single-operation writes between commits are not reflected (documented
   * 8.0 history granularity).
   *
   * The lower bound may be:
   * - **a `Db`** — an earlier view of the SAME store (its pinned generation);
   * - **a generation number** — used directly as the exclusive lower bound;
   * - **a `Date`** — resolved to the generation committed at or before it.
   *
   * `db.since(prior)` equals `db.since(prior.generation)` by construction.
   *
   * @param lowerBound - An earlier `Db`, a generation number, or a `Date`.
   * @returns Changed entity and relation ids, sorted.
   * @throws RangeError when the resolved lower bound is newer than this view.
   * @throws Error when a `Db` lower bound belongs to a different store.
   * @throws GenerationCompactedError when a number/`Date` lower bound is below the
   *   compaction horizon.
   * @example
   * const before = brain.now()
   * await brain.transact(ops)
   * await brain.now().since(before)        // ids changed by the transaction
   * await brain.now().since(before.generation)
   * await brain.now().since(new Date(Date.now() - 3_600_000))
   */
  async since(lowerBound: Db<T> | number | Date): Promise<ChangedIds> {
    this.assertUsable('since')

    let fromGen: number
    if (lowerBound instanceof Db) {
      if (lowerBound.host.store !== this.host.store) {
        throw new Error('since(): both Db values must come from the same Brainy instance')
      }
      fromGen = lowerBound.gen
    } else {
      // number = exclusive lower-bound generation; Date resolves via the shared
      // asOf resolver (same reachability gate). changedBetween treats fromGen as
      // exclusive, so since(db) === since(db.generation).
      fromGen = (await this.host.resolveGeneration(lowerBound)).generation
    }

    if (fromGen > this.gen) {
      throw new RangeError(
        `since(): lower-bound generation ${fromGen} is newer than this view ` +
          `(generation ${this.gen}). Pass an earlier generation/Date, or call newerDb.since(olderDb).`
      )
    }
    return this.host.store.changedBetween(fromGen, this.gen)
  }

  // ==========================================================================
  // Durability
  // ==========================================================================

  /**
   * @description Snapshot this view to `path` as a self-contained store:
   * flush, then hard-link every immutable file into the target
   * (Cassandra-style — instant and space-shared; cross-device targets fall
   * back to byte copies, and the small append-in-place file list is always
   * byte-copied). The result opens with `Brainy.load(path)` with the full
   * query surface, and later mutations of the source can never alter it —
   * rewrites swap inodes, the snapshot keeps the old bytes.
   *
   * `persist()` requires this view to still be the store's **latest**
   * generation (snapshotting captures current bytes): pin with `brain.now()`
   * and persist before further writes. A view that history has moved past
   * throws {@link GenerationConflictError}; speculative overlays throw
   * {@link SpeculativeOverlayError} (commit them with `brain.transact()`
   * first).
   *
   * @param path - Absolute directory for the snapshot (created; must be
   *   empty or absent).
   * @throws GenerationConflictError when this view is no longer the latest
   *   generation.
   * @throws SpeculativeOverlayError when this view is a speculative overlay.
   */
  async persist(path: string): Promise<void> {
    this.assertUsable('persist')
    if (this.overlay) {
      throw new SpeculativeOverlayError('persist', this.gen)
    }
    await this.host.persistPinned(path, this.gen)
  }

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

  /**
   * @description Release this view's pin (store + versioned index
   * providers) and free its at-generation index materialization, if one was
   * built. Idempotent. After release, every read throws. Views from
   * `Brainy.load()` / `asOf(path)` also close their underlying read-only
   * brain here. A `FinalizationRegistry` backstop releases leaked pins (and
   * materializations) when a `Db` is garbage-collected, but explicit
   * release is what makes `compactHistory()` deterministic — prefer it.
   */
  async release(): Promise<void> {
    if (this.isReleased) return
    this.isReleased = true
    this.host.unregisterDbFromFinalization(this)
    this.host.releaseGeneration(this.gen)
    if (this.materializedHandle) {
      // A failed materialization already surfaced to the query that
      // triggered it; release() must still succeed.
      const handle = await this.materializedHandle.catch(() => null)
      if (handle) await handle.close()
    }
    if (this.closeOnRelease) {
      await this.closeOnRelease()
    }
  }

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

  /** Whether any transaction committed past the pinned generation. */
  private isHistorical(): boolean {
    return this.host.store.hasCommittedAfter(this.gen)
  }

  /**
   * Build (once) and cache the at-generation index materialization. The
   * first index-accelerated query at a historical pin pays the O(n at G)
   * build; every later one reuses the cached handle until release(). The
   * GC-backstop registration is refreshed so a leaked `Db` also closes its
   * materialized reader (whose flush timers would otherwise outlive it).
   */
  private materialize(): Promise<HistoricalQueryHandle<T>> {
    if (!this.materializedHandle) {
      this.materializedHandle = this.host.materializeAt(this.gen).then((handle) => {
        this.host.unregisterDbFromFinalization(this)
        this.host.registerDbForFinalization(this, this.gen, async () => {
          await handle.close()
          if (this.closeOnRelease) await this.closeOnRelease()
        })
        return handle
      })
      // A failed build must not poison the cache — the next query retries.
      this.materializedHandle = this.materializedHandle.catch((err) => {
        this.materializedHandle = undefined
        throw err
      })
    }
    return this.materializedHandle
  }

  /** Throw on use-after-release. */
  private assertUsable(method: string): void {
    if (this.isReleased) {
      throw new Error(
        `Cannot call ${method}() on a released Db (generation ${this.gen}). ` +
          `Pin a fresh view with brain.now() or brain.asOf().`
      )
    }
  }

  /**
   * Reject find() dimensions that require index machinery a speculative
   * overlay cannot answer (overlay entities carry no embeddings and are in
   * no index — see {@link SpeculativeOverlayError}).
   */
  private assertOverlayCompatibleFind(params: FindParams<T>): void {
    for (const [capability, present] of indexOnlyFindDimensions(params)) {
      if (present) {
        throw new SpeculativeOverlayError(capability, this.gen)
      }
    }
  }
}

/**
 * @description The find() dimensions that only index machinery can answer
 * (everything beyond metadata filters + ordering + offset/limit windows),
 * as `[capability, present]` pairs for the given params.
 */
function indexOnlyFindDimensions(params: FindParams): Array<[string, boolean]> {
  return [
    ['semantic query', params.query !== undefined],
    ['vector search', params.vector !== undefined],
    ['proximity search (near)', params.near !== undefined],
    ['graph traversal (connected)', params.connected !== undefined],
    ['cursor pagination', params.cursor !== undefined],
    ['aggregation', params.aggregate !== undefined],
    ['relation expansion (includeRelations)', params.includeRelations === true],
    [
      `search mode '${params.mode ?? params.searchMode}'`,
      (params.mode !== undefined && params.mode !== 'metadata' && params.mode !== 'auto') ||
        (params.searchMode !== undefined && params.searchMode !== 'auto')
    ]
  ]
}

/**
 * @description Whether the find() params include any index-only dimension —
 * the routing predicate that decides between the record path (cheap,
 * metadata-only) and the at-generation index materialization at historical
 * generations.
 */
function findRequiresIndexes(params: FindParams): boolean {
  return indexOnlyFindDimensions(params).some(([, present]) => present)
}

/**
 * @description Build a `Result` row from a record/overlay-resolved entity,
 * mirroring the live metadata-only find path (flattened fields, score `1.0`).
 */
function resultFromEntity<T>(entity: Entity<T>): Result<T> {
  return {
    id: entity.id,
    score: 1.0,
    type: entity.type,
    subtype: entity.subtype,
    metadata: entity.metadata,
    data: entity.data,
    confidence: entity.confidence,
    weight: entity.weight,
    _rev: entity._rev,
    entity
  }
}

/**
 * @description In-place stable sort of merged results by a find() `orderBy`
 * field, mirroring live ordering semantics (`asc` default, undefined values
 * last in both directions).
 */
function sortResultsBy<T>(results: Result<T>[], orderBy: string, order: 'asc' | 'desc'): void {
  const direction = order === 'desc' ? -1 : 1
  results.sort((a, b) => {
    const va = resolveEntityField(a.entity as Entity, orderBy)
    const vb = resolveEntityField(b.entity as Entity, orderBy)
    if (va === undefined && vb === undefined) return 0
    if (va === undefined) return 1
    if (vb === undefined) return -1
    if (typeof va === 'number' && typeof vb === 'number') return (va - vb) * direction
    const sa = String(va)
    const sb = String(vb)
    return (sa < sb ? -1 : sa > sb ? 1 : 0) * direction
  })
}

/**
 * @description Filter one relation against `RelatedParams`, mirroring
 * the storage-level filter `brain.related()` builds (`from` → sourceId,
 * `to` → targetId, type/subtype set membership, service equality).
 */
function relationMatchesParams<T>(relation: Relation<T>, params: RelatedParams): boolean {
  if (params.from !== undefined && relation.from !== params.from) return false
  if (params.to !== undefined && relation.to !== params.to) return false
  if (params.type !== undefined) {
    const types = Array.isArray(params.type) ? params.type : [params.type]
    if (!types.includes(relation.type)) return false
  }
  if (params.subtype !== undefined) {
    const subtypes = Array.isArray(params.subtype) ? params.subtype : [params.subtype]
    if (relation.subtype === undefined || !subtypes.includes(relation.subtype)) return false
  }
  if (params.service !== undefined && relation.service !== params.service) return false
  return true
}