//! Database abstraction layer. //! //! Provides a backend-agnostic `Database` trait that unifies all persistence //! operations. Two implementations exist behind feature flags: //! //! - `postgres` (default): Uses `deadpool-postgres` + `tokio-postgres` //! - `libsql`: Uses libSQL (Turso's SQLite fork) for embedded/edge deployment //! //! The existing `Store`, `Repository`, `SecretsStore`, and `WasmToolStore` //! types become thin wrappers that delegate to `Arc`. #[cfg(feature = "postgres")] pub mod postgres; #[cfg(feature = "postgres")] pub mod tls; #[cfg(feature = "libsql")] pub mod libsql; #[cfg(feature = "libsql")] pub mod libsql_migrations; use std::collections::HashMap; use std::sync::Arc; use async_trait::async_trait; use chrono::{DateTime, Utc}; use rust_decimal::Decimal; use uuid::Uuid; use crate::agent::BrokenTool; use crate::agent::routine::{Routine, RoutineRun, RunStatus}; use crate::context::{ActionRecord, JobContext, JobState}; use crate::error::DatabaseError; use crate::error::WorkspaceError; use crate::history::{ AgentJobRecord, AgentJobSummary, ConversationMessage, ConversationSummary, JobEventRecord, LlmCallRecord, SandboxJobRecord, SandboxJobSummary, SettingRow, }; 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> { let (db, _handles) = connect_with_handles(config).await?; Ok(db) } /// Backend-specific handles retained after database connection. /// /// These are needed by satellite stores (e.g., `SecretsStore`) that require /// a backend-specific handle rather than the generic `Arc`. #[derive(Default)] pub struct DatabaseHandles { #[cfg(feature = "postgres")] pub pg_pool: Option, #[cfg(feature = "libsql")] pub libsql_db: Option>, } /// Connect to the database, run migrations, and return both the generic /// `Database` trait object and the backend-specific handles. pub async fn connect_with_handles( config: &crate::config::DatabaseConfig, ) -> Result<(Arc, DatabaseHandles), DatabaseError> { let mut handles = DatabaseHandles::default(); 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::LibSqlBackend::new_remote_replica(db_path, url, token.expose_secret()) .await .map_err(|e| DatabaseError::Pool(e.to_string()))? } else { libsql::LibSqlBackend::new_local(db_path) .await .map_err(|e| DatabaseError::Pool(e.to_string()))? }; backend.run_migrations().await?; tracing::debug!("libSQL database connected and migrations applied"); handles.libsql_db = Some(backend.shared_db()); Ok((Arc::new(backend) as Arc, handles)) } #[cfg(feature = "postgres")] crate::config::DatabaseBackend::Postgres => { let pg = postgres::PgBackend::new(config) .await .map_err(|e| DatabaseError::Pool(e.to_string()))?; pg.run_migrations().await?; tracing::info!("PostgreSQL database connected and migrations applied"); handles.pg_pool = Some(pg.pool()); Ok((Arc::new(pg) as Arc, handles)) } #[allow(unreachable_patterns)] _ => Err(DatabaseError::Pool(format!( "Database backend '{}' is not available. Rebuild with the appropriate feature flag.", config.backend ))), } } /// Create a secrets store from database and secrets configuration. /// /// This is the shared factory for CLI commands and other call sites that need /// a `SecretsStore` without going through the full `AppBuilder`. Mirrors the /// pattern of [`connect_from_config`] but returns a secrets-specific store. pub async fn create_secrets_store( config: &crate::config::DatabaseConfig, crypto: Arc, ) -> 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::LibSqlBackend::new_remote_replica(db_path, url, token.expose_secret()) .await .map_err(|e| DatabaseError::Pool(e.to_string()))? } else { libsql::LibSqlBackend::new_local(db_path) .await .map_err(|e| DatabaseError::Pool(e.to_string()))? }; backend.run_migrations().await?; Ok(Arc::new(crate::secrets::LibSqlSecretsStore::new( backend.shared_db(), crypto, ))) } #[cfg(feature = "postgres")] crate::config::DatabaseBackend::Postgres => { let pg = postgres::PgBackend::new(config) .await .map_err(|e| DatabaseError::Pool(e.to_string()))?; pg.run_migrations().await?; Ok(Arc::new(crate::secrets::PostgresSecretsStore::new( pg.pool(), crypto, ))) } #[allow(unreachable_patterns)] _ => Err(DatabaseError::Pool(format!( "Database backend '{}' is not available for secrets. Rebuild with the appropriate feature flag.", config.backend ))), } } // ==================== Wizard / testing helpers ==================== /// Connect to the database WITHOUT running migrations, validating /// prerequisites when applicable (PostgreSQL version, pgvector). /// /// Returns both the `Database` trait object and backend-specific handles. /// Used by the wizard to test connectivity before committing — call /// [`Database::run_migrations`] on the returned trait object when ready. pub async fn connect_without_migrations( config: &crate::config::DatabaseConfig, ) -> Result<(Arc, DatabaseHandles), DatabaseError> { let mut handles = DatabaseHandles::default(); 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::LibSqlBackend::new_remote_replica(db_path, url, token.expose_secret()) .await .map_err(|e| DatabaseError::Pool(e.to_string()))? } else { libsql::LibSqlBackend::new_local(db_path) .await .map_err(|e| DatabaseError::Pool(e.to_string()))? }; handles.libsql_db = Some(backend.shared_db()); Ok((Arc::new(backend) as Arc, handles)) } #[cfg(feature = "postgres")] crate::config::DatabaseBackend::Postgres => { let pg = postgres::PgBackend::new(config) .await .map_err(|e| DatabaseError::Pool(e.to_string()))?; handles.pg_pool = Some(pg.pool()); // Validate PostgreSQL prerequisites (version, pgvector) validate_postgres(&pg.pool()).await?; Ok((Arc::new(pg) as Arc, handles)) } #[allow(unreachable_patterns)] _ => Err(DatabaseError::Pool(format!( "Database backend '{}' is not available. Rebuild with the appropriate feature flag.", config.backend ))), } } /// Validate PostgreSQL prerequisites (version >= 15, pgvector available). /// /// Returns `Ok(())` if all prerequisites are met, or a `DatabaseError` /// with a user-facing message describing the issue. #[cfg(feature = "postgres")] async fn validate_postgres(pool: &deadpool_postgres::Pool) -> Result<(), DatabaseError> { let client = pool .get() .await .map_err(|e| DatabaseError::Pool(format!("Failed to connect: {}", e)))?; // Check PostgreSQL server version (need 15+ for pgvector). let version_row = client .query_one("SHOW server_version", &[]) .await .map_err(|e| DatabaseError::Query(format!("Failed to query server version: {}", e)))?; let version_str: &str = version_row.get(0); let major_version = version_str .split('.') .next() .and_then(|v| v.parse::().ok()) .ok_or_else(|| { DatabaseError::Pool(format!( "Could not parse PostgreSQL version from '{}'. \ Expected a numeric major version (e.g., '15.2').", version_str )) })?; const MIN_PG_MAJOR_VERSION: u32 = 15; if major_version < MIN_PG_MAJOR_VERSION { return Err(DatabaseError::Pool(format!( "PostgreSQL {} detected. IronClaw requires PostgreSQL {} or later \ for pgvector support.\n\ Upgrade: https://www.postgresql.org/download/", version_str, MIN_PG_MAJOR_VERSION ))); } // Check if pgvector extension is available. let pgvector_row = client .query_opt( "SELECT 1 FROM pg_available_extensions WHERE name = 'vector'", &[], ) .await .map_err(|e| { DatabaseError::Query(format!("Failed to check pgvector availability: {}", e)) })?; if pgvector_row.is_none() { return Err(DatabaseError::Pool(format!( "pgvector extension not found on your PostgreSQL server.\n\n\ Install it:\n \ macOS: brew install pgvector\n \ Ubuntu: apt install postgresql-{0}-pgvector\n \ Docker: use the pgvector/pgvector:pg{0} image\n \ Source: https://github.com/pgvector/pgvector#installation\n\n\ Then restart PostgreSQL and re-run: ironclaw onboard", major_version ))); } Ok(()) } // ==================== User management record types ==================== /// A registered user. #[derive(Debug, Clone)] pub struct UserRecord { /// User identifier (string, matches existing `user_id` throughout the codebase). pub id: String, pub email: Option, pub display_name: String, /// `active`, `suspended`, or `deactivated`. pub status: String, /// `admin` or `member`. pub role: String, pub created_at: DateTime, pub updated_at: DateTime, pub last_login_at: Option>, /// Who created/invited this user (nullable for bootstrap users). pub created_by: Option, pub metadata: serde_json::Value, } /// An API token for authenticating requests (hash stored, never plaintext). #[derive(Debug, Clone)] pub struct ApiTokenRecord { pub id: Uuid, pub user_id: String, /// Human label (e.g. "my-laptop", "ci-bot"). pub name: String, /// First 8 hex chars of the plaintext token for display/identification. pub token_prefix: String, pub expires_at: Option>, pub last_used_at: Option>, pub created_at: DateTime, /// Soft-revoke timestamp. Non-null means revoked. pub revoked_at: Option>, } // ==================== Sub-traits ==================== // // Each sub-trait groups related persistence methods. The `Database` supertrait // combines them all, so existing `Arc` consumers keep working. // Leaf consumers can depend on a specific sub-trait instead. #[async_trait] pub trait ConversationStore: Send + Sync { async fn create_conversation( &self, channel: &str, user_id: &str, thread_id: Option<&str>, ) -> Result; async fn touch_conversation(&self, id: Uuid) -> Result<(), DatabaseError>; async fn add_conversation_message( &self, conversation_id: Uuid, role: &str, content: &str, ) -> Result; async fn ensure_conversation( &self, id: Uuid, channel: &str, user_id: &str, thread_id: Option<&str>, ) -> Result; async fn list_conversations_with_preview( &self, user_id: &str, channel: &str, limit: i64, ) -> Result, DatabaseError>; async fn list_conversations_all_channels( &self, user_id: &str, limit: i64, ) -> Result, DatabaseError>; async fn get_or_create_routine_conversation( &self, routine_id: Uuid, routine_name: &str, user_id: &str, ) -> Result; async fn get_or_create_heartbeat_conversation( &self, user_id: &str, ) -> Result; async fn get_or_create_assistant_conversation( &self, user_id: &str, channel: &str, ) -> Result; async fn create_conversation_with_metadata( &self, channel: &str, user_id: &str, metadata: &serde_json::Value, ) -> Result; async fn list_conversation_messages_paginated( &self, conversation_id: Uuid, before: Option>, limit: i64, ) -> Result<(Vec, bool), DatabaseError>; async fn update_conversation_metadata_field( &self, id: Uuid, key: &str, value: &serde_json::Value, ) -> Result<(), DatabaseError>; async fn get_conversation_metadata( &self, id: Uuid, ) -> Result, DatabaseError>; async fn list_conversation_messages( &self, conversation_id: Uuid, ) -> Result, DatabaseError>; async fn conversation_belongs_to_user( &self, conversation_id: Uuid, user_id: &str, ) -> Result; } #[async_trait] pub trait JobStore: Send + Sync { async fn save_job(&self, ctx: &JobContext) -> Result<(), DatabaseError>; async fn get_job(&self, id: Uuid) -> Result, DatabaseError>; async fn update_job_status( &self, id: Uuid, status: JobState, failure_reason: Option<&str>, ) -> Result<(), DatabaseError>; async fn mark_job_stuck(&self, id: Uuid) -> Result<(), DatabaseError>; async fn get_stuck_jobs(&self) -> Result, DatabaseError>; async fn list_agent_jobs(&self) -> Result, DatabaseError>; async fn list_agent_jobs_for_user( &self, user_id: &str, ) -> Result, DatabaseError>; async fn agent_job_summary(&self) -> Result; async fn agent_job_summary_for_user( &self, user_id: &str, ) -> Result; /// Get the failure reason for a single agent job (O(1) lookup). async fn get_agent_job_failure_reason(&self, id: Uuid) -> Result, DatabaseError>; async fn save_action(&self, job_id: Uuid, action: &ActionRecord) -> Result<(), DatabaseError>; async fn get_job_actions(&self, job_id: Uuid) -> Result, DatabaseError>; async fn record_llm_call(&self, record: &LlmCallRecord<'_>) -> Result; async fn save_estimation_snapshot( &self, job_id: Uuid, category: &str, tool_names: &[String], estimated_cost: Decimal, estimated_time_secs: i32, estimated_value: Decimal, ) -> Result; async fn update_estimation_actuals( &self, id: Uuid, actual_cost: Decimal, actual_time_secs: i32, actual_value: Option, ) -> Result<(), DatabaseError>; } #[async_trait] pub trait SandboxStore: Send + Sync { async fn save_sandbox_job(&self, job: &SandboxJobRecord) -> Result<(), DatabaseError>; async fn get_sandbox_job(&self, id: Uuid) -> Result, DatabaseError>; async fn list_sandbox_jobs(&self) -> Result, DatabaseError>; async fn update_sandbox_job_status( &self, id: Uuid, status: &str, success: Option, message: Option<&str>, started_at: Option>, completed_at: Option>, ) -> Result<(), DatabaseError>; async fn cleanup_stale_sandbox_jobs(&self) -> Result; async fn sandbox_job_summary(&self) -> Result; async fn list_sandbox_jobs_for_user( &self, user_id: &str, ) -> Result, DatabaseError>; async fn sandbox_job_summary_for_user( &self, user_id: &str, ) -> Result; async fn sandbox_job_belongs_to_user( &self, job_id: Uuid, user_id: &str, ) -> Result; async fn update_sandbox_job_mode(&self, id: Uuid, mode: &str) -> Result<(), DatabaseError>; async fn get_sandbox_job_mode(&self, id: Uuid) -> Result, DatabaseError>; async fn save_job_event( &self, job_id: Uuid, event_type: &str, data: &serde_json::Value, ) -> Result<(), DatabaseError>; async fn list_job_events( &self, job_id: Uuid, limit: Option, ) -> Result, DatabaseError>; } #[async_trait] pub trait RoutineStore: Send + Sync { async fn create_routine(&self, routine: &Routine) -> Result<(), DatabaseError>; async fn get_routine(&self, id: Uuid) -> Result, DatabaseError>; async fn get_routine_by_name( &self, user_id: &str, name: &str, ) -> Result, DatabaseError>; async fn list_routines(&self, user_id: &str) -> Result, DatabaseError>; async fn list_all_routines(&self) -> Result, DatabaseError>; async fn list_event_routines(&self) -> Result, DatabaseError>; async fn list_due_cron_routines(&self) -> Result, DatabaseError>; async fn update_routine(&self, routine: &Routine) -> Result<(), DatabaseError>; async fn update_routine_runtime( &self, id: Uuid, last_run_at: DateTime, next_fire_at: Option>, run_count: u64, consecutive_failures: u32, state: &serde_json::Value, ) -> Result<(), DatabaseError>; async fn delete_routine(&self, id: Uuid) -> Result; async fn create_routine_run(&self, run: &RoutineRun) -> Result<(), DatabaseError>; async fn complete_routine_run( &self, id: Uuid, status: RunStatus, result_summary: Option<&str>, tokens_used: Option, ) -> Result<(), DatabaseError>; async fn list_routine_runs( &self, routine_id: Uuid, limit: i64, ) -> Result, DatabaseError>; async fn count_running_routine_runs(&self, routine_id: Uuid) -> Result; async fn count_running_routine_runs_batch( &self, routine_ids: &[Uuid], ) -> Result, DatabaseError>; /// Fetch the last run status for multiple routines in a single query. /// Returns a map from routine_id to its most recent RunStatus. /// Routines with no runs are omitted from the result. async fn batch_get_last_run_status( &self, routine_ids: &[Uuid], ) -> Result, DatabaseError>; async fn link_routine_run_to_job( &self, run_id: Uuid, job_id: Uuid, ) -> Result<(), DatabaseError>; async fn get_webhook_routine_by_path( &self, path: &str, user_id: Option<&str>, ) -> Result, DatabaseError>; /// List routine runs that were dispatched as full_job but have not yet /// been finalized (status='running' with a linked job_id). async fn list_dispatched_routine_runs(&self) -> Result, DatabaseError>; } #[async_trait] pub trait ToolFailureStore: Send + Sync { async fn record_tool_failure( &self, tool_name: &str, error_message: &str, ) -> Result<(), DatabaseError>; async fn get_broken_tools(&self, threshold: i32) -> Result, DatabaseError>; async fn mark_tool_repaired(&self, tool_name: &str) -> Result<(), DatabaseError>; async fn increment_repair_attempts(&self, tool_name: &str) -> Result<(), DatabaseError>; } #[async_trait] pub trait SettingsStore: Send + Sync { async fn get_setting( &self, user_id: &str, key: &str, ) -> Result, DatabaseError>; async fn get_setting_full( &self, user_id: &str, key: &str, ) -> Result, DatabaseError>; async fn set_setting( &self, user_id: &str, key: &str, value: &serde_json::Value, ) -> Result<(), DatabaseError>; async fn delete_setting(&self, user_id: &str, key: &str) -> Result; async fn list_settings(&self, user_id: &str) -> Result, DatabaseError>; async fn get_all_settings( &self, user_id: &str, ) -> Result, DatabaseError>; async fn set_all_settings( &self, user_id: &str, settings: &HashMap, ) -> Result<(), DatabaseError>; async fn has_settings(&self, user_id: &str) -> Result; } #[async_trait] pub trait WorkspaceStore: Send + Sync { async fn get_document_by_path( &self, user_id: &str, agent_id: Option, path: &str, ) -> Result; async fn get_document_by_id(&self, id: Uuid) -> Result; async fn get_or_create_document_by_path( &self, user_id: &str, agent_id: Option, path: &str, ) -> Result; async fn update_document(&self, id: Uuid, content: &str) -> Result<(), WorkspaceError>; async fn delete_document_by_path( &self, user_id: &str, agent_id: Option, path: &str, ) -> Result<(), WorkspaceError>; async fn list_directory( &self, user_id: &str, agent_id: Option, directory: &str, ) -> Result, WorkspaceError>; async fn list_all_paths( &self, user_id: &str, agent_id: Option, ) -> Result, WorkspaceError>; async fn list_documents( &self, user_id: &str, agent_id: Option, ) -> Result, WorkspaceError>; async fn delete_chunks(&self, document_id: Uuid) -> Result<(), WorkspaceError>; async fn insert_chunk( &self, document_id: Uuid, chunk_index: i32, content: &str, embedding: Option<&[f32]>, ) -> Result; async fn update_chunk_embedding( &self, chunk_id: Uuid, embedding: &[f32], ) -> Result<(), WorkspaceError>; async fn get_chunks_without_embeddings( &self, user_id: &str, agent_id: Option, limit: usize, ) -> Result, WorkspaceError>; async fn hybrid_search( &self, user_id: &str, agent_id: Option, query: &str, embedding: Option<&[f32]>, config: &SearchConfig, ) -> Result, WorkspaceError>; // ==================== Multi-scope read methods ==================== // // Default implementations loop over user_ids calling single-scope methods, // then merge results. Backends can override with efficient SQL (e.g., // `WHERE user_id = ANY($1::text[])`). /// Hybrid search across multiple user scopes, merging results by score. /// /// **Note:** The default implementation calls `hybrid_search` per scope and /// merges by raw score. Because RRF scores are normalized independently /// within each scope, scores are not directly comparable across scopes. /// The Postgres backend overrides this with a single combined query that /// applies RRF once to the unified result set. async fn hybrid_search_multi( &self, user_ids: &[String], agent_id: Option, query: &str, embedding: Option<&[f32]>, config: &SearchConfig, ) -> Result, WorkspaceError> { if user_ids.len() > 1 { tracing::debug!( scope_count = user_ids.len(), "hybrid_search_multi: using default per-scope RRF merge; \ cross-scope score comparison may be unreliable" ); } let mut all_results = Vec::new(); for uid in user_ids { let results = self .hybrid_search(uid, agent_id, query, embedding, config) .await?; all_results.extend(results); } // Re-sort by score descending and truncate to limit all_results.sort_by(|a, b| { b.score .partial_cmp(&a.score) .unwrap_or(std::cmp::Ordering::Equal) }); all_results.truncate(config.limit); Ok(all_results) } /// List all file paths across multiple user scopes. async fn list_all_paths_multi( &self, user_ids: &[String], agent_id: Option, ) -> Result, WorkspaceError> { let mut all_paths = Vec::new(); for uid in user_ids { let paths = self.list_all_paths(uid, agent_id).await?; all_paths.extend(paths); } all_paths.sort(); all_paths.dedup(); Ok(all_paths) } /// Get a document by path, searching across multiple user scopes. /// /// Returns the first match found (tries each user_id in order). async fn get_document_by_path_multi( &self, user_ids: &[String], agent_id: Option, path: &str, ) -> Result { for uid in user_ids { match self.get_document_by_path(uid, agent_id, path).await { Ok(doc) => return Ok(doc), Err(WorkspaceError::DocumentNotFound { .. }) => continue, Err(e) => return Err(e), } } Err(WorkspaceError::DocumentNotFound { doc_type: path.to_string(), user_id: format!("[{}]", user_ids.join(", ")), }) } /// List directory contents across multiple user scopes. async fn list_directory_multi( &self, user_ids: &[String], agent_id: Option, directory: &str, ) -> Result, WorkspaceError> { let mut all_entries = Vec::new(); for uid in user_ids { all_entries.extend(self.list_directory(uid, agent_id, directory).await?); } Ok(crate::workspace::merge_workspace_entries(all_entries)) } } #[async_trait] pub trait UserStore: Send + Sync { // ---- Users ---- /// Create a new user record. async fn create_user(&self, user: &UserRecord) -> Result<(), DatabaseError>; /// Get a user by their string id. async fn get_user(&self, id: &str) -> Result, DatabaseError>; /// Get a user by email address. async fn get_user_by_email(&self, email: &str) -> Result, DatabaseError>; /// List users, optionally filtered by status. async fn list_users(&self, status: Option<&str>) -> Result, DatabaseError>; /// Update a user's status (active/suspended/deactivated). async fn update_user_status(&self, id: &str, status: &str) -> Result<(), DatabaseError>; /// Update a user's role (admin/member). async fn update_user_role(&self, id: &str, role: &str) -> Result<(), DatabaseError>; /// Update a user's display name and metadata. async fn update_user_profile( &self, id: &str, display_name: &str, metadata: &serde_json::Value, ) -> Result<(), DatabaseError>; /// Record a login timestamp. async fn record_login(&self, id: &str) -> Result<(), DatabaseError>; // ---- API Tokens ---- /// Create a new API token. The `token_hash` is SHA-256 of the plaintext. async fn create_api_token( &self, user_id: &str, name: &str, token_hash: &[u8; 32], token_prefix: &str, expires_at: Option>, ) -> Result; /// List tokens for a user (never includes the hash). async fn list_api_tokens(&self, user_id: &str) -> Result, DatabaseError>; /// Soft-revoke a token. Returns false if the token doesn't exist or doesn't belong to the user. async fn revoke_api_token(&self, token_id: Uuid, user_id: &str) -> Result; /// Look up a token by hash, returning the token record and its owning user. /// Only returns active (non-revoked, non-expired) tokens for active users. async fn authenticate_token( &self, token_hash: &[u8; 32], ) -> Result, DatabaseError>; /// Update `last_used_at` for a token. async fn record_token_usage(&self, token_id: Uuid) -> Result<(), DatabaseError>; /// Check whether any user records exist (for first-run bootstrap detection). async fn has_any_users(&self) -> Result; /// Delete a user and all their data across all user-scoped tables. /// Returns false if the user doesn't exist. async fn delete_user(&self, id: &str) -> Result; /// Get per-user LLM usage stats for a time period. /// Aggregates from llm_calls via agent_jobs.user_id. async fn user_usage_stats( &self, user_id: Option<&str>, since: DateTime, ) -> Result, DatabaseError>; /// Lightweight per-user summary stats (job count, total cost, last active). /// Used by the admin users list to show inline stats. async fn user_summary_stats( &self, user_id: Option<&str>, ) -> Result, DatabaseError>; /// Create a user and their initial API token atomically. /// If either operation fails, both are rolled back. async fn create_user_with_token( &self, user: &UserRecord, token_name: &str, token_hash: &[u8; 32], token_prefix: &str, expires_at: Option>, ) -> Result; } /// Per-user LLM usage statistics. #[derive(Debug, Clone)] pub struct UserUsageStats { pub user_id: String, pub model: String, pub call_count: i64, pub input_tokens: i64, pub output_tokens: i64, pub total_cost: Decimal, } /// Lightweight per-user summary for the admin users list. #[derive(Debug, Clone)] pub struct UserSummaryStats { pub user_id: String, /// Total agent jobs created by this user. pub job_count: i64, /// Total LLM spend across all jobs (all-time). pub total_cost: Decimal, /// Most recent activity (latest job or LLM call timestamp). pub last_active_at: Option>, } /// Backend-agnostic database supertrait. /// /// Combines all sub-traits into one. Existing `Arc` consumers /// continue to work; leaf consumers can depend on a specific sub-trait instead. #[async_trait] pub trait Database: ConversationStore + JobStore + SandboxStore + RoutineStore + ToolFailureStore + SettingsStore + WorkspaceStore + UserStore + Send + Sync { /// Run schema migrations for this backend. async fn run_migrations(&self) -> Result<(), DatabaseError>; } #[cfg(test)] mod tests { use super::*; /// Regression test: `create_secrets_store` selects the correct backend at /// runtime based on `DatabaseConfig`, not at compile time. Previously the /// CLI duplicated this logic with compile-time `#[cfg]` gates that always /// chose postgres when both features were enabled (PR #209). #[cfg(feature = "libsql")] #[tokio::test] async fn test_create_secrets_store_libsql_backend() { use secrecy::SecretString; let tmp = tempfile::tempdir().unwrap(); let db_path = tmp.path().join("test.db"); let config = crate::config::DatabaseConfig { backend: crate::config::DatabaseBackend::LibSql, libsql_path: Some(db_path), libsql_url: None, libsql_auth_token: None, url: SecretString::from("unused://libsql".to_string()), pool_size: 1, ssl_mode: crate::config::SslMode::default(), }; let master_key = SecretString::from("a]".repeat(16)); let crypto = Arc::new(crate::secrets::SecretsCrypto::new(master_key).unwrap()); let store = create_secrets_store(&config, crypto).await; assert!( store.is_ok(), "create_secrets_store should succeed for libsql backend" ); // Verify basic operation works let store = store.unwrap(); let exists = store.exists("test_user", "nonexistent_secret").await; assert!(exists.is_ok()); assert!(!exists.unwrap()); } }