gitops/optimclaw
1
0
Fork 0
mirror of https://github.com/outbackdingo/optimclaw.git synced 2026-08-30 16:19:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: add libSQL/Turso database backend with full feature parity

Browse Source 
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]>
This commit is contained in:
Zaki
2026-02-12 04:35:05 -08:00
co-authored by Claude Opus 4.6
parent 115b7f38fe
commit 0de6f6aabb
44 changed files with 6282 additions and 285 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/workspace/mod.rs
+221 -24
View File
@@ -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,14 @@ 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 +477,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 +496,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 +592,7 @@ impl Workspace {
None
};
self.repo
self.storage
.hybrid_search(
&self.user_id,
self.agent_id,
@@ -411,13 +608,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 +631,7 @@ impl Workspace {
None
};
self.repo
self.storage
.insert_chunk(document_id, index as i32, &content, embedding.as_deref())
.await?;
}
@@ -542,7 +739,7 @@ impl Workspace {
};
let chunks = self
.repo
.storage
.get_chunks_without_embeddings(&self.user_id, self.agent_id, 100)
.await?;
@@ -550,7 +747,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;