---
title: "UniKVS"
description: "Explains how to use the main package UniKVS, which provides the configuration builder and KVS client."
---

## Overview [#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`).

```package-install
npm i unikvs
```

## Configuration Builder [#config-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.

**With schema**

Pass Valibot schemas to `PlainValue()`, `StreamValue()`, and `Value()`. The mapping type is inferred automatically. Install `valibot` separately when using Valibot schemas.

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

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

Use the array form for dynamic keys. List pairs of key and value schemas; the first matching definition wins.

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

**Without schema**

The type parameter `T` is a mapping from keys to values (`{ [key]: PlainValue | StreamValue | Value }`), and decides which operations are available per key at the type level. No runtime validation is performed.

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

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

:::warning[When the schema definition is invalid]
A malformed `schema` throws `InvalidInputError` at the `config()` call.
:::

Configure in the following order.

1. **`setVariables(vars)`**

    Sets the initial runtime variables (optional). Each call replaces the existing values.

2. **`appendTransformer(transformer)`**

    Adds a transformer (optional). You can also add transformers after registering storages.

3. **`appendStorage(storage)`**

    Adds a storage (required, multiple allowed).

4. **`create()`**

    Creates a `UniKvs` client. It throws `MissingStorageError` if no storage is registered.

:::warning[No storage registered]
Call `appendStorage()` at least once before `create()`.
:::

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 [#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).

| Method          | Description                                        |
| --------------- | -------------------------------------------------- |
| `open()`          | Initializes the storages and transformers.           |
| `close()`         | Closes all storages and transformers.                |
| `set(key, value)` | Saves a value under a key.                           |
| `get(key)`        | Gets the value for a key.                            |
| `stream(key)`     | Gets a stream for a key.                             |
| `has(key)`        | Checks whether a key exists.                         |
| `delete(key)`     | Deletes a key.                                       |
| `clear()`         | Deletes all data.                                    |

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

```ts
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 [#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.

| Type               | Write                              | Read            | Stream read                   |
| ------------------ | ---------------------------------- | --------------- | ----------------------------- |
| `PlainValue<T>`  | `set(key, T)`                      | `get(key): T` | Not available.                |
| `StreamValue<T>` | `set(key, T \| ReadableStream<T>)` | Not available. | `stream(key): ValueStream<T>` |
| `Value<T>`       | `set(key, T \| ReadableStream<T>)` | `get(key): T` | `stream(key): ValueStream<T>` |

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

**Specify with types**

```ts
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");
```

**Specify with schemas**

```ts
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.

```ts
// 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 [#schema-validation]

Specifying `schema` validates input and output values against Valibot schemas.

| Operation | What is validated | Error on failure |
| --------- | ----------------- | ---------------- |
| `set` | Validates the input value. | `InvalidInputError` |
| `get` | Validates the decoded value. | `InvalidOutputError` |
| `stream` | Validates each decoded chunk. | `InvalidOutputError` |
| `has` / `delete` | Only in the array form, validates whether the key matches a key schema. | `InvalidInputError` |

```ts
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", "");
```

:::note[Mixing up plain and stream]
Passing a `ReadableStream` to a plain-only key, or a plain value to a stream-only key, throws `InvalidInputError`. Keys matching no key schema in the array form are also `InvalidInputError`.
:::

## ValueStream [#value-stream]

The return value of `stream()` is `ValueStream<T>`, a wrapper around `ReadableStream`. It supports `for await...of` and automatic disposal with `await using`.

**getReader**

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

**for-await**

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

**await using**

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

:::warning[Dispose streams when you are done]
Don't forget automatic disposal via `await using` or calling `dispose()`.
:::

## Variables and Cancellation [#variables-and-cancellation]

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

```ts
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.

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

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

## Write-back on read [#read-repair]

`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.

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

## Errors [#errors]

| Error class | Description |
| --------------------------------- | -------------------------------------------------------------------- |
| `KeyNotFoundError` | The key does not exist in any storage. |
| `UniKvsIsNotOpenError` | Operated before opening. |
| `UniKvsIsOpenError` | Called `open()` while already open. |
| `MissingStorageError` | Called `create()` with no storage registered. |
| `StorageIsNotOpenError` | A storage is not open. |
| `TransformerIsNotOpenError` | A transformer is not open. |
| `PluginOperationAggregateError` | Aggregate error for multiple plugin operations. Chunk validation failures during stream writes are also reported in this form. |
| `InvalidInputError` | The input format is invalid. With `schema`, `set` input validation and key validation failures, as well as malformed `schema` definitions, are also this error. |
| `InvalidOutputError` | The output format is invalid. With `schema`, `get` and `stream` output validation failures are also this error. |
| `ReadableStreamNotSupportedError` | Read streams are not supported. |
| `WritableStreamNotSupportedError` | Write streams are not supported. |
| `EncodableStreamNotSupportedError` | Encode streams are not supported. |
| `DecodableStreamNotSupportedError` | Decode streams are not supported. |

```ts
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 [#examples]

A minimal example using `@unikvs/memory`.

**Minimal**

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

**With validation**

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