diff --git a/DESIGN.md b/DESIGN.md index a921750..49f8083 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -484,6 +484,73 @@ Decisions taken where the brief was silent. Each is the simplest option consiste - **Mutex mode** (driver without `setIfAbsent`): a per-stream in-process promise chain serialises appends. It is safe only with a single writer process. +### M6 — Documents + +- **Projection value** is `{ state, version, commit }`, or `{ deleted: true, version, commit }` for a + tombstone. Tombstones keep the version, so re-creating a document continues its stream. +- **`expectedCommit` append hint.** Writes pass the projection's `commit` to `append` as + `expectedCommit`, so a write is exactly projection `getMany` + `setIfAbsent` + projection `batch` + (3 driver calls), even with a cold head cache. A hint taken from a projection can only be stale + (which causes a collision, then catch-up and retry), never ahead of the stream, so it cannot + create a gap. Non-KV backends may ignore it. +- **Per-document in-process write lock.** Writes to the same document within one storage instance + are serialised (`KeyedMutex`), so optimistic retries (default 5) are spent only on cross-process + contention. +- **Return values**: `set` → `void`; `insert` → id; `update` → new state, or `undefined` if the document + doesn't exist (pass `{ upsert: true }` to treat missing as `{}`); `delete` → whether a document + was deleted. No-op updates (resulting state equal to current) append nothing. +- **Arity for documents**: in forms with an optional leading key (`get`, `set`, `update`), a non-null + object first argument selects the self form, and any other first argument is a key (so `undefined` + throws). `delete`/`history` use argument count. `getAt(version)` vs `getAt(key, version)` also uses + argument count, because a version and a numeric key are both numbers. +- **The root document node has no own document** (consistent with KV). +- **`keys()` on a document node** lists children holding a live document. Unlike KV `keys()`, it skips + children that only have nested documents or tombstones. +- **`find` results** are `{ key, data }`. `findOne` returns the same shape or `undefined`. +- **Filter semantics vs MongoDB** (see `src/documents/filter.ts`): + - Comparisons are type-bracketed (`$gt: 5` matches only numbers), and cross-type order follows + the index encoding: `null/missing < false < true < numbers < strings < other`. + - On array fields, every positive operator in one condition (`$eq $gt $gte $lt $lte $in`) must be + satisfied by the **same element** (like `$elemMatch`). In MongoDB, `{ s: { $gt: 5, $lt: 10 } }` + matches `[1, 20]`; here it doesn't. This makes index range scans exact. To get MongoDB's + behaviour, split the condition with `$and`. + - `{ f: null }` matches missing fields. `$not` works both at top level (negating a filter) and at + field level. Unknown operators throw. + - `$in` is not index-accelerated in v1. +- **Index entries** for each field value: scalars as-is, arrays as their scalar elements (multikey; + empty arrays → `null`), and missing or non-scalar values as `null`. Every live document therefore + has at least one entry per index, so a full index scan sees all documents. Indexing a document + where more than one field of a compound index has multiple values throws (as MongoDB does). +- **Index names** default to the fields joined by `,`. +- **Unique indexes**: claims are scoped **per parent node** (`/~u//`), like one + MongoDB collection per parent. Value tuples where every field is missing/null are not claimed + (sparse). Claims are taken before appending, then released if the append fails. A claim value is + `[ownerKey, claimedAtMs]`. A claim held by another document is taken over only if that + document, after catching up from its stream, no longer holds the value, **and** the claim is older + than `claimGraceMs` (default 30 s). The owner might otherwise be between claiming and appending. + The residual risk is a writer stalling for longer than the grace period between claiming and + appending. +- **Planner**: per index, count leading equality fields (k), then whether field k has a range; + score = 2k + (range ? 1 : 0). Ties prefer an index whose order satisfies `sort` after dropping + equality-constrained sort fields. With no usable predicate, an index matching `sort` is still used + as an ordered full index scan, which streams and stops early with `limit`. Range bounds are clamped + to the bound's type class. Index-backed results are re-read from projections, re-checked against the + full filter, de-duplicated, and stale entries are read-repaired. +- **Result order** when `sort` is not given is the plan's order (index order, or child-key order for + a full scan). Ties in sorted results are broken by child key in the direction of the first sort + field, both for index-ordered and in-memory sorts. +- **In-memory sort** keeps a bounded top `skip + limit` buffer when a limit is given. Without a limit, + it buffers all matches; this is the one place results are materialised. +- **Full-scan warning**: `console.warn` once per filter *shape* (values replaced by `?`) per storage + instance, only for non-empty filters. On by default unless `NODE_ENV === 'production'`; set it with + `warnOnFullScan`. +- **Catch-up** re-asserts unique claims for the caught-up state with plain writes, since the log is + the source of truth. A slow catch-up can briefly overwrite a newer projection with an older one; + the next write's `expectedVersion` check detects this and heals it. `catchUp()` walks the subtree + via event-stream `keys()` and returns the number of documents repaired. +- **Projections for non-KV event stores** must be passed explicitly (`projections: KVNode`); the + projection key of a document is then `projections.prefix + docKey`. + ## 13. Benchmark baselines diff --git a/src/documents/event-sourced.ts b/src/documents/event-sourced.ts new file mode 100644 index 0000000..f9af064 --- /dev/null +++ b/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 + /** 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 + 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 + parentEv: EventNode + 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 implements EventSourcedDocumentNode { + readonly path: readonly string[] + /** Document-space key (escaped), relative to the store root. */ + readonly selfKey: string + readonly #ctx: DocContext + readonly #ev: EventNode + readonly #parentEv: EventNode | undefined + readonly #esc: string + readonly #prefix: string + readonly #state: TrieState + readonly #selfIdx: DocIndexes | undefined + readonly #childIdx: DocIndexes | undefined + + /** @internal Use `EventSourcedDocumentStorage` and `child()`. */ + constructor( + ctx: DocContext, + ev: EventNode, + parentEv: EventNode | undefined, + path: readonly string[], + selfKey: string, + esc: string, + state: TrieState, + 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>(key: K): EventSourcedDocumentNodeImpl> { + 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(): EventSourcedDocumentNodeImpl { + return this as unknown as EventSourcedDocumentNodeImpl + } + + // ---------------------------------------------------------------- reads + + get(opts?: GetOptions): Promise | undefined> + get>(key: K, opts?: GetOptions): Promise> | undefined> + async get(...args: unknown[]): Promise { + 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> + history>(key: K): AsyncIterable> + history(...args: unknown[]): AsyncIterable> { + const t = args.length === 0 ? this.#self() : this.#child(args[0]) + return t.ev.read() as AsyncIterable> + } + + getAt(version: number): Promise | undefined> + getAt>(key: K, version: number): Promise> | undefined> + async getAt(...args: unknown[]): Promise { + 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>> { + 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 } + 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 } + if (++emitted >= limit) return + } + } + + async findOne(filter: Filter = {}, opts: FindOptions = {}): Promise> | undefined> { + for await (const r of this.find(filter, { ...opts, limit: 1 })) return r + return undefined + } + + async *keys(opts: DocKeysOptions = {}): AsyncGenerator> { + 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 + if (--remaining <= 0) return + } + } + + // ---------------------------------------------------------------- writes + + set(doc: ValueOf): Promise + set>(key: K, doc: ValueOf>): Promise + async set(...args: unknown[]): Promise { + 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): Promise { + 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 | undefined> + update>(key: K, spec: UpdateSpec, opts?: UpdateOptions): Promise> | undefined> + async update(...args: unknown[]): Promise { + 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 + delete>(key: K): Promise + async delete(...args: unknown[]): Promise { + 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 { + 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).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 { + 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 { + return this.#ctx.writeLock.run(t.projKey, () => this.#writeUnlocked(t, compute)) + } + + async #writeUnlocked(t: Target, compute: (cur: Doc | undefined) => DocEvent | null): Promise { + 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, acquired: string[]): Promise { + 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 => { + 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 { + 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 { + 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 { + 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 { + const idx = this.#childIdx! + const { driver, readRepair } = this.#ctx + const parentProj = this.#ctx.projBase + this.#prefix + const seen = new Set() + 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 { + 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): (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 extends EventSourcedDocumentNodeImpl { + constructor(events: EventNode, 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) + } +} diff --git a/src/documents/filter.ts b/src/documents/filter.ts new file mode 100644 index 0000000..f2a4f9e --- /dev/null +++ b/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 { + 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()): Map { + 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 } diff --git a/src/documents/index.ts b/src/documents/index.ts new file mode 100644 index 0000000..8a2ad21 --- /dev/null +++ b/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' diff --git a/src/documents/indexes.ts b/src/documents/indexes.ts new file mode 100644 index 0000000..11d2e39 --- /dev/null +++ b/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//` 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 + +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 + readonly #enc: (v: IndexValue) => string + + constructor( + list: CompiledDocIndex[], + /** `~i` */ + readonly base: string, + /** `~u` */ + 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 = new Set()): Set { + 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 { + const out = new Map() + 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, 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 + 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() + 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 (``). */ + bind(state: TrieState, 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) + } +} diff --git a/src/documents/types.ts b/src/documents/types.ts new file mode 100644 index 0000000..2435656 --- /dev/null +++ b/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 +} + +export interface FindResult { + 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 = ValueOf>> + +/** + * 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 { + readonly path: readonly string[] + child>(key: K): DocumentNode> + as(): DocumentNode + + get(opts?: GetOptions): Promise | undefined> + get>(key: K, opts?: GetOptions): Promise> | undefined> + + set(doc: ValueOf): Promise + set>(key: K, doc: ValueOf>): Promise + + /** Inserts a child with a generated time-sortable id; returns the id. */ + insert(doc: ChildDoc): Promise + + /** Returns the new state, or `undefined` if the document doesn't exist (and no `upsert`). */ + update(spec: UpdateSpec, opts?: UpdateOptions): Promise | undefined> + update>(key: K, spec: UpdateSpec, opts?: UpdateOptions): Promise> | undefined> + + /** Returns whether a document was deleted. */ + delete(): Promise + delete>(key: K): Promise + + /** Queries direct children. */ + find(filter?: Filter, opts?: FindOptions): AsyncIterable>> + findOne(filter?: Filter, opts?: FindOptions): Promise> | undefined> + + /** Keys of direct children that hold a (non-deleted) document. */ + keys(opts?: DocKeysOptions): AsyncIterable> +} + +/** Document store backed by an event log: adds history and point-in-time reads. */ +export interface EventSourcedDocumentNode extends DocumentNode { + child>(key: K): EventSourcedDocumentNode> + as(): EventSourcedDocumentNode + + history(): AsyncIterable> + history>(key: K): AsyncIterable> + + /** State as of `version` (`undefined` if missing or deleted then). */ + getAt(version: number): Promise | undefined> + getAt>(key: K, version: number): Promise> | undefined> + + /** Rebuilds lagging projections for this node and all descendants; returns how many were repaired. */ + catchUp(): Promise +} + +export type DocumentStorage = DocumentNode diff --git a/src/documents/ulid.ts b/src/documents/ulid.ts new file mode 100644 index 0000000..ecf27ac --- /dev/null +++ b/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() diff --git a/src/documents/update.ts b/src/documents/update.ts new file mode 100644 index 0000000..6ce09b3 --- /dev/null +++ b/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 + $unset?: Record + $inc?: Record + $push?: Record + $pull?: Record + $addToSet?: Record +} + +/** Stored document events: resolved deltas, never operators, so replay is deterministic. */ +export type DocEvent = + | { op: 'set'; doc: Doc } + | { op: 'patch'; set?: Record; unset?: string[] } + | { op: 'delete' } + +export interface Patch { + set: Record + 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; 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; unset?: string[] } = { op: 'patch' } + if (Object.keys(p.set).length) e.set = p.set + if (p.unset.length) e.unset = p.unset + return e +} diff --git a/src/documents/values.ts b/src/documents/values.ts new file mode 100644 index 0000000..4406b8e --- /dev/null +++ b/src/documents/values.ts @@ -0,0 +1,157 @@ +import type { IndexValue } from '../core/encoding.js' + +/** Plain JSON object. */ +export type Doc = Record + +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 + 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(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 +} diff --git a/src/events/kv.ts b/src/events/kv.ts index 963b2e5..2ccf9fa 100644 --- a/src/events/kv.ts +++ b/src/events/kv.ts @@ -96,7 +96,8 @@ export class KVEventNode implements EventNode { 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>> @@ -145,7 +146,7 @@ export class KVEventNode implements EventNode { return EMPTY_HEAD } - async #append(s: Stream, events: readonly unknown[], expected: number | 'any'): Promise { + async #append(s: Stream, events: readonly unknown[], expected: number | 'any', hint?: Head): Promise { const { driver, heads, maxRetries } = this.#ctx if (events.length === 0) { const head = await this.#loadHead(s) @@ -155,7 +156,7 @@ export class KVEventNode implements EventNode { 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) diff --git a/src/events/types.ts b/src/events/types.ts index 34c76f6..db798d8 100644 --- a/src/events/types.ts +++ b/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 { diff --git a/src/index.ts b/src/index.ts index 3f1e020..caf70c3 100644 --- a/src/index.ts +++ b/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' diff --git a/test/documents/event-sourced.test.ts b/test/documents/event-sourced.test.ts new file mode 100644 index 0000000..5094d08 --- /dev/null +++ b/test/documents/event-sourced.test.ts @@ -0,0 +1,367 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' +import fc from 'fast-check' +import { KVStorage } from '../../src/kv/node.js' +import { InMemoryKVDriver } from '../../src/kv/memory.js' +import { KVEventStorage } from '../../src/events/kv.js' +import { EventSourcedDocumentStorage, type EventSourcedDocumentStorageOptions } from '../../src/documents/event-sourced.js' +import type { DocIndexDeclarations } from '../../src/documents/indexes.js' +import type { Filter } from '../../src/documents/filter.js' +import { CapabilityError, UniqueConstraintError } from '../../src/errors.js' +import { collect, CountingDriver, type Capability } from '../helpers.js' + +type User = { name?: string; email?: string; age?: number; tags?: string[]; logins?: number; address?: { city?: string } } + +const userIndexes: DocIndexDeclarations = { + users: [{ fields: ['email'], unique: true }, { fields: ['age'] }, { name: 'cityAge', fields: ['address.city', 'age'] }, { fields: ['tags'] }], + 'users/*/posts': [{ fields: ['createdAt'] }], +} + +function setup(opts: { hide?: Capability[]; inner?: InMemoryKVDriver } & EventSourcedDocumentStorageOptions = {}) { + const { hide = [], inner = new InMemoryKVDriver(), ...docOpts } = opts + const driver = new CountingDriver(inner, hide) + const kv = new KVStorage(driver) + const events = new KVEventStorage(kv) + const docs = new EventSourcedDocumentStorage(events, { indexes: userIndexes, warnOnFullScan: false, ...docOpts }) + return { driver, kv, events, docs, users: docs.child('users') } +} + +afterEach(() => vi.restoreAllMocks()) + +describe.each([ + ['atomic batch', [] as Capability[]], + ['non-atomic', ['batch'] as Capability[]], +])('EventSourcedDocumentStorage (%s)', (_, hide) => { + describe('CRUD', () => { + it('inserts with time-sortable ids and reads back', async () => { + const { users } = setup({ hide }) + const a = await users.insert({ name: 'Ada', age: 36 }) + const b = await users.insert({ name: 'Bob', age: 20 }) + expect(a).toMatch(/^[0-9A-HJKMNP-TV-Z]{26}$/) + expect(a < b).toBe(true) + expect(await users.get(a)).toEqual({ name: 'Ada', age: 36 }) + expect(await collect(users.keys())).toEqual([a, b]) + }) + + it('set / get / delete in self and child forms', async () => { + const { users } = setup({ hide }) + await users.set('123', { name: 'Ada', age: 36 }) + expect(await users.child('123').get()).toEqual({ name: 'Ada', age: 36 }) + await users.child('123').set({ name: 'Ada L.' }) + expect(await users.get('123')).toEqual({ name: 'Ada L.' }) + expect(await users.delete('123')).toBe(true) + expect(await users.delete('123')).toBe(false) + expect(await users.get('123')).toBeUndefined() + expect(await collect(users.keys())).toEqual([]) + await users.set('123', { name: 'Again' }) + expect(await users.get('123')).toEqual({ name: 'Again' }) + }) + + it('applies arity rules', async () => { + const { users, docs } = setup({ hide }) + const id: string | undefined = undefined + await expect(users.get(id as never)).rejects.toThrow(TypeError) + await expect(users.update(id as never, { $set: { a: 1 } })).rejects.toThrow(TypeError) + await expect(users.delete(id as never)).rejects.toThrow(TypeError) + await expect(docs.get()).rejects.toThrow(/root/) + await expect(users.set('x', [] as never)).rejects.toThrow(/plain object/) + }) + + it('nests documents under documents', async () => { + const { users } = setup({ hide }) + await users.set('u1', { name: 'Ada' }) + const posts = users.child('u1').child('posts') + await posts.set('p1', { title: 'Hello', createdAt: '2024-01' }) + expect(await posts.get('p1')).toEqual({ title: 'Hello', createdAt: '2024-01' }) + expect((await collect(users.find())).map((r) => r.key)).toEqual(['u1']) + expect((await collect(posts.find({ createdAt: { $gte: '2024' } }))).map((r) => r.key)).toEqual(['p1']) + }) + }) + + describe('update', () => { + it('supports $set $unset $inc $push $pull $addToSet with dot paths', async () => { + const { users } = setup({ hide }) + await users.set('1', { name: 'Ada', tags: ['a'], address: { city: 'X' }, nickname: 'A' } as User) + const r = await users.update('1', { + $set: { name: 'Ada L.', 'address.zip': '123' }, + $inc: { logins: 1 }, + $push: { tags: { $each: ['b', 'c'] } }, + $unset: { nickname: true }, + }) + expect(r).toEqual({ name: 'Ada L.', tags: ['a', 'b', 'c'], address: { city: 'X', zip: '123' }, logins: 1 }) + await users.update('1', { $pull: { tags: { $in: ['a', 'c'] } }, $addToSet: { 'address.list': { $each: [1, 1, 2] } } }) + await users.child('1').update({ $inc: { logins: 5 }, $addToSet: { tags: 'b' } }) + expect(await users.get('1')).toEqual({ name: 'Ada L.', tags: ['b'], address: { city: 'X', zip: '123', list: [1, 2] }, logins: 6 }) + }) + + it('returns undefined for missing docs unless upsert', async () => { + const { users } = setup({ hide }) + expect(await users.update('nope', { $set: { a: 1 } })).toBeUndefined() + expect(await users.get('nope')).toBeUndefined() + expect(await users.update('nope', { $inc: { n: 2 } }, { upsert: true })).toEqual({ n: 2 }) + }) + + it('rejects conflicting paths and bad operators', async () => { + const { users } = setup({ hide }) + await users.set('1', { age: 1 }) + await expect(users.update('1', { $set: { a: 1 }, $unset: { 'a.b': 1 } })).rejects.toThrow(/conflict/) + await expect(users.update('1', { $rename: {} } as never)).rejects.toThrow(/Unsupported/) + await expect(users.update('1', { $inc: { name: 1 } }).then(() => users.update('1', { $push: { age: 1 } }))).rejects.toThrow(/non-array/) + }) + + it('stores resolved deltas, not operators, and skips no-op updates', async () => { + const { users } = setup({ hide }) + await users.set('1', { logins: 5 }) + await users.update('1', { $inc: { logins: 1 } }) + await users.update('1', { $set: { logins: 6 } }) // no-op + expect((await collect(users.history('1'))).map((e) => e.data)).toEqual([ + { op: 'set', doc: { logins: 5 } }, + { op: 'patch', set: { logins: 6 } }, + ]) + }) + }) + + describe('history and getAt', () => { + it('replays to any version', async () => { + const { users } = setup({ hide }) + await users.set('1', { n: 1 }) + await users.update('1', { $inc: { n: 1 } }) + await users.delete('1') + await users.set('1', { n: 10 }) + expect(await users.getAt('1', 0)).toBeUndefined() + expect(await users.getAt('1', 1)).toEqual({ n: 1 }) + expect(await users.getAt('1', 2)).toEqual({ n: 2 }) + expect(await users.getAt('1', 3)).toBeUndefined() + expect(await users.child('1').getAt(4)).toEqual({ n: 10 }) + expect(await users.getAt('1', 99)).toEqual({ n: 10 }) + expect((await collect(users.child('1').history())).map((e) => e.version)).toEqual([1, 2, 3, 4]) + }) + }) + + describe('find', () => { + async function seed() { + const ctx = setup({ hide }) + const rows: [string, User][] = [ + ['a', { name: 'Ada', age: 36, email: 'ada@x', tags: ['x', 'y'], address: { city: 'London' } }], + ['b', { name: 'Bob', age: 20, email: 'bob@x', tags: ['y'], address: { city: 'Paris' } }], + ['c', { name: 'Cy', age: 51, tags: [], address: { city: 'London' } }], + ['d', { name: 'Di', age: 20, email: 'di@x' }], + ['e', { name: 'Ed', email: 'ed@x', address: { city: 'London' } }], + ] + for (const [k, v] of rows) await ctx.users.set(k, v) + await ctx.users.set('gone', { name: 'Gone', age: 20 }) + await ctx.users.delete('gone') + return ctx + } + const keysOf = async (it: AsyncIterable<{ key: string }>) => (await collect(it)).map((r) => r.key) + + it('evaluates comparison, logical and element operators', async () => { + const { users } = await seed() + // Unsorted results come in index or key order depending on the plan; compare as sets. + const q = async (f: Filter) => (await keysOf(users.find(f))).sort() + expect(await q({})).toEqual(['a', 'b', 'c', 'd', 'e']) + expect(await q({ age: 20 })).toEqual(['b', 'd']) + expect(await q({ age: { $gte: 20, $lt: 40 } })).toEqual(['a', 'b', 'd']) + expect(await q({ age: { $gt: 20 } })).toEqual(['a', 'c']) + expect(await q({ age: { $ne: 20 } })).toEqual(['a', 'c', 'e']) + expect(await q({ age: { $in: [36, 51] } })).toEqual(['a', 'c']) + expect(await q({ age: { $nin: [36, 51] } })).toEqual(['b', 'd', 'e']) + expect(await q({ age: { $exists: false } })).toEqual(['e']) + expect(await q({ age: null })).toEqual(['e']) + expect(await q({ email: { $exists: true }, age: { $lte: 20 } })).toEqual(['b', 'd']) + expect(await q({ 'address.city': 'London' })).toEqual(['a', 'c', 'e']) + expect(await q({ 'address.city': 'London', age: { $gt: 40 } })).toEqual(['c']) + expect(await q({ tags: 'y' })).toEqual(['a', 'b']) + expect(await q({ tags: ['y'] })).toEqual(['b']) + expect(await q({ $or: [{ age: 51 }, { name: 'Bob' }] })).toEqual(['b', 'c']) + expect(await q({ $and: [{ age: { $gte: 20 } }, { age: { $lte: 20 } }] })).toEqual(['b', 'd']) + expect(await q({ $not: { age: 20 } })).toEqual(['a', 'c', 'e']) + expect(await q({ age: { $not: { $gte: 30 } } })).toEqual(['b', 'd', 'e']) + expect(await q({ name: { $gt: 5 } })).toEqual([]) // type-bracketed + }) + + it('sorts (index-backed and in-memory), skips, limits and projects', async () => { + const { users } = await seed() + expect(await keysOf(users.find({}, { sort: { age: 1 } }))).toEqual(['e', 'b', 'd', 'a', 'c']) + expect(await keysOf(users.find({}, { sort: { age: -1 } }))).toEqual(['c', 'a', 'd', 'b', 'e']) + expect(await keysOf(users.find({}, { sort: { name: -1 }, skip: 1, limit: 2 }))).toEqual(['d', 'c']) + expect(await keysOf(users.find({ age: { $gte: 20 } }, { sort: { age: -1 }, limit: 2 }))).toEqual(['c', 'a']) + expect(await keysOf(users.find({ 'address.city': 'London' }, { sort: { age: 1 } }))).toEqual(['e', 'a', 'c']) + expect(await collect(users.find({ age: 36 }, { projection: { name: 1, 'address.city': 1 } }))).toEqual([{ key: 'a', data: { name: 'Ada', address: { city: 'London' } } }]) + expect(await collect(users.find({ age: 36 }, { projection: { tags: 0, address: 0, email: 0 } }))).toEqual([{ key: 'a', data: { name: 'Ada', age: 36 } }]) + expect(await users.findOne({ email: 'bob@x' })).toEqual({ key: 'b', data: expect.objectContaining({ name: 'Bob' }) }) + expect(await users.findOne({ email: 'nobody' })).toBeUndefined() + }) + + it('uses an index: findOne costs one entry scan and one projection read', async () => { + const { users, driver } = await seed() + driver.reset() + await users.findOne({ email: 'di@x' }) + expect(driver.calls).toEqual({ scan: 1, getMany: 1 }) + }) + + it('warns once per filter shape on full scans', async () => { + const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}) + const { users } = setup({ hide, warnOnFullScan: true }) + await collect(users.find({ name: 'x' })) + await collect(users.find({ name: 'y' })) + await collect(users.find({ age: 1 })) + expect(warn).toHaveBeenCalledTimes(1) + expect(warn.mock.calls[0]![0]).toMatch(/full scan for filter \{"name":"\?"\}/) + }) + }) + + describe('unique indexes', () => { + it('rejects duplicates and frees values on change and delete', async () => { + const { users } = setup({ hide }) + await users.set('1', { email: 'a@x' }) + await expect(users.set('2', { email: 'a@x' })).rejects.toBeInstanceOf(UniqueConstraintError) + await expect(users.insert({ email: 'a@x' })).rejects.toMatchObject({ index: 'email', existingKey: '1' }) + expect(await collect(users.keys())).toEqual(['1']) + await users.update('1', { $set: { email: 'b@x' } }) + await users.set('2', { email: 'a@x' }) + await users.delete('2') + await users.set('3', { email: 'a@x' }) + await users.set('4', {}) // missing values are not claimed + await users.set('5', {}) + await users.set('3', { email: 'a@x', name: 'same value, same doc' }) + }) + + it('exactly one of concurrent writers claiming the same value wins', async () => { + const { users } = setup({ hide }) + const results = await Promise.allSettled(['1', '2', '3'].map((k) => users.set(k, { email: 'same' }))) + expect(results.filter((r) => r.status === 'fulfilled')).toHaveLength(1) + expect((await collect(users.find({ email: 'same' }))).length).toBe(1) + }) + + it('takes over a stale claim only after the grace period', async () => { + let t = 0 + const { users, driver } = setup({ hide, now: () => t, claimGraceMs: 1000 }) + // Leave an orphaned claim: doc 1's projection and events vanish but its claim remains. + await users.set('1', { email: 'a@x' }) + await driver.inner.deleteMany(['~p/users/1']) + await driver.inner.deleteMany((await collect(driver.inner.scan({ gte: 'users/1/', lt: 'users/10' }))).map(([k]) => k)) + await expect(users.set('2', { email: 'a@x' })).rejects.toBeInstanceOf(UniqueConstraintError) + t = 5000 + await users.set('2', { email: 'a@x' }) + expect(await users.get('2')).toEqual({ email: 'a@x' }) + }) + + it('requires setIfAbsent', () => { + const kv = new KVStorage(new CountingDriver(new InMemoryKVDriver(), ['setIfAbsent'])) + expect(() => new EventSourcedDocumentStorage(new KVEventStorage(kv), { indexes: userIndexes })).toThrow(CapabilityError) + }) + }) + + describe('projection lag (crash between append and projection write)', () => { + async function lagged() { + const ctx = setup({ hide }) + await ctx.users.set('1', { n: 1, email: 'a@x' }) + // Simulate a crash: the event is appended but the projection is never written. + await ctx.events.child('users').append('1', [{ op: 'patch', set: { n: 2, email: 'b@x' } }]) + return ctx + } + + it('reads see the stale projection; consistent reads catch up', async () => { + const { users } = await lagged() + expect(await users.get('1')).toEqual({ n: 1, email: 'a@x' }) + expect(await users.get('1', { consistent: true })).toEqual({ n: 2, email: 'b@x' }) + expect(await users.findOne({ email: 'b@x' })).toMatchObject({ key: '1' }) + expect(await users.findOne({ email: 'a@x' })).toBeUndefined() + }) + + it('the next write self-heals', async () => { + const { users } = await lagged() + expect(await users.update('1', { $inc: { n: 1 } })).toEqual({ n: 3, email: 'b@x' }) + expect((await collect(users.history('1'))).map((e) => e.version)).toEqual([1, 2, 3]) + await users.set('2', { email: 'a@x' }) // old claim was released by the catch-up + }) + + it('catchUp() repairs a subtree', async () => { + const { users, docs, events } = await lagged() + await events.child('users').child('9').child('posts').append('p', [{ op: 'set', doc: { createdAt: 'x' } }]) + expect(await docs.catchUp()).toBe(2) + expect(await users.get('1')).toEqual({ n: 2, email: 'b@x' }) + expect(await users.child('9').child('posts').get('p')).toEqual({ createdAt: 'x' }) + expect(await docs.catchUp()).toBe(0) + }) + }) + + describe('concurrency', () => { + it('two storage instances incrementing the same doc converge', async () => { + const inner = new InMemoryKVDriver() + const a = setup({ hide, inner }).users + const b = setup({ hide, inner }).users + await a.set('1', { n: 0 }) + await Promise.all([...Array(5)].flatMap(() => [a.update('1', { $inc: { n: 1 } }), b.update('1', { $inc: { n: 1 } })])) + expect(await a.get('1')).toEqual({ n: 10 }) + expect(await b.get('1', { consistent: true })).toEqual({ n: 10 }) + }) + }) +}) + +describe('costs', () => { + it('get is one driver call; update is projection read + append + projection write, even cold', async () => { + const inner = new InMemoryKVDriver() + await setup({ inner }).users.set('1', { name: 'a', age: 1 }) + const { users, driver } = setup({ inner }) // fresh instance: cold head cache + driver.reset() + await users.get('1') + expect(driver.calls).toEqual({ getMany: 1 }) + driver.reset() + await users.update('1', { $set: { age: 2 } }) + expect(driver.calls).toEqual({ getMany: 1, setIfAbsent: 1, batch: 1 }) + }) +}) + +describe('property: indexed find equals unindexed find', () => { + const doc = fc.record( + { + age: fc.option(fc.integer({ min: 0, max: 5 }), { nil: undefined }), + email: fc.constantFrom('a', 'b', 'c', 'd', 'e', 'f'), + tags: fc.array(fc.constantFrom('x', 'y', 'z'), { maxLength: 3 }), + address: fc.option(fc.record({ city: fc.constantFrom('L', 'P') }), { nil: undefined }), + }, + { requiredKeys: [] }, + ) + const bound = fc.integer({ min: -1, max: 6 }) + const filter: fc.Arbitrary = fc.oneof( + fc.record({ age: bound }), + fc.record({ age: fc.record({ $gte: bound, $lt: bound }, { requiredKeys: [] }) }), + fc.record({ age: fc.record({ $gt: bound, $lte: bound }, { requiredKeys: [] }) }), + fc.record({ tags: fc.constantFrom('x', 'y', 'z') }), + fc.record({ tags: fc.record({ $gte: fc.constantFrom('x', 'y'), $lt: fc.constantFrom('y', 'zz') }) }), + fc.record({ 'address.city': fc.constantFrom('L', 'P'), age: fc.record({ $gt: bound }) }), + fc.record({ 'address.city': fc.constantFrom('L', 'P') }), + fc.constant({ age: null }), + fc.constant({}), + ) + const sort = fc.option(fc.record({ age: fc.constantFrom(1 as const, -1 as const) }), { nil: undefined }) + + it('returns the same results in the same order', async () => { + await fc.assert( + fc.asyncProperty(fc.array(fc.tuple(fc.constantFrom('1', '2', '3', '4', '5'), doc), { maxLength: 12 }), filter, sort, fc.option(fc.integer({ min: 1, max: 4 }), { nil: undefined }), async (writes, f, s, limit) => { + const plain = new EventSourcedDocumentStorage(new KVEventStorage(new KVStorage(new InMemoryKVDriver())), { warnOnFullScan: false }).child('users') + const indexed = new EventSourcedDocumentStorage(new KVEventStorage(new KVStorage(new InMemoryKVDriver())), { + warnOnFullScan: false, + indexes: { users: userIndexes.users!.map((i) => ({ ...i, unique: false })) }, + }).child('users') + for (const [k, v] of writes) { + await plain.set(k, v) + await indexed.set(k, v) + } + const opts = { ...(s ? { sort: s } : {}), ...(limit ? { limit } : {}) } + const a = await collect(plain.find(f, opts)) + const b = await collect(indexed.find(f, opts)) + if (s) { + expect(b).toEqual(a) + return + } + // Unsorted order depends on the plan: compare as sets (a limited result is any subset of that size). + const all = new Set((await collect(plain.find(f))).map((r) => r.key)) + expect(b).toHaveLength(a.length) + for (const r of b) expect(all.has(r.key)).toBe(true) + }), + { numRuns: 300 }, + ) + }) +}) diff --git a/test/documents/units.test.ts b/test/documents/units.test.ts new file mode 100644 index 0000000..e206270 --- /dev/null +++ b/test/documents/units.test.ts @@ -0,0 +1,56 @@ +import { describe, expect, it } from 'vitest' +import { createIdGenerator } from '../../src/documents/ulid.js' +import { matches } from '../../src/documents/filter.js' +import { applyPatch, resolveUpdate } from '../../src/documents/update.js' + +describe('ulid', () => { + it('is strictly increasing within one millisecond and sorts by time', () => { + let t = 1_700_000_000_000 + const gen = createIdGenerator(() => t) + const ids = Array.from({ length: 1000 }, () => gen()) + expect([...ids].sort()).toEqual(ids) + expect(new Set(ids).size).toBe(1000) + t += 1 + const later = gen() + expect(later > ids.at(-1)!).toBe(true) + expect(later.slice(0, 10) > ids[0]!.slice(0, 10)).toBe(true) + }) +}) + +describe('filter semantics', () => { + it('requires one array element to satisfy all positive operators of a condition', () => { + expect(matches({ s: [1, 20] }, { s: { $gt: 5, $lt: 10 } })).toBe(false) // MongoDB would match + expect(matches({ s: [1, 7] }, { s: { $gt: 5, $lt: 10 } })).toBe(true) + expect(matches({ s: [1, 20] }, { $and: [{ s: { $gt: 5 } }, { s: { $lt: 10 } }] })).toBe(true) + }) + + it('traverses arrays of sub-documents on dot paths', () => { + expect(matches({ a: [{ b: 1 }, { b: 2 }] }, { 'a.b': 2 })).toBe(true) + expect(matches({ a: [{ b: 1 }] }, { 'a.0.b': 1 })).toBe(true) + expect(matches({ a: [{ b: 1 }] }, { 'a.b': { $exists: true } })).toBe(true) + expect(matches({ a: [] }, { 'a.b': { $exists: true } })).toBe(false) + }) + + it('treats null as matching missing', () => { + expect(matches({}, { x: null })).toBe(true) + expect(matches({ x: null }, { x: { $ne: null } })).toBe(false) + expect(matches({ x: 0 }, { x: { $ne: null } })).toBe(true) + }) + + it('rejects unknown operators', () => { + expect(() => matches({}, { x: { $regex: 'a' } })).toThrow(/Unknown operator/) + expect(() => matches({}, { $nor: [] } as never)).toThrow(/Unknown top-level/) + }) +}) + +describe('update resolution', () => { + it('$pull by sub-document filter and by condition', () => { + const state = { items: [{ k: 1, v: 'a' }, { k: 2, v: 'b' }], n: [1, 5, 9] } + const patch = resolveUpdate(state, { $pull: { items: { k: 2 }, n: { $gte: 5 } } }) + expect(applyPatch(state, patch)).toEqual({ items: [{ k: 1, v: 'a' }], n: [1] }) + }) + + it('$unset of an array position nulls it', () => { + expect(applyPatch({ a: [1, 2] }, resolveUpdate({ a: [1, 2] }, { $unset: { 'a.0': true } }))).toEqual({ a: [null, 2] }) + }) +})