Authenticated, scoped sync: signed requests and per-path subscriptions

- RequestAuth: clients sign `${timestamp}:${path}:${body}` (as in
  sync-server-example); headers carry public key, timestamp, signature and
  algorithm. Pluggable signers/verifiers; Ed25519 via WebCrypto built in.
  Timestamp window and replay protection.
- KVStorageHttpServer requires auth ('none' must be explicit). With auth, every
  key, scan range and subscription path must be under the caller's public key.
- POST /events with { paths }: subscribers only receive (and on resume only
  replay) changes under their authorized paths.
- KVStorageHttp: signer option; subscribe(path)/unsubscribe(path)/close()
  replace watch(); 401/403 reject instead of retrying.
- KVStorageComposite: prefix option scopes sync, cache and subscription.

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

src/index.ts

@@ -16,6 +16,8 @@ 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/hex.js'
+export * from './utils/request-auth.js'
 export * from './utils/json.js'
 export * from './utils/keys.js'
 export * from './utils/schema.js'

src/kv-storage/kv-storage-composite.ts

@@ -1,13 +1,20 @@
 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 { Keys } from '../utils/keys.js'
+import type { AnySpec, ChildOf, KeyOf, 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
+  /**
+   * Only this key and its subtree are synced, cached and subscribed to (e.g. the user's public
+   * key). Keys outside it are rejected. Default `''`: the whole source (only for sources the
+   * client may read entirely).
+   */
+  prefix?: string
   /**
    * 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.
@@ -19,8 +26,8 @@ export interface KVStorageCompositeOptions {
 
 /** Implemented by sources that can push remote changes (e.g. `KVStorageHttp`). */
 interface WatchableSource {
-  watch(): Promise<void>
-  unwatch(): Promise<void>
+  subscribe(path: string): Promise<void>
+  unsubscribe(path: string): Promise<void>
   lastEventId: string | undefined
 }
 
@@ -38,16 +45,22 @@ const OUTBOX_PREFIX = '~q/o/'
 const OUTBOX_END = '~q/o0'
 const SYNC_BATCH = 500
 
+// Types only: children of a KVStorageComposite are KVStorageComposite nodes too (child() creates them with
+// Object.create, so they share this class's prototype).
+export interface KVStorageComposite<S extends Spec = AnySpec> {
+  child<K extends KeyOf<S>>(key: K): KVStorageComposite<ChildOf<S, K>>
+}
+
 /**
  * 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
+ * - **Staying in sync:** remote changes pushed by the source (`subscribe(prefix)`, 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.
+ *   resume (`resync`). Without `subscribe()`, 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
@@ -59,6 +72,8 @@ const SYNC_BATCH = 500
  */
 export class KVStorageComposite<S extends Spec = AnySpec> extends BaseKVStorage<S> {
   private readonly source: BaseKVStorage
+  /** Synced key range: `[prefix, prefix + '0')` is the prefix's own key and its subtree. */
+  private readonly scope: { prefix: string; gte?: string; lt?: string }
   private readonly replicas: BaseKVStorage[]
   /** Persistent replica: seeds the others and holds the outbox and metadata. */
   private readonly persistent: BaseKVStorage
@@ -102,6 +117,8 @@ export class KVStorageComposite<S extends Spec = AnySpec> extends BaseKVStorage<
     super()
     if (options.replicas.length === 0) throw new TypeError('KVStorageComposite needs at least one replica')
     this.source = options.source
+    const prefix = options.prefix ?? ''
+    this.scope = prefix === '' ? { prefix } : { prefix, gte: prefix, lt: prefix + Keys.afterSeparator }
     this.replicas = options.replicas
     this.persistent = options.replicas.at(-1)!
     this.backoff = { baseDelay: 1000, maxDelay: 30_000, ...options.retry }
@@ -134,21 +151,27 @@ export class KVStorageComposite<S extends Spec = AnySpec> extends BaseKVStorage<
     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()
+    if (this.watchable) await (this.source as unknown as WatchableSource).unsubscribe(this.scope.prefix)
   }
 
   async readMany(keys: string[]): Promise<unknown[]> {
+    this.checkScope(keys)
     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)
+    // Narrow the range to the synced scope (and, against an authenticated server, to what it allows).
+    const gte = this.scope.gte !== undefined && (options.gte === undefined || options.gte < this.scope.gte) ? this.scope.gte : options.gte
+    const lt = this.scope.lt !== undefined && (options.lt === undefined || options.lt > this.scope.lt) ? this.scope.lt : options.lt
+    if (gte !== undefined && lt !== undefined && gte >= lt) return
+    const scoped = { ...options, gte, lt }
+    if (!this.state.hasSnapshot) return yield* this.source.scan(scoped)
     // 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 })) {
+    for await (const entry of this.replicas[0]!.scan({ ...scoped, limit: undefined })) {
       if (entry[0] >= INTERNAL_PREFIX && entry[0] < INTERNAL_END) continue
       yield entry
       if (--remaining <= 0) return
@@ -156,16 +179,19 @@ export class KVStorageComposite<S extends Spec = AnySpec> extends BaseKVStorage<
   }
 
   protected async applyWrite(entries: [string, unknown][]): Promise<void> {
+    this.checkScope(entries.map(([key]) => key))
     await this.start()
     await this.writeThrough({ type: 'write', entries })
   }
 
   protected async applyRemove(keys: string[]): Promise<void> {
+    this.checkScope(keys)
     await this.start()
     await this.writeThrough({ type: 'remove', keys })
   }
 
   protected async applyWriteIfAbsent(entries: [string, unknown][]): Promise<boolean[]> {
+    this.checkScope(entries.map(([key]) => key))
     await this.start()
     const written = await this.source.writeIfAbsent(entries)
     const changes = entries.filter((_, i) => written[i]).map(([key, value]) => ({ key, value }))
@@ -176,8 +202,17 @@ export class KVStorageComposite<S extends Spec = AnySpec> extends BaseKVStorage<
 
   // ------------------------------------------------------------------ start
 
+  private inScope(key: string): boolean {
+    const { prefix } = this.scope
+    return prefix === '' || key === prefix || key.startsWith(prefix + Keys.separator)
+  }
+
+  private checkScope(keys: string[]): void {
+    for (const key of keys) if (!this.inScope(key)) throw new RangeError(`"${key}" is outside this composite's prefix "${this.scope.prefix}"`)
+  }
+
   private get watchable(): boolean {
-    return typeof (this.source as unknown as Partial<WatchableSource>).watch === 'function'
+    return typeof (this.source as unknown as Partial<WatchableSource>).subscribe === 'function'
   }
 
   private async open(): Promise<void> {
@@ -201,8 +236,9 @@ export class KVStorageComposite<S extends Spec = AnySpec> extends BaseKVStorage<
       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()))
+      const watching = source.subscribe(this.scope.prefix)
+      if (!meta?.lastEventId) void watching.then(() => this.enqueueApply(() => this.fullSync())).catch((error) => console.error(error))
+      else void watching.catch((error) => console.error(error))
     } else {
       this.enqueueApply(() => this.fullSync())
     }
@@ -238,7 +274,9 @@ export class KVStorageComposite<S extends Spec = AnySpec> extends BaseKVStorage<
   }
 
   private async applyRemote(event: KVChangeEvent): Promise<void> {
-    const candidates = event.changes.filter(({ key }) => !this.consumeEcho(key) && !this.pending.has(key) && !key.startsWith(INTERNAL_PREFIX))
+    const candidates = event.changes.filter(
+      ({ key }) => this.inScope(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]))
@@ -276,8 +314,9 @@ export class KVStorageComposite<S extends Spec = AnySpec> extends BaseKVStorage<
 
   /** Changes that turn the persistent replica into the source (both scanned in key order). */
   private async *diff(): AsyncGenerator<KVChange> {
-    const remote = this.source.scan({ values: true })[Symbol.asyncIterator]()
-    const local = this.persistent.scan({ values: true })[Symbol.asyncIterator]()
+    const range = { gte: this.scope.gte, lt: this.scope.lt, values: true }
+    const remote = this.source.scan(range)[Symbol.asyncIterator]()
+    const local = this.persistent.scan(range)[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
     }

src/kv-storage/kv-storage-http-server.ts

@@ -1,7 +1,16 @@
+import { AuthError } from '../utils/errors.js'
 import { ExtendedJson } from '../utils/extended-json.js'
+import { Keys } from '../utils/keys.js'
+import { RequestAuth, type RequestAuthOptions, type SignatureVerifier } from '../utils/request-auth.js'
 import type { BaseKVStorage, KVChange, ScanOptions } from './base-kv-storage.js'
 
 export interface KVStorageHttpServerOptions {
+  /**
+   * `{ verifiers }`: requests must be signed (see `RequestAuth`), and a caller may only touch keys
+   * under its own public key (`<publicKey>` and `<publicKey>/...`).
+   * `'none'`: no authentication, everything is open; for local development and tests only.
+   */
+  auth: 'none' | ({ verifiers: SignatureVerifier[] } & RequestAuthOptions)
   /** Path prefix to strip, e.g. `/kv`. Default `''`. */
   basePath?: string
   /** Recent change batches kept for clients resuming with `Last-Event-ID`. Default 10 000. */
@@ -15,32 +24,45 @@ interface HistoryEntry {
   changes: KVChange[]
 }
 
+interface Subscriber {
+  /** Keys whose own value and subtree this subscriber receives (`''` = everything). */
+  paths: string[]
+  send: (frame: string) => void
+}
+
+type RequestBody = { keys: string[]; entries: [string, unknown][]; paths: string[] } & ScanOptions
+
 /**
  * Serves any KV storage over HTTP for `KVStorageHttp`. Framework-agnostic: `handle(Request)`
- * returns a `Response`.
+ * returns a `Response`. Every route is `POST` with an Extended JSON body:
  *
- * - `POST /<primitive>` with an Extended JSON body (`scan` streams NDJSON).
- * - `GET /events`: Server-Sent Events of every change to the storage. Event ids are
- *   `<epoch>-<seq>` (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.
+ * - `/readMany`, `/writeMany`, `/removeMany`, `/writeIfAbsent`, `/scan` (streams NDJSON).
+ * - `/events` with `{ paths }`: Server-Sent Events of changes to those keys and their subtrees.
+ *   Event ids are `<epoch>-<seq>` (epoch = this server instance). A client sending
+ *   `Last-Event-ID` gets the matching changes it missed if they are still in the history,
+ *   otherwise a `resync` event; either way a `ready` event ends the catch-up.
+ *
+ * With authentication, each request is checked first (401), then every key, scan range and
+ * subscription path must be the caller's public key or under it (403).
  *
  * 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 auth: RequestAuth | undefined
   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>()
+  private readonly subscribers = new Set<Subscriber>()
 
   constructor(
     private readonly storage: BaseKVStorage,
-    options: KVStorageHttpServerOptions = {},
+    options: KVStorageHttpServerOptions,
   ) {
+    this.auth = options.auth === 'none' ? undefined : new RequestAuth(options.auth.verifiers, options.auth)
     this.basePath = options.basePath ?? ''
     this.historySize = options.historySize ?? 10_000
     this.keepAliveMs = options.keepAliveMs ?? 15_000
@@ -49,58 +71,99 @@ export class KVStorageHttpServer {
 
   async handle(request: Request): Promise<Response> {
     const path = new URL(request.url).pathname
-    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')
+    if (request.method !== 'POST' || !path.startsWith(this.basePath)) return KVStorageHttpServer.error(404, 'Not found')
     try {
-      const body = ExtendedJson.parse(await request.text()) as { keys: string[]; entries: [string, unknown][] } & ScanOptions
-      switch (route) {
-        case 'POST /readMany': {
+      const text = await request.text()
+      const owner = this.auth ? await this.auth.verify(request.headers, path, text) : undefined
+      const body = ExtendedJson.parse(text) as RequestBody
+      switch (path.slice(this.basePath.length)) {
+        case '/readMany': {
+          this.authorize(owner, body.keys)
           const values = await this.storage.readMany(body.keys)
           const missing = values.flatMap((v, i) => (v === undefined ? [i] : []))
           return KVStorageHttpServer.json({ values: values.map((v) => v ?? null), missing })
         }
-        case 'POST /writeMany':
+        case '/writeMany':
+          this.authorize(
+            owner,
+            body.entries.map(([key]) => key),
+          )
           await this.storage.writeMany(body.entries)
           return new Response(null, { status: 204 })
-        case 'POST /removeMany':
+        case '/removeMany':
+          this.authorize(owner, body.keys)
           await this.storage.removeMany(body.keys)
           return new Response(null, { status: 204 })
-        case 'POST /writeIfAbsent':
+        case '/writeIfAbsent':
+          this.authorize(
+            owner,
+            body.entries.map(([key]) => key),
+          )
           return KVStorageHttpServer.json(await this.storage.writeIfAbsent(body.entries))
-        case 'POST /scan':
+        case '/scan':
+          this.authorizeRange(owner, body)
           return this.scan(body)
+        case '/events':
+          if (!Array.isArray(body.paths) || body.paths.length === 0) return KVStorageHttpServer.error(400, 'Subscribe to at least one path')
+          this.authorize(owner, body.paths)
+          return this.events(body.paths, request.headers.get('last-event-id'))
         default:
           return KVStorageHttpServer.error(404, 'Not found')
       }
     } catch (error) {
+      if (error instanceof AuthError) return KVStorageHttpServer.error(error.status, error.message)
       return KVStorageHttpServer.error(500, error instanceof Error ? error.message : String(error))
     }
   }
 
+  // ------------------------------------------------------------------ authorization
+
+  /** Every key must be the owner's key or under it (no-op without authentication). */
+  private authorize(owner: string | undefined, keys: string[]): void {
+    if (owner === undefined) return
+    for (const key of keys) {
+      if (typeof key !== 'string' || !KVStorageHttpServer.within(owner, key)) throw new AuthError(403, `No access to "${key}"`)
+    }
+  }
+
+  /** A scan must stay inside `[owner, owner + '0')`: the owner's key and its subtree. */
+  private authorizeRange(owner: string | undefined, options: ScanOptions): void {
+    if (owner === undefined) return
+    const { gte, lt } = options
+    if (gte === undefined || lt === undefined || gte < owner || lt > owner + Keys.afterSeparator) throw new AuthError(403, 'Scan outside your keys')
+  }
+
+  /** Whether `key` is `path` or below it (`''` covers everything). */
+  private static within(path: string, key: string): boolean {
+    return path === '' || key === path || key.startsWith(path + Keys.separator)
+  }
+
+  // ------------------------------------------------------------------ change stream
+
   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)
+    for (const subscriber of this.subscribers) {
+      const frame = this.changeFrame(entry, subscriber.paths)
+      if (frame) subscriber.send(frame)
+    }
   }
 
-  private events(lastEventId: string | null): Response {
+  private events(paths: string[], lastEventId: string | null): Response {
     const encoder = new TextEncoder()
-    let send: ((frame: string) => void) | undefined
+    let subscriber: Subscriber | undefined
     let keepAlive: ReturnType<typeof setInterval> | undefined
     const stream = new ReadableStream<Uint8Array>({
       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)
+        subscriber = { paths, send: (frame) => controller.enqueue(encoder.encode(frame)) }
+        subscriber.send(this.catchUpFrames(paths, lastEventId))
+        this.subscribers.add(subscriber)
+        keepAlive = setInterval(() => subscriber!.send(': keep-alive\n\n'), this.keepAliveMs)
         ;(keepAlive as { unref?: () => void }).unref?.()
       },
       cancel: () => {
-        if (send) this.subscribers.delete(send)
+        if (subscriber) this.subscribers.delete(subscriber)
         clearInterval(keepAlive)
       },
     })
@@ -108,10 +171,10 @@ export class KVStorageHttpServer {
   }
 
   /**
-   * 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).
+   * What a (re)connecting client gets before live changes: the matching changes it missed, or
+   * `resync` when they can't be replayed; then always `ready` (caught up as of that id).
    */
-  private catchUpFrames(lastEventId: string | null): string {
+  private catchUpFrames(paths: string[], lastEventId: string | null): string {
     const ready = this.frame('ready', this.seq, null)
     if (lastEventId === null) return ready
     const [epoch, seqText] = lastEventId.split('-')
@@ -119,14 +182,26 @@ export class KVStorageHttpServer {
     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
+    return (
+      this.history
+        .filter((entry) => entry.seq > seq)
+        .map((entry) => this.changeFrame(entry, paths))
+        .join('') + ready
+    )
+  }
+
+  /** A `change` frame with only the changes under `paths`, or `''` if none match. */
+  private changeFrame(entry: HistoryEntry, paths: string[]): string {
+    const changes = entry.changes.filter(({ key }) => paths.some((path) => KVStorageHttpServer.within(path, key)))
+    return changes.length ? this.frame('change', entry.seq, changes) : ''
   }
 
   private frame(event: string, seq: number, data: unknown): string {
     return `id: ${this.epoch}-${seq}\nevent: ${event}\ndata: ${ExtendedJson.stringify(data)}\n\n`
   }
 
+  // ------------------------------------------------------------------ scan
+
   private scan(options: ScanOptions): Response {
     const iterator = this.storage.scan(options)[Symbol.asyncIterator]()
     const encoder = new TextEncoder()

src/kv-storage/kv-storage-http.ts

@@ -1,39 +1,58 @@
-import { StorageUnavailableError } from '../utils/errors.js'
+import { AuthError, StorageUnavailableError } from '../utils/errors.js'
 import { ExtendedJson } from '../utils/extended-json.js'
-import type { AnySpec, Spec } from '../utils/schema.js'
+import { RequestAuth, type RequestSigner } from '../utils/request-auth.js'
+import type { AnySpec, ChildOf, KeyOf, Spec } from '../utils/schema.js'
+import { ExponentialBackoff } from '../utils/sse-session/exponential-backoff.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
+  /** Signs every request (required by servers with authentication). */
+  signer?: RequestSigner
   headers?: Record<string, string>
   fetch?: (url: string, init: RequestInit) => Promise<Response>
   /** Called when the change stream fails (it keeps reconnecting). Default: ignore. */
-  onWatchError?: (error: Error) => void
+  onSubscriptionError?: (error: Error) => void
+}
+
+// Types only: children of a KVStorageHttp are KVStorageHttp nodes too (child() creates them with
+// Object.create, so they share this class's prototype).
+export interface KVStorageHttp<S extends Spec = AnySpec> {
+  child<K extends KeyOf<S>>(key: K): KVStorageHttp<ChildOf<S, K>>
 }
 
 /**
  * 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).
+ * `subscribe(path)` receives changes under a key (pushed over SSE) as `change` events with
+ * `origin: 'remote'`; `resync` is emitted when the server can't resume from `lastEventId`. With a
+ * `signer`, every request is signed, and an authenticated server only allows keys under the
+ * signer's public key. Values travel as Extended JSON (`bigint` and `Uint8Array` survive).
  */
 export class KVStorageHttp<S extends Spec = AnySpec> extends BaseKVStorage<S> {
   private readonly url: string
+  private readonly signer: RequestSigner | undefined
   private readonly headers: Record<string, string>
   private readonly fetcher: (url: string, init: RequestInit) => Promise<Response>
-  private readonly onWatchError: (error: Error) => void
+  private readonly onSubscriptionError: (error: Error) => void
   /** Mutable state lives here: child nodes share this object (see `child()`). */
-  private readonly state: { session?: SSESession; lastEventId?: string } = {}
+  private readonly state: {
+    paths: Set<string>
+    session?: SSESession
+    lastEventId?: string
+    /** Subscription changes are applied one at a time. */
+    updating: Promise<void>
+  } = { paths: new Set(), updating: Promise.resolve() }
 
   constructor(options: KVStorageHttpOptions) {
     super()
     this.url = options.url.replace(/\/$/, '')
+    this.signer = options.signer
     this.headers = options.headers ?? {}
     this.fetcher = options.fetch ?? ((url, init) => globalThis.fetch(url, init))
-    this.onWatchError = options.onWatchError ?? (() => {})
+    this.onSubscriptionError = options.onSubscriptionError ?? (() => {})
   }
 
   /** Position in the server's change stream; sent as `Last-Event-ID` when (re)connecting. */
@@ -45,6 +64,11 @@ export class KVStorageHttp<S extends Spec = AnySpec> extends BaseKVStorage<S> {
     this.state.lastEventId = id
   }
 
+  /** Keys currently subscribed to. */
+  get subscriptions(): ReadonlySet<string> {
+    return this.state.paths
+  }
+
   async readMany(keys: string[]): Promise<unknown[]> {
     if (keys.length === 0) return []
     const { values, missing } = (await this.post('readMany', { keys })) as { values: unknown[]; missing: number[] }
@@ -75,39 +99,22 @@ export class KVStorageHttp<S extends Spec = AnySpec> extends BaseKVStorage<S> {
   }
 
   /**
-   * 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.
+   * Receives changes to `path` (default: this node's key) and everything under it. Resolves once
+   * the stream is connected and caught up (missed changes replayed, or `resync` emitted). While
+   * the server is unreachable this stays pending, so don't await it on a path that must work offline.
    */
-  async watch(): Promise<void> {
-    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<string, string>), '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
+  subscribe(path: string = this.key): Promise<void> {
+    return this.updateSubscriptions(() => this.state.paths.add(path))
   }
 
-  async unwatch(): Promise<void> {
-    const session = this.state.session
-    this.state.session = undefined
-    await session?.disconnect()
+  /** Stops receiving changes to `path` (default: this node's key). */
+  unsubscribe(path: string = this.key): Promise<void> {
+    return this.updateSubscriptions(() => this.state.paths.delete(path))
+  }
+
+  /** Drops every subscription and closes the stream. */
+  close(): Promise<void> {
+    return this.updateSubscriptions(() => this.state.paths.clear())
   }
 
   protected async applyWrite(entries: [string, unknown][]): Promise<void> {
@@ -122,24 +129,102 @@ export class KVStorageHttp<S extends Spec = AnySpec> extends BaseKVStorage<S> {
     return (await this.post('writeIfAbsent', { entries })) as boolean[]
   }
 
+  // ------------------------------------------------------------------ subscriptions
+
+  /** Applies a subscription change, then reconnects the stream with the new path list. */
+  private updateSubscriptions(change: () => unknown): Promise<void> {
+    const run = async () => {
+      const before = JSON.stringify([...this.state.paths])
+      change()
+      if (JSON.stringify([...this.state.paths]) === before && (this.state.session || this.state.paths.size === 0)) return
+      await this.state.session?.disconnect()
+      this.state.session = undefined
+      if (this.state.paths.size > 0) await this.connect([...this.state.paths])
+    }
+    const result = this.state.updating.then(run)
+    this.state.updating = result.catch(() => {})
+    return result
+  }
+
+  private async connect(paths: string[]): Promise<void> {
+    const url = `${this.url}/events`
+    const body = ExtendedJson.stringify({ paths })
+    const session = new SSESession(url, {
+      method: 'POST',
+      body,
+      fetch: this.fetcher,
+      headers: { ...this.headers, 'content-type': 'application/json' },
+      persistent: true,
+      onError: this.onSubscriptionError,
+      retry: KVStorageHttp.retryUnlessDenied(),
+      // Signed on every (re)connect, so the timestamp is always fresh.
+      onRequest: async (request) => ({
+        ...request,
+        headers: {
+          ...(request.headers as Record<string, string>),
+          ...(this.state.lastEventId ? { 'Last-Event-ID': this.state.lastEventId } : {}),
+          ...(await this.authHeaders(url, body)),
+        },
+      }),
+    })
+    session.on('message', (message) => {
+      if (message.id) this.state.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')
+    try {
+      await session.connect()
+    } catch (error) {
+      this.state.session = undefined
+      throw (error as Error).cause instanceof AuthError ? (error as Error).cause : error
+    }
+    await caughtUp
+  }
+
+  // ------------------------------------------------------------------ requests
+
   private async post(path: string, body: unknown): Promise<unknown> {
     const response = await this.request(path, body)
     return response.status === 204 ? undefined : ExtendedJson.parse(await response.text())
   }
 
-  private async request(path: string, body: unknown): Promise<Response> {
+  private async request(path: string, payload: unknown): Promise<Response> {
+    const url = `${this.url}/${path}`
+    const body = ExtendedJson.stringify(payload)
     let response: Response
     try {
-      response = await this.fetcher(`${this.url}/${path}`, {
+      response = await this.fetcher(url, {
         method: 'POST',
-        headers: { ...this.headers, 'content-type': 'application/json' },
-        body: ExtendedJson.stringify(body),
+        headers: { ...this.headers, 'content-type': 'application/json', ...(await this.authHeaders(url, body)) },
+        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.status === 401 || response.status === 403) throw new AuthError(response.status, await response.text())
     if (!response.ok) throw new Error(`HTTP ${response.status}: ${await response.text()}`)
     return response
   }
+
+  /** Retries connecting with backoff, except when the server denies access (401/403): that won't fix itself. */
+  private static retryUnlessDenied(): { run<T>(fn: () => Promise<T>): Promise<T> } {
+    const backoff = new ExponentialBackoff({ baseDelay: 1000, maxDelay: 10_000, maxAttempts: 0, growthRate: 1.3, jitter: 0.3 })
+    return {
+      run: (fn) =>
+        backoff.run(fn, (error, context) => {
+          const status = /^HTTP error: (40[13])/.exec(error.message)?.[1]
+          if (status) context.abort(new AuthError(Number(status) as 401 | 403, error.message))
+        }),
+    }
+  }
+
+  private async authHeaders(url: string, body: string): Promise<Record<string, string>> {
+    return this.signer ? RequestAuth.headers(this.signer, new URL(url).pathname, body) : {}
+  }
 }

src/utils/errors.ts

@@ -25,3 +25,14 @@ export class StorageUnavailableError extends Error {
     this.name = 'StorageUnavailableError'
   }
 }
+
+/** A request was not authenticated (401) or not allowed for this caller (403). */
+export class AuthError extends Error {
+  constructor(
+    readonly status: 401 | 403,
+    message: string,
+  ) {
+    super(message)
+    this.name = 'AuthError'
+  }
+}

src/utils/extended-json.ts

@@ -1,6 +1,8 @@
 // Adapted from xo-cash-utils `source/extended-json.ts` (branch installable-sse), with local hex
 // encoding instead of @bitauth/libauth.
 
+import { Hex } from './hex.js'
+
 /** Matches a bigint encoded as `<bigint: 123n>`. */
 const BIGINT_PATTERN = /^<bigint: (?<bigint>[+-]?[0-9]+)n>$/u
 
@@ -22,7 +24,7 @@ export class ExtendedJson {
 
   /** `JSON.stringify` replacer. */
   static replacer(this: void, _key: string, value: unknown): unknown {
-    if (value instanceof Uint8Array) return `<uint8array: ${ExtendedJson.toHex(value)}>`
+    if (value instanceof Uint8Array) return `<uint8array: ${Hex.encode(value)}>`
     if (typeof value === 'bigint') return `<bigint: ${value.toString()}n>`
     return value
   }
@@ -33,19 +35,7 @@ export class ExtendedJson {
     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!)
+    if (bytes) return Hex.decode(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
-  }
 }

src/utils/hex.ts

@@ -0,0 +1,16 @@
+/** Lowercase hex encoding of bytes. */
+export class Hex {
+  static encode(bytes: Uint8Array): string {
+    let hex = ''
+    for (const byte of bytes) hex += byte.toString(16).padStart(2, '0')
+    return hex
+  }
+
+  /** Throws on odd length or non-hex characters. */
+  static decode(hex: string): Uint8Array {
+    if (hex.length % 2 !== 0 || !/^[0-9a-fA-F]*$/.test(hex)) throw new TypeError('Invalid hex string')
+    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
+  }
+}

src/utils/request-auth.ts

@@ -0,0 +1,150 @@
+import { AuthError } from './errors.js'
+import { Hex } from './hex.js'
+
+/** Signs requests. Implement this to use another algorithm (e.g. secp256k1 via libauth). */
+export interface RequestSigner {
+  /** Algorithm name sent in `X-Signature-Algorithm`; the server must have a matching verifier. */
+  readonly algorithm: string
+  /** Public key as lowercase hex. Also the owner segment of the caller's keys (`/<publicKey>/...`). */
+  readonly publicKey: string
+  sign(payload: Uint8Array): Promise<Uint8Array>
+}
+
+/** Verifies signatures for one algorithm. */
+export interface SignatureVerifier {
+  readonly algorithm: string
+  verify(publicKey: string, payload: Uint8Array, signature: Uint8Array): Promise<boolean>
+}
+
+export const AUTH_HEADERS = {
+  publicKey: 'X-Public-Key',
+  timestamp: 'X-Timestamp',
+  signature: 'X-Signature',
+  algorithm: 'X-Signature-Algorithm',
+} as const
+
+export interface RequestAuthOptions {
+  /** Accepted difference between the request timestamp and the server clock. Default 60 000 ms. */
+  maxSkewMs?: number
+  now?: () => number
+}
+
+/**
+ * Signed-request authentication. The client signs `${timestamp}:${path}:${body}` (path = URL
+ * pathname, body = the exact request body) and sends the public key, timestamp, signature (hex)
+ * and algorithm as headers. The server checks the timestamp is recent, verifies the signature
+ * with the verifier for that algorithm, and rejects a signature it has already seen within the
+ * timestamp window (replay protection).
+ */
+export class RequestAuth {
+  private readonly verifiers: Map<string, SignatureVerifier>
+  private readonly maxSkewMs: number
+  private readonly now: () => number
+  /** Signatures accepted within the window → when they expire; insertion order ≈ expiry order. */
+  private readonly seen = new Map<string, number>()
+
+  constructor(verifiers: SignatureVerifier[], options: RequestAuthOptions = {}) {
+    if (verifiers.length === 0) throw new TypeError('RequestAuth needs at least one verifier')
+    this.verifiers = new Map(verifiers.map((v) => [v.algorithm, v]))
+    this.maxSkewMs = options.maxSkewMs ?? 60_000
+    this.now = options.now ?? Date.now
+  }
+
+  static payload(timestamp: string, path: string, body: string): Uint8Array {
+    return new TextEncoder().encode(`${timestamp}:${path}:${body}`)
+  }
+
+  /** Auth headers for a request to `path` with `body`. */
+  static async headers(signer: RequestSigner, path: string, body: string, now: number = Date.now()): Promise<Record<string, string>> {
+    const timestamp = String(now)
+    const signature = await signer.sign(RequestAuth.payload(timestamp, path, body))
+    return {
+      [AUTH_HEADERS.publicKey]: signer.publicKey,
+      [AUTH_HEADERS.timestamp]: timestamp,
+      [AUTH_HEADERS.signature]: Hex.encode(signature),
+      [AUTH_HEADERS.algorithm]: signer.algorithm,
+    }
+  }
+
+  /** Returns the caller's public key, or throws `AuthError` (401). */
+  async verify(headers: Headers, path: string, body: string): Promise<string> {
+    const publicKey = headers.get(AUTH_HEADERS.publicKey)
+    const timestamp = headers.get(AUTH_HEADERS.timestamp)
+    const signature = headers.get(AUTH_HEADERS.signature)
+    const algorithm = headers.get(AUTH_HEADERS.algorithm)
+    if (!publicKey || !timestamp || !signature || !algorithm) throw new AuthError(401, 'Missing authentication headers')
+    const verifier = this.verifiers.get(algorithm)
+    if (!verifier) throw new AuthError(401, `Unsupported signature algorithm "${algorithm}"`)
+    if (!/^[0-9a-f]+$/.test(publicKey)) throw new AuthError(401, 'Public key must be lowercase hex')
+    const now = this.now()
+    const time = Number(timestamp)
+    if (!Number.isInteger(time) || Math.abs(now - time) > this.maxSkewMs) throw new AuthError(401, 'Timestamp outside the accepted window')
+    this.forgetExpired(now)
+    if (this.seen.has(signature)) throw new AuthError(401, 'Replayed request')
+    let valid = false
+    try {
+      valid = await verifier.verify(publicKey, RequestAuth.payload(timestamp, path, body), Hex.decode(signature))
+    } catch {
+      valid = false
+    }
+    if (!valid) throw new AuthError(401, 'Invalid signature')
+    this.seen.set(signature, time + this.maxSkewMs)
+    return publicKey
+  }
+
+  private forgetExpired(now: number): void {
+    for (const [signature, expires] of this.seen) {
+      if (expires >= now) break
+      this.seen.delete(signature)
+    }
+  }
+}
+
+/** Ed25519 signer using WebCrypto (Node 22+ and current browsers; no dependencies). */
+export class Ed25519Signer implements RequestSigner {
+  readonly algorithm = 'ed25519'
+
+  private constructor(
+    private readonly privateKey: CryptoKey,
+    readonly publicKey: string,
+  ) {}
+
+  /** A new random key pair. */
+  static async generate(): Promise<Ed25519Signer> {
+    const pair = (await crypto.subtle.generateKey({ name: 'Ed25519' }, true, ['sign', 'verify'])) as CryptoKeyPair
+    return new Ed25519Signer(pair.privateKey, Hex.encode(new Uint8Array(await crypto.subtle.exportKey('raw', pair.publicKey))))
+  }
+
+  /** Restores a key exported with `exportPrivateKey()` (PKCS #8). */
+  static async fromPrivateKey(pkcs8: Uint8Array): Promise<Ed25519Signer> {
+    const privateKey = await crypto.subtle.importKey('pkcs8', pkcs8 as BufferSource, { name: 'Ed25519' }, true, ['sign'])
+    // WebCrypto can't derive the public key directly; the JWK export of a private key includes it (`x`).
+    const jwk = await crypto.subtle.exportKey('jwk', privateKey)
+    const publicKey = Uint8Array.from(atob(jwk.x!.replace(/-/g, '+').replace(/_/g, '/')), (c) => c.charCodeAt(0))
+    return new Ed25519Signer(privateKey, Hex.encode(publicKey))
+  }
+
+  async exportPrivateKey(): Promise<Uint8Array> {
+    return new Uint8Array(await crypto.subtle.exportKey('pkcs8', this.privateKey))
+  }
+
+  async sign(payload: Uint8Array): Promise<Uint8Array> {
+    return new Uint8Array(await crypto.subtle.sign({ name: 'Ed25519' }, this.privateKey, payload as BufferSource))
+  }
+}
+
+/** Ed25519 verifier using WebCrypto. */
+export class Ed25519Verifier implements SignatureVerifier {
+  readonly algorithm = 'ed25519'
+  private readonly keys = new Map<string, Promise<CryptoKey>>()
+
+  async verify(publicKey: string, payload: Uint8Array, signature: Uint8Array): Promise<boolean> {
+    let key = this.keys.get(publicKey)
+    if (!key) {
+      key = crypto.subtle.importKey('raw', Hex.decode(publicKey) as BufferSource, { name: 'Ed25519' }, false, ['verify'])
+      this.keys.set(publicKey, key)
+      if (this.keys.size > 10_000) this.keys.delete(this.keys.keys().next().value!)
+    }
+    return crypto.subtle.verify({ name: 'Ed25519' }, await key, signature as BufferSource, payload as BufferSource)
+  }
+}