UniKVS

Explains how to use the main package UniKVS, which provides the configuration builder and KVS client.

Overview

unikvs is the main client package. Main Client Assemble transformers and storages with the configuration builder (UniKvs.config() / UniKvsConfig), then read and write with the generated client (UniKvs).

npm install unikvs

pnpm add unikvs

yarn add unikvs

bun add unikvs

nub add unikvs

aube add unikvs

Configuration Builder

UniKvs.config() creates a builder. You can specify the mapping directly with a type parameter, or let it be inferred from Valibot schemas via the schema option. The latter also enables runtime validation of input and output values.

import { PlainValue, UniKvs } from "unikvs";
import * as v from "valibot";

const kvs = UniKvs.config({
  schema: {
    foo: PlainValue(v.instance(Uint8Array)),
  },
})
  .appendStorage(storage)
  .create();

import { PlainValue, StreamValue, UniKvs } from "unikvs";
import * as v from "valibot";

const kvs = UniKvs.config({
  schema: [
    [v.pipe(v.string(), v.regex(/^msg-.+/)), PlainValue(v.string())],
    [v.pipe(v.string(), v.regex(/^img-.+/)), StreamValue(v.instance(Uint8Array))],
  ],
})
  .appendStorage(storage)
  .create();

import { UniKvs, type Value } from "unikvs";

const kvs = UniKvs.config<{
  foo: Value<Uint8Array>;
}>()
  .appendStorage(storage)
  .create();

Configure in the following order.

Each storage uses the transformers added up to its registration.

Writes go to all storages in parallel. Reads decode by applying the upstream pipeline of the found storage in reverse.

Client Operations

The created client provides the following operations. Initialize it with open() and close it with close() when you are done (await using also works).

All operations support AbortSignal cancellation and passing runtime variables (vars). Arguments can be specified either as an options object or as positional arguments.

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 });

Value Types

PlainValue, StreamValue, and Value are both types and functions. Used as types, they restrict the methods available per key; used as functions, they infer types from Valibot schemas.

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

import { UniKvs, type PlainValue, type StreamValue } from "unikvs";

const kvs = UniKvs.config<{
  message: PlainValue<string>;
  logs: StreamValue<Uint8Array>;
}>()
  .appendStorage(storage)
  .create();

await kvs.open();

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

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

import { PlainValue, StreamValue, UniKvs } from "unikvs";
import * as v from "valibot";

const kvs = UniKvs.config({
  schema: {
    message: PlainValue(v.string()),
    logs: StreamValue(v.instance(Uint8Array)),
  },
})
  .appendStorage(storage)
  .create();

await kvs.open();

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

Combinations that are not allowed 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

Specifying schema validates input and output values against Valibot schemas.

import { PlainValue, UniKvs } from "unikvs";
import * as v from "valibot";

const kvs = UniKvs.config({
  schema: {
    message: PlainValue(v.pipe(v.string(), v.minLength(1))),
  },
})
  .appendStorage(storage)
  .create();

await kvs.open();

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

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

ValueStream

The return value of stream() is ValueStream<T>, a wrapper around ReadableStream. It supports for await...of and automatic 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 (const valueStream = await kvs.stream("logs")) {
  const reader = valueStream.getReader();
  const { value } = await reader.read();
  console.log(value);
}

Variables and Cancellation

vars passes per-operation settings. Values are merged over those set with setVariables().

const kvs = UniKvs.config<{ foo: Value<Uint8Array> }>()
  .setVariables({ region: "ap-northeast-1" })
  .appendStorage(storage)
  .create();

await kvs.open();

// Per-operation vars overwrite the same keys
await kvs.set("foo", new Uint8Array([1]), {
  vars: { region: "us-east-1" },
});

// You can also use the array form
await kvs.get("foo", {
  vars: [["region", "us-east-1"]],
});

Cancellation is done with an AbortSignal. An interrupted operation fails with the abort reason.

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

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

Write-back on read

get and stream accept repair: true to write a value found in a later storage back to earlier storages while decoding (disabled by default). Because the intermediate decoded value already matches each storage’s format, no re-encoding is needed. A failed write-back does not fail the read. See the multi-storage guide for details.

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

Errors

import { KeyNotFoundError } from "unikvs";

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

See the cross-cutting docs for error classification and handling.

Examples

A minimal example using @unikvs/memory.

import { Memory } from "@unikvs/memory";
import { UniKvs, type Value } from "unikvs";

const kvs = UniKvs.config<{
  foo: Value<Uint8Array>;
}>()
  .appendStorage(new Memory())
  .create();

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 { PlainValue, UniKvs } from "unikvs";
import * as v from "valibot";

const kvs = UniKvs.config({
  schema: {
    foo: PlainValue(v.instance(Uint8Array)),
  },
})
  .appendStorage(new Memory())
  .create();

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();