diff --git a/.gitea/workflows/ci.yml b/.gitea/workflows/ci.yml index 4789154..abbcc9b 100644 --- a/.gitea/workflows/ci.yml +++ b/.gitea/workflows/ci.yml @@ -12,7 +12,7 @@ jobs: strategy: fail-fast: false matrix: - node: [20, 22] + node: [22, 24, 26] steps: - uses: actions/checkout@v4 - uses: actions/setup-node@v4 diff --git a/CLAUDE.md b/CLAUDE.md index 9d9b3f6..507cc91 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -10,7 +10,7 @@ Layered TypeScript storage library: KV → Events → Documents. The design is i - `npm run typecheck`, `npm run build`, `npm run bench` - `npm run format` / `npm run format:check`: Prettier -CI (`.gitea/workflows/ci.yml`, Gitea Actions) runs the format check, typecheck, tests and build on Node 20 and 22. +CI (`.gitea/workflows/ci.yml`, Gitea Actions) runs the format check, typecheck, tests and build on Node 22, 24 and 26. ## Conventions @@ -18,7 +18,11 @@ CI (`.gitea/workflows/ci.yml`, Gitea Actions) runs the format check, typecheck, primitives, and the user-facing API lives in the base class. - Classes over loose functions; TS `private`, not `#`; no overloads (use `param?` or tuple-union rest params); don't override base methods in subclasses. -- No runtime dependencies. Tests live only in `test/`, mirroring `src/`. +- **Never reassign a field after construction.** `child()` nodes are frozen `Object.create(root)` + views, so `this.x = ...` on a child throws. Keep mutable state in a `private readonly state` + object or in Maps (see `KVStorageComposite`, `KVStorageHttp`). +- No runtime dependencies; third-party code is vendored (`src/utils/sse-session/`, `event-emitter.ts`, + `extended-json.ts`, with a header naming the source). Tests live only in `test/`, mirroring `src/`. - New KV/document implementations must pass `test/kv-storage/kv-storage.conformance.ts` / `test/document-storage/document-storage.suite.ts`. - Ask the owner before pushing. diff --git a/DESIGN.md b/DESIGN.md index 4021168..bca1a8c 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -14,6 +14,9 @@ section 7 lists what changed and why. that delegates the primitives. 2. **One navigation primitive: `child(key)`.** Every instance is a node; `child` returns a view of the same storage rooted deeper (via `Object.create`, so it shares the instance's state). Sync, no I/O. + Child nodes are **frozen**: an implementation that reassigns a field while running on a child + would silently shadow the root's value, so that throws instead. Implementations keep mutable + state in a `readonly state` object or in Maps. 3. **Self or child by argument.** `get()` reads this node, `get(key)` a child. A key that is passed but `undefined`/`null` throws (`InvalidKeyError`) instead of silently meaning "self". Methods use optional parameters or tuple-union rest parameters, not overloads. @@ -21,16 +24,21 @@ section 7 lists what changed and why. suits the backend (in-memory equality maps, native Mongo/SQLite/IndexedDB indexes, or nothing). 5. **Simple over clever.** Code style: classes, TS `private` (not `#`), no method overrides in subclasses, tests only under `test/`. +6. **No runtime dependencies.** Third-party code is vendored into `src/utils/` with a header naming + its source: `sse-session/` (Harvmaster/sse-session), `event-emitter.ts` and `extended-json.ts` + (xo-cash-utils, branch installable-sse). Node ≥ 22; CI runs Node 22, 24 and 26. ## 2. Layout ``` src/ - kv-storage/ base-kv-storage.ts, kv-storage-in-memory.ts, kv-storage-http.ts, kv-storage-http-server.ts + kv-storage/ base-kv-storage.ts, kv-storage-in-memory.ts, kv-storage-indexeddb.ts, + kv-storage-http.ts, kv-storage-http-server.ts, kv-storage-composite.ts event-storage/ base-event-storage.ts, event-storage-kv.ts document-storage/ base-document-storage.ts, document-storage-in-memory.ts, document-storage-kv.ts, event-sourced-document-storage.ts - utils/ keys, sorted-map, json, ulid, errors, schema (types), document-path/query/update + utils/ keys, sorted-map, json, extended-json, event-emitter, ulid, errors, schema (types), + document-path/query/update, sse-session/ (vendored) test/ mirrors src/; shared suites: kv-storage.conformance.ts, document-storage.suite.ts bench/ ``` @@ -49,15 +57,24 @@ bench/ ## 4. KV storage -`BaseKVStorage` primitives (absolute keys): +`BaseKVStorage` primitives (absolute keys). Implementations provide the reads and the protected +`apply*` writes; the public `writeMany` / `removeMany` / `writeIfAbsent` live in the base class and +emit a `change` event after each write, so every store emits change events. -| Primitive | | -| ------------------------------------------- | ---------------------------------------------------------------------------------------------------- | -| `readMany(keys)` | values in order, `undefined` for missing | -| `writeMany(entries)` | | -| `removeMany(keys)` | | -| `writeIfAbsent(entries)` | `true` = written; atomic per key (needed by events) | -| `scan({ gte, lt, limit, reverse, values })` | key-ordered; must keep working if keys change mid-scan (keys present throughout appear exactly once) | +| Primitive | | +| ------------------------------------------------------ | ---------------------------------------------------------------------------------------------------- | +| `readMany(keys)` | values in order, `undefined` for missing | +| `scan({ gte, lt, limit, reverse, values })` | key-ordered; must keep working if keys change mid-scan (keys present throughout appear exactly once) | +| `applyWrite(entries)` → public `writeMany` | | +| `applyRemove(keys)` → public `removeMany` | | +| `applyWriteIfAbsent(entries)` → public `writeIfAbsent` | `true` = written; atomic per key (needed by events) | + +**Values** are JSON plus `bigint` and `Uint8Array`. Anything crossing the network is encoded as +Extended JSON (``, ``); IndexedDB stores them natively. + +**Events** (`EventEmitter`, shared by all nodes of a storage): `change` → +`{ changes: { key, value? }[], origin: 'local' | 'remote', id? }` (no `value` = removed), and +`resync` → changes may have been missed, so reload. Node API (defined once): `child`, `as`, `get(key?)`, `getMany`, `set(value)` / `set(key, value)`, `setMany`, `delete(key?)`, `deleteMany`, `deleteTree(key?)`, `keys({ limit, after, reverse })`, @@ -68,14 +85,66 @@ Node API (defined once): `child`, `as`, `get(key?)`, `getMany`, `set(value)` / ` reserved segments. - **The only KV "index" is the ordered key space.** There is no `find`: with KV you know where to look, or you list a node's children. -- **Implementations:** `KVStorageInMemory` (`SortedMap`: sorted chunks of ≤1024 keys; values deep-copied); - `KVStorageHttp` (client) + `KVStorageHttpServer` (wraps any KV storage; `handle(Request) → Response`, - `POST /` with JSON bodies, `scan` streamed as NDJSON). The server exposes the whole - underlying store (primitives use absolute keys). +- **Implementations:** + - `KVStorageInMemory`: `SortedMap` (sorted chunks of ≤1024 keys); values deep-copied. + - `KVStorageIndexedDB`: one object store with string keys. Scans read in batches, each in its own + transaction, resuming after the last key (IndexedDB transactions close when the code awaits + anything else). `writeIfAbsent` runs in one readwrite transaction. + - `KVStorageHttp` (client) and `KVStorageHttpServer` (wraps any KV storage; + `handle(Request) → Response`): `POST /` with Extended JSON bodies; `scan` streams + NDJSON. Unreachable server (fetch failure, 502/503/504) → `StorageUnavailableError`. The server + exposes the whole underlying store (primitives use absolute keys). + - `KVStorageComposite`: see 4.1. +- **Change stream (SSE).** `GET /events` on the server streams every change to the wrapped storage + as `event: change` with id `-` (epoch = server instance). The server keeps the last + `historySize` (10 000) change batches. + - A client reconnecting with `Last-Event-ID` gets the batches it missed, or `event: resync` if it + can't be resumed (another epoch, or too old). + - Every catch-up ends with `event: ready`. + - Keep-alive comments are sent every 15 s. + - `KVStorageHttp.watch()` subscribes (via the vendored `SSESession`, which reconnects with + backoff), emits `change` events with `origin: 'remote'` and `resync`, and resolves once caught up. + - Limitation: a server only sees writes made through its own storage instance, so several server + processes over one database need a shared pub/sub (not built). - **Wrappers** (e.g. a future encryption layer) extend `BaseKVStorage` and delegate the primitives. Keys must stay plaintext (or be deterministically encrypted) for prefix scans to work; values can be encrypted freely. +### 4.1 `KVStorageComposite({ source, replicas, retry? })` + +Keeps full local copies of a source (e.g. `KVStorageHttp`) so reads are local and the app works +offline. Replicas are listed fastest first (e.g. `[inMemory, indexedDb]`). The last replica is the +persistent one: it seeds the others on start and stores the composite's internal keys under `~q/` +(hidden from reads): `~q/meta` (`{ lastEventId }`, present once a sync has completed) and the +outbox `~q/o/`. + +- **Start** (lazy, on first use): + 1. Load the outbox and copy the persistent replica into the faster ones. + 2. If a previous sync exists, serve reads locally straight away. + 3. Watch the source, resuming from `lastEventId`. + 4. Run a full sync only when there's no resume point, when the source sends `resync`, or when the + source can't be watched. + 5. Before the first sync completes, reads go to the source. +- **Remote changes** (and full syncs) are applied in order through one queue. A change that arrives + during a full sync waits behind it, so the end state always matches the source. `lastEventId` is + saved after each applied batch. Changes the replicas already hold are skipped. +- **Full sync** merge-scans the source against the persistent replica (both key-ordered) and applies + the difference to every replica in batches of 500. +- **Writes** go to the source, then every replica. + - On `StorageUnavailableError` (or while anything is queued, to keep order), the write is applied + locally and appended to the outbox. The outbox is replayed in order with exponential backoff, + and also whenever the source shows signs of life. Last writer wins on conflicts. + - Remote changes to keys with queued writes are ignored until those writes are replayed. + - `writeIfAbsent` always needs the source. +- **Echo tracking.** Each write sent to a watched source expects one echo per key. A remote change + to a key with an outstanding echo is consumed instead of applied. The stream is in server order, + so that change is either the echo or older than it, and this avoids briefly showing a stale value. + A full sync clears the counts. +- Write through the composite, not directly through the same source instance (its `local`-origin + changes are not mirrored). +- `start()`, `synced()` (complete copy, all received changes applied), `queued` (outbox size), + `close()`. + ## 5. Event storage `BaseEventStorage` primitives over stream paths: `appendTo`, `readFrom`, `versionOf`, `streamsOf`. @@ -153,6 +222,10 @@ Node API: `get(key?)`, `getMany`, `set(doc)` / `set(key, doc)`, `insert(doc)` ## 8. Open questions +- Composite: replicate a subtree instead of the whole store; conflict policies beyond last writer + wins for offline writes. +- Multi-process servers need a shared change feed (pub/sub) for SSE. + - Hard deletes from immutable event logs (privacy requests): compaction vs crypto-shredding. The encryption wrapper could support per-subject keys. - Order-preserving numeric keys (`9` before `10`) as an opt-in key encoding. diff --git a/package-lock.json b/package-lock.json index 1a4a335..40b5f5e 100644 --- a/package-lock.json +++ b/package-lock.json @@ -9,6 +9,7 @@ "version": "0.1.0", "devDependencies": { "@types/node": "^22.20.5", + "fake-indexeddb": "^6.2.6", "fast-check": "^4.10.2", "prettier": "^3.9.10", "typescript": "^5.9.3", @@ -16,14 +17,6 @@ }, "engines": { "node": ">=20" - }, - "peerDependencies": { - "vitest": ">=3" - }, - "peerDependenciesMeta": { - "vitest": { - "optional": true - } } }, "node_modules/@esbuild/aix-ppc64": { @@ -1136,6 +1129,16 @@ "node": ">=12.0.0" } }, + "node_modules/fake-indexeddb": { + "version": "6.2.6", + "resolved": "https://registry.npmjs.org/fake-indexeddb/-/fake-indexeddb-6.2.6.tgz", + "integrity": "sha512-ihhmr0KoET51jT8R4YhbUYuLLvr1qqO6INISygtN16i8loR8kAxtAUr4+EKHTSMjL+lggaAn4OC7DRqzrNgnuQ==", + "dev": true, + "license": "Apache-2.0", + "engines": { + "node": ">=18" + } + }, "node_modules/fast-check": { "version": "4.10.2", "resolved": "https://registry.npmjs.org/fast-check/-/fast-check-4.10.2.tgz", diff --git a/package.json b/package.json index 1f1e75a..8c502b3 100644 --- a/package.json +++ b/package.json @@ -4,7 +4,7 @@ "description": "Layered TypeScript storage: KV -> Events -> Documents", "type": "module", "engines": { - "node": ">=20" + "node": ">=22" }, "main": "./dist/index.js", "types": "./dist/index.d.ts", @@ -29,6 +29,7 @@ }, "devDependencies": { "@types/node": "^22.20.5", + "fake-indexeddb": "^6.2.6", "fast-check": "^4.10.2", "prettier": "^3.9.10", "typescript": "^5.9.3", diff --git a/src/document-storage/base-document-storage.ts b/src/document-storage/base-document-storage.ts index 7730ebf..cf9bf09 100644 --- a/src/document-storage/base-document-storage.ts +++ b/src/document-storage/base-document-storage.ts @@ -27,7 +27,7 @@ export type ChildDoc = ValueOf>> */ export abstract class BaseDocumentStorage { /** Unescaped path of this node (`[]` for the root, which has no own document). */ - path: readonly string[] = [] + readonly path: readonly string[] = [] protected readonly ids = new Ulid() abstract readMany(paths: (readonly string[])[]): Promise<(Doc | undefined)[]> @@ -40,8 +40,11 @@ export abstract class BaseDocumentStorage { abstract query(parent: readonly string[], query: DocumentQuery): AsyncIterable child>(key: K): BaseDocumentStorage> { - const node = Object.create(this) as BaseDocumentStorage> - node.path = [...this.path, Keys.segment(key)] + const node = Object.create(this, { path: { value: [...this.path, Keys.segment(key)], enumerable: true } }) as BaseDocumentStorage> + // Frozen: child nodes share the root's state through the prototype, so an implementation that + // reassigns a field (`this.x = ...`) while running on a child would silently shadow it. + // Keep mutable state in objects (Maps, a `state` object) instead; this makes mistakes throw. + Object.freeze(node) return node } diff --git a/src/event-storage/base-event-storage.ts b/src/event-storage/base-event-storage.ts index 173e361..4daaabf 100644 --- a/src/event-storage/base-event-storage.ts +++ b/src/event-storage/base-event-storage.ts @@ -39,7 +39,7 @@ export interface StreamKeysOptions { */ export abstract class BaseEventStorage { /** Unescaped path of this node (`[]` for the root, which is not a stream). */ - path: readonly string[] = [] + readonly path: readonly string[] = [] abstract appendTo(stream: readonly string[], events: unknown[], expectedVersion: number | 'any'): Promise abstract readFrom(stream: readonly string[], options: ReadOptions): AsyncIterable> @@ -47,8 +47,11 @@ export abstract class BaseEventStorage { abstract streamsOf(parent: readonly string[], options: StreamKeysOptions): AsyncIterable child>(key: K): BaseEventStorage> { - const node = Object.create(this) as BaseEventStorage> - node.path = [...this.path, Keys.segment(key)] + const node = Object.create(this, { path: { value: [...this.path, Keys.segment(key)], enumerable: true } }) as BaseEventStorage> + // Frozen: child nodes share the root's state through the prototype, so an implementation that + // reassigns a field (`this.x = ...`) while running on a child would silently shadow it. + // Keep mutable state in objects (Maps, a `state` object) instead; this makes mistakes throw. + Object.freeze(node) return node } diff --git a/src/index.ts b/src/index.ts index 78841b9..4ef085e 100644 --- a/src/index.ts +++ b/src/index.ts @@ -2,6 +2,8 @@ export * from './kv-storage/base-kv-storage.js' export * from './kv-storage/kv-storage-in-memory.js' export * from './kv-storage/kv-storage-http.js' export * from './kv-storage/kv-storage-http-server.js' +export * from './kv-storage/kv-storage-indexeddb.js' +export * from './kv-storage/kv-storage-composite.js' export * from './event-storage/base-event-storage.js' export * from './event-storage/event-storage-kv.js' export * from './document-storage/base-document-storage.js' @@ -12,6 +14,8 @@ export * from './utils/document-path.js' export * from './utils/document-query.js' export * from './utils/document-update.js' export * from './utils/errors.js' +export * from './utils/event-emitter.js' +export * from './utils/extended-json.js' export * from './utils/json.js' export * from './utils/keys.js' export * from './utils/schema.js' diff --git a/src/kv-storage/base-kv-storage.ts b/src/kv-storage/base-kv-storage.ts index e13a874..08dc823 100644 --- a/src/kv-storage/base-kv-storage.ts +++ b/src/kv-storage/base-kv-storage.ts @@ -1,3 +1,4 @@ +import { EventEmitter } from '../utils/event-emitter.js' import { Keys } from '../utils/keys.js' import type { AnySpec, ChildOf, EntryOf, KeyOf, KeyString, Spec, StringEntryOf, ValueAt, ValueOf } from '../utils/schema.js' @@ -19,39 +20,90 @@ export interface KeysOptions { reverse?: boolean } +/** One key's new value; `value: undefined` means it was removed. */ +export interface KVChange { + key: string + value?: unknown +} + +export interface KVChangeEvent { + changes: KVChange[] + /** `local`: made through this instance. `remote`: observed from elsewhere (e.g. pushed by a server). */ + origin: 'local' | 'remote' + /** Position in the source's change stream, when it has one (used to resume). */ + id?: string +} + +export type KVStorageEvents = { + change: KVChangeEvent + /** Changes may have been missed (e.g. a resume point was too old): consumers should reload. */ + resync: { id?: string } +} + /** - * Ordered key-value storage. Implementations provide five raw primitives over absolute keys; - * everything else (navigation, self/child reads and writes, child listing) is defined here once. + * Ordered key-value storage. Implementations provide five raw primitives over absolute keys + * (`readMany`, `scan`, and the protected `apply*` writes); everything else (navigation, self/child + * reads and writes, child listing, change events) is defined here once. + * + * Every write emits a `change` event from the storage (shared by all its child nodes). * * Every instance is a node: `child(key)` returns a view of the same storage rooted at that key. * Methods taking an optional `key` act on that child, or on this node's own value when omitted. * A key that is passed but `undefined`/`null` throws rather than silently meaning "self". */ -export abstract class BaseKVStorage { +export abstract class BaseKVStorage extends EventEmitter { /** Unescaped path of this node (`[]` for the root). */ - path: readonly string[] = [] + readonly path: readonly string[] = [] /** This node's own key (`''` for the root). */ - key = '' + readonly key: string = '' /** Prefix of every descendant key (`''` for the root). */ - prefix = '' + readonly prefix: string = '' /** Values for absolute keys, in order; `undefined` for missing. */ abstract readMany(keys: string[]): Promise - abstract writeMany(entries: [key: string, value: unknown][]): Promise - abstract removeMany(keys: string[]): Promise - /** Writes each entry only if its key is absent; `true` = written. Must be atomic per key. */ - abstract writeIfAbsent(entries: [key: string, value: unknown][]): Promise /** Entries in key order (descending when `reverse`). */ abstract scan(options?: ScanOptions): AsyncIterable<[key: string, value: unknown]> + protected abstract applyWrite(entries: [key: string, value: unknown][]): Promise + protected abstract applyRemove(keys: string[]): Promise + /** Writes each entry only if its key is absent; `true` = written. Must be atomic per key. */ + protected abstract applyWriteIfAbsent(entries: [key: string, value: unknown][]): Promise + + // ------------------------------------------------------------------ raw writes (absolute keys) + + async writeMany(entries: [key: string, value: unknown][]): Promise { + if (entries.length === 0) return + await this.applyWrite(entries) + this.emit('change', { changes: entries.map(([key, value]) => ({ key, value })), origin: 'local' }) + } + + async removeMany(keys: string[]): Promise { + if (keys.length === 0) return + await this.applyRemove(keys) + this.emit('change', { changes: keys.map((key) => ({ key })), origin: 'local' }) + } + + async writeIfAbsent(entries: [key: string, value: unknown][]): Promise { + if (entries.length === 0) return [] + const written = await this.applyWriteIfAbsent(entries) + const changes = entries.filter((_, i) => written[i]).map(([key, value]) => ({ key, value })) + if (changes.length) this.emit('change', { changes, origin: 'local' }) + return written + } // ------------------------------------------------------------------ navigation /** A view of this storage rooted at `key`. Synchronous; no I/O. */ child>(key: K): BaseKVStorage> { - const node = Object.create(this) as BaseKVStorage> - node.path = [...this.path, Keys.segment(key)] - node.key = this.prefix + Keys.escape(Keys.segment(key)) - node.prefix = node.key + Keys.separator + const own = this.prefix + Keys.escape(Keys.segment(key)) + const node = Object.create(this, { + path: { value: [...this.path, Keys.segment(key)], enumerable: true }, + key: { value: own, enumerable: true }, + prefix: { value: own + Keys.separator, enumerable: true }, + }) as BaseKVStorage> + // Frozen: child nodes share the root's state through the prototype, so an implementation that + // reassigns a field (`this.x = ...`) while running on a child would silently shadow it. + // Keep mutable state in objects (Maps, a `state` object) instead; this makes mistakes throw. + Object.freeze(node) return node } diff --git a/src/kv-storage/kv-storage-composite.ts b/src/kv-storage/kv-storage-composite.ts new file mode 100644 index 0000000..a61ab3d --- /dev/null +++ b/src/kv-storage/kv-storage-composite.ts @@ -0,0 +1,379 @@ +import { StorageUnavailableError } from '../utils/errors.js' +import type { OffCallback } from '../utils/event-emitter.js' +import { Json } from '../utils/json.js' +import type { AnySpec, Spec } from '../utils/schema.js' +import { ExponentialBackoff } from '../utils/sse-session/exponential-backoff.js' +import { BaseKVStorage, type KVChange, type KVChangeEvent, type ScanOptions } from './base-kv-storage.js' + +export interface KVStorageCompositeOptions { + /** The source of truth (e.g. `KVStorageHttp`). */ + source: BaseKVStorage + /** + * Local full copies, fastest first (e.g. `[inMemory, indexedDb]`). The last one should be + * persistent: it also stores the offline outbox and sync metadata, and seeds the others on start. + */ + replicas: BaseKVStorage[] + /** Delays between attempts to replay queued writes or re-sync. */ + retry?: Partial<{ baseDelay: number; maxDelay: number }> +} + +/** Implemented by sources that can push remote changes (e.g. `KVStorageHttp`). */ +interface WatchableSource { + watch(): Promise + unwatch(): Promise + lastEventId: string | undefined +} + +type OutboxEntry = { type: 'write'; entries: [string, unknown][] } | { type: 'remove'; keys: string[] } + +interface SyncMeta { + lastEventId?: string +} + +/** Internal keys in the persistent replica; hidden from reads. */ +const INTERNAL_PREFIX = '~q/' +const INTERNAL_END = '~q0' +const META_KEY = '~q/meta' +const OUTBOX_PREFIX = '~q/o/' +const OUTBOX_END = '~q/o0' +const SYNC_BATCH = 500 + +/** + * Keeps local replicas of a source in sync, so reads (including `keys()`, `entries()` and document + * queries) are served locally, and the app keeps working offline. + * + * - **Start** (lazily, on first use): the persistent replica seeds the faster ones. If it holds a + * previous sync, reads are served from it straight away (possibly stale until caught up). + * - **Staying in sync:** remote changes pushed by the source (`watch()`, e.g. SSE) are applied to + * every replica. On restart the source resumes from the saved `lastEventId`; a full sync (diffing + * the source against the replicas) only runs the first time, or when the source says it can't + * resume (`resync`). Without `watch()`, a full sync runs on start. + * - **Writes** go to the source, then the replicas. If the source is unreachable + * (`StorageUnavailableError`), they are applied locally and queued in a persisted outbox, then + * replayed in order when it's reachable again (last writer wins on conflict). While any write is + * queued, later writes queue too, to keep their order. `writeIfAbsent` needs the source and + * fails offline. + * + * Write through the composite, not the source: changes made directly through the same source + * instance (`origin: 'local'`) are not mirrored. + */ +export class KVStorageComposite extends BaseKVStorage { + private readonly source: BaseKVStorage + private readonly replicas: BaseKVStorage[] + /** Persistent replica: seeds the others and holds the outbox and metadata. */ + private readonly persistent: BaseKVStorage + private readonly backoff: { baseDelay: number; maxDelay: number } + /** Queued-write count per key; remote changes to these keys are ignored until replayed. */ + private readonly pending = new Map() + /** + * Echoes still expected per key: one per write this client sent to a watchable source. A remote + * change to such a key is the echo or older than it (the stream is in server order), so it is + * consumed instead of applied; this avoids briefly showing a stale value. + */ + private readonly echoes = new Map() + private readonly offs: OffCallback[] = [] + /** Mutable state lives here: child nodes share this object (see `child()`). */ + private readonly state: { + started?: Promise + /** Replicas hold a complete (possibly stale) copy, so reads can be served locally. */ + hasSnapshot: boolean + snapshotReady: Promise + markSnapshot: () => void + /** + * Remote changes and full syncs are applied strictly in order through this chain. Changes that + * arrive during a full sync wait behind it, so the end state matches the source even if it + * changed during the scan. + */ + applying: Promise + replaying?: Promise + outboxSeq: number + retryTimer?: ReturnType + retryAttempt: number + closed: boolean + } + + constructor(options: KVStorageCompositeOptions) { + super() + if (options.replicas.length === 0) throw new TypeError('KVStorageComposite needs at least one replica') + this.source = options.source + this.replicas = options.replicas + this.persistent = options.replicas.at(-1)! + this.backoff = { baseDelay: 1000, maxDelay: 30_000, ...options.retry } + let markSnapshot!: () => void + const snapshotReady = new Promise((resolve) => (markSnapshot = resolve)) + this.state = { hasSnapshot: false, snapshotReady, markSnapshot, applying: Promise.resolve(), outboxSeq: 0, retryAttempt: 0, closed: false } + } + + /** Starts syncing (also happens on first use). Resolves once local data is loaded. */ + start(): Promise { + return (this.state.started ??= this.open()) + } + + /** Resolves once the replicas hold a complete copy and all received remote changes are applied. */ + async synced(): Promise { + await this.start() + await this.state.snapshotReady + await this.state.applying + } + + /** Number of writes waiting to be replayed to the source. */ + get queued(): number { + let total = 0 + for (const count of this.pending.values()) total += count + return total + } + + /** Stops watching the source and cancels retries. */ + async close(): Promise { + this.state.closed = true + clearTimeout(this.state.retryTimer) + for (const off of this.offs.splice(0)) off() + if (this.watchable) await (this.source as unknown as WatchableSource).unwatch() + } + + async readMany(keys: string[]): Promise { + await this.start() + return (this.state.hasSnapshot ? this.replicas[0]! : this.source).readMany(keys) + } + + async *scan(options: ScanOptions = {}): AsyncGenerator<[string, unknown]> { + await this.start() + if (!this.state.hasSnapshot) return yield* this.source.scan(options) + // Skip internal keys before applying the limit. + let remaining = options.limit ?? Infinity + if (remaining <= 0) return + for await (const entry of this.replicas[0]!.scan({ ...options, limit: undefined })) { + if (entry[0] >= INTERNAL_PREFIX && entry[0] < INTERNAL_END) continue + yield entry + if (--remaining <= 0) return + } + } + + protected async applyWrite(entries: [string, unknown][]): Promise { + await this.start() + await this.writeThrough({ type: 'write', entries }) + } + + protected async applyRemove(keys: string[]): Promise { + await this.start() + await this.writeThrough({ type: 'remove', keys }) + } + + protected async applyWriteIfAbsent(entries: [string, unknown][]): Promise { + await this.start() + const written = await this.source.writeIfAbsent(entries) + await this.applyToReplicas(entries.filter((_, i) => written[i]).map(([key, value]) => ({ key, value }))) + return written + } + + // ------------------------------------------------------------------ start + + private get watchable(): boolean { + return typeof (this.source as unknown as Partial).watch === 'function' + } + + private async open(): Promise { + const [meta] = (await this.persistent.readMany([META_KEY])) as (SyncMeta | undefined)[] + for await (const [key, value] of this.persistent.scan({ gte: OUTBOX_PREFIX, lt: OUTBOX_END, values: true })) { + this.state.outboxSeq = Number(key.slice(OUTBOX_PREFIX.length)) + for (const k of KVStorageComposite.keysOf(value as OutboxEntry)) this.pending.set(k, (this.pending.get(k) ?? 0) + 1) + } + await this.seedFasterReplicas() + if (meta) this.snapshotTaken() + + this.offs.push( + this.source.on('change', (event) => { + if (event.origin === 'remote') this.enqueueApply(() => this.applyRemote(event)) + if (this.queued > 0) void this.replay() + }), + this.source.on('resync', () => this.enqueueApply(() => this.fullSync())), + ) + + if (this.watchable) { + const source = this.source as unknown as WatchableSource + source.lastEventId = meta?.lastEventId + // Without a resume point, catch up on the stream first so no change after the snapshot is missed. + const watching = source.watch() + if (!meta?.lastEventId) void watching.then(() => this.enqueueApply(() => this.fullSync())) + } else { + this.enqueueApply(() => this.fullSync()) + } + if (this.queued > 0) void this.replay() + } + + /** Copies the persistent replica into the faster ones. */ + private async seedFasterReplicas(): Promise { + const faster = this.replicas.slice(0, -1) + if (faster.length === 0) return + let batch: [string, unknown][] = [] + const flush = async () => { + await Promise.all(faster.map((replica) => replica.writeMany(batch))) + batch = [] + } + for await (const entry of this.persistent.scan({ values: true })) { + if (!entry[0].startsWith(INTERNAL_PREFIX) && batch.push(entry) >= SYNC_BATCH) await flush() + } + await flush() + } + + private snapshotTaken(): void { + this.state.hasSnapshot = true + this.state.markSnapshot() + } + + // ------------------------------------------------------------------ remote → replicas + + private enqueueApply(task: () => Promise): void { + this.state.applying = this.state.applying.then(task).catch((error) => { + if (!(error instanceof StorageUnavailableError)) console.error(error) + }) + } + + private async applyRemote(event: KVChangeEvent): Promise { + const candidates = event.changes.filter(({ key }) => !this.consumeEcho(key) && !this.pending.has(key) && !key.startsWith(INTERNAL_PREFIX)) + // Skip changes the replicas already reflect. + const current = await this.replicas[0]!.readMany(candidates.map(({ key }) => key)) + const changes = candidates.filter(({ value }, i) => !Json.equal(value, current[i])) + await this.applyToReplicas(changes) + if (event.id) await this.persistent.writeMany([[META_KEY, { lastEventId: event.id } satisfies SyncMeta]]) + if (changes.length) this.emit('change', { changes, origin: 'remote', id: event.id }) + } + + /** Diffs the source against the persistent replica and applies the differences to every replica. */ + private async fullSync(): Promise { + // The diff re-reads the source, so outstanding echoes no longer matter (and may never arrive). + this.echoes.clear() + try { + const id = this.watchable ? (this.source as unknown as WatchableSource).lastEventId : undefined + let changes: KVChange[] = [] + const flush = async () => { + const applied = changes.filter(({ key }) => !this.pending.has(key)) + await this.applyToReplicas(applied) + if (applied.length) this.emit('change', { changes: applied, origin: 'remote' }) + changes = [] + } + for await (const change of this.diff()) if (changes.push(change) >= SYNC_BATCH) await flush() + await flush() + await this.persistent.writeMany([[META_KEY, { lastEventId: id } satisfies SyncMeta]]) + this.snapshotTaken() + this.state.retryAttempt = 0 + } catch (error) { + if (!(error instanceof StorageUnavailableError)) throw error + this.scheduleRetry(() => this.enqueueApply(() => this.fullSync())) + } + } + + /** Changes that turn the persistent replica into the source (both scanned in key order). */ + private async *diff(): AsyncGenerator { + const remote = this.source.scan({ values: true })[Symbol.asyncIterator]() + const local = this.persistent.scan({ values: true })[Symbol.asyncIterator]() + const nextLocal = async () => { + for (let next = await local.next(); ; next = await local.next()) if (next.done || !next.value[0].startsWith(INTERNAL_PREFIX)) return next + } + let r = await remote.next() + let l = await nextLocal() + while (!r.done || !l.done) { + if (l.done || (!r.done && r.value[0] < l.value[0])) { + yield { key: r.value![0], value: r.value![1] } + r = await remote.next() + } else if (r.done || l.value[0] < r.value[0]) { + yield { key: l.value[0] } + l = await nextLocal() + } else { + if (!Json.equal(r.value[1], l.value[1])) yield { key: r.value[0], value: r.value[1] } + r = await remote.next() + l = await nextLocal() + } + } + } + + private async applyToReplicas(changes: KVChange[]): Promise { + const writes = changes.filter((c) => c.value !== undefined).map((c): [string, unknown] => [c.key, c.value]) + const removes = changes.filter((c) => c.value === undefined).map((c) => c.key) + await Promise.all(this.replicas.map(async (replica) => (await replica.writeMany(writes), await replica.removeMany(removes)))) + } + + // ------------------------------------------------------------------ local writes → source + + private async writeThrough(entry: OutboxEntry): Promise { + if (this.queued === 0) { + try { + await this.send(entry) + return this.applyToReplicas(KVStorageComposite.changesOf(entry)) + } catch (error) { + if (!(error instanceof StorageUnavailableError)) throw error + } + } + // Offline, or earlier writes are still queued: apply locally and queue for the source. + const seq = ++this.state.outboxSeq + for (const key of KVStorageComposite.keysOf(entry)) this.pending.set(key, (this.pending.get(key) ?? 0) + 1) + await this.persistent.writeMany([[OUTBOX_PREFIX + String(seq).padStart(12, '0'), entry]]) + await this.applyToReplicas(KVStorageComposite.changesOf(entry)) + void this.replay() + } + + /** Sends queued writes to the source in order; stops (and retries later) if it's unreachable. */ + private replay(): Promise { + const run = async () => { + try { + for (;;) { + let next: [string, unknown] | undefined + for await (const entry of this.persistent.scan({ gte: OUTBOX_PREFIX, lt: OUTBOX_END, limit: 1, values: true })) next = entry + if (!next) return + const [key, entry] = next as [string, OutboxEntry] + await this.send(entry) + await this.persistent.removeMany([key]) + for (const k of KVStorageComposite.keysOf(entry)) { + const count = this.pending.get(k)! - 1 + if (count > 0) this.pending.set(k, count) + else this.pending.delete(k) + } + this.state.retryAttempt = 0 + } + } catch (error) { + if (!(error instanceof StorageUnavailableError)) console.error(error) + this.scheduleRetry(() => void this.replay()) + } finally { + this.state.replaying = undefined + } + } + return (this.state.replaying ??= run()) + } + + /** Sends a write to the source, expecting its echo when the source is watched. */ + private async send(entry: OutboxEntry): Promise { + const keys = this.watchable ? KVStorageComposite.keysOf(entry) : [] + for (const key of keys) this.echoes.set(key, (this.echoes.get(key) ?? 0) + 1) + try { + await (entry.type === 'write' ? this.source.writeMany(entry.entries) : this.source.removeMany(entry.keys)) + } catch (error) { + for (const key of keys) this.consumeEcho(key) + throw error + } + } + + /** Whether `key` had an outstanding echo (now consumed). */ + private consumeEcho(key: string): boolean { + const count = this.echoes.get(key) + if (!count) return false + if (count > 1) this.echoes.set(key, count - 1) + else this.echoes.delete(key) + return true + } + + private scheduleRetry(task: () => void): void { + if (this.state.closed) return + clearTimeout(this.state.retryTimer) + const delay = ExponentialBackoff.calculateDelay({ ...this.backoff, growthRate: 2, jitter: 0.2, maxAttempts: 0 }, this.state.retryAttempt++) + this.state.retryTimer = setTimeout(task, delay) + ;(this.state.retryTimer as { unref?: () => void }).unref?.() + } + + private static keysOf(entry: OutboxEntry): string[] { + return entry.type === 'write' ? entry.entries.map(([key]) => key) : entry.keys + } + + private static changesOf(entry: OutboxEntry): KVChange[] { + return entry.type === 'write' ? entry.entries.map(([key, value]) => ({ key, value })) : entry.keys.map((key) => ({ key })) + } +} diff --git a/src/kv-storage/kv-storage-http-server.ts b/src/kv-storage/kv-storage-http-server.ts index 54be60d..208262a 100644 --- a/src/kv-storage/kv-storage-http-server.ts +++ b/src/kv-storage/kv-storage-http-server.ts @@ -1,36 +1,75 @@ -import type { BaseKVStorage, ScanOptions } from './base-kv-storage.js' +import { ExtendedJson } from '../utils/extended-json.js' +import type { BaseKVStorage, KVChange, ScanOptions } from './base-kv-storage.js' + +export interface KVStorageHttpServerOptions { + /** Path prefix to strip, e.g. `/kv`. Default `''`. */ + basePath?: string + /** Recent change batches kept for clients resuming with `Last-Event-ID`. Default 10 000. */ + historySize?: number + /** Interval of SSE keep-alive comments. Default 15 000 ms. */ + keepAliveMs?: number +} + +interface HistoryEntry { + seq: number + changes: KVChange[] +} /** * Serves any KV storage over HTTP for `KVStorageHttp`. Framework-agnostic: `handle(Request)` - * returns a `Response`. All routes are `POST /` with a JSON body; `scan` streams NDJSON. - * The raw primitives use absolute keys, so this exposes the whole store behind `storage`. + * returns a `Response`. + * + * - `POST /` with an Extended JSON body (`scan` streams NDJSON). + * - `GET /events`: Server-Sent Events of every change to the storage. Event ids are + * `-` (epoch = this server instance). A client sending `Last-Event-ID` gets the + * changes it missed if they are still in the history; otherwise it gets a `resync` event. + * A client without one gets a `ready` event carrying the current id. + * + * Only changes made through the wrapped storage instance are seen; several server processes over + * one database each see only their own writes. */ export class KVStorageHttpServer { + private readonly basePath: string + private readonly historySize: number + private readonly keepAliveMs: number + private readonly epoch = Math.random().toString(36).slice(2, 10) + private seq = 0 + private readonly history: HistoryEntry[] = [] + private readonly subscribers = new Set<(frame: string) => void>() + constructor( private readonly storage: BaseKVStorage, - private readonly basePath = '', - ) {} + options: KVStorageHttpServerOptions = {}, + ) { + this.basePath = options.basePath ?? '' + this.historySize = options.historySize ?? 10_000 + this.keepAliveMs = options.keepAliveMs ?? 15_000 + storage.on('change', ({ changes }) => this.publish(changes)) + } async handle(request: Request): Promise { const path = new URL(request.url).pathname - if (request.method !== 'POST' || !path.startsWith(this.basePath)) return KVStorageHttpServer.error(404, 'Not found') + if (!path.startsWith(this.basePath)) return KVStorageHttpServer.error(404, 'Not found') + const route = `${request.method} ${path.slice(this.basePath.length)}` + if (route === 'GET /events') return this.events(request.headers.get('last-event-id')) + if (request.method !== 'POST') return KVStorageHttpServer.error(404, 'Not found') try { - const body = (await request.json()) as { keys: string[]; entries: [string, unknown][] } & ScanOptions - switch (path.slice(this.basePath.length)) { - case '/readMany': { + const body = ExtendedJson.parse(await request.text()) as { keys: string[]; entries: [string, unknown][] } & ScanOptions + switch (route) { + case 'POST /readMany': { const values = await this.storage.readMany(body.keys) const missing = values.flatMap((v, i) => (v === undefined ? [i] : [])) - return Response.json({ values: values.map((v) => v ?? null), missing }) + return KVStorageHttpServer.json({ values: values.map((v) => v ?? null), missing }) } - case '/writeMany': + case 'POST /writeMany': await this.storage.writeMany(body.entries) return new Response(null, { status: 204 }) - case '/removeMany': + case 'POST /removeMany': await this.storage.removeMany(body.keys) return new Response(null, { status: 204 }) - case '/writeIfAbsent': - return Response.json(await this.storage.writeIfAbsent(body.entries)) - case '/scan': + case 'POST /writeIfAbsent': + return KVStorageHttpServer.json(await this.storage.writeIfAbsent(body.entries)) + case 'POST /scan': return this.scan(body) default: return KVStorageHttpServer.error(404, 'Not found') @@ -40,6 +79,54 @@ export class KVStorageHttpServer { } } + private publish(changes: KVChange[]): void { + const entry = { seq: ++this.seq, changes } + this.history.push(entry) + if (this.history.length > this.historySize) this.history.splice(0, this.history.length - this.historySize) + const frame = this.frame('change', entry.seq, changes) + for (const send of this.subscribers) send(frame) + } + + private events(lastEventId: string | null): Response { + const encoder = new TextEncoder() + let send: ((frame: string) => void) | undefined + let keepAlive: ReturnType | undefined + const stream = new ReadableStream({ + start: (controller) => { + send = (frame) => controller.enqueue(encoder.encode(frame)) + send(this.catchUpFrames(lastEventId)) + this.subscribers.add(send) + keepAlive = setInterval(() => send!(': keep-alive\n\n'), this.keepAliveMs) + ;(keepAlive as { unref?: () => void }).unref?.() + }, + cancel: () => { + if (send) this.subscribers.delete(send) + clearInterval(keepAlive) + }, + }) + return new Response(stream, { headers: { 'content-type': 'text/event-stream', 'cache-control': 'no-cache' } }) + } + + /** + * What a (re)connecting client gets before live changes: the changes it missed, or `resync` when + * they can't be replayed; then always `ready` (the client is caught up as of that id). + */ + private catchUpFrames(lastEventId: string | null): string { + const ready = this.frame('ready', this.seq, null) + if (lastEventId === null) return ready + const [epoch, seqText] = lastEventId.split('-') + const seq = Number(seqText) + const oldest = this.history[0]?.seq ?? this.seq + 1 + // Resumable only if it's our epoch and nothing after `seq` has been dropped from the history. + if (epoch !== this.epoch || !Number.isInteger(seq) || seq > this.seq || seq < oldest - 1) return this.frame('resync', this.seq, null) + ready + const missed = this.history.filter((entry) => entry.seq > seq).map((entry) => this.frame('change', entry.seq, entry.changes)) + return missed.join('') + ready + } + + private frame(event: string, seq: number, data: unknown): string { + return `id: ${this.epoch}-${seq}\nevent: ${event}\ndata: ${ExtendedJson.stringify(data)}\n\n` + } + private scan(options: ScanOptions): Response { const iterator = this.storage.scan(options)[Symbol.asyncIterator]() const encoder = new TextEncoder() @@ -53,7 +140,7 @@ export class KVStorageHttpServer { if (chunk) controller.enqueue(encoder.encode(chunk)) return controller.close() } - chunk += JSON.stringify(options.values ? next.value : [next.value[0]]) + '\n' + chunk += ExtendedJson.stringify(options.values ? next.value : [next.value[0]]) + '\n' } controller.enqueue(encoder.encode(chunk)) }, @@ -64,6 +151,10 @@ export class KVStorageHttpServer { return new Response(stream, { headers: { 'content-type': 'application/x-ndjson' } }) } + private static json(body: unknown): Response { + return new Response(ExtendedJson.stringify(body), { headers: { 'content-type': 'application/json' } }) + } + private static error(status: number, message: string): Response { return Response.json({ error: message }, { status }) } diff --git a/src/kv-storage/kv-storage-http.ts b/src/kv-storage/kv-storage-http.ts index ced0b4e..4e43026 100644 --- a/src/kv-storage/kv-storage-http.ts +++ b/src/kv-storage/kv-storage-http.ts @@ -1,24 +1,48 @@ +import { StorageUnavailableError } from '../utils/errors.js' +import { ExtendedJson } from '../utils/extended-json.js' import type { AnySpec, Spec } from '../utils/schema.js' -import { BaseKVStorage, type ScanOptions } from './base-kv-storage.js' +import { SSESession } from '../utils/sse-session/sse-session.js' +import { BaseKVStorage, type KVChange, type ScanOptions } from './base-kv-storage.js' export interface KVStorageHttpOptions { /** Base URL of a `KVStorageHttpServer`, e.g. `https://host/kv`. */ url: string headers?: Record - fetch?: typeof fetch + fetch?: (url: string, init: RequestInit) => Promise + /** Called when the change stream fails (it keeps reconnecting). Default: ignore. */ + onWatchError?: (error: Error) => void } -/** KV storage client for a `KVStorageHttpServer`. Same API as any other KV storage. */ +/** + * KV storage client for a `KVStorageHttpServer`. Same API as any other KV storage. + * + * `watch()` subscribes to the server's change stream (SSE): changes made by anyone are emitted + * here as `change` events with `origin: 'remote'`, and `resync` is emitted when the server can't + * resume from `lastEventId`. Values travel as Extended JSON (`bigint` and `Uint8Array` survive). + */ export class KVStorageHttp extends BaseKVStorage { private readonly url: string private readonly headers: Record - private readonly fetcher: typeof fetch + private readonly fetcher: (url: string, init: RequestInit) => Promise + private readonly onWatchError: (error: Error) => void + /** Mutable state lives here: child nodes share this object (see `child()`). */ + private readonly state: { session?: SSESession; lastEventId?: string } = {} constructor(options: KVStorageHttpOptions) { super() this.url = options.url.replace(/\/$/, '') this.headers = options.headers ?? {} - this.fetcher = options.fetch ?? globalThis.fetch.bind(globalThis) + this.fetcher = options.fetch ?? ((url, init) => globalThis.fetch(url, init)) + this.onWatchError = options.onWatchError ?? (() => {}) + } + + /** Position in the server's change stream; sent as `Last-Event-ID` when (re)connecting. */ + get lastEventId(): string | undefined { + return this.state.lastEventId + } + + set lastEventId(id: string | undefined) { + this.state.lastEventId = id } async readMany(keys: string[]): Promise { @@ -28,18 +52,6 @@ export class KVStorageHttp extends BaseKVStorage { return values } - async writeMany(entries: [string, unknown][]): Promise { - if (entries.length) await this.post('writeMany', { entries }) - } - - async removeMany(keys: string[]): Promise { - if (keys.length) await this.post('removeMany', { keys }) - } - - async writeIfAbsent(entries: [string, unknown][]): Promise { - return entries.length ? ((await this.post('writeIfAbsent', { entries })) as boolean[]) : [] - } - async *scan(options: ScanOptions = {}): AsyncGenerator<[string, unknown]> { if (options.limit !== undefined && options.limit <= 0) return const response = await this.request('scan', options) @@ -53,8 +65,8 @@ export class KVStorageHttp extends BaseKVStorage { const lines = buffer.split('\n') buffer = lines.pop()! for (const line of lines) { - const [key, value] = JSON.parse(line) as [string, unknown?] - yield [key, value] + const [key, entryValue] = ExtendedJson.parse(line) as [string, unknown?] + yield [key, entryValue] } } } finally { @@ -62,17 +74,71 @@ export class KVStorageHttp extends BaseKVStorage { } } + /** + * Subscribes to the server's change stream. Resolves once connected and caught up (missed + * changes replayed, or `resync` emitted); keeps reconnecting and resuming from `lastEventId` + * until `unwatch()`. While the server is unreachable this stays pending, so don't await it on a + * path that must work offline. + */ + async watch(): Promise { + if (this.state.session) return + const session = new SSESession(`${this.url}/events`, { + fetch: this.fetcher, + headers: this.headers, + persistent: true, + onError: this.onWatchError, + onRequest: async (request) => + this.lastEventId ? { ...request, headers: { ...(request.headers as Record), 'Last-Event-ID': this.lastEventId } } : request, + }) + session.on('message', (message) => { + if (message.id) this.lastEventId = message.id + if (message.event === 'change') { + this.emit('change', { changes: ExtendedJson.parse(message.data) as KVChange[], origin: 'remote', id: message.id }) + } else if (message.event === 'resync') { + this.emit('resync', { id: message.id }) + } + }) + this.state.session = session + const caughtUp = session.waitFor('message', (message) => message.event === 'ready') + await session.connect() + await caughtUp + } + + async unwatch(): Promise { + const session = this.state.session + this.state.session = undefined + await session?.disconnect() + } + + protected async applyWrite(entries: [string, unknown][]): Promise { + await this.post('writeMany', { entries }) + } + + protected async applyRemove(keys: string[]): Promise { + await this.post('removeMany', { keys }) + } + + protected async applyWriteIfAbsent(entries: [string, unknown][]): Promise { + return (await this.post('writeIfAbsent', { entries })) as boolean[] + } + private async post(path: string, body: unknown): Promise { const response = await this.request(path, body) - return response.status === 204 ? undefined : response.json() + return response.status === 204 ? undefined : ExtendedJson.parse(await response.text()) } private async request(path: string, body: unknown): Promise { - const response = await this.fetcher(`${this.url}/${path}`, { - method: 'POST', - headers: { ...this.headers, 'content-type': 'application/json' }, - body: JSON.stringify(body), - }) + let response: Response + try { + response = await this.fetcher(`${this.url}/${path}`, { + method: 'POST', + headers: { ...this.headers, 'content-type': 'application/json' }, + body: ExtendedJson.stringify(body), + }) + } catch (error) { + throw new StorageUnavailableError(`Cannot reach ${this.url}`, { cause: error }) + } + if ([502, 503, 504].includes(response.status)) throw new StorageUnavailableError(`HTTP ${response.status} from ${this.url}`) if (!response.ok) throw new Error(`HTTP ${response.status}: ${await response.text()}`) return response } diff --git a/src/kv-storage/kv-storage-in-memory.ts b/src/kv-storage/kv-storage-in-memory.ts index 6883e1a..3ac5f6e 100644 --- a/src/kv-storage/kv-storage-in-memory.ts +++ b/src/kv-storage/kv-storage-in-memory.ts @@ -11,15 +11,15 @@ export class KVStorageInMemory extends BaseKVStorage Json.clone(this.data.get(k))) } - async writeMany(entries: [string, unknown][]): Promise { + protected async applyWrite(entries: [string, unknown][]): Promise { for (const [k, v] of entries) this.data.set(k, Json.clone(v)) } - async removeMany(keys: string[]): Promise { + protected async applyRemove(keys: string[]): Promise { for (const k of keys) this.data.delete(k) } - async writeIfAbsent(entries: [string, unknown][]): Promise { + protected async applyWriteIfAbsent(entries: [string, unknown][]): Promise { return entries.map(([k, v]) => !this.data.has(k) && (this.data.set(k, Json.clone(v)), true)) } diff --git a/src/kv-storage/kv-storage-indexeddb.ts b/src/kv-storage/kv-storage-indexeddb.ts new file mode 100644 index 0000000..4c0e7c7 --- /dev/null +++ b/src/kv-storage/kv-storage-indexeddb.ts @@ -0,0 +1,129 @@ +import type { AnySpec, Spec } from '../utils/schema.js' +import { BaseKVStorage, type ScanOptions } from './base-kv-storage.js' + +export interface KVStorageIndexedDBOptions { + /** Database name. */ + name: string + /** Object store name. Default `kv`. */ + storeName?: string + /** IndexedDB factory. Default `globalThis.indexedDB`. */ + indexedDB?: IDBFactory + /** Entries read per transaction while scanning. Default 256. */ + scanBatchSize?: number +} + +/** + * KV storage in the browser's IndexedDB (one object store, string keys). Values are stored with the + * structured clone algorithm, so `bigint` and `Uint8Array` survive. IndexedDB orders string keys by + * UTF-16 code unit, like JavaScript string comparison. + */ +export class KVStorageIndexedDB extends BaseKVStorage { + private readonly storeName: string + private readonly scanBatchSize: number + private readonly db: Promise + + constructor(options: KVStorageIndexedDBOptions) { + super() + this.storeName = options.storeName ?? 'kv' + this.scanBatchSize = options.scanBatchSize ?? 256 + const factory = options.indexedDB ?? globalThis.indexedDB + const open = factory.open(options.name, 1) + open.onupgradeneeded = () => open.result.createObjectStore(this.storeName) + this.db = KVStorageIndexedDB.done(open) + } + + async readMany(keys: string[]): Promise { + const store = await this.store('readonly') + return Promise.all(keys.map((key) => KVStorageIndexedDB.done(store.get(key)))) + } + + async *scan(options: ScanOptions = {}): AsyncGenerator<[string, unknown]> { + let remaining = options.limit ?? Infinity + let last: string | undefined + while (remaining > 0) { + const batch = await this.readBatch(options, last, Math.min(remaining, this.scanBatchSize)) + for (const entry of batch) { + yield entry + remaining-- + } + if (batch.length < this.scanBatchSize) return + last = batch.at(-1)![0] + } + } + + /** Closes the database connection. */ + async close(): Promise { + ;(await this.db).close() + } + + protected async applyWrite(entries: [string, unknown][]): Promise { + const store = await this.store('readwrite') + for (const [key, value] of entries) store.put(value, key) + await KVStorageIndexedDB.committed(store.transaction) + } + + protected async applyRemove(keys: string[]): Promise { + const store = await this.store('readwrite') + for (const key of keys) store.delete(key) + await KVStorageIndexedDB.committed(store.transaction) + } + + protected async applyWriteIfAbsent(entries: [string, unknown][]): Promise { + // One readwrite transaction: checks and writes can't interleave with other writers. + const store = await this.store('readwrite') + const written = await Promise.all( + entries.map(async ([key, value]) => { + if ((await KVStorageIndexedDB.done(store.getKey(key))) !== undefined) return false + store.add(value, key) + return true + }), + ) + await KVStorageIndexedDB.committed(store.transaction) + return written + } + + /** Up to `count` entries after `last` (exclusive), in scan order, within one transaction. */ + private async readBatch(options: ScanOptions, last: string | undefined, count: number): Promise<[string, unknown][]> { + const reverse = options.reverse ?? false + // The upper bound is always exclusive (`lt`, or the last key seen in reverse); the lower bound + // is exclusive only when resuming forwards after `last`. + const lower = reverse ? options.gte : (last ?? options.gte) + const upper = reverse ? (last ?? options.lt) : options.lt + const lowerOpen = !reverse && last !== undefined + if (lower !== undefined && upper !== undefined && lower >= upper) return [] + let range: IDBKeyRange | undefined + if (lower !== undefined && upper !== undefined) range = IDBKeyRange.bound(lower, upper, lowerOpen, true) + else if (lower !== undefined) range = IDBKeyRange.lowerBound(lower, lowerOpen) + else if (upper !== undefined) range = IDBKeyRange.upperBound(upper, true) + const store = await this.store('readonly') + const out: [string, unknown][] = [] + const request = options.values ? store.openCursor(range, reverse ? 'prev' : 'next') : store.openKeyCursor(range, reverse ? 'prev' : 'next') + return new Promise((resolve, reject) => { + request.onerror = () => reject(request.error) + request.onsuccess = () => { + const cursor = request.result + if (!cursor || out.length >= count) return resolve(out) + out.push([cursor.key as string, options.values ? (cursor as IDBCursorWithValue).value : undefined]) + cursor.continue() + } + }) + } + + private async store(mode: IDBTransactionMode): Promise { + return (await this.db).transaction(this.storeName, mode).objectStore(this.storeName) + } + + private static done(request: IDBRequest): Promise { + return new Promise((resolve, reject) => { + request.onsuccess = () => resolve(request.result) + request.onerror = () => reject(request.error) + }) + } + + private static committed(transaction: IDBTransaction): Promise { + return new Promise((resolve, reject) => { + transaction.oncomplete = () => resolve() + transaction.onerror = transaction.onabort = () => reject(transaction.error) + }) + } +} diff --git a/src/utils/errors.ts b/src/utils/errors.ts index 8b95d83..c92957d 100644 --- a/src/utils/errors.ts +++ b/src/utils/errors.ts @@ -17,3 +17,11 @@ export class InvalidKeyError extends TypeError { this.name = 'InvalidKeyError' } } + +/** A remote storage could not be reached (network failure or 502/503/504). Writes may be retried later. */ +export class StorageUnavailableError extends Error { + constructor(message: string, options?: { cause?: unknown }) { + super(message, options) + this.name = 'StorageUnavailableError' + } +} diff --git a/src/utils/event-emitter.ts b/src/utils/event-emitter.ts new file mode 100644 index 0000000..6ef1090 --- /dev/null +++ b/src/utils/event-emitter.ts @@ -0,0 +1,134 @@ +// Adapted from xo-cash-utils `source/event-emitter.ts` (branch installable-sse): payloads are passed +// through as-is (no cloning, freezing or DeeplyReadonly), and timeouts reject with a plain Error. + +export type EventMap = Record + +type Listener = (detail: T) => void + +/** Lets listeners for individual payloads live in one collection typed with the union of payloads. */ +type StoredListener = { + bivarianceHack(detail: T): void +}['bivarianceHack'] + +interface ListenerEntry { + listener: StoredListener + wrappedListener: StoredListener + cancel: () => void +} + +/** Returned by {@link EventEmitter.on} and {@link EventEmitter.once} to remove the listener. */ +export type OffCallback = () => void + +/** A small typed event emitter. */ +export class EventEmitter { + private readonly listeners: Map>> = new Map() + + /** + * Adds a listener. + * @param debounceMilliseconds - When > 0, only the last event within this window is delivered. + */ + on(type: K, listener: Listener, debounceMilliseconds: number = 0): OffCallback { + const { cancel, listener: cancellableListener } = this.cancellable(listener) + const wrappedListener = debounceMilliseconds > 0 ? this.debounce(cancellableListener, debounceMilliseconds) : cancellableListener + this.entriesFor(type).add({ listener, wrappedListener, cancel } as ListenerEntry) + return () => this.off(type, listener) + } + + /** Adds a listener that is removed after its first call. */ + once(type: K, listener: Listener, debounceMilliseconds: number = 0): OffCallback { + const wrappedListener: Listener = (detail) => { + this.off(type, listener) + listener(detail) + } + const { cancel, listener: cancellableListener } = this.cancellable(wrappedListener) + const debouncedListener = debounceMilliseconds > 0 ? this.debounce(cancellableListener, debounceMilliseconds) : cancellableListener + this.entriesFor(type).add({ listener, wrappedListener: debouncedListener, cancel } as ListenerEntry) + return () => this.off(type, listener) + } + + /** Removes one listener, or all listeners for `type` when none is given. */ + off(type: K, listener?: Listener): void { + const listeners = this.listeners.get(type) + if (!listeners) return + for (const entry of Array.from(listeners)) { + if (listener && entry.listener !== listener && entry.wrappedListener !== listener) continue + // Cancel so a pending debounced call doesn't fire after removal. + entry.cancel() + listeners.delete(entry) + } + if (!listener || listeners.size === 0) this.listeners.delete(type) + } + + /** + * Calls every listener for `type` with `payload`. Listener errors are logged, not thrown. + * @returns Whether any listener was registered. + */ + emit(type: K, payload: T[K]): boolean { + const listeners = this.listeners.get(type) + if (!listeners) return false + for (const entry of Array.from(listeners)) { + try { + entry.wrappedListener(payload) + } catch (error) { + console.error(error) + } + } + return listeners.size > 0 + } + + removeAllListeners(): void { + for (const type of Array.from(this.listeners.keys())) this.off(type) + } + + /** Resolves with the first payload of `type` matching `predicate`; rejects after `timeoutMs` if given. */ + waitFor(type: K, predicate: (payload: T[K]) => boolean, timeoutMs?: number): Promise { + return new Promise((resolve, reject) => { + let timeoutId: ReturnType | undefined + const cleanup = (): void => { + this.off(type, listener) + if (timeoutId !== undefined) clearTimeout(timeoutId) + } + const listener = (payload: T[K]): void => { + try { + if (!predicate(payload)) return + cleanup() + resolve(payload) + } catch (error) { + cleanup() + reject(error) + } + } + if (timeoutMs !== undefined) { + timeoutId = setTimeout(() => { + this.off(type, listener) + reject(new Error(`Timed out waiting for "${String(type)}"`)) + }, timeoutMs) + } + this.on(type, listener) + }) + } + + private entriesFor(type: K): Set> { + let entries = this.listeners.get(type) + if (!entries) this.listeners.set(type, (entries = new Set())) + return entries + } + + private debounce(func: Listener, wait: number): Listener { + let timeout: ReturnType | undefined + return (detail) => { + if (timeout !== undefined) clearTimeout(timeout) + timeout = setTimeout(() => func(detail), wait) + } + } + + private cancellable(func: Listener): { cancel: () => void; listener: Listener } { + let cancelled = false + return { + cancel: () => void (cancelled = true), + listener: (detail) => { + if (!cancelled) func(detail) + }, + } + } +} diff --git a/src/utils/extended-json.ts b/src/utils/extended-json.ts new file mode 100644 index 0000000..13888e3 --- /dev/null +++ b/src/utils/extended-json.ts @@ -0,0 +1,51 @@ +// Adapted from xo-cash-utils `source/extended-json.ts` (branch installable-sse), with local hex +// encoding instead of @bitauth/libauth. + +/** Matches a bigint encoded as ``. */ +const BIGINT_PATTERN = /^[+-]?[0-9]+)n>$/u + +/** Matches a Uint8Array encoded as ``. */ +const UINT8ARRAY_PATTERN = /^[0-9a-f]*)>$/u + +/** + * JSON that also round-trips `bigint` (``) and `Uint8Array` (``). + * Used for every value that crosses the network. + */ +export class ExtendedJson { + static stringify(value: unknown): string { + return JSON.stringify(value, ExtendedJson.replacer) + } + + static parse(text: string): unknown { + return JSON.parse(text, ExtendedJson.reviver) + } + + /** `JSON.stringify` replacer. */ + static replacer(this: void, _key: string, value: unknown): unknown { + if (value instanceof Uint8Array) return `` + if (typeof value === 'bigint') return `` + return value + } + + /** `JSON.parse` reviver. */ + static reviver(this: void, _key: string, value: unknown): unknown { + if (typeof value !== 'string') return value + const bigint = value.match(BIGINT_PATTERN) + if (bigint) return BigInt(bigint.groups!.bigint!) + const bytes = value.match(UINT8ARRAY_PATTERN) + if (bytes) return ExtendedJson.fromHex(bytes.groups!.hex!) + return value + } + + private static toHex(bytes: Uint8Array): string { + let hex = '' + for (const byte of bytes) hex += byte.toString(16).padStart(2, '0') + return hex + } + + private static fromHex(hex: string): Uint8Array { + const bytes = new Uint8Array(hex.length / 2) + for (let i = 0; i < bytes.length; i++) bytes[i] = parseInt(hex.slice(i * 2, i * 2 + 2), 16) + return bytes + } +} diff --git a/src/utils/json.ts b/src/utils/json.ts index 8c666b7..84d13eb 100644 --- a/src/utils/json.ts +++ b/src/utils/json.ts @@ -1,8 +1,11 @@ -/** Helpers for JSON-compatible values (all stored values are JSON by contract). */ +/** + * Helpers for stored values: JSON plus `bigint` and `Uint8Array` (see `ExtendedJson`). + */ export class Json { - /** Deep copy; much cheaper than `structuredClone` for plain JSON. */ + /** Deep copy; much cheaper than `structuredClone` for plain data. */ static clone(value: T): T { if (value === null || typeof value !== 'object') return value + if (value instanceof Uint8Array) return value.slice() as T if (Array.isArray(value)) return value.map((v) => Json.clone(v)) as T const out: Record = {} for (const k in value) out[k] = Json.clone((value as Record)[k]) @@ -12,6 +15,9 @@ export class Json { static equal(a: unknown, b: unknown): boolean { if (a === b) return true if (typeof a !== 'object' || typeof b !== 'object' || a === null || b === null) return false + if (a instanceof Uint8Array || b instanceof Uint8Array) { + return a instanceof Uint8Array && b instanceof Uint8Array && a.length === b.length && a.every((byte, i) => byte === b[i]) + } if (Array.isArray(a) !== Array.isArray(b)) return false const ka = Object.keys(a) const kb = Object.keys(b) @@ -19,6 +25,6 @@ export class Json { } static isObject(value: unknown): value is Record { - return typeof value === 'object' && value !== null && !Array.isArray(value) + return typeof value === 'object' && value !== null && !Array.isArray(value) && !(value instanceof Uint8Array) } } diff --git a/src/utils/sse-session/async-push-iterator.ts b/src/utils/sse-session/async-push-iterator.ts new file mode 100644 index 0000000..462285f --- /dev/null +++ b/src/utils/sse-session/async-push-iterator.ts @@ -0,0 +1,85 @@ +// Vendored from https://git.harvmaster.com/Harvmaster/sse-session (commit c08267e85c). +// Local changes: import paths, uses ../event-emitter.ts, tryAsync folded into SSESession. + +/** + * An async iterable queue that bridges push-based producers and pull-based consumers. + * + * Composes an internal {@link ReadableStream} instead of extending it, so producers + * call {@link push} while consumers use standard async iteration (`for await...of`). + * + * ```ts + * const messages = new AsyncPushIterator(); + * + * // Producer (elsewhere) + * messages.push(event); + * + * // Consumer + * for await (const event of messages) { + * handle(event); + * } + * ``` + * + * {@link Symbol.asyncIterator} returns `stream.values({ preventCancel: true })` so + * breaking out of `for await...of` does not cancel the underlying stream. That + * matters for long-lived sessions where the producer keeps pushing after a consumer + * stops reading early (for example, test helpers that only collect a fixed count). + */ +export class AsyncPushIterator { + /** ReadableStream backing the async iterator returned from {@link Symbol.asyncIterator}. */ + private readonly stream: ReadableStream + + /** Controller used to enqueue values and close the stream from {@link push} and {@link close}. */ + private controller: ReadableStreamDefaultController | undefined + + /** When true, no more values are accepted and iteration eventually completes. */ + public closed = false + + public constructor() { + // `start`'s `this` is the underlying source object when using a plain method. + // An arrow function captures the class instance so the controller is stored here. + this.stream = new ReadableStream({ + start: (controller: ReadableStreamDefaultController): void => { + this.controller = controller + }, + }) + } + + /** + * Enqueues a value for the consumer. + * + * After {@link close}, pushes are silently dropped. + * + * @param value - The next value to yield from the iterator. + */ + push(value: T): void { + if (this.closed) return + + this.controller?.enqueue(value) + } + + /** + * Ends the stream. + * + * Marks the iterator closed so future {@link push} calls are ignored. + * Buffered values are still yielded before iteration completes. + */ + close(): void { + this.closed = true + + try { + this.controller?.close() + } catch { + // The reader may already have released or cancelled the stream. + } + } + + /** + * Returns an async iterator over the composed stream. + * + * Uses `preventCancel: true` so early `break` from `for await...of` does not + * close the stream and block later pushes. + */ + [Symbol.asyncIterator](): AsyncIterableIterator { + return this.stream.values({ preventCancel: true }) + } +} diff --git a/src/utils/sse-session/exponential-backoff.ts b/src/utils/sse-session/exponential-backoff.ts new file mode 100644 index 0000000..ca95335 --- /dev/null +++ b/src/utils/sse-session/exponential-backoff.ts @@ -0,0 +1,251 @@ +// Vendored from https://git.harvmaster.com/Harvmaster/sse-session (commit c08267e85c). +// Local changes: import paths, uses ../event-emitter.ts, tryAsync folded into SSESession. + +/** + * Exponential backoff is a technique used to retry a function after a delay. + * + * The delay increases exponentially with each attempt, up to a maximum delay. + * + * The jitter is a random amount of time added to the delay to prevent thundering herd problems. + * + * The growth rate is the factor by which the delay increases with each attempt. + */ +export class ExponentialBackoff { + /** + * Create a new ExponentialBackoff instance + * + * @param config - The configuration for the exponential backoff + * @returns The ExponentialBackoff instance + */ + static from(config?: Partial): ExponentialBackoff { + const backoff = new ExponentialBackoff(config) + + return backoff + } + + /** + * Run the function with exponential backoff + * + * @param fn - The function to run + * @param onError - The callback to call when an error occurs + * @param options - The configuration for the exponential backoff + * + * @throws An AggregateError with all the errors that were thrown by the task function + * + * @returns The result of the function + */ + static run( + fn: (context: ExpontialBackoffCallbackContext) => Promise, + onError = (_error: Error, _context: ExpontialBackoffCallbackContext): void => {}, + options?: Partial, + ): Promise { + const backoff = ExponentialBackoff.from(options) + + return backoff.run(fn, onError) + } + + private readonly options: ExponentialBackoffOptions + + /** + * Creates a new exponential-backoff instance. + * + * Unspecified options use the defaults listed below. + * + * @param options - Exponential-backoff configuration overrides. + * @param options.maxDelay - Maximum delay between retries. Default: `10_000` ms. + * @param options.maxAttempts - Maximum number of attempts; `0` retries indefinitely. Default: `10`. + * @param options.baseDelay - Delay used as the basis for the first retry. Default: `1_000` ms. + * @param options.growthRate - Multiplier applied to the delay after each attempt. Default: `2`. + * @param options.jitter - Random proportional variation applied to each delay. Default: `0.1`. + */ + constructor(options: Partial = {}) { + this.options = { + maxDelay: 10_000, + maxAttempts: 10, + baseDelay: 1_000, + growthRate: 2, + jitter: 0.1, + ...options, + } + + ExponentialBackoff.validateOptions(this.options) + } + + /** + * Run the function with exponential backoff + * + * If the function fails but we have not hit the max attempts, the error will be passed to the onError callback + * and the function will be retried with an exponential delay + * + * If the function fails and we have hit the max attempts, an AggregateError will be thrown with all the errors that were thrown by the task function + * + * @param fn - The function to run + * @param onError - The callback to call when an error occurs + * + * @throws An AggregateError with all the errors that were thrown by the task function + * + * @returns The result of the function + */ + async run( + fn: (context: ExpontialBackoffCallbackContext) => Promise, + onError = (_error: Error, _context: ExpontialBackoffCallbackContext): void => {}, + ): Promise { + const abortController = new AbortController() + + const context: ExpontialBackoffCallbackContext = { + errors: [], + attempt: 1, + options: { ...this.options }, + nextDelay: 0, + signal: abortController.signal, + abort: (reason?: unknown) => abortController.abort(reason), + } + + while (context.options.maxAttempts === 0 || context.attempt <= context.options.maxAttempts) { + context.nextDelay = ExponentialBackoff.calculateDelay(context.options, context.attempt - 1) + + try { + // Await the promise before returning so its execution context remains in the try-catch + // If we didn't await, this `run` function would successfully return and any errors would not be caught here. + return await fn(context) + } catch (error) { + // Store the error in case we fail every attempt + const errorInstance = error instanceof Error ? error : new Error(`${error}`) + context.errors.push(errorInstance) + await onError(errorInstance, context) + } + + // If the abort controller is aborted, throw an error + if (abortController.signal.aborted) { + const reasonError = abortController.signal.reason instanceof Error ? abortController.signal.reason : new Error(`${abortController.signal.reason}`) + throw new Error('Exponential backoff aborted', { cause: reasonError }) + } + + // If we have hit the max attempts, break out of the loop + const unlimitedAttempts = context.options.maxAttempts === 0 + if (!unlimitedAttempts && context.attempt >= context.options.maxAttempts) { + break + } + + // Wait before going to the next attempt + await this.wait(context.nextDelay, abortController.signal) + + context.attempt++ + } + + // We completed the loop without ever succeeding. Throw an ExponentialBackoffMaxRetriesHitError with all the errors we got + throw new AggregateError(context.errors, 'Exponential backoff max retries hit') + } + + private wait(delay: number, signal: AbortSignal): Promise { + return new Promise((resolve, reject) => { + // Set a timeout to resolve the promise + const timeout = setTimeout(() => { + resolve() + }, delay) + + // If the signal is aborted, reject the promise + signal.addEventListener('abort', () => { + clearTimeout(timeout) + reject(new Error('Exponential backoff aborted')) + }) + }) + } + + /** + * Calculate the delay before we should attempt to retry + * + * NOTE: The maximum delay is (maxDelay * (1 + jitter)) + * + * @param attempt + * @returns The time in milliseconds before another attempt should be made + */ + public static calculateDelay(options: ExponentialBackoffOptions, attempt: number): number { + // Get the power of the growth rate + const power = options.growthRate ** attempt + + // Get the delay before jitter or limit + const rawDelay = options.baseDelay * power + + // Cap the delay to the maximum. Do this before the jitter so jitter does not become larger than delay + const cappedDelay = Math.min(rawDelay, options.maxDelay) + + // Get a random number for the amount to "jitter" the delay by + const jitterAmount = Math.random() + + // Calculate the jitter + const jitter = jitterAmount * options.jitter * cappedDelay + + // Subtract the jitter from the delay + return cappedDelay - jitter + } + + /** + * Validate the options, ensure that no values are NaN + * @param options + */ + public static validateOptions(options: ExponentialBackoffOptions): void { + const keys = Object.keys(options) as (keyof ExponentialBackoffOptions)[] + + for (const key of keys) { + if (!isFinite(options[key])) { + throw new Error(`Invalid option: ${key} is not finite`) + } + + if (key === 'jitter' && (options[key] < 0 || options[key] > 1)) { + throw new Error(`Invalid option: jitter is not between 0 and 1`) + } + + if (options[key] < 0) { + throw new Error(`Invalid option: ${key} is less than 0`) + } + } + } +} + +export type ExponentialBackoffOptions = { + /** + * The maximum delay between attempts in milliseconds + */ + maxDelay: number + + /** + * The maximum number of attempts. Passing 0 will result in infinite attempts. + */ + maxAttempts: number + + /** + * The base delay between attempts in milliseconds + */ + baseDelay: number + + /** + * The growth rate of the delay + */ + growthRate: number + + /** + * The jitter of the delay as a percentage of growthRate + */ + jitter: number +} + +export type ExpontialBackoffCallbackContext = { + /** The errors that have occurred */ + errors: Error[] + + /** The current attempt */ + attempt: number + + /** The options for the exponential backoff */ + options: ExponentialBackoffOptions + + /** The next delay before the next attempt */ + nextDelay: number + + /** The abort signal */ + signal: AbortSignal + + /** The abort function */ + abort: (reason?: unknown) => void +} diff --git a/src/utils/sse-session/index.ts b/src/utils/sse-session/index.ts new file mode 100644 index 0000000..02b167c --- /dev/null +++ b/src/utils/sse-session/index.ts @@ -0,0 +1,5 @@ +export * from './types.js' +export * from './sse-session.js' +export * from './sse-event-parser.js' +export * from './async-push-iterator.js' +export * from './exponential-backoff.js' diff --git a/src/utils/sse-session/sse-event-parser.ts b/src/utils/sse-session/sse-event-parser.ts new file mode 100644 index 0000000..a445348 --- /dev/null +++ b/src/utils/sse-session/sse-event-parser.ts @@ -0,0 +1,214 @@ +// Vendored from https://git.harvmaster.com/Harvmaster/sse-session (commit c08267e85c). +// Local changes: import paths, uses ../event-emitter.ts, tryAsync folded into SSESession. + +import type { SSEvent } from './types.js' + +/** + * Optional encoders used when decoding incoming SSE bytes and re-encoding + * any buffered remainder between chunks. + */ +export interface SSEEventParserOptions { + /** Decodes raw stream bytes into text. Defaults to a new `TextDecoder`. */ + textDecoder: TextDecoder +} + +/** + * Regex that splits decoded SSE text into lines. + * + * The SSE wire format is line-oriented (`field: value` per line). Servers may + * send `\r\n` (HTTP default), `\n` (Unix), or `\r` (legacy Mac). Matching all + * three keeps parsing correct regardless of platform or server implementation. + */ +export const SSE_LINE_ENDINGS = /\r\n|\r|\n/ + +/** + * Regex that matches the single optional leading space in an SSE field value. + * + * Per the SSE spec, `field: value` may include one space immediately after the + * colon; that space is not part of the value. Used with `.replace()` to strip + * it when parsing lines such as `data: hello` → `hello`. + */ +export const SSE_FIELD_VALUE_REGEX = /^ / + +/** + * The newline character used when normalizing SSE text internally. + * + * Used to join consecutive `data:` lines into one payload and to reassemble + * buffered partial lines between streamed chunks before the next parse call. + */ +export const NEW_LINE = '\n' + +/** + * Incrementally parses Server-Sent Events (SSE) from streamed byte chunks. + * + * SSE payloads are line-oriented: each event is a sequence of `field: value` + * lines terminated by a blank line. This parser accepts arbitrary chunk + * boundaries from a live HTTP response body and emits only complete events. + * + * Typical usage is one parser instance per connection, calling {@link parseEvents} + * for each chunk received from the stream: + * + * ```ts + * const parser = new SSEEventParser(); + * + * for await (const chunk of response.body) { + * for (const event of parser.parseEvents(chunk)) { + * // handle event.data, event.event, event.id, event.retry + * } + * } + * ``` + * + * Supported fields follow the SSE spec: `data`, `event`, `id`, and `retry`. + * Multiple `data:` lines in one event are joined with `\n`. An event is only + * emitted once a blank line is seen and at least one `data` field was collected. + */ +export class SSEEventParser { + private readonly textDecoder: TextDecoder + + /** Bytes from a partial line or incomplete event, carried over to the next chunk. */ + private messageBuffer: string = '' + + /** + * Creates a parser for one SSE stream. + * + * Inject custom encoders in tests or when a non-default character encoding + * is required; production callers can rely on the defaults. + * + * @param options - Optional text encoders for decode/encode of stream bytes. + */ + constructor(options: Partial = {}) { + this.textDecoder = options.textDecoder ?? new TextDecoder() + } + + /** + * Clears any buffered bytes from a partial line or incomplete event. + * + * Call when abandoning a transport so the next connection does not prepend + * stale bytes to incoming chunks. + */ + public reset(): void { + // Clear the message buffer + this.messageBuffer = '' + + // Reset the decoder to clear any buffered bytes + this.textDecoder.decode() + } + + /** + * Parses all complete SSE events contained in a newly received chunk. + * + * The chunk is appended to any bytes buffered from earlier calls. Complete + * events (blank-line delimited blocks with at least one `data` field) are + * returned immediately; any trailing partial line or in-progress event stays + * in the internal buffer until a later chunk completes it. + * + * @param chunk - Newly received SSE stream bytes. + * @returns Zero or more complete parsed SSE events from this chunk. + */ + public parseEvents(chunk: Uint8Array): SSEvent[] { + const lines = this.getBufferedLines(chunk) + + // The final split item is the incomplete remainder after the final line + // ending. It must not be processed as a complete SSE line. + const completeLines = lines.slice(0, -1) + + const events: SSEvent[] = [] + let event: Partial = {} + let processedLineCount = 0 + + for (const [index, line] of completeLines.entries()) { + // A blank line indicates the end of an event. If we have received data, we can complete the event + if (line === '') { + if (event.data !== undefined) { + // Cast event when pushing to an SSE Event so it changes from Partial to SSEvent + // The "Partial" can safely be removed as we have guaranteed that `data` is defined + events.push(event as SSEvent) + } + + event = {} + processedLineCount = index + 1 + + continue + } + + this.parseLine(line, event) + } + + this.storeRemainingLines(lines, processedLineCount) + + return events + } + + /** + * Appends a new chunk to the buffered bytes and splits the combined payload + * into lines. + * + * Accepts `\r\n`, `\r`, and `\n` line endings so events parse correctly + * regardless of server or platform conventions. + */ + private getBufferedLines(chunk: Uint8Array): string[] { + this.messageBuffer += this.textDecoder.decode(chunk, { stream: true }) + + return this.messageBuffer.split(SSE_LINE_ENDINGS) + } + + /** + * Parses one SSE field line into an in-progress event. + * + * Lines starting with a colon are ignored as comments. A single optional space after the colon + * is stripped from the field value, per the SSE spec. + */ + private parseLine(line: string, event: Partial): void { + // Split the line at the first colon. Before the first colon is the field, after the first colon is the value. + const [field, ...valueArray] = line.split(':') + + // Join the rest of the array back together to get the value as it was before the split. + // If the value has a space at the start, remove it (according to the SSE spec) + const value = valueArray.join(':').replace(SSE_FIELD_VALUE_REGEX, '') + + switch (field) { + case 'data': + event.data = event.data !== undefined ? `${event.data}${NEW_LINE}${value}` : value + + return + + case 'event': + event.event = value + + return + + case 'id': + event.id = value + + return + + case 'retry': + this.parseRetry(value, event) + + return + } + } + + /** + * Applies a numeric `retry:` field to an in-progress event. + * + * Non-numeric values are ignored rather than failing the parse. + */ + private parseRetry(value: string, event: Partial): void { + const retry = parseInt(value, 10) + + if (!isNaN(retry)) { + event.retry = retry + } + } + + /** + * Preserves incomplete trailing lines for the next received chunk. + * + * Only lines that were fully processed (through a completed event boundary) + * are discarded; the remainder is re-encoded into {@link messageBuffer}. + */ + private storeRemainingLines(lines: string[], processedLineCount: number): void { + this.messageBuffer = lines.slice(processedLineCount).join(NEW_LINE) + } +} diff --git a/src/utils/sse-session/sse-session.ts b/src/utils/sse-session/sse-session.ts new file mode 100644 index 0000000..9211cae --- /dev/null +++ b/src/utils/sse-session/sse-session.ts @@ -0,0 +1,471 @@ +// Vendored from https://git.harvmaster.com/Harvmaster/sse-session (commit c08267e85c). +// Local changes: import paths, uses ../event-emitter.ts, tryAsync folded into SSESession. + +import type { SSESessionOptions, SSESessionEventMap, SSEvent } from './types.js' + +import { EventEmitter } from '../event-emitter.js' +import { AsyncPushIterator } from './async-push-iterator.js' +import { ExponentialBackoff } from './exponential-backoff.js' + +import { SSEEventParser } from './sse-event-parser.js' + +/** + * A fetch-based Server-Sent Events (SSE) client with reconnect and optional + * browser tab visibility handling. + * + * Each session maintains one HTTP streaming connection at a time. Incoming + * bytes are parsed into {@link SSEvent} objects and delivered through two + * surfaces: + * + * - **Events** — `"connected"`, `"message"`, `"disconnected"`, `"error"`, + * and `"closed"` on the session itself (extends {@link EventEmitter}). + * - **Messages** — {@link messages}, an async iterable for `for await...of` + * consumers. + * + * Typical usage: + * + * ```ts + * const session = await SSESession.create("/events"); + * + * session.on("message", (event) => console.log(event.data)); + * + * for await (const event of session.messages) { + * handle(event); + * } + * ``` + * + * ## Lifecycle + * + * - {@link connect} opens (or reopens) the transport. It resolves once the + * HTTP stream is established; reading continues in the background. + * - {@link abort} stops the in-flight fetch without ending the session. + * Used internally for tab visibility. The {@link messages} iterator stays + * open so an existing consumer resumes when the tab becomes visible again. + * - {@link disconnect} aborts the transport, closes {@link messages}, emits + * `"closed"`, and disables attached visibility handlers until the next + * manual {@link connect}. + * + * Automatic reconnect is controlled by {@link SSESessionOptions.persistent} + * (server closed the stream) and + * {@link SSESessionOptions.attemptReconnect} (transport error). + * + * ## Connection supersession + * + * Each {@link connect} or {@link abort} bumps an internal `connectionId`. + * Background read loops capture their id at start and exit quietly when a + * newer connection supersedes them, avoiding duplicate events or errors from + * stale transports. + */ +export class SSESession extends EventEmitter { + /** + * Creates a session and waits until the first connection is established. + * + * @param url - The SSE endpoint URL. + * @param options - Configuration merged with instance defaults. + * @returns A connected session. + * @throws When the initial connection cannot be established. + */ + static async create(url: string, options: Partial = {}): Promise { + const client = new SSESession(url, options) + await client.connect() + + return client + } + + /** + * Enables SSE resume semantics by sending `Last-Event-ID` on reconnect. + * + * Listens for incoming `"message"` events and remembers the most recent + * {@link SSEvent.id}. On every subsequent connect or reconnect, the session's + * {@link onRequest} hook is wrapped so that header is attached when an id is + * known, allowing the server to replay only events the client has not yet + * received. + * + * The existing {@link onRequest} callback is preserved and runs after the + * header is applied, so auth or other header mutations continue to work. + * + * Attach as early in the session lifetime as possible. When added after + * {@link create}, the initial connection omits the header (no id yet); + * all later reconnects include it. To instrument before the first connect, + * call this on the session returned from {@link withBrowserVisibility} + * before awaiting a separate {@link connect} when the tab starts hidden. + * + * ```ts + * const session = await SSESession.create(url); + * await SSESession.addLastEventIdReconnect(session); + * // Reconnects send Last-Event-ID once an event with an id is received. + * ``` + * + * @param client - The session to instrument. + * @returns The same session, for chaining. + */ + static async addLastEventIdReconnect(client: SSESession): Promise { + let lastEventId: string | undefined + + client.on('message', (event) => { + // If the event has no id, we should not update the lastEventId and instead keep the old one. + lastEventId = event.id ?? lastEventId + }) + + const originalOnRequest = client.options.onRequest + + client.options.onRequest = async (request: RequestInit): Promise => { + if (lastEventId) { + request.headers = { ...request.headers, 'Last-Event-ID': lastEventId } + } + + return originalOnRequest(request) + } + + return client + } + + /** + * Pauses and resumes a session based on browser tab visibility. + * + * Uses the Page Visibility API (`document.visibilitychange`): + * + * - **hidden** — {@link abort} stops the active fetch. {@link messages} + * stays open; `"disconnected"` fires but `"closed"` does not. + * - **visible** — {@link connect} re-establishes the stream if needed. + * + * The listener is removed when {@link disconnect} emits `"closed"`, and + * re-attached automatically on the next `"connected"` event. + * + * No-op in non-browser environments where `document` is undefined. + * + * @param client - The session to manage. + */ + static addBrowserVisibilityHandler(client: SSESession): SSESession { + if (typeof document === 'undefined') return client + + const handleVisibilityChange = (): void => { + if (document.visibilityState === 'hidden') { + void client.abort() + + return + } + + client.connect().catch(() => { + // connect() reports failures via onError and the "error" event. + }) + } + + document.addEventListener('visibilitychange', handleVisibilityChange) + + // Stop managing visibility after an explicit disconnect; re-register + // when the same instance is manually connected again. + client.once('closed', () => { + document.removeEventListener('visibilitychange', handleVisibilityChange) + + client.once('connected', () => { + SSESession.addBrowserVisibilityHandler(client) + }) + }) + + return client + } + + /** SSE endpoint URL for this session. */ + private readonly url: string + + /** + * Per-instance configuration. + * + * Defaults live on the instance field (not a shared static) so each session + * gets its own {@link SSEEventParser} and {@link ExponentialBackoff}. + */ + public options: SSESessionOptions = { + fetch: (...args) => fetch(...args), + method: 'GET', + headers: { + Accept: 'text/event-stream', + 'Cache-Control': 'no-cache', + }, + body: new FormData(), + + onRequest: (request) => Promise.resolve(request), + onConnected: () => {}, + onDisconnected: () => {}, + onError: (error) => console.error('SSEClient error:', error), + + // Retry the initial fetch until it succeeds (maxAttempts: 0 = unlimited). + retry: new ExponentialBackoff({ + baseDelay: 1000, + maxDelay: 10000, + maxAttempts: 0, + growthRate: 1.3, + jitter: 0.3, + }), + + attemptReconnect: true, + persistent: false, + + eventParser: new SSEEventParser(), + } + + /** AbortController for the currently active fetch, if any. */ + private controller: AbortController | null = null + + /** + * Asynchronous stream of parsed SSE events for the active connection. + * + * Stays open across {@link abort} and automatic reconnects so an existing + * `for await` consumer keeps receiving events after visibility resumes. + * + * Closes when: + * - the server ends the stream and {@link SSESessionOptions.persistent} + * is false, + * - {@link disconnect} is called, or + * - a transport error occurs with + * {@link SSESessionOptions.attemptReconnect} disabled. + * + * A later {@link connect} replaces this with a new iterator when the + * previous one was closed. Consumers should read from `session.messages` + * rather than caching a reference across terminal disconnects. + */ + public messages: AsyncPushIterator = new AsyncPushIterator() + + public constructor(url: string, options: Partial = {}) { + super() + + this.url = url + this.options = { + ...this.options, + ...options, + // Shallow merge would drop default headers when options.headers is set. + headers: { ...this.options.headers, ...options.headers }, + } + } + + /** + * Connects or reconnects to the SSE endpoint. + * + * Resolves once the HTTP stream is established and `"connected"` has been + * emitted. Body reading continues asynchronously in the background via + * {@link readStream}. + * + * @throws When the fetch retry policy exhausts attempts or the connection + * is superseded before the reader is handed off (in the latter case the + * promise resolves without throwing). + */ + public async connect(): Promise { + // If there is already a controller present, we are already connected. + if (this.controller) return + + // Prepare for a fresh transport. Parser state from an abandoned connection + // must not bleed into the next one; reopen messages if a prior terminal + // close ended the consumer's iteration loop. + // NOTE: The resetEventParser may be out of spec and should not be done here: https://html.spec.whatwg.org/multipage/server-sent-events.html#event-stream-interpretation + this.options.eventParser.reset() + + // If the last messages iterator wasnt closed, we keep it around as this is likely a reconnect. + if (this.messages.closed) { + this.messages = new AsyncPushIterator() + } + + const controller = new AbortController() + this.controller = controller + + const { method, headers, body } = this.options + + const fetchBody = method === 'POST' ? body : null + + const fetchOptions: RequestInit = { + method, + headers: headers || {}, + body: fetchBody ?? null, + signal: controller.signal, + cache: 'no-store', + } + + let reader: ReadableStreamDefaultReader + + try { + reader = await this.options.retry.run(() => this.createReader(fetchOptions)) + } catch (error) { + // A newer abort/connect superseded this attempt — leave state to the winner. + if (this.controller !== controller) return + + this.controller = null + + await this.notifyDisconnected() + await this.notifyError(error) + this.messages.close() + + throw error + } + + // Connection succeeded but was already replaced (for example abort during fetch). + if (this.controller !== controller) { + await reader.cancel() + + return + } + + await SSESession.tryAsync( + () => this.options.onConnected(), + (error) => this.notifyError(error), + ) + this.emit('connected', undefined) + + // Fire-and-forget: connect() resolves while the stream is consumed. + this.readStream(reader, controller).catch((error) => { + this.options.onError(error) + }) + } + + /** + * Aborts only the currently active transport. + * + * The session remains reusable: {@link messages} stays open, visibility + * handling stays attached, and {@link connect} can reopen the stream. + * Partial parser state from the abandoned transport is discarded. + * + * Emits `"disconnected"` but not `"closed"`. + */ + public async abort(): Promise { + if (!this.controller) return + + // Grab the current controller to ensure we are aborting the correct one. + const controller = this.controller + this.controller = null + + // Invalidate any in-flight read loop and fetch for this transport. + controller.abort() + this.options.eventParser.reset() + + await this.notifyDisconnected() + } + + /** + * Terminates the session and disables attached visibility handling until + * the same instance is manually {@link connect connected} again. + * + * Closes {@link messages} and emits `"closed"`. + */ + public async disconnect(): Promise { + this.messages.close() + this.emit('closed', undefined) + + if (this.controller) { + await this.abort() + } else { + this.options.eventParser.reset() + } + } + + /** + * Performs the HTTP request and returns a reader for the response body. + * + * {@link SSESessionOptions.onRequest} may mutate headers (for example auth + * tokens or `Last-Event-ID`) before the fetch runs. + */ + private async createReader(fetchOptions: RequestInit): Promise> { + const requestOptions = await this.options.onRequest(fetchOptions) + const response = await this.options.fetch(this.url, requestOptions) + + if (!response.ok) { + const responseCode = response.status + const responseText = await response.text() + + const error = new Error(`HTTP error: ${responseCode} ${responseText}`) + void this.notifyError(error) + throw error + } + + if (!response.body) { + const error = new Error('HTTP error: Response body is null') + void this.notifyError(error) + throw error + } + + return response.body.getReader() + } + + /** + * Reads bytes from an established stream until it ends, errors, or is + * superseded by a newer connection. + */ + private async readStream(reader: ReadableStreamDefaultReader, controller: AbortController): Promise { + try { + while (this.controller === controller) { + const { done, value } = await reader.read() + + // abort() or a newer connect() may have landed while we were awaiting. + if (this.controller !== controller) return + + if (done) { + this.controller = null + + await this.notifyDisconnected() + + if (this.options.persistent) { + // Server closed gracefully — reopen unless the consumer opted out. + await this.connect() + } else { + this.messages.close() + } + + return + } + + // Some environments yield `{ done: false, value: undefined }`. + if (!value) continue + + for (const event of this.options.eventParser.parseEvents(value)) { + this.emit('message', event) + this.messages.push(event) + } + } + } catch (error) { + // If the controller is different, we already started a new connection and it would be confusing to handle this error. + if (controller !== this.controller) return + + // Invalidate the current controller to allow for reconnection if needed + this.controller = null + + await this.notifyDisconnected() + + // Expected path for abort() — do not treat as an error or reconnect. + if (controller.signal.aborted) return + + await this.notifyError(error) + + if (this.options.attemptReconnect) { + await this.connect() + } else { + this.messages.close() + } + } + } + + /** Invokes {@link SSESessionOptions.onDisconnected} and emits `"disconnected"`. */ + private async notifyDisconnected(): Promise { + await SSESession.tryAsync( + () => this.options.onDisconnected(), + (error) => this.options.onError(error), + ) + this.emit('disconnected', undefined) + } + + /** Invokes {@link SSESessionOptions.onError} and emits `"error"`. */ + private async notifyError(error: unknown): Promise { + const errorInstance = error instanceof Error ? error : new Error(String(error)) + + await SSESession.tryAsync( + () => this.options.onError(errorInstance), + (callbackError) => console.error('SSESession error:', callbackError), + ) + this.emit('error', errorInstance) + } + + /** Runs `fn`, passing any error to `onError` instead of throwing. */ + private static async tryAsync(fn: () => unknown, onError?: (error: Error) => void): Promise { + try { + await fn() + } catch (error) { + onError?.(error instanceof Error ? error : new Error(`${error}`)) + } + } +} diff --git a/src/utils/sse-session/types.ts b/src/utils/sse-session/types.ts new file mode 100644 index 0000000..997b0ce --- /dev/null +++ b/src/utils/sse-session/types.ts @@ -0,0 +1,145 @@ +// Vendored from https://git.harvmaster.com/Harvmaster/sse-session (commit c08267e85c). +// Local changes: import paths, uses ../event-emitter.ts, tryAsync folded into SSESession. + +export type SSERequestInit = { + /** + * The HTTP method to use. + */ + method: 'GET' | 'POST' + + /** + * Request headers sent on every connect and reconnect. + */ + headers?: Record + + /** + * Request body for POST-based SSE endpoints. + */ + body?: string | FormData +} + +/** + * The fetch function to use. + * + * NOTE: This is compatible with Browser/Node's native "fetch" function. + * We use this in place of "typeof fetch" so that we can accept non-standard URLs ("url" is a "string" here). + * For example, a LibP2P adapter might not use a standardized URL format (and might only include "path"). + * This would cause a type error as native fetch expects type "URL". + */ +export type SSERequestFunction = { + fetch: (url: string, options: RequestInit) => Promise +} + +/** + * Lifecycle hooks invoked by {@link SSESession} during connect, read, and teardown. + */ +export type SSESessionCallbacks = { + /** + * Called before each fetch so callers can attach auth headers, cookies, or + * a `Last-Event-ID` for resume semantics. + */ + onRequest: (request: RequestInit) => Promise + + /** + * Called after the HTTP stream is established and before body reading begins. + */ + onConnected: () => void + + /** + * Called when the active transport ends — including {@link SSESession.abort}, + * server stream completion, and errors. Not paired with {@link SSESessionCallbacks.onConnected} + * when the initial connect never succeeds. + */ + onDisconnected: () => void + + /** + * Called on fetch or read failures. Not invoked for intentional + * {@link SSESession.abort} aborts. + */ + onError: (error: Error) => void +} + +export type SSESessionRetryInterface = { + /** + * Retry policy used while establishing the HTTP connection in + * {@link SSESession.connect}. Defaults to {@link ExponentialBackoff} with + * unlimited attempts. + */ + retry: { + run(fn: () => Promise, onError?: (error: Error) => void): Promise + } +} + +export interface SSEParser { + /** + * Incremental SSE frame parser for the response body. + * + * {@link SSEEventParser.reset} is called by the session when abandoning a + * transport so partial frames do not carry over to the next connection. + */ + eventParser: { + parseEvents(buffer: Uint8Array): SSEvent[] + reset(): void + } +} + +export type SSELifecycleOptions = { + /** + * When true, {@link SSESession} calls {@link SSESession.connect} again after + * a transport **error** (not an intentional abort). + */ + attemptReconnect: boolean + + /** + * When true, {@link SSESession} calls {@link SSESession.connect} again after + * the **server** closes the stream normally (`done`). + */ + persistent: boolean +} + +/** + * Events emitted by {@link SSESession}. + * + * - `"connected"` — HTTP stream established. + * - `"message"` — A complete SSE event was parsed. + * - `"disconnected"` — The active transport ended (including {@link SSESession.abort}). + * - `"error"` — An unexpected fetch or read failure. + * - `"closed"` — {@link SSESession.disconnect} was called; visibility handling is detached. + */ +export type SSESessionEventMap = { + connected: void + disconnected: void + error: Error + message: SSEvent + closed: void +} + +/** + * Configuration for {@link SSESession}. + */ +export type SSESessionOptions = SSESessionCallbacks & SSERequestInit & SSERequestFunction & SSESessionRetryInterface & SSELifecycleOptions & SSEParser + +/** + * Represents a Server-Sent Event. + */ +export interface SSEvent { + /** + * Event data. + */ + data: string + + /** + * Event type. + */ + event?: string + + /** + * Event ID. + */ + id?: string + + /** + * Reconnection time in milliseconds. + */ + retry?: number +} diff --git a/test/helpers.ts b/test/helpers.ts index 0c00824..65550dc 100644 --- a/test/helpers.ts +++ b/test/helpers.ts @@ -10,6 +10,7 @@ export async function collect(iterable: AsyncIterable): Promise { /** * Wraps another KV storage and counts primitive calls. Also shows the wrapper pattern * (e.g. an encryption layer): extend BaseKVStorage and delegate the five primitives. + * Counts are reported under the public method names (writeMany, removeMany, writeIfAbsent). */ export class KVStorageCounting extends BaseKVStorage { calls: Record = {} @@ -25,13 +26,13 @@ export class KVStorageCounting extends BaseKVStorage { readMany(keys: string[]) { return this.count('readMany', () => this.inner.readMany(keys)) } - writeMany(entries: [string, unknown][]) { + protected applyWrite(entries: [string, unknown][]) { return this.count('writeMany', () => this.inner.writeMany(entries)) } - removeMany(keys: string[]) { + protected applyRemove(keys: string[]) { return this.count('removeMany', () => this.inner.removeMany(keys)) } - writeIfAbsent(entries: [string, unknown][]) { + protected applyWriteIfAbsent(entries: [string, unknown][]) { return this.count('writeIfAbsent', () => this.inner.writeIfAbsent(entries)) } scan(options?: ScanOptions) { diff --git a/test/kv-storage/kv-storage-composite.test.ts b/test/kv-storage/kv-storage-composite.test.ts new file mode 100644 index 0000000..c6bbd22 --- /dev/null +++ b/test/kv-storage/kv-storage-composite.test.ts @@ -0,0 +1,144 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' +import { KVStorageComposite } from '../../src/kv-storage/kv-storage-composite.js' +import { KVStorageInMemory } from '../../src/kv-storage/kv-storage-in-memory.js' +import type { BaseKVStorage } from '../../src/kv-storage/base-kv-storage.js' +import { DocumentStorageKV } from '../../src/document-storage/document-storage-kv.js' +import { collect } from '../helpers.js' +import { runKVStorageConformance } from './kv-storage.conformance.js' +import { TestNetwork } from './sync-helpers.js' + +runKVStorageConformance( + 'KVStorageComposite (in-memory source, 1 replica)', + () => new KVStorageComposite({ source: new KVStorageInMemory(), replicas: [new KVStorageInMemory()] }), +) +runKVStorageConformance('KVStorageComposite (http source, 2 replicas)', () => { + network = new TestNetwork() + return new KVStorageComposite({ source: network.client(), replicas: [new KVStorageInMemory(), new KVStorageInMemory()] }) +}) + +let network: TestNetwork +const composites: KVStorageComposite[] = [] +afterEach(async () => { + await Promise.all(composites.splice(0).map((c) => c.close())) + await network?.close() +}) + +function composite(persistent: BaseKVStorage = new KVStorageInMemory()) { + const c = new KVStorageComposite({ source: network.client(), replicas: [new KVStorageInMemory(), persistent], retry: { baseDelay: 5, maxDelay: 20 } }) + composites.push(c) + return c +} + +describe('KVStorageComposite sync', () => { + it('loads existing data with one full sync, then serves reads locally', async () => { + network = new TestNetwork() + await network.store.writeMany([ + ['users/1', { name: 'Ada' }], + ['users/2', { name: 'Bob' }], + ]) + const db = composite() + await db.synced() + network.store.reset() + expect(await db.child('users').get('1')).toEqual({ name: 'Ada' }) + expect(await collect(db.child('users').keys())).toEqual(['1', '2']) + expect(network.store.calls).toEqual({}) + }) + + it('applies changes pushed from other clients to every replica', async () => { + network = new TestNetwork() + const memory = new KVStorageInMemory() + const db = composite(memory) + await db.synced() + const seen = db.waitFor('change', (e) => e.origin === 'remote') + await network.client().child('users').set('9', { name: 'Remote' }) + expect((await seen).changes).toEqual([{ key: 'users/9', value: { name: 'Remote' } }]) + expect(await db.child('users').get('9')).toEqual({ name: 'Remote' }) + expect(await memory.readMany(['users/9'])).toEqual([{ name: 'Remote' }]) + }) + + it('queues writes while offline and replays them in order when back online', async () => { + network = new TestNetwork() + const db = composite() + await db.synced() + network.online = false + await db.child('users').set('1', { v: 1 }) + await db.child('users').set('1', { v: 2 }) + await db.child('users').delete('1') + await db.child('users').set('2', { v: 3 }) + expect(db.queued).toBe(4) + expect(await db.child('users').get('2')).toEqual({ v: 3 }) // served locally + await expect(db.writeIfAbsent([['x', 1]])).rejects.toThrow(/Cannot reach/) + network.online = true + await vi.waitFor(() => expect(db.queued).toBe(0)) + expect(await network.store.readMany(['users/1', 'users/2'])).toEqual([undefined, { v: 3 }]) + }) + + it('keeps queued local values when the server pushes an older value for the same key', async () => { + network = new TestNetwork() + const db = composite() + await db.synced() + network.online = false + await db.child('k').set({ from: 'local' }) + await network.store.writeMany([['k', { from: 'remote' }]]) // pushed over the still-open stream + await new Promise((r) => setTimeout(r, 20)) + expect(await db.child('k').get()).toEqual({ from: 'local' }) + network.online = true + await vi.waitFor(() => expect(db.queued).toBe(0)) + expect(await network.store.readMany(['k'])).toEqual([{ from: 'local' }]) + }) + + it('restarts from the persistent replica and resumes without a full sync', async () => { + network = new TestNetwork() + const persistent = new KVStorageInMemory() + const first = composite(persistent) + await first.child('a').set(1) + await first.synced() + await first.close() + await network.client().writeMany([['b', 2]]) // missed while "closed" + + network.store.reset() + const second = composite(persistent) + expect(await second.child('a').get()).toBe(1) // available before contacting the server + await vi.waitFor(async () => expect(await second.child('b').get()).toBe(2)) + expect(network.store.calls.scan).toBeUndefined() // resumed via Last-Event-ID, no full scan + }) + + it('falls back to a full sync when the server cannot resume', async () => { + network = new TestNetwork({ historySize: 1 }) + const persistent = new KVStorageInMemory() + const first = composite(persistent) + await first.child('a').set(1) + await first.synced() + await first.close() + await network.client().writeMany([['b', 2]]) + await network.client().writeMany([['c', 3]]) + await network.client().removeMany(['a']) + const second = composite(persistent) + await vi.waitFor(async () => expect(await collect(second.keys())).toEqual(['b', 'c'])) + }) + + it('replays an outbox left by a previous session', async () => { + network = new TestNetwork() + const persistent = new KVStorageInMemory() + const first = composite(persistent) + await first.synced() + network.online = false + await first.child('offline').set('queued') + await first.close() + network.online = true + const second = composite(persistent) + await second.start() + await vi.waitFor(async () => expect(await network.store.readMany(['offline'])).toEqual(['queued'])) + }) + + it('serves document queries from the local replica, offline too', async () => { + network = new TestNetwork() + const db = composite() + const users = new DocumentStorageKV(db).child('users') + await users.set('1', { age: 30 }) + await users.set('2', { age: 40 }) + await db.synced() + network.online = false + expect((await collect(users.find({ age: { $gt: 35 } }))).map((r) => r.key)).toEqual(['2']) + }) +}) diff --git a/test/kv-storage/kv-storage-http-watch.test.ts b/test/kv-storage/kv-storage-http-watch.test.ts new file mode 100644 index 0000000..889a9e9 --- /dev/null +++ b/test/kv-storage/kv-storage-http-watch.test.ts @@ -0,0 +1,61 @@ +import { afterEach, describe, expect, it } from 'vitest' +import type { KVChangeEvent } from '../../src/kv-storage/base-kv-storage.js' +import { StorageUnavailableError } from '../../src/utils/errors.js' +import { TestNetwork } from './sync-helpers.js' + +let network: TestNetwork +afterEach(() => network?.close()) + +describe('KVStorageHttp.watch (SSE)', () => { + it('receives changes made by other clients as remote change events', async () => { + network = new TestNetwork() + const watcher = network.client() + await watcher.watch() + const received = watcher.waitFor('change', (e) => e.origin === 'remote') + await network.client().child('users').set('1', { name: 'Ada', n: 1n }) + const event = await received + expect(event.changes).toEqual([{ key: 'users/1', value: { name: 'Ada', n: 1n } }]) + expect(event.id).toMatch(/^\w+-1$/) + expect(watcher.lastEventId).toBe(event.id) + }) + + it('resumes from lastEventId and replays missed changes', async () => { + network = new TestNetwork() + const writer = network.client() + const watcher = network.client() + await watcher.watch() + const first = watcher.waitFor('change', (e) => e.origin === 'remote') + await writer.writeMany([['a', 1]]) + await first + await watcher.unwatch() + await writer.writeMany([['b', 2]]) + await writer.writeMany([['c', 3]]) + const replayed: KVChangeEvent[] = [] + watcher.on('change', (e) => replayed.push(e)) + await watcher.watch() // resolves once caught up, so the missed changes have been emitted + expect(replayed.map((e) => e.changes)).toEqual([[{ key: 'b', value: 2 }], [{ key: 'c', value: 3 }]]) + }) + + it('asks for a resync when the missed changes are no longer in the history', async () => { + network = new TestNetwork({ historySize: 1 }) + const writer = network.client() + const watcher = network.client() + await watcher.watch() + const seen = watcher.waitFor('change', () => true) + await writer.writeMany([['a', 1]]) + await seen + await watcher.unwatch() + await writer.writeMany([['b', 2]]) + await writer.writeMany([['c', 3]]) + const resync = watcher.waitFor('resync', () => true) + await watcher.watch() + expect(await resync).toEqual({ id: expect.stringMatching(/-3$/) }) + }) + + it('reports an unreachable server as StorageUnavailableError', async () => { + network = new TestNetwork() + const client = network.client() + network.online = false + await expect(client.readMany(['x'])).rejects.toBeInstanceOf(StorageUnavailableError) + }) +}) diff --git a/test/kv-storage/kv-storage-http.test.ts b/test/kv-storage/kv-storage-http.test.ts index 8a5a3b9..2dc295e 100644 --- a/test/kv-storage/kv-storage-http.test.ts +++ b/test/kv-storage/kv-storage-http.test.ts @@ -9,7 +9,7 @@ import { runKVStorageConformance } from './kv-storage.conformance.js' import { serve } from './node-server.js' runKVStorageConformance('KVStorageHttp (in-process)', () => { - const server = new KVStorageHttpServer(new KVStorageInMemory(), '/kv') + const server = new KVStorageHttpServer(new KVStorageInMemory(), { basePath: '/kv' }) return new KVStorageHttp({ url: 'http://test/kv', fetch: async (input, init) => server.handle(new Request(input, init)) }) }) @@ -22,7 +22,7 @@ describe('over a real socket', () => { }) afterAll(() => server.close()) const fresh = () => { - stores.push(new KVStorageHttpServer(new KVStorageInMemory(), `/${stores.length}`)) + stores.push(new KVStorageHttpServer(new KVStorageInMemory(), { basePath: `/${stores.length}` })) return new KVStorageHttp({ url: `${server.url}/${stores.length - 1}` }) } diff --git a/test/kv-storage/kv-storage-in-memory.test.ts b/test/kv-storage/kv-storage-in-memory.test.ts index 68c09a5..20f5e2a 100644 --- a/test/kv-storage/kv-storage-in-memory.test.ts +++ b/test/kv-storage/kv-storage-in-memory.test.ts @@ -76,3 +76,12 @@ describe('model-based: random writes vs a reference map', () => { ) }) }) + +describe('child nodes', () => { + it('are frozen, so implementations cannot accidentally shadow shared state', () => { + const child = new KVStorageInMemory().child('a').child('b') + expect(Object.isFrozen(child)).toBe(true) + expect(() => ((child as unknown as { state: number }).state = 1)).toThrow(TypeError) + expect(child.key).toBe('a/b') + }) +}) diff --git a/test/kv-storage/kv-storage-indexeddb.test.ts b/test/kv-storage/kv-storage-indexeddb.test.ts new file mode 100644 index 0000000..8fe639c --- /dev/null +++ b/test/kv-storage/kv-storage-indexeddb.test.ts @@ -0,0 +1,8 @@ +import 'fake-indexeddb/auto' +import { KVStorageIndexedDB } from '../../src/kv-storage/kv-storage-indexeddb.js' +import { runKVStorageConformance } from './kv-storage.conformance.js' + +let n = 0 +runKVStorageConformance('KVStorageIndexedDB', () => new KVStorageIndexedDB({ name: `db-${n++}` })) +// Tiny batches exercise resuming scans across transactions. +runKVStorageConformance('KVStorageIndexedDB (scan batches of 2)', () => new KVStorageIndexedDB({ name: `db-${n++}`, scanBatchSize: 2 })) diff --git a/test/kv-storage/kv-storage.conformance.ts b/test/kv-storage/kv-storage.conformance.ts index 939afe4..f5814bd 100644 --- a/test/kv-storage/kv-storage.conformance.ts +++ b/test/kv-storage/kv-storage.conformance.ts @@ -71,6 +71,28 @@ export function runKVStorageConformance(name: string, make: () => BaseKVStorage expect(seen.filter((k) => k !== 'a/1')).toEqual(['10', '9', 'a', 'a/2', 'a0', 'b', 'c']) }) + it('emits change events for writes and removals', async () => { + const events: unknown[] = [] + const off = db.on('change', (event) => events.push(event)) + await db.writeMany([['x', 1]]) + await db.removeMany(['x']) + await db.writeIfAbsent([ + ['a', 'no'], + ['y', 2], + ]) + off() + expect(events).toEqual([ + { changes: [{ key: 'x', value: 1 }], origin: 'local' }, + { changes: [{ key: 'x' }], origin: 'local' }, + { changes: [{ key: 'y', value: 2 }], origin: 'local' }, + ]) + }) + + it('stores bigint and Uint8Array values', async () => { + await db.writeMany([['bin', { n: 10n ** 20n, bytes: new Uint8Array([0, 255]) }]]) + expect(await db.readMany(['bin'])).toEqual([{ n: 10n ** 20n, bytes: new Uint8Array([0, 255]) }]) + }) + it('does not share references with callers', async () => { const [value] = (await db.readMany(['a/2'])) as [{ n: string }] value.n = 'changed' diff --git a/test/kv-storage/sync-helpers.ts b/test/kv-storage/sync-helpers.ts new file mode 100644 index 0000000..e37e1a3 --- /dev/null +++ b/test/kv-storage/sync-helpers.ts @@ -0,0 +1,35 @@ +import { KVStorageHttp } from '../../src/kv-storage/kv-storage-http.js' +import { KVStorageHttpServer, type KVStorageHttpServerOptions } from '../../src/kv-storage/kv-storage-http-server.js' +import { KVStorageInMemory } from '../../src/kv-storage/kv-storage-in-memory.js' +import { KVStorageCounting } from '../helpers.js' + +/** + * An in-process server with a network switch. While `online` is false, new requests fail like a + * dropped connection (open SSE streams stay up, so pushes can still be simulated). + */ +export class TestNetwork { + online = true + readonly store = new KVStorageCounting(new KVStorageInMemory()) + readonly server: KVStorageHttpServer + private readonly clients: KVStorageHttp[] = [] + + constructor(options: KVStorageHttpServerOptions = {}) { + this.server = new KVStorageHttpServer(this.store, options) + } + + client(): KVStorageHttp { + const client = new KVStorageHttp({ + url: 'http://server', + fetch: async (url, init) => { + if (!this.online) throw new TypeError('fetch failed') + return this.server.handle(new Request(url, init)) + }, + }) + this.clients.push(client) + return client + } + + async close(): Promise { + await Promise.all(this.clients.map((c) => c.unwatch())) + } +} diff --git a/test/utils/event-emitter.test.ts b/test/utils/event-emitter.test.ts new file mode 100644 index 0000000..f6b47f5 --- /dev/null +++ b/test/utils/event-emitter.test.ts @@ -0,0 +1,58 @@ +import { describe, expect, it, vi } from 'vitest' +import { EventEmitter } from '../../src/utils/event-emitter.js' + +type Events = { tick: number; done: void } + +describe('EventEmitter', () => { + it('delivers payloads as-is (no cloning or freezing) and supports off', () => { + const emitter = new EventEmitter() + const seen: unknown[] = [] + const payload = { n: 1 } + const off = emitter.on('obj', (p) => seen.push(p)) + expect(emitter.emit('obj', payload)).toBe(true) + off() + expect(emitter.emit('obj', payload)).toBe(false) + expect(seen[0]).toBe(payload) + expect(Object.isFrozen(payload)).toBe(false) + }) + + it('once fires a single time; listener errors are logged, not thrown', () => { + const error = vi.spyOn(console, 'error').mockImplementation(() => {}) + const emitter = new EventEmitter() + const once = vi.fn() + emitter.once('tick', once) + emitter.on('tick', () => { + throw new Error('boom') + }) + emitter.emit('tick', 1) + emitter.emit('tick', 2) + expect(once).toHaveBeenCalledTimes(1) + expect(error).toHaveBeenCalledTimes(2) + error.mockRestore() + }) + + it('debounces and cancels pending debounced calls on off', async () => { + vi.useFakeTimers() + const emitter = new EventEmitter() + const listener = vi.fn() + const off = emitter.on('tick', listener, 50) + emitter.emit('tick', 1) + emitter.emit('tick', 2) + vi.advanceTimersByTime(60) + expect(listener).toHaveBeenCalledExactlyOnceWith(2) + emitter.emit('tick', 3) + off() + vi.advanceTimersByTime(60) + expect(listener).toHaveBeenCalledTimes(1) + vi.useRealTimers() + }) + + it('waitFor resolves on a matching payload and rejects on timeout', async () => { + const emitter = new EventEmitter() + const waiting = emitter.waitFor('tick', (n) => n > 1) + emitter.emit('tick', 1) + emitter.emit('tick', 2) + expect(await waiting).toBe(2) + await expect(emitter.waitFor('done', () => true, 5)).rejects.toThrow(/Timed out waiting for "done"/) + }) +}) diff --git a/test/utils/extended-json.test.ts b/test/utils/extended-json.test.ts new file mode 100644 index 0000000..c4dd21f --- /dev/null +++ b/test/utils/extended-json.test.ts @@ -0,0 +1,32 @@ +import { describe, expect, it } from 'vitest' +import fc from 'fast-check' +import { ExtendedJson } from '../../src/utils/extended-json.js' + +describe('ExtendedJson', () => { + it('round-trips bigint and Uint8Array, nested anywhere', () => { + const value = { a: 1n, b: [new Uint8Array([0, 1, 255]), -(2n ** 70n)], c: { d: new Uint8Array(0), e: 'plain', f: null } } + expect(ExtendedJson.stringify({ n: 5n, bytes: new Uint8Array([171, 205]) })).toBe('{"n":"","bytes":""}') + expect(ExtendedJson.parse(ExtendedJson.stringify(value))).toEqual(value) + }) + + it('round-trips arbitrary data (property)', () => { + const data = fc.letrec((tie) => ({ + value: fc.oneof( + { depthSize: 'small' }, + fc.bigInt(), + fc.uint8Array(), + fc.string(), + fc.double({ noNaN: true, noDefaultInfinity: true }), + fc.boolean(), + fc.constant(null), + fc.array(tie('value')), + fc.dictionary(fc.string(), tie('value')), + ), + })).value + fc.assert( + fc.property(data, (v) => { + expect(ExtendedJson.parse(ExtendedJson.stringify(v))).toEqual(v) + }), + ) + }) +}) diff --git a/test/utils/sse-session.test.ts b/test/utils/sse-session.test.ts new file mode 100644 index 0000000..38cba5b --- /dev/null +++ b/test/utils/sse-session.test.ts @@ -0,0 +1,32 @@ +import { describe, expect, it } from 'vitest' +import { SSEEventParser } from '../../src/utils/sse-session/sse-event-parser.js' +import { SSESession } from '../../src/utils/sse-session/sse-session.js' + +const encode = (text: string) => new TextEncoder().encode(text) + +describe('vendored sse-session', () => { + it('parses events across arbitrary chunk boundaries', () => { + const parser = new SSEEventParser() + const text = 'id: 1\nevent: change\ndata: {"a":1}\n\n: comment\ndata: line1\ndata: line2\n\n' + const events = [] + for (let i = 0; i < text.length; i += 3) events.push(...parser.parseEvents(encode(text.slice(i, i + 3)))) + expect(events).toEqual([{ id: '1', event: 'change', data: '{"a":1}' }, { data: 'line1\nline2' }]) + }) + + it('connects, delivers messages and reconnects when persistent', async () => { + let connections = 0 + const session = new SSESession('http://test/events', { + persistent: true, + fetch: async () => { + connections++ + return new Response(encode(`data: hello ${connections}\n\n`)) + }, + }) + const received: string[] = [] + session.on('message', (m) => received.push(m.data)) + await session.connect() + await session.waitFor('message', (m) => m.data === 'hello 2') + await session.disconnect() + expect(received.slice(0, 2)).toEqual(['hello 1', 'hello 2']) + }) +}) diff --git a/tsconfig.json b/tsconfig.json index a33d2d0..568f991 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -3,7 +3,7 @@ "target": "ES2022", "module": "NodeNext", "moduleResolution": "NodeNext", - "lib": ["ES2022"], + "lib": ["ES2022", "DOM", "DOM.Iterable", "DOM.AsyncIterable"], "types": ["node"], "strict": true, "noUncheckedIndexedAccess": true,