use std::collections::{HashMap, HashSet, VecDeque};use std::convert::TryInto;use std::ffi::OsString;use std::path::{Path, PathBuf};use std::sync::mpsc::SyncSender;use std::sync::{Arc, Once};use std::{fs, thread}; use ai::project_context::model::ProjectRulePath;use anyhow::{Context, Result, anyhow, bail};use chrono::Utc;use cloud_object_models::folder::persistence as folder_persistence;use cloud_object_models::folder::persistence::upsert_folders;use cloud_object_models::json_model::persistence::{ self as generic_string_persistence, PersistedGenericStringObject,};use cloud_object_models::notebook::persistence as notebook_persistence;use cloud_object_models::notebook::persistence::upsert_notebooks;use cloud_object_models::workflow::persistence as workflow_persistence;use cloud_object_models::workflow::persistence::upsert_workflows;use cloud_object_persistence::{ GenericStringObjectPersistenceData, delete_cloud_object, delete_generic_string_object, increment_retry_count, load_cloud_object_read_context, mark_object_as_synced, read_time_of_next_force_object_refresh, record_time_of_next_refresh, update_object_after_server_creation, update_object_metadata, upsert_generic_string_objects as upsert_generic_string_object_rows,};use diesel::connection::{DefaultLoadingMode, SimpleConnection};use diesel::result::Error;use diesel::sqlite::SqliteConnection;use diesel::{ BelongingToDsl, BoolExpressionMethods, Connection, ExpressionMethods, GroupedBy, OptionalExtension, QueryDsl, RunQueryDsl, SelectableHelper,};use diesel_migrations::MigrationHarness;use itertools::Itertools;use libsqlite3_sys as sqlite3;use lsp::supported_servers::LSPServerType;use num_traits::FromPrimitive;use pathfinder_geometry::rect::RectF;use pathfinder_geometry::vector::Vector2F;use persistence::model::AMBIENT_AGENT_PANE_KIND;use uuid::Uuid;use warp_core::features::FeatureFlag;use warp_errors::{report_error, report_if_error};use warpui::platform::FullscreenState;use warpui::windowing::{MIN_WINDOW_HEIGHT, MIN_WINDOW_WIDTH};use warpui::{AppContext, SingletonEntity}; use super::agent::{ backfill_conversation_summaries, delete_agent_conversations, read_agent_conversation_metadata, upsert_agent_conversation,};use super::block_list::{ delete_ai_conversation, delete_blocks, save_block, update_block_agent_view_visibility, upsert_ai_query,};use super::model::{ self, AI_DOCUMENT_PANE_KIND, AI_FACT_PANE_KIND, ActiveMCPServer, CODE_PANE_KIND, CurrentUserInformation, ENV_VAR_COLLECTION_PANE_KIND, EXECUTION_PROFILE_EDITOR_PANE_KIND, MCP_SERVER_PANE_KIND, MCPEnvironmentVariables, NOTEBOOK_PANE_KIND, NewActiveMCPServer, NewApp, NewCommand, NewServerExperiment, NewTab, NewTabGroup, NewTeam, NewWindow, NewWorkspace, NewWorkspaceMetadata, NewWorkspaceTeam, Project, SETTINGS_PANE_KIND, TERMINAL_PANE_KIND, Tab, TabGroup, WORKFLOW_PANE_KIND, Window, WorkspaceMetadata as WorkspaceMetadataModel,};use super::{ BlockCompleted, FinishedCommandMetadata, ModelEvent, PersistedData, PersistedDataScope, PersistenceScope, StartedCommandMetadata, WriterHandles, schema,};use crate::ai::agent::conversation::AIConversationId;use crate::ai::ambient_agents::AmbientAgentTaskId;use crate::ai::mcp::templatable_installation::VariableValue;use crate::ai::mcp::{TemplatableMCPServer, TemplatableMCPServerInstallation};use crate::ai::persisted_workspace::EnablementState;use crate::app_state::{ AIFactPaneSnapshot, AmbientAgentPaneSnapshot, AppState, BranchSnapshot, CodePaneSnapShot, CodePaneTabSnapshot, CodeReviewPaneSnapshot, EnvVarCollectionPaneSnapshot, LeafContents, LeafSnapshot, LeftPanelSnapshot, NotebookPaneSnapshot, PaneFlex, PaneNodeSnapshot, RightPanelSnapshot, SettingsPaneSnapshot, SplitDirection, TabGroupSnapshot, TabSnapshot, TerminalPaneSnapshot, WindowSnapshot, WorkflowPaneSnapshot,};use crate::auth::UserUid;use crate::auth::auth_manager::PersistedCurrentUserInformation;use crate::auth::auth_state::AuthStateProvider;use crate::cloud_object::model::actions::{ ObjectAction, ObjectActionSubtype, object_action_from_persisted,};use crate::cloud_object::model::generic_string_model::{CloudStringObject, GenericStringObjectId};use crate::cloud_object::{CloudObject, ObjectIdType};use crate::code::editor_management::CodeSource;use crate::drive::OpenWarpDriveObjectSettings;use crate::notebooks::NotebookId;use crate::persistence::block_list::{ get_all_restored_blocks, process_ai_queries_for_nld_history_match, process_ai_queries_for_uparrow_prompt, read_recent_ai_queries,};use crate::persistence::model::{ CODE_REVIEW_PANE_KIND, GET_STARTED_PANE_KIND, NewPersistedObjectAction, NewTeamSettings, ProjectRules, UserProfile,};use crate::server::experiments::ServerExperiment;use crate::server::ids::{ClientId, HashableId, ServerId, SyncId};use crate::server::telemetry::TelemetryEvent;use crate::settings_view::SettingsSection;use crate::suggestions::ignored_suggestions_model::SuggestionType;use crate::tab::SelectedTabColor;use crate::terminal::ShellLaunchData;use crate::terminal::history::PersistedCommand;use crate::themes::theme::AnsiColorIdentifier;use crate::workflows::WorkflowId;use crate::workspace::tab_group::TabGroupId;use crate::workspaces::team::Team as TeamMetadata;use crate::workspaces::user_profiles::{UserProfileWithUID, user_profile_from_persistence};use crate::workspaces::workspace::{Workspace as WorkspaceMetadata, WorkspaceUid};use crate::{safe_info, send_telemetry_from_app_ctx}; diesel::define_sql_function! { fn json_extract(target: diesel::sql_types::Text, path: diesel::sql_types::Text) -> diesel::sql_types::Text;} // Choose a power of 2 that seems to be a reasonable upper bound for how many// events to queue.const CHANNEL_SIZE: usize = 1024;const COMMANDS_COUNT_LIMIT: i64 = 10000; const WARP_SQLITE_FILE_NAME: &str = "warp.sqlite"; /// Runs any migrations and creates the Sqlite database if it doesn't exist./// Reads from the sqlite database to get the app state for session restoration./// Starts a writer thread that listens for ModelEvents and processes them.pub fn initialize( ctx: &mut AppContext, scope: PersistenceScope, data_scope: PersistedDataScope,) -> (Option<Box<PersistedData>>, Option<WriterHandles>) { unsafe { // Set up logging before any SQLite calls. init_logging(); } let database_path = database_file_path_for_scope(&scope); match init_db(&scope) { Ok(mut conn) => { let mut persisted_data = read_persisted_data(&mut conn, ctx, data_scope); let writer_handles = match start_writer(conn, database_path.clone()) { Ok(writer_handles) => Some(writer_handles), Err(err) => { send_telemetry_from_app_ctx!( TelemetryEvent::DatabaseWriteError(err.to_string()), ctx ); report_db_error("starting writer", err, &database_path); None } }; // Persist any read-time-derived conversation summaries so the // derivation only happens once per pre-`summary`-column row. if let (Some(persisted_data), Some(writer_handles)) = (persisted_data.as_mut(), writer_handles.as_ref()) { let backfills = std::mem::take(&mut persisted_data.conversation_summary_backfills); if !backfills.is_empty() { log::info!("Backfilling {} conversation summaries", backfills.len()); report_if_error!( writer_handles .sender .send(ModelEvent::BackfillConversationSummaries { backfills }) .context("Error requesting conversation summary backfill") ); } } (persisted_data, writer_handles) } Err(err) => { send_telemetry_from_app_ctx!( TelemetryEvent::DatabaseStartUpError(err.to_string()), ctx ); report_db_error("initialization", err, &database_path); (None, None) } }} fn read_persisted_data( conn: &mut SqliteConnection, ctx: &mut AppContext, data_scope: PersistedDataScope,) -> Option<Box<PersistedData>> { let user_uid = AuthStateProvider::as_ref(ctx).get().user_id(); match read_sqlite_data(conn, user_uid, data_scope) { Ok(app_state) => Some(Box::new(app_state)), Err(err) => { send_telemetry_from_app_ctx!(TelemetryEvent::DatabaseReadError(err.to_string()), ctx); report_error!(anyhow::Error::new(err).context("Failed to read persisted data")); None } }} /// Returns a read-only connection to the sqlite database./// We want only one write connection to exist and use event processing to write any data needed.pub fn establish_ro_connection(database_url: &str) -> Result<SqliteConnection> { establish_connection(database_url, true)} fn establish_connection(database_url: &str, read_only: bool) -> Result<SqliteConnection> { let full_database_url = if read_only { &format!("file:{database_url}?mode=ro") } else { database_url }; let mut conn = SqliteConnection::establish(full_database_url)?; conn.batch_execute( r#" PRAGMA foreign_keys = ON; -- enforce foreign key constraints PRAGMA busy_timeout = 1000; -- sleep for up to 1s if the database is busy "#, )?; // Enable WAL mode, checkpointing whenever the log is at least 500 pages long (in theory, // around 2MB). In addition, SQLite will automatically checkpoint when the app closes its // database connection. // The auto-checkpoint interval is lowered from the default of 1000 because all writes // already run in a background thread and can afford to checkpoint slightly more often. // At the default value, the WAL can grow larger than a typical database (for our usage). conn.batch_execute( r#" PRAGMA journal_mode=WAL; PRAGMA wal_autocheckpoint=500; "#, ) .context("Failed to enable WAL")?; Ok(conn)} /// Set up SQLite [error logging](https://www.sqlite.org/errlog.html)////// ## Safety/// Setting up SQLite logging is not thread-safe. No other SQLite calls may be made while this/// function is running.unsafe fn init_logging() { unsafe { use std::ffi::{CStr, c_char, c_int, c_void}; use std::{panic, ptr}; extern "C-unwind" fn log_callback(_data: *mut c_void, err_code: c_int, msg: *const c_char) { // `err_code` is an extended error code (https://www.sqlite.org/rescode.html#primary_result_codes_versus_extended_result_codes). // In general, the least-significant byte of an extended error code is the primary error // code it belongs to. Each primary error code can also be used where an extended error // code is expected (for example, `SQLITE_SCHEMA` has no extended error codes). let primary_error_code = err_code & 0xFF; let level = match (primary_error_code, err_code) { // This usually means that a schema change invalidated a prepared statement. (sqlite3::SQLITE_SCHEMA, _) => log::Level::Debug, // These are used with sqlite3_log, in extensions. (sqlite3::SQLITE_NOTICE | sqlite3::SQLITE_WARNING, _) => log::Level::Warn, // According to the docs, this error means that the database file was moved (or deleted), // so SQLite can't safely modify it and the rollback journal: // https://www.sqlite.org/rescode.html#readonly_dbmoved // This is mostly outside of Warp's control (e.g. the user or some system program is // moving around files in the user data directory), so downgrade to a warning. (_, sqlite3::SQLITE_READONLY_DBMOVED) => log::Level::Warn, _ => log::Level::Error, }; // Safety: the message pointer came from the SQLite library, which promises that it's a // valid C string pointer. let msg = unsafe { CStr::from_ptr(msg) }; let err_message = String::from_utf8_lossy(msg.to_bytes()); // Sentry shouldn't panic, but to be safe, make sure we don't unwind across the FFI // boundary. let _ = panic::catch_unwind(|| { // We report SQLite errors to Sentry in a more-structured format so that they have // better grouping (all are under the same Sentry issue, with details for the specific // error kind). Warning and debug SQLite messages are logged - with the default // sentry_log configuration, warnings are added as breadcrumbs to other events and // debug messages are ignored. // In local builds without crash reporting, all SQLite messages get logged locally. #[cfg(feature = "crash_reporting")] if level == log::Level::Error { use std::sync::atomic::{AtomicU64, Ordering}; // Each bit represents a primary SQLite error code (0-63). Primary codes are // the least-significant byte of the extended code; real error codes are ≤ 28. static REPORTED_PRIMARY_CODES: AtomicU64 = AtomicU64::new(0); let primary_code = (primary_error_code as u64).min(63); let bit = 1u64 << primary_code; if REPORTED_PRIMARY_CODES.fetch_or(bit, Ordering::Relaxed) & bit == 0 { // First occurrence of this primary error code — report to Sentry. sentry::with_scope( |scope| { let mut context = std::collections::BTreeMap::new(); context.insert("message".to_string(), err_message.into()); context.insert("code".to_string(), err_code.into()); context.insert( "code_description".to_string(), sqlite3::code_to_str(err_code).into(), ); scope.set_context( "sqlite", sentry::protocol::Context::Other(context), ); }, || { sentry::capture_message( "Sqlite Error", sentry_log::convert_log_level(level), ) }, ); // The structured Sentry event is the record; skip the redundant log line. return; } } log::log!( level, "SQLite error {} ({}): {}", err_code, sqlite3::code_to_str(err_code), err_message ); }); } static INIT: Once = Once::new(); INIT.call_once(|| { let null: *const c_void = ptr::null(); // Diesel doesn't expose SQLite's logging/tracing APIs, but the FFI bindings do. let status = sqlite3::sqlite3_config( sqlite3::SQLITE_CONFIG_LOG, log_callback as extern "C-unwind" fn(_, _, _), null, ); if status != sqlite3::SQLITE_OK { report_error!( "Error setting up SQLite logging", extra: { "status" => %sqlite3::code_to_str(status) } ); } }); }} /// Determines the db path, establishes a connection and runs any migrations.pub(super) fn init_db(scope: &PersistenceScope) -> Result<SqliteConnection> { // First, make sure the parent directory of the file exists, otherwise // we'll get an error if the file doesn't already exist. let db_path = database_file_path_for_scope(scope); // If we fail to create the necessary directories, log a warning and // continue; we'll return a sqlite error if it actually fails to initialize // a database connection. let db_parent = db_path .parent() .expect("database file path should be absolute"); if let Err(err) = std::fs::create_dir_all(db_parent) { log::warn!( "Encountered an error while creating parent directories for sqlite database: {err:#}" ); } if matches!(scope, PersistenceScope::RemoteServerDaemon { .. }) { ensure_owner_only_dir(db_parent)?; } if matches!(scope, PersistenceScope::App) { migrate_old_sqlite_into_secure_container_if_needed(&db_path); } let conn = setup_database(&db_path)?; if matches!(scope, PersistenceScope::RemoteServerDaemon { .. }) { ensure_owner_only_file(&db_path)?; } Ok(conn)} fn migrate_old_sqlite_into_secure_container_if_needed(db_path: &Path) { let old_db_path = warp_core::paths::state_dir().join(WARP_SQLITE_FILE_NAME); if old_db_path == db_path || !old_db_path.exists() || db_path.exists() { return; } match std::fs::rename(&old_db_path, db_path) { Ok(_) => { safe_info!( safe: ("Migrated SQLite database into application container"), full: ("Migrated SQLite database from `{}` to `{}`", old_db_path.display(), db_path.display()) ); // Also migrate the associated WAL and SHM files. let old_wal = old_db_path.with_extension("sqlite-wal"); let old_shm = old_db_path.with_extension("sqlite-shm"); let new_wal = db_path.with_extension("sqlite-wal"); let new_shm = db_path.with_extension("sqlite-shm"); if let Err(err) = std::fs::rename(&old_wal, &new_wal) { if err.kind() != std::io::ErrorKind::NotFound { report_error!( anyhow::Error::new(err) .context("Failed to migrate SQLite WAL into application container") ); } } else { log::info!("Migrated SQLite WAL into application container"); } if let Err(err) = std::fs::rename(&old_shm, &new_shm) { if err.kind() != std::io::ErrorKind::NotFound { report_error!( anyhow::Error::new(err) .context("Failed to migrate SQLite SHM into application container") ); } } else { log::info!("Migrated SQLite shared memory file into application container"); } } Err(err) => { report_error!( anyhow::Error::new(err) .context("Failed to migrate SQLite database into application container") ); } }} /// Creates or connects to the database at `database_path` and runs any migrations.fn setup_database(database_path: &Path) -> Result<SqliteConnection> { let db_url = database_path .to_str() .ok_or_else(|| anyhow!("Failed to convert db path to a string"))?; let mut conn = establish_connection(db_url, false)?; safe_info!( safe: ("Connecting to SQLite database"), full: ("Connecting to SQLite database at {db_url}") ); conn.run_pending_migrations(persistence::MIGRATIONS) .map_err(|e| anyhow!(e)) .context("Failed to perform migrations")?; Ok(conn)} /// The path at which the sqlite database is located for the given scope.////// Integration tests that initialize the database with known data should use/// this function to determine where to create the database file.pub fn database_file_path_for_scope(scope: &PersistenceScope) -> PathBuf { match scope { PersistenceScope::App => app_database_file_path(), PersistenceScope::Tui => tui_database_file_path(), PersistenceScope::RemoteServerDaemon { identity_key } => { remote_server_daemon_database_file_path(identity_key) } }} /// The database file path for the scope this process's persistence was/// initialized with (see [`super::current_scope`]).////// Ad-hoc read-only connections should use this instead of hardcoding/// [`PersistenceScope::App`], so that a TUI process never reads the GUI's/// database.pub fn database_file_path_for_current_scope() -> PathBuf { database_file_path_for_scope(&super::current_scope())} fn app_database_file_path() -> PathBuf { warp_core::paths::secure_state_dir() .unwrap_or_else(warp_core::paths::state_dir) .join(WARP_SQLITE_FILE_NAME)} fn tui_database_file_path() -> PathBuf { warp_core::paths::tui_state_dir().join(WARP_SQLITE_FILE_NAME)} fn remote_server_daemon_database_file_path(identity_key: &str) -> PathBuf { let data_dir = remote_server::setup::remote_server_daemon_data_dir(identity_key); let expanded_data_dir = shellexpand::tilde(&data_dir).into_owned(); PathBuf::from(expanded_data_dir).join(WARP_SQLITE_FILE_NAME)} #[cfg(unix)]fn ensure_owner_only_dir(path: &Path) -> Result<()> { use std::fs::Permissions; use std::os::unix::fs::PermissionsExt; std::fs::set_permissions(path, Permissions::from_mode(0o700)) .with_context(|| format!("setting permissions on directory {}", path.display()))} #[cfg(not(unix))]fn ensure_owner_only_dir(_path: &Path) -> Result<()> { Ok(())Lines 1–500 of 3246. Load more
use std::collections::{HashMap, HashSet, VecDeque}; use std::convert::TryInto; use std::ffi::OsString; use std::path::{Path, PathBuf}; use std::sync::mpsc::SyncSender; use std::sync::{Arc, Once}; use std::{fs, thread}; use ai::project_context::model::ProjectRulePath; use anyhow::{Context, Result, anyhow, bail}; use chrono::Utc; use cloud_object_models::folder::persistence as folder_persistence; use cloud_object_models::folder::persistence::upsert_folders; use cloud_object_models::json_model::persistence::{ self as generic_string_persistence, PersistedGenericStringObject, }; use cloud_object_models::notebook::persistence as notebook_persistence; use cloud_object_models::notebook::persistence::upsert_notebooks; use cloud_object_models::workflow::persistence as workflow_persistence; use cloud_object_models::workflow::persistence::upsert_workflows; use cloud_object_persistence::{ GenericStringObjectPersistenceData, delete_cloud_object, delete_generic_string_object, increment_retry_count, load_cloud_object_read_context, mark_object_as_synced, read_time_of_next_force_object_refresh, record_time_of_next_refresh, update_object_after_server_creation, update_object_metadata, upsert_generic_string_objects as upsert_generic_string_object_rows, }; use diesel::connection::{DefaultLoadingMode, SimpleConnection}; use diesel::result::Error; use diesel::sqlite::SqliteConnection; use diesel::{ BelongingToDsl, BoolExpressionMethods, Connection, ExpressionMethods, GroupedBy, OptionalExtension, QueryDsl, RunQueryDsl, SelectableHelper, }; use diesel_migrations::MigrationHarness; use itertools::Itertools; use libsqlite3_sys as sqlite3; use lsp::supported_servers::LSPServerType; use num_traits::FromPrimitive; use pathfinder_geometry::rect::RectF; use pathfinder_geometry::vector::Vector2F; use persistence::model::AMBIENT_AGENT_PANE_KIND; use uuid::Uuid; use warp_core::features::FeatureFlag; use warp_errors::{report_error, report_if_error}; use warpui::platform::FullscreenState; use warpui::windowing::{MIN_WINDOW_HEIGHT, MIN_WINDOW_WIDTH}; use warpui::{AppContext, SingletonEntity}; use super::agent::{ backfill_conversation_summaries, delete_agent_conversations, read_agent_conversation_metadata, upsert_agent_conversation, }; use super::block_list::{ delete_ai_conversation, delete_blocks, save_block, update_block_agent_view_visibility, upsert_ai_query, }; use super::model::{ self, AI_DOCUMENT_PANE_KIND, AI_FACT_PANE_KIND, ActiveMCPServer, CODE_PANE_KIND, CurrentUserInformation, ENV_VAR_COLLECTION_PANE_KIND, EXECUTION_PROFILE_EDITOR_PANE_KIND, MCP_SERVER_PANE_KIND, MCPEnvironmentVariables, NOTEBOOK_PANE_KIND, NewActiveMCPServer, NewApp, NewCommand, NewServerExperiment, NewTab, NewTabGroup, NewTeam, NewWindow, NewWorkspace, NewWorkspaceMetadata, NewWorkspaceTeam, Project, SETTINGS_PANE_KIND, TERMINAL_PANE_KIND, Tab, TabGroup, WORKFLOW_PANE_KIND, Window, WorkspaceMetadata as WorkspaceMetadataModel, }; use super::{ BlockCompleted, FinishedCommandMetadata, ModelEvent, PersistedData, PersistedDataScope, PersistenceScope, StartedCommandMetadata, WriterHandles, schema, }; use crate::ai::agent::conversation::AIConversationId; use crate::ai::ambient_agents::AmbientAgentTaskId; use crate::ai::mcp::templatable_installation::VariableValue; use crate::ai::mcp::{TemplatableMCPServer, TemplatableMCPServerInstallation}; use crate::ai::persisted_workspace::EnablementState; use crate::app_state::{ AIFactPaneSnapshot, AmbientAgentPaneSnapshot, AppState, BranchSnapshot, CodePaneSnapShot, CodePaneTabSnapshot, CodeReviewPaneSnapshot, EnvVarCollectionPaneSnapshot, LeafContents, LeafSnapshot, LeftPanelSnapshot, NotebookPaneSnapshot, PaneFlex, PaneNodeSnapshot, RightPanelSnapshot, SettingsPaneSnapshot, SplitDirection, TabGroupSnapshot, TabSnapshot, TerminalPaneSnapshot, WindowSnapshot, WorkflowPaneSnapshot, }; use crate::auth::UserUid; use crate::auth::auth_manager::PersistedCurrentUserInformation; use crate::auth::auth_state::AuthStateProvider; use crate::cloud_object::model::actions::{ ObjectAction, ObjectActionSubtype, object_action_from_persisted, }; use crate::cloud_object::model::generic_string_model::{CloudStringObject, GenericStringObjectId}; use crate::cloud_object::{CloudObject, ObjectIdType}; use crate::code::editor_management::CodeSource; use crate::drive::OpenWarpDriveObjectSettings; use crate::notebooks::NotebookId; use crate::persistence::block_list::{ get_all_restored_blocks, process_ai_queries_for_nld_history_match, process_ai_queries_for_uparrow_prompt, read_recent_ai_queries, }; use crate::persistence::model::{ CODE_REVIEW_PANE_KIND, GET_STARTED_PANE_KIND, NewPersistedObjectAction, NewTeamSettings, ProjectRules, UserProfile, }; use crate::server::experiments::ServerExperiment; use crate::server::ids::{ClientId, HashableId, ServerId, SyncId}; use crate::server::telemetry::TelemetryEvent; use crate::settings_view::SettingsSection; use crate::suggestions::ignored_suggestions_model::SuggestionType; use crate::tab::SelectedTabColor; use crate::terminal::ShellLaunchData; use crate::terminal::history::PersistedCommand; use crate::themes::theme::AnsiColorIdentifier; use crate::workflows::WorkflowId; use crate::workspace::tab_group::TabGroupId; use crate::workspaces::team::Team as TeamMetadata; use crate::workspaces::user_profiles::{UserProfileWithUID, user_profile_from_persistence}; use crate::workspaces::workspace::{Workspace as WorkspaceMetadata, WorkspaceUid}; use crate::{safe_info, send_telemetry_from_app_ctx}; diesel::define_sql_function! { fn json_extract(target: diesel::sql_types::Text, path: diesel::sql_types::Text) -> diesel::sql_types::Text; } // Choose a power of 2 that seems to be a reasonable upper bound for how many // events to queue. const CHANNEL_SIZE: usize = 1024; const COMMANDS_COUNT_LIMIT: i64 = 10000; const WARP_SQLITE_FILE_NAME: &str = "warp.sqlite"; /// Runs any migrations and creates the Sqlite database if it doesn't exist. /// Reads from the sqlite database to get the app state for session restoration. /// Starts a writer thread that listens for ModelEvents and processes them. pub fn initialize( ctx: &mut AppContext, scope: PersistenceScope, data_scope: PersistedDataScope, ) -> (Option<Box<PersistedData>>, Option<WriterHandles>) { unsafe { // Set up logging before any SQLite calls. init_logging(); } let database_path = database_file_path_for_scope(&scope); match init_db(&scope) { Ok(mut conn) => { let mut persisted_data = read_persisted_data(&mut conn, ctx, data_scope); let writer_handles = match start_writer(conn, database_path.clone()) { Ok(writer_handles) => Some(writer_handles), Err(err) => { send_telemetry_from_app_ctx!( TelemetryEvent::DatabaseWriteError(err.to_string()), ctx ); report_db_error("starting writer", err, &database_path); None } }; // Persist any read-time-derived conversation summaries so the // derivation only happens once per pre-`summary`-column row. if let (Some(persisted_data), Some(writer_handles)) = (persisted_data.as_mut(), writer_handles.as_ref()) { let backfills = std::mem::take(&mut persisted_data.conversation_summary_backfills); if !backfills.is_empty() { log::info!("Backfilling {} conversation summaries", backfills.len()); report_if_error!( writer_handles .sender .send(ModelEvent::BackfillConversationSummaries { backfills }) .context("Error requesting conversation summary backfill") ); } } (persisted_data, writer_handles) } Err(err) => { send_telemetry_from_app_ctx!( TelemetryEvent::DatabaseStartUpError(err.to_string()), ctx ); report_db_error("initialization", err, &database_path); (None, None) } } } fn read_persisted_data( conn: &mut SqliteConnection, ctx: &mut AppContext, data_scope: PersistedDataScope, ) -> Option<Box<PersistedData>> { let user_uid = AuthStateProvider::as_ref(ctx).get().user_id(); match read_sqlite_data(conn, user_uid, data_scope) { Ok(app_state) => Some(Box::new(app_state)), Err(err) => { send_telemetry_from_app_ctx!(TelemetryEvent::DatabaseReadError(err.to_string()), ctx); report_error!(anyhow::Error::new(err).context("Failed to read persisted data")); None } } } /// Returns a read-only connection to the sqlite database. /// We want only one write connection to exist and use event processing to write any data needed. pub fn establish_ro_connection(database_url: &str) -> Result<SqliteConnection> { establish_connection(database_url, true) } fn establish_connection(database_url: &str, read_only: bool) -> Result<SqliteConnection> { let full_database_url = if read_only { &format!("file:{database_url}?mode=ro") } else { database_url }; let mut conn = SqliteConnection::establish(full_database_url)?; conn.batch_execute( r#" PRAGMA foreign_keys = ON; -- enforce foreign key constraints PRAGMA busy_timeout = 1000; -- sleep for up to 1s if the database is busy "#, )?; // Enable WAL mode, checkpointing whenever the log is at least 500 pages long (in theory, // around 2MB). In addition, SQLite will automatically checkpoint when the app closes its // database connection. // The auto-checkpoint interval is lowered from the default of 1000 because all writes // already run in a background thread and can afford to checkpoint slightly more often. // At the default value, the WAL can grow larger than a typical database (for our usage). conn.batch_execute( r#" PRAGMA journal_mode=WAL; PRAGMA wal_autocheckpoint=500; "#, ) .context("Failed to enable WAL")?; Ok(conn) } /// Set up SQLite [error logging](https://www.sqlite.org/errlog.html) /// /// ## Safety /// Setting up SQLite logging is not thread-safe. No other SQLite calls may be made while this /// function is running. unsafe fn init_logging() { unsafe { use std::ffi::{CStr, c_char, c_int, c_void}; use std::{panic, ptr}; extern "C-unwind" fn log_callback(_data: *mut c_void, err_code: c_int, msg: *const c_char) { // `err_code` is an extended error code (https://www.sqlite.org/rescode.html#primary_result_codes_versus_extended_result_codes). // In general, the least-significant byte of an extended error code is the primary error // code it belongs to. Each primary error code can also be used where an extended error // code is expected (for example, `SQLITE_SCHEMA` has no extended error codes). let primary_error_code = err_code & 0xFF; let level = match (primary_error_code, err_code) { // This usually means that a schema change invalidated a prepared statement. (sqlite3::SQLITE_SCHEMA, _) => log::Level::Debug, // These are used with sqlite3_log, in extensions. (sqlite3::SQLITE_NOTICE | sqlite3::SQLITE_WARNING, _) => log::Level::Warn, // According to the docs, this error means that the database file was moved (or deleted), // so SQLite can't safely modify it and the rollback journal: // https://www.sqlite.org/rescode.html#readonly_dbmoved // This is mostly outside of Warp's control (e.g. the user or some system program is // moving around files in the user data directory), so downgrade to a warning. (_, sqlite3::SQLITE_READONLY_DBMOVED) => log::Level::Warn, _ => log::Level::Error, }; // Safety: the message pointer came from the SQLite library, which promises that it's a // valid C string pointer. let msg = unsafe { CStr::from_ptr(msg) }; let err_message = String::from_utf8_lossy(msg.to_bytes()); // Sentry shouldn't panic, but to be safe, make sure we don't unwind across the FFI // boundary. let _ = panic::catch_unwind(|| { // We report SQLite errors to Sentry in a more-structured format so that they have // better grouping (all are under the same Sentry issue, with details for the specific // error kind). Warning and debug SQLite messages are logged - with the default // sentry_log configuration, warnings are added as breadcrumbs to other events and // debug messages are ignored. // In local builds without crash reporting, all SQLite messages get logged locally. #[cfg(feature = "crash_reporting")] if level == log::Level::Error { use std::sync::atomic::{AtomicU64, Ordering}; // Each bit represents a primary SQLite error code (0-63). Primary codes are // the least-significant byte of the extended code; real error codes are ≤ 28. static REPORTED_PRIMARY_CODES: AtomicU64 = AtomicU64::new(0); let primary_code = (primary_error_code as u64).min(63); let bit = 1u64 << primary_code; if REPORTED_PRIMARY_CODES.fetch_or(bit, Ordering::Relaxed) & bit == 0 { // First occurrence of this primary error code — report to Sentry. sentry::with_scope( |scope| { let mut context = std::collections::BTreeMap::new(); context.insert("message".to_string(), err_message.into()); context.insert("code".to_string(), err_code.into()); context.insert( "code_description".to_string(), sqlite3::code_to_str(err_code).into(), ); scope.set_context( "sqlite", sentry::protocol::Context::Other(context), ); }, || { sentry::capture_message( "Sqlite Error", sentry_log::convert_log_level(level), ) }, ); // The structured Sentry event is the record; skip the redundant log line. return; } } log::log!( level, "SQLite error {} ({}): {}", err_code, sqlite3::code_to_str(err_code), err_message ); }); } static INIT: Once = Once::new(); INIT.call_once(|| { let null: *const c_void = ptr::null(); // Diesel doesn't expose SQLite's logging/tracing APIs, but the FFI bindings do. let status = sqlite3::sqlite3_config( sqlite3::SQLITE_CONFIG_LOG, log_callback as extern "C-unwind" fn(_, _, _), null, ); if status != sqlite3::SQLITE_OK { report_error!( "Error setting up SQLite logging", extra: { "status" => %sqlite3::code_to_str(status) } ); } }); } } /// Determines the db path, establishes a connection and runs any migrations. pub(super) fn init_db(scope: &PersistenceScope) -> Result<SqliteConnection> { // First, make sure the parent directory of the file exists, otherwise // we'll get an error if the file doesn't already exist. let db_path = database_file_path_for_scope(scope); // If we fail to create the necessary directories, log a warning and // continue; we'll return a sqlite error if it actually fails to initialize // a database connection. let db_parent = db_path .parent() .expect("database file path should be absolute"); if let Err(err) = std::fs::create_dir_all(db_parent) { log::warn!( "Encountered an error while creating parent directories for sqlite database: {err:#}" ); } if matches!(scope, PersistenceScope::RemoteServerDaemon { .. }) { ensure_owner_only_dir(db_parent)?; } if matches!(scope, PersistenceScope::App) { migrate_old_sqlite_into_secure_container_if_needed(&db_path); } let conn = setup_database(&db_path)?; if matches!(scope, PersistenceScope::RemoteServerDaemon { .. }) { ensure_owner_only_file(&db_path)?; } Ok(conn) } fn migrate_old_sqlite_into_secure_container_if_needed(db_path: &Path) { let old_db_path = warp_core::paths::state_dir().join(WARP_SQLITE_FILE_NAME); if old_db_path == db_path || !old_db_path.exists() || db_path.exists() { return; } match std::fs::rename(&old_db_path, db_path) { Ok(_) => { safe_info!( safe: ("Migrated SQLite database into application container"), full: ("Migrated SQLite database from `{}` to `{}`", old_db_path.display(), db_path.display()) ); // Also migrate the associated WAL and SHM files. let old_wal = old_db_path.with_extension("sqlite-wal"); let old_shm = old_db_path.with_extension("sqlite-shm"); let new_wal = db_path.with_extension("sqlite-wal"); let new_shm = db_path.with_extension("sqlite-shm"); if let Err(err) = std::fs::rename(&old_wal, &new_wal) { if err.kind() != std::io::ErrorKind::NotFound { report_error!( anyhow::Error::new(err) .context("Failed to migrate SQLite WAL into application container") ); } } else { log::info!("Migrated SQLite WAL into application container"); } if let Err(err) = std::fs::rename(&old_shm, &new_shm) { if err.kind() != std::io::ErrorKind::NotFound { report_error!( anyhow::Error::new(err) .context("Failed to migrate SQLite SHM into application container") ); } } else { log::info!("Migrated SQLite shared memory file into application container"); } } Err(err) => { report_error!( anyhow::Error::new(err) .context("Failed to migrate SQLite database into application container") ); } } } /// Creates or connects to the database at `database_path` and runs any migrations. fn setup_database(database_path: &Path) -> Result<SqliteConnection> { let db_url = database_path .to_str() .ok_or_else(|| anyhow!("Failed to convert db path to a string"))?; let mut conn = establish_connection(db_url, false)?; safe_info!( safe: ("Connecting to SQLite database"), full: ("Connecting to SQLite database at {db_url}") ); conn.run_pending_migrations(persistence::MIGRATIONS) .map_err(|e| anyhow!(e)) .context("Failed to perform migrations")?; Ok(conn) } /// The path at which the sqlite database is located for the given scope. /// /// Integration tests that initialize the database with known data should use /// this function to determine where to create the database file. pub fn database_file_path_for_scope(scope: &PersistenceScope) -> PathBuf { match scope { PersistenceScope::App => app_database_file_path(), PersistenceScope::Tui => tui_database_file_path(), PersistenceScope::RemoteServerDaemon { identity_key } => { remote_server_daemon_database_file_path(identity_key) } } } /// The database file path for the scope this process's persistence was /// initialized with (see [`super::current_scope`]). /// /// Ad-hoc read-only connections should use this instead of hardcoding /// [`PersistenceScope::App`], so that a TUI process never reads the GUI's /// database. pub fn database_file_path_for_current_scope() -> PathBuf { database_file_path_for_scope(&super::current_scope()) } fn app_database_file_path() -> PathBuf { warp_core::paths::secure_state_dir() .unwrap_or_else(warp_core::paths::state_dir) .join(WARP_SQLITE_FILE_NAME) } fn tui_database_file_path() -> PathBuf { warp_core::paths::tui_state_dir().join(WARP_SQLITE_FILE_NAME) } fn remote_server_daemon_database_file_path(identity_key: &str) -> PathBuf { let data_dir = remote_server::setup::remote_server_daemon_data_dir(identity_key); let expanded_data_dir = shellexpand::tilde(&data_dir).into_owned(); PathBuf::from(expanded_data_dir).join(WARP_SQLITE_FILE_NAME) } #[cfg(unix)] fn ensure_owner_only_dir(path: &Path) -> Result<()> { use std::fs::Permissions; use std::os::unix::fs::PermissionsExt; std::fs::set_permissions(path, Permissions::from_mode(0o700)) .with_context(|| format!("setting permissions on directory {}", path.display())) } #[cfg(not(unix))] fn ensure_owner_only_dir(_path: &Path) -> Result<()> { Ok(())