@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