Rust API Reference¶

use uni_db::{Uni, Value};

#[tokio::main]
async fn main() -> uni_db::Result<()> {
    // Open or create a database
    let db = Uni::open("./my-graph").build().await?;

    // Define schema
    db.schema()
        .label("Person")
            .property("name", uni_db::DataType::String)
            .property("age", uni_db::DataType::Int64)
        .edge_type("KNOWS", &["Person"], &["Person"])
            .property("since", uni_db::DataType::Date)
        .apply()
        .await?;

    // All data access goes through Sessions
    let session = db.session();

    // Read with Cypher
    let results = session.query("MATCH (p:Person) RETURN p.name, p.age").await?;
    for row in results.rows() {
        let name: String = row.get("p.name")?;
        println!("{name}");
    }

    // Write through Transactions
    let tx = session.tx().await?;
    tx.execute("CREATE (:Person {name: 'Alice', age: 30})").await?;
    let result = tx.commit().await?;
    println!("Committed {} mutations at version {}", result.mutations_committed, result.version);

    db.shutdown().await?;
    Ok(())
}

Uni  (lifecycle / admin)
 |
 +-- Session  (read scope, cheap to create)
      |
      +-- Transaction  (write scope, ACID)

// Open or create at path
let db = Uni::open("./my-graph").build().await?;

// Open existing only (fails if missing)
let db = Uni::open_existing("./my-graph").build().await?;

// Create new (fails if already exists)
let db = Uni::create("./new-graph").build().await?;

// In-memory / temporary database
let db = Uni::in_memory().build().await?;
let db = Uni::temporary().build().await?;

let db = Uni::open("./my-graph")
    .schema_file("schema.json")
    .config(UniConfig { /* ... */ })
    .read_only()
    .write_lease(WriteLease::Local)
    .remote_storage("s3://bucket/data", cloud_config)
    .build()
    .await?;

let session = db.session();

// Queries (read-only)
let rows = session.query("MATCH (n:Person) RETURN n.name").await?;

// Parameterized queries
let rows = session.query_with("MATCH (n) WHERE n.age > $min RETURN n")
    .param("min", 21)
    .timeout(Duration::from_secs(5))
    .fetch_all()
    .await?;

// Transactions for writes
let tx = session.tx().await?;
tx.execute("CREATE (:Person {name: 'Bob'})").await?;
tx.commit().await?;

// Cloning (shares plan cache, independent params/metrics)
let session2 = session.clone();

let session = db.session();
session.params().set("tenant", 42);
session.params().set("region", "us-east");

// Parameters are auto-merged into queries as $session.tenant, $session.region
let rows = session.query("MATCH (n) WHERE n.tenant = $session.tenant RETURN n").await?;

let mut session = db.session();

// Pin to a named snapshot (read-only after pinning)
session.pin_to_version("snapshot_20240101").await?;

// Pin to a timestamp
session.pin_to_timestamp(chrono::Utc::now() - chrono::Duration::hours(1)).await?;

// Unpin and return to live state
session.refresh().await?;

let result = session.query_with("MATCH (n:Person) WHERE n.age > $min RETURN n")
    .param("min", 25)
    .param("limit", 100)
    .timeout(Duration::from_secs(10))
    .max_memory(512 * 1024 * 1024) // 512 MB
    .fetch_all()
    .await?;

let tx = session.tx().await?;

// Reads see uncommitted writes
tx.execute("CREATE (:Person {name: 'Alice', age: 30})").await?;
let rows = tx.query("MATCH (p:Person {name: 'Alice'}) RETURN p.age").await?;

// Parameterized mutations
tx.execute_with("CREATE (:Person {name: $name, age: $age})")
    .param("name", "Bob")
    .param("age", 25)
    .run()
    .await?;

let result = tx.commit().await?;
println!("Version: {}, Mutations: {}", result.version, result.mutations_committed);

// Evaluate a Locy DERIVE at session level (collects derived facts)
let result = session.locy("
    reachable(X, Y) :- knows(X, Y).
    reachable(X, Z) :- reachable(X, Y), knows(Y, Z).
    ?DERIVE reachable(X, Y).
").await?;

// Apply the derived facts inside a transaction
let tx = session.tx().await?;
let apply_result = tx.apply(result.derived().unwrap().clone()).await?;
println!("Applied {} facts, version gap: {}", apply_result.facts_applied, apply_result.version_gap);
tx.commit().await?;

let result = tx.commit().await?;
println!("Version: {}", result.version);
println!("Mutations: {}", result.mutations_committed);
println!("Duration: {:?}", result.duration);
println!("Version gap: {}", result.version_gap());

pub enum IsolationLevel {
    Serialized, // default — commit-time serialization with private L0 per tx
}

db.schema()
    .label("Person")
        .property("name", DataType::String)
        .property("age", DataType::Int64)
        .property_nullable("email", DataType::String)
        .vector("embedding", 1536)
        .index("name", IndexType::Scalar(ScalarType::BTree))
    .label("Document")
        .property("title", DataType::String)
        .index("title", IndexType::FullText)
    .edge_type("KNOWS", &["Person"], &["Person"])
        .property("since", DataType::Date)
        .property_nullable("weight", DataType::Float64)
    .apply()
    .await?;

pub enum DataType {
    String,
    Int64,
    Float64,
    Boolean,
    Date,
    DateTime,
    Duration,
    Bytes,
    Json,
    Vector { dimensions: usize },
    // ... additional variants
}

pub enum IndexType {
    Vector(VectorIndexCfg),
    FullText,
    Scalar(ScalarType),
    Inverted(InvertedIndexConfig),
}

pub enum ScalarType {
    BTree,
    Hash,
    Bitmap,
    LabelList,
}

pub struct VectorIndexCfg {
    pub algorithm: VectorAlgo,
    pub metric: VectorMetric,
    pub embedding: Option<EmbeddingCfg>,
}

pub enum VectorAlgo {
    Flat,
    IvfFlat { partitions: u32 },
    IvfPq { partitions: u32, sub_vectors: u32 },
    IvfSq { partitions: u32 },
    IvfRq { partitions: u32, num_bits: Option<u8> },
    Hnsw { m: u32, ef_construction: u32, partitions: Option<u32> },      // alias for HnswSq
    HnswFlat { m: u32, ef_construction: u32, partitions: Option<u32> },
    HnswSq { m: u32, ef_construction: u32, partitions: Option<u32> },
    HnswPq { m: u32, ef_construction: u32, sub_vectors: u32, partitions: Option<u32> },
}

pub enum VectorMetric {
    Cosine,
    L2,
    Dot,
}

let tx = session.tx().await?;

let mut writer = tx.bulk_writer()
    .batch_size(10_000)
    .defer_vector_indexes(true)
    .async_indexes(false)
    .validate_constraints(true)
    .max_buffer_size_bytes(1_073_741_824) // 1 GB
    .on_progress(|p| println!("{:?}", p.phase))
    .build()?;

let vids = writer.insert_vertices("Person", vec![
    HashMap::from([("name".into(), Value::from("Alice")), ("age".into(), Value::from(30))]),
    HashMap::from([("name".into(), Value::from("Bob")), ("age".into(), Value::from(25))]),
]).await?;

writer.insert_edges("KNOWS", vec![
    EdgeData::new(vids[0], vids[1], HashMap::new()),
]).await?;

let stats = writer.commit().await?;
println!("Inserted {} vertices, {} edges", stats.vertices_inserted, stats.edges_inserted);

tx.commit().await?;

let tx = session.tx().await?;

let mut appender = tx.appender("Person")
    .batch_size(5000)
    .defer_vector_indexes(true)
    .build()?;

appender.append(HashMap::from([("name".into(), Value::from("Alice"))])).await?;
appender.append(HashMap::from([("name".into(), Value::from("Bob"))])).await?;

// Also supports Arrow RecordBatch
// appender.write_batch(&record_batch).await?;

let stats = appender.finish().await?;

tx.commit().await?;

let result = session.locy_with("
    edge_rule(X, Y) :- knows(X, Y).
    ?edge_rule(X, Y).
")
    .param("threshold", 0.5)
    .timeout(Duration::from_secs(30))
    .max_iterations(1000)
    .run()
    .await?;

for row in result.rows() {
    println!("{:?}", row);
}

let result = session.locy("?QUERY knows(X, Y).").await?;

// Access rows (via Deref to inner LocyResult)
for row in result.rows() { /* ... */ }

// Execution metrics
let metrics = result.metrics();
println!("Duration: {:?}", metrics.duration);

// Derived facts (from DERIVE commands)
if let Some(derived) = result.derived() {
    let tx = session.tx().await?;
    tx.apply(derived.clone()).await?;
    tx.commit().await?;
}

// Warnings (from inner LocyResult via Deref)
for warning in result.warnings() {
    println!("Warning: {}", warning.message);
}

let prepared = session.prepare("MATCH (n:Person) WHERE n.age > $min RETURN n").await?;

// Execute with positional params
let result = prepared.execute(&[("min", Value::from(21))]).await?;

// Or use the fluent binder
let result = prepared.bind()
    .param("min", 21)
    .execute()
    .await?;

// Can be shared across threads via Arc
let shared = Arc::new(prepared);

let prepared = session.prepare_locy("edge_rule(X, Y) :- knows(X, Y). ?edge_rule(X, Y).").await?;

let result = prepared.execute(&[("threshold", Value::from(0.5))]).await?;

let result = prepared.bind()
    .param("threshold", 0.5)
    .execute()
    .await?;

let mut stream = session.watch();
while let Some(notif) = stream.next().await {
    println!("Version {} committed with {} mutations", notif.version, notif.mutation_count);
}

let mut stream = session.watch_with()
    .labels(&["Person", "Document"])
    .edge_types(&["KNOWS"])
    .exclude_session(session.id())
    .debounce(Duration::from_millis(100))
    .build();

pub trait SessionHook: Send + Sync {
    /// Called before a query. Return Err to reject.
    fn before_query(&self, ctx: &HookContext) -> Result<()> { Ok(()) }

    /// Called after a query completes. Panics are caught and logged.
    fn after_query(&self, ctx: &HookContext, metrics: &QueryMetrics) {}

    /// Called before commit. Return Err to reject.
    fn before_commit(&self, ctx: &CommitHookContext) -> Result<()> { Ok(()) }

    /// Called after commit succeeds. Panics are caught and logged.
    fn after_commit(&self, ctx: &CommitHookContext, result: &CommitResult) {}
}

struct AuditHook;

impl SessionHook for AuditHook {
    fn before_query(&self, ctx: &HookContext) -> uni_db::Result<()> {
        tracing::info!(session = %ctx.session_id, query = %ctx.query_text, "Query started");
        Ok(())
    }

    fn before_commit(&self, ctx: &CommitHookContext) -> uni_db::Result<()> {
        if ctx.mutation_count > 10_000 {
            return Err(uni_db::UniError::HookRejected {
                message: "Too many mutations in one commit".into(),
            });
        }
        Ok(())
    }
}

let mut session = db.session();
session.add_hook("audit", AuditHook);

pub enum WriteLease {
    Local,                          // Single-process lock (default)
    DynamoDB { table: String },     // DynamoDB-based distributed lease
    Custom(Box<dyn WriteLeaseProvider>),  // Custom provider
}

#[async_trait]
pub trait WriteLeaseProvider: Send + Sync {
    async fn acquire(&self) -> Result<LeaseGuard>;
    async fn heartbeat(&self, guard: &LeaseGuard) -> Result<()>;
    async fn release(&self, guard: LeaseGuard) -> Result<()>;
}

let db = Uni::open("./shared-db")
    .write_lease(WriteLease::Local)
    .build()
    .await?;

// Or with read-only mode (no writer)
let db = Uni::open("./shared-db")
    .read_only()
    .build()
    .await?;

let db = UniSync::in_memory()?;
let session = db.session();
let rows = session.query("MATCH (n) RETURN count(n)")?;
db.shutdown()?;

let template = db.session_template()
    .param("tenant", 42)
    .rules("edge_rule(X, Y) :- knows(X, Y).")?
    .hook("audit", AuditHook)
    .query_timeout(Duration::from_secs(30))
    .transaction_timeout(Duration::from_secs(60))
    .build()?;

// Cheap per-request session creation (sync, no recompilation)
let session = template.create();

let m = db.metrics();
println!("L0 mutations: {}", m.l0_mutation_count);
println!("Active sessions: {}", m.active_sessions);
println!("Uptime: {:?}", m.uptime);
println!("Total queries: {}", m.total_queries);

let result = session.query("MATCH (n:Person) RETURN n.name, n.age").await?;

// Iterate rows
for row in result.rows() {
    let name: String = row.get("n.name")?;
    let age: i64 = row.get("n.age")?;
}

// Metadata
println!("Columns: {:?}", result.columns());
println!("Row count: {}", result.len());
println!("Empty: {}", result.is_empty());

// Metrics
let metrics = result.metrics();
println!("Duration: {:?}", metrics.duration);

// Consume into owned rows
let rows: Vec<Row> = result.into_rows();

let result = tx.execute("CREATE (:Person {name: 'Alice'})").await?;
println!("Affected rows: {}", result.affected_rows);
println!("Nodes created: {}", result.nodes_created);
println!("Relationships created: {}", result.relationships_created);
println!("Properties set: {}", result.properties_set);

let row: &Row = result.rows().first().unwrap();

// Typed access by column name
let name: String = row.get("n.name")?;
let age: i64 = row.get("n.age")?;
let maybe_email: Option<String> = row.get("n.email").ok();

// Access by index
let first_value: &Value = row.get_by_index(0)?;

let mut cursor = session.query_with("MATCH (n) RETURN n")
    .cursor()
    .await?;

while let Some(row) = cursor.next().await? {
    // Process row without loading entire result set
}

pub enum Value {
    Null,
    Bool(bool),
    Int(i64),
    Float(f64),
    String(String),
    Bytes(Vec<u8>),
    List(Vec<Value>),
    Map(HashMap<String, Value>),
    Vector(Vec<f32>),
    Node { id: Vid, labels: Vec<String>, properties: HashMap<String, Value> },
    Edge { id: Eid, src: Vid, dst: Vid, edge_type: String, properties: HashMap<String, Value> },
    Path { nodes: Vec<Value>, edges: Vec<Value> },
    // ... additional variants (Date, DateTime, Duration)
}

Value::from(42_i64)        // Int
Value::from(3.14_f64)      // Float
Value::from("hello")       // String
Value::from(true)          // Bool
Value::from(vec![1.0_f32, 2.0, 3.0])  // Vector

// Compact a specific label or edge type
let stats = db.compaction().compact("Person").await?;

// Wait for all background compaction
db.compaction().wait().await?;

// List all indexes
let all_indexes = db.indexes().list(None);

// List indexes for a label
let person_indexes = db.indexes().list(Some("Person"));

// Rebuild indexes (blocking)
db.indexes().rebuild("Person", false).await?;

// Rebuild indexes (background)
let task_id = db.indexes().rebuild("Person", true).await?;

// Check rebuild status
let status = db.indexes().rebuild_status().await?;

// Retry failed rebuilds
let retried = db.indexes().retry_failed().await?;

// Create a snapshot
let snapshot_id = db.create_snapshot("before_migration").await?;

// List snapshots
let snapshots = db.list_snapshots().await?;
for s in &snapshots {
    println!("{}: {:?}", s.id, s.created_at);
}

// Pin a session to a snapshot
let mut session = db.session();
session.pin_to_version(&snapshot_id).await?;
let old_data = session.query("MATCH (n) RETURN count(n)").await?;
session.refresh().await?;  // Return to live state

// Restore to a snapshot (requires restart to fully take effect)
db.restore_snapshot(&snapshot_id).await?;

// Database-level (global)
db.rules().register("edge_rule(X, Y) :- knows(X, Y).")?;

// Session-level
let session = db.session();
session.rules().register("path_rule(X, Z) :- edge_rule(X, Y), edge_rule(Y, Z).")?;

// Query registered rules
let names = session.rules().list();
let info = session.rules().get("edge_rule");
println!("Rules: {:?}, count: {}", names, session.rules().count());

// Remove and clear
session.rules().remove("edge_rule")?;
session.rules().clear();