<p align="center">
  <picture>
    <source media="(prefers-color-scheme: dark)" srcset="./assets/logo.svg">
    <img src="./assets/logo.svg" alt="LivenDB" width="80" height="80">
  </picture>
  <br/>
  <strong>LivenDB</strong>
</p>

<p align="center">
  <a href="https://github.com/livendb/liven/actions/workflows/build.yml">
    <img src="https://img.shields.io/github/actions/workflow/status/livendb/liven/build.yml?branch=main&label=Build&logo=github" alt="Build">
  </a>
  <a href="https://github.com/livendb/liven/pkgs/container/liven">
    <img src="https://img.shields.io/badge/ghcr.io-liven-blue?logo=docker" alt="GHCR">
  </a>
  <a href="https://crates.io/crates/liven">
    <img src="https://img.shields.io/crates/v/liven?logo=rust&label=crates.io" alt="crates.io">
  </a>
  <a href="https://crates.io/crates/liven">
    <img src="https://img.shields.io/crates/d/liven?logo=rust&label=downloads" alt="downloads">
  </a>
</p>

---

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

```sh
# 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

```bash
# 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:

```toml
[dependencies]
liven = "0.0.6"                        # full build (server, TUI, TLS)
```

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

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

Select individual features:

```toml
[dependencies]
liven = { version = "0.0.6", default-features = false, features = ["tls"] }   # core + TLS
liven = { version = "0.0.6", default-features = false, features = ["server", "tls"] }  # core + server + TLS
liven = { version = "0.0.6", 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.

```rust
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:

```rust
// 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

```rust
// ── 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

```rust
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()`:

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

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

```rust
use liven::query::Query as Q;

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

### Real-time subscriptions

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

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

```rust
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 &rarr;](https://docs.rs/liven)**

---

## How It Works (at a glance)

```mermaid
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]
```

- **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

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

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

Benchmark source: [`benches/engine_bench.rs`](./benches/engine_bench.rs)
Docker runner: [`Dockerfile.bench`](./Dockerfile.bench), [`run-bench.sh`](./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           |
| `tui`    | Interactive terminal dashboard                   |
| `tls`    | mTLS support with X.509 certificates             |

```sh
# 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
```

---

## Licensing

**SSPL 1.0 OR Commercial**

- **SSPL** — Free for self-hosting, development, and personal use.
- **Commercial** — Required for managed services or proprietary embedding.

Contact `team@livendb.com` for commercial licensing.

[**Full license &rarr;**](./LICENSE-SSPL)

## Contributing

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

All contributors are expected to follow our [Code of Conduct](./CODE_OF_CONDUCT.md).
