mirror of
https://github.com/outbackdingo/optimclaw.git
synced 2026-09-01 00:59:33 +00:00
feat: add libSQL/Turso embedded database backend (#47)
* feat: add libSQL/Turso database backend with full feature parity Introduce a Database trait abstraction (~60 async methods) enabling compile-time backend selection between PostgreSQL and libSQL/Turso. Convert all modules from concrete Store to Arc<dyn Database>, add LibSqlSecretsStore and LibSqlWasmToolStore implementations, wire libsql stores throughout CLI and main entry points, and make the setup wizard backend-agnostic. Key changes: - src/db/: Database trait, PostgresDatabase adapter, LibSqlBackend with native SQLite-dialect SQL, and idempotent migration system - src/secrets/store.rs: LibSqlSecretsStore (all 8 trait methods) - src/tools/wasm/storage.rs: LibSqlWasmToolStore (all 7 trait methods) - src/main.rs, cli/tool.rs, cli/mcp.rs: backend-conditional wiring - src/setup/channels.rs: SecretsContext uses Arc<dyn SecretsStore> - Feature-gate postgres-only tests and examples Co-Authored-By: Claude Opus 4.6 <[email protected]> * feat: enable onboarding wizard for libSQL builds Refactor the setup wizard to work with both postgres and libsql feature flags. Previously the wizard was gated behind #[cfg(feature = "postgres")] only, so libsql-only builds would print an error on `ironclaw onboard`. - Add libsql fields to Settings (database_backend, libsql_path, libsql_url) - Split wizard database/migration/secrets methods into feature-gated variants - Add step_database_libsql() with local path and Turso remote replica prompts - Update setup/mod.rs and main.rs feature gates to any(postgres, libsql) - Extend check_onboard_needed() to detect libsql database presence Co-Authored-By: Claude Opus 4.6 <[email protected]> * fix: address PR review feedback for libSQL backend - P0: Switch libsql_backend to connection-per-operation pattern to fix shared Connection concurrency issue across tokio tasks - P0: Wrap secrets store INSERT+SELECT in transaction to fix TOCTOU race - P0: Document encryption-at-rest limitations and json_patch divergence - P1: Fix get_opt_text removing .filter(|s| !s.is_empty()) that conflated empty strings with NULL - P1: Replace datetime('now') with fmt_ts(&Utc::now()) for consistent RFC 3339 timestamps across all queries - P2: Use explicit _rowid column in FTS5 triggers and joins for stability across VACUUM operations - P2: Add tracing::warn when embedding provided but vector search disabled in hybrid_search - Extract shared connect_from_config() helper to deduplicate DB connection logic across main.rs, cli/config.rs, and cli/mcp.rs Co-Authored-By: Claude Opus 4.6 <[email protected]> * fix: add missing JobContext fields and resolve fmt/clippy warnings Add total_tokens_used and max_tokens fields to JobContext in libsql_backend.rs, apply cargo fmt, and fix clippy warnings. Co-Authored-By: Claude Opus 4.6 <[email protected]> * fix: review fixes for libSQL backend (shared connections, panics, indexes) - Replace .expect() with proper error propagation in 3 call sites - Share Arc<Database> between backend and stores instead of single Connection - Add connect-per-operation pattern to LibSqlSecretsStore and LibSqlWasmToolStore - Wrap store() INSERT + SELECT-back in a transaction - Add ~22 missing indexes for parity with PostgreSQL schema - Add 18 leak_detection_patterns seed rows matching PostgreSQL V2 migration - Fix super:: import to use crate:: style - Gate mask_password_in_url behind #[cfg(feature = "postgres")] - Rewrite secrets store init with or_else chain for runtime backend selection Co-Authored-By: Claude Opus 4.6 <[email protected]> * fix: Resolve clippy lints (collapsible_if, too_many_arguments) Collapse nested if blocks into let_chains to satisfy clippy's collapsible_if lint (CI uses -D warnings). Suppress too_many_arguments on libsql_row_to_tool_at since refactoring the positional index pattern would be a larger change. Co-Authored-By: Claude Opus 4.6 <[email protected]> --------- Co-authored-by: Claude Opus 4.6 <[email protected]> Co-authored-by: Illia Polosukhin <[email protected]>
This commit is contained in:
Zaki Manian
co-authored by
Claude Opus 4.6
Illia Polosukhin
Claude Opus 4.6
Illia Polosukhin
parent
54e9206f0b
commit
e843c18141
+223
-24
@@ -43,23 +43,206 @@
|
||||
mod chunker;
|
||||
mod document;
|
||||
mod embeddings;
|
||||
#[cfg(feature = "postgres")]
|
||||
mod repository;
|
||||
mod search;
|
||||
|
||||
pub use chunker::{ChunkConfig, chunk_document};
|
||||
pub use document::{MemoryChunk, MemoryDocument, WorkspaceEntry, paths};
|
||||
pub use embeddings::{EmbeddingProvider, MockEmbeddings, NearAiEmbeddings, OpenAiEmbeddings};
|
||||
#[cfg(feature = "postgres")]
|
||||
pub use repository::Repository;
|
||||
pub use search::{SearchConfig, SearchResult};
|
||||
pub use search::{RankedResult, SearchConfig, SearchResult, reciprocal_rank_fusion};
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use chrono::{NaiveDate, Utc};
|
||||
#[cfg(feature = "postgres")]
|
||||
use deadpool_postgres::Pool;
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::error::WorkspaceError;
|
||||
|
||||
/// Internal storage abstraction for Workspace.
|
||||
///
|
||||
/// Allows Workspace to work with either a PostgreSQL `Repository` (the original
|
||||
/// path) or any `Database` trait implementation (e.g. libSQL backend).
|
||||
enum WorkspaceStorage {
|
||||
/// PostgreSQL-backed repository (uses connection pool directly).
|
||||
#[cfg(feature = "postgres")]
|
||||
Repo(Repository),
|
||||
/// Generic backend implementing the Database trait.
|
||||
Db(Arc<dyn crate::db::Database>),
|
||||
}
|
||||
|
||||
impl WorkspaceStorage {
|
||||
async fn get_document_by_path(
|
||||
&self,
|
||||
user_id: &str,
|
||||
agent_id: Option<Uuid>,
|
||||
path: &str,
|
||||
) -> Result<MemoryDocument, WorkspaceError> {
|
||||
match self {
|
||||
#[cfg(feature = "postgres")]
|
||||
Self::Repo(repo) => repo.get_document_by_path(user_id, agent_id, path).await,
|
||||
Self::Db(db) => db.get_document_by_path(user_id, agent_id, path).await,
|
||||
}
|
||||
}
|
||||
|
||||
async fn get_document_by_id(&self, id: Uuid) -> Result<MemoryDocument, WorkspaceError> {
|
||||
match self {
|
||||
#[cfg(feature = "postgres")]
|
||||
Self::Repo(repo) => repo.get_document_by_id(id).await,
|
||||
Self::Db(db) => db.get_document_by_id(id).await,
|
||||
}
|
||||
}
|
||||
|
||||
async fn get_or_create_document_by_path(
|
||||
&self,
|
||||
user_id: &str,
|
||||
agent_id: Option<Uuid>,
|
||||
path: &str,
|
||||
) -> Result<MemoryDocument, WorkspaceError> {
|
||||
match self {
|
||||
#[cfg(feature = "postgres")]
|
||||
Self::Repo(repo) => {
|
||||
repo.get_or_create_document_by_path(user_id, agent_id, path)
|
||||
.await
|
||||
}
|
||||
Self::Db(db) => {
|
||||
db.get_or_create_document_by_path(user_id, agent_id, path)
|
||||
.await
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn update_document(&self, id: Uuid, content: &str) -> Result<(), WorkspaceError> {
|
||||
match self {
|
||||
#[cfg(feature = "postgres")]
|
||||
Self::Repo(repo) => repo.update_document(id, content).await,
|
||||
Self::Db(db) => db.update_document(id, content).await,
|
||||
}
|
||||
}
|
||||
|
||||
async fn delete_document_by_path(
|
||||
&self,
|
||||
user_id: &str,
|
||||
agent_id: Option<Uuid>,
|
||||
path: &str,
|
||||
) -> Result<(), WorkspaceError> {
|
||||
match self {
|
||||
#[cfg(feature = "postgres")]
|
||||
Self::Repo(repo) => repo.delete_document_by_path(user_id, agent_id, path).await,
|
||||
Self::Db(db) => db.delete_document_by_path(user_id, agent_id, path).await,
|
||||
}
|
||||
}
|
||||
|
||||
async fn list_directory(
|
||||
&self,
|
||||
user_id: &str,
|
||||
agent_id: Option<Uuid>,
|
||||
directory: &str,
|
||||
) -> Result<Vec<WorkspaceEntry>, WorkspaceError> {
|
||||
match self {
|
||||
#[cfg(feature = "postgres")]
|
||||
Self::Repo(repo) => repo.list_directory(user_id, agent_id, directory).await,
|
||||
Self::Db(db) => db.list_directory(user_id, agent_id, directory).await,
|
||||
}
|
||||
}
|
||||
|
||||
async fn list_all_paths(
|
||||
&self,
|
||||
user_id: &str,
|
||||
agent_id: Option<Uuid>,
|
||||
) -> Result<Vec<String>, WorkspaceError> {
|
||||
match self {
|
||||
#[cfg(feature = "postgres")]
|
||||
Self::Repo(repo) => repo.list_all_paths(user_id, agent_id).await,
|
||||
Self::Db(db) => db.list_all_paths(user_id, agent_id).await,
|
||||
}
|
||||
}
|
||||
|
||||
async fn delete_chunks(&self, document_id: Uuid) -> Result<(), WorkspaceError> {
|
||||
match self {
|
||||
#[cfg(feature = "postgres")]
|
||||
Self::Repo(repo) => repo.delete_chunks(document_id).await,
|
||||
Self::Db(db) => db.delete_chunks(document_id).await,
|
||||
}
|
||||
}
|
||||
|
||||
async fn insert_chunk(
|
||||
&self,
|
||||
document_id: Uuid,
|
||||
chunk_index: i32,
|
||||
content: &str,
|
||||
embedding: Option<&[f32]>,
|
||||
) -> Result<Uuid, WorkspaceError> {
|
||||
match self {
|
||||
#[cfg(feature = "postgres")]
|
||||
Self::Repo(repo) => {
|
||||
repo.insert_chunk(document_id, chunk_index, content, embedding)
|
||||
.await
|
||||
}
|
||||
Self::Db(db) => {
|
||||
db.insert_chunk(document_id, chunk_index, content, embedding)
|
||||
.await
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn update_chunk_embedding(
|
||||
&self,
|
||||
chunk_id: Uuid,
|
||||
embedding: &[f32],
|
||||
) -> Result<(), WorkspaceError> {
|
||||
match self {
|
||||
#[cfg(feature = "postgres")]
|
||||
Self::Repo(repo) => repo.update_chunk_embedding(chunk_id, embedding).await,
|
||||
Self::Db(db) => db.update_chunk_embedding(chunk_id, embedding).await,
|
||||
}
|
||||
}
|
||||
|
||||
async fn get_chunks_without_embeddings(
|
||||
&self,
|
||||
user_id: &str,
|
||||
agent_id: Option<Uuid>,
|
||||
limit: usize,
|
||||
) -> Result<Vec<MemoryChunk>, WorkspaceError> {
|
||||
match self {
|
||||
#[cfg(feature = "postgres")]
|
||||
Self::Repo(repo) => {
|
||||
repo.get_chunks_without_embeddings(user_id, agent_id, limit)
|
||||
.await
|
||||
}
|
||||
Self::Db(db) => {
|
||||
db.get_chunks_without_embeddings(user_id, agent_id, limit)
|
||||
.await
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn hybrid_search(
|
||||
&self,
|
||||
user_id: &str,
|
||||
agent_id: Option<Uuid>,
|
||||
query: &str,
|
||||
embedding: Option<&[f32]>,
|
||||
config: &SearchConfig,
|
||||
) -> Result<Vec<SearchResult>, WorkspaceError> {
|
||||
match self {
|
||||
#[cfg(feature = "postgres")]
|
||||
Self::Repo(repo) => {
|
||||
repo.hybrid_search(user_id, agent_id, query, embedding, config)
|
||||
.await
|
||||
}
|
||||
Self::Db(db) => {
|
||||
db.hybrid_search(user_id, agent_id, query, embedding, config)
|
||||
.await
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Default template seeded into HEARTBEAT.md on first access.
|
||||
///
|
||||
/// Intentionally comment-only so the heartbeat runner treats it as
|
||||
@@ -80,25 +263,39 @@ const HEARTBEAT_SEED: &str = "\
|
||||
/// Workspace provides database-backed memory storage for an agent.
|
||||
///
|
||||
/// Each workspace is scoped to a user (and optionally an agent).
|
||||
/// Documents are persisted to PostgreSQL and indexed for search.
|
||||
/// Documents are persisted to the database and indexed for search.
|
||||
/// Supports both PostgreSQL (via Repository) and libSQL (via Database trait).
|
||||
pub struct Workspace {
|
||||
/// User identifier (from channel).
|
||||
user_id: String,
|
||||
/// Optional agent ID for multi-agent isolation.
|
||||
agent_id: Option<Uuid>,
|
||||
/// Database repository.
|
||||
repo: Repository,
|
||||
/// Database storage backend.
|
||||
storage: WorkspaceStorage,
|
||||
/// Embedding provider for semantic search.
|
||||
embeddings: Option<Arc<dyn EmbeddingProvider>>,
|
||||
}
|
||||
|
||||
impl Workspace {
|
||||
/// Create a new workspace for a user.
|
||||
/// Create a new workspace backed by a PostgreSQL connection pool.
|
||||
#[cfg(feature = "postgres")]
|
||||
pub fn new(user_id: impl Into<String>, pool: Pool) -> Self {
|
||||
Self {
|
||||
user_id: user_id.into(),
|
||||
agent_id: None,
|
||||
repo: Repository::new(pool),
|
||||
storage: WorkspaceStorage::Repo(Repository::new(pool)),
|
||||
embeddings: None,
|
||||
}
|
||||
}
|
||||
|
||||
/// Create a new workspace backed by any Database implementation.
|
||||
///
|
||||
/// Use this for libSQL or any other backend that implements the Database trait.
|
||||
pub fn new_with_db(user_id: impl Into<String>, db: Arc<dyn crate::db::Database>) -> Self {
|
||||
Self {
|
||||
user_id: user_id.into(),
|
||||
agent_id: None,
|
||||
storage: WorkspaceStorage::Db(db),
|
||||
embeddings: None,
|
||||
}
|
||||
}
|
||||
@@ -138,7 +335,7 @@ impl Workspace {
|
||||
/// ```
|
||||
pub async fn read(&self, path: &str) -> Result<MemoryDocument, WorkspaceError> {
|
||||
let path = normalize_path(path);
|
||||
self.repo
|
||||
self.storage
|
||||
.get_document_by_path(&self.user_id, self.agent_id, &path)
|
||||
.await
|
||||
}
|
||||
@@ -155,14 +352,14 @@ impl Workspace {
|
||||
pub async fn write(&self, path: &str, content: &str) -> Result<MemoryDocument, WorkspaceError> {
|
||||
let path = normalize_path(path);
|
||||
let doc = self
|
||||
.repo
|
||||
.storage
|
||||
.get_or_create_document_by_path(&self.user_id, self.agent_id, &path)
|
||||
.await?;
|
||||
self.repo.update_document(doc.id, content).await?;
|
||||
self.storage.update_document(doc.id, content).await?;
|
||||
self.reindex_document(doc.id).await?;
|
||||
|
||||
// Return updated doc
|
||||
self.repo.get_document_by_id(doc.id).await
|
||||
self.storage.get_document_by_id(doc.id).await
|
||||
}
|
||||
|
||||
/// Append content to a file.
|
||||
@@ -172,7 +369,7 @@ impl Workspace {
|
||||
pub async fn append(&self, path: &str, content: &str) -> Result<(), WorkspaceError> {
|
||||
let path = normalize_path(path);
|
||||
let doc = self
|
||||
.repo
|
||||
.storage
|
||||
.get_or_create_document_by_path(&self.user_id, self.agent_id, &path)
|
||||
.await?;
|
||||
|
||||
@@ -182,7 +379,7 @@ impl Workspace {
|
||||
format!("{}\n{}", doc.content, content)
|
||||
};
|
||||
|
||||
self.repo.update_document(doc.id, &new_content).await?;
|
||||
self.storage.update_document(doc.id, &new_content).await?;
|
||||
self.reindex_document(doc.id).await?;
|
||||
Ok(())
|
||||
}
|
||||
@@ -191,7 +388,7 @@ impl Workspace {
|
||||
pub async fn exists(&self, path: &str) -> Result<bool, WorkspaceError> {
|
||||
let path = normalize_path(path);
|
||||
match self
|
||||
.repo
|
||||
.storage
|
||||
.get_document_by_path(&self.user_id, self.agent_id, &path)
|
||||
.await
|
||||
{
|
||||
@@ -206,7 +403,7 @@ impl Workspace {
|
||||
/// Also deletes associated chunks.
|
||||
pub async fn delete(&self, path: &str) -> Result<(), WorkspaceError> {
|
||||
let path = normalize_path(path);
|
||||
self.repo
|
||||
self.storage
|
||||
.delete_document_by_path(&self.user_id, self.agent_id, &path)
|
||||
.await
|
||||
}
|
||||
@@ -229,14 +426,16 @@ impl Workspace {
|
||||
/// ```
|
||||
pub async fn list(&self, directory: &str) -> Result<Vec<WorkspaceEntry>, WorkspaceError> {
|
||||
let directory = normalize_directory(directory);
|
||||
self.repo
|
||||
self.storage
|
||||
.list_directory(&self.user_id, self.agent_id, &directory)
|
||||
.await
|
||||
}
|
||||
|
||||
/// List all files recursively (flat list of all paths).
|
||||
pub async fn list_all(&self) -> Result<Vec<String>, WorkspaceError> {
|
||||
self.repo.list_all_paths(&self.user_id, self.agent_id).await
|
||||
self.storage
|
||||
.list_all_paths(&self.user_id, self.agent_id)
|
||||
.await
|
||||
}
|
||||
|
||||
// ==================== Convenience Methods ====================
|
||||
@@ -280,7 +479,7 @@ impl Workspace {
|
||||
|
||||
/// Helper to read or create a file.
|
||||
async fn read_or_create(&self, path: &str) -> Result<MemoryDocument, WorkspaceError> {
|
||||
self.repo
|
||||
self.storage
|
||||
.get_or_create_document_by_path(&self.user_id, self.agent_id, path)
|
||||
.await
|
||||
}
|
||||
@@ -299,7 +498,7 @@ impl Workspace {
|
||||
} else {
|
||||
format!("{}\n\n{}", doc.content, entry)
|
||||
};
|
||||
self.repo.update_document(doc.id, &new_content).await?;
|
||||
self.storage.update_document(doc.id, &new_content).await?;
|
||||
self.reindex_document(doc.id).await?;
|
||||
Ok(())
|
||||
}
|
||||
@@ -395,7 +594,7 @@ impl Workspace {
|
||||
None
|
||||
};
|
||||
|
||||
self.repo
|
||||
self.storage
|
||||
.hybrid_search(
|
||||
&self.user_id,
|
||||
self.agent_id,
|
||||
@@ -411,13 +610,13 @@ impl Workspace {
|
||||
/// Re-index a document (chunk and generate embeddings).
|
||||
async fn reindex_document(&self, document_id: Uuid) -> Result<(), WorkspaceError> {
|
||||
// Get the document
|
||||
let doc = self.repo.get_document_by_id(document_id).await?;
|
||||
let doc = self.storage.get_document_by_id(document_id).await?;
|
||||
|
||||
// Chunk the content
|
||||
let chunks = chunk_document(&doc.content, ChunkConfig::default());
|
||||
|
||||
// Delete old chunks
|
||||
self.repo.delete_chunks(document_id).await?;
|
||||
self.storage.delete_chunks(document_id).await?;
|
||||
|
||||
// Insert new chunks
|
||||
for (index, content) in chunks.into_iter().enumerate() {
|
||||
@@ -434,7 +633,7 @@ impl Workspace {
|
||||
None
|
||||
};
|
||||
|
||||
self.repo
|
||||
self.storage
|
||||
.insert_chunk(document_id, index as i32, &content, embedding.as_deref())
|
||||
.await?;
|
||||
}
|
||||
@@ -542,7 +741,7 @@ impl Workspace {
|
||||
};
|
||||
|
||||
let chunks = self
|
||||
.repo
|
||||
.storage
|
||||
.get_chunks_without_embeddings(&self.user_id, self.agent_id, 100)
|
||||
.await?;
|
||||
|
||||
@@ -550,7 +749,7 @@ impl Workspace {
|
||||
for chunk in chunks {
|
||||
match provider.embed(&chunk.content).await {
|
||||
Ok(embedding) => {
|
||||
self.repo
|
||||
self.storage
|
||||
.update_chunk_embedding(chunk.id, &embedding)
|
||||
.await?;
|
||||
count += 1;
|
||||
|
||||
Reference in New Issue
Block a user