M3: KVNode/KVStorage with arity dispatch, skip-scan keys, deleteTree, tree
Children are listed in subtree order (escaped segment + separator) so the skip-scan stays O(children) even when sibling keys extend each other with characters below the separator. Model-based property test vs a reference Map. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_017Qxbkp4tA7G5SCWjzWuxmE
This commit is contained in:
Claude
@@ -0,0 +1,236 @@
|
||||
import { escapeSegment, succ } from '../core/path.js'
|
||||
import type { KVDriver } from './driver.js'
|
||||
|
||||
/**
|
||||
* Direct-children skip-scan, shared by every layer.
|
||||
*
|
||||
* ## Child order
|
||||
*
|
||||
* Children are ordered by **escaped segment followed by the separator** ("subtree order": the
|
||||
* order in which their descendants appear in key space). For segments without characters that
|
||||
* sort below the separator this is plain lexicographic order. When one segment extends another
|
||||
* with such a character (`a` vs `a-1` with `/`), the longer one comes first (`a-1/` < `a/`).
|
||||
*
|
||||
* This is the one order that can be produced without extra reads: a child's own key
|
||||
* (`p/a`) sorts *before* siblings that extend it (`p/a-1`), while its descendants (`p/a/…`) sort
|
||||
* after them. The forward scan therefore defers yielding a child seen only through its own key
|
||||
* until the scan passes `p/a/`, using a stack bounded by segment length.
|
||||
*
|
||||
* ## Cost
|
||||
*
|
||||
* - Forward: one seek per child that has descendants (to jump over its subtree), one at the start,
|
||||
* and one per reserved (`~…`) region. Siblings without descendants share an iterator.
|
||||
* `after` adds one `getMany` when the `after` segment contains a below-separator character.
|
||||
* - Reverse: as forward, plus one `getMany` per newly found child whose segment contains a
|
||||
* below-separator character (to find shorter siblings that only have an own key), or that was
|
||||
* found through a descendant when `own` is requested.
|
||||
*/
|
||||
export interface ChildHit {
|
||||
/** Escaped segment. */
|
||||
esc: string
|
||||
/** Whether the child has an own value. Only accurate when `own` was requested. */
|
||||
hasOwn: boolean
|
||||
/** The own value, when `values` was requested and `hasOwn`. */
|
||||
value: unknown
|
||||
}
|
||||
|
||||
export interface ChildScanOptions {
|
||||
/** Unescaped segment to start after (exclusive), in the direction of travel. */
|
||||
after?: string
|
||||
reverse?: boolean
|
||||
limit?: number
|
||||
/** Fetch own values. Implies `own`. */
|
||||
values?: boolean
|
||||
/** Require accurate `hasOwn`. */
|
||||
own?: boolean
|
||||
}
|
||||
|
||||
const RESERVED_START = '~'
|
||||
const RESERVED_END = succ('~')
|
||||
|
||||
export async function* scanChildren(
|
||||
driver: KVDriver,
|
||||
prefix: string,
|
||||
end: string | undefined,
|
||||
sep: string,
|
||||
opts: ChildScanOptions = {},
|
||||
): AsyncGenerator<ChildHit> {
|
||||
const limit = opts.limit ?? Infinity
|
||||
if (limit <= 0) return
|
||||
if (opts.reverse) yield* reverseScan(driver, prefix, end, sep, opts, limit)
|
||||
else yield* forwardScan(driver, prefix, end, sep, opts, limit)
|
||||
}
|
||||
|
||||
/** Prefixes `E[0..j)` (j ≥ 1) where `E[j]` sorts below the separator. */
|
||||
function lowPrefixes(esc: string, sepCode: number): string[] {
|
||||
let out: string[] | undefined
|
||||
for (let j = 1; j < esc.length; j++) {
|
||||
if (esc.charCodeAt(j) < sepCode) (out ??= []).push(esc.slice(0, j))
|
||||
}
|
||||
return out ?? NONE
|
||||
}
|
||||
|
||||
const NONE: string[] = []
|
||||
|
||||
async function* forwardScan(
|
||||
driver: KVDriver,
|
||||
prefix: string,
|
||||
end: string | undefined,
|
||||
sep: string,
|
||||
opts: ChildScanOptions,
|
||||
limit: number,
|
||||
): AsyncGenerator<ChildHit> {
|
||||
const sepCode = sep.charCodeAt(0)
|
||||
const succSep = succ(sep)
|
||||
const values = opts.values === true
|
||||
const pending: ChildHit[] = []
|
||||
let emitted = 0
|
||||
let gte = prefix
|
||||
|
||||
if (opts.after !== undefined) {
|
||||
const escA = escapeSegment(opts.after, sep)
|
||||
gte = prefix + escA + succSep
|
||||
// Shorter siblings that extend into `after` with a low char come after it, but their own
|
||||
// keys sort before `gte`; probe for them.
|
||||
const cands = lowPrefixes(escA, sepCode)
|
||||
if (cands.length) {
|
||||
const found = await driver.getMany(cands.map((g) => prefix + g))
|
||||
for (let i = 0; i < cands.length; i++) {
|
||||
if (found[i] !== undefined) pending.push({ esc: cands[i]!, hasOwn: true, value: values ? found[i] : undefined })
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const prefixLen = prefix.length
|
||||
let it = driver.scan({ gte, lt: end, values })[Symbol.asyncIterator]()
|
||||
const reseek = async (from: string) => {
|
||||
await it.return?.()
|
||||
it = driver.scan({ gte: from, lt: end, values })[Symbol.asyncIterator]()
|
||||
}
|
||||
|
||||
try {
|
||||
outer: for (;;) {
|
||||
const r = await it.next()
|
||||
if (r.done) {
|
||||
while (pending.length) {
|
||||
yield pending.pop()!
|
||||
if (++emitted >= limit) return
|
||||
}
|
||||
return
|
||||
}
|
||||
const key = r.value[0]
|
||||
const cut = key.indexOf(sep, prefixLen)
|
||||
const esc = cut === -1 ? key.slice(prefixLen) : key.slice(prefixLen, cut)
|
||||
|
||||
while (pending.length) {
|
||||
const top = pending[pending.length - 1]!
|
||||
if (top.esc === esc) {
|
||||
// First descendant of a deferred child: emit it and jump over its subtree.
|
||||
pending.pop()
|
||||
yield top
|
||||
if (++emitted >= limit) return
|
||||
await reseek(prefix + esc + succSep)
|
||||
continue outer
|
||||
}
|
||||
if (top.esc + sep < esc + sep) {
|
||||
pending.pop()
|
||||
yield top
|
||||
if (++emitted >= limit) return
|
||||
} else break
|
||||
}
|
||||
|
||||
if (esc.startsWith(RESERVED_START)) {
|
||||
await reseek(prefix + RESERVED_END)
|
||||
continue
|
||||
}
|
||||
if (cut === -1) {
|
||||
// Own key: defer until we know no sibling extending it with a low char comes next.
|
||||
pending.push({ esc, hasOwn: true, value: r.value[1] })
|
||||
continue
|
||||
}
|
||||
// Descendant of a child with no own value (an own key would have sorted first).
|
||||
yield { esc, hasOwn: false, value: undefined }
|
||||
if (++emitted >= limit) return
|
||||
await reseek(prefix + esc + succSep)
|
||||
}
|
||||
} finally {
|
||||
await it.return?.()
|
||||
}
|
||||
}
|
||||
|
||||
async function* reverseScan(
|
||||
driver: KVDriver,
|
||||
prefix: string,
|
||||
end: string | undefined,
|
||||
sep: string,
|
||||
opts: ChildScanOptions,
|
||||
limit: number,
|
||||
): AsyncGenerator<ChildHit> {
|
||||
const sepCode = sep.charCodeAt(0)
|
||||
const values = opts.values === true
|
||||
const needOwn = values || opts.own === true
|
||||
// Segments already emitted whose own key may still appear (it sorts after everything else of theirs).
|
||||
const done: string[] = []
|
||||
let emitted = 0
|
||||
let lt = end
|
||||
|
||||
if (opts.after !== undefined) {
|
||||
const escA = escapeSegment(opts.after, sep)
|
||||
lt = prefix + escA + sep
|
||||
// `after` itself and shorter siblings extending into it with a low char sort after it in
|
||||
// child order but their own keys fall below `lt`: mark them done.
|
||||
done.push(...lowPrefixes(escA, sepCode), escA)
|
||||
}
|
||||
|
||||
const prefixLen = prefix.length
|
||||
let it = driver.scan({ gte: prefix, lt, reverse: true, values })[Symbol.asyncIterator]()
|
||||
const reseek = async (below: string) => {
|
||||
await it.return?.()
|
||||
it = driver.scan({ gte: prefix, lt: below, reverse: true, values })[Symbol.asyncIterator]()
|
||||
}
|
||||
|
||||
try {
|
||||
for (;;) {
|
||||
const r = await it.next()
|
||||
if (r.done) return
|
||||
const key = r.value[0]
|
||||
const cut = key.indexOf(sep, prefixLen)
|
||||
const esc = cut === -1 ? key.slice(prefixLen) : key.slice(prefixLen, cut)
|
||||
|
||||
if (esc.startsWith(RESERVED_START)) {
|
||||
await reseek(prefix + RESERVED_START)
|
||||
continue
|
||||
}
|
||||
while (done.length && !esc.startsWith(done[done.length - 1]!)) done.pop()
|
||||
if (done.length && done[done.length - 1] === esc) {
|
||||
done.pop()
|
||||
if (cut !== -1) await reseek(prefix + esc + sep)
|
||||
continue
|
||||
}
|
||||
|
||||
// New child. Shorter siblings that it extends with a low char come before it in reverse
|
||||
// order; any with descendants were already emitted, so only own-key-only ones are unknown.
|
||||
const cands = lowPrefixes(esc, sepCode).filter((g) => !done.includes(g))
|
||||
const probeOwn = cut !== -1 && needOwn
|
||||
let found: unknown[] = NONE
|
||||
if (cands.length || probeOwn) {
|
||||
const probe = cands.map((g) => prefix + g)
|
||||
if (probeOwn) probe.push(prefix + esc)
|
||||
found = await driver.getMany(probe)
|
||||
}
|
||||
for (let i = 0; i < cands.length; i++) {
|
||||
if (found[i] === undefined) continue
|
||||
yield { esc: cands[i]!, hasOwn: true, value: values ? found[i] : undefined }
|
||||
done.push(cands[i]!)
|
||||
if (++emitted >= limit) return
|
||||
}
|
||||
const own = cut === -1 ? r.value[1] : probeOwn ? found[cands.length] : undefined
|
||||
yield { esc, hasOwn: cut === -1 || own !== undefined, value: values ? own : undefined }
|
||||
done.push(esc)
|
||||
if (++emitted >= limit) return
|
||||
if (cut !== -1) await reseek(prefix + esc + sep)
|
||||
}
|
||||
} finally {
|
||||
await it.return?.()
|
||||
}
|
||||
}
|
||||
@@ -1,3 +1,6 @@
|
||||
export * from './driver.js'
|
||||
export * from './memory.js'
|
||||
export * from './sorted.js'
|
||||
export * from './node.js'
|
||||
export * from './indexes.js'
|
||||
export * from './children.js'
|
||||
|
||||
@@ -0,0 +1,157 @@
|
||||
import { indexValueCodec, isIndexValue, type IndexValue, type IndexValueCodec } from '../core/encoding.js'
|
||||
import { escapeSegment, RESERVED, succ } from '../core/path.js'
|
||||
import { PatternTrie, patternsOverlap, type TrieState } from '../core/trie.js'
|
||||
|
||||
export type { IndexValue }
|
||||
|
||||
/** Maps a child's value to index value(s). An array means multi-entry; `undefined` means not indexed. */
|
||||
export type IndexFn<V = any> = (value: V) => IndexValue | readonly IndexValue[] | undefined
|
||||
|
||||
/** Index name → function, for the direct children of nodes matching one pattern. */
|
||||
export type IndexDefs = Record<string, IndexFn>
|
||||
|
||||
/** Path pattern (`*` = one segment, `''` = root) → index definitions. */
|
||||
export type IndexDeclarations = Record<string, IndexDefs>
|
||||
|
||||
/** A value or a range over index values. Bounds combine (`{ gte, lt }`). */
|
||||
export type IndexQuery = IndexValue | IndexRange
|
||||
|
||||
export interface IndexRange {
|
||||
gt?: IndexValue
|
||||
gte?: IndexValue
|
||||
lt?: IndexValue
|
||||
lte?: IndexValue
|
||||
}
|
||||
|
||||
export function isIndexRange(q: unknown): q is IndexRange {
|
||||
return typeof q === 'object' && q !== null
|
||||
}
|
||||
|
||||
interface CompiledIndex {
|
||||
name: string
|
||||
/** `escapeSegment(name) + sep` */
|
||||
head: string
|
||||
fn: IndexFn
|
||||
}
|
||||
|
||||
/** Indexes over the direct children of one concrete node. Entry keys are rooted at `base`. */
|
||||
export class BoundIndexes {
|
||||
readonly #byName: Map<string, CompiledIndex>
|
||||
readonly #list: CompiledIndex[]
|
||||
readonly #codec: IndexValueCodec
|
||||
|
||||
constructor(
|
||||
defs: CompiledIndex[],
|
||||
/** `<nodePrefix>~i<sep>` */
|
||||
readonly base: string,
|
||||
readonly sep: string,
|
||||
) {
|
||||
this.#list = defs
|
||||
this.#byName = new Map(defs.map((d) => [d.name, d]))
|
||||
this.#codec = indexValueCodec(sep)
|
||||
}
|
||||
|
||||
has(name: string): boolean {
|
||||
return this.#byName.has(name)
|
||||
}
|
||||
|
||||
/** All index entry keys for a child (escaped segment) holding `value`. */
|
||||
entriesFor(escChild: string, value: unknown, out: Set<string> = new Set()): Set<string> {
|
||||
if (value === undefined) return out
|
||||
for (const idx of this.#list) this.#entriesOf(idx, escChild, value, out)
|
||||
return out
|
||||
}
|
||||
|
||||
#entriesOf(idx: CompiledIndex, escChild: string, value: unknown, out: Set<string>): void {
|
||||
const r = idx.fn(value)
|
||||
if (r === undefined) return
|
||||
const stem = this.base + idx.head
|
||||
const tail = this.sep + escChild
|
||||
if (Array.isArray(r)) {
|
||||
for (const v of r as IndexValue[]) out.add(stem + this.#encode(idx.name, v) + tail)
|
||||
} else out.add(stem + this.#encode(idx.name, r as IndexValue) + tail)
|
||||
}
|
||||
|
||||
#encode(name: string, v: unknown): string {
|
||||
if (!isIndexValue(v)) throw new TypeError(`Index "${name}" returned a non-indexable value: ${String(v)}`)
|
||||
return this.#codec.encode(v)
|
||||
}
|
||||
|
||||
/** Whether `entryKey` is among the entries `value` currently produces for index `name`. */
|
||||
matches(name: string, escChild: string, value: unknown, entryKey: string): boolean {
|
||||
if (value === undefined) return false
|
||||
const out = new Set<string>()
|
||||
this.#entriesOf(this.#get(name), escChild, value, out)
|
||||
return out.has(entryKey)
|
||||
}
|
||||
|
||||
/** Scan bounds over the entries of index `name` for a query. */
|
||||
range(name: string, q: IndexQuery): { gte: string; lt: string } {
|
||||
const stem = this.base + this.#get(name).head
|
||||
const succSep = succ(this.sep)
|
||||
const enc = (v: IndexValue) => stem + this.#encode(name, v)
|
||||
if (!isIndexRange(q)) {
|
||||
const k = enc(q)
|
||||
return { gte: k + this.sep, lt: k + succSep }
|
||||
}
|
||||
let gte = stem
|
||||
let lt = stem.slice(0, -1) + succSep
|
||||
if (q.gte !== undefined) gte = maxStr(gte, enc(q.gte))
|
||||
if (q.gt !== undefined) gte = maxStr(gte, enc(q.gt) + succSep)
|
||||
if (q.lt !== undefined) lt = minStr(lt, enc(q.lt))
|
||||
if (q.lte !== undefined) lt = minStr(lt, enc(q.lte) + succSep)
|
||||
return { gte, lt }
|
||||
}
|
||||
|
||||
/** Splits an entry key under index `name` into the escaped child segment. */
|
||||
childOf(name: string, entryKey: string): string {
|
||||
const start = this.base.length + this.#get(name).head.length
|
||||
return entryKey.slice(entryKey.indexOf(this.sep, start) + 1)
|
||||
}
|
||||
|
||||
#get(name: string): CompiledIndex {
|
||||
const idx = this.#byName.get(name)
|
||||
if (!idx) throw new TypeError(`Unknown index "${name}"`)
|
||||
return idx
|
||||
}
|
||||
}
|
||||
|
||||
const maxStr = (a: string, b: string) => (a > b ? a : b)
|
||||
const minStr = (a: string, b: string) => (a < b ? a : b)
|
||||
|
||||
/** Compiles index declarations into a pattern trie and binds them to nodes. */
|
||||
export class IndexRegistry {
|
||||
readonly trie: PatternTrie<IndexDefs>
|
||||
readonly empty: boolean
|
||||
|
||||
constructor(
|
||||
decls: IndexDeclarations,
|
||||
readonly sep: string,
|
||||
) {
|
||||
this.trie = new PatternTrie(sep)
|
||||
const patterns = Object.keys(decls)
|
||||
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 clash = Object.keys(decls[a]!).filter((n) => n in decls[b]!)
|
||||
if (clash.length) throw new TypeError(`Index name(s) ${clash.join(', ')} declared on overlapping patterns "${a}" and "${b}"`)
|
||||
}
|
||||
}
|
||||
for (const p of patterns) this.trie.add(p, decls[p]!)
|
||||
this.empty = patterns.length === 0
|
||||
}
|
||||
|
||||
/** Indexes over the children of the node with trie state `state` and key prefix `prefix`. */
|
||||
bind(state: TrieState<IndexDefs>, prefix: string): BoundIndexes | undefined {
|
||||
if (this.empty) return undefined
|
||||
const found = PatternTrie.values(state)
|
||||
if (found.length === 0) return undefined
|
||||
const defs: CompiledIndex[] = []
|
||||
for (const { value } of found) {
|
||||
for (const [name, fn] of Object.entries(value)) defs.push({ name, head: escapeSegment(name, this.sep) + this.sep, fn })
|
||||
}
|
||||
return new BoundIndexes(defs, prefix + RESERVED.index + this.sep, this.sep)
|
||||
}
|
||||
}
|
||||
+403
@@ -0,0 +1,403 @@
|
||||
import { DEFAULT_SEPARATOR, escapeSegment, succ, toSegment, unescapeSegment, validateSeparator } from '../core/path.js'
|
||||
import type { TrieState } from '../core/trie.js'
|
||||
import type { AnySpec, ChildOf, EntryOf, KeyOf, KeyString, Spec, StringEntryOf, ValueOf } from '../types/schema.js'
|
||||
import { scanChildren } from './children.js'
|
||||
import type { BatchOp, KVDriver } from './driver.js'
|
||||
import { BoundIndexes, IndexRegistry, isIndexRange, type IndexDeclarations, type IndexDefs, type IndexQuery } from './indexes.js'
|
||||
|
||||
export interface KVStorageOptions {
|
||||
/** Single printable ASCII character other than `%` and `~`. Default `/`. */
|
||||
separator?: string
|
||||
/** Path pattern → index definitions over that node's direct children. */
|
||||
indexes?: IndexDeclarations
|
||||
/**
|
||||
* Delete stale index entries found by `findBy` (read-repair). Default `true`.
|
||||
* See DESIGN.md §12 (M4) for the race with concurrent non-atomic writers.
|
||||
*/
|
||||
readRepair?: boolean
|
||||
}
|
||||
|
||||
export interface KeysOptions {
|
||||
limit?: number
|
||||
/** Start after this child key (exclusive), in the direction of travel. */
|
||||
after?: string | number
|
||||
reverse?: boolean
|
||||
}
|
||||
|
||||
export interface TreeOptions {
|
||||
/** Levels below this node to include; `1` = direct children only. */
|
||||
depth: number
|
||||
}
|
||||
|
||||
export interface FindByOptions {
|
||||
limit?: number
|
||||
reverse?: boolean
|
||||
}
|
||||
|
||||
/** @internal */
|
||||
export interface KVContext {
|
||||
readonly driver: KVDriver
|
||||
readonly sep: string
|
||||
readonly registry: IndexRegistry
|
||||
readonly readRepair: boolean
|
||||
}
|
||||
|
||||
interface Slot {
|
||||
key: string
|
||||
esc: string
|
||||
value: unknown
|
||||
}
|
||||
|
||||
const FIND_CHUNK = 128
|
||||
const DELETE_CHUNK = 1000
|
||||
|
||||
/**
|
||||
* A store over its direct children. `child(key)` re-roots to that key (sync, no I/O).
|
||||
* Methods take an optional leading key: without it they act on this node's own value,
|
||||
* with it on that child's. The choice is made by argument count; a passed-but-`undefined`
|
||||
* key throws.
|
||||
*/
|
||||
export class KVNode<S extends Spec = AnySpec> {
|
||||
/** Unescaped path segments. */
|
||||
readonly path: readonly string[]
|
||||
/** This node's own key (escaped). `''` for the root. */
|
||||
readonly selfKey: string
|
||||
/** Prefix of every descendant key. `''` for the root. */
|
||||
readonly prefix: string
|
||||
/** Exclusive upper bound of the subtree range; `undefined` for the root. */
|
||||
readonly end: string | undefined
|
||||
|
||||
readonly #ctx: KVContext
|
||||
readonly #esc: string
|
||||
readonly #state: TrieState<IndexDefs>
|
||||
readonly #selfIdx: BoundIndexes | undefined
|
||||
readonly #childIdx: BoundIndexes | undefined
|
||||
|
||||
/** @internal Use `KVStorage` and `child()`. */
|
||||
constructor(ctx: KVContext, path: readonly string[], selfKey: string, esc: string, state: TrieState<IndexDefs>, selfIdx: BoundIndexes | undefined) {
|
||||
this.#ctx = ctx
|
||||
this.path = path
|
||||
this.selfKey = selfKey
|
||||
this.#esc = esc
|
||||
const root = path.length === 0
|
||||
this.prefix = root ? '' : selfKey + ctx.sep
|
||||
this.end = root ? undefined : selfKey + succ(ctx.sep)
|
||||
this.#state = state
|
||||
this.#selfIdx = selfIdx
|
||||
this.#childIdx = ctx.registry.bind(state, this.prefix)
|
||||
}
|
||||
|
||||
get driver(): KVDriver {
|
||||
return this.#ctx.driver
|
||||
}
|
||||
|
||||
get separator(): string {
|
||||
return this.#ctx.sep
|
||||
}
|
||||
|
||||
get isRoot(): boolean {
|
||||
return this.path.length === 0
|
||||
}
|
||||
|
||||
child<K extends KeyOf<S>>(key: K): KVNode<ChildOf<S, K>> {
|
||||
const seg = toSegment(key)
|
||||
const esc = escapeSegment(seg, this.#ctx.sep)
|
||||
return new KVNode(this.#ctx, [...this.path, seg], this.prefix + esc, esc, this.#ctx.registry.trie.step(this.#state, seg), this.#childIdx)
|
||||
}
|
||||
|
||||
/** Re-types this node. No runtime effect. */
|
||||
as<S2 extends Spec>(): KVNode<S2> {
|
||||
return this as unknown as KVNode<S2>
|
||||
}
|
||||
|
||||
get(): Promise<ValueOf<S> | undefined>
|
||||
get<K extends KeyOf<S>>(key: K): Promise<ValueOf<ChildOf<S, K>> | undefined>
|
||||
async get(...args: unknown[]): Promise<unknown> {
|
||||
const t = this.#resolve(args, 0)
|
||||
return (await this.#ctx.driver.getMany([t.key]))[0]
|
||||
}
|
||||
|
||||
async getMany<K extends KeyOf<S>>(keys: readonly K[]): Promise<(ValueOf<ChildOf<S, K>> | undefined)[]> {
|
||||
if (keys.length === 0) return []
|
||||
return (await this.#ctx.driver.getMany(keys.map((k) => this.#childKey(k)))) as (ValueOf<ChildOf<S, K>> | undefined)[]
|
||||
}
|
||||
|
||||
set(value: ValueOf<S>): Promise<void>
|
||||
set<K extends KeyOf<S>>(key: K, value: ValueOf<ChildOf<S, K>>): Promise<void>
|
||||
async set(...args: unknown[]): Promise<void> {
|
||||
const t = this.#resolve(args, 1)
|
||||
const value = args[args.length - 1]
|
||||
if (value === undefined) throw new TypeError('Cannot set undefined; use delete()')
|
||||
return this.#write(t.idx, [{ key: t.key, esc: t.esc, value }])
|
||||
}
|
||||
|
||||
async setMany(entries: readonly EntryOf<S>[]): Promise<void> {
|
||||
if (entries.length === 0) return
|
||||
const slots = entries.map(([k, value]): Slot => {
|
||||
if (value === undefined) throw new TypeError('Cannot set undefined; use delete()')
|
||||
const esc = escapeSegment(toSegment(k), this.#ctx.sep)
|
||||
return { key: this.prefix + esc, esc, value }
|
||||
})
|
||||
return this.#write(this.#childIdx, slots)
|
||||
}
|
||||
|
||||
/** Deletes the own value (or a child's own value) only, not descendants. */
|
||||
delete(): Promise<void>
|
||||
delete<K extends KeyOf<S>>(key: K): Promise<void>
|
||||
async delete(...args: unknown[]): Promise<void> {
|
||||
const t = this.#resolve(args, 0)
|
||||
return this.#write(t.idx, [{ key: t.key, esc: t.esc, value: undefined }])
|
||||
}
|
||||
|
||||
async deleteMany(keys: readonly KeyOf<S>[]): Promise<void> {
|
||||
if (keys.length === 0) return
|
||||
const slots = keys.map((k): Slot => {
|
||||
const esc = escapeSegment(toSegment(k), this.#ctx.sep)
|
||||
return { key: this.prefix + esc, esc, value: undefined }
|
||||
})
|
||||
return this.#write(this.#childIdx, slots)
|
||||
}
|
||||
|
||||
/** Deletes the own value and all descendants (or a child's). On the root, deletes everything. */
|
||||
deleteTree(): Promise<void>
|
||||
deleteTree<K extends KeyOf<S>>(key: K): Promise<void>
|
||||
async deleteTree(...args: unknown[]): Promise<void> {
|
||||
if (args.length === 0 && this.isRoot) return this.#deleteRange('', undefined)
|
||||
const t = this.#resolve(args, 0)
|
||||
await this.#write(t.idx, [{ key: t.key, esc: t.esc, value: undefined }])
|
||||
await this.#deleteRange(t.key + this.#ctx.sep, t.key + succ(this.#ctx.sep))
|
||||
}
|
||||
|
||||
/**
|
||||
* Direct child keys (unescaped), including children that only have descendants.
|
||||
* Order: see `scanChildren` — lexicographic by escaped segment followed by the separator.
|
||||
*/
|
||||
async *keys(opts: KeysOptions = {}): AsyncGenerator<KeyString<S>> {
|
||||
const after = opts.after === undefined ? undefined : toSegment(opts.after)
|
||||
for await (const hit of scanChildren(this.#ctx.driver, this.prefix, this.end, this.#ctx.sep, {
|
||||
after,
|
||||
reverse: opts.reverse,
|
||||
limit: opts.limit,
|
||||
})) {
|
||||
yield unescapeSegment(hit.esc) as KeyString<S>
|
||||
}
|
||||
}
|
||||
|
||||
/** Direct children that have an own value, as `[key, value]`. */
|
||||
async *entries(opts: KeysOptions = {}): AsyncGenerator<StringEntryOf<S>> {
|
||||
const after = opts.after === undefined ? undefined : toSegment(opts.after)
|
||||
let remaining = opts.limit ?? Infinity
|
||||
if (remaining <= 0) return
|
||||
for await (const hit of scanChildren(this.#ctx.driver, this.prefix, this.end, this.#ctx.sep, {
|
||||
after,
|
||||
reverse: opts.reverse,
|
||||
values: true,
|
||||
})) {
|
||||
if (!hit.hasOwn) continue
|
||||
yield [unescapeSegment(hit.esc), hit.value] as StringEntryOf<S>
|
||||
if (--remaining <= 0) return
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Assembles descendants up to `depth` levels into a plain object: leaves become values,
|
||||
* nodes with children become objects. A node with both an own value and children exposes the
|
||||
* own value under the `''` property (empty keys are invalid, so it can't collide).
|
||||
* Reserved (`~…`) subtrees are skipped. This node's own value is not included.
|
||||
*/
|
||||
async tree<T = Record<string, unknown>>(opts: TreeOptions): Promise<T> {
|
||||
const { depth } = opts
|
||||
if (!(depth >= 1)) throw new RangeError('tree() depth must be >= 1')
|
||||
const { driver, sep } = this.#ctx
|
||||
type TNode = { own?: unknown; has: boolean; kids?: Map<string, TNode> }
|
||||
const root: TNode = { has: false }
|
||||
const prefix = this.prefix
|
||||
let it = driver.scan({ gte: prefix, lt: this.end, values: true })[Symbol.asyncIterator]()
|
||||
const reseek = async (gte: string) => {
|
||||
await it.return?.()
|
||||
it = driver.scan({ gte, lt: this.end, values: true })[Symbol.asyncIterator]()
|
||||
}
|
||||
try {
|
||||
outer: for (;;) {
|
||||
const r = await it.next()
|
||||
if (r.done) break
|
||||
const [key, value] = r.value
|
||||
const segs = key.slice(prefix.length).split(sep)
|
||||
let base = prefix
|
||||
for (let i = 0; i < segs.length && i < depth; i++) {
|
||||
if (segs[i]!.startsWith('~')) {
|
||||
await reseek(base + succ('~'))
|
||||
continue outer
|
||||
}
|
||||
base += segs[i]! + sep
|
||||
}
|
||||
if (segs.length > depth) {
|
||||
await reseek(prefix + segs.slice(0, depth).join(sep) + succ(sep))
|
||||
continue
|
||||
}
|
||||
let node = root
|
||||
for (const s of segs) {
|
||||
const kids = (node.kids ??= new Map())
|
||||
const seg = unescapeSegment(s)
|
||||
let next = kids.get(seg)
|
||||
if (!next) kids.set(seg, (next = { has: false }))
|
||||
node = next
|
||||
}
|
||||
node.own = value
|
||||
node.has = true
|
||||
}
|
||||
} finally {
|
||||
await it.return?.()
|
||||
}
|
||||
const build = (n: TNode): unknown => {
|
||||
if (!n.kids) return n.own
|
||||
const obj: Record<string, unknown> = {}
|
||||
if (n.has) obj[''] = n.own
|
||||
for (const [k, v] of n.kids) obj[k] = build(v)
|
||||
return obj
|
||||
}
|
||||
return (root.kids ? build(root) : {}) as T
|
||||
}
|
||||
|
||||
/**
|
||||
* Looks up direct children by an index declared for this node, by exact value or range.
|
||||
* Each hit's value is re-read and re-checked; stale entries are skipped (and deleted when
|
||||
* `readRepair` is on). Multi-entry hits are de-duplicated.
|
||||
*/
|
||||
async *findBy(index: string, query: IndexQuery, opts: FindByOptions = {}): AsyncGenerator<StringEntryOf<S>> {
|
||||
const idx = this.#childIdx
|
||||
if (!idx || !idx.has(index)) throw new TypeError(`No index "${index}" on node "${this.selfKey}"`)
|
||||
const { driver, readRepair } = this.#ctx
|
||||
const { gte, lt } = idx.range(index, query)
|
||||
let remaining = opts.limit ?? Infinity
|
||||
if (remaining <= 0) return
|
||||
const seen = isIndexRange(query) ? new Set<string>() : undefined
|
||||
let chunk: string[] = []
|
||||
const it = driver.scan({ gte, lt, reverse: opts.reverse })[Symbol.asyncIterator]()
|
||||
|
||||
// Processes one chunk of entry keys; returns false when the limit is reached.
|
||||
const flush = async function* (this: KVNode<S>): AsyncGenerator<StringEntryOf<S>, boolean> {
|
||||
const entries = chunk
|
||||
chunk = []
|
||||
const escs = entries.map((e) => idx.childOf(index, e))
|
||||
const values = await driver.getMany(escs.map((e) => this.prefix + e))
|
||||
const stale: string[] = []
|
||||
try {
|
||||
for (let i = 0; i < entries.length; i++) {
|
||||
const esc = escs[i]!
|
||||
if (!idx.matches(index, esc, values[i], entries[i]!)) {
|
||||
stale.push(entries[i]!)
|
||||
continue
|
||||
}
|
||||
if (seen) {
|
||||
if (seen.has(esc)) continue
|
||||
seen.add(esc)
|
||||
}
|
||||
yield [unescapeSegment(esc), values[i]] as StringEntryOf<S>
|
||||
if (--remaining <= 0) return false
|
||||
}
|
||||
return true
|
||||
} 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])
|
||||
// Small first chunk so `findOne`-style lookups cost one probe.
|
||||
if (chunk.length >= (remaining < FIND_CHUNK ? remaining : FIND_CHUNK)) {
|
||||
if (!(yield* flush.call(this))) return
|
||||
}
|
||||
}
|
||||
if (chunk.length) yield* flush.call(this)
|
||||
} finally {
|
||||
await it.return?.()
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------- internals
|
||||
|
||||
#childKey(key: unknown): string {
|
||||
return this.prefix + escapeSegment(toSegment(key), this.#ctx.sep)
|
||||
}
|
||||
|
||||
/** Self vs child by argument count; `selfArity` is the self form's argument count. */
|
||||
#resolve(args: readonly unknown[], selfArity: number): { key: string; esc: string; idx: BoundIndexes | undefined } {
|
||||
if (args.length <= selfArity) {
|
||||
if (this.isRoot) throw new TypeError('The root node has no own value')
|
||||
return { key: this.selfKey, esc: this.#esc, idx: this.#selfIdx }
|
||||
}
|
||||
const esc = escapeSegment(toSegment(args[0]), this.#ctx.sep)
|
||||
return { key: this.prefix + esc, esc, idx: this.#childIdx }
|
||||
}
|
||||
|
||||
/** Writes slots (`value: undefined` = delete), maintaining indexes when present. */
|
||||
async #write(idx: BoundIndexes | undefined, slots: Slot[]): Promise<void> {
|
||||
const { driver } = this.#ctx
|
||||
if (!idx) {
|
||||
const sets: [string, unknown][] = []
|
||||
const dels: string[] = []
|
||||
for (const s of slots) {
|
||||
if (s.value === undefined) dels.push(s.key)
|
||||
else sets.push([s.key, s.value])
|
||||
}
|
||||
if (sets.length) await driver.setMany(sets)
|
||||
if (dels.length) await driver.deleteMany(dels)
|
||||
return
|
||||
}
|
||||
// Last write per key wins.
|
||||
const byKey = new Map<string, Slot>()
|
||||
for (const s of slots) byKey.set(s.key, s)
|
||||
const unique = [...byKey.values()]
|
||||
const old = await driver.getMany(unique.map((s) => s.key))
|
||||
const adds: string[] = []
|
||||
const removes: string[] = []
|
||||
for (let i = 0; i < unique.length; i++) {
|
||||
const s = unique[i]!
|
||||
const before = idx.entriesFor(s.esc, old[i])
|
||||
const after = idx.entriesFor(s.esc, s.value)
|
||||
for (const e of after) if (!before.has(e)) adds.push(e)
|
||||
for (const e of before) if (!after.has(e)) removes.push(e)
|
||||
}
|
||||
if (driver.batch) {
|
||||
const ops: BatchOp[] = []
|
||||
for (const k of adds) ops.push({ type: 'set', key: k, value: true })
|
||||
for (const s of unique) ops.push(s.value === undefined ? { type: 'delete', key: s.key } : { type: 'set', key: s.key, value: s.value })
|
||||
for (const k of removes) ops.push({ type: 'delete', key: k })
|
||||
return driver.batch(ops)
|
||||
}
|
||||
// Non-atomic: adds first, then values, then removes. Entries are hints; findBy re-checks.
|
||||
if (adds.length) await driver.setMany(adds.map((k) => [k, true]))
|
||||
const sets = unique.filter((s) => s.value !== undefined)
|
||||
const dels = unique.filter((s) => s.value === undefined)
|
||||
if (sets.length) await driver.setMany(sets.map((s) => [s.key, s.value]))
|
||||
if (dels.length) await driver.deleteMany(dels.map((s) => s.key))
|
||||
if (removes.length) await driver.deleteMany(removes)
|
||||
}
|
||||
|
||||
async #deleteRange(gte: string, lt: string | undefined): Promise<void> {
|
||||
const { driver } = this.#ctx
|
||||
if (lt !== undefined && driver.deleteRange) return driver.deleteRange(gte, lt)
|
||||
for (;;) {
|
||||
const keys: string[] = []
|
||||
for await (const [k] of driver.scan({ gte, lt, limit: DELETE_CHUNK })) keys.push(k)
|
||||
if (keys.length === 0) return
|
||||
await driver.deleteMany(keys)
|
||||
if (keys.length < DELETE_CHUNK) return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** The root `KVNode` over a driver. */
|
||||
export class KVStorage<S extends Spec = AnySpec> extends KVNode<S> {
|
||||
constructor(driver: KVDriver, opts: KVStorageOptions = {}) {
|
||||
const sep = opts.separator ?? DEFAULT_SEPARATOR
|
||||
validateSeparator(sep)
|
||||
const registry = new IndexRegistry(opts.indexes ?? {}, sep)
|
||||
super({ driver, sep, registry, readRepair: opts.readRepair ?? true }, [], '', '', registry.trie.rootState, undefined)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user