Skip to main content

Advanced Operations

This guide covers advanced features available in ekoDB client libraries, including search operations, chat functionality, and real-time data handling.

Complete Examples Available

For complete, runnable examples, visit the ekoDB Examples Repository. It contains 265+ examples across Rust, Go, Python, TypeScript, Kotlin, and JavaScript.

Options Structs​

New in v0.8.0

Cleaner method signatures using builder pattern for operation options (Rust, TypeScript, Kotlin).

Overview​

Instead of long parameter lists, use options structs for cleaner, more maintainable code:

Before (v0.7.x):

await client.insert("users", record, "1h", true, "tx_123", false);
// What do these parameters mean? 🤔

After (v0.8.0):

await client.insert("users", record, {
  ttl: "1h",
  bypassRipple: true,
  transactionId: "tx_123",
  bypassCache: false
});
// Much clearer! ✨

Insert with Options​

use ekodb_client::{Client, Record, InsertOptions};

let mut record = Record::new();
record.insert("name", "Alice");
record.insert("email", "alice@example.com");

// Use builder pattern for options
let options = InsertOptions::new()
    .ttl("1h")
    .bypass_ripple(true)
    .transaction_id("tx_123")
    .bypass_cache(false);

let result = client.insert("users", record, Some(options)).await?;

Update with Options​

use ekodb_client::UpdateOptions;

let options = UpdateOptions::new()
    .bypass_ripple(true)
    .transaction_id("tx_456");

let updated = client.update("users", "user-123", updates, Some(options)).await?;

Available Options Structs​

StructAvailable FieldsLanguages
InsertOptionsttl, bypass_ripple, transaction_id, bypass_cacheAll
UpdateOptionsbypass_ripple, transaction_id, bypass_cacheAll
UpsertOptionsbypass_ripple, transaction_idAll
DeleteOptionsbypass_ripple, transaction_idAll
FindOptionsbypass_cache, transaction_idAll
Language-Specific Patterns
  • Rust/TypeScript/Kotlin: Dedicated options structs with builder pattern
  • Python: Optional keyword arguments (Pythonic)
  • Go: Variadic options with pointers (idiomatic)

All approaches provide the same functionality with language-appropriate ergonomics.

Search Operations​

ekoDB provides powerful search capabilities including full-text search, fuzzy search, and vector search.

Full-Text Search​

Search across all fields in your documents:

use ekodb_client::{Client, SearchQuery};

let search = SearchQuery::builder()
    .query("database")
    .min_score(0.1)
    .limit(10)
    .build();

let results = client.search("articles", search).await?;

for result in results.results {
    println!("Score: {:.4} - {:?}", result.score, result.record);
}

Search with custom field weights to prioritize certain fields:

use std::collections::HashMap;

let mut weights = HashMap::new();
weights.insert("title".to_string(), 2.0);
weights.insert("description".to_string(), 1.0);

let search = SearchQuery::builder()
    .query("rust database")
    .fields(vec!["title".to_string(), "description".to_string()])
    .weights(weights)
    .limit(5)
    .build();

let results = client.search("articles", search).await?;

Enable typo tolerance with fuzzy matching:

let search = SearchQuery::builder()
    .query("databse")  // Typo: "databse" instead of "database"
    .fuzzy(true)
    .fuzziness(2)  // Allow up to 2 character differences
    .limit(10)
    .build();

let results = client.search("articles", search).await?;

Perform semantic similarity search using embeddings:

// First, create a collection with vector index
let schema = Schema::builder()
    .add_field("content", FieldType::String)
    .add_field("embedding", FieldType::Vector(384))  // 384-dimensional vector
    .build();

client.create_collection("documents", schema).await?;

// Insert document with embedding
let mut doc = Record::new();
doc.insert("content", "ekoDB is a high-performance database");
doc.insert("embedding", vec![0.1, 0.2, 0.3, /* ... 384 dimensions */]);
client.insert("documents", doc).await?;

// Search by vector similarity
let query_vector = vec![0.1, 0.2, 0.3, /* ... */];
let search = SearchQuery::builder()
    .vector(query_vector)
    .limit(10)
    .build();

let results = client.search("documents", search).await?;

Combine text and vector search for best results:

let search = SearchQuery::builder()
    .query("database performance")  // Text query
    .vector(query_vector)           // Vector query
    .limit(10)
    .build();

let results = client.search("documents", search).await?;

Chat Operations​

Build AI-powered chat applications with built-in context management and session handling.

Basic Chat​

use ekodb_client::{Client, CreateChatSessionRequest, ChatMessageRequest, CollectionConfig};

// Create a chat session
let session = client.create_chat_session(CreateChatSessionRequest {
    collections: vec![CollectionConfig {
        collection_name: "products".to_string(),
        fields: vec![],
        search_options: None,
    }],
    llm_provider: "openai".to_string(),
    llm_model: Some("gpt-4.1".to_string()),
    system_prompt: Some("You are a helpful assistant.".to_string()),
    ..Default::default()
}).await?;

// Send a message
let response = client.chat_message(
    &session.chat_id,
    ChatMessageRequest::new("What products do you have?")
).await?;

println!("AI: {:?}", response.responses);

Chat Sessions​

Manage conversation history with sessions:

// Send multiple messages in the same session
let response1 = client.chat_message(
    &session.chat_id,
    ChatMessageRequest::new("What's the price of ekoDB Pro?")
).await?;

let response2 = client.chat_message(
    &session.chat_id,
    ChatMessageRequest::new("What features does it include?")
).await?;

// Get session message history
let messages = client.get_chat_session_messages(
    &session.chat_id, None
).await?;
println!("Total messages: {}", messages.messages.len());
Schema-Aware Chat Queries

ekoDB's chat system features intelligent query understanding that works with your natural data structure:

  • Field Name Matching: Queries like "What is the price?" automatically find records with a price field, even if it contains numeric data (not text-searchable)
  • Multi-Turn Context: Follow-up questions use conversation history to enhance search relevance
  • No Denormalization Required: Works with structured data as you'd naturally model it

Example:

// Your data structure
{
  "product": "ekoDB",
  "description": "High-performance database",
  "price": 99  // Numeric field
}

When a user asks "What is the price?", ekoDB:

  1. Checks the collection schema for fields matching "price"
  2. Finds records with that field name
  3. Provides the full record to the LLM
  4. LLM responds: "The price is $99"

See complete Chat Session examples in all languages:

  • Rust: client_chat_sessions.rs
  • Python: client_chat_sessions.py
  • TypeScript: client_chat_sessions.ts
  • Go: client_chat_sessions.go
  • Kotlin: ClientChatSessions.kt

Real-Time Operations​

WebSocket Queries​

Subscribe to real-time data changes:

use ekodb_client::WebSocketClient;

// Connect to WebSocket
let ws_url = "wss://your-subdomain.production.google.ekodb.net";
let mut ws_client = client.websocket(ws_url).await?;

// Subscribe to collection changes
let results = ws_client.find_all("users").await?;

// Process real-time updates
for record in results {
    println!("New/Updated record: {:?}", record);
}

ws_client.close().await?;

WebSocket CRUD Operations​

All 14 server-supported CRUD operations are available over WebSocket. The persistent WS connection eliminates HTTP overhead per request — zero TLS handshake, reuses the authenticated connection. All methods support messageId for concurrent request correlation.

// Connect via convenience method (derives WS URL, attaches schema cache)
let ws = client.connect_ws().await?;

// Insert
let record = ws.insert("users", json!({"name": "Alice"}), None).await?;

// Query with filter
let results = ws.query("users",
    Some(json!({"field": "status", "operator": "Eq", "value": "active"})),
    None, Some(10), None,
).await?;

// Find by ID
let user = ws.find_by_id("users", "record-id").await?;

// Update
ws.update("users", "record-id", json!({"name": "Updated"}), None).await?;

// Delete
ws.delete("users", "record-id", None).await?;

// Batch operations
ws.batch_insert("logs", vec![json!({"msg": "a"}), json!({"msg": "b"})], None).await?;

// Search
let hits = ws.text_search("docs", "rust async", None, Some(10)).await?;

// Collection management
let collections = ws.list_collections().await?;
ws.create_collection("new_coll", None).await?;

// Atomic field actions
ws.update_with_action("counters", "views", "increment", "count", Some(json!(1))).await?;

Schema Cache​

The schema cache stores each collection's primary_key_alias and version in memory. This ensures extractRecordId() works correctly regardless of how users configure their ID field names. The cache is LRU with configurable TTL, and auto-invalidates via WebSocket SchemaChanged events.

// Enable at client creation
let client = Client::builder()
    .base_url("https://api.ekodb.net")
    .api_key("key")
    .schema_cache(true)
    .schema_cache_ttl(300)  // seconds
    .schema_cache_max(100)  // max collections
    .build()?;

// Extract IDs correctly with any primary_key_alias
let id = client.extract_id("users", &record);

// Auto-invalidates when connected via WS
let ws = client.connect_ws().await?;

SSE Subscriptions​

Subscribe to collection mutations via Server-Sent Events. Works behind reverse proxies that block WebSocket upgrades. Also delivers schema_changed events for automatic schema cache invalidation.

let rx = client.subscribe_sse("orders", None, None).await?;
while let Some(event) = rx.recv().await {
    println!("{}: {} on {}",
        event.event, event.record_ids.join(", "), event.collection);
}

// With filter — only receive mutations where status = "active"
let rx = client.subscribe_sse(
    "orders",
    Some("status"),
    Some("active"),
).await?;

SSE also delivers schema_changed events, automatically invalidating the client's schema cache when a collection's configuration changes.

Joins​

ekoDB supports cross-collection joins to combine data from multiple collections in a single query.

Single Collection Join​

Join users with their department data:

use ekodb_client::{Client, QueryBuilder, JoinBuilder};

// Join users with departments
let join = JoinBuilder::single(
    "departments",      // Target collection
    "department_id",    // Local field (in users)
    "id",               // Foreign field (in departments)
    "department"        // Output field name
);

let query = QueryBuilder::new()
    .join(join)
    .limit(10)
    .build();

let users = client.find("users", query).await?;
for user in users {
    println!("User: {:?}, Department: {:?}", user["name"], user["department"]);
}
Complete Join Examples

Join examples - Single and multi-collection joins with filtering:

  • Rust: client_joins.rs
  • Python: client_joins.py
  • TypeScript: client_joins.ts
  • JavaScript: client_joins.js
  • Go: client_joins.go
  • Kotlin: ClientJoins.kt

TTL (Time-To-Live)​

Set automatic expiration for documents:

use ekodb_client::options::InsertOptions;

// Insert with 1 hour TTL
let mut session = Record::new();
session.insert("user_id", "user-123");
session.insert("token", "abc123");

let options = InsertOptions::new().ttl("1h");  // Expires in 1 hour
let result = client.insert("sessions", session, Some(options)).await?;
Complete TTL Examples

Document TTL examples - Insert with expiration, verify expiration works:

  • Rust: client_document_ttl.rs
  • Python: client_document_ttl.py
  • TypeScript: client_document_ttl.ts
  • JavaScript: client_document_ttl.js
  • Go: client_document_ttl.go
  • Kotlin: ClientDocumentTtl.kt

WebSocket TTL examples - TTL with real-time connections:

  • Rust: client_websocket_ttl.rs
  • Python: client_websocket_ttl.py
  • TypeScript: client_websocket_ttl.ts
  • JavaScript: client_websocket_ttl.js
  • Go: client_websocket_ttl.go
  • Kotlin: ClientWebsocketTtl.kt

Transactions​

ekoDB supports ACID transactions with multiple isolation levels. Transactions ensure data consistency when performing multiple operations that must succeed or fail together.

Isolation Levels​

LevelDescription
READ_UNCOMMITTEDCan read uncommitted changes from other transactions
READ_COMMITTEDOnly reads committed data (default)
REPEATABLE_READSame data on re-read within transaction
SERIALIZABLEHighest isolation, transactions appear sequential

Basic Transaction​

// Start a transaction
let tx_id = client.begin_transaction("SERIALIZABLE").await?;

// Perform operations within transaction context

// Commit when done
client.commit_transaction(&tx_id).await?;

// Or rollback on failure
// client.rollback_transaction(&tx_id).await?;
Transaction Best Practices
  • Use the lowest isolation level that meets your consistency requirements
  • Keep transactions short to minimize lock contention
  • Always handle rollback in error cases
  • For detailed transaction patterns, see Transactions

RAG Helpers​

The Go client includes convenience methods for RAG (Retrieval-Augmented Generation) workflows:

Generate Embeddings​

Generate embedding vectors from text using ekoDB's native Functions:

// Generate embedding for text
embedding, err := client.Embed("Hello world", "text-embedding-3-small")
if err != nil {
    log.Fatal(err)
}
fmt.Printf("Generated %d dimensions\n", len(embedding))

Text Search​

Perform full-text search with stemming and fuzzy matching:

// Search for documents by text
results, err := client.TextSearch("documents", "database performance", 10)
if err != nil {
    log.Fatal(err)
}
for _, doc := range results {
    fmt.Printf("Found: %v\n", doc["title"])
}

Hybrid Search​

Combine semantic similarity (vector) with keyword matching (text):

// Generate embedding for query
embedding, _ := client.Embed("How to optimize queries?", "text-embedding-3-small")

// Perform hybrid search
results, err := client.HybridSearch("documents", "optimize queries", embedding, 5)
if err != nil {
    log.Fatal(err)
}

Find All Records​

Simple method to retrieve all records from a collection:

// Get all messages (up to limit)
allMessages, err := client.FindAll("messages", 1000)
if err != nil {
    log.Fatal(err)
}
fmt.Printf("Found %d messages\n", len(allMessages))

Functions​

Server-Side Feature

Functions are ekoDB's stored procedures system that runs on the server. They can be called from any client library or via REST API to execute complex business logic, queries, CRUD operations, AI workflows, and batch processing.

Deep Dive

For comprehensive architecture details, operation types, and advanced patterns, see Functions Architecture.

Functions let you create, store, and execute complete business logic as composable operations. Define your data logic once in ekoDB, then call it like puzzle pieces from any client.

What You Can Do​

  • ✅ Complete business logic - Queries, CRUD, AI operations in one place
  • ✅ Parameterize everything - Dynamic values via {{param_name}}
  • ✅ Version control - Track function versions
  • ✅ Compose like puzzles - Chain operations together
  • ✅ Call from anywhere - REST API or any client library

Function Capabilities​

  • Query Operations: Find, filter, search, vector search, hybrid search
  • CRUD Operations: Insert, update, delete (single and batch)
  • Transformations: Group, project, count
  • AI Operations: Chat completions, embeddings generation
  • Conditional Logic: If/then/else, foreach loops
  • External Integrations: HTTP requests to any REST API

Basic Example​

Create a function to query active users:

POST /api/functions
Content-Type: application/json

{
  "label": "get_active_users",
  "name": "Get Active Users",
  "description": "Returns active users with a limit",
  "parameters": {
    "limit": {
      "default": 10,
      "required": false
    }
  },
  "functions": [
    {
      "type": "Query",
      "collection": "users",
      "filter": {
        "type": "Condition",
        "field": "status",
        "operator": "Eq",
        "value": "active"
      },
      "limit": "{{limit}}"
    }
  ]
}

Call a Function​

Execute via REST API:

POST /api/functions/get_active_users
Content-Type: application/json

{
  "limit": 20
}

Managing Functions​

REST API:

# List all functions
GET /api/functions

# Get a specific function by ID or label
GET /api/functions/get_active_users

# Update a function by ID or label
PUT /api/functions/{id_or_label}

# Delete a function by ID or label
DELETE /api/functions/get_active_users

Client Library Methods:

use ekodb_client::{Client, Function, ParameterDefinition, UserFunction};

// Create a user function using the builder pattern
let user_func = UserFunction::new("get_active_users", "Get Active Users")
    .with_version("1.0.0")
    .with_parameter(ParameterDefinition {
        name: "collection".to_string(),
        required: true,
        description: Some("Collection to query".to_string()),
        default: None,
    })
    .with_function(Function::FindAll {
        collection: "{{collection}}".to_string(),
    })
    .with_tag("users")
    .with_tag("query");

let func_id = client.save_user_function(user_func).await?;

// Get a user function by label
let func = client.get_user_function("get_active_users").await?;

// List all user functions (optionally filter by tags)
let all_funcs = client.list_user_functions(None).await?;
let tagged_funcs = client.list_user_functions(Some(vec!["users".to_string()])).await?;

// Update a user function
client.update_user_function("get_active_users", updated_func).await?;

// Delete a user function
client.delete_user_function("get_active_users").await?;

Parameters​

Make functions dynamic with parameters:

{
  "parameters": {
    "status": {
      "default": "active",
      "required": false,
      "description": "Filter by status"
    },
    "min_amount": {
      "required": true,
      "description": "Minimum amount"
    }
  }
}

Reference parameters in your function definitions using {{param_name}}:

{
  "type": "Query",
  "collection": "orders",
  "filter": {
    "type": "Condition",
    "field": "status",
    "operator": "Eq",
    "value": "{{status}}"
  }
}

Available Operations​

Functions support these operation types:

Query Operations​

  • FindAll - Retrieve all records
  • Query - Advanced filtering, sorting, pagination
  • VectorSearch - Semantic similarity search
  • HybridSearch - Combine text + vector search
  • TextSearch - Full-text search
  • FindById - Get specific record by ID
  • FindOne - Find one by key/value

CRUD Operations​

  • Insert - Insert single record
  • BatchInsert - Insert multiple records
  • Update - Update with filter
  • UpdateById - Update specific record
  • Delete - Delete with filter
  • DeleteById - Delete specific record
  • BatchDelete - Delete multiple records

Transformations​

  • Group - Group and aggregate
  • Project - Select/exclude fields
  • Count - Count records

AI Operations​

  • Chat - AI chat completions
  • Embed - Generate embeddings

Logic & Control​

  • If - Conditional execution
  • ForEach - Loop over records
  • CallFunction - Call another function

External Integrations​

  • HttpRequest - Call external APIs (Stripe, SendGrid, etc.)

Combine embeddings with search:

{
  "label": "smart_search",
  "name": "Smart Product Search",
  "parameters": {
    "query": { "required": true }
  },
  "functions": [
    {
      "type": "Embed",
      "input_field": "query",
      "output_field": "query_embedding"
    },
    {
      "type": "HybridSearch",
      "collection": "products",
      "query_text": "{{query}}",
      "query_vector": "{{query_embedding}}",
      "limit": 10
    }
  ]
}

Example: Batch Processing​

Process multiple records with AI:

{
  "label": "enrich_articles",
  "name": "AI Content Enrichment",
  "functions": [
    {
      "type": "Query",
      "collection": "articles",
      "filter": {
        "type": "Condition",
        "field": "embedding",
        "operator": "Eq",
        "value": null
      },
      "limit": 100
    },
    {
      "type": "ForEach",
      "functions": [
        {
          "type": "Embed",
          "input_field": "content",
          "output_field": "embedding"
        },
        {
          "type": "UpdateById",
          "collection": "articles",
          "record_id": "{{id}}",
          "updates": {
            "embedding": "{{embedding}}"
          }
        }
      ]
    }
  ]
}

Best Practices​

  • Keep it simple - Start with single operations, build up
  • Use parameters - Make functions reusable with dynamic values
  • Filter early - Reduce data before expensive operations
  • Add descriptions - Document what each function does
  • Tag for organization - Use tags like analytics, users, ai
  • Version your functions - Track changes with version field
  • Test thoroughly - Validate with edge cases

Storage​

Functions are stored in a dedicated collection: functions_{db_name} (configurable)

Complete Documentation​

For complete details including:

  • All operation types and parameters
  • Advanced parameter resolution
  • Conditional logic patterns
  • External API integration examples
  • Error handling

See complete Function examples in all languages:

  • Rust: client_function_composition.rs
  • Python: client_function_composition.py
  • TypeScript: client_function_composition.ts
  • JavaScript: client_function_composition.js
  • Go: client_functions.go
  • Kotlin: ClientFunctionComposition.kt

See User Functions CRUD examples:

  • Rust: client_user_functions.rs
  • Python: client_user_functions.py
  • TypeScript: client_user_functions.ts
  • Go: client_user_functions.go
  • Kotlin: ClientUserFunctions.kt

WebSocket Chat Streaming​

Stream real-time LLM responses via WebSocket. The server sends ChatStreamEvent messages as the model generates text, calls tools, or completes.

Event Types​

EventDescription
chunkA text token from the LLM (stream these as they arrive)
endStream completed — includes messageId, executionTimeMs, tokenUsage, contextWindow
toolCallThe LLM wants to execute a client-side tool
errorAn error occurred during streaming

Streaming Example​

let ws = client.websocket("ws://localhost:8080").await?;

let mut stream = ws.chat_send_with_tools(
    &chat_id, "What is the capital of France?",
    None, None, None, None,
).await?;

while let Some(event) = stream.recv().await {
    match event {
        ChatStreamEvent::Chunk(text) => print!("{}", text),
        ChatStreamEvent::End { execution_time_ms, context_window, .. } => {
            println!("\nDone in {}ms", execution_time_ms);
            if let Some(cw) = context_window {
                println!("Context window: {} tokens", cw);
            }
        }
        ChatStreamEvent::ToolCall { tool_name, call_id, arguments, .. } => {
            println!("[Tool] {}", tool_name);
            let result = serde_json::json!({"result": "done"});
            ws.send_tool_result(&chat_id, &call_id, true, Some(result), None).await?;
        }
        ChatStreamEvent::Error(err) => eprintln!("Error: {}", err),
    }
}

Goals, Tasks & Agents​

Manage AI planning workflows with goals (multi-step plans), tasks (scheduled/triggered jobs), and agents (named AI profiles).

Goals​

Goals are multi-step plans the AI can create, execute, and track. Each goal has steps that move through a lifecycle: active → pending_review → approved/rejected.

// Create a goal
let goal = client.goal_create(serde_json::json!({
    "title": "Migrate user data",
    "description": "Move users from legacy to new schema",
    "status": "active",
})).await?;

// List and search
let goals = client.goal_list().await?;
let results = client.goal_search("migrate").await?;

// Lifecycle
client.goal_complete(&goal_id, serde_json::json!({"summary": "Done"})).await?;
client.goal_approve(&goal_id).await?;

// Steps
client.goal_step_start(&goal_id, 0).await?;
client.goal_step_complete(&goal_id, 0, serde_json::json!({"result": "done"})).await?;

Tasks​

Tasks represent scheduled or triggered jobs with lifecycle management.

MethodDescription
taskCreateCreate a new task
taskList / taskGetList or get task details
taskDue(now)Get tasks due at a given time
taskStart / taskPause / taskResumeLifecycle transitions
taskSucceed / taskFailTerminal states with result/error data
taskDeleteRemove a task

Agents​

Agents are named AI profiles with specific models and configurations.

MethodDescription
agentCreateCreate a new agent profile
agentList / agentGetList or get agent details
agentGetByNameLook up by name
agentUpdate / agentDeleteModify or remove
agentsByDeploymentList agents on a specific deployment

Schedule Management​

Manage cron-based scheduled execution of functions and tasks.

let sched = client.create_schedule(serde_json::json!({
    "name": "nightly-backup",
    "cron": "0 2 * * *",
    "task_type": "backup",
})).await?;

let sched_id = sched["id"].as_str().unwrap();

let schedules = client.list_schedules().await?;
client.pause_schedule(&sched_id).await?;
client.resume_schedule(&sched_id).await?;
client.delete_schedule(&sched_id).await?;

Next Steps​

Need Help?​