UniKVS

How to use UniKVS, the main package providing the builder and KVS client.

Overview

unikvs is the main client package. Main client Combine middleware and storage with the builder (Builder / UniKvsBuilder), then read and write with the resulting client (UniKvs).

npm install unikvs

pnpm add unikvs

yarn add unikvs

bun add unikvs

nub add unikvs

aube add unikvs

Builder

Create a builder with new Builder(). You can pass the mapping directly as a type parameter, or let it be inferred from a Valibot schema via the schema option. The latter also enables runtime validation of input and output values.

import * as unikvs from "unikvs";
import * as v from "valibot";

const kvs = new unikvs.Builder({
  schema: {
    foo: unikvs.PlainValue(v.instance(Uint8Array)),
  },
})
  .persist(storage)
  .build();

import * as unikvs from "unikvs";
import * as v from "valibot";

const kvs = new unikvs.Builder({
  schema: [
    [v.pipe(v.string(), v.regex(/^msg-.+/)), unikvs.PlainValue(v.string())],
    [v.pipe(v.string(), v.regex(/^img-.+/)), unikvs.StreamValue(v.instance(Uint8Array))],
  ],
})
  .persist(storage)
  .build();

import * as unikvs from "unikvs";

const kvs = new unikvs.Builder<{
  foo: unikvs.Value<Uint8Array>;
}>()
  .persist(storage)
  .build();

Configure in the following order.

Each storage uses a snapshot of the pipeline taken when persist() was called.

Writes run against all storages in parallel; reads decode by applying the preceding stages in reverse order, starting from the storage where the key was found.

Branching

Use fork() to add storage on a branched pipeline. Create branches with new Branch(). Additions on a branch do not leak to the parent pipeline. Calling fork() with a branch that reaches no storage throws InvalidForkError.

The following example registers different storages on a branch and on the parent.

import * as unikvs from "unikvs";

const kvs = new unikvs.Builder({
  vars: {},
  schema: {},
})
  .use(new Middleware())
  .fork(new unikvs.Branch().use(new Middleware()).persist(new Storage()))
  .persist(new Storage())
  .build();

Middleware at the beginning

Pass an array of middleware to beforeAll in build() to insert it before every configured middleware. It applies to the main pipeline and every forked branch. Use it when you want to apply a common transformation at the position closest to the caller’s data.

const kvs = new unikvs.Builder()
  .use(new Middleware())
  .persist(storage)
  .build({ beforeAll: [new Middleware()] });

The first element is closest to the caller and the last element is closest to the following middleware.

Middleware immediately before storage

Pass an array of middleware to beforeStorage in build() to insert it immediately before every configured storage. It applies to the main pipeline and every forked branch. Use it when you want to apply a common transformation to multiple storages at the final stage.

const kvs = new unikvs.Builder().persist(storage).build({ beforeStorage: [new Middleware()] });

The first element is closer to the existing pipeline and the last element is closer to the storage.

Validating the configuration

isValid() reports whether at least one storage is registered and the pipeline ends with a storage. A branch that reaches no storage throws InvalidForkError at fork() time, so branch validity is guaranteed when the fork is added.

const withStorage = new unikvs.Builder().persist(mainStorage);
withStorage.isValid(); // true

const withoutStorage = new unikvs.Builder();
withoutStorage.isValid(); // false

Client operations

The created client provides the following operations. Call open() to initialize and close() when done (await using also auto-closes).

All operations support cancellation via AbortSignal and per-operation runtime variables (vars). Arguments can be passed either as an options object or positionally.

const ac = new AbortController();
setTimeout(() => ac.abort(), 1000);

await kvs.set("foo", new Uint8Array([0, 1, 2]), {
  signal: ac.signal,
  vars: { traceId: "abc-123" },
});

const bytes = await kvs.get("foo", { signal: ac.signal });

await kvs.copy("foo", "backup");
await kvs.move({ key: "backup", destinationKey: "archive" });

copy and move take the source key and the destination destinationKey. An existing destination value is overwritten. move removes the source only after the destination write succeeds. If the source does not exist, they throw KeyNotFoundError. If both keys are the same, they only verify that the source exists and do not delete it.

With multiple storages, the operation is applied to every storage where the source exists. Middleware mapKey is applied to both the source and destination keys. When a storage does not implement copy or move, the operation is reproduced with read, write, and delete.

Value types

PlainValue, StreamValue, and Value are both types and functions. As types they restrict which methods each key supports; as functions they infer types from Valibot schemas.

Value<T> is sugar for both, equivalent to PlainValue<T> | StreamValue<T>.

import * as unikvs from "unikvs";

const kvs = new unikvs.Builder<{
  message: unikvs.PlainValue<string>;
  logs: unikvs.StreamValue<Uint8Array>;
}>()
  .persist(storage)
  .build();

await kvs.open();

await kvs.set("message", "hello");
const msg = await kvs.get("message");
// msg is a string

await kvs.set("logs", new Uint8Array([0x01]));
const valueStream = await kvs.stream("logs");

import * as unikvs from "unikvs";
import * as v from "valibot";

const kvs = new unikvs.Builder({
  schema: {
    message: unikvs.PlainValue(v.string()),
    logs: unikvs.StreamValue(v.instance(Uint8Array)),
  },
})
  .persist(storage)
  .build();

await kvs.open();

await kvs.set("message", "hello");
const msg = await kvs.get("message");
// msg is a string

Disallowed combinations are type errors. With schema, they also throw InvalidInputError at runtime.

// Type error: stream() cannot be used on a PlainValue key
kvs.stream("message");

// Type error: get() cannot be used on a StreamValue key
kvs.get("logs");

Schema validation

When schema is given, input and output values are validated with Valibot schemas.

import * as unikvs from "unikvs";
import * as v from "valibot";

const kvs = new unikvs.Builder({
  schema: {
    message: unikvs.PlainValue(v.pipe(v.string(), v.minLength(1))),
  },
})
  .persist(storage)
  .build();

await kvs.open();

await kvs.set("message", "hello");
await kvs.get("message"); // "hello"

// Fails input validation and throws InvalidInputError
await kvs.set("message", "");

ValueStream

stream() returns a ValueStream<T>, a wrapper around ReadableStream. It supports for await...of and auto-disposal with await using.

const reader = (await kvs.stream("logs")).getReader();
while (true) {
  const { done, value } = await reader.read();
  if (done) {
    break;
  }
  console.log(value); // Uint8Array
}

for await (const chunk of await kvs.stream("logs")) {
  console.log(chunk); // Uint8Array
}

await using valueStream = await kvs.stream("logs");
const reader = valueStream.getReader();
const { value } = await reader.read();
console.log(value);

Variables and cancellation

Pass per-operation settings with vars.

const kvs = new unikvs.Builder<{ foo: unikvs.Value<Uint8Array> }>().persist(storage).build();

await kvs.open();

// Pass vars per operation
await kvs.set("foo", new Uint8Array([1]), {
  vars: { region: "us-east-1" },
});

// Array form also works
await kvs.get("foo", {
  vars: [["region", "us-east-1"]],
});

Cancel with an AbortSignal; an aborted operation fails with the abort reason.

const ac = new AbortController();
setTimeout(() => ac.abort(), 1000);

await kvs.set("foo", new Uint8Array([1]), { signal: ac.signal });

Base vars

Set base vars with new Builder({ vars }). They are merged with each operation’s vars, with the per-operation side winning. You can pass a Record, Map, entry array, or Variables. Malformed input throws InvalidInputError.

const kvs = new unikvs.Builder<{ foo: unikvs.Value<Uint8Array> }>({
  vars: { region: "us-east-1" },
})
  .persist(storage)
  .build();

await kvs.open();

// Merged with per-operation vars. The per-operation `region` wins
await kvs.set("foo", new Uint8Array([1]), {
  vars: { region: "eu-west-1", traceId: "abc-123" },
});
// Runtime variables are { region: "eu-west-1", traceId: "abc-123" }

build() snapshots the base vars at call time. Builders are reusable; later builder changes do not affect already created clients.

Both the builder (Builder) and the client (UniKvs) have same-named methods to change base vars. Each returns itself for chaining. Client-side changes take effect from subsequent operations.

setVars

Replace base vars entirely.

builder.setVars({ region: "us-east-1" });
kvs.setVars({ region: "us-east-1" });

assignVars

Overwrite base vars with the given vars. Keys in the argument win.

// When base vars are { a: 1, b: 2 }
builder.assignVars({ b: 20, c: 3 });
// Base vars become { a: 1, b: 20, c: 3 }

defaultsVars

Fill defaults: the current base vars win over the given vars.

// When base vars are { a: 1, b: 2 }
builder.defaultsVars({ b: 20, c: 3 });
// Base vars become { a: 1, b: 2, c: 3 }

Read repair

get and stream accept repair: true to write values found in later storage back to earlier storage (default false). A denied write-back still succeeds the read by default and leaves a warning log. Pass throwOnRepairDenied: true to fail the read instead (default false). Control write-back eligibility with @unikvs/allow-repair; on denial, meta.names holds the display names of the write-back targets. See using multiple backends for details.

const bytes = await kvs.get("foo", { repair: true });

Errors

import * as unikvs from "unikvs";

try {
  const bytes = await kvs.get("missing-key");
} catch (ex) {
  if (ex instanceof KeyNotFoundError) {
    console.log("Key not found");
  } else {
    throw ex;
  }
}

For error classification and handling, see handling errors.

Examples

Minimal example using @unikvs/memory.

import { Memory } from "@unikvs/memory";
import * as unikvs from "unikvs";

const kvs = new unikvs.Builder<{
  foo: unikvs.Value<Uint8Array>;
}>()
  .persist(new Memory())
  .build();

await kvs.open();

await kvs.set("foo", new Uint8Array([0, 1, 2]));

const bytes = await kvs.get("foo");
console.log(bytes); // Uint8Array(3) [0, 1, 2]

console.log(await kvs.has("foo")); // true

await kvs.delete("foo");
await kvs.close();

import { Memory } from "@unikvs/memory";
import * as unikvs from "unikvs";
import * as v from "valibot";

const kvs = new unikvs.Builder({
  schema: {
    foo: unikvs.PlainValue(v.instance(Uint8Array)),
  },
})
  .persist(new Memory())
  .build();

await kvs.open();

await kvs.set("foo", new Uint8Array([0, 1, 2]));

const bytes = await kvs.get("foo");
console.log(bytes); // Uint8Array(3) [0, 1, 2]

await kvs.close();