LivenDB

---

Stream. Process. Store. One Engine.

Liven is a database built for data that moves. It ingests streaming data, transforms it on the fly, and stores it durably — all with a single pipeline query language. One binary.

# Install via crates.io
cargo install liven

# One-liner install
curl --proto '=https' --tlsv1.2 -sSfL https://livendb.com/install | sh

# Install via Docker
docker pull ghcr.io/livendb/liven:latest
docker run -p 43121:43121 -p 43120:43120 -v ./data:/var/lib/liven livendb/liven:latest

# Launch the server
liven start

---

Why Liven?

Databases today make you choose: batch or stream? Historical or real-time? Key-value or vector? Liven was built to erase those lines.

One query language, two modes:

• Historical queries against stored data
• Real-time subscriptions on the same pipeline — just add .listen()

One engine, three deployment models:

• Embedded library (~1.5 MB) — runs inside your Rust process
• Network server — TCP + WebSocket, thousands of clients
• Interactive TUI or web dashboard — for ad-hoc queries and monitoring

Built-in capabilities that usually require separate systems:

• Vector similarity search (int8 quantized, cosine similarity)
• Stream joins (time-bounded correlate, multi-hop chain)
• Event pattern detection (sequence FSM)
• Time-windowed aggregations
• Full-text substring matching

---

Quick Start

# One-liner install
curl --proto '=https' --tlsv1.2 -sSfL https://livendb.com/install | sh

# Or via Docker
docker run -p 43121:43121 -p 43120:43120 ghcr.io/livendb/liven

# Or build from source
cargo build --release
./target/release/liven start
# → Open http://localhost:43120
# → Admin auth key printed on first start — save it

# Insert and query via the embedded Web UI at http://localhost:43120

# Or use the interactive TUI shell
liven vibe

# Tail a stream in real time
liven tail events

# List streams
liven list

Embedded in Rust

Add the dependency with the features you need:

[dependencies]
liven = "0.1.1"                        # full build (server, TUI, TLS)

For a minimal embedded build with no server, TUI, or TLS:

[dependencies]
liven = { version = "0.1.1", default-features = false }      # core only

Select individual features:

[dependencies]
liven = { version = "0.1.1", default-features = false, features = ["tls"] }   # core + TLS
liven = { version = "0.1.1", default-features = false, features = ["server", "tls"] }  # core + server + TLS
liven = { version = "0.1.1", features = ["tui"] }  # full + TUI (already included)

---

Rust Crate API

Liven provides two usage modes via the same unified method signatures:

Mode	Initialization	Runtime
Embedded	Liven::open("./data")?	In-process, no server needed
Wire	LivenClient::connect("127.0.0.1:43121").await?	Remote server over TCP

Both modes expose the same methods (insert, get, filter, enrich, etc.) — the embedded versions are synchronous, the wire versions are async.

use liven::Liven;
use liven::client::LivenClient;
use liven::query::{Pipeline, Filter};
use serde_json::json;

// ── Embedded ──
let db = Liven::open("./data")?;
db.insert("events", "e1", json!({"type": "click"}))?;
let results = db.run(
    Pipeline::from("events")
        .filter(Filter::field("type").eq("click"))
        .limit(10)
)?;

// ── Wire (async) ──
let mut client = LivenClient::connect("127.0.0.1:43121").await?;
client.insert("events", "e1", json!({"type": "click"})).await?;
let results = client.run(
    &Pipeline::from("events")
        .filter(Filter::field("type").eq("click"))
        .limit(10)
        .build(),
).await?;

Tip: Use db.query("...") for ad-hoc string queries and db.insert(...) / db.get(...) etc. for the typed API. Both work identically in embedded mode and over the wire.

Connection URL

The wire client supports connection URLs with optional auth key:

// Plain TCP
LivenClient::connect("127.0.0.1:43121").await?;

// With auth key in URL
LivenClient::connect("127.0.0.1:43121?auth_key=my_secret").await?;

CRUD operations

// ── Embedded ──        // ── Wire (async) ──
db.insert("users", "u1", json!({"name":"Alice"}))?;
                        // client.insert("users", "u1", json!(...)).await?;

db.upsert("users", "u1", json!({"name":"Alice"}))?;
db.update("users", "u1", json!({"status":"active"}))?;
db.get("users", "u1")?;
db.delete("users", "u1")?;
db.clear("logs")?;
db.drop_stream("temp")?;
db.insert_many("orders", vec![("o1".into(), json!({"amount":100}))])?;
db.upsert_many("orders", vec![("o1".into(), json!({"amount":200}))])?;

// Metadata
db.streams()?;
db.status()?;

Pipeline operations

use liven::query::{Pipeline, Filter};
use liven::types::AggregateStrategy;

// ── Embedded ──                    // ── Wire (async) ──
db.filter("events",                   // client.filter("events",
    Filter::field("type").eq("click"), //   Filter::field("type").eq("click"),
)?;                                     // ).await?;

db.limit("events", 10)?;             // client.limit("events", 10).await?;
db.count("events")?;                  // client.count("events").await?;
db.sort("events", "timestamp", true)?; // client.sort("events","timestamp",true).await?;
db.page("events", 1, 50)?;           // client.page("events", 1, 50).await?;
db.map("users", vec!["name".into(), "email".into()])?;
db.window("metrics", 60_000, AggregateStrategy::avg())?;
db.group("events", "type", vec!["count".into()])?;
db.distinct("users", "email")?;
db.page_cursor("events", "cursor_abc", 50)?;

// Vector similarity
db.vector_filter("embeddings", "vector", vec![12, -5, 3], 0.85)?;

// Stream joins
db.enrich("logs", "users", "user_id")?;
db.correlate("events", "orders", "user_id", 5000)?;
db.chain("prompts", "responses", "prompt_id")?;
db.sequence("system_events",
    vec![Filter::field("event").eq("disk_full"),
         Filter::field("event").eq("crash")],
    10_000)?;

Pipeline builder (for complex chains)

When you need multiple stages, use the builder and execute with db.run():

use liven::query::{Pipeline, Filter};

let pipeline = Pipeline::from("orders")
    .filter(Filter::field("amount").gte(100.0))
    .filter(Filter::field("status").eq("completed"))
    .sort("amount", true)
    .limit(10);

// Embedded
db.run(pipeline.clone())?;

// Wire
client.run(&pipeline.build()).await?;

Pipeline update / delete

let pipeline = Pipeline::from("orders")
    .filter(Filter::field("status").eq("pending"));

// Update all matching records
db.pipeline_update(pipeline.clone(), json!({"status": "cancelled"}))?;

// Delete all matching records
db.pipeline_delete(pipeline)?;

Explain

use liven::query::Query as Q;

let plan = db.explain(Q::insert("events", "e1", json!({"x": 1})))?;

Real-time subscriptions

use liven::query::{Pipeline, Filter};

// Blocking subscription (embedded, non-async)
loop {
    if let Some(record) = db.subscribe_sync(std::time::Duration::from_millis(100))? {
        println!("New: {}", record.key);
    }
}

// Async subscription (embedded)
let mut rx = db.subscribe();
tokio::spawn(async move {
    while let Ok(record) = rx.recv().await {
        println!("Live: {:?}", record);
    }
});

// Wire streaming (client)
use futures_util::StreamExt;
let mut stream = client.listen("events").await?;
while let Some(Ok(record)) = stream.next().await {
    println!("Got: {}", record.key);
}

Custom configuration

use liven::embed::{LivenConfig, Liven};

let config = LivenConfig {
    max_streams: 128,
    max_index_ram_mb: 1024,
    ..Default::default()
};
let db = Liven::open_with_config("./data", config)?;

Full example

use liven::Liven;
use liven::query::{Pipeline, Filter};
use serde_json::json;

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let dir = format!("./liven_demo_{}", std::process::id());
    let db = Liven::open(&dir)?;

    // Insert
    db.insert("events", "e1", json!({"type": "click", "value": 10}))?;
    db.insert("events", "e2", json!({"type": "purchase", "value": 50}))?;
    db.insert("events", "e3", json!({"type": "click", "value": 20}))?;

    // Query
    let clicks = db.filter("events", Filter::field("type").eq("click"))?;
    println!("Clicks: {:?}", clicks);

    // Count
    let count = db.count("events")?;
    println!("Total: {:?}", count);

    let _ = std::fs::remove_dir_all(&dir);
    Ok(())
}

Full API documentation →

---

How It Works (at a glance)

flowchart LR
    Client -->|query / subscribe| Query[Pipeline Query<br/>Engine]
    Query -->|write| Storage[Append-Only<br/>Storage]
    Query -->|read| Index[In-Memory Index]
    Storage -->|flusher updates| Index
    Index -->|point lookup| Query
    Index -->|broadcast| Subscriber[Live Subscribers]

Loading

• Writes are appended to segment files. A background flusher batches them for throughput without sacrificing durability.
• Reads go through a lock-free in-memory index. Point lookups resolve in microseconds.
• Subscriptions broadcast every write to all listeners. The server evaluates pipeline filters before delivery.
• Compaction reclaims space from deleted records automatically.
• Recovery replays segments on startup. Checksums catch corruption.

---

Benchmarks

Reproducible benchmarks run inside a pinned Docker image (x86-64-v2 CPU features) to ensure consistent results across hardware.

Operation	Performance	Notes
Point lookup (existing key)	~2.3 µs	Microsecond, independent of dataset size
Point lookup (missing key)	~65 ns	Near-zero cost (hash miss)
Range index (timestamp)	~88 ns	Billion elements/second
Full scan	~600–700K elem/s	Linear, predictable throughput
Scan with limit	~67M elem/s	Short-circuits after limit
Append (single)	~5 ms	Fsync-bound per operation
Append (batch 500)	~707K ops/s	Batch for throughput
Upsert (new key)	~5.3 ms	Same cost as append
Upsert (existing key)	~10.7 ms	Includes tombstone write
Compaction	~500–570 µs	Sub-linear growth
Parse (simple)	~437 ns	Not a bottleneck
Parse (complex)	~1.15 µs	Still sub-microsecond
Vector lookup (quantized 512d)	223 MiB/s	9× faster than msgpack
Wire encode (16 KB)	15 GiB/s	Far beyond network limits

Run benchmarks

# Local (requires Rust nightly for CPU features)
cargo bench

# Reproducible Docker (recommended)
./run-bench.sh

Benchmark source: benches/engine_bench.rs Docker runner: Dockerfile.bench, run-bench.sh

---

Security

Auth-key mode (default)

Symmetric keys with BLAKE3 hashing. Four role levels:

Role	Read	Insert	Delete	Admin
read-only	✅	❌	❌	❌
write	✅	✅	❌	❌
write-delete	✅	✅	✅	❌
admin	✅	✅	✅	✅

Keys can be generated, revoked, and role-changed at runtime via the Web UI or REST API — no server restart required.

mTLS / ZTNA

Mutual TLS with X.509 certificates. Client CN maps to capabilities. Single-port mode multiplexes cleartext and TLS on the same listener.

Master key

Stored in ./liven.key (mode 0600). Override with LIVEN_SECURITY_MASTER_KEY environment variable.

---

Feature flags

Liven uses Cargo feature flags for modular builds.

The default feature enables everything by pulling in full, which bundles all three optional capabilities.

Feature	What's included
full	All features below (enabled by default)
server	REST API + WebSocket + embedded Web UI + mTLS
tui	Interactive terminal dashboard
tls	mTLS support with X.509 certificates

# Minimal embedded build (no server, no TUI, no TLS)
cargo build --release --no-default-features

# Embedded with TLS support
cargo build --release --no-default-features --features tls

# Server without TUI (includes TLS for ZTNA/mTLS)
cargo build --release --no-default-features --features server

---

License

Licensed under the Apache License, Version 2.0.

Full license →

Contributing

Contributions are welcome! See CONTRIBUTING.md for guidelines on submitting pull requests, code style, and development setup.

All contributors are expected to follow our Code of Conduct.