diff --git a/DESIGN.md b/DESIGN.md index 65087af..346784f 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -566,6 +566,29 @@ Decisions taken where the brief was silent. Each is the simplest option consiste - **Type tests** live in `test/types.test-d.ts` and are enforced by `tsc --noEmit` (part of `npm test`), including the `@ts-expect-error` negative cases from section 7. +### M8 — HTTP driver + +- **Extra endpoints** beyond the brief's list: `POST /deleteRange {gte, lt}` and + `GET /capabilities`. Unsupported optional operations return `501`, driver errors `500`, and unknown + routes `404`. +- **Missing vs `null`**: `/getMany` returns `{ values, missing }`, where `missing` lists the indices of + absent keys, because `null` is a legitimate stored value. +- **Codec scope**: the codec serialises whole request/response bodies and each NDJSON scan line + (`[key]` or `[key, value]`), so it must not emit newlines. The default is JSON. +- **Capabilities on the client**: `HttpKVDriver` defines `setIfAbsent`/`batch` only when the server + supports them, so stores' feature detection (e.g. unique indexes requiring `setIfAbsent`, atomic + index writes using `batch`) behaves as it would locally. `HttpKVDriver.connect()` discovers them via + `/capabilities`. The constructor takes them explicitly (default: all). Without native + `deleteRange` the client falls back to `BaseKVDriver`'s chunked delete. +- **Scan streaming**: the server pulls the driver iterator in chunks that grow from 16 to 256 + entries, so short reads (skip-scan re-seeks) don't make it run far ahead. Cancelling the client + iterator cancels the response body, which calls `return()` on the server-side iterator. +- **Optional `headers`** (e.g. auth) and an injectable `fetch` on the client; an optional `basePath` on + the handler. +- The conformance suite runs end-to-end in-process (fetch → handler), over a real `node:http` + socket, and against a minimal server driver with discovered capabilities. A smoke test runs KV + indexes, events and unique-indexed documents over HTTP. + ## 13. Benchmark baselines diff --git a/src/http/client.ts b/src/http/client.ts new file mode 100644 index 0000000..86e4916 --- /dev/null +++ b/src/http/client.ts @@ -0,0 +1,115 @@ +import { BaseKVDriver, JsonCodec, type BatchOp, type Codec, type ScanOptions } from '../kv/driver.js' +import type { HttpCapabilities } from './handler.js' + +export interface HttpKVDriverOptions { + /** Base URL of a `createKVHttpHandler` endpoint, e.g. `https://host/kv`. */ + url: string + /** Optional capabilities the server's driver supports. Default: all. `connect()` discovers them. */ + capabilities?: Partial + codec?: Codec + headers?: Record + fetch?: typeof fetch +} + +/** + * Driver client for `createKVHttpHandler`. Optional methods exist only when the server + * supports them, so stores' feature detection works across HTTP. + */ +export class HttpKVDriver extends BaseKVDriver { + readonly #url: string + readonly #codec: Codec + readonly #headers: Record + readonly #fetch: typeof fetch + readonly #nativeDeleteRange: boolean + + setIfAbsent?: (entries: [string, unknown][]) => Promise + batch?: (ops: BatchOp[]) => Promise + + /** Creates a client after asking the server which optional capabilities it has. */ + static async connect(opts: Omit): Promise { + const probe = new HttpKVDriver({ ...opts, capabilities: {} }) + const capabilities = (await probe.#request('GET', '/capabilities')) as HttpCapabilities + return new HttpKVDriver({ ...opts, capabilities }) + } + + constructor(opts: HttpKVDriverOptions) { + super() + this.#url = opts.url.replace(/\/$/, '') + this.#codec = opts.codec ?? JsonCodec + this.#headers = opts.headers ?? {} + this.#fetch = opts.fetch ?? globalThis.fetch.bind(globalThis) + const caps = { setIfAbsent: true, batch: true, deleteRange: true, ...opts.capabilities } + this.#nativeDeleteRange = caps.deleteRange + if (caps.setIfAbsent) this.setIfAbsent = async (entries) => ((await this.#request('POST', '/setIfAbsent', { entries })) as { written: boolean[] }).written + if (caps.batch) this.batch = async (ops) => void (await this.#request('POST', '/batch', { ops })) + } + + async getMany(keys: string[]): Promise { + if (keys.length === 0) return [] + const { values, missing } = (await this.#request('POST', '/getMany', { keys })) as { values: unknown[]; missing: number[] } + for (const i of missing) values[i] = undefined + return values + } + + async setMany(entries: [string, unknown][]): Promise { + if (entries.length) await this.#request('POST', '/setMany', { entries }) + } + + async deleteMany(keys: string[]): Promise { + if (keys.length) await this.#request('POST', '/deleteMany', { keys }) + } + + override async deleteRange(gte: string, lt: string): Promise { + if (!this.#nativeDeleteRange) return super.deleteRange(gte, lt) + await this.#request('POST', '/deleteRange', { gte, lt }) + } + + async *scan(opts: ScanOptions): AsyncGenerator<[string, unknown]> { + if (opts.limit !== undefined && opts.limit <= 0) return + const q = new URLSearchParams() + if (opts.gte !== undefined) q.set('gte', opts.gte) + if (opts.lt !== undefined) q.set('lt', opts.lt) + if (opts.limit !== undefined && opts.limit !== Infinity) q.set('limit', String(opts.limit)) + if (opts.reverse) q.set('reverse', 'true') + if (opts.values) q.set('values', 'true') + const res = await this.#fetch(`${this.#url}/scan?${q}`, { headers: this.#headers }) + if (!res.ok || !res.body) throw await httpError(res) + const reader = res.body.getReader() + const decoder = new TextDecoder() + let buf = '' + try { + for (;;) { + const { done, value } = await reader.read() + if (done) break + buf += decoder.decode(value, { stream: true }) + let nl: number + while ((nl = buf.indexOf('\n')) !== -1) { + const line = buf.slice(0, nl) + buf = buf.slice(nl + 1) + const [key, v] = this.#codec.decode(line) as [string, unknown?] + yield [key, v] + } + } + } finally { + await reader.cancel().catch(() => {}) + } + } + + async #request(method: 'GET' | 'POST', path: string, body?: unknown): Promise { + const res = await this.#fetch(this.#url + path, { + method, + headers: { ...this.#headers, ...(body === undefined ? {} : { 'content-type': 'application/json' }) }, + body: body === undefined ? undefined : this.#codec.encode(body), + }) + if (!res.ok) throw await httpError(res) + return res.status === 204 ? undefined : this.#codec.decode(await res.text()) + } +} + +async function httpError(res: Response): Promise { + let detail = '' + try { + detail = await res.text() + } catch {} + return new Error(`HTTP ${res.status}: ${detail}`) +} diff --git a/src/http/handler.ts b/src/http/handler.ts new file mode 100644 index 0000000..c557ecd --- /dev/null +++ b/src/http/handler.ts @@ -0,0 +1,115 @@ +import { JsonCodec, type BatchOp, type Codec, type KVDriver, type ScanOptions } from '../kv/driver.js' + +export interface KVHttpHandlerOptions { + /** Serialises request/response bodies and NDJSON scan lines. Must not emit newlines. Default JSON. */ + codec?: Codec + /** Path prefix to strip, e.g. `/kv`. Default `''`. */ + basePath?: string +} + +export interface HttpCapabilities { + setIfAbsent: boolean + batch: boolean + deleteRange: boolean +} + +/** + * Framework-agnostic HTTP front for a driver (`Request → Response`). + * + * - `POST /getMany {keys}` → `{values, missing}` (`missing` = indices of absent keys) + * - `POST /setMany {entries}`, `POST /deleteMany {keys}`, `POST /batch {ops}` → 204 + * - `POST /setIfAbsent {entries}` → `{written}` + * - `POST /deleteRange {gte, lt}` → 204 + * - `GET /scan?gte<&limit&reverse&values` → NDJSON, one `[key]` or `[key, value]` per line + * - `GET /capabilities` → `{setIfAbsent, batch, deleteRange}` + */ +export function createKVHttpHandler(driver: KVDriver, opts: KVHttpHandlerOptions = {}): (req: Request) => Promise { + const codec = opts.codec ?? JsonCodec + const base = opts.basePath ?? '' + const reply = (body: unknown, status = 200) => new Response(codec.encode(body), { status, headers: { 'content-type': 'application/json' } }) + const noContent = () => new Response(null, { status: 204 }) + const fail = (status: number, message: string) => reply({ error: message }, status) + + return async (req) => { + const url = new URL(req.url) + if (!url.pathname.startsWith(base)) return fail(404, 'Not found') + const route = `${req.method} ${url.pathname.slice(base.length)}` + try { + const body = async () => codec.decode(await req.text()) as T + switch (route) { + case 'POST /getMany': { + const { keys } = await body<{ keys: string[] }>() + const found = await driver.getMany(keys) + const missing: number[] = [] + const values = found.map((v, i) => (v === undefined ? (missing.push(i), null) : v)) + return reply({ values, missing }) + } + case 'POST /setMany': + await driver.setMany((await body<{ entries: [string, unknown][] }>()).entries) + return noContent() + case 'POST /deleteMany': + await driver.deleteMany((await body<{ keys: string[] }>()).keys) + return noContent() + case 'POST /setIfAbsent': + if (!driver.setIfAbsent) return fail(501, 'setIfAbsent not supported') + return reply({ written: await driver.setIfAbsent((await body<{ entries: [string, unknown][] }>()).entries) }) + case 'POST /batch': + if (!driver.batch) return fail(501, 'batch not supported') + await driver.batch((await body<{ ops: BatchOp[] }>()).ops) + return noContent() + case 'POST /deleteRange': { + if (!driver.deleteRange) return fail(501, 'deleteRange not supported') + const { gte, lt } = await body<{ gte: string; lt: string }>() + await driver.deleteRange(gte, lt) + return noContent() + } + case 'GET /capabilities': + return reply({ setIfAbsent: !!driver.setIfAbsent, batch: !!driver.batch, deleteRange: !!driver.deleteRange } satisfies HttpCapabilities) + case 'GET /scan': + return scanResponse(driver, parseScan(url.searchParams), codec) + default: + return fail(404, `No route ${route}`) + } + } catch (err) { + return fail(500, err instanceof Error ? err.message : String(err)) + } + } +} + +function parseScan(q: URLSearchParams): ScanOptions { + const opts: ScanOptions = {} + if (q.has('gte')) opts.gte = q.get('gte')! + if (q.has('lt')) opts.lt = q.get('lt')! + if (q.has('limit')) opts.limit = Number(q.get('limit')) + if (q.get('reverse') === 'true') opts.reverse = true + if (q.get('values') === 'true') opts.values = true + return opts +} + +function scanResponse(driver: KVDriver, opts: ScanOptions, codec: Codec): Response { + const it = driver.scan(opts)[Symbol.asyncIterator]() + const encoder = new TextEncoder() + // Chunks grow so short (skip-scan) reads don't make the server run far ahead. + let chunk = 16 + const stream = new ReadableStream({ + async pull(controller) { + let buf = '' + for (let i = 0; i < chunk; i++) { + const r = await it.next() + if (r.done) { + if (buf) controller.enqueue(encoder.encode(buf)) + controller.close() + return + } + const [key, value] = r.value + buf += codec.encode(opts.values ? [key, value] : [key]) + '\n' + } + chunk = Math.min(chunk * 2, 256) + controller.enqueue(encoder.encode(buf)) + }, + async cancel() { + await it.return?.() + }, + }) + return new Response(stream, { headers: { 'content-type': 'application/x-ndjson' } }) +} diff --git a/src/http/index.ts b/src/http/index.ts new file mode 100644 index 0000000..13ba7ec --- /dev/null +++ b/src/http/index.ts @@ -0,0 +1,2 @@ +export * from './handler.js' +export * from './client.js' diff --git a/src/index.ts b/src/index.ts index caf70c3..9400bff 100644 --- a/src/index.ts +++ b/src/index.ts @@ -5,3 +5,4 @@ export * from './kv/index.js' export * from './types/schema.js' export * from './events/index.js' export * from './documents/index.js' +export * from './http/index.js' diff --git a/test/http/http.test.ts b/test/http/http.test.ts new file mode 100644 index 0000000..274efaf --- /dev/null +++ b/test/http/http.test.ts @@ -0,0 +1,92 @@ +import { afterAll, beforeAll, describe, expect, it } from 'vitest' +import { InMemoryKVDriver } from '../../src/kv/memory.js' +import type { KVDriver } from '../../src/kv/driver.js' +import { KVStorage } from '../../src/kv/node.js' +import { KVEventStorage } from '../../src/events/kv.js' +import { EventSourcedDocumentStorage } from '../../src/documents/event-sourced.js' +import { createKVHttpHandler } from '../../src/http/handler.js' +import { HttpKVDriver } from '../../src/http/client.js' +import { runKVDriverConformance } from '../../src/testing/kv-conformance.js' +import { collect, CountingDriver, MinimalDriver } from '../helpers.js' +import { serve } from './node-server.js' + +/** A client wired straight to a handler, no socket. */ +function inProcess(server: KVDriver = new InMemoryKVDriver(), caps?: ConstructorParameters[0]['capabilities']) { + const handler = createKVHttpHandler(server, { basePath: '/kv' }) + return new HttpKVDriver({ url: 'http://test/kv', capabilities: caps, fetch: async (input, init) => handler(new Request(input, init)) }) +} + +runKVDriverConformance('HttpKVDriver (in-process)', () => inProcess()) +runKVDriverConformance('HttpKVDriver (minimal server driver, discovered capabilities)', () => { + const handler = createKVHttpHandler(new MinimalDriver()) + return HttpKVDriver.connect({ url: 'http://test', fetch: async (input, init) => handler(new Request(input, init)) }) +}) + +describe('over a real socket', () => { + let server: Awaited> + const backends: InMemoryKVDriver[] = [] + beforeAll(async () => { + // One server; each driver gets a fresh backend selected by path. + server = await serve(async (req) => { + const m = /^\/db(\d+)/.exec(new URL(req.url).pathname) + const backend = backends[Number(m?.[1])] + if (!backend) return new Response('no db', { status: 404 }) + return createKVHttpHandler(backend, { basePath: `/db${m![1]}` })(req) + }) + }) + afterAll(() => server.close()) + + const fresh = () => { + backends.push(new InMemoryKVDriver()) + return HttpKVDriver.connect({ url: `${server.url}/db${backends.length - 1}` }) + } + + runKVDriverConformance('HttpKVDriver (node:http)', fresh) + + it('cancels a server-side scan when the client stops early', async () => { + const d = await fresh() + await d.setMany(Array.from({ length: 5000 }, (_, i) => [`k${String(i).padStart(5, '0')}`, i] as [string, number])) + const first: string[] = [] + for await (const [k] of d.scan({})) { + first.push(k) + if (first.length === 3) break + } + expect(first).toEqual(['k00000', 'k00001', 'k00002']) + }) + + it('runs the whole stack (KV indexes, events, documents) over HTTP', async () => { + const kv = new KVStorage(await fresh(), { indexes: { users: { byAge: (u: { age: number }) => u.age } } }) + await kv.child('users').set('1', { age: 30 }) + expect(await collect(kv.child('users').findBy('byAge', { gte: 18 }))).toEqual([['1', { age: 30 }]]) + const docs = new EventSourcedDocumentStorage(new KVEventStorage(kv), { indexes: { people: [{ fields: ['email'], unique: true }] } }) + const id = await docs.child('people').insert({ email: 'a@x', n: 1 }) + await docs.child('people').update(id, { $inc: { n: 1 } }) + await expect(docs.child('people').insert({ email: 'a@x' })).rejects.toThrow(/Unique/) + expect(await docs.child('people').findOne({ email: 'a@x' })).toEqual({ key: id, data: { email: 'a@x', n: 2 } }) + }) +}) + +describe('handler', () => { + it('reports capabilities and rejects unsupported operations with 501', async () => { + const handler = createKVHttpHandler(new MinimalDriver()) + const caps = await (await handler(new Request('http://x/capabilities'))).json() + expect(caps).toEqual({ setIfAbsent: false, batch: false, deleteRange: true }) + const res = await handler(new Request('http://x/batch', { method: 'POST', body: JSON.stringify({ ops: [] }) })) + expect(res.status).toBe(501) + }) + + it('maps driver errors to 500 and unknown routes to 404', async () => { + const broken = new CountingDriver(new InMemoryKVDriver()) + broken.getMany = () => Promise.reject(new Error('boom')) + const handler = createKVHttpHandler(broken) + expect((await handler(new Request('http://x/getMany', { method: 'POST', body: '{"keys":["a"]}' }))).status).toBe(500) + expect((await handler(new Request('http://x/nope'))).status).toBe(404) + await expect(inProcess(new MinimalDriver(), { batch: true }).batch!([])).rejects.toThrow(/HTTP 501/) + }) + + it('distinguishes null values from missing keys', async () => { + const d = inProcess() + await d.setMany([['n', null]]) + expect(await d.getMany(['n', 'missing'])).toEqual([null, undefined]) + }) +}) diff --git a/test/http/node-server.ts b/test/http/node-server.ts new file mode 100644 index 0000000..05b1551 --- /dev/null +++ b/test/http/node-server.ts @@ -0,0 +1,36 @@ +import { createServer, type IncomingHttpHeaders } from 'node:http' +import type { AddressInfo } from 'node:net' + +/** Serves a `Request → Response` handler over a real socket (test helper). */ +export async function serve(handler: (req: Request) => Promise): Promise<{ url: string; close: () => Promise }> { + const server = createServer(async (req, res) => { + const chunks: Buffer[] = [] + for await (const c of req) chunks.push(c as Buffer) + const headers = Object.entries(req.headers as IncomingHttpHeaders).flatMap(([k, v]) => (v === undefined ? [] : [[k, String(v)] as [string, string]])) + const request = new Request(`http://localhost${req.url}`, { + method: req.method, + headers, + body: req.method === 'GET' || req.method === 'HEAD' ? undefined : Buffer.concat(chunks), + }) + const response = await handler(request) + res.writeHead(response.status, Object.fromEntries(response.headers)) + if (!response.body) return void res.end() + const reader = response.body.getReader() + res.on('close', () => void reader.cancel().catch(() => {})) + try { + for (;;) { + const { done, value } = await reader.read() + if (done) break + res.write(value) + } + } finally { + res.end() + } + }) + await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve)) + const { port } = server.address() as AddressInfo + return { + url: `http://127.0.0.1:${port}`, + close: () => new Promise((resolve) => (server.closeAllConnections(), server.close(() => resolve()))), + } +}