From 7474fd4c529b3978fe737d3bb86311fc43af8e72 Mon Sep 17 00:00:00 2001 From: Zaki Date: Thu, 12 Feb 2026 07:02:21 -0800 Subject: [PATCH] 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 --- CLAUDE.md | 2 + src/cli/config.rs | 39 +--- src/cli/mcp.rs | 28 +-- src/db/libsql_backend.rs | 417 +++++++++++++++++++----------------- src/db/libsql_migrations.rs | 15 +- src/db/mod.rs | 55 +++++ src/main.rs | 43 +--- src/secrets/store.rs | 17 +- 8 files changed, 317 insertions(+), 299 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index fa564c7f..aeaf1dd2 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -409,6 +409,8 @@ LIBSQL_AUTH_TOKEN=your-token # Required when LIBSQL_URL is set - **Hybrid search** uses FTS5 only (vector search via libsql_vector_idx not yet implemented) - **Settings reload from DB** skipped (Config::from_db requires Store) - No incremental migration versioning (schema is CREATE IF NOT EXISTS, no ALTER TABLE support yet) +- **No encryption at rest** -- The local SQLite database file stores conversation content, job data, workspace memory, and other application data in plaintext. Only secrets (API tokens, credentials) are encrypted via AES-256-GCM before storage. Users handling sensitive data should use full-disk encryption (FileVault, LUKS, BitLocker) or consider the PostgreSQL backend with TDE/encrypted storage. +- **JSON merge patch vs path-targeted update** -- The libSQL backend uses RFC 7396 JSON Merge Patch (`json_patch`) for metadata updates, while PostgreSQL uses path-targeted `jsonb_set`. Merge patch replaces top-level keys entirely, which may drop nested keys not present in the patch. Callers should avoid relying on partial nested object updates in metadata fields. ## Safety Layer diff --git a/src/cli/config.rs b/src/cli/config.rs index e47e7f54..f6b52516 100644 --- a/src/cli/config.rs +++ b/src/cli/config.rs @@ -3,6 +3,8 @@ //! Commands for viewing and modifying settings. //! Settings are stored in the database (env > DB > default). +use std::sync::Arc; + use clap::Subcommand; use crate::settings::Settings; @@ -49,7 +51,7 @@ pub async fn run_config_command(cmd: ConfigCommand) -> anyhow::Result<()> { let _ = dotenvy::dotenv(); // Try to connect to the DB for settings access - let db: Option> = match connect_db().await { + let db: Option> = match connect_db().await { Ok(d) => Some(d), Err(e) => { eprintln!( @@ -71,38 +73,11 @@ pub async fn run_config_command(cmd: ConfigCommand) -> anyhow::Result<()> { } /// Bootstrap a DB connection for config commands (backend-agnostic). -async fn connect_db() -> anyhow::Result> { - use crate::db::Database as _; +async fn connect_db() -> anyhow::Result> { let config = crate::config::Config::from_env().await.map_err(|e| anyhow::anyhow!("{}", e))?; - - match config.database.backend { - #[cfg(feature = "libsql")] - crate::config::DatabaseBackend::LibSql => { - use secrecy::ExposeSecret as _; - let default_path = crate::config::default_libsql_path(); - let db_path = config.database.libsql_path.as_deref().unwrap_or(&default_path); - let backend = if let Some(ref url) = config.database.libsql_url { - let token = config.database.libsql_auth_token.as_ref() - .ok_or_else(|| anyhow::anyhow!("LIBSQL_AUTH_TOKEN required when LIBSQL_URL is set"))?; - crate::db::libsql_backend::LibSqlBackend::new_remote_replica(db_path, url, token.expose_secret()).await? - } else { - crate::db::libsql_backend::LibSqlBackend::new_local(db_path).await? - }; - backend.run_migrations().await?; - Ok(Box::new(backend)) - } - #[cfg(feature = "postgres")] - _ => { - let pg = crate::db::postgres::PgBackend::new(&config.database).await - .map_err(|e| anyhow::anyhow!("{}", e))?; - pg.run_migrations().await.map_err(|e| anyhow::anyhow!("{}", e))?; - Ok(Box::new(pg)) - } - #[cfg(not(feature = "postgres"))] - _ => { - anyhow::bail!("No database backend available. Enable 'postgres' or 'libsql' feature."); - } - } + crate::db::connect_from_config(&config.database) + .await + .map_err(|e| anyhow::anyhow!("{}", e)) } const DEFAULT_USER_ID: &str = "default"; diff --git a/src/cli/mcp.rs b/src/cli/mcp.rs index 7692a089..a575abf3 100644 --- a/src/cli/mcp.rs +++ b/src/cli/mcp.rs @@ -468,33 +468,9 @@ async fn toggle_server(name: String, enable: bool, disable: bool) -> anyhow::Res const DEFAULT_USER_ID: &str = "default"; /// Try to connect to the database (backend-agnostic). -async fn connect_db() -> Option> { - use crate::db::Database as _; +async fn connect_db() -> Option> { let config = Config::from_env().await.ok()?; - match config.database.backend { - #[cfg(feature = "libsql")] - crate::config::DatabaseBackend::LibSql => { - use secrecy::ExposeSecret as _; - let default_path = crate::config::default_libsql_path(); - let db_path = config.database.libsql_path.as_deref().unwrap_or(&default_path); - let backend = if let Some(ref url) = config.database.libsql_url { - let token = config.database.libsql_auth_token.as_ref()?; - crate::db::libsql_backend::LibSqlBackend::new_remote_replica(db_path, url, token.expose_secret()).await.ok()? - } else { - crate::db::libsql_backend::LibSqlBackend::new_local(db_path).await.ok()? - }; - backend.run_migrations().await.ok()?; - Some(Box::new(backend)) - } - #[cfg(feature = "postgres")] - _ => { - let pg = crate::db::postgres::PgBackend::new(&config.database).await.ok()?; - pg.run_migrations().await.ok()?; - Some(Box::new(pg)) - } - #[cfg(not(feature = "postgres"))] - _ => None, - } + crate::db::connect_from_config(&config.database).await.ok() } /// Load MCP servers (DB if available, else disk). diff --git a/src/db/libsql_backend.rs b/src/db/libsql_backend.rs index 7b129f7c..c9bebaa6 100644 --- a/src/db/libsql_backend.rs +++ b/src/db/libsql_backend.rs @@ -50,7 +50,6 @@ const ROUTINE_RUN_COLUMNS: &str = "\ /// libSQL/Turso database backend. pub struct LibSqlBackend { db: LibSqlDatabase, - conn: Connection, } impl LibSqlBackend { @@ -68,11 +67,7 @@ impl LibSqlBackend { .await .map_err(|e| DatabaseError::Pool(format!("Failed to open libSQL database: {}", e)))?; - let conn = db.connect().map_err(|e| { - DatabaseError::Pool(format!("Failed to connect to libSQL database: {}", e)) - })?; - - Ok(Self { db, conn }) + Ok(Self { db }) } /// Create a new in-memory database (for testing). @@ -82,11 +77,7 @@ impl LibSqlBackend { .await .map_err(|e| DatabaseError::Pool(format!("Failed to create in-memory database: {}", e)))?; - let conn = db.connect().map_err(|e| { - DatabaseError::Pool(format!("Failed to connect to in-memory database: {}", e)) - })?; - - Ok(Self { db, conn }) + Ok(Self { db }) } /// Create with Turso cloud sync (embedded replica). @@ -108,11 +99,7 @@ impl LibSqlBackend { DatabaseError::Pool(format!("Failed to open remote replica: {}", e)) })?; - let conn = db.connect().map_err(|e| { - DatabaseError::Pool(format!("Failed to connect to remote replica: {}", e)) - })?; - - Ok(Self { db, conn }) + Ok(Self { db }) } /// Get the underlying database handle (for sync operations). @@ -190,10 +177,9 @@ fn get_text(row: &libsql::Row, idx: i32) -> String { } /// Extract an optional text column. -/// Returns None for both SQL NULL and empty strings (since empty string was -/// previously used as a proxy for NULL in optional fields). +/// Returns None for SQL NULL, preserves empty strings as Some(""). fn get_opt_text(row: &libsql::Row, idx: i32) -> Option { - row.get::(idx).ok().filter(|s| !s.is_empty()) + row.get::(idx).ok() } /// Convert an `Option<&str>` to a `libsql::Value` (Text or Null). @@ -285,8 +271,8 @@ fn get_opt_ts(row: &libsql::Row, idx: i32) -> Option> { #[async_trait] impl Database for LibSqlBackend { async fn run_migrations(&self) -> Result<(), DatabaseError> { - self.conn - .execute_batch(libsql_migrations::SCHEMA) + let conn = self.connect()?; + conn.execute_batch(libsql_migrations::SCHEMA) .await .map_err(|e| DatabaseError::Migration(format!("libSQL migration failed: {}", e)))?; Ok(()) @@ -300,9 +286,9 @@ impl Database for LibSqlBackend { user_id: &str, thread_id: Option<&str>, ) -> Result { + let conn = self.connect()?; let id = Uuid::new_v4(); - self.conn - .execute( + conn.execute( "INSERT INTO conversations (id, channel, user_id, thread_id) VALUES (?1, ?2, ?3, ?4)", params![id.to_string(), channel, user_id, opt_text(thread_id)], ) @@ -312,10 +298,11 @@ impl Database for LibSqlBackend { } async fn touch_conversation(&self, id: Uuid) -> Result<(), DatabaseError> { - self.conn - .execute( - "UPDATE conversations SET last_activity = datetime('now') WHERE id = ?1", - params![id.to_string()], + let conn = self.connect()?; + let now = fmt_ts(&Utc::now()); + conn.execute( + "UPDATE conversations SET last_activity = ?2 WHERE id = ?1", + params![id.to_string(), now], ) .await .map_err(|e| DatabaseError::Query(e.to_string()))?; @@ -328,9 +315,9 @@ impl Database for LibSqlBackend { role: &str, content: &str, ) -> Result { + let conn = self.connect()?; let id = Uuid::new_v4(); - self.conn - .execute( + conn.execute( "INSERT INTO conversation_messages (id, conversation_id, role, content) VALUES (?1, ?2, ?3, ?4)", params![id.to_string(), conversation_id.to_string(), role, content], ) @@ -347,14 +334,15 @@ impl Database for LibSqlBackend { user_id: &str, thread_id: Option<&str>, ) -> Result<(), DatabaseError> { - self.conn - .execute( + let conn = self.connect()?; + let now = fmt_ts(&Utc::now()); + conn.execute( r#" INSERT INTO conversations (id, channel, user_id, thread_id) VALUES (?1, ?2, ?3, ?4) - ON CONFLICT (id) DO UPDATE SET last_activity = datetime('now') + ON CONFLICT (id) DO UPDATE SET last_activity = ?5 "#, - params![id.to_string(), channel, user_id, opt_text(thread_id)], + params![id.to_string(), channel, user_id, opt_text(thread_id), now], ) .await .map_err(|e| DatabaseError::Query(e.to_string()))?; @@ -367,8 +355,8 @@ impl Database for LibSqlBackend { channel: &str, limit: i64, ) -> Result, DatabaseError> { - let mut rows = self - .conn + let conn = self.connect()?; + let mut rows = conn .query( r#" SELECT @@ -417,9 +405,9 @@ impl Database for LibSqlBackend { user_id: &str, channel: &str, ) -> Result { + let conn = self.connect()?; // Try to find existing - let mut rows = self - .conn + let mut rows = conn .query( r#" SELECT id FROM conversations @@ -442,8 +430,7 @@ impl Database for LibSqlBackend { // Create new let id = Uuid::new_v4(); let metadata = serde_json::json!({"thread_type": "assistant", "title": "Assistant"}); - self.conn - .execute( + conn.execute( "INSERT INTO conversations (id, channel, user_id, metadata) VALUES (?1, ?2, ?3, ?4)", params![id.to_string(), channel, user_id, metadata.to_string()], ) @@ -458,9 +445,9 @@ impl Database for LibSqlBackend { user_id: &str, metadata: &serde_json::Value, ) -> Result { + let conn = self.connect()?; let id = Uuid::new_v4(); - self.conn - .execute( + conn.execute( "INSERT INTO conversations (id, channel, user_id, metadata) VALUES (?1, ?2, ?3, ?4)", params![id.to_string(), channel, user_id, metadata.to_string()], ) @@ -475,12 +462,12 @@ impl Database for LibSqlBackend { before: Option>, limit: i64, ) -> Result<(Vec, bool), DatabaseError> { + let conn = self.connect()?; let fetch_limit = limit + 1; let cid = conversation_id.to_string(); let mut rows = if let Some(before_ts) = before { - self.conn - .query( + conn.query( r#" SELECT id, role, content, created_at FROM conversation_messages @@ -492,8 +479,7 @@ impl Database for LibSqlBackend { ) .await } else { - self.conn - .query( + conn.query( r#" SELECT id, role, content, created_at FROM conversation_messages @@ -529,10 +515,10 @@ impl Database for LibSqlBackend { key: &str, value: &serde_json::Value, ) -> Result<(), DatabaseError> { + let conn = self.connect()?; // SQLite: use json_patch to merge the key let patch = serde_json::json!({ key: value }); - self.conn - .execute( + conn.execute( "UPDATE conversations SET metadata = json_patch(metadata, ?2) WHERE id = ?1", params![id.to_string(), patch.to_string()], ) @@ -545,8 +531,8 @@ impl Database for LibSqlBackend { &self, id: Uuid, ) -> Result, DatabaseError> { - let mut rows = self - .conn + let conn = self.connect()?; + let mut rows = conn .query( "SELECT metadata FROM conversations WHERE id = ?1", params![id.to_string()], @@ -564,8 +550,8 @@ impl Database for LibSqlBackend { &self, conversation_id: Uuid, ) -> Result, DatabaseError> { - let mut rows = self - .conn + let conn = self.connect()?; + let mut rows = conn .query( r#" SELECT id, role, content, created_at @@ -593,10 +579,11 @@ impl Database for LibSqlBackend { // ==================== Jobs ==================== async fn save_job(&self, ctx: &JobContext) -> Result<(), DatabaseError> { + let conn = self.connect()?; let status = ctx.state.to_string(); let estimated_time_secs = ctx.estimated_duration.map(|d| d.as_secs() as i64); - self.conn + conn .execute( r#" INSERT INTO agent_jobs ( @@ -642,8 +629,8 @@ impl Database for LibSqlBackend { } async fn get_job(&self, id: Uuid) -> Result, DatabaseError> { - let mut rows = self - .conn + let conn = self.connect()?; + let mut rows = conn .query( r#" SELECT id, conversation_id, title, description, category, status, user_id, @@ -695,8 +682,8 @@ impl Database for LibSqlBackend { status: JobState, failure_reason: Option<&str>, ) -> Result<(), DatabaseError> { - self.conn - .execute( + let conn = self.connect()?; + conn.execute( "UPDATE agent_jobs SET status = ?2, failure_reason = ?3 WHERE id = ?1", params![id.to_string(), status.to_string(), opt_text(failure_reason)], ) @@ -706,10 +693,11 @@ impl Database for LibSqlBackend { } async fn mark_job_stuck(&self, id: Uuid) -> Result<(), DatabaseError> { - self.conn - .execute( - "UPDATE agent_jobs SET status = 'stuck', stuck_since = datetime('now') WHERE id = ?1", - params![id.to_string()], + let conn = self.connect()?; + let now = fmt_ts(&Utc::now()); + conn.execute( + "UPDATE agent_jobs SET status = 'stuck', stuck_since = ?2 WHERE id = ?1", + params![id.to_string(), now], ) .await .map_err(|e| DatabaseError::Query(e.to_string()))?; @@ -717,8 +705,8 @@ impl Database for LibSqlBackend { } async fn get_stuck_jobs(&self) -> Result, DatabaseError> { - let mut rows = self - .conn + let conn = self.connect()?; + let mut rows = conn .query("SELECT id FROM agent_jobs WHERE status = 'stuck'", ()) .await .map_err(|e| DatabaseError::Query(e.to_string()))?; @@ -741,12 +729,12 @@ impl Database for LibSqlBackend { job_id: Uuid, action: &ActionRecord, ) -> Result<(), DatabaseError> { + let conn = self.connect()?; let duration_ms = action.duration.as_millis() as i64; let warnings_json = serde_json::to_string(&action.sanitization_warnings) .map_err(|e| DatabaseError::Serialization(e.to_string()))?; - self.conn - .execute( + conn.execute( r#" INSERT INTO job_actions ( id, job_id, sequence_num, tool_name, input, output_raw, output_sanitized, @@ -778,8 +766,8 @@ impl Database for LibSqlBackend { &self, job_id: Uuid, ) -> Result, DatabaseError> { - let mut rows = self - .conn + let conn = self.connect()?; + let mut rows = conn .query( r#" SELECT id, sequence_num, tool_name, input, output_raw, output_sanitized, @@ -816,9 +804,9 @@ impl Database for LibSqlBackend { // ==================== LLM Calls ==================== async fn record_llm_call(&self, record: &LlmCallRecord<'_>) -> Result { + let conn = self.connect()?; let id = Uuid::new_v4(); - self.conn - .execute( + conn.execute( r#" INSERT INTO llm_calls (id, job_id, conversation_id, provider, model, input_tokens, output_tokens, cost, purpose) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9) @@ -851,12 +839,12 @@ impl Database for LibSqlBackend { estimated_time_secs: i32, estimated_value: Decimal, ) -> Result { + let conn = self.connect()?; let id = Uuid::new_v4(); let tools_json = serde_json::to_string(tool_names) .map_err(|e| DatabaseError::Serialization(e.to_string()))?; - self.conn - .execute( + conn.execute( r#" INSERT INTO estimation_snapshots (id, job_id, category, tool_names, estimated_cost, estimated_time_secs, estimated_value) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7) @@ -883,8 +871,8 @@ impl Database for LibSqlBackend { actual_time_secs: i32, actual_value: Option, ) -> Result<(), DatabaseError> { - self.conn - .execute( + let conn = self.connect()?; + conn.execute( "UPDATE estimation_snapshots SET actual_cost = ?2, actual_time_secs = ?3, actual_value = ?4 WHERE id = ?1", params![ id.to_string(), @@ -901,8 +889,8 @@ impl Database for LibSqlBackend { // ==================== Sandbox Jobs ==================== async fn save_sandbox_job(&self, job: &SandboxJobRecord) -> Result<(), DatabaseError> { - self.conn - .execute( + let conn = self.connect()?; + conn.execute( r#" INSERT INTO agent_jobs ( id, title, description, status, source, user_id, project_dir, @@ -937,8 +925,8 @@ impl Database for LibSqlBackend { &self, id: Uuid, ) -> Result, DatabaseError> { - let mut rows = self - .conn + let conn = self.connect()?; + let mut rows = conn .query( r#" SELECT id, title, status, user_id, project_dir, @@ -968,8 +956,8 @@ impl Database for LibSqlBackend { } async fn list_sandbox_jobs(&self) -> Result, DatabaseError> { - let mut rows = self - .conn + let conn = self.connect()?; + let mut rows = conn .query( r#" SELECT id, title, status, user_id, project_dir, @@ -1009,8 +997,8 @@ impl Database for LibSqlBackend { started_at: Option>, completed_at: Option>, ) -> Result<(), DatabaseError> { - self.conn - .execute( + let conn = self.connect()?; + conn.execute( r#" UPDATE agent_jobs SET status = ?2, @@ -1035,17 +1023,18 @@ impl Database for LibSqlBackend { } async fn cleanup_stale_sandbox_jobs(&self) -> Result { - let count = self - .conn + let conn = self.connect()?; + let now = fmt_ts(&Utc::now()); + let count = conn .execute( r#" UPDATE agent_jobs SET status = 'interrupted', failure_reason = 'Process restarted', - completed_at = datetime('now') + completed_at = ?1 WHERE source = 'sandbox' AND status IN ('running', 'creating') "#, - (), + params![now], ) .await .map_err(|e| DatabaseError::Query(e.to_string()))?; @@ -1056,8 +1045,8 @@ impl Database for LibSqlBackend { } async fn sandbox_job_summary(&self) -> Result { - let mut rows = self - .conn + let conn = self.connect()?; + let mut rows = conn .query( "SELECT status, COUNT(*) as cnt FROM agent_jobs WHERE source = 'sandbox' GROUP BY status", (), @@ -1087,8 +1076,8 @@ impl Database for LibSqlBackend { id: Uuid, mode: &str, ) -> Result<(), DatabaseError> { - self.conn - .execute( + let conn = self.connect()?; + conn.execute( "UPDATE agent_jobs SET job_mode = ?2 WHERE id = ?1", params![id.to_string(), mode], ) @@ -1101,8 +1090,8 @@ impl Database for LibSqlBackend { &self, id: Uuid, ) -> Result, DatabaseError> { - let mut rows = self - .conn + let conn = self.connect()?; + let mut rows = conn .query( "SELECT job_mode FROM agent_jobs WHERE id = ?1", params![id.to_string()], @@ -1124,8 +1113,8 @@ impl Database for LibSqlBackend { event_type: &str, data: &serde_json::Value, ) -> Result<(), DatabaseError> { - self.conn - .execute( + let conn = self.connect()?; + conn.execute( "INSERT INTO job_events (job_id, event_type, data) VALUES (?1, ?2, ?3)", params![job_id.to_string(), event_type, data.to_string()], ) @@ -1138,8 +1127,8 @@ impl Database for LibSqlBackend { &self, job_id: Uuid, ) -> Result, DatabaseError> { - let mut rows = self - .conn + let conn = self.connect()?; + let mut rows = conn .query( r#" SELECT id, job_id, event_type, data, created_at @@ -1166,6 +1155,7 @@ impl Database for LibSqlBackend { // ==================== Routines ==================== async fn create_routine(&self, routine: &Routine) -> Result<(), DatabaseError> { + let conn = self.connect()?; let trigger_type = routine.trigger.type_tag(); let trigger_config = routine.trigger.to_config_json(); let action_type = routine.action.type_tag(); @@ -1177,8 +1167,7 @@ impl Database for LibSqlBackend { .dedup_window .map(|d| d.as_secs() as i64); - self.conn - .execute( + conn.execute( r#" INSERT INTO routines ( id, name, description, user_id, enabled, @@ -1224,8 +1213,8 @@ impl Database for LibSqlBackend { } async fn get_routine(&self, id: Uuid) -> Result, DatabaseError> { - let mut rows = self - .conn + let conn = self.connect()?; + let mut rows = conn .query( &format!("SELECT {} FROM routines WHERE id = ?1", ROUTINE_COLUMNS), params![id.to_string()], @@ -1244,8 +1233,8 @@ impl Database for LibSqlBackend { user_id: &str, name: &str, ) -> Result, DatabaseError> { - let mut rows = self - .conn + let conn = self.connect()?; + let mut rows = conn .query( &format!("SELECT {} FROM routines WHERE user_id = ?1 AND name = ?2", ROUTINE_COLUMNS), params![user_id, name], @@ -1260,8 +1249,8 @@ impl Database for LibSqlBackend { } async fn list_routines(&self, user_id: &str) -> Result, DatabaseError> { - let mut rows = self - .conn + let conn = self.connect()?; + let mut rows = conn .query( &format!("SELECT {} FROM routines WHERE user_id = ?1 ORDER BY name", ROUTINE_COLUMNS), params![user_id], @@ -1277,8 +1266,8 @@ impl Database for LibSqlBackend { } async fn list_event_routines(&self) -> Result, DatabaseError> { - let mut rows = self - .conn + let conn = self.connect()?; + let mut rows = conn .query( &format!("SELECT {} FROM routines WHERE enabled = 1 AND trigger_type = 'event'", ROUTINE_COLUMNS), (), @@ -1294,9 +1283,9 @@ impl Database for LibSqlBackend { } async fn list_due_cron_routines(&self) -> Result, DatabaseError> { + let conn = self.connect()?; let now = fmt_ts(&Utc::now()); - let mut rows = self - .conn + let mut rows = conn .query( &format!( "SELECT {} FROM routines WHERE enabled = 1 AND trigger_type = 'cron' AND next_fire_at IS NOT NULL AND next_fire_at <= ?1", @@ -1315,6 +1304,7 @@ impl Database for LibSqlBackend { } async fn update_routine(&self, routine: &Routine) -> Result<(), DatabaseError> { + let conn = self.connect()?; let trigger_type = routine.trigger.type_tag(); let trigger_config = routine.trigger.to_config_json(); let action_type = routine.action.type_tag(); @@ -1325,9 +1315,9 @@ impl Database for LibSqlBackend { .guardrails .dedup_window .map(|d| d.as_secs() as i64); + let now = fmt_ts(&Utc::now()); - self.conn - .execute( + conn.execute( r#" UPDATE routines SET name = ?2, description = ?3, enabled = ?4, @@ -1337,7 +1327,7 @@ impl Database for LibSqlBackend { notify_channel = ?12, notify_user = ?13, notify_on_success = ?14, notify_on_failure = ?15, notify_on_attention = ?16, state = ?17, next_fire_at = ?18, - updated_at = datetime('now') + updated_at = ?19 WHERE id = ?1 "#, params![ @@ -1359,6 +1349,7 @@ impl Database for LibSqlBackend { routine.notify.on_attention as i64, routine.state.to_string(), fmt_opt_ts(&routine.next_fire_at), + now, ], ) .await @@ -1375,13 +1366,14 @@ impl Database for LibSqlBackend { consecutive_failures: u32, state: &serde_json::Value, ) -> Result<(), DatabaseError> { - self.conn - .execute( + let conn = self.connect()?; + let now = fmt_ts(&Utc::now()); + conn.execute( r#" UPDATE routines SET last_run_at = ?2, next_fire_at = ?3, run_count = ?4, consecutive_failures = ?5, - state = ?6, updated_at = datetime('now') + state = ?6, updated_at = ?7 WHERE id = ?1 "#, params![ @@ -1391,6 +1383,7 @@ impl Database for LibSqlBackend { run_count as i64, consecutive_failures as i64, state.to_string(), + now, ], ) .await @@ -1399,8 +1392,8 @@ impl Database for LibSqlBackend { } async fn delete_routine(&self, id: Uuid) -> Result { - let count = self - .conn + let conn = self.connect()?; + let count = conn .execute( "DELETE FROM routines WHERE id = ?1", params![id.to_string()], @@ -1413,8 +1406,8 @@ impl Database for LibSqlBackend { // ==================== Routine Runs ==================== async fn create_routine_run(&self, run: &RoutineRun) -> Result<(), DatabaseError> { - self.conn - .execute( + let conn = self.connect()?; + conn.execute( r#" INSERT INTO routine_runs ( id, routine_id, trigger_type, trigger_detail, @@ -1443,11 +1436,12 @@ impl Database for LibSqlBackend { result_summary: Option<&str>, tokens_used: Option, ) -> Result<(), DatabaseError> { - self.conn - .execute( + let conn = self.connect()?; + let now = fmt_ts(&Utc::now()); + conn.execute( r#" UPDATE routine_runs SET - completed_at = datetime('now'), status = ?2, + completed_at = ?5, status = ?2, result_summary = ?3, tokens_used = ?4 WHERE id = ?1 "#, @@ -1456,6 +1450,7 @@ impl Database for LibSqlBackend { status.to_string(), opt_text(result_summary), tokens_used.map(|t| t as i64), + now, ], ) .await @@ -1468,8 +1463,8 @@ impl Database for LibSqlBackend { routine_id: Uuid, limit: i64, ) -> Result, DatabaseError> { - let mut rows = self - .conn + let conn = self.connect()?; + let mut rows = conn .query( &format!( "SELECT {} FROM routine_runs WHERE routine_id = ?1 ORDER BY started_at DESC LIMIT ?2", @@ -1491,8 +1486,8 @@ impl Database for LibSqlBackend { &self, routine_id: Uuid, ) -> Result { - let mut rows = self - .conn + let conn = self.connect()?; + let mut rows = conn .query( "SELECT COUNT(*) as cnt FROM routine_runs WHERE routine_id = ?1 AND status = 'running'", params![routine_id.to_string()], @@ -1513,17 +1508,18 @@ impl Database for LibSqlBackend { tool_name: &str, error_message: &str, ) -> Result<(), DatabaseError> { - self.conn - .execute( + let conn = self.connect()?; + let now = fmt_ts(&Utc::now()); + conn.execute( r#" INSERT INTO tool_failures (id, tool_name, error_message, error_count, last_failure) - VALUES (?1, ?2, ?3, 1, datetime('now')) + VALUES (?1, ?2, ?3, 1, ?4) ON CONFLICT (tool_name) DO UPDATE SET error_message = ?3, error_count = tool_failures.error_count + 1, - last_failure = datetime('now') + last_failure = ?4 "#, - params![Uuid::new_v4().to_string(), tool_name, error_message], + params![Uuid::new_v4().to_string(), tool_name, error_message, now], ) .await .map_err(|e| DatabaseError::Query(e.to_string()))?; @@ -1534,8 +1530,8 @@ impl Database for LibSqlBackend { &self, threshold: i32, ) -> Result, DatabaseError> { - let mut rows = self - .conn + let conn = self.connect()?; + let mut rows = conn .query( r#" SELECT tool_name, error_message, error_count, first_failure, last_failure, @@ -1566,10 +1562,11 @@ impl Database for LibSqlBackend { } async fn mark_tool_repaired(&self, tool_name: &str) -> Result<(), DatabaseError> { - self.conn - .execute( - "UPDATE tool_failures SET repaired_at = datetime('now'), error_count = 0 WHERE tool_name = ?1", - params![tool_name], + let conn = self.connect()?; + let now = fmt_ts(&Utc::now()); + conn.execute( + "UPDATE tool_failures SET repaired_at = ?2, error_count = 0 WHERE tool_name = ?1", + params![tool_name, now], ) .await .map_err(|e| DatabaseError::Query(e.to_string()))?; @@ -1577,8 +1574,8 @@ impl Database for LibSqlBackend { } async fn increment_repair_attempts(&self, tool_name: &str) -> Result<(), DatabaseError> { - self.conn - .execute( + let conn = self.connect()?; + conn.execute( "UPDATE tool_failures SET repair_attempts = repair_attempts + 1 WHERE tool_name = ?1", params![tool_name], ) @@ -1594,8 +1591,8 @@ impl Database for LibSqlBackend { user_id: &str, key: &str, ) -> Result, DatabaseError> { - let mut rows = self - .conn + let conn = self.connect()?; + let mut rows = conn .query( "SELECT value FROM settings WHERE user_id = ?1 AND key = ?2", params![user_id, key], @@ -1614,8 +1611,8 @@ impl Database for LibSqlBackend { user_id: &str, key: &str, ) -> Result, DatabaseError> { - let mut rows = self - .conn + let conn = self.connect()?; + let mut rows = conn .query( "SELECT key, value, updated_at FROM settings WHERE user_id = ?1 AND key = ?2", params![user_id, key], @@ -1639,16 +1636,17 @@ impl Database for LibSqlBackend { key: &str, value: &serde_json::Value, ) -> Result<(), DatabaseError> { - self.conn - .execute( + let conn = self.connect()?; + let now = fmt_ts(&Utc::now()); + conn.execute( r#" INSERT INTO settings (user_id, key, value, updated_at) - VALUES (?1, ?2, ?3, datetime('now')) + VALUES (?1, ?2, ?3, ?4) ON CONFLICT (user_id, key) DO UPDATE SET value = excluded.value, - updated_at = datetime('now') + updated_at = ?4 "#, - params![user_id, key, value.to_string()], + params![user_id, key, value.to_string(), now], ) .await .map_err(|e| DatabaseError::Query(e.to_string()))?; @@ -1660,8 +1658,8 @@ impl Database for LibSqlBackend { user_id: &str, key: &str, ) -> Result { - let count = self - .conn + let conn = self.connect()?; + let count = conn .execute( "DELETE FROM settings WHERE user_id = ?1 AND key = ?2", params![user_id, key], @@ -1675,8 +1673,8 @@ impl Database for LibSqlBackend { &self, user_id: &str, ) -> Result, DatabaseError> { - let mut rows = self - .conn + let conn = self.connect()?; + let mut rows = conn .query( "SELECT key, value, updated_at FROM settings WHERE user_id = ?1 ORDER BY key", params![user_id], @@ -1699,8 +1697,8 @@ impl Database for LibSqlBackend { &self, user_id: &str, ) -> Result, DatabaseError> { - let mut rows = self - .conn + let conn = self.connect()?; + let mut rows = conn .query( "SELECT key, value FROM settings WHERE user_id = ?1", params![user_id], @@ -1720,41 +1718,40 @@ impl Database for LibSqlBackend { user_id: &str, settings: &HashMap, ) -> Result<(), DatabaseError> { - self.conn - .execute("BEGIN", ()) + let conn = self.connect()?; + let now = fmt_ts(&Utc::now()); + conn.execute("BEGIN", ()) .await .map_err(|e| DatabaseError::Query(e.to_string()))?; for (key, value) in settings { - if let Err(e) = self - .conn + if let Err(e) = conn .execute( r#" INSERT INTO settings (user_id, key, value, updated_at) - VALUES (?1, ?2, ?3, datetime('now')) + VALUES (?1, ?2, ?3, ?4) ON CONFLICT (user_id, key) DO UPDATE SET value = excluded.value, - updated_at = datetime('now') + updated_at = ?4 "#, - params![user_id, key.as_str(), value.to_string()], + params![user_id, key.as_str(), value.to_string(), now.as_str()], ) .await { - let _ = self.conn.execute("ROLLBACK", ()).await; + let _ = conn.execute("ROLLBACK", ()).await; return Err(DatabaseError::Query(e.to_string())); } } - self.conn - .execute("COMMIT", ()) + conn.execute("COMMIT", ()) .await .map_err(|e| DatabaseError::Query(e.to_string()))?; Ok(()) } async fn has_settings(&self, user_id: &str) -> Result { - let mut rows = self - .conn + let conn = self.connect()?; + let mut rows = conn .query( "SELECT COUNT(*) as cnt FROM settings WHERE user_id = ?1", params![user_id], @@ -1776,9 +1773,11 @@ impl Database for LibSqlBackend { agent_id: Option, path: &str, ) -> Result { + let conn = self.connect().map_err(|e| WorkspaceError::SearchFailed { + reason: e.to_string(), + })?; let agent_id_str = agent_id.map(|id| id.to_string()); - let mut rows = self - .conn + let mut rows = conn .query( r#" SELECT id, user_id, agent_id, path, content, @@ -1805,8 +1804,10 @@ impl Database for LibSqlBackend { } async fn get_document_by_id(&self, id: Uuid) -> Result { - let mut rows = self - .conn + let conn = self.connect().map_err(|e| WorkspaceError::SearchFailed { + reason: e.to_string(), + })?; + let mut rows = conn .query( r#" SELECT id, user_id, agent_id, path, content, @@ -1845,10 +1846,12 @@ impl Database for LibSqlBackend { } // Create + let conn = self.connect().map_err(|e| WorkspaceError::SearchFailed { + reason: e.to_string(), + })?; let id = Uuid::new_v4(); let agent_id_str = agent_id.map(|id| id.to_string()); - self.conn - .execute( + conn.execute( r#" INSERT INTO memory_documents (id, user_id, agent_id, path, content, metadata) VALUES (?1, ?2, ?3, ?4, '', '{}') @@ -1865,10 +1868,13 @@ impl Database for LibSqlBackend { } async fn update_document(&self, id: Uuid, content: &str) -> Result<(), WorkspaceError> { - self.conn - .execute( - "UPDATE memory_documents SET content = ?2, updated_at = datetime('now') WHERE id = ?1", - params![id.to_string(), content], + let conn = self.connect().map_err(|e| WorkspaceError::SearchFailed { + reason: e.to_string(), + })?; + let now = fmt_ts(&Utc::now()); + conn.execute( + "UPDATE memory_documents SET content = ?2, updated_at = ?3 WHERE id = ?1", + params![id.to_string(), content, now], ) .await .map_err(|e| WorkspaceError::SearchFailed { @@ -1886,9 +1892,11 @@ impl Database for LibSqlBackend { let doc = self.get_document_by_path(user_id, agent_id, path).await?; self.delete_chunks(doc.id).await?; + let conn = self.connect().map_err(|e| WorkspaceError::SearchFailed { + reason: e.to_string(), + })?; let agent_id_str = agent_id.map(|id| id.to_string()); - self.conn - .execute( + conn.execute( "DELETE FROM memory_documents WHERE user_id = ?1 AND agent_id IS ?2 AND path = ?3", params![user_id, agent_id_str.as_deref(), path], ) @@ -1905,6 +1913,9 @@ impl Database for LibSqlBackend { agent_id: Option, directory: &str, ) -> Result, WorkspaceError> { + let conn = self.connect().map_err(|e| WorkspaceError::SearchFailed { + reason: e.to_string(), + })?; // Implement the list_workspace_files logic in Rust instead of PL/pgSQL. let dir = if !directory.is_empty() && !directory.ends_with('/') { format!("{}/", directory) @@ -1919,8 +1930,7 @@ impl Database for LibSqlBackend { format!("{}%", dir) }; - let mut rows = self - .conn + let mut rows = conn .query( r#" SELECT path, updated_at, substr(content, 1, 200) as content_preview @@ -2004,9 +2014,11 @@ impl Database for LibSqlBackend { user_id: &str, agent_id: Option, ) -> Result, WorkspaceError> { + let conn = self.connect().map_err(|e| WorkspaceError::SearchFailed { + reason: e.to_string(), + })?; let agent_id_str = agent_id.map(|id| id.to_string()); - let mut rows = self - .conn + let mut rows = conn .query( "SELECT path FROM memory_documents WHERE user_id = ?1 AND agent_id IS ?2 ORDER BY path", params![user_id, agent_id_str.as_deref()], @@ -2030,9 +2042,11 @@ impl Database for LibSqlBackend { user_id: &str, agent_id: Option, ) -> Result, WorkspaceError> { + let conn = self.connect().map_err(|e| WorkspaceError::SearchFailed { + reason: e.to_string(), + })?; let agent_id_str = agent_id.map(|id| id.to_string()); - let mut rows = self - .conn + let mut rows = conn .query( r#" SELECT id, user_id, agent_id, path, content, @@ -2060,8 +2074,10 @@ impl Database for LibSqlBackend { // ==================== Workspace: Chunks ==================== async fn delete_chunks(&self, document_id: Uuid) -> Result<(), WorkspaceError> { - self.conn - .execute( + let conn = self.connect().map_err(|e| WorkspaceError::ChunkingFailed { + reason: e.to_string(), + })?; + conn.execute( "DELETE FROM memory_chunks WHERE document_id = ?1", params![document_id.to_string()], ) @@ -2079,6 +2095,9 @@ impl Database for LibSqlBackend { content: &str, embedding: Option<&[f32]>, ) -> Result { + let conn = self.connect().map_err(|e| WorkspaceError::ChunkingFailed { + reason: e.to_string(), + })?; let id = Uuid::new_v4(); let embedding_blob = embedding.map(|e| { // Convert f32 slice to bytes for F32_BLOB @@ -2086,8 +2105,7 @@ impl Database for LibSqlBackend { bytes }); - self.conn - .execute( + conn.execute( r#" INSERT INTO memory_chunks (id, document_id, chunk_index, content, embedding) VALUES (?1, ?2, ?3, ?4, ?5) @@ -2112,10 +2130,12 @@ impl Database for LibSqlBackend { chunk_id: Uuid, embedding: &[f32], ) -> Result<(), WorkspaceError> { + let conn = self.connect().map_err(|e| WorkspaceError::EmbeddingFailed { + reason: e.to_string(), + })?; let bytes: Vec = embedding.iter().flat_map(|f| f.to_le_bytes()).collect(); - self.conn - .execute( + conn.execute( "UPDATE memory_chunks SET embedding = ?2 WHERE id = ?1", params![chunk_id.to_string(), libsql::Value::Blob(bytes)], ) @@ -2132,9 +2152,11 @@ impl Database for LibSqlBackend { agent_id: Option, limit: usize, ) -> Result, WorkspaceError> { + let conn = self.connect().map_err(|e| WorkspaceError::SearchFailed { + reason: e.to_string(), + })?; let agent_id_str = agent_id.map(|id| id.to_string()); - let mut rows = self - .conn + let mut rows = conn .query( r#" SELECT c.id, c.document_id, c.chunk_index, c.content, c.created_at @@ -2177,18 +2199,20 @@ impl Database for LibSqlBackend { embedding: Option<&[f32]>, config: &SearchConfig, ) -> Result, WorkspaceError> { + let conn = self.connect().map_err(|e| WorkspaceError::SearchFailed { + reason: e.to_string(), + })?; let agent_id_str = agent_id.map(|id| id.to_string()); let pre_limit = config.pre_fusion_limit as i64; // FTS search using FTS5 let fts_results = if config.use_fts { - let mut rows = self - .conn + let mut rows = conn .query( r#" SELECT c.id, c.document_id, c.content FROM memory_chunks_fts fts - JOIN memory_chunks c ON c.rowid = fts.rowid + JOIN memory_chunks c ON c._rowid = fts.rowid JOIN memory_documents d ON d.id = c.document_id WHERE d.user_id = ?1 AND d.agent_id IS ?2 AND memory_chunks_fts MATCH ?3 @@ -2233,13 +2257,12 @@ impl Database for LibSqlBackend { // vector_top_k returns rowids from the vector index. // We join back to memory_chunks and filter by user/agent. - let mut rows = self - .conn + let mut rows = conn .query( r#" SELECT c.id, c.document_id, c.content FROM vector_top_k('idx_memory_chunks_embedding', vector(?1), ?2) AS top_k - JOIN memory_chunks c ON c.rowid = top_k.id + JOIN memory_chunks c ON c._rowid = top_k.id JOIN memory_documents d ON d.id = c.document_id WHERE d.user_id = ?3 AND d.agent_id IS ?4 "#, @@ -2268,6 +2291,10 @@ impl Database for LibSqlBackend { Vec::new() }; + if embedding.is_some() && !config.use_vector { + tracing::warn!("Embedding provided but vector search is disabled in config; using FTS-only results"); + } + Ok(reciprocal_rank_fusion(fts_results, vector_results, config)) } } diff --git a/src/db/libsql_migrations.rs b/src/db/libsql_migrations.rs index d064aea4..6de9ff5a 100644 --- a/src/db/libsql_migrations.rs +++ b/src/db/libsql_migrations.rs @@ -216,7 +216,8 @@ CREATE TRIGGER IF NOT EXISTS update_memory_documents_updated_at -- ==================== Workspace: Memory Chunks ==================== CREATE TABLE IF NOT EXISTS memory_chunks ( - id TEXT PRIMARY KEY, + _rowid INTEGER PRIMARY KEY AUTOINCREMENT, + id TEXT NOT NULL UNIQUE, document_id TEXT NOT NULL REFERENCES memory_documents(id) ON DELETE CASCADE, chunk_index INTEGER NOT NULL, content TEXT NOT NULL, @@ -234,24 +235,24 @@ CREATE INDEX IF NOT EXISTS idx_memory_chunks_embedding -- FTS5 virtual table for full-text search CREATE VIRTUAL TABLE IF NOT EXISTS memory_chunks_fts USING fts5( content, - content=memory_chunks, - content_rowid=rowid + content='memory_chunks', + content_rowid='_rowid' ); -- Triggers to keep FTS5 in sync with memory_chunks CREATE TRIGGER IF NOT EXISTS memory_chunks_fts_insert AFTER INSERT ON memory_chunks BEGIN - INSERT INTO memory_chunks_fts(rowid, content) VALUES (new.rowid, new.content); + INSERT INTO memory_chunks_fts(rowid, content) VALUES (new._rowid, new.content); END; CREATE TRIGGER IF NOT EXISTS memory_chunks_fts_delete AFTER DELETE ON memory_chunks BEGIN INSERT INTO memory_chunks_fts(memory_chunks_fts, rowid, content) - VALUES ('delete', old.rowid, old.content); + VALUES ('delete', old._rowid, old.content); END; CREATE TRIGGER IF NOT EXISTS memory_chunks_fts_update AFTER UPDATE ON memory_chunks BEGIN INSERT INTO memory_chunks_fts(memory_chunks_fts, rowid, content) - VALUES ('delete', old.rowid, old.content); - INSERT INTO memory_chunks_fts(rowid, content) VALUES (new.rowid, new.content); + VALUES ('delete', old._rowid, old.content); + INSERT INTO memory_chunks_fts(rowid, content) VALUES (new._rowid, new.content); END; -- ==================== Workspace: Heartbeat State ==================== diff --git a/src/db/mod.rs b/src/db/mod.rs index 2b2e4a59..ffc13c84 100644 --- a/src/db/mod.rs +++ b/src/db/mod.rs @@ -19,6 +19,7 @@ pub mod libsql_backend; pub mod libsql_migrations; use std::collections::HashMap; +use std::sync::Arc; use async_trait::async_trait; use chrono::{DateTime, Utc}; @@ -37,6 +38,60 @@ use crate::history::{ use crate::workspace::{MemoryChunk, MemoryDocument, WorkspaceEntry}; use crate::workspace::{SearchConfig, SearchResult}; +/// Create a database backend from configuration, run migrations, and return it. +/// +/// This is the shared helper for CLI commands and other call sites that need +/// a simple `Arc` without retaining backend-specific handles +/// (e.g., `pg_pool` or `libsql_conn` for the secrets store). The main agent +/// startup in `main.rs` uses its own initialization block because it also +/// captures those backend-specific handles. +pub async fn connect_from_config( + config: &crate::config::DatabaseConfig, +) -> Result, DatabaseError> { + match config.backend { + #[cfg(feature = "libsql")] + crate::config::DatabaseBackend::LibSql => { + use secrecy::ExposeSecret as _; + + let default_path = crate::config::default_libsql_path(); + let db_path = config.libsql_path.as_deref().unwrap_or(&default_path); + + let backend = if let Some(ref url) = config.libsql_url { + let token = config.libsql_auth_token.as_ref().ok_or_else(|| { + DatabaseError::Pool( + "LIBSQL_AUTH_TOKEN required when LIBSQL_URL is set".to_string(), + ) + })?; + libsql_backend::LibSqlBackend::new_remote_replica( + db_path, + url, + token.expose_secret(), + ) + .await + .map_err(|e| DatabaseError::Pool(e.to_string()))? + } else { + libsql_backend::LibSqlBackend::new_local(db_path) + .await + .map_err(|e| DatabaseError::Pool(e.to_string()))? + }; + backend.run_migrations().await?; + Ok(Arc::new(backend)) + } + #[cfg(feature = "postgres")] + _ => { + let pg = postgres::PgBackend::new(config) + .await + .map_err(|e| DatabaseError::Pool(e.to_string()))?; + pg.run_migrations().await?; + Ok(Arc::new(pg)) + } + #[cfg(not(feature = "postgres"))] + _ => Err(DatabaseError::Pool( + "No database backend available. Enable 'postgres' or 'libsql' feature.".to_string(), + )), + } +} + /// Backend-agnostic database trait. /// /// Combines all persistence operations from Store, Repository, and related diff --git a/src/main.rs b/src/main.rs index 054f8f81..c8564780 100644 --- a/src/main.rs +++ b/src/main.rs @@ -133,40 +133,10 @@ async fn main() -> anyhow::Result<()> { }; // Create a Database-trait-backed workspace for the memory command - let db: Arc = match config.database.backend { - #[cfg(feature = "libsql")] - ironclaw::config::DatabaseBackend::LibSql => { - use ironclaw::db::libsql_backend::LibSqlBackend; - use ironclaw::db::Database as _; - use secrecy::ExposeSecret as _; - - let default_path = ironclaw::config::default_libsql_path(); - let db_path = config.database.libsql_path.as_deref() - .unwrap_or(&default_path); - - let backend = if let Some(ref url) = config.database.libsql_url { - let token = config.database.libsql_auth_token.as_ref() - .expect("LIBSQL_AUTH_TOKEN required when LIBSQL_URL is set"); - LibSqlBackend::new_remote_replica(db_path, url, token.expose_secret()).await? - } else { - LibSqlBackend::new_local(db_path).await? - }; - backend.run_migrations().await?; - Arc::new(backend) - } - #[cfg(feature = "postgres")] - _ => { - use ironclaw::db::Database as _; - let pg = ironclaw::db::postgres::PgBackend::new(&config.database).await - .map_err(|e| anyhow::anyhow!("{}", e))?; - pg.run_migrations().await.map_err(|e| anyhow::anyhow!("{}", e))?; - Arc::new(pg) - } - #[cfg(not(feature = "postgres"))] - _ => { - anyhow::bail!("No database backend available. Enable 'postgres' or 'libsql' feature."); - } - }; + let db: Arc = + ironclaw::db::connect_from_config(&config.database) + .await + .map_err(|e| anyhow::anyhow!("{}", e))?; return ironclaw::cli::run_memory_command_with_db(mem_cmd.clone(), db, embeddings).await; } @@ -370,6 +340,11 @@ async fn main() -> anyhow::Result<()> { // // Creates an `Arc` that all consumers share. // Backend is selected by the `DATABASE_BACKEND` env var / config. + // + // NOTE: For simpler call sites (CLI commands, Memory handler) use the shared + // helper `ironclaw::db::connect_from_config()`. This block is kept inline + // because it also captures backend-specific handles (`pg_pool`, `libsql_conn`) + // needed by the secrets store. #[cfg(feature = "postgres")] let mut pg_pool: Option = None; #[cfg(feature = "libsql")] diff --git a/src/secrets/store.rs b/src/secrets/store.rs index f12639e8..34675bb1 100644 --- a/src/secrets/store.rs +++ b/src/secrets/store.rs @@ -339,8 +339,11 @@ impl SecretsStore for LibSqlSecretsStore { .expires_at .map(|dt| dt.to_rfc3339_opts(chrono::SecondsFormat::Millis, true)); - self.conn - .execute( + // Start transaction for atomic upsert + read-back + let tx = self.conn.transaction().await + .map_err(|e| SecretError::Database(e.to_string()))?; + + tx.execute( r#" INSERT INTO secrets (id, user_id, name, encrypted_value, key_salt, provider, expires_at, created_at, updated_at) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?8) @@ -366,8 +369,7 @@ impl SecretsStore for LibSqlSecretsStore { .map_err(|e| SecretError::Database(e.to_string()))?; // Read back the row (may have been upserted) - let mut rows = self - .conn + let mut rows = tx .query( r#" SELECT id, user_id, name, encrypted_value, key_salt, provider, expires_at, @@ -386,7 +388,12 @@ impl SecretsStore for LibSqlSecretsStore { .map_err(|e| SecretError::Database(e.to_string()))? .ok_or_else(|| SecretError::Database("Insert succeeded but row not found".into()))?; - libsql_row_to_secret(&row) + let secret = libsql_row_to_secret(&row)?; + + tx.commit().await + .map_err(|e| SecretError::Database(e.to_string()))?; + + Ok(secret) } async fn get(&self, user_id: &str, name: &str) -> Result {