diff --git a/rust/src/api/messages/incoming/client2client/additional_data.rs b/rust/src/api/messages/incoming/client2client/additional_data.rs index 171b0d51..2ef17c0a 100644 --- a/rust/src/api/messages/incoming/client2client/additional_data.rs +++ b/rust/src/api/messages/incoming/client2client/additional_data.rs @@ -62,7 +62,7 @@ async fn verify_shared_contacts( return Ok(()); } - let signal_database = ctx.get_rust_db().await; + let signal_database = ctx.rust_db.read().await.clone(); for contact in data.contacts { let stored_identity = sqlx::query_scalar!( @@ -91,7 +91,7 @@ async fn verify_shared_contacts( .await?; let verified_at = chrono::Utc::now().timestamp_millis(); - ctx.get_user_discovery() + ctx.user_discovery .get() .await .update_verification_state_for_user(contact.user_id, Some(verified_at), tr) diff --git a/rust/src/api/messages/incoming/client2client/groups.rs b/rust/src/api/messages/incoming/client2client/groups.rs index 0a1f4035..76d033ff 100644 --- a/rust/src/api/messages/incoming/client2client/groups.rs +++ b/rust/src/api/messages/incoming/client2client/groups.rs @@ -260,7 +260,7 @@ pub(crate) async fn handle_flame_sync( }; let update_counters = flame.force_update - || (is_today(group_last_flame_counter_change) && is_today(last_flame_counter_change)); + || (is_today(group_last_flame_counter_change) & is_today(last_flame_counter_change)); let flame_counter = if update_counters { group.flame_counter.max(flame.flame_counter) diff --git a/rust/src/api/messages/incoming/client2client/messages.rs b/rust/src/api/messages/incoming/client2client/messages.rs index b48eaa96..afe41f34 100644 --- a/rust/src/api/messages/incoming/client2client/messages.rs +++ b/rust/src/api/messages/incoming/client2client/messages.rs @@ -154,7 +154,7 @@ async fn queue_retry_control( } pub(crate) async fn ensure_contact_exists(ctx: &Arc, from_user_id: i64) -> Result<()> { - let db_app = ctx.get_app_database().await; + let db_app = ctx.app_db.read().await.clone(); let exists = sqlx::query_scalar!( "SELECT EXISTS(SELECT 1 FROM contacts WHERE user_id = ?)", @@ -206,7 +206,7 @@ pub(crate) async fn ensure_contact_exists(ctx: &Arc, from_user_id: i64) .await?; if let Some(identity_key) = user.public_identity_key { - let rust_database = ctx.get_rust_db().await; + let rust_database = ctx.rust_db.read().await.clone(); sqlx::query!( r#" INSERT INTO signal_identities(name, identity_key, timestamp) @@ -271,7 +271,7 @@ async fn encrypt_v2_with_session_recovery( plaintext: Vec, ) -> Result> { let encrypt = |plaintext| async move { - let engine = ctx.get_signal_engine().lock().await; + let engine = ctx.signal_engine.lock().await; engine .as_ref() .ok_or(TwonlyError::SignalIdentityNotFound)? @@ -281,7 +281,7 @@ async fn encrypt_v2_with_session_recovery( match encrypt(plaintext.clone()).await { Err(TwonlyError::Signal(message)) - if message.contains("session with") && message.contains("not found") => + if message.contains("session with") & message.contains("not found") => { tracing::warn!( contact_id, @@ -314,7 +314,7 @@ pub(crate) async fn send_queued_receipt(ctx: &Arc, receipt_id: &str) -> locks.retain(|_, time| time.elapsed() < std::time::Duration::from_secs(120)); } - let app_db = ctx.get_app_database().await; + let app_db = ctx.app_db.read().await.clone(); let row = sqlx::query!( r#" SELECT r.contact_id, r.message, r.message_id, r.contact_will_sends_receipt, @@ -418,7 +418,7 @@ pub(crate) async fn prepare_queued_receipt( ctx: &Arc, receipt_id: &str, ) -> Result, Option>)>> { - let database = ctx.get_app_database().await; + let database = ctx.app_db.read().await.clone(); let row = sqlx::query!( r#"SELECT contact_id, message, message_id, retry_count FROM receipts WHERE receipt_id = ?"#, @@ -462,7 +462,7 @@ pub(crate) async fn prepare_queued_receipt( } pub(crate) async fn retransmit_queued_receipts(ctx: &Arc) -> Result<()> { - let database = ctx.get_app_database().await; + let database = ctx.app_db.read().await.clone(); let receipt_ids = sqlx::query_scalar!( r#" SELECT receipt_id FROM receipts diff --git a/rust/src/api/messages/incoming/client2client/recovery.rs b/rust/src/api/messages/incoming/client2client/recovery.rs index 343b78c2..1d38ab26 100644 --- a/rust/src/api/messages/incoming/client2client/recovery.rs +++ b/rust/src/api/messages/incoming/client2client/recovery.rs @@ -90,7 +90,7 @@ pub(crate) async fn perform_heartbeat(ctx: &Arc) -> Result<()> { .last_contact_heartbeat .is_none_or(|last| now.signed_duration_since(last).num_hours() >= 24); if contacts_due { - let database = ctx.get_app_database().await; + let database = ctx.app_db.read().await.clone(); let contacts = sqlx::query!( r#"SELECT user_id, recovery_secret_share FROM contacts WHERE recovery_is_trusted_friend = 1 @@ -121,7 +121,7 @@ pub(crate) async fn perform_heartbeat(ctx: &Arc) -> Result<()> { } } - let database = ctx.get_app_database().await; + let database = ctx.app_db.read().await.clone(); let contacts = sqlx::query!( r#"SELECT user_id, recovery_contacts_secret_share FROM contacts WHERE recovery_contacts_secret_share IS NOT NULL @@ -153,7 +153,7 @@ pub(crate) async fn perform_heartbeat(ctx: &Arc) -> Result<()> { .call() .await?; - let database = ctx.get_app_database().await; + let database = ctx.app_db.read().await.clone(); sqlx::query!( "UPDATE contacts SET recovery_contacts_last_heartbeat = ? WHERE user_id = ?", now.timestamp(), @@ -253,7 +253,7 @@ mod tests { } async fn insert_contact(ctx: &Arc, user_id: i64) -> Result<()> { - let database = ctx.get_app_database().await; + let database = ctx.app_db.read().await.clone(); sqlx::query!( "INSERT INTO contacts(user_id, username, accepted) VALUES (?, ?, 1)", user_id, @@ -268,7 +268,7 @@ mod tests { async fn recovery_share_is_stored_and_deleted() -> anyhow::Result<()> { let (_temp, ctx) = context().await?; insert_contact(&ctx, 7).await?; - let database = ctx.get_app_database().await; + let database = ctx.app_db.read().await.clone(); let mut transaction = database.pool.begin().await?; handle_passwordless_recovery( @@ -317,7 +317,7 @@ mod tests { async fn valid_and_invalid_heartbeat_update_the_expected_state() -> anyhow::Result<()> { let (_temp, ctx) = context().await?; insert_contact(&ctx, 8).await?; - let database = ctx.get_app_database().await; + let database = ctx.app_db.read().await.clone(); let share = vec![4, 5, 6]; sqlx::query!( "UPDATE contacts SET recovery_secret_share = ? WHERE user_id = 8", @@ -362,7 +362,7 @@ mod tests { async fn heartbeat_without_share_queues_deletion_response() -> anyhow::Result<()> { let (_temp, ctx) = context().await?; insert_contact(&ctx, 9).await?; - let database = ctx.get_app_database().await; + let database = ctx.app_db.read().await.clone(); let mut transaction = database.pool.begin().await?; handle_passwordless_recovery_heartbeat( &mut transaction, diff --git a/rust/src/api/messages/incoming/client2client/user_discovery.rs b/rust/src/api/messages/incoming/client2client/user_discovery.rs index 6780b7be..5d19a9a5 100644 --- a/rust/src/api/messages/incoming/client2client/user_discovery.rs +++ b/rust/src/api/messages/incoming/client2client/user_discovery.rs @@ -87,7 +87,7 @@ pub(crate) async fn handle_user_discovery_request( )); } let messages = ctx - .get_user_discovery() + .user_discovery .get() .await .get_new_messages(from_user_id, &request.current_version, t) @@ -124,7 +124,7 @@ pub(crate) async fn handle_user_discovery_update( } Ok(ctx - .get_user_discovery() + .user_discovery .get() .await .handle_new_messages(from_user_id, None, update.messages, t) diff --git a/rust/src/api/messages/incoming/mod.rs b/rust/src/api/messages/incoming/mod.rs index 05136fb6..4a0c6785 100644 --- a/rust/src/api/messages/incoming/mod.rs +++ b/rust/src/api/messages/incoming/mod.rs @@ -101,7 +101,7 @@ pub(crate) async fn handle_sealed_message(ctx: &Arc, bytes: Vec) -> pub(crate) async fn handle_request_new_pqc_prekeys( ctx: &Arc, ) -> Result { - let engine = ctx.get_signal_engine().lock().await; + let engine = ctx.signal_engine.lock().await; let prekeys = engine .as_ref() @@ -160,7 +160,7 @@ pub(crate) async fn handle_decoded_server_message( ensure_contact_exists(ctx, from_user_id).await?; } - let database = ctx.get_app_database().await; + let database = ctx.app_db.read().await.clone(); let mut tr = database.pool.begin().await?; @@ -232,7 +232,7 @@ pub(crate) async fn handle_decoded_server_message( TwonlyError::Generic("V2 encrypted client message has no ciphertext".into()) })?; let decrypted = { - let engine = ctx.get_signal_engine().lock().await; + let engine = ctx.signal_engine.lock().await; engine .as_ref() .ok_or(TwonlyError::SignalIdentityNotFound)? @@ -278,7 +278,7 @@ pub(crate) async fn handle_decoded_server_message( Type::TestNotification => {} } - if is_encrypted_message && !sends_error_response { + if is_encrypted_message & !sends_error_response { queue_sender_delivery_receipt(&mut tr, from_user_id, &message.receipt_id).await?; } diff --git a/rust/src/api/messages/outgoing.rs b/rust/src/api/messages/outgoing.rs index 32324849..9d56452d 100644 --- a/rust/src/api/messages/outgoing.rs +++ b/rust/src/api/messages/outgoing.rs @@ -25,7 +25,7 @@ pub(crate) async fn decorate_content( content.sender_profile_counter = Some(config.avatar_counter); if config.ask_for_friend_promotions { - let database = ctx.get_app_database().await; + let database = ctx.app_db.read().await.clone(); let accepted = sqlx::query_scalar!("SELECT COUNT(*) FROM contacts WHERE accepted = 1") .fetch_one(&database.pool) .await?; @@ -34,9 +34,9 @@ pub(crate) async fn decorate_content( } } - if config.is_user_discovery_enabled && is_persisted_message { + if config.is_user_discovery_enabled & is_persisted_message { ctx.initialize_user_discovery_from_config().await?; - let database = ctx.get_app_database().await; + let database = ctx.app_db.read().await.clone(); let allowed = sqlx::query_scalar!( r#"SELECT EXISTS(SELECT 1 FROM contacts WHERE user_id = ? AND accepted = 1 AND blocked = 0 AND media_send_counter >= ? AND user_discovery_excluded = 0 @@ -49,7 +49,7 @@ pub(crate) async fn decorate_content( .await?; if allowed != 0 { content.sender_user_discovery_version = Some( - ctx.get_user_discovery() + ctx.user_discovery .get() .await .get_current_version() @@ -74,7 +74,7 @@ pub async fn send_c2c_message_to_contact( decorate_content(ctx, contact_id, &mut content, message_id.is_some()).await?; - let db_app = ctx.get_app_database().await; + let db_app = ctx.app_db.read().await.clone(); let mut t = db_app.pool.begin().await?; if only_send_if_no_receipts_are_open { diff --git a/rust/src/api/runtime.rs b/rust/src/api/runtime.rs index ef644abd..de9b65b7 100644 --- a/rust/src/api/runtime.rs +++ b/rust/src/api/runtime.rs @@ -24,7 +24,7 @@ pub struct ApiRuntime {} impl ApiRuntime { pub(crate) async fn initialize(ctx: &Arc) -> Result<()> { let config = ApiConfig::from_rust_state(ctx).await?; - ctx.get_api_client() + ctx.api_client .set(tokio::sync::RwLock::new(ApiClient::new(ctx, config))) .map_err(|_| TwonlyError::Initialization) } @@ -33,7 +33,7 @@ impl ApiRuntime { let replacement = ApiClient::new(ctx, ApiConfig::from_rust_state(ctx).await?); let current = Self::client(ctx).await?; let slot = ctx - .get_api_client() + .api_client .get() .ok_or(TwonlyError::Initialization)?; *slot.write().await = replacement; @@ -129,7 +129,7 @@ impl ApiRuntime { } pub(crate) async fn replay_outbox(ctx: &Arc) -> Result<()> { - let database = ctx.get_app_database().await; + let database = ctx.app_db.read().await.clone(); let rows = sqlx::query!("SELECT sequence_id, payload FROM api_outbox ORDER BY created_at") .fetch_all(&database.pool) .await?; @@ -154,7 +154,7 @@ impl ApiRuntime { pub(crate) async fn replay_legacy_raw_outbox(ctx: &Arc) -> Result<()> { use base64::Engine as _; - let path = std::path::Path::new(&ctx.get_config()?.data_dir) + let path = std::path::Path::new(&ctx.config.data_dir) .join("keyvalue") .join("rawbytes-to-retransmit.json"); if !path.exists() { @@ -229,7 +229,7 @@ impl ApiRuntime { pub(crate) async fn client(ctx: &Arc) -> Result> { let client = ctx - .get_api_client() + .api_client .get() .ok_or(TwonlyError::Initialization)?; Ok(client.read().await.clone()) diff --git a/rust/src/api/runtime/auth.rs b/rust/src/api/runtime/auth.rs index fdd6fdb1..64268419 100644 --- a/rust/src/api/runtime/auth.rs +++ b/rust/src/api/runtime/auth.rs @@ -103,8 +103,9 @@ impl ApiAuthHandshaker { ) -> Result<()> { let login_token = self .context - .get_key_manager() - .await? + .key_manager + .lock() + .await .main_key .get_login_token() .to_vec(); @@ -131,7 +132,7 @@ impl ApiAuthHandshaker { user_id: i64, ) -> Result { use base64::Engine as _; - let Some(encoded) = self.context.get_secure_storage().read("api_auth_token")? else { + let Some(encoded) = self.context.secure_storage.read("api_auth_token")? else { return Ok(false); }; let auth_token = base64::engine::general_purpose::STANDARD @@ -186,8 +187,9 @@ impl ApiAuthHandshaker { }; let serialized_identity = self .context - .get_key_manager() - .await? + .key_manager + .lock() + .await .signal_identity .as_ref() .ok_or(TwonlyError::SignalIdentityNotFound)? @@ -218,7 +220,7 @@ impl ApiAuthHandshaker { "auth-token response has unexpected payload".into(), )); }; - self.context.get_secure_storage().write( + self.context.secure_storage.write( "api_auth_token", &base64::engine::general_purpose::STANDARD.encode(token), )?; @@ -232,8 +234,9 @@ impl ApiAuthHandshaker { ) -> Result<()> { let token = self .context - .get_key_manager() - .await? + .key_manager + .lock() + .await .main_key .get_login_token() .to_vec(); @@ -269,7 +272,7 @@ impl ApiAuthHandshaker { ServerResult::Ok(_) => {} ServerResult::ErrorCode(code) => return Err(TwonlyError::Api(code)), } - self.context.get_secure_storage().delete("api_auth_token")?; + self.context.secure_storage.delete("api_auth_token")?; let _ = self.events.send(ApiEvent { kind: ApiEventKind::LoginTokenMigrated, state: None, diff --git a/rust/src/api/runtime/client.rs b/rust/src/api/runtime/client.rs index 1b5130d6..81f4cbf8 100644 --- a/rust/src/api/runtime/client.rs +++ b/rust/src/api/runtime/client.rs @@ -202,8 +202,8 @@ impl ApiClient { pub(crate) async fn set_network_available(self: &Arc, available: bool) -> Result<()> { self.network_available.store(available, Ordering::Release); if available - && !self.in_background.load(Ordering::Acquire) - && self.ws_client.lock().await.is_none() + & !self.in_background.load(Ordering::Acquire) + & self.ws_client.lock().await.is_none() { self.connect().await?; } diff --git a/rust/src/api/runtime/helpers.rs b/rust/src/api/runtime/helpers.rs index 02ee4a62..0666d8b2 100644 --- a/rust/src/api/runtime/helpers.rs +++ b/rust/src/api/runtime/helpers.rs @@ -79,7 +79,7 @@ pub(crate) fn schedule_post_authentication(ctx: &Arc, in_background: bo } if let Err(error) = ctx - .get_user_discovery() + .user_discovery .get() .await .on_connected(&ctx) @@ -88,7 +88,7 @@ pub(crate) fn schedule_post_authentication(ctx: &Arc, in_background: bo tracing::warn!("user-discovery post-connection refresh failed: {error}"); } - let signal_engine = ctx.get_signal_engine().lock().await; + let signal_engine = ctx.signal_engine.lock().await; if let Some(engine) = signal_engine.as_ref() { if let Err(error) = engine.on_connected(&ctx).await { tracing::warn!("Signal key maintenance failed: {error}"); diff --git a/rust/src/api/runtime/request.rs b/rust/src/api/runtime/request.rs index 561f2721..43d56918 100644 --- a/rust/src/api/runtime/request.rs +++ b/rust/src/api/runtime/request.rs @@ -105,7 +105,7 @@ impl ApiClient { ) -> Result> { let sequence = self.next_sequence().await; let context = self.context.upgrade().ok_or(TwonlyError::Initialization)?; - let database = context.get_app_database().await; + let database = context.app_db.read().await.clone(); sqlx::query!( "INSERT INTO api_outbox(sequence_id, operation_kind, payload) VALUES(?, ?, ?)", @@ -224,7 +224,7 @@ impl ApiClient { if code == ErrorCode::UserIdNotFound as i32 { if let Some(contact_id) = contact_id { let context = self.context.upgrade().ok_or(TwonlyError::Initialization)?; - let database = context.get_app_database().await; + let database = context.app_db.read().await.clone(); let mut transaction = database.pool.begin().await?; sqlx::query!( "UPDATE contacts SET account_deleted = 1 WHERE user_id = ?", diff --git a/rust/src/api/server/account.rs b/rust/src/api/server/account.rs index 4bce3ab2..eaccf16b 100644 --- a/rust/src/api/server/account.rs +++ b/rust/src/api/server/account.rs @@ -21,7 +21,7 @@ impl Server { lang_code: String, is_ios: bool, ) -> Result> { - let key_manager = ctx.get_key_manager().await?; + let key_manager = ctx.key_manager.lock().await; let identity = key_manager .signal_identity .as_ref() diff --git a/rust/src/api/server/contacts.rs b/rust/src/api/server/contacts.rs index 8b7dd4e5..3208caa1 100644 --- a/rust/src/api/server/contacts.rs +++ b/rust/src/api/server/contacts.rs @@ -36,7 +36,7 @@ impl Server { } pub async fn check_for_deleted_usernames(ctx: &Arc) -> Result<()> { - let database = ctx.get_app_database().await; + let database = ctx.app_db.read().await.clone(); let contacts = sqlx::query_scalar!( "SELECT user_id FROM contacts WHERE username IN ('[deleted]', '[Unknown]')" ) diff --git a/rust/src/api/server/prekeys.rs b/rust/src/api/server/prekeys.rs index 6d29fe65..b8338646 100644 --- a/rust/src/api/server/prekeys.rs +++ b/rust/src/api/server/prekeys.rs @@ -20,7 +20,7 @@ pub struct PqcPreKeyInput { impl Server { #[doc(hidden)] pub async fn generate_and_upload_pqc_pre_keys(ctx: &Arc) -> Result> { - let engine = ctx.get_signal_engine().lock().await; + let engine = ctx.signal_engine.lock().await; let bundle = engine .as_ref() .ok_or(crate::error::TwonlyError::SignalIdentityNotFound)? diff --git a/rust/src/backup/backup_archive.rs b/rust/src/backup/backup_archive.rs index 98a0a89e..6d1fa503 100644 --- a/rust/src/backup/backup_archive.rs +++ b/rust/src/backup/backup_archive.rs @@ -14,6 +14,7 @@ use std::collections::BTreeMap; use std::fs::{remove_file, File}; use std::io::{copy, Cursor}; use std::path::{Path, PathBuf}; +use std::sync::Arc; use walkdir::WalkDir; use zeroize::Zeroize; use zip::write::SimpleFileOptions; @@ -44,7 +45,7 @@ impl BackupArchive { ctx: &Context, keys: &KeyManager, ) -> Result)>> { - let config = ctx.get_config()?; + let config = &ctx.config; let database_dir = PathBuf::from(&config.database_dir); let data_dir = PathBuf::from(&config.data_dir); let rust_db_key = keys.main_key.get_database_key(DatabaseKey::RustDb); @@ -65,7 +66,7 @@ impl BackupArchive { } pub(crate) async fn create_backup(ctx: &Context) -> Result { - let config = ctx.get_config()?; + let config = &ctx.config; let data_dir = PathBuf::from(&config.data_dir); let backup_data_dir = data_dir.join("temp_backup_dir"); @@ -74,7 +75,7 @@ impl BackupArchive { } std::fs::create_dir_all(&backup_data_dir)?; - let keys = ctx.get_key_manager().await?; + let keys = ctx.key_manager.lock().await; for (file_name, source_dir, is_db, mut encryption_key) in Self::get_backup_files(ctx, &keys)? @@ -91,7 +92,7 @@ impl BackupArchive { if is_db { if file_name == APP_DATABASE_FILE { let backup_database_file = backup_data_dir.join(file_name); - let app_database = ctx.get_app_database().await; + let app_database = ctx.app_db.read().await.clone(); app_database .create_backup( &backup_database_file.display().to_string(), @@ -178,8 +179,8 @@ impl BackupArchive { } pub(crate) async fn restore_from_backup(ctx: &Context, file_path: &Path) -> Result<()> { - let data_dir = PathBuf::from(&ctx.get_config()?.data_dir); - let key_manager = ctx.get_key_manager().await?; + let data_dir = PathBuf::from(&ctx.config.data_dir); + let key_manager = ctx.key_manager.lock().await; let encrypted_zip = std::fs::read(file_path)?; let zip_content = key_manager.main_key.decrypt_backup(&encrypted_zip)?; @@ -260,9 +261,9 @@ impl BackupArchive { // app_db.sqlite is owned by a replaceable Rust handle. Close it before // replacing the file so subsequent DAO calls cannot continue using an // unlinked pre-restore database. - let current_app_database = ctx.get_app_database().await; + let current_app_database = ctx.app_db.read().await.clone(); current_app_database.pool.close().await; - let current_rust_database = ctx.get_rust_db().await; + let current_rust_database = ctx.rust_db.read().await.clone(); current_rust_database.pool.close().await; for (file_name, target_dir, is_db, _) in Self::get_backup_files(ctx, &key_manager)? { @@ -281,7 +282,7 @@ impl BackupArchive { } } - let database_dir = PathBuf::from(&ctx.get_config()?.database_dir); + let database_dir = PathBuf::from(&ctx.config.database_dir); let app_database_path = database_dir.join(APP_DATABASE_FILE); let app_database = crate::database::app::AppDatabase::new( &app_database_path.display().to_string(), @@ -290,7 +291,8 @@ impl BackupArchive { ) .await?; app_database.run_migrations().await?; - ctx.replace_app_database(app_database).await; + + *ctx.app_db.write().await = Arc::new(app_database); let rust_database_path = database_dir.join("rust_db.sqlite"); let rust_database = Database::new( @@ -397,8 +399,8 @@ mod tests { // 1. Add some data let original_login_token = { let secure_storage = SecureStorage::new("testing"); - let config = ctx.get_config().unwrap(); - let key_manager = ctx.get_key_manager().await.unwrap(); + let config = &ctx.config; + let key_manager = ctx.key_manager.lock().await; key_manager.store_to_keychain(&secure_storage).unwrap(); // Add a file @@ -407,7 +409,7 @@ mod tests { key_manager.main_key.get_login_token() }; { - let app_db = ctx.get_app_database().await; + let app_db = ctx.app_db.read().await.clone(); sqlx::query!( r#" INSERT INTO contacts(user_id, username) @@ -425,12 +427,12 @@ mod tests { // 3. Modify data (to simulate state before restore) { - let config = ctx.get_config().unwrap(); + let config = &ctx.config; let config_file = PathBuf::from(&config.data_dir).join("user_discovery_config.json"); std::fs::write(config_file, "new config").unwrap(); - let app_db = ctx.get_app_database().await; + let app_db = ctx.app_db.read().await.clone(); sqlx::query!( r#" UPDATE contacts @@ -450,8 +452,8 @@ mod tests { // 5. Verify restored data { - let config = ctx.get_config().unwrap(); - let key_manager = ctx.get_key_manager().await.unwrap(); + let config = &ctx.config; + let key_manager = ctx.key_manager.lock().await; let config_file = PathBuf::from(&config.data_dir).join("user_discovery_config.json"); let config_content = std::fs::read_to_string(config_file).unwrap(); @@ -459,7 +461,7 @@ mod tests { assert_eq!(key_manager.main_key.get_login_token(), original_login_token); - let app_db = ctx.get_app_database().await; + let app_db = ctx.app_db.read().await.clone(); let username = sqlx::query_scalar!( r#" SELECT username @@ -484,7 +486,7 @@ mod tests { ) .await .unwrap(); - let database_dir = PathBuf::from(&ctx.get_config().unwrap().database_dir); + let database_dir = PathBuf::from(&ctx.config.database_dir); let legacy_path = database_dir.join("twonly.sqlite"); let legacy = crate::database::app::AppDatabase::new(&legacy_path.display().to_string(), None, false) @@ -509,7 +511,7 @@ mod tests { let archive_path = BackupArchive::create_backup(&ctx).await.unwrap(); remove_file_from_encrypted_archive(&ctx, &archive_path, APP_DATABASE_FILE).await; - let app_db = ctx.get_app_database().await; + let app_db = ctx.app_db.read().await.clone(); sqlx::query!( r#" INSERT INTO contacts(user_id, username) @@ -524,7 +526,7 @@ mod tests { .await .unwrap(); - let restored = ctx.get_app_database().await; + let restored = ctx.app_db.read().await.clone(); let contacts = sqlx::query!( r#" SELECT user_id, username @@ -546,7 +548,7 @@ mod tests { archive_path: &Path, excluded_name: &str, ) { - let keys = ctx.get_key_manager().await.unwrap(); + let keys = ctx.key_manager.lock().await; let encrypted = std::fs::read(archive_path).unwrap(); let decrypted = keys.main_key.decrypt_backup(&encrypted).unwrap(); let mut source = ZipArchive::new(Cursor::new(decrypted)).unwrap(); diff --git a/rust/src/bridge/mod.rs b/rust/src/bridge/mod.rs index 9ac1bba8..dcfde22c 100644 --- a/rust/src/bridge/mod.rs +++ b/rust/src/bridge/mod.rs @@ -10,24 +10,13 @@ pub mod groups; pub mod user_config; pub mod wrapper; -use std::sync::Arc; -use crate::api::runtime::ApiClient; use crate::context::Context; -use crate::database::app::AppDatabase; -use crate::database::signal::Database; use crate::error::Result; -use crate::error::TwonlyError; -use crate::keys::KeyManager; -use crate::secure_storage::SecureStorage; -use crate::signal::engine::RustSignalEngine; -use crate::user_discovery::UserDiscovery; -use crate::utils::Shared; use flutter_rust_bridge::frb; pub use crate::user_discovery::AnnouncedUser; pub use crate::user_discovery::OtherPromotion; -use tokio::sync::{Mutex, OnceCell, RwLock}; pub struct InitConfig { pub database_dir: String, @@ -51,26 +40,9 @@ pub struct _AnnouncedUser { pub public_id: i64, } -pub(crate) struct TwonlyFlutter { - #[allow(dead_code)] - pub(crate) config: InitConfig, - pub(crate) user_discovery: Shared, - #[allow(dead_code)] - pub(crate) rust_db: Arc>>, - pub(crate) app_db: Arc>>, - pub(crate) secure_storage: SecureStorage, - pub(crate) key_manager: Arc>, - pub(crate) signal_engine: Arc>>, - pub(crate) api_client: OnceCell>>, -} - -pub(super) fn get_twonly_flutter() -> Result<&'static TwonlyFlutter> { +pub(super) fn get_twonly_flutter() -> Result<&'static Context> { let ctx = Context::get_static()?; - if let Context::Flutter(twonly) = &**ctx { - Ok(twonly) - } else { - Err(TwonlyError::Initialization) - } + Ok(&**ctx) } pub async fn initialize_twonly_flutter(config: InitConfig) -> Result<()> { diff --git a/rust/src/bridge/user_config.rs b/rust/src/bridge/user_config.rs index f434ff90..28ebb0b5 100644 --- a/rust/src/bridge/user_config.rs +++ b/rust/src/bridge/user_config.rs @@ -37,10 +37,10 @@ impl UserConfigApi { let ctx = Context::get_static()?; let normalized = UserConfig::save_json(ctx, &serde_json::to_string(&config)?)?; let config: UserConfig = serde_json::from_str(&normalized)?; - let mut key_manager = ctx.get_key_manager().await?; + let mut key_manager = ctx.key_manager.lock().await; if key_manager.user_id != Some(config.user_id) { key_manager.user_id = Some(config.user_id); - key_manager.store_to_keychain(ctx.get_secure_storage())?; + key_manager.store_to_keychain(&ctx.secure_storage)?; } drop(key_manager); if let Ok(callbacks) = crate::bridge::callbacks::get_callbacks() { diff --git a/rust/src/context.rs b/rust/src/context.rs index fec5ecbf..dead523b 100644 --- a/rust/src/context.rs +++ b/rust/src/context.rs @@ -11,29 +11,22 @@ use crate::error::Result; use crate::error::TwonlyError; use crate::keys::DatabaseKey; use crate::keys::KeyManager; -#[cfg(not(test))] use crate::log::init_tracing; +use crate::secure_storage::SecureStorage; use crate::signal::engine::RustSignalEngine; use crate::user_discovery::UserDiscovery; use crate::utils::Shared; -use crate::{bridge::TwonlyFlutter, secure_storage::SecureStorage}; use libsignal_protocol::IdentityKey; -use libsignal_protocol::IdentityKeyPair; use std::{path::PathBuf, sync::Arc}; use tokio::sync::{Mutex, OnceCell, RwLock}; -#[cfg(not(test))] use zeroize::Zeroize; -#[cfg(not(test))] static GLOBAL_CONTEXT: OnceCell> = OnceCell::const_new(); -pub struct TwonlyStandalone { - #[allow(dead_code)] +pub struct Context { pub(crate) config: InitConfig, - #[allow(dead_code)] pub(crate) rust_db: Arc>>, - pub(crate) app_db: Arc>>, - #[allow(dead_code)] + pub app_db: Arc>>, pub(crate) secure_storage: SecureStorage, pub(crate) key_manager: Arc>, pub(crate) user_discovery: Shared, @@ -41,45 +34,7 @@ pub struct TwonlyStandalone { pub(crate) api_client: OnceCell>>, } -#[allow(private_interfaces)] // for the test -pub enum Context { - Flutter(TwonlyFlutter), - Standalone(TwonlyStandalone), -} - impl Context { - pub(crate) fn get_api_client(&self) -> &OnceCell>> { - match self { - Self::Flutter(value) => &value.api_client, - Self::Standalone(value) => &value.api_client, - } - } - - pub(crate) fn data_dir(&self) -> &str { - match self { - Self::Flutter(value) => &value.config.data_dir, - Self::Standalone(value) => &value.config.data_dir, - } - } - - pub(crate) fn get_user_discovery(&self) -> &Shared { - match self { - Self::Flutter(value) => &value.user_discovery, - Self::Standalone(value) => &value.user_discovery, - } - } - - pub(crate) fn get_signal_engine(&self) -> &Arc>> { - match self { - Self::Flutter(value) => &value.signal_engine, - Self::Standalone(value) => &value.signal_engine, - } - } - - pub fn from_standalone(standalone: TwonlyStandalone) -> Self { - Self::Standalone(standalone) - } - pub(crate) async fn init_flutter(config: InitConfig) -> Result<()> { Self::init_common(config, true).await } @@ -135,7 +90,7 @@ impl Context { rust_db.clone(), )?); - let ctx = Arc::new(Context::from_standalone(TwonlyStandalone { + let ctx = Arc::new(Context { config, rust_db, app_db, @@ -144,7 +99,7 @@ impl Context { user_discovery, signal_engine: Arc::new(Mutex::new(None)), api_client: OnceCell::const_new(), - })); + }); ApiRuntime::initialize(&ctx).await?; ApiRuntime::connect(&ctx).await?; Ok(ctx) @@ -158,7 +113,7 @@ impl Context { registration_id: i64, pre_key_store: std::collections::HashMap>, ) -> Result<()> { - let mut key_manager = self.get_key_manager().await?; + let mut key_manager = self.key_manager.lock().await; key_manager.signal_identity = Some(crate::keys::SignalIdentityKey { identity_key_pair_structure, registration_id, @@ -170,9 +125,9 @@ impl Context { #[doc(hidden)] #[cfg(any(test, debug_assertions))] pub async fn inject_test_user_id(&self, user_id: i64) -> Result<()> { - let mut key_manager = self.get_key_manager().await?; + let mut key_manager = self.key_manager.lock().await; key_manager.user_id = Some(user_id); - key_manager.store_to_keychain(self.get_secure_storage())?; + key_manager.store_to_keychain(&self.secure_storage)?; let signal_identity = key_manager.signal_identity.as_ref().map(|identity| { ( identity.identity_key_pair_structure.clone(), @@ -182,8 +137,8 @@ impl Context { drop(key_manager); if let Some((identity_key_pair_structure, registration_id)) = signal_identity { - let database = self.get_rust_db().await; - *self.get_signal_engine().lock().await = Some(RustSignalEngine::new_with_pool( + let database = self.rust_db.read().await.clone(); + *self.signal_engine.lock().await = Some(RustSignalEngine::new_with_pool( database.pool.clone(), identity_key_pair_structure, registration_id as u32, @@ -195,72 +150,13 @@ impl Context { } pub(crate) async fn initialize_user_discovery_from_config(&self) -> Result<()> { - let Some(config) = crate::user_config::UserConfig::load_from(self)? else { - return Ok(()); - }; - if !config.is_user_discovery_enabled { - return Ok(()); - } - - let discovery_config_path = - PathBuf::from(self.data_dir()).join("user_discovery_config.json"); - let settings_are_current = std::fs::read_to_string(&discovery_config_path) - .ok() - .and_then(|value| serde_json::from_str::(&value).ok()) - .is_some_and(|value| { - value.get("threshold").and_then(serde_json::Value::as_u64) - == Some(u64::from(config.user_discovery_threshold)) - && value - .get("share_promotion") - .and_then(serde_json::Value::as_bool) - == Some(config.user_discovery_share_promotion) - }); - if settings_are_current { - let database = self.get_app_database().await; - let has_shares = - sqlx::query_scalar!("SELECT EXISTS(SELECT 1 FROM user_discovery_shares LIMIT 1)") - .fetch_one(&database.pool) - .await? - != 0; - if has_shares { - return Ok(()); - } - } - - let key_manager = self.get_key_manager().await?; - let user_id = key_manager.user_id.ok_or_else(|| { - TwonlyError::Generic("cannot initialize user discovery without user ID".into()) - })?; - let identity = key_manager - .signal_identity - .as_ref() - .ok_or(TwonlyError::SignalIdentityNotFound)?; - let identity = IdentityKeyPair::try_from(identity.identity_key_pair_structure.as_slice()) - .map_err(|error| { - TwonlyError::Generic(format!("invalid Signal identity: {error}")) - })?; - let public_key = identity.identity_key().serialize().to_vec(); - drop(key_manager); - - let database = self.get_app_database().await; - let mut transaction = database.pool.begin().await?; - self.get_user_discovery() + self.user_discovery .get() .await - .initialize_or_update( - config.user_discovery_threshold, - user_id, - public_key, - config.user_discovery_share_promotion, - &mut transaction, - ) - .await?; - transaction.commit().await?; - database.notify_committed(["user_discovery_shares"]); - Ok(()) + .initialize_from_config(self) + .await } - #[cfg(not(test))] async fn init_common(config: InitConfig, is_flutter: bool) -> Result<()> { if GLOBAL_CONTEXT.initialized() { tracing::info!("twonly already initialized. Ensuring storage directories exist."); @@ -322,7 +218,6 @@ impl Context { app_db.run_migrations().await?; let app_db = Arc::new(RwLock::new(Arc::new(app_db))); app_db_key.zeroize(); - rust_db_key.zeroize(); if is_flutter { @@ -350,7 +245,7 @@ impl Context { key_manager.clone(), rust_db_handle.clone(), )?); - let ctx = Arc::new(Context::Flutter(TwonlyFlutter { + let ctx = Arc::new(Context { config, secure_storage, rust_db: rust_db_handle, @@ -359,7 +254,7 @@ impl Context { user_discovery, signal_engine, api_client: OnceCell::const_new(), - })); + }); if let Err(error) = ctx.initialize_user_discovery_from_config().await { tracing::warn!("failed to initialize user discovery: {error}"); } @@ -385,7 +280,7 @@ impl Context { key_manager.clone(), rust_db_handle.clone(), )?); - let ctx = Arc::new(Context::Standalone(TwonlyStandalone { + let ctx = Arc::new(Context { config, rust_db: rust_db_handle, app_db, @@ -394,7 +289,7 @@ impl Context { user_discovery, signal_engine, api_client: OnceCell::const_new(), - })); + }); if let Err(error) = ctx.initialize_user_discovery_from_config().await { tracing::warn!("failed to initialize user discovery: {error}"); } @@ -408,65 +303,20 @@ impl Context { Ok(()) } - #[cfg(test)] - async fn init_common(_config: InitConfig, _is_flutter: bool) -> Result<()> { - Err(TwonlyError::Initialization) - } - - #[cfg(not(test))] pub(super) fn get_static() -> Result<&'static Arc> { GLOBAL_CONTEXT.get().ok_or(TwonlyError::Initialization) } - #[cfg(test)] - pub(super) fn get_static() -> Result<&'static Arc> { - Err(TwonlyError::Initialization) - } - - pub(crate) fn get_secure_storage(&self) -> &SecureStorage { - match self { - Self::Flutter(twonly) => &twonly.secure_storage, - Self::Standalone(twonly) => &twonly.secure_storage, - } - } - - pub(crate) fn get_config(&self) -> Result<&InitConfig> { - match self { - Self::Flutter(twonly) => Ok(&twonly.config), - Self::Standalone(twonly) => Ok(&twonly.config), - } - } - - pub(crate) async fn get_key_manager(&self) -> Result> { - match self { - Self::Flutter(twonly) => Ok(twonly.key_manager.lock().await), - Self::Standalone(twonly) => Ok(twonly.key_manager.lock().await), - } - } - pub(crate) async fn user_id(&self) -> Result { - self.get_key_manager() - .await? + self.key_manager + .lock() + .await .user_id .ok_or_else(|| TwonlyError::Generic("local user ID is missing".into())) } - pub async fn get_app_database(&self) -> Arc { - match self { - Self::Flutter(twonly) => twonly.app_db.read().await.clone(), - Self::Standalone(twonly) => twonly.app_db.read().await.clone(), - } - } - - pub(crate) async fn get_rust_db(&self) -> Arc { - match self { - Self::Flutter(twonly) => twonly.rust_db.read().await.clone(), - Self::Standalone(twonly) => twonly.rust_db.read().await.clone(), - } - } - pub(crate) async fn get_identity(&self, user_id: i64) -> Result> { - let database = self.get_rust_db().await; + let database = self.rust_db.read().await.clone(); let user_id = user_id.to_string(); let identity_key = sqlx::query_scalar!( r#"SELECT identity_key FROM signal_identities WHERE name = ?"#, @@ -488,29 +338,17 @@ impl Context { key_manager: &KeyManager, ) -> Result<()> { let database = Arc::new(database); - match self { - Self::Flutter(twonly) => { - *twonly.rust_db.write().await = database.clone(); - let engine = match (key_manager.user_id, &key_manager.signal_identity) { - (Some(user_id), Some(identity)) => Some(RustSignalEngine::new_with_pool( - database.pool.clone(), - identity.identity_key_pair_structure.clone(), - identity.registration_id as u32, - user_id.to_string(), - )?), - _ => None, - }; - *twonly.signal_engine.lock().await = engine; - } - Self::Standalone(twonly) => *twonly.rust_db.write().await = database, - } + *self.rust_db.write().await = database.clone(); + let engine = match (key_manager.user_id, &key_manager.signal_identity) { + (Some(user_id), Some(identity)) => Some(RustSignalEngine::new_with_pool( + database.pool.clone(), + identity.identity_key_pair_structure.clone(), + identity.registration_id as u32, + user_id.to_string(), + )?), + _ => None, + }; + *self.signal_engine.lock().await = engine; Ok(()) } - - pub(crate) async fn replace_app_database(&self, database: AppDatabase) { - match self { - Self::Flutter(twonly) => *twonly.app_db.write().await = Arc::new(database), - Self::Standalone(twonly) => *twonly.app_db.write().await = Arc::new(database), - } - } } diff --git a/rust/src/database/app/tables/contact.rs b/rust/src/database/app/tables/contact.rs index 0698e9c5..527dca74 100644 --- a/rust/src/database/app/tables/contact.rs +++ b/rust/src/database/app/tables/contact.rs @@ -6,6 +6,7 @@ use crate::{ context::Context, error::{Result, TwonlyError}, + user_config::UserConfig, }; use sqlx::{Sqlite, Transaction}; @@ -76,7 +77,7 @@ impl Contact { .unwrap_or_else(|| self.username.clone()) } - pub async fn update(tr: &mut Transaction<'_, Sqlite>, contact: UpdateContact) -> Result<()> { + pub async fn update(t: &mut Transaction<'_, Sqlite>, contact: UpdateContact) -> Result<()> { let update_display_name = contact.display_name.is_some(); let display_name = contact.display_name.flatten(); let update_avatar = contact.avatar_svg_compressed.is_some(); @@ -109,14 +110,14 @@ impl Contact { contact.user_id, contact.only_if_not_requested, ) - .execute(&mut **tr) + .execute(&mut **t) .await?; Ok(()) } pub async fn insert_on_conflict_update( - tr: &mut Transaction<'_, Sqlite>, + t: &mut Transaction<'_, Sqlite>, contact: UpdateContact, ) -> Result<()> { sqlx::query!( @@ -147,25 +148,25 @@ impl Contact { contact.blocked, contact.only_if_not_requested, ) - .execute(&mut **tr) + .execute(&mut **t) .await?; Ok(()) } pub async fn get_contact_by_id( - transaction: &mut Transaction<'_, Sqlite>, + t: &mut Transaction<'_, Sqlite>, user_id: i64, ) -> Result> { let contact = sqlx::query_as!(Self, "SELECT * FROM contacts WHERE user_id = ?", user_id) - .fetch_optional(&mut **transaction) + .fetch_optional(&mut **t) .await?; Ok(contact) } pub async fn update_ask_for_friend_promotions( - transaction: &mut Transaction<'_, Sqlite>, + t: &mut Transaction<'_, Sqlite>, user_id: i64, ) -> Result<()> { sqlx::query!( @@ -176,7 +177,7 @@ impl Contact { "#, user_id, ) - .execute(&mut **transaction) + .execute(&mut **t) .await?; Ok(()) } @@ -186,7 +187,7 @@ impl Contact { t: &mut Transaction<'_, Sqlite>, contact_id: i64, ) -> Result { - let config = crate::user_config::UserConfig::load_required_from(context)?; + let config = UserConfig::load_required_from(context)?; let contact = sqlx::query!( r#"SELECT accepted, blocked, media_send_counter, user_discovery_excluded, user_discovery_manual_approved @@ -205,22 +206,19 @@ impl Contact { })) } - pub async fn exists(transaction: &mut Transaction<'_, Sqlite>, user_id: i64) -> Result { + pub async fn exists(t: &mut Transaction<'_, Sqlite>, user_id: i64) -> Result { let exists = sqlx::query_scalar!( "SELECT EXISTS(SELECT 1 FROM contacts WHERE user_id = ?)", user_id ) - .fetch_one(&mut **transaction) + .fetch_one(&mut **t) .await? != 0; Ok(exists) } - pub async fn ensure_exists( - transaction: &mut Transaction<'_, Sqlite>, - user_id: i64, - ) -> Result<()> { - if !Self::exists(transaction, user_id).await? { + pub async fn ensure_exists(t: &mut Transaction<'_, Sqlite>, user_id: i64) -> Result<()> { + if !Self::exists(t, user_id).await? { return Err(TwonlyError::Generic(format!( "contact {user_id} does not exist" ))); diff --git a/rust/src/sealed_sender.rs b/rust/src/sealed_sender.rs index 1d46495e..d9f5015d 100644 --- a/rust/src/sealed_sender.rs +++ b/rust/src/sealed_sender.rs @@ -180,7 +180,7 @@ impl SealedSender { context: &Context, ) -> Result { let (recipient_user_id, recipient_identity) = { - let key_manager = context.get_key_manager().await.map_err(context_error)?; + let key_manager = context.key_manager.lock().await; let recipient_user_id = key_manager .user_id .ok_or(SealedSenderError::MissingLocalUserId)?; @@ -355,7 +355,7 @@ mod tests { .await .unwrap(); { - let mut key_manager = context.get_key_manager().await.unwrap(); + let mut key_manager = context.key_manager.lock().await; key_manager.user_id = Some(42); key_manager.signal_identity = Some(crate::keys::SignalIdentityKey { identity_key_pair_structure: recipient.serialize().to_vec(), @@ -363,7 +363,7 @@ mod tests { pre_key_store: Default::default(), }); } - let database = context.get_rust_db().await; + let database = context.rust_db.read().await.clone(); let sender_identity = sender.identity_key().serialize(); sqlx::query!( r#" diff --git a/rust/src/services/contacts.rs b/rust/src/services/contacts.rs index 802c8aae..ec03565c 100644 --- a/rust/src/services/contacts.rs +++ b/rust/src/services/contacts.rs @@ -38,7 +38,7 @@ impl ContactService { self.process_user_prekey_bundle(&user).await?; - let database = self.ctx.get_app_database().await; + let database = self.ctx.app_db.read().await.clone(); let mut transaction = database.pool.begin().await?; UpdateContact::builder() .user_id(user.user_id) @@ -107,7 +107,7 @@ impl ContactService { }; self.ctx - .get_signal_engine() + .signal_engine .lock() .await .as_ref() @@ -134,7 +134,7 @@ impl ContactService { } pub async fn accept_request(&self, contact_id: i64, blocking: bool) -> Result<()> { - let database = self.ctx.get_app_database().await; + let database = self.ctx.app_db.read().await.clone(); let mut transaction = database.pool.begin().await?; let contact = Contact::get_contact_by_id(&mut transaction, contact_id) .await? @@ -168,7 +168,7 @@ impl ContactService { } pub async fn reject_request(&self, contact_id: i64, blocking: bool) -> Result<()> { - let db_app = self.ctx.get_app_database().await; + let db_app = self.ctx.app_db.read().await.clone(); let mut t = db_app.pool.begin().await?; diff --git a/rust/src/services/groups/mod.rs b/rust/src/services/groups/mod.rs index e5a0c5d4..661e760c 100644 --- a/rust/src/services/groups/mod.rs +++ b/rust/src/services/groups/mod.rs @@ -45,7 +45,7 @@ impl GroupService { } async fn sync_flame_counters(&self) -> Result<()> { - let db = self.ctx.get_app_database().await; + let db = self.ctx.app_db.read().await.clone(); let groups = Group::flame_sync_candidates(&db.pool).await?; let Some(best_friend) = groups.iter().max_by_key(|group| group.total_media_counter) else { @@ -184,7 +184,7 @@ impl GroupService { }) .await?; - let db = self.ctx.get_app_database().await; + let db = self.ctx.app_db.read().await.clone(); let mut tr = db.pool.begin().await?; let serialized_identity = identity.serialize().to_vec(); InsertGroup::builder() @@ -227,7 +227,7 @@ impl GroupService { } pub async fn fetch_group_state(&self, group_id: String) -> Result { - let database = self.ctx.get_app_database().await; + let database = self.ctx.app_db.read().await.clone(); let mut t = database.pool.begin().await?; let updated = self .fetch_group_state_in_transaction(&mut t, &group_id) @@ -293,7 +293,7 @@ impl GroupService { appended_changes = true; } } - if appended_changes && group_state.admin_ids.contains(&self.ctx.user_id().await?) { + if appended_changes & group_state.admin_ids.contains(&self.ctx.user_id().await?) { GroupApi::update_remote(&group, server.version_id, &group_state, None, None).await?; } self.apply_state(t, group_id, server.version_id as i64, &group_state) @@ -521,7 +521,7 @@ impl GroupService { user.username .ok_or_else(|| TwonlyError::Generic("user response has no username".into()))?, )?; - let database = self.ctx.get_app_database().await; + let database = self.ctx.app_db.read().await.clone(); let mut tr = database.pool.begin().await?; UpdateContact::builder() .user_id(contact_id) @@ -592,7 +592,7 @@ impl GroupService { } pub async fn fetch_group_states_for_unjoined_groups(&self) -> Result<()> { - let db = self.ctx.get_app_database().await; + let db = self.ctx.app_db.read().await.clone(); let mut t = db.pool.begin().await?; self.fetch_group_states_for_unjoined_groups_in_transaction(&mut t) .await?; @@ -617,7 +617,7 @@ impl GroupService { } pub async fn fetch_missing_group_public_keys(&self) -> Result<()> { - let db = self.ctx.get_app_database().await; + let db = self.ctx.app_db.read().await.clone(); let rows = GetMissingGroupPublicKeys::builder() .build() .fetch_all(&db.pool) @@ -646,7 +646,7 @@ impl GroupService { &self, group_id: &str, ) -> Result<(Arc, GroupRecord)> { - let db = self.ctx.get_app_database().await; + let db = self.ctx.app_db.read().await.clone(); let row = GroupRecord::load(&db.pool, group_id).await?; Ok((db, row)) } diff --git a/rust/src/services/mediafiles.rs b/rust/src/services/mediafiles.rs index ed250dec..12655fa5 100644 --- a/rust/src/services/mediafiles.rs +++ b/rust/src/services/mediafiles.rs @@ -45,7 +45,7 @@ impl MediaFileService { } pub async fn download_pending(&self) -> Result<()> { - let database = self.ctx.get_app_database().await; + let database = self.ctx.app_db.read().await.clone(); // A terminated process can leave a download in this intermediate state. sqlx::query!( "UPDATE media_files SET download_state = 'pending' WHERE download_state = 'downloading'" @@ -68,7 +68,7 @@ impl MediaFileService { } pub async fn download(&self, media_id: &str) -> Result<()> { - let database = self.ctx.get_app_database().await; + let database = self.ctx.app_db.read().await.clone(); let claimed = sqlx::query!( r#"UPDATE media_files SET download_state = 'downloading' WHERE media_id = ? AND download_state = 'pending'"#, @@ -99,7 +99,7 @@ impl MediaFileService { /// handler. Wait until that transaction becomes visible before claiming it. pub async fn download_when_available(&self, media_id: &str) -> Result<()> { for _ in 0..40 { - let database = self.ctx.get_app_database().await; + let database = self.ctx.app_db.read().await.clone(); let state = sqlx::query_scalar!( "SELECT download_state FROM media_files WHERE media_id = ?", media_id, @@ -119,7 +119,7 @@ impl MediaFileService { } async fn download_claimed(&self, media_id: &str) -> Result<()> { - let database = self.ctx.get_app_database().await; + let database = self.ctx.app_db.read().await.clone(); let messages = sqlx::query!( r#"SELECT m.message_id, m.sender_id, c.account_deleted FROM messages m @@ -267,7 +267,7 @@ impl MediaFileService { // Keep the file update and state transition ordered: ready is only // visible after the plaintext has been written successfully. - let database = self.ctx.get_app_database().await; + let database = self.ctx.app_db.read().await.clone(); sqlx::query!( r#"UPDATE media_files SET download_state = 'ready', stored_file_hash = ? WHERE media_id = ?"#, @@ -282,7 +282,7 @@ impl MediaFileService { } pub async fn request_reupload(&self, media_id: &str) -> Result<()> { - let database = self.ctx.get_app_database().await; + let database = self.ctx.app_db.read().await.clone(); sqlx::query!( "UPDATE media_files SET download_state = 'reuploadRequested' WHERE media_id = ?", media_id, @@ -353,7 +353,7 @@ impl MediaFileService { fn paths(&self, media_id: &str, media_type: &str) -> Vec { let extension = Self::extension(media_type); - let base = PathBuf::from(self.ctx.data_dir()).join("mediafiles"); + let base = PathBuf::from(&self.ctx.config.data_dir).join("mediafiles"); vec![ base.join("tmp").join(format!("{media_id}.{extension}")), base.join("tmp") @@ -372,13 +372,13 @@ impl MediaFileService { } fn temp_path(&self, media_id: &str, media_type: &str) -> PathBuf { - PathBuf::from(self.ctx.data_dir()) + PathBuf::from(&self.ctx.config.data_dir) .join("mediafiles/tmp") .join(format!("{media_id}.{}", Self::extension(media_type))) } fn encrypted_path(&self, media_id: &str, media_type: &str) -> PathBuf { - PathBuf::from(self.ctx.data_dir()) + PathBuf::from(&self.ctx.config.data_dir) .join("mediafiles/tmp") .join(format!( "{media_id}.encrypted.{}", diff --git a/rust/src/services/messages.rs b/rust/src/services/messages.rs index 6e23317c..44b62425 100644 --- a/rust/src/services/messages.rs +++ b/rust/src/services/messages.rs @@ -33,7 +33,7 @@ impl MessageService { ) -> Result<()> { let mut content = proto::EncryptedContent::decode(encrypted_content.as_slice())?; content.group_id = Some(group_id.clone()); - let database = self.ctx.get_app_database().await; + let database = self.ctx.app_db.read().await.clone(); if message_id.is_some() || content.reaction.is_some() || content.media.is_some() @@ -174,7 +174,7 @@ impl MessageService { text: String, quote_message_id: Option, ) -> Result { - let database = self.ctx.get_app_database().await; + let database = self.ctx.app_db.read().await.clone(); let message_id = uuid::Uuid::new_v4().to_string(); let timestamp = chrono::Utc::now().timestamp_millis(); sqlx::query!( @@ -221,7 +221,7 @@ impl MessageService { message_type: String, additional_data: Vec, ) -> Result { - let database = self.ctx.get_app_database().await; + let database = self.ctx.app_db.read().await.clone(); let message_id = uuid::Uuid::new_v4().to_string(); let timestamp = chrono::Utc::now().timestamp_millis(); sqlx::query!( @@ -260,8 +260,8 @@ impl MessageService { group_id: String, contact_ids: Vec, ) -> Result { - let app = self.ctx.get_app_database().await; - let signal = self.ctx.get_rust_db().await; + let app = self.ctx.app_db.read().await.clone(); + let signal = self.ctx.rust_db.read().await.clone(); let mut contacts = Vec::new(); for contact_id in contact_ids { let contact = sqlx::query!( @@ -303,12 +303,11 @@ impl MessageService { ) -> Result { let local_user_id = self .ctx - .get_key_manager() - .await? + .key_manager.lock().await .user_id .ok_or_else(|| TwonlyError::Generic("local user ID is unavailable".into()))?; let group_id = Group::direct_chat_id(local_user_id, contact_id); - let database = self.ctx.get_app_database().await; + let database = self.ctx.app_db.read().await.clone(); let mut transaction = database.pool.begin().await?; let contact = Contact::get_contact_by_id(&mut transaction, contact_id) .await? @@ -438,7 +437,7 @@ impl MessageService { .encrypted_content(content.encode_to_vec()) .call() .await?; - let database = self.ctx.get_app_database().await; + let database = self.ctx.app_db.read().await.clone(); for message_id in message_ids { sqlx::query!( "UPDATE messages SET opened_at = ?, opened_by_all = ? WHERE message_id = ?", diff --git a/rust/src/signal/engine.rs b/rust/src/signal/engine.rs index a68d895c..963cea4d 100644 --- a/rust/src/signal/engine.rs +++ b/rust/src/signal/engine.rs @@ -23,7 +23,7 @@ use crate::signal::store::DbSignalProtocolStore; use crate::utils::current_time; use rand::SeedableRng; -pub struct RustSignalEngine { +pub(crate) struct RustSignalEngine { store: Arc>, local_name: String, } @@ -141,45 +141,13 @@ impl RustSignalEngine { }) } - pub fn generate_identity_key_pair() -> Result> { + #[cfg(test)] + fn generate_identity_key_pair() -> Result> { let mut csprng = rand::rngs::StdRng::from_os_rng(); let key_pair = IdentityKeyPair::generate(&mut csprng); Ok(key_pair.serialize().to_vec()) } - pub async fn generate_prekeys(&self, count: usize) -> Result)>> { - let mut store_guard = self.store.lock().await; - let store = &mut *store_guard; - let mut csprng = rand::rngs::StdRng::from_os_rng(); - let mut next_id = sqlx::query_scalar!( - r#"SELECT COALESCE(MAX(pre_key_id), 0) AS "id!: u32" FROM signal_pre_keys"#, - ) - .fetch_one(&store.pool) - .await?; - let mut prekeys = Vec::with_capacity(count); - - for _ in 0..count { - next_id = if next_id >= 16_777_215 { - 1 - } else { - next_id + 1 - }; - let key_pair = libsignal_protocol::KeyPair::generate(&mut csprng); - store - .pre_key_store - .save_pre_key( - next_id.into(), - &libsignal_protocol::PreKeyRecord::new(next_id.into(), &key_pair), - ) - .assert_send() - .await - .map_err(|error| TwonlyError::Signal(error.to_string()))?; - prekeys.push((next_id, key_pair.public_key.serialize().to_vec())); - } - - Ok(prekeys) - } - pub async fn generate_bundle(&self) -> Result { let mut store_guard = self.store.lock().await; let store = &mut *store_guard; diff --git a/rust/src/user_config.rs b/rust/src/user_config.rs index 16da3129..806b811e 100644 --- a/rust/src/user_config.rs +++ b/rust/src/user_config.rs @@ -333,7 +333,7 @@ pub struct UserConfig { impl UserConfig { fn path(context: &Context) -> PathBuf { - PathBuf::from(context.data_dir()) + PathBuf::from(&context.config.data_dir) .join("keyvalue") .join("user.json") } diff --git a/rust/src/user_discovery.rs b/rust/src/user_discovery.rs index 2a266909..67f9ef0d 100644 --- a/rust/src/user_discovery.rs +++ b/rust/src/user_discovery.rs @@ -77,7 +77,7 @@ impl UserDiscovery { /// has authenticated. Cryptographic discovery state remains owned here; /// the API runtime only invokes this lifecycle hook. pub async fn on_connected(&self, ctx: &Arc) -> Result<()> { - let database = ctx.get_app_database().await; + let database = ctx.app_db.read().await.clone(); let announcements = sqlx::query!( r#"SELECT announced_user_id, announced_public_key FROM user_discovery_announced_users WHERE username IS NULL"# @@ -135,6 +135,70 @@ impl UserDiscovery { }) } + pub async fn initialize_from_config(&self, ctx: &Context) -> Result<()> { + let config = crate::user_config::UserConfig::load_required_from(ctx)?; + + if !config.is_user_discovery_enabled { + return Ok(()); + } + + let discovery_config_path = + PathBuf::from(&ctx.config.data_dir).join("user_discovery_config.json"); + + let settings_are_current = std::fs::read_to_string(&discovery_config_path) + .ok() + .and_then(|value| serde_json::from_str::(&value).ok()) + .is_some_and(|value| { + value.get("threshold").and_then(serde_json::Value::as_u64) + == Some(u64::from(config.user_discovery_threshold)) + && value + .get("share_promotion") + .and_then(serde_json::Value::as_bool) + == Some(config.user_discovery_share_promotion) + }); + + if settings_are_current { + let database = ctx.app_db.read().await.clone(); + let has_shares = + sqlx::query_scalar!("SELECT EXISTS(SELECT 1 FROM user_discovery_shares LIMIT 1)") + .fetch_one(&database.pool) + .await? + != 0; + if has_shares { + return Ok(()); + } + } + + let key_manager = ctx.key_manager.lock().await; + let user_id = key_manager.user_id.ok_or_else(|| { + TwonlyError::Generic("cannot initialize user discovery without user ID".into()) + })?; + let identity = key_manager + .signal_identity + .as_ref() + .ok_or(TwonlyError::SignalIdentityNotFound)?; + let identity = IdentityKeyPair::try_from(identity.identity_key_pair_structure.as_slice()) + .map_err(|error| { + TwonlyError::Generic(format!("invalid Signal identity: {error}")) + })?; + let public_key = identity.identity_key().serialize().to_vec(); + drop(key_manager); + + let database = ctx.app_db.read().await.clone(); + let mut transaction = database.pool.begin().await?; + self.initialize_or_update( + config.user_discovery_threshold, + user_id, + public_key, + config.user_discovery_share_promotion, + &mut transaction, + ) + .await?; + transaction.commit().await?; + database.notify_committed(["user_discovery_shares"]); + Ok(()) + } + async fn sign_data(&self, input_data: &[u8]) -> Result> { let key_manager = self.key_manager.lock().await; let identity = key_manager diff --git a/rust/tests/api.rs b/rust/tests/api.rs index 2e6d46aa..d564a900 100644 --- a/rust/tests/api.rs +++ b/rust/tests/api.rs @@ -208,7 +208,7 @@ async fn test_connect_to_dev_server() -> anyhow::Result<()> { // Find the group_id that tester_a just created (it's the only non-direct-chat group) let group_id = { - let database = tester_a.context.get_app_database().await; + let database = tester_a.context.app_db.read().await.clone(); sqlx::query_scalar!( "SELECT group_id FROM groups WHERE is_direct_chat = 0 ORDER BY rowid DESC LIMIT 1" ) @@ -260,7 +260,7 @@ async fn test_connect_to_dev_server() -> anyhow::Result<()> { // tester_a needs tester_b's public key to promote them. We simulate a message from tester_b // so that tester_a can request the missing public key. { - let db_a = tester_a.context.get_app_database().await; + let db_a = tester_a.context.app_db.read().await.clone(); sqlx::query!( "UPDATE group_members SET last_message = CAST(strftime('%s','now') AS INTEGER) WHERE group_id = ? AND contact_id = ?", group_id, @@ -284,7 +284,7 @@ async fn test_connect_to_dev_server() -> anyhow::Result<()> { .fetch_group_state(group_id.clone()) .await?; { - let database = tester_b.context.get_app_database().await; + let database = tester_b.context.app_db.read().await.clone(); let is_admin = sqlx::query_scalar!( "SELECT is_group_admin FROM groups WHERE group_id = ?", group_id @@ -305,7 +305,7 @@ async fn test_connect_to_dev_server() -> anyhow::Result<()> { .fetch_group_state(group_id.clone()) .await?; { - let database = tester_b.context.get_app_database().await; + let database = tester_b.context.app_db.read().await.clone(); let is_admin = sqlx::query_scalar!( "SELECT is_group_admin FROM groups WHERE group_id = ?", group_id @@ -380,7 +380,7 @@ async fn test_connect_to_dev_server() -> anyhow::Result<()> { .await?; let additional_data = { - let database = tester_a.context.get_app_database().await; + let database = tester_a.context.app_db.read().await.clone(); sqlx::query_scalar!( "SELECT additional_message_data FROM messages WHERE message_id = ?", message_id, diff --git a/rust/tests/api/tester.rs b/rust/tests/api/tester.rs index fd9c1044..73a36191 100644 --- a/rust/tests/api/tester.rs +++ b/rust/tests/api/tester.rs @@ -20,7 +20,7 @@ pub(crate) struct Tester { impl Tester { pub async fn set_contact_verified(&self, user_id: i64, verified: bool) -> anyhow::Result<()> { - let database = self.context.get_app_database().await; + let database = self.context.app_db.read().await.clone(); if verified { sqlx::query!( "INSERT INTO key_verifications(contact_id, type) VALUES (?, 'manualTest')", @@ -48,7 +48,7 @@ impl Tester { expected_data: &[u8], ) -> anyhow::Result<()> { for _ in 0..100 { - let database = self.context.get_app_database().await; + let database = self.context.app_db.read().await.clone(); let message = sqlx::query!( "SELECT sender_id, type, additional_message_data FROM messages WHERE message_id = ?", message_id, @@ -75,7 +75,7 @@ impl Tester { verified_by: i64, ) -> anyhow::Result<()> { for _ in 0..100 { - let database = self.context.get_app_database().await; + let database = self.context.app_db.read().await.clone(); let exists = sqlx::query_scalar!( "SELECT EXISTS(SELECT 1 FROM key_verifications WHERE contact_id = ? AND type = 'contactSharedByVerified' AND verified_by = ?)", contact_id, @@ -94,7 +94,7 @@ impl Tester { } pub async fn is_contact_verified(&self, contact_id: i64) -> anyhow::Result { - let database = self.context.get_app_database().await; + let database = self.context.app_db.read().await.clone(); let verified = sqlx::query_scalar!( r#" SELECT EXISTS( @@ -124,7 +124,7 @@ impl Tester { requested: bool, ) -> anyhow::Result<()> { for _ in 0..100 { - let database = self.context.get_app_database().await; + let database = self.context.app_db.read().await.clone(); let state = sqlx::query!( "SELECT accepted, requested FROM contacts WHERE user_id = ?", user_id @@ -149,7 +149,7 @@ impl Tester { expected_username: &str, ) -> anyhow::Result<()> { for _ in 0..100 { - let database = self.context.get_app_database().await; + let database = self.context.app_db.read().await.clone(); let username = sqlx::query_scalar!("SELECT username FROM contacts WHERE user_id = ?", user_id) .fetch_optional(&database.pool) @@ -175,7 +175,7 @@ impl Tester { expected_text: &str, ) -> anyhow::Result<()> { for _ in 0..100 { - let database = self.context.get_app_database().await; + let database = self.context.app_db.read().await.clone(); let message = sqlx::query!( "SELECT sender_id, content, is_deleted_from_sender FROM messages WHERE message_id = ?", message_id @@ -203,7 +203,7 @@ impl Tester { emoji: &str, ) -> anyhow::Result<()> { for _ in 0..100 { - let database = self.context.get_app_database().await; + let database = self.context.app_db.read().await.clone(); let exists = sqlx::query_scalar!( "SELECT EXISTS(SELECT 1 FROM reactions WHERE message_id = ? AND sender_id = ? AND emoji = ?)", message_id, @@ -229,7 +229,7 @@ impl Tester { emoji: &str, ) -> anyhow::Result<()> { for _ in 0..100 { - let database = self.context.get_app_database().await; + let database = self.context.app_db.read().await.clone(); let exists = sqlx::query_scalar!( "SELECT EXISTS(SELECT 1 FROM reactions WHERE message_id = ? AND sender_id = ? AND emoji = ?)", message_id, @@ -250,7 +250,7 @@ impl Tester { pub async fn wait_for_message_deleted(&self, message_id: &str) -> anyhow::Result<()> { for _ in 0..100 { - let database = self.context.get_app_database().await; + let database = self.context.app_db.read().await.clone(); let deleted = sqlx::query_scalar!( "SELECT is_deleted_from_sender FROM messages WHERE message_id = ?", message_id @@ -273,7 +273,7 @@ impl Tester { group_name: &str, ) -> anyhow::Result<()> { for _ in 0..100 { - let database = self.context.get_app_database().await; + let database = self.context.app_db.read().await.clone(); let row = sqlx::query!("SELECT group_name FROM groups WHERE group_id = ?", group_id) .fetch_optional(&database.pool) .await?; @@ -293,7 +293,7 @@ impl Tester { contact_id: i64, ) -> anyhow::Result<()> { for _ in 0..100 { - let database = self.context.get_app_database().await; + let database = self.context.app_db.read().await.clone(); let state = sqlx::query_scalar!( "SELECT member_state FROM group_members WHERE group_id = ? AND contact_id = ?", group_id, @@ -319,7 +319,7 @@ impl Tester { expected_name: &str, ) -> anyhow::Result<()> { for _ in 0..100 { - let database = self.context.get_app_database().await; + let database = self.context.app_db.read().await.clone(); let name = sqlx::query_scalar!("SELECT group_name FROM groups WHERE group_id = ?", group_id) .fetch_optional(&database.pool) @@ -336,7 +336,7 @@ impl Tester { pub async fn wait_for_group_left(&self, group_id: &str) -> anyhow::Result<()> { for _ in 0..100 { - let database = self.context.get_app_database().await; + let database = self.context.app_db.read().await.clone(); let left = sqlx::query_scalar!("SELECT left_group FROM groups WHERE group_id = ?", group_id) .fetch_optional(&database.pool) diff --git a/rust/tests/api/user_discovery.rs b/rust/tests/api/user_discovery.rs index 5d72168b..5b0aecdb 100644 --- a/rust/tests/api/user_discovery.rs +++ b/rust/tests/api/user_discovery.rs @@ -42,7 +42,7 @@ async fn send_trigger(from: &Tester, to: &Tester, label: &str) -> anyhow::Result async fn wait_for_promotion(relay: &Tester, contact_id: i64) -> anyhow::Result<()> { for _ in 0..300 { - let database = relay.context.get_app_database().await; + let database = relay.context.app_db.read().await.clone(); let exists = sqlx::query_scalar!( r#"SELECT EXISTS( SELECT 1 FROM user_discovery_own_promotions @@ -70,7 +70,7 @@ async fn wait_for_discovery( expected_relations: i64, ) -> anyhow::Result<()> { for _ in 0..300 { - let database = observer.context.get_app_database().await; + let database = observer.context.app_db.read().await.clone(); let announced = sqlx::query_scalar!( "SELECT EXISTS(SELECT 1 FROM user_discovery_announced_users WHERE announced_user_id = ?)", discovered_user_id, @@ -103,7 +103,7 @@ async fn user_discovery_reconstructs_an_unknown_user_from_three_contacts() -> an let discoverable = create_tester().await?; for tester in [&observer, &relay_a, &relay_b, &relay_c, &discoverable] { - let database = tester.context.get_app_database().await; + let database = tester.context.app_db.read().await.clone(); let share_count = sqlx::query_scalar!("SELECT COUNT(*) FROM user_discovery_shares") .fetch_one(&database.pool) .await?; @@ -134,7 +134,7 @@ async fn user_discovery_reconstructs_an_unknown_user_from_three_contacts() -> an wait_for_discovery(&observer, discoverable.user_id, 3).await?; - let database = observer.context.get_app_database().await; + let database = observer.context.app_db.read().await.clone(); let direct_contact = sqlx::query_scalar!( "SELECT EXISTS(SELECT 1 FROM contacts WHERE user_id = ?)", discoverable.user_id,