@livendb/liven-client

TypeScript client SDK for LivenDB

import { LivenClient, Pipeline, Filter } from "@livendb/liven-client";

const client = await LivenClient.connect("127.0.0.1:43121");
await client.insert("users", "u1", { name: "Alice", role: "admin" });
const user = await client.get("users", "u1");
console.log(user);
// → [{ stream_name: "users", key: "u1", value: { name: "Alice", role: "admin" }, … }]

const results = await client.run(
  Pipeline.from("users")
    .filter(Filter.field("role").eq("admin"))
    .limit(10)
);
client.close();

---

Installation

npm install @livendb/liven-client

Requires Node.js 18+.

Quick Start

1. Connect

import { LivenClient } from "@livendb/liven-client";

// Default port is 43121
const client = await LivenClient.connect("localhost");

// With authentication key
const client = await LivenClient.connect("localhost:43121?auth_key=your-api-key");

2. CRUD Operations

// Insert
await client.insert("events", "e1", { type: "click", value: 42 });

// Upsert (insert or replace)
await client.upsert("events", "e1", { type: "click", value: 99 });

// Update (merge with existing)
await client.update("users", "u1", { last_login: Date.now() });

// Get by key
const record = await client.get("users", "u1");

// Delete
await client.delete("events", "e1");

// Clear all records from a stream
await client.clear("sessions");

// Drop an entire stream
await client.dropStream("old_data");

3. Batch Operations

await client.insertMany("events", [
  ["e1", { type: "click" }],
  ["e2", { type: "view" }],
  ["e3", { type: "purchase" }],
]);

await client.upsertMany("users", [
  ["u1", { name: "Alice", score: 100 }],
  ["u2", { name: "Bob", score: 85 }],
]);

4. Query with Pipelines

import { Pipeline, Filter } from "@livendb/liven-client";

const results = await client.run(
  Pipeline.from("orders")
    .filter(Filter.field("amount").gte(100))
    .filter(Filter.field("status").eq("completed"))
    .sort("amount", true)
    .limit(10)
);

5. Shorthand Pipeline Methods

// Filter
await client.filter("events", Filter.field("type").eq("click"));

// Limit
await client.limit("events", 50);

// Count
const [countResult] = await client.count("events");

// Sort
await client.sort("orders", "total", true); // descending

// Pagination
await client.page("events", 1, 25);            // page 1, 25 per page
await client.pageCursor("feed", "cursor_abc", 25); // cursor-based

// Field projection
await client.map("users", ["name", "email"]);

// Time-windowed aggregation
await client.window("pageviews", 60000, "count"); // 1min window
await client.window("sales", 3600000, "sum");     // 1hr sum
await client.window("sensors", 300000, "avg");    // 5min avg

// Group by
await client.group("events", "type", ["count", "sum(value)"]);

// Distinct
await client.distinct("visitors", "ip_address");

// Vector similarity search
await client.vectorFilter("documents", "embedding", [1, 0, -1, …], 0.75);

6. Joins & Event Patterns

// Enrich (left join)
await client.enrich("orders", "users", "user_id");

// Correlate (time-bounded join)
await client.correlate("events", "sessions", "session_id", 5000);

// Chain (multi-hop join)
await client.chain("reviews", "orders", "order_id");

// Sequence (event pattern matching)
await client.sequence(
  "events",
  [
    Filter.field("type").eq("login"),
    Filter.field("type").eq("purchase"),
  ],
  300000 // within 5 minutes
);

7. Pipeline Mutation Operations

// Update records matching a pipeline
await client.runUpdate(
  Pipeline.from("events").filter(Filter.field("status").eq("pending")),
  { status: "processed" }
);

// Delete records matching a pipeline
await client.runDelete(
  Pipeline.from("events").filter(Filter.field("type").eq("temp"))
);

8. Live Subscriptions

// Subscribe to matching records in real-time
await client.runListen(
  Pipeline.from("alerts").filter(Filter.field("priority").eq("critical"))
);

9. Metadata

// List all streams
const streams = await client.streams();

// Server status
const status = await client.status();

API Reference

LivenClient

Method	Description
LivenClient.connect(addr)	Connect to a Liven server. Supports ?auth_key= in URL
close()	Close the connection
query(dsl)	Execute a raw LIVEN DSL string
insert(stream, key, value)	Insert a record
upsert(stream, key, value)	Insert or replace a record
update(stream, key, value)	Merge value into existing record
delete(stream, key)	Delete a record by key
get(stream, key)	Get a record by key
clear(stream)	Remove all records from a stream
dropStream(stream)	Remove an entire stream
insertMany(stream, batch)	Batch insert
upsertMany(stream, batch)	Batch upsert
streams()	List all streams
status()	Server status
filter(stream, filter)	Query with filter
limit(stream, count)	Query with limit
count(stream)	Count records
sort(stream, field, descending)	Query with sort
page(stream, page, size)	Page-based pagination
pageCursor(stream, cursor, size)	Cursor-based pagination
map(stream, fields)	Field projection
window(stream, durationMs, strategy)	Time-windowed aggregation
group(stream, field, aggregations)	Group-by aggregation
distinct(stream, field)	Distinct values
vectorFilter(stream, field, vector, threshold)	Vector similarity search
enrich(stream, source, joinKey)	Left join
correlate(stream, source, joinKey, withinMs)	Time-bounded join
chain(stream, target, joinKey)	Multi-hop join
sequence(stream, steps, withinMs)	Event pattern matching
run(pipeline)	Execute a Pipeline builder
runListen(pipeline)	Subscribe to pipeline results
runUpdate(pipeline, value)	Update matching records
runDelete(pipeline)	Delete matching records

Pipeline

Builder for constructing pipeline queries.

Pipeline.from("stream")
  .filter(filter)
  .get(key)
  .map(["field1", "field2"])
  .window(5000, "count")
  .limit(50)
  .sort("field", true)
  .page(1, 20)
  .pageCursor("cursor", 25)
  .count()
  .group("field", ["count", "sum(x)"])
  .correlate("source", "key", 5000)
  .chain("target", "key")
  .distinct("field")
  .enrich("source", "key")
  .vectorFilter("embedding", [1, 2, 3], 0.8)
  .sequence([filter1, filter2], 300000)
  .build()             // → DSL string
  .buildListen()       // → DSL string + .listen()
  .buildUpdate(value)  // → DSL string + update(value)
  .buildDelete()       // → DSL string + delete()

Filter

Builder for typed filter expressions.

Method	Description
Filter.field(name)	Start building a filter for a field
Filter.and([...filters])	Combine filters with AND
Filter.or([...filters])	Combine filters with OR
Filter.not(filter)	Negate a filter

Field comparisons:

Filter.field("status").eq("active")
Filter.field("status").ne("deleted")
Filter.field("amount").gt(100)
Filter.field("amount").lt(50)
Filter.field("amount").gte(100)
Filter.field("amount").lte(50)
Filter.field("name").contains("alice")
Filter.field("name").startsWith("al")
Filter.field("name").endsWith("ice")
Filter.field("amount").between(10, 100)
Filter.field("status").in(["active", "pending"])

Compound filters:

Filter.and([
  Filter.field("status").eq("active"),
  Filter.field("age").gte(18),
])

Filter.or([
  Filter.field("role").eq("admin"),
  Filter.field("role").eq("moderator"),
])

Filter.not(Filter.field("status").eq("deleted"))

Wire Protocol

The client communicates with LivenDB over TCP using framed msgpack messages:

[version: u8][length: u32 BE][discriminator: u8][msgpack_payload]

• version: 0x01
• discriminator: 0x02 (msgpack), 0x03 (raw vector bytes)
• length: payload size in bytes (big-endian)

Legacy v0 frames (4-byte length prefix, no version/discriminator) are supported for decoding.

Development

# Install dependencies
npm install

# Build
npm run build

# Run tests
npm test

# Watch mode
npm run test:watch

License

MIT