M6: EventSourcedDocumentStorage with projections, update operators, queries, planner, unique indexes, catch-up

Writes append resolved deltas with optimistic concurrency and self-heal lagging
projections. Reads hit projections only. Index-backed find is property-tested
against unindexed find.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_017Qxbkp4tA7G5SCWjzWuxmE

src/documents/event-sourced.ts

@@ -0,0 +1,588 @@
+import type { IndexValue } from '../core/encoding.js'
+import { escapeSegment, RESERVED, succ, toSegment, unescapeSegment } from '../core/path.js'
+import type { TrieState } from '../core/trie.js'
+import { CapabilityError, ConcurrencyError, UniqueConstraintError } from '../errors.js'
+import { KeyedMutex } from '../core/mutex.js'
+import { KVEventNode } from '../events/kv.js'
+import type { EventNode, ReadOptions, StoredEvent } from '../events/types.js'
+import { scanChildren } from '../kv/children.js'
+import type { BatchOp, KVDriver } from '../kv/driver.js'
+import type { KVNode } from '../kv/node.js'
+import type { AnySpec, ChildOf, KeyOf, KeyString, Spec, ValueOf } from '../types/schema.js'
+import { analyze, matches, type Filter } from './filter.js'
+import { DocIndexes, DocIndexRegistry, type DocIndexDeclarations, type Plan } from './indexes.js'
+import type { ChildDoc, DocKeysOptions, EventSourcedDocumentNode, FindOptions, FindResult, GetOptions, UpdateOptions } from './types.js'
+import { ulid } from './ulid.js'
+import { applyEvent, applyPatch, resolveUpdate, toPatchEvent, type DocEvent, type UpdateSpec } from './update.js'
+import { clone, compareValues, deepEqual, getPath, isPlainObject, setPath, sortKey, unsetPath, type Doc } from './values.js'
+
+export interface EventSourcedDocumentStorageOptions {
+  /** Where projections live. Default: the `~p` root of the event store's KV. Required for non-KV event stores. */
+  projections?: KVNode<AnySpec>
+  /** Path pattern → indexes over that node's direct children. */
+  indexes?: DocIndexDeclarations
+  /** Delete stale index entries found by `find`. Default `true`. */
+  readRepair?: boolean
+  /** Retries after a write conflict (each preceded by a catch-up). Default 5. */
+  maxRetries?: number
+  /** Warn once per filter shape when `find` falls back to a full scan. Default: `NODE_ENV !== 'production'`. */
+  warnOnFullScan?: boolean
+  /** Id generator for `insert`. Default: ULID-like. */
+  newId?: () => string
+  /**
+   * A unique claim whose owner no longer holds the value is only taken over once it is older
+   * than this, since its owner may be mid-write. Default 30 000 ms.
+   */
+  claimGraceMs?: number
+  /** Clock for claim timestamps. Default `Date.now`. */
+  now?: () => number
+}
+
+/** Stored projection of one document. */
+export type Projection = { state: Doc; version: number; commit: number; deleted?: undefined } | { deleted: true; version: number; commit: number; state?: undefined }
+
+interface DocContext {
+  readonly driver: KVDriver
+  readonly sep: string
+  /** Projection-space prefix: projection key of a doc = `projBase + docSelfKey`. */
+  readonly projBase: string
+  readonly registry: DocIndexRegistry
+  readonly readRepair: boolean
+  readonly maxRetries: number
+  readonly warnOnFullScan: boolean
+  readonly warned: Set<string>
+  readonly newId: () => string
+  readonly claimGraceMs: number
+  readonly now: () => number
+  /** Serialises writes to one document within this process, so retries only arise across processes. */
+  readonly writeLock: KeyedMutex
+}
+
+/** Stored value of a unique claim: owner child segment (escaped) and claim time. */
+type Claim = [owner: string, at: number]
+
+interface Target {
+  ev: EventNode<AnySpec>
+  parentEv: EventNode<AnySpec>
+  esc: string
+  /** Projection key of the document. */
+  projKey: string
+  /** Projection-space prefix of its parent (sibling projection key = `parentProj + esc`). */
+  parentProj: string
+  idx: DocIndexes | undefined
+}
+
+type Hit = { esc: string; state: Doc }
+
+const FIND_CHUNK = 128
+
+/**
+ * Document node backed by an event stream per document and a projection per document.
+ * Writes: read projection → resolve delta → append (expectedVersion = projection.version) →
+ * write projection + index entries. Reads hit the projection only.
+ */
+export class EventSourcedDocumentNodeImpl<S extends Spec = AnySpec> implements EventSourcedDocumentNode<S> {
+  readonly path: readonly string[]
+  /** Document-space key (escaped), relative to the store root. */
+  readonly selfKey: string
+  readonly #ctx: DocContext
+  readonly #ev: EventNode<AnySpec>
+  readonly #parentEv: EventNode<AnySpec> | undefined
+  readonly #esc: string
+  readonly #prefix: string
+  readonly #state: TrieState<any>
+  readonly #selfIdx: DocIndexes | undefined
+  readonly #childIdx: DocIndexes | undefined
+
+  /** @internal Use `EventSourcedDocumentStorage` and `child()`. */
+  constructor(
+    ctx: DocContext,
+    ev: EventNode<AnySpec>,
+    parentEv: EventNode<AnySpec> | undefined,
+    path: readonly string[],
+    selfKey: string,
+    esc: string,
+    state: TrieState<any>,
+    selfIdx: DocIndexes | undefined,
+  ) {
+    this.#ctx = ctx
+    this.#ev = ev
+    this.#parentEv = parentEv
+    this.path = path
+    this.selfKey = selfKey
+    this.#esc = esc
+    this.#prefix = path.length === 0 ? '' : selfKey + ctx.sep
+    this.#state = state
+    this.#selfIdx = selfIdx
+    this.#childIdx = ctx.registry.bind(state, ctx.projBase + this.#prefix)
+  }
+
+  child<K extends KeyOf<S>>(key: K): EventSourcedDocumentNodeImpl<ChildOf<S, K>> {
+    const seg = toSegment(key)
+    const esc = escapeSegment(seg, this.#ctx.sep)
+    return new EventSourcedDocumentNodeImpl(this.#ctx, this.#ev.child(seg), this.#ev, [...this.path, seg], this.#prefix + esc, esc, this.#ctx.registry.trie.step(this.#state, seg), this.#childIdx)
+  }
+
+  as<S2 extends Spec>(): EventSourcedDocumentNodeImpl<S2> {
+    return this as unknown as EventSourcedDocumentNodeImpl<S2>
+  }
+
+  // ---------------------------------------------------------------- reads
+
+  get(opts?: GetOptions): Promise<ValueOf<S> | undefined>
+  get<K extends KeyOf<S>>(key: K, opts?: GetOptions): Promise<ValueOf<ChildOf<S, K>> | undefined>
+  async get(...args: unknown[]): Promise<unknown> {
+    const [t, rest] = this.#split(args)
+    const opts = (rest[0] ?? {}) as GetOptions
+    let proj = await this.#projection(t)
+    if (opts.consistent && (await t.ev.version()) > (proj?.version ?? 0)) {
+      await this.#catchUpDoc(t)
+      proj = await this.#projection(t)
+    }
+    return proj?.state
+  }
+
+  history(): AsyncIterable<StoredEvent<DocEvent>>
+  history<K extends KeyOf<S>>(key: K): AsyncIterable<StoredEvent<DocEvent>>
+  history(...args: unknown[]): AsyncIterable<StoredEvent<DocEvent>> {
+    const t = args.length === 0 ? this.#self() : this.#child(args[0])
+    return t.ev.read() as AsyncIterable<StoredEvent<DocEvent>>
+  }
+
+  getAt(version: number): Promise<ValueOf<S> | undefined>
+  getAt<K extends KeyOf<S>>(key: K, version: number): Promise<ValueOf<ChildOf<S, K>> | undefined>
+  async getAt(...args: unknown[]): Promise<unknown> {
+    const t = args.length <= 1 ? this.#self() : this.#child(args[0])
+    const version = args[args.length - 1] as number
+    if (!Number.isInteger(version) || version < 0) throw new TypeError(`Invalid version ${String(version)}`)
+    let state: Doc | undefined
+    for await (const e of t.ev.read({ to: version } satisfies ReadOptions)) state = applyEvent(state, e.data as DocEvent)
+    return state
+  }
+
+  async *find(filter: Filter = {}, opts: FindOptions = {}): AsyncGenerator<FindResult<ChildDoc<S>>> {
+    const skip = opts.skip ?? 0
+    const limit = opts.limit ?? Infinity
+    if (limit <= 0) return
+    const sort = Object.entries(opts.sort ?? {}) as [string, 1 | -1][]
+    const plan = this.#childIdx?.plan(analyze(filter), sort)
+    if (!plan && Object.keys(filter).length > 0) this.#warnFullScan(filter)
+    const source = plan ? this.#indexScan(plan, skip + limit) : this.#projectionScan()
+    const project = opts.projection ? projector(opts.projection) : (d: Doc) => d
+
+    if (sort.length && !plan?.sorted) {
+      // In-memory sort, bounded to skip + limit when a limit is given.
+      const cap = skip + limit
+      const cmp = comparator(sort)
+      const top: Hit[] = []
+      for await (const hit of source) {
+        if (!matches(hit.state, filter)) continue
+        if (top.length >= cap && cmp(hit, top[top.length - 1]!) >= 0) continue
+        let lo = 0
+        let hi = top.length
+        while (lo < hi) {
+          const mid = (lo + hi) >>> 1
+          if (cmp(top[mid]!, hit) <= 0) lo = mid + 1
+          else hi = mid
+        }
+        top.splice(lo, 0, hit)
+        if (top.length > cap) top.pop()
+      }
+      for (let i = skip; i < top.length; i++) yield { key: unescapeSegment(top[i]!.esc), data: project(top[i]!.state) as ChildDoc<S> }
+      return
+    }
+
+    let skipped = 0
+    let emitted = 0
+    for await (const hit of source) {
+      if (!matches(hit.state, filter)) continue
+      if (skipped < skip) {
+        skipped++
+        continue
+      }
+      yield { key: unescapeSegment(hit.esc), data: project(hit.state) as ChildDoc<S> }
+      if (++emitted >= limit) return
+    }
+  }
+
+  async findOne(filter: Filter = {}, opts: FindOptions = {}): Promise<FindResult<ChildDoc<S>> | undefined> {
+    for await (const r of this.find(filter, { ...opts, limit: 1 })) return r
+    return undefined
+  }
+
+  async *keys(opts: DocKeysOptions = {}): AsyncGenerator<KeyString<S>> {
+    let remaining = opts.limit ?? Infinity
+    if (remaining <= 0) return
+    const after = opts.after === undefined ? undefined : toSegment(opts.after)
+    const { prefix, end } = this.#childRange()
+    for await (const hit of scanChildren(this.#ctx.driver, prefix, end, this.#ctx.sep, { after, reverse: opts.reverse, values: true })) {
+      if (!hit.hasOwn || (hit.value as Projection).deleted) continue
+      yield unescapeSegment(hit.esc) as KeyString<S>
+      if (--remaining <= 0) return
+    }
+  }
+
+  // ---------------------------------------------------------------- writes
+
+  set(doc: ValueOf<S>): Promise<void>
+  set<K extends KeyOf<S>>(key: K, doc: ValueOf<ChildOf<S, K>>): Promise<void>
+  async set(...args: unknown[]): Promise<void> {
+    const [t, rest] = this.#split(args)
+    const doc = rest[0]
+    if (!isPlainObject(doc)) throw new TypeError('A document must be a plain object')
+    const snapshot = clone(doc)
+    await this.#write(t, () => ({ op: 'set', doc: snapshot }))
+  }
+
+  async insert(doc: ChildDoc<S>): Promise<string> {
+    if (!isPlainObject(doc)) throw new TypeError('A document must be a plain object')
+    const id = this.#ctx.newId()
+    const t = this.#child(id)
+    const snapshot = clone(doc)
+    await this.#write(t, (cur) => {
+      if (cur) throw new Error(`Document "${id}" already exists`)
+      return { op: 'set', doc: snapshot }
+    })
+    return id
+  }
+
+  update(spec: UpdateSpec, opts?: UpdateOptions): Promise<ValueOf<S> | undefined>
+  update<K extends KeyOf<S>>(key: K, spec: UpdateSpec, opts?: UpdateOptions): Promise<ValueOf<ChildOf<S, K>> | undefined>
+  async update(...args: unknown[]): Promise<unknown> {
+    const [t, rest] = this.#split(args)
+    const spec = rest[0] as UpdateSpec
+    const opts = (rest[1] ?? {}) as UpdateOptions
+    if (!isPlainObject(spec)) throw new TypeError('update() expects an update spec')
+    const result = await this.#write(t, (cur) => {
+      if (!cur && !opts.upsert) return null
+      const base = cur ?? {}
+      const patch = resolveUpdate(base, spec)
+      if (cur && deepEqual(applyPatch(base, patch), cur)) return null // no-op: no event
+      return toPatchEvent(patch)
+    })
+    return result
+  }
+
+  delete(): Promise<boolean>
+  delete<K extends KeyOf<S>>(key: K): Promise<boolean>
+  async delete(...args: unknown[]): Promise<boolean> {
+    const t = args.length === 0 ? this.#self() : this.#child(args[0])
+    let deleted = false
+    await this.#write(t, (cur) => {
+      deleted = cur !== undefined
+      return cur ? { op: 'delete' } : null
+    })
+    return deleted
+  }
+
+  // ---------------------------------------------------------------- maintenance
+
+  async catchUp(): Promise<number> {
+    let repaired = 0
+    if (this.path.length > 0 && (await this.#catchUpDoc(this.#self()))) repaired++
+    for await (const key of this.#ev.keys()) repaired += await this.child(key as KeyOf<S>).catchUp()
+    return repaired
+  }
+
+  // ---------------------------------------------------------------- internals
+
+  #self(): Target {
+    if (this.path.length === 0) throw new TypeError('The root node has no own document')
+    const parentProj = this.#ctx.projBase + this.selfKey.slice(0, this.selfKey.length - this.#esc.length)
+    return { ev: this.#ev, parentEv: this.#parentEv!, esc: this.#esc, projKey: this.#ctx.projBase + this.selfKey, parentProj, idx: this.#selfIdx }
+  }
+
+  #child(key: unknown): Target {
+    const seg = toSegment(key)
+    const esc = escapeSegment(seg, this.#ctx.sep)
+    const parentProj = this.#ctx.projBase + this.#prefix
+    return { ev: this.#ev.child(seg), parentEv: this.#ev, esc, projKey: parentProj + esc, parentProj, idx: this.#childIdx }
+  }
+
+  /** Target for a sibling of `t` given its escaped segment. */
+  #sibling(t: Target, esc: string): Target {
+    return { ev: t.parentEv.child(unescapeSegment(esc)), parentEv: t.parentEv, esc, projKey: t.parentProj + esc, parentProj: t.parentProj, idx: t.idx }
+  }
+
+  /** Self form when there are no args or the first is a non-null object; otherwise key form. */
+  #split(args: unknown[]): [Target, unknown[]] {
+    if (args.length === 0 || (typeof args[0] === 'object' && args[0] !== null)) return [this.#self(), args]
+    return [this.#child(args[0]), args.slice(1)]
+  }
+
+  #childRange(): { prefix: string; end: string | undefined } {
+    const prefix = this.#ctx.projBase + this.#prefix
+    return { prefix, end: prefix === '' ? undefined : prefix.slice(0, -1) + succ(this.#ctx.sep) }
+  }
+
+  async #projection(t: Target): Promise<Projection | undefined> {
+    return (await this.#ctx.driver.getMany([t.projKey]))[0] as Projection | undefined
+  }
+
+  /**
+   * The write path. `compute` maps the current state to an event (or `null` for a no-op).
+   * Returns the resulting state.
+   */
+  #write(t: Target, compute: (cur: Doc | undefined) => DocEvent | null): Promise<Doc | undefined> {
+    return this.#ctx.writeLock.run(t.projKey, () => this.#writeUnlocked(t, compute))
+  }
+
+  async #writeUnlocked(t: Target, compute: (cur: Doc | undefined) => DocEvent | null): Promise<Doc | undefined> {
+    const { maxRetries } = this.#ctx
+    for (let attempt = 0; ; attempt++) {
+      const proj = await this.#projection(t)
+      const cur = proj?.state
+      const event = compute(cur)
+      if (!event) return cur
+      const next = applyEvent(cur, event)
+
+      const oldClaims = t.idx?.claimsFor(cur)
+      const newClaims = t.idx?.claimsFor(next)
+      const acquired: string[] = []
+      if (newClaims?.size) {
+        const wanted = new Map([...newClaims].filter(([k]) => !oldClaims!.has(k)))
+        if (wanted.size) await this.#claim(t, wanted, acquired)
+      }
+
+      let result
+      try {
+        result = await t.ev.append([event], { expectedVersion: proj?.version ?? 0, expectedCommit: proj?.commit ?? 0 })
+      } catch (err) {
+        if (acquired.length) await this.#ctx.driver.deleteMany(acquired)
+        if (err instanceof ConcurrencyError && attempt < maxRetries) {
+          await this.#catchUpDoc(t)
+          continue
+        }
+        throw err
+      }
+      const releases = oldClaims ? [...oldClaims.keys()].filter((k) => !newClaims!.has(k)) : []
+      const projection: Projection = next ? { state: next, version: result.version, commit: result.commit } : { deleted: true, version: result.version, commit: result.commit }
+      await this.#writeProjection(t, cur, next, projection, releases)
+      return next
+    }
+  }
+
+  /**
+   * Claims unique values for `t`; records successful claims in `acquired`. Throws
+   * `UniqueConstraintError` on conflict. A claim held by another document is taken over only if
+   * that document, caught up to its stream head, no longer holds the value and the claim is older
+   * than `claimGraceMs` (its owner might otherwise be between claiming and appending).
+   */
+  async #claim(t: Target, claims: Map<string, { index: string; values: IndexValue[] }>, acquired: string[]): Promise<void> {
+    const { driver, now, claimGraceMs } = this.#ctx
+    const keys = [...claims.keys()]
+    const written = await driver.setIfAbsent!(keys.map((k) => [k, [t.esc, now()] satisfies Claim]))
+    const lost = keys.filter((k, i) => {
+      if (written[i]) acquired.push(k)
+      return !written[i]
+    })
+    if (!lost.length) return
+    const holders = (await driver.getMany(lost)) as (Claim | undefined)[]
+    const fail = async (k: string, owner?: string): Promise<never> => {
+      if (acquired.length) await driver.deleteMany(acquired.splice(0))
+      const c = claims.get(k)!
+      throw new UniqueConstraintError(c.index, c.values, owner === undefined ? undefined : unescapeSegment(owner))
+    }
+    for (let i = 0; i < lost.length; i++) {
+      const k = lost[i]!
+      const holder = holders[i]
+      if (holder?.[0] === t.esc) continue // already ours (e.g. left by an interrupted write)
+      if (holder) {
+        const owner = this.#sibling(t, holder[0])
+        await this.#catchUpDoc(owner)
+        const ownerProj = await this.#projection(owner)
+        if (t.idx!.claimsFor(ownerProj?.state).has(k)) await fail(k, holder[0])
+        if (now() - holder[1] < claimGraceMs) await fail(k, holder[0])
+        await driver.deleteMany([k])
+      }
+      const [ok] = await driver.setIfAbsent!([[k, [t.esc, now()] satisfies Claim]])
+      if (!ok) await fail(k)
+      acquired.push(k)
+    }
+  }
+
+  async #writeProjection(t: Target, cur: Doc | undefined, next: Doc | undefined, projection: Projection, releases: string[]): Promise<void> {
+    const { driver } = this.#ctx
+    if (!t.idx) {
+      await driver.setMany([[t.projKey, projection]])
+      return
+    }
+    const before = t.idx.entriesFor(t.esc, cur)
+    const after = t.idx.entriesFor(t.esc, next)
+    const adds = [...after].filter((e) => !before.has(e))
+    const removes = [...before].filter((e) => !after.has(e)).concat(releases)
+    if (driver.batch) {
+      const ops: BatchOp[] = adds.map((key) => ({ type: 'set', key, value: true }))
+      ops.push({ type: 'set', key: t.projKey, value: projection })
+      for (const key of removes) ops.push({ type: 'delete', key })
+      await driver.batch(ops)
+      return
+    }
+    if (adds.length) await driver.setMany(adds.map((k) => [k, true]))
+    await driver.setMany([[t.projKey, projection]])
+    if (removes.length) await driver.deleteMany(removes)
+  }
+
+  /** Replays events newer than the projection into it. Returns whether anything changed. */
+  async #catchUpDoc(t: Target): Promise<boolean> {
+    const proj = await this.#projection(t)
+    const cur = proj?.state
+    let state = cur
+    let version = proj?.version ?? 0
+    let commit = proj?.commit ?? 0
+    for await (const e of t.ev.read({ from: version + 1 })) {
+      state = applyEvent(state, e.data as DocEvent)
+      version = e.version
+      commit = e.commit
+    }
+    if (version === (proj?.version ?? 0)) return false
+    const oldClaims = t.idx?.claimsFor(cur)
+    const newClaims = t.idx?.claimsFor(state)
+    if (newClaims?.size) {
+      // The log is the source of truth: (re)assert claims for the caught-up state.
+      const missing = [...newClaims.keys()].filter((k) => !oldClaims!.has(k))
+      if (missing.length) await this.#ctx.driver.setMany(missing.map((k) => [k, [t.esc, this.#ctx.now()] satisfies Claim]))
+    }
+    const releases = oldClaims ? [...oldClaims.keys()].filter((k) => !newClaims!.has(k)) : []
+    const projection: Projection = state ? { state, version, commit } : { deleted: true, version, commit }
+    await this.#writeProjection(t, cur, state, projection, releases)
+    return true
+  }
+
+  async *#projectionScan(): AsyncGenerator<Hit> {
+    const { prefix, end } = this.#childRange()
+    for await (const hit of scanChildren(this.#ctx.driver, prefix, end, this.#ctx.sep, { values: true })) {
+      const p = hit.value as Projection | undefined
+      if (hit.hasOwn && p?.state) yield { esc: hit.esc, state: p.state }
+    }
+  }
+
+  async *#indexScan(plan: Plan, want: number): AsyncGenerator<Hit> {
+    const idx = this.#childIdx!
+    const { driver, readRepair } = this.#ctx
+    const parentProj = this.#ctx.projBase + this.#prefix
+    const seen = new Set<string>()
+    let chunk: string[] = []
+    let chunkSize = Math.min(FIND_CHUNK, Math.max(1, want))
+    const it = driver.scan({ gte: plan.gte, lt: plan.lt, reverse: plan.reverse })[Symbol.asyncIterator]()
+
+    async function* flush(): AsyncGenerator<Hit> {
+      const entries = chunk
+      chunk = []
+      chunkSize = FIND_CHUNK
+      const escs = entries.map((e) => idx.childOf(plan.index, e))
+      const projs = (await driver.getMany(escs.map((e) => parentProj + e))) as (Projection | undefined)[]
+      const stale: string[] = []
+      try {
+        for (let i = 0; i < entries.length; i++) {
+          const esc = escs[i]!
+          const state = projs[i]?.state
+          if (!idx.entriesFor(esc, state).has(entries[i]!)) {
+            stale.push(entries[i]!)
+            continue
+          }
+          if (seen.has(esc)) continue
+          seen.add(esc)
+          yield { esc, state: state! }
+        }
+      } finally {
+        if (readRepair && stale.length) await driver.deleteMany(stale)
+      }
+    }
+
+    try {
+      for (;;) {
+        const r = await it.next()
+        if (r.done) break
+        chunk.push(r.value[0])
+        if (chunk.length >= chunkSize) yield* flush()
+      }
+      if (chunk.length) yield* flush()
+    } finally {
+      await it.return?.()
+    }
+  }
+
+  #warnFullScan(filter: Filter): void {
+    if (!this.#ctx.warnOnFullScan) return
+    const shape = JSON.stringify(filterShape(filter))
+    if (this.#ctx.warned.has(shape)) return
+    this.#ctx.warned.add(shape)
+    console.warn(`[potential-storage] find() on "${this.selfKey || '(root)'}" has no usable index; full scan for filter ${shape}`)
+  }
+}
+
+/** Filter with values replaced by `?`, for de-duplicating warnings. */
+function filterShape(f: unknown): unknown {
+  if (Array.isArray(f)) return f.map(filterShape)
+  if (isPlainObject(f)) return Object.fromEntries(Object.entries(f).map(([k, v]) => [k, k.startsWith('$') && (k === '$and' || k === '$or' || k === '$not') ? filterShape(v) : isPlainObject(v) ? filterShape(v) : '?']))
+  return '?'
+}
+
+function comparator(sort: [string, 1 | -1][]): (a: Hit, b: Hit) => number {
+  const tieDir = sort[0]?.[1] ?? 1
+  return (a, b) => {
+    for (const [field, dir] of sort) {
+      const d = compareValues(sortKey(a.state, field, dir), sortKey(b.state, field, dir))
+      if (d !== 0) return d * dir
+    }
+    return a.esc === b.esc ? 0 : (a.esc < b.esc ? -1 : 1) * tieDir
+  }
+}
+
+function projector(spec: Record<string, 0 | 1 | boolean>): (d: Doc) => Doc {
+  const fields = Object.entries(spec)
+  const include = fields.some(([, v]) => v === 1 || v === true)
+  if (include && fields.some(([, v]) => v === 0 || v === false)) throw new TypeError('Projection cannot mix inclusion and exclusion')
+  return (d) => {
+    if (include) {
+      const out: Doc = {}
+      for (const [f] of fields) {
+        const v = getPath(d, f)
+        if (v !== undefined) setPath(out, f, clone(v))
+      }
+      return out
+    }
+    const out = clone(d)
+    for (const [f] of fields) unsetPath(out, f)
+    return out
+  }
+}
+
+/** Event-sourced document storage: the root document node. */
+export class EventSourcedDocumentStorage<S extends Spec = AnySpec> extends EventSourcedDocumentNodeImpl<S> {
+  constructor(events: EventNode<AnySpec>, opts: EventSourcedDocumentStorageOptions = {}) {
+    let driver: KVDriver
+    let sep: string
+    let projBase: string
+    if (opts.projections) {
+      driver = opts.projections.driver
+      sep = opts.projections.separator
+      projBase = opts.projections.prefix
+    } else if (events instanceof KVEventNode) {
+      const kv = events.kv
+      driver = kv.driver
+      sep = kv.separator
+      projBase = kv.prefix + RESERVED.projections + sep
+    } else {
+      throw new TypeError('EventSourcedDocumentStorage needs `projections` when the event store is not KV-backed')
+    }
+    const registry = new DocIndexRegistry(opts.indexes ?? {}, sep)
+    if (registry.hasUnique && !driver.setIfAbsent) throw new CapabilityError('setIfAbsent', 'unique document indexes require it')
+    const ctx: DocContext = {
+      driver,
+      sep,
+      projBase,
+      registry,
+      readRepair: opts.readRepair ?? true,
+      maxRetries: opts.maxRetries ?? 5,
+      warnOnFullScan: opts.warnOnFullScan ?? (typeof process === 'undefined' || process.env?.NODE_ENV !== 'production'),
+      warned: new Set(),
+      newId: opts.newId ?? ulid,
+      claimGraceMs: opts.claimGraceMs ?? 30_000,
+      now: opts.now ?? Date.now,
+      writeLock: new KeyedMutex(),
+    }
+    super(ctx, events, undefined, [], '', '', registry.trie.rootState, undefined)
+  }
+}

src/documents/filter.ts

@@ -0,0 +1,155 @@
+import type { IndexValue } from '../core/encoding.js'
+import { compareValues, deepEqual, isPlainObject, isScalar, resolvePath, typeRank, type Doc } from './values.js'
+
+/**
+ * Query filter (Mongo subset): field conditions with implicit equality and dot paths,
+ * `$eq $ne $gt $gte $lt $lte $in $nin $exists`, and `$and $or $not`.
+ *
+ * Semantics differences from MongoDB (see DESIGN.md §12, M6):
+ * - Comparisons are type-bracketed: `$gt: 5` only matches numbers.
+ * - On array fields, all positive operators in one condition (`$eq $gt $gte $lt $lte $in`) must be
+ *   satisfied by the same element (like `$elemMatch`), so index range scans are exact.
+ */
+export type Filter = { [field: string]: unknown; $and?: Filter[]; $or?: Filter[]; $not?: Filter }
+
+const POSITIVE = new Set(['$eq', '$gt', '$gte', '$lt', '$lte', '$in'])
+const FIELD_LEVEL = new Set(['$ne', '$nin', '$exists', '$not'])
+
+export function isOperatorObject(v: unknown): v is Record<string, unknown> {
+  if (!isPlainObject(v)) return false
+  const keys = Object.keys(v)
+  return keys.length > 0 && keys.every((k) => k.startsWith('$'))
+}
+
+export function matches(doc: unknown, filter: Filter): boolean {
+  for (const key of Object.keys(filter)) {
+    const cond = filter[key]
+    switch (key) {
+      case '$and':
+        if (!(cond as Filter[]).every((f) => matches(doc, f))) return false
+        break
+      case '$or':
+        if (!(cond as Filter[]).some((f) => matches(doc, f))) return false
+        break
+      case '$not':
+        if (matches(doc, cond as Filter)) return false
+        break
+      default:
+        if (key.startsWith('$')) throw new TypeError(`Unknown top-level operator ${key}`)
+        if (!matchesField(resolvePath(doc, key), cond)) return false
+    }
+  }
+  return true
+}
+
+/** Whether the values found at a path satisfy a field condition. */
+export function matchesField(values: unknown[], cond: unknown): boolean {
+  if (!isOperatorObject(cond)) return someCandidate(values, (c) => eq(c, cond))
+  const positive: [string, unknown][] = []
+  for (const [op, arg] of Object.entries(cond)) {
+    if (POSITIVE.has(op)) positive.push([op, arg])
+    else if (!FIELD_LEVEL.has(op)) throw new TypeError(`Unknown operator ${op}`)
+  }
+  for (const [op, arg] of Object.entries(cond)) {
+    switch (op) {
+      case '$ne':
+        if (someCandidate(values, (c) => eq(c, arg))) return false
+        break
+      case '$nin':
+        if (someCandidate(values, (c) => (arg as unknown[]).some((x) => eq(c, x)))) return false
+        break
+      case '$exists':
+        if (values.length > 0 !== Boolean(arg)) return false
+        break
+      case '$not':
+        if (matchesField(values, arg)) return false
+        break
+    }
+  }
+  if (positive.length === 0) return true
+  return someCandidate(values, (c) => positive.every(([op, arg]) => applyPositive(op, c, arg)))
+}
+
+/** Tests each value and, for arrays, each element. Missing paths test `undefined` once. */
+function someCandidate(values: unknown[], test: (c: unknown) => boolean): boolean {
+  if (values.length === 0) return test(undefined)
+  for (const v of values) {
+    if (test(v)) return true
+    if (Array.isArray(v) && v.some(test)) return true
+  }
+  return false
+}
+
+function eq(c: unknown, x: unknown): boolean {
+  if (x === null) return c === null || c === undefined
+  return deepEqual(c, x)
+}
+
+function applyPositive(op: string, c: unknown, arg: unknown): boolean {
+  switch (op) {
+    case '$eq':
+      return eq(c, arg)
+    case '$in':
+      if (!Array.isArray(arg)) throw new TypeError('$in expects an array')
+      return arg.some((x) => eq(c, x))
+    default: {
+      const rc = typeRank(c)
+      if (rc !== typeRank(arg) || rc === 4) return false
+      const d = compareValues(c, arg)
+      return op === '$gt' ? d > 0 : op === '$gte' ? d >= 0 : op === '$lt' ? d < 0 : d <= 0
+    }
+  }
+}
+
+// ------------------------------------------------------------------ planner analysis
+
+export interface RangePred {
+  gt?: IndexValue
+  gte?: IndexValue
+  lt?: IndexValue
+  lte?: IndexValue
+}
+
+export interface FieldPred {
+  eq?: IndexValue
+  range?: RangePred
+}
+
+/**
+ * Index-usable predicates per field, from the top level and top-level `$and`.
+ * Only scalar `$eq`/implicit equality and same-typed scalar range bounds qualify.
+ */
+export function analyze(filter: Filter, out = new Map<string, FieldPred>()): Map<string, FieldPred> {
+  for (const key of Object.keys(filter)) {
+    const cond = filter[key]
+    if (key === '$and') {
+      for (const f of cond as Filter[]) analyze(f, out)
+      continue
+    }
+    if (key.startsWith('$')) continue
+    const pred = out.get(key) ?? {}
+    if (!isOperatorObject(cond)) {
+      if (isScalar(cond)) pred.eq ??= cond
+    } else {
+      if ('$eq' in cond && isScalar(cond.$eq)) pred.eq ??= cond.$eq
+      const range: RangePred = {}
+      let rank = -1
+      let ok = false
+      for (const op of ['$gt', '$gte', '$lt', '$lte'] as const) {
+        if (!(op in cond)) continue
+        const v = cond[op]
+        if (!isScalar(v)) continue
+        const r = typeRank(v)
+        if (rank !== -1 && r !== rank) continue
+        rank = r
+        range[op.slice(1) as keyof RangePred] = v
+        ok = true
+      }
+      if (ok && !pred.range) pred.range = range
+    }
+    if (pred.eq !== undefined || pred.range) out.set(key, pred)
+  }
+  return out
+}
+
+export type { Doc }

src/documents/index.ts

@@ -0,0 +1,7 @@
+export * from './types.js'
+export * from './event-sourced.js'
+export * from './filter.js'
+export * from './update.js'
+export * from './indexes.js'
+export * from './ulid.js'
+export { compareValues, deepEqual, type Doc } from './values.js'

src/documents/indexes.ts

@@ -0,0 +1,215 @@
+import { encodeIndexValues, indexValueCodec, type IndexValue } from '../core/encoding.js'
+import { escapeSegment, RESERVED, succ } from '../core/path.js'
+import { PatternTrie, patternsOverlap, type TrieState } from '../core/trie.js'
+import type { FieldPred, RangePred } from './filter.js'
+import { indexScalars, resolvePath, typeRank, type Doc } from './values.js'
+
+export interface DocIndexSpec {
+  /** Default: the fields joined with `,`. */
+  name?: string
+  fields: string[]
+  /** Claims `~u/<name>/<values>` per value tuple. Documents where every field is missing/null are not claimed. */
+  unique?: boolean
+}
+
+/** Path pattern (`*` = one segment, `''` = root) → indexes over that node's direct children. */
+export type DocIndexDeclarations = Record<string, DocIndexSpec[]>
+
+export interface SortSpec {
+  [field: string]: 1 | -1
+}
+
+interface CompiledDocIndex {
+  name: string
+  fields: string[]
+  segs: string[][]
+  unique: boolean
+  /** `escapeSegment(name) + sep` */
+  head: string
+}
+
+export interface Plan {
+  index: string
+  gte: string
+  lt: string
+  reverse: boolean
+  /** Index order satisfies the requested sort. */
+  sorted: boolean
+}
+
+/** Document indexes over the children of one node. */
+export class DocIndexes {
+  readonly #list: CompiledDocIndex[]
+  readonly #byName: Map<string, CompiledDocIndex>
+  readonly #enc: (v: IndexValue) => string
+
+  constructor(
+    list: CompiledDocIndex[],
+    /** `<projBase><nodePrefix>~i<sep>` */
+    readonly base: string,
+    /** `<projBase><nodePrefix>~u<sep>` */
+    readonly uniqueBase: string,
+    readonly sep: string,
+  ) {
+    this.#list = list
+    this.#byName = new Map(list.map((i) => [i.name, i]))
+    const codec = indexValueCodec(sep)
+    this.#enc = (v) => codec.encode(v)
+  }
+
+  get hasUnique(): boolean {
+    return this.#list.some((i) => i.unique)
+  }
+
+  /** Value tuples a state produces for one index (cartesian over multikey fields). */
+  #tuples(idx: CompiledDocIndex, state: Doc): IndexValue[][] {
+    let tuples: IndexValue[][] = [[]]
+    let multi = 0
+    for (const segs of idx.segs) {
+      const found = resolvePath(state, segs)
+      const scalars = found.length === 0 ? [null] : [...new Set(found.flatMap(indexScalars))]
+      if (scalars.length > 1 && ++multi > 1) throw new TypeError(`Index "${idx.name}": cannot index more than one array field per document`)
+      tuples = tuples.flatMap((t) => scalars.map((s) => [...t, s]))
+    }
+    return tuples
+  }
+
+  entriesFor(escChild: string, state: Doc | undefined, out: Set<string> = new Set()): Set<string> {
+    if (!state) return out
+    for (const idx of this.#list) {
+      const stem = this.base + idx.head
+      for (const t of this.#tuples(idx, state)) out.add(stem + encodeIndexValues(t, this.sep) + this.sep + escChild)
+    }
+    return out
+  }
+
+  /** Unique claim key → index name and values. */
+  claimsFor(state: Doc | undefined): Map<string, { index: string; values: IndexValue[] }> {
+    const out = new Map<string, { index: string; values: IndexValue[] }>()
+    if (!state) return out
+    for (const idx of this.#list) {
+      if (!idx.unique) continue
+      for (const t of this.#tuples(idx, state)) {
+        if (t.every((v) => v === null)) continue
+        out.set(this.uniqueBase + idx.head + encodeIndexValues(t, this.sep), { index: idx.name, values: t })
+      }
+    }
+    return out
+  }
+
+  /** Escaped child segment of an entry key of index `name`. */
+  childOf(name: string, entryKey: string): string {
+    const idx = this.#byName.get(name)!
+    let pos = this.base.length + idx.head.length
+    for (let i = 0; i < idx.fields.length; i++) pos = entryKey.indexOf(this.sep, pos) + 1
+    return entryKey.slice(pos)
+  }
+
+  /**
+   * Chooses the index covering the most leading equality fields, then at most one range field.
+   * Ties prefer an index whose order satisfies `sort`. With no usable predicate, an index whose
+   * order satisfies `sort` is still used (a full, ordered index scan). Returns `undefined` for a
+   * full projection scan.
+   */
+  plan(preds: Map<string, FieldPred>, sort: [string, 1 | -1][]): Plan | undefined {
+    let best: { idx: CompiledDocIndex; k: number; range: RangePred | undefined; sorted: boolean; dir: 1 | -1; score: number } | undefined
+    for (const idx of this.#list) {
+      let k = 0
+      while (k < idx.fields.length && preds.get(idx.fields[k]!)?.eq !== undefined) k++
+      const range = k < idx.fields.length ? preds.get(idx.fields[k]!)?.range : undefined
+      // Sort fields constrained by equality are constant; drop them before matching.
+      const eqFields = new Set(idx.fields.slice(0, k))
+      const rest = sort.filter(([f]) => !eqFields.has(f))
+      // Ties are broken by child key in the direction of the first sort field.
+      const dir = rest[0]?.[1] ?? sort[0]?.[1] ?? 1
+      const sorted = rest.length === 0 || (rest.every(([f, d], i) => idx.fields[k + i] === f && d === dir) && rest.length <= idx.fields.length - k)
+      const score = k * 2 + (range ? 1 : 0)
+      if (score === 0 && !(sorted && rest.length > 0)) continue
+      if (!best || score > best.score || (score === best.score && sorted && !best.sorted)) best = { idx, k, range, sorted, dir, score }
+    }
+    if (!best) return undefined
+    const { idx, k, range, sorted, dir } = best
+    const succSep = succ(this.sep)
+    let prefix = this.base + idx.head
+    for (let i = 0; i < k; i++) prefix += this.#enc(preds.get(idx.fields[i]!)!.eq!) + this.sep
+    // prefix ends with sep: the range of everything under it is [prefix, prefix-without-sep + succ(sep)).
+    const all = { gte: prefix, lt: prefix.slice(0, -1) + succSep }
+    let { gte, lt } = all
+    if (range) {
+      const sample = (range.gt ?? range.gte ?? range.lt ?? range.lte) as IndexValue
+      const [lo, hi] = typeBounds(typeRank(sample), this.#enc)
+      gte = prefix + lo
+      lt = hi === undefined ? all.lt : prefix + hi
+      if (range.gte !== undefined) gte = max(gte, prefix + this.#enc(range.gte))
+      if (range.gt !== undefined) gte = max(gte, prefix + this.#enc(range.gt) + succSep)
+      if (range.lt !== undefined) lt = min(lt, prefix + this.#enc(range.lt))
+      if (range.lte !== undefined) lt = min(lt, prefix + this.#enc(range.lte) + succSep)
+    }
+    return { index: idx.name, gte, lt, reverse: sorted && dir === -1, sorted }
+  }
+}
+
+/** Encoded `[lower, upper)` of one type class; `undefined` upper = end of index. */
+function typeBounds(rank: number, enc: (v: IndexValue) => string): [string, string | undefined] {
+  switch (rank) {
+    case 0:
+      return [enc(null), enc(false)]
+    case 1:
+      return [enc(false), enc(-Infinity)]
+    case 2:
+      return [enc(-Infinity), enc('')]
+    default:
+      return [enc(''), undefined]
+  }
+}
+
+const max = (a: string, b: string) => (a > b ? a : b)
+const min = (a: string, b: string) => (a < b ? a : b)
+
+export class DocIndexRegistry {
+  readonly trie: PatternTrie<CompiledDocIndex[]>
+  readonly empty: boolean
+  readonly hasUnique: boolean
+
+  constructor(
+    decls: DocIndexDeclarations,
+    readonly sep: string,
+  ) {
+    this.trie = new PatternTrie(sep)
+    const patterns = Object.keys(decls)
+    const compiled = new Map<string, CompiledDocIndex[]>()
+    for (const p of patterns) {
+      const list = decls[p]!.map((spec): CompiledDocIndex => {
+        if (!spec.fields.length) throw new TypeError(`Index on "${p}" has no fields`)
+        const name = spec.name ?? spec.fields.join(',')
+        return { name, fields: spec.fields, segs: spec.fields.map((f) => f.split('.')), unique: spec.unique === true, head: escapeSegment(name, sep) + sep }
+      })
+      const names = list.map((i) => i.name)
+      const dup = names.find((n, i) => names.indexOf(n) !== i)
+      if (dup) throw new TypeError(`Duplicate index name "${dup}" on "${p}"`)
+      compiled.set(p, list)
+    }
+    for (let i = 0; i < patterns.length; i++) {
+      for (let j = i + 1; j < patterns.length; j++) {
+        const a = patterns[i]!
+        const b = patterns[j]!
+        if (!patternsOverlap(a, b, sep)) continue
+        const bn = new Set(compiled.get(b)!.map((x) => x.name))
+        const clash = compiled.get(a)!.filter((x) => bn.has(x.name))
+        if (clash.length) throw new TypeError(`Index name(s) ${clash.map((c) => c.name).join(', ')} declared on overlapping patterns "${a}" and "${b}"`)
+      }
+    }
+    for (const p of patterns) this.trie.add(p, compiled.get(p)!)
+    this.empty = patterns.length === 0
+    this.hasUnique = [...compiled.values()].some((l) => l.some((i) => i.unique))
+  }
+
+  /** `projPrefix` is the projection-space prefix of the node (`<projBase><nodePrefix>`). */
+  bind(state: TrieState<CompiledDocIndex[]>, projPrefix: string): DocIndexes | undefined {
+    if (this.empty) return undefined
+    const found = PatternTrie.values(state)
+    if (found.length === 0) return undefined
+    const list = found.flatMap((f) => f.value)
+    return new DocIndexes(list, projPrefix + RESERVED.index + this.sep, projPrefix + RESERVED.unique + this.sep, this.sep)
+  }
+}

src/documents/types.ts

@@ -0,0 +1,90 @@
+import type { StoredEvent } from '../events/types.js'
+import type { AnySpec, ChildOf, KeyOf, KeyString, Spec, ValueOf } from '../types/schema.js'
+import type { Filter } from './filter.js'
+import type { SortSpec } from './indexes.js'
+import type { DocEvent, UpdateSpec } from './update.js'
+
+export interface FindOptions {
+  sort?: SortSpec
+  limit?: number
+  skip?: number
+  /** Inclusion (`{ a: 1 }`) or exclusion (`{ a: 0 }`) of dot-path fields. */
+  projection?: Record<string, 0 | 1 | boolean>
+}
+
+export interface FindResult<T> {
+  key: string
+  data: T
+}
+
+export interface GetOptions {
+  /** Check the stream head and catch the projection up first (one extra scan). */
+  consistent?: boolean
+}
+
+export interface UpdateOptions {
+  /** Treat a missing document as `{}`. */
+  upsert?: boolean
+}
+
+export interface DocKeysOptions {
+  limit?: number
+  after?: string | number
+  reverse?: boolean
+}
+
+/** Document type of a node's children. */
+export type ChildDoc<S> = ValueOf<ChildOf<S, KeyOf<S>>>
+
+/**
+ * One node of a document store: a document (its own) and a collection of child documents.
+ * Same navigation and self/child arity rules as `KVNode`; in forms with an optional leading key,
+ * a non-null object first argument selects the self form.
+ */
+export interface DocumentNode<S extends Spec = AnySpec> {
+  readonly path: readonly string[]
+  child<K extends KeyOf<S>>(key: K): DocumentNode<ChildOf<S, K>>
+  as<S2 extends Spec>(): DocumentNode<S2>
+
+  get(opts?: GetOptions): Promise<ValueOf<S> | undefined>
+  get<K extends KeyOf<S>>(key: K, opts?: GetOptions): Promise<ValueOf<ChildOf<S, K>> | undefined>
+
+  set(doc: ValueOf<S>): Promise<void>
+  set<K extends KeyOf<S>>(key: K, doc: ValueOf<ChildOf<S, K>>): Promise<void>
+
+  /** Inserts a child with a generated time-sortable id; returns the id. */
+  insert(doc: ChildDoc<S>): Promise<string>
+
+  /** Returns the new state, or `undefined` if the document doesn't exist (and no `upsert`). */
+  update(spec: UpdateSpec, opts?: UpdateOptions): Promise<ValueOf<S> | undefined>
+  update<K extends KeyOf<S>>(key: K, spec: UpdateSpec, opts?: UpdateOptions): Promise<ValueOf<ChildOf<S, K>> | undefined>
+
+  /** Returns whether a document was deleted. */
+  delete(): Promise<boolean>
+  delete<K extends KeyOf<S>>(key: K): Promise<boolean>
+
+  /** Queries direct children. */
+  find(filter?: Filter, opts?: FindOptions): AsyncIterable<FindResult<ChildDoc<S>>>
+  findOne(filter?: Filter, opts?: FindOptions): Promise<FindResult<ChildDoc<S>> | undefined>
+
+  /** Keys of direct children that hold a (non-deleted) document. */
+  keys(opts?: DocKeysOptions): AsyncIterable<KeyString<S>>
+}
+
+/** Document store backed by an event log: adds history and point-in-time reads. */
+export interface EventSourcedDocumentNode<S extends Spec = AnySpec> extends DocumentNode<S> {
+  child<K extends KeyOf<S>>(key: K): EventSourcedDocumentNode<ChildOf<S, K>>
+  as<S2 extends Spec>(): EventSourcedDocumentNode<S2>
+
+  history(): AsyncIterable<StoredEvent<DocEvent>>
+  history<K extends KeyOf<S>>(key: K): AsyncIterable<StoredEvent<DocEvent>>
+
+  /** State as of `version` (`undefined` if missing or deleted then). */
+  getAt(version: number): Promise<ValueOf<S> | undefined>
+  getAt<K extends KeyOf<S>>(key: K, version: number): Promise<ValueOf<ChildOf<S, K>> | undefined>
+
+  /** Rebuilds lagging projections for this node and all descendants; returns how many were repaired. */
+  catchUp(): Promise<number>
+}
+
+export type DocumentStorage<S extends Spec = AnySpec> = DocumentNode<S>

src/documents/ulid.ts

@@ -0,0 +1,37 @@
+const ALPHABET = '0123456789ABCDEFGHJKMNPQRSTVWXYZ' // Crockford base32
+
+/**
+ * ULID-like id generator: 10 chars of millisecond timestamp + 16 chars of randomness,
+ * lexicographically sortable by creation time. Ids generated within the same millisecond
+ * increment the random part, so they stay strictly increasing per generator.
+ */
+export function createIdGenerator(now: () => number = Date.now): () => string {
+  let lastTime = -1
+  const rand = new Uint8Array(16) // one base32 digit (0–31) per byte
+  return () => {
+    let t = now()
+    if (t <= lastTime) {
+      t = lastTime
+      // Increment the random part (base 32, right to left).
+      let i = 15
+      while (i >= 0 && rand[i] === 31) rand[i--] = 0
+      if (i < 0) throw new Error('ULID random overflow within one millisecond')
+      rand[i]!++
+    } else {
+      crypto.getRandomValues(rand)
+      for (let i = 0; i < 16; i++) rand[i]! &= 31
+      // Leave headroom so increments rarely overflow.
+      rand[0]! &= 15
+      lastTime = t
+    }
+    let out = ''
+    for (let i = 9; i >= 0; i--) {
+      out = ALPHABET[t % 32]! + out
+      t = Math.floor(t / 32)
+    }
+    for (let i = 0; i < 16; i++) out += ALPHABET[rand[i]!]
+    return out
+  }
+}
+
+export const ulid = createIdGenerator()

src/documents/update.ts

@@ -0,0 +1,113 @@
+import { isOperatorObject, matches, matchesField, type Filter } from './filter.js'
+import { clone, deepEqual, getPath, isPlainObject, setPath, unsetPath, type Doc } from './values.js'
+
+/** Update operators supported in v1. Paths may use dots. */
+export interface UpdateSpec {
+  $set?: Record<string, unknown>
+  $unset?: Record<string, unknown>
+  $inc?: Record<string, number>
+  $push?: Record<string, unknown>
+  $pull?: Record<string, unknown>
+  $addToSet?: Record<string, unknown>
+}
+
+/** Stored document events: resolved deltas, never operators, so replay is deterministic. */
+export type DocEvent =
+  | { op: 'set'; doc: Doc }
+  | { op: 'patch'; set?: Record<string, unknown>; unset?: string[] }
+  | { op: 'delete' }
+
+export interface Patch {
+  set: Record<string, unknown>
+  unset: string[]
+}
+
+const OPS = new Set(['$set', '$unset', '$inc', '$push', '$pull', '$addToSet'])
+
+/** Resolves operators against `state` into a patch of final values. */
+export function resolveUpdate(state: Doc, spec: UpdateSpec): Patch {
+  const keys = Object.keys(spec)
+  if (keys.length === 0) throw new TypeError('Empty update')
+  for (const k of keys) if (!OPS.has(k)) throw new TypeError(`Unsupported update operator ${k}`)
+
+  const patch: Patch = { set: {}, unset: [] }
+  const touched: string[] = []
+  const touch = (path: string) => {
+    for (const t of touched) {
+      if (t === path || t.startsWith(path + '.') || path.startsWith(t + '.')) throw new TypeError(`Update paths conflict: "${t}" and "${path}"`)
+    }
+    touched.push(path)
+  }
+  const arrayAt = (path: string, op: string): unknown[] => {
+    const cur = getPath(state, path)
+    if (cur === undefined || cur === null) return []
+    if (!Array.isArray(cur)) throw new TypeError(`${op} on non-array field "${path}"`)
+    return cur
+  }
+  const eachOf = (v: unknown): unknown[] =>
+    isPlainObject(v) && Object.keys(v).length === 1 && Array.isArray(v.$each) ? (v.$each as unknown[]) : [v]
+
+  for (const [path, v] of Object.entries(spec.$set ?? {})) {
+    touch(path)
+    patch.set[path] = clone(v)
+  }
+  for (const path of Object.keys(spec.$unset ?? {})) {
+    touch(path)
+    patch.unset.push(path)
+  }
+  for (const [path, by] of Object.entries(spec.$inc ?? {})) {
+    touch(path)
+    if (typeof by !== 'number') throw new TypeError(`$inc by non-number at "${path}"`)
+    const cur = getPath(state, path)
+    if (cur !== undefined && cur !== null && typeof cur !== 'number') throw new TypeError(`$inc on non-number field "${path}"`)
+    patch.set[path] = ((cur as number | undefined) ?? 0) + by
+  }
+  for (const [path, v] of Object.entries(spec.$push ?? {})) {
+    touch(path)
+    patch.set[path] = [...clone(arrayAt(path, '$push')), ...clone(eachOf(v))]
+  }
+  for (const [path, v] of Object.entries(spec.$addToSet ?? {})) {
+    touch(path)
+    const next = clone(arrayAt(path, '$addToSet'))
+    for (const item of eachOf(v)) if (!next.some((x) => deepEqual(x, item))) next.push(clone(item))
+    patch.set[path] = next
+  }
+  for (const [path, cond] of Object.entries(spec.$pull ?? {})) {
+    touch(path)
+    patch.set[path] = clone(arrayAt(path, '$pull')).filter((el) => !pullMatches(el, cond))
+  }
+  return patch
+}
+
+function pullMatches(el: unknown, cond: unknown): boolean {
+  if (isOperatorObject(cond)) return matchesField([el], cond)
+  if (isPlainObject(cond) && isPlainObject(el)) return matches(el, cond as Filter)
+  return deepEqual(el, cond)
+}
+
+/** Applies a resolved patch to a copy of `state`. */
+export function applyPatch(state: Doc, patch: { set?: Record<string, unknown>; unset?: string[] }): Doc {
+  const next = clone(state)
+  for (const [path, v] of Object.entries(patch.set ?? {})) setPath(next, path, clone(v))
+  for (const path of patch.unset ?? []) unsetPath(next, path)
+  return next
+}
+
+/** Folds one event into a state (`undefined` = no document). */
+export function applyEvent(state: Doc | undefined, e: DocEvent): Doc | undefined {
+  switch (e.op) {
+    case 'set':
+      return clone(e.doc)
+    case 'patch':
+      return applyPatch(state ?? {}, e)
+    case 'delete':
+      return undefined
+  }
+}
+
+export function toPatchEvent(p: Patch): DocEvent {
+  const e: { op: 'patch'; set?: Record<string, unknown>; unset?: string[] } = { op: 'patch' }
+  if (Object.keys(p.set).length) e.set = p.set
+  if (p.unset.length) e.unset = p.unset
+  return e
+}

src/documents/values.ts

@@ -0,0 +1,157 @@
+import type { IndexValue } from '../core/encoding.js'
+
+/** Plain JSON object. */
+export type Doc = Record<string, unknown>
+
+export function isPlainObject(v: unknown): v is Doc {
+  return typeof v === 'object' && v !== null && !Array.isArray(v)
+}
+
+export function isScalar(v: unknown): v is IndexValue {
+  return v === null || typeof v === 'string' || typeof v === 'boolean' || (typeof v === 'number' && !Number.isNaN(v))
+}
+
+export function splitPath(path: string): string[] {
+  return path.split('.')
+}
+
+/**
+ * Values reachable at a dot path. Arrays are traversed: a non-numeric segment applied to an
+ * array maps over its elements. Missing paths yield no values.
+ */
+export function resolvePath(doc: unknown, path: string | readonly string[]): unknown[] {
+  const segs = typeof path === 'string' ? splitPath(path) : path
+  const out: unknown[] = []
+  walk(doc, segs, 0, out)
+  return out
+}
+
+function walk(v: unknown, segs: readonly string[], i: number, out: unknown[]): void {
+  if (i === segs.length) {
+    if (v !== undefined) out.push(v)
+    return
+  }
+  const seg = segs[i]!
+  if (Array.isArray(v)) {
+    if (/^\d+$/.test(seg)) walk(v[Number(seg)], segs, i + 1, out)
+    else for (const el of v) if (isPlainObject(el) || Array.isArray(el)) walk(el, segs, i, out)
+    return
+  }
+  if (isPlainObject(v) && Object.hasOwn(v, seg)) walk(v[seg], segs, i + 1, out)
+}
+
+/** Value at a dot path without array traversal (numeric segments index arrays). */
+export function getPath(doc: unknown, path: string): unknown {
+  let v = doc
+  for (const seg of splitPath(path)) {
+    if (Array.isArray(v) && /^\d+$/.test(seg)) v = v[Number(seg)]
+    else if (isPlainObject(v) && Object.hasOwn(v, seg)) v = v[seg]
+    else return undefined
+  }
+  return v
+}
+
+/** Sets a dot path in place, creating intermediate objects. */
+export function setPath(doc: Doc, path: string, value: unknown): void {
+  const segs = splitPath(path)
+  let v: unknown = doc
+  for (let i = 0; i < segs.length - 1; i++) {
+    const seg = segs[i]!
+    const container = v as Record<string, unknown>
+    let next = Array.isArray(container) && /^\d+$/.test(seg) ? container[Number(seg)] : container[seg]
+    if (next === undefined || next === null) {
+      next = {}
+      if (Array.isArray(container)) container[Number(seg)] = next
+      else container[seg] = next
+    } else if (typeof next !== 'object') {
+      throw new TypeError(`Cannot set "${path}": "${segs.slice(0, i + 1).join('.')}" is not an object`)
+    }
+    v = next
+  }
+  const last = segs.at(-1)!
+  if (Array.isArray(v) && /^\d+$/.test(last)) v[Number(last)] = value
+  else (v as Doc)[last] = value
+}
+
+export function unsetPath(doc: Doc, path: string): void {
+  const segs = splitPath(path)
+  let v: unknown = doc
+  for (let i = 0; i < segs.length - 1; i++) {
+    v = isPlainObject(v) ? v[segs[i]!] : Array.isArray(v) && /^\d+$/.test(segs[i]!) ? v[Number(segs[i])] : undefined
+    if (typeof v !== 'object' || v === null) return
+  }
+  const last = segs.at(-1)!
+  if (Array.isArray(v) && /^\d+$/.test(last)) v[Number(last)] = null // Mongo semantics: keep positions
+  else if (isPlainObject(v)) delete v[last]
+}
+
+export function deepEqual(a: unknown, b: unknown): boolean {
+  if (a === b) return true
+  if (typeof a !== 'object' || typeof b !== 'object' || a === null || b === null) return false
+  if (Array.isArray(a) !== Array.isArray(b)) return false
+  if (Array.isArray(a)) {
+    const bb = b as unknown[]
+    return a.length === bb.length && a.every((x, i) => deepEqual(x, bb[i]))
+  }
+  const ka = Object.keys(a)
+  const kb = Object.keys(b)
+  return ka.length === kb.length && ka.every((k) => Object.hasOwn(b, k) && deepEqual((a as Doc)[k], (b as Doc)[k]))
+}
+
+export function clone<T>(v: T): T {
+  return v === null || typeof v !== 'object' ? v : structuredClone(v)
+}
+
+/** Type class used for bracketing comparisons, in index-encoding order. */
+export function typeRank(v: unknown): number {
+  if (v === null || v === undefined) return 0
+  if (typeof v === 'boolean') return 1
+  if (typeof v === 'number') return 2
+  if (typeof v === 'string') return 3
+  return 4
+}
+
+/**
+ * Total order matching the index encoding for scalars:
+ * `null/missing < false < true < numbers < strings < anything else`.
+ */
+export function compareValues(a: unknown, b: unknown): number {
+  const ra = typeRank(a)
+  const rb = typeRank(b)
+  if (ra !== rb) return ra < rb ? -1 : 1
+  if (ra === 1 || ra === 2 || ra === 3) return a === b ? 0 : (a as number) < (b as number) ? -1 : 1
+  if (ra === 4) {
+    const sa = JSON.stringify(a)
+    const sb = JSON.stringify(b)
+    return sa === sb ? 0 : sa < sb ? -1 : 1
+  }
+  return 0
+}
+
+/**
+ * The scalar(s) an index stores for a field value: scalars as-is, arrays as their scalar
+ * elements (multikey; empty → null), and missing or non-scalar values as `null`.
+ */
+export function indexScalars(v: unknown): IndexValue[] {
+  if (Array.isArray(v)) {
+    const out: IndexValue[] = []
+    for (const el of v) out.push(isScalar(el) ? el : null)
+    return out.length ? out : [null]
+  }
+  return [isScalar(v) ? v : null]
+}
+
+/** Sort key for a field under a direction; arrays sort by their min (asc) or max (desc) element. */
+export function sortKey(doc: unknown, path: string, dir: 1 | -1): unknown {
+  const vals = resolvePath(doc, path)
+  if (vals.length === 0) return null
+  let best: unknown
+  let first = true
+  for (const v of vals) {
+    for (const s of indexScalars(v)) {
+      if (first || compareValues(s, best) * dir < 0) best = s
+      first = false
+    }
+  }
+  return best
+}

src/events/kv.ts

@@ -96,7 +96,8 @@ export class KVEventNode<S extends Spec = AnySpec> implements EventNode<S> {
     if (!Array.isArray(events)) throw new TypeError('append() expects an array of events')
     const expected = opts.expectedVersion ?? 'any'
     if (expected !== 'any' && !(Number.isInteger(expected) && expected >= 0)) throw new TypeError(`Invalid expectedVersion ${String(expected)}`)
-    return this.#append(stream, events, expected)
+    const hint = expected !== 'any' && Number.isInteger(opts.expectedCommit) ? { commit: opts.expectedCommit!, version: expected } : undefined
+    return this.#append(stream, events, expected, hint)
   }
 
   read(opts?: ReadOptions): AsyncIterable<StoredEvent<ValueOf<S>>>
@@ -145,7 +146,7 @@ export class KVEventNode<S extends Spec = AnySpec> implements EventNode<S> {
     return EMPTY_HEAD
   }
 
-  async #append(s: Stream, events: readonly unknown[], expected: number | 'any'): Promise<AppendResult> {
+  async #append(s: Stream, events: readonly unknown[], expected: number | 'any', hint?: Head): Promise<AppendResult> {
     const { driver, heads, maxRetries } = this.#ctx
     if (events.length === 0) {
       const head = await this.#loadHead(s)
@@ -155,7 +156,7 @@ export class KVEventNode<S extends Spec = AnySpec> implements EventNode<S> {
     if (!driver.setIfAbsent) return this.#ctx.mutex.run(s.key, () => this.#appendSingleWriter(s, events, expected))
 
     for (let attempt = 0; ; attempt++) {
-      let head = heads.get(s.key)
+      let head = heads.get(s.key) ?? (attempt === 0 ? hint : undefined)
       if (!head) head = await this.#loadHead(s)
       else if (expected !== 'any' && head.version !== expected) head = await this.#loadHead(s) // cache may be stale
       if (expected !== 'any' && head.version !== expected) throw new ConcurrencyError(s.key, expected, head.version)

src/events/types.ts

@@ -3,6 +3,13 @@ import type { AnySpec, ChildOf, KeyOf, KeyString, Spec, ValueOf } from '../types
 export interface AppendOptions {
   /** Version the stream must be at (0 = empty). Default `'any'`. */
   expectedVersion?: number | 'any'
+  /**
+   * Commit sequence of the head at `expectedVersion`, if the caller knows it (e.g. from a
+   * projection). Lets a cold append skip the head lookup. Must come from the stream itself: a
+   * stale hint only causes a collision, but an invented one could leave a gap. Ignored unless
+   * `expectedVersion` is a number; backends may ignore it.
+   */
+  expectedCommit?: number
 }
 
 export interface AppendResult {

src/index.ts

@@ -4,3 +4,4 @@ export * from './core/encoding.js'
 export * from './kv/index.js'
 export * from './types/schema.js'
 export * from './events/index.js'
+export * from './documents/index.js'