M8: HTTP driver (framework-agnostic handler + HttpKVDriver client) passing conformance end-to-end

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

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.
+
 <!-- decisions appended per milestone -->
 
 ## 13. Benchmark baselines

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<HttpCapabilities>
+  codec?: Codec
+  headers?: Record<string, string>
+  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<string, string>
+  readonly #fetch: typeof fetch
+  readonly #nativeDeleteRange: boolean
+
+  setIfAbsent?: (entries: [string, unknown][]) => Promise<boolean[]>
+  batch?: (ops: BatchOp[]) => Promise<void>
+
+  /** Creates a client after asking the server which optional capabilities it has. */
+  static async connect(opts: Omit<HttpKVDriverOptions, 'capabilities'>): Promise<HttpKVDriver> {
+    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<unknown[]> {
+    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<void> {
+    if (entries.length) await this.#request('POST', '/setMany', { entries })
+  }
+
+  async deleteMany(keys: string[]): Promise<void> {
+    if (keys.length) await this.#request('POST', '/deleteMany', { keys })
+  }
+
+  override async deleteRange(gte: string, lt: string): Promise<void> {
+    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<unknown> {
+    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<Error> {
+  let detail = ''
+  try {
+    detail = await res.text()
+  } catch {}
+  return new Error(`HTTP ${res.status}: ${detail}`)
+}

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&lt&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<Response> {
+  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 <T>() => 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<Uint8Array>({
+    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' } })
+}

src/http/index.ts

@@ -0,0 +1,2 @@
+export * from './handler.js'
+export * from './client.js'

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'

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<typeof HttpKVDriver>[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<ReturnType<typeof serve>>
+  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])
+  })
+})

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<Response>): Promise<{ url: string; close: () => Promise<void> }> {
+  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<void>((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()))),
+  }
+}