Simplify: one abstract base class per layer

- BaseKVStorage / BaseEventStorage / BaseDocumentStorage define the API once;
  implementations provide only raw primitives (client and server share an API)
- Drop KV indexes, query planner, unique indexes; document stores take index fields
- Escape every char <= '/' so child keys stay contiguous; skip-scan is now simple
- Event-sourced documents = event storage + any document store for projections
- New folder layout (kv-storage, event-storage, document-storage, utils); tests only in test/

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

src/document-storage/base-document-storage.ts

@@ -0,0 +1,117 @@
+import type { Doc } from '../utils/document-path.js'
+import { DocumentQuery, type Filter, type FindOptions, type FindResult } from '../utils/document-query.js'
+import { DocumentUpdate, type DocChange, type UpdateSpec } from '../utils/document-update.js'
+import { Keys } from '../utils/keys.js'
+import type { AnySpec, ChildOf, KeyOf, Spec, ValueAt, ValueOf } from '../utils/schema.js'
+import { Ulid } from '../utils/ulid.js'
+
+/**
+ * Fields to index, per collection path pattern (`*` matches one segment), e.g.
+ * `{ users: ['email', 'age'], 'users/*\/posts': ['createdAt'] }`. Each implementation decides
+ * how (or whether) to use them; a native database maps them onto its own indexes.
+ */
+export type IndexDeclarations = Record<string, string[]>
+
+export interface UpdateOptions {
+  /** Treat a missing document as `{}`. */
+  upsert?: boolean
+}
+
+/** Document type of a node's children. */
+export type ChildDoc<S> = ValueOf<ChildOf<S, KeyOf<S>>>
+
+/**
+ * Document storage. Every node is a document (its own) and a collection of child documents.
+ * Implementations provide three primitives over document paths; the user-facing API is
+ * defined here once.
+ */
+export abstract class BaseDocumentStorage<S extends Spec = AnySpec> {
+  /** Unescaped path of this node (`[]` for the root, which has no own document). */
+  path: readonly string[] = []
+  protected readonly ids = new Ulid()
+
+  abstract readMany(paths: (readonly string[])[]): Promise<(Doc | undefined)[]>
+  /**
+   * Maps the current state of one document to a change (`null` = no change), applies it and
+   * returns the new state. Implementations decide how this is made safe against concurrent writers.
+   */
+  abstract mutate(path: readonly string[], change: (current: Doc | undefined) => DocChange | null): Promise<Doc | undefined>
+  /** Children of `parent` matching the query, filtered, sorted and paged. */
+  abstract query(parent: readonly string[], query: DocumentQuery): AsyncIterable<FindResult>
+
+  child<K extends KeyOf<S>>(key: K): BaseDocumentStorage<ChildOf<S, K>> {
+    const node = Object.create(this) as BaseDocumentStorage<ChildOf<S, K>>
+    node.path = [...this.path, Keys.segment(key)]
+    return node
+  }
+
+  as<S2 extends Spec>(): BaseDocumentStorage<S2> {
+    return this as unknown as BaseDocumentStorage<S2>
+  }
+
+  async get<K extends KeyOf<S> | undefined = undefined>(key?: K): Promise<ValueAt<S, K> | undefined> {
+    const [doc] = await this.readMany([this.pathOf(arguments.length > 0, key)])
+    return doc as ValueAt<S, K> | undefined
+  }
+
+  async getMany<K extends KeyOf<S>>(keys: readonly K[]): Promise<(ValueOf<ChildOf<S, K>> | undefined)[]> {
+    return (await this.readMany(keys.map((k) => this.pathOf(true, k)))) as (ValueOf<ChildOf<S, K>> | undefined)[]
+  }
+
+  /** `set(doc)` replaces this document; `set(key, doc)` a child document. */
+  async set<K extends KeyOf<S>>(...args: [doc: ValueOf<S>] | [key: K, doc: ValueOf<ChildOf<S, K>>]): Promise<void> {
+    const doc = args[args.length - 1]
+    if (typeof doc !== 'object' || doc === null || Array.isArray(doc)) throw new TypeError('A document must be a plain object')
+    await this.mutate(this.pathOf(args.length > 1, args[0]), () => ({ op: 'set', doc: doc as Doc }))
+  }
+
+  /** Adds a child document under a new time-sortable id and returns the id. */
+  async insert(doc: ChildDoc<S>): Promise<string> {
+    const id = this.ids.generate()
+    await this.set(id as KeyOf<S>, doc as never)
+    return id
+  }
+
+  /**
+   * `update(spec, options?)` this document; `update(key, spec, options?)` a child.
+   * Returns the new state, or `undefined` if the document doesn't exist (unless `upsert`).
+   */
+  async update<K extends KeyOf<S>>(
+    ...args: [spec: UpdateSpec, options?: UpdateOptions] | [key: K, spec: UpdateSpec, options?: UpdateOptions]
+  ): Promise<ValueAt<S, K> | undefined> {
+    const hasKey = typeof args[0] !== 'object' || args[0] === null
+    const spec = args[hasKey ? 1 : 0] as UpdateSpec
+    const options = (args[hasKey ? 2 : 1] ?? {}) as UpdateOptions
+    const next = await this.mutate(this.pathOf(hasKey, args[0]), (current) => (current || options.upsert ? DocumentUpdate.resolve(current ?? {}, spec) : null))
+    return next as ValueAt<S, K> | undefined
+  }
+
+  /** Deletes this document (or a child). Returns whether one existed. */
+  async delete(key?: KeyOf<S>): Promise<boolean> {
+    let existed = false
+    await this.mutate(this.pathOf(arguments.length > 0, key), (current) => ((existed = current !== undefined) ? { op: 'delete' } : null))
+    return existed
+  }
+
+  /** Child documents matching `filter`. */
+  find(filter: Filter = {}, options: FindOptions = {}): AsyncIterable<FindResult<ChildDoc<S>>> {
+    return this.query(this.path, new DocumentQuery(filter, options)) as AsyncIterable<FindResult<ChildDoc<S>>>
+  }
+
+  async findOne(filter: Filter = {}, options: FindOptions = {}): Promise<FindResult<ChildDoc<S>> | undefined> {
+    for await (const result of this.find(filter, { ...options, limit: 1 })) return result
+    return undefined
+  }
+
+  /** Keys of child documents. */
+  async *keys(): AsyncGenerator<string> {
+    for await (const { key } of this.find()) yield key
+  }
+
+  /** Path of a child (`hasKey`) or this node; validates keys. */
+  protected pathOf(hasKey: boolean, key: unknown): readonly string[] {
+    if (hasKey) return [...this.path, Keys.segment(key)]
+    if (this.path.length === 0) throw new TypeError('The root node has no own document')
+    return this.path
+  }
+}

src/document-storage/document-storage-in-memory.ts

@@ -0,0 +1,112 @@
+import type { Doc } from '../utils/document-path.js'
+import { DocumentPath } from '../utils/document-path.js'
+import type { DocumentQuery, FindResult } from '../utils/document-query.js'
+import { DocumentUpdate, type DocChange } from '../utils/document-update.js'
+import { Json } from '../utils/json.js'
+import { Keys } from '../utils/keys.js'
+import type { AnySpec, Spec } from '../utils/schema.js'
+import { BaseDocumentStorage, type IndexDeclarations } from './base-document-storage.js'
+
+export interface DocumentStorageInMemoryOptions {
+  /** Fields with an equality index, per collection pattern. */
+  indexes?: IndexDeclarations
+}
+
+/** Field → JSON-encoded value → child keys. */
+type CollectionIndex = Map<string, Map<string, Set<string>>>
+
+/**
+ * In-memory document storage. Declared index fields get an equality index per collection, used
+ * when a filter has `{ field: value }` or `{ field: { $eq: value } }`; everything else is a scan.
+ */
+export class DocumentStorageInMemory<S extends Spec = AnySpec> extends BaseDocumentStorage<S> {
+  private readonly collections = new Map<string, Map<string, Doc>>()
+  private readonly indexes = new Map<string, CollectionIndex>()
+  private readonly declarations: [pattern: string[], fields: string[]][]
+
+  constructor(options: DocumentStorageInMemoryOptions = {}) {
+    super()
+    this.declarations = Object.entries(options.indexes ?? {}).map(([pattern, fields]) => [pattern === '' ? [] : pattern.split('/'), fields])
+  }
+
+  async readMany(paths: (readonly string[])[]): Promise<(Doc | undefined)[]> {
+    return paths.map((path) => Json.clone(this.collection(path.slice(0, -1)).get(path.at(-1)!)))
+  }
+
+  async mutate(path: readonly string[], change: (current: Doc | undefined) => DocChange | null): Promise<Doc | undefined> {
+    const parent = path.slice(0, -1)
+    const key = path.at(-1)!
+    const docs = this.collection(parent)
+    const current = docs.get(key)
+    const resolved = change(Json.clone(current))
+    if (!resolved) return Json.clone(current)
+    const next = DocumentUpdate.apply(current, resolved)
+    if (next) docs.set(key, next)
+    else docs.delete(key)
+    this.reindex(parent, key, current, next)
+    return Json.clone(next)
+  }
+
+  async *query(parent: readonly string[], query: DocumentQuery): AsyncGenerator<FindResult> {
+    const docs = this.collection(parent)
+    const keys = [...(this.lookup(parent, query) ?? docs.keys())].sort()
+    yield* query.run(
+      (async function* () {
+        for (const key of keys) {
+          const doc = docs.get(key)
+          if (doc) yield { key, data: Json.clone(doc) }
+        }
+      })(),
+    )
+  }
+
+  private collection(parent: readonly string[]): Map<string, Doc> {
+    const id = Keys.join(parent)
+    let docs = this.collections.get(id)
+    if (!docs) this.collections.set(id, (docs = new Map()))
+    return docs
+  }
+
+  /** Candidate keys from an equality index, if the filter allows one. */
+  private lookup(parent: readonly string[], query: DocumentQuery): Iterable<string> | undefined {
+    const index = this.indexes.get(Keys.join(parent))
+    for (const field of this.fieldsFor(parent)) {
+      let value = query.filter[field]
+      if (Json.isObject(value) && Object.keys(value).length === 1 && '$eq' in value) value = value.$eq
+      if (value === undefined || typeof value === 'object') continue
+      return index?.get(field)?.get(JSON.stringify(value)) ?? []
+    }
+    return undefined
+  }
+
+  private reindex(parent: readonly string[], key: string, before: Doc | undefined, after: Doc | undefined): void {
+    const fields = this.fieldsFor(parent)
+    if (fields.length === 0) return
+    const id = Keys.join(parent)
+    let index = this.indexes.get(id)
+    if (!index) this.indexes.set(id, (index = new Map()))
+    for (const field of fields) {
+      let byValue = index.get(field)
+      if (!byValue) index.set(field, (byValue = new Map()))
+      for (const value of DocumentStorageInMemory.indexValues(before, field)) byValue.get(value)?.delete(key)
+      for (const value of DocumentStorageInMemory.indexValues(after, field)) {
+        let keys = byValue.get(value)
+        if (!keys) byValue.set(value, (keys = new Set()))
+        keys.add(key)
+      }
+    }
+  }
+
+  private fieldsFor(parent: readonly string[]): string[] {
+    return this.declarations
+      .filter(([pattern]) => pattern.length === parent.length && pattern.every((s, i) => s === '*' || s === parent[i]))
+      .flatMap(([, fields]) => fields)
+  }
+
+  /** Index keys for a field: its value, plus each element of an array (missing indexes as `null`). */
+  private static indexValues(doc: Doc | undefined, field: string): string[] {
+    if (!doc) return []
+    const value = DocumentPath.get(doc, field) ?? null
+    return Array.isArray(value) ? value.map((v) => JSON.stringify(v)) : [JSON.stringify(value)]
+  }
+}

src/document-storage/document-storage-kv.ts

@@ -0,0 +1,46 @@
+import type { BaseKVStorage } from '../kv-storage/base-kv-storage.js'
+import type { Doc } from '../utils/document-path.js'
+import type { DocumentQuery, FindResult } from '../utils/document-query.js'
+import { DocumentUpdate, type DocChange } from '../utils/document-update.js'
+import { Keys } from '../utils/keys.js'
+import type { AnySpec, Spec } from '../utils/schema.js'
+import { BaseDocumentStorage } from './base-document-storage.js'
+
+/**
+ * Document storage on any KV storage: each document is the value at its path. Queries scan the
+ * collection's direct children (no indexes). Writes are read-modify-write, so concurrent writers
+ * to the same document are last-writer-wins; use `EventSourcedDocumentStorage` if that matters.
+ */
+export class DocumentStorageKV<S extends Spec = AnySpec> extends BaseDocumentStorage<S> {
+  constructor(private readonly kv: BaseKVStorage) {
+    super()
+  }
+
+  async readMany(paths: (readonly string[])[]): Promise<(Doc | undefined)[]> {
+    return (await this.kv.readMany(paths.map((p) => this.keyOf(p)))) as (Doc | undefined)[]
+  }
+
+  async mutate(path: readonly string[], change: (current: Doc | undefined) => DocChange | null): Promise<Doc | undefined> {
+    const key = this.keyOf(path)
+    const [current] = (await this.kv.readMany([key])) as (Doc | undefined)[]
+    const resolved = change(current)
+    if (!resolved) return current
+    const next = DocumentUpdate.apply(current, resolved)
+    if (next) await this.kv.writeMany([[key, next]])
+    else await this.kv.removeMany([key])
+    return next
+  }
+
+  async *query(parent: readonly string[], query: DocumentQuery): AsyncGenerator<FindResult> {
+    const collection = parent.reduce<BaseKVStorage>((node, segment) => node.child(segment), this.kv)
+    yield* query.run(
+      (async function* () {
+        for await (const [key, data] of collection.entries()) yield { key, data: data as Doc }
+      })(),
+    )
+  }
+
+  private keyOf(path: readonly string[]): string {
+    return this.kv.prefix + Keys.join(path)
+  }
+}

src/document-storage/event-sourced-document-storage.ts

@@ -0,0 +1,125 @@
+import type { BaseEventStorage, StoredEvent } from '../event-storage/base-event-storage.js'
+import type { Doc } from '../utils/document-path.js'
+import type { DocumentQuery, FindResult } from '../utils/document-query.js'
+import { DocumentUpdate, type DocChange } from '../utils/document-update.js'
+import { ConcurrencyError } from '../utils/errors.js'
+import type { AnySpec, ChildOf, KeyOf, Spec, ValueAt } from '../utils/schema.js'
+import { BaseDocumentStorage } from './base-document-storage.js'
+
+export interface EventSourcedDocumentStorageOptions {
+  /** Retries after a write conflict, each preceded by a catch-up. Default 5. */
+  maxRetries?: number
+}
+
+// Types only: children of an event-sourced store are event-sourced stores too (child() creates
+// them with Object.create, so they share this class's prototype).
+export interface EventSourcedDocumentStorage<S extends Spec = AnySpec> {
+  child<K extends KeyOf<S>>(key: K): EventSourcedDocumentStorage<ChildOf<S, K>>
+}
+
+/**
+ * Documents as event streams: each document's changes are appended to its own stream, and the
+ * current state is kept in a projection store (any `BaseDocumentStorage`), which serves all
+ * reads and queries.
+ *
+ * Projections carry a `_version` field (stripped from results) used as `expectedVersion` on the
+ * next append. If a write crashes between appending and updating the projection, the next
+ * write's append conflicts, the projection is rebuilt from the stream, and the write retries.
+ * `_version` is therefore reserved in documents.
+ */
+export class EventSourcedDocumentStorage<S extends Spec = AnySpec> extends BaseDocumentStorage<S> {
+  private readonly maxRetries: number
+  /** Tail of the pending writes per document, so writes in one process don't conflict with each other. */
+  private readonly locks = new Map<string, Promise<unknown>>()
+
+  constructor(
+    private readonly events: BaseEventStorage,
+    private readonly projections: BaseDocumentStorage,
+    options: EventSourcedDocumentStorageOptions = {},
+  ) {
+    super()
+    this.maxRetries = options.maxRetries ?? 5
+  }
+
+  async readMany(paths: (readonly string[])[]): Promise<(Doc | undefined)[]> {
+    return (await this.projections.readMany(paths)).map((doc) => EventSourcedDocumentStorage.strip(doc))
+  }
+
+  async *query(parent: readonly string[], query: DocumentQuery): AsyncGenerator<FindResult> {
+    for await (const { key, data } of this.projections.query(parent, query)) yield { key, data: EventSourcedDocumentStorage.strip(data)! }
+  }
+
+  mutate(path: readonly string[], change: (current: Doc | undefined) => DocChange | null): Promise<Doc | undefined> {
+    const id = JSON.stringify(path)
+    const result = (this.locks.get(id) ?? Promise.resolve()).then(() => this.write(path, change))
+    const tail = result.catch(() => {})
+    this.locks.set(id, tail)
+    void tail.then(() => this.locks.get(id) === tail && this.locks.delete(id))
+    return result
+  }
+
+  private async write(path: readonly string[], change: (current: Doc | undefined) => DocChange | null): Promise<Doc | undefined> {
+    for (let attempt = 0; ; attempt++) {
+      const [stored] = await this.projections.readMany([path])
+      const version = (stored?._version as number | undefined) ?? (await this.events.versionOf(path))
+      if (!stored && version > 0) {
+        // No projection but the stream has history (deleted, or a crashed first write): rebuild.
+        if ((await this.rebuild(path)) !== undefined) continue
+      }
+      const current = EventSourcedDocumentStorage.strip(stored)
+      const resolved = change(current)
+      if (!resolved) return current
+      const next = DocumentUpdate.apply(current, resolved)
+      try {
+        await this.events.appendTo(path, [resolved], version)
+      } catch (error) {
+        if (!(error instanceof ConcurrencyError) || attempt >= this.maxRetries) throw error
+        await this.rebuild(path)
+        continue
+      }
+      await this.project(path, next, version + 1)
+      return next
+    }
+  }
+
+  /** Events of this document (or a child). */
+  history(key?: KeyOf<S>): AsyncIterable<StoredEvent<DocChange>> {
+    return this.events.readFrom(this.pathOf(arguments.length > 0, key), {}) as AsyncIterable<StoredEvent<DocChange>>
+  }
+
+  /** State as of `version`: `getAt(version)` for this document, `getAt(key, version)` for a child. */
+  async getAt<K extends KeyOf<S>>(...args: [version: number] | [key: K, version: number]): Promise<ValueAt<S, K> | undefined> {
+    const path = this.pathOf(args.length > 1, args[0])
+    let state: Doc | undefined
+    for await (const event of this.events.readFrom(path, { to: args[args.length - 1] as number })) {
+      state = DocumentUpdate.apply(state, event.data as DocChange)
+    }
+    return state as ValueAt<S, K> | undefined
+  }
+
+  /** Rebuilds the projection of this document (or a child) from its stream. Returns the state. */
+  catchUp(key?: KeyOf<S>): Promise<Doc | undefined> {
+    return this.rebuild(this.pathOf(arguments.length > 0, key))
+  }
+
+  private async rebuild(path: readonly string[]): Promise<Doc | undefined> {
+    let state: Doc | undefined
+    let version = 0
+    for await (const event of this.events.readFrom(path, {})) {
+      state = DocumentUpdate.apply(state, event.data as DocChange)
+      version = event.version
+    }
+    await this.project(path, state, version)
+    return state
+  }
+
+  private async project(path: readonly string[], state: Doc | undefined, version: number): Promise<void> {
+    await this.projections.mutate(path, () => (state ? { op: 'set', doc: { ...state, _version: version } } : { op: 'delete' }))
+  }
+
+  private static strip(doc: Doc | undefined): Doc | undefined {
+    if (!doc) return undefined
+    const { _version, ...state } = doc
+    return state
+  }
+}