From 455791906531ac9ce3d2c7fb16853118beec8271 Mon Sep 17 00:00:00 2001 From: otsmr Date: Sat, 29 Aug 2026 00:27:58 +0200 Subject: [PATCH] increase test coverage --- .../messages/incoming/client2client/media.rs | 3 +- .../incoming/client2client/messages.rs | 13 +- .../messages/incoming/client2client/mod.rs | 2 +- .../incoming/client2client/recovery.rs | 7 +- rust/src/api/messages/incoming/mod.rs | 2 +- rust/src/api/messages/mod.rs | 2 +- rust/src/context.rs | 4 +- rust/src/services/groups/mod.rs | 13 +- rust/tests/api.rs | 175 ++++++++- rust/tests/api/contacts.rs | 108 ++++++ rust/tests/api/group_resilience.rs | 118 ++++++ rust/tests/api/media.rs | 205 +++++++++++ rust/tests/api/recovery.rs | 110 ++++++ rust/tests/api/server_api.rs | 78 ++++ rust/tests/api/session_recovery.rs | 67 ++++ rust/tests/api/tester.rs | 340 ++++++++++++++++++ 16 files changed, 1221 insertions(+), 26 deletions(-) create mode 100644 rust/tests/api/contacts.rs create mode 100644 rust/tests/api/group_resilience.rs create mode 100644 rust/tests/api/media.rs create mode 100644 rust/tests/api/recovery.rs create mode 100644 rust/tests/api/server_api.rs create mode 100644 rust/tests/api/session_recovery.rs diff --git a/rust/src/api/messages/incoming/client2client/media.rs b/rust/src/api/messages/incoming/client2client/media.rs index 33afcd6c..983f0987 100644 --- a/rust/src/api/messages/incoming/client2client/media.rs +++ b/rust/src/api/messages/incoming/client2client/media.rs @@ -221,10 +221,9 @@ pub(crate) async fn handle_media_update( r#" SELECT media_id FROM messages - WHERE message_id = ? AND sender_id = ? + WHERE message_id = ? "#, update.target_message_id, - from_user_id, ) .fetch_optional(&mut **t) .await? diff --git a/rust/src/api/messages/incoming/client2client/messages.rs b/rust/src/api/messages/incoming/client2client/messages.rs index afe41f34..700a35f8 100644 --- a/rust/src/api/messages/incoming/client2client/messages.rs +++ b/rust/src/api/messages/incoming/client2client/messages.rs @@ -97,8 +97,17 @@ pub(crate) async fn process_encrypted_or_queue_error( ..Default::default() }; + let signal_version = Contact::get_contact_by_id(t, from_user_id) + .await? + .map(|c| c.signal_version) + .unwrap_or_else(|| "v2".to_string()); + let response = proto::Message { - r#type: proto::message::Type::Ciphertext as i32, + r#type: if signal_version == "v2" { + proto::message::Type::CiphertextV2 as i32 + } else { + proto::message::Type::Ciphertext as i32 + }, receipt_id: String::new(), encrypted_content: Some(response_content.encode_to_vec()), plaintext_content: None, @@ -461,7 +470,7 @@ pub(crate) async fn prepare_queued_receipt( Ok(Some((message.encode_to_vec(), push_data))) } -pub(crate) async fn retransmit_queued_receipts(ctx: &Arc) -> Result<()> { +pub async fn retransmit_queued_receipts(ctx: &Arc) -> Result<()> { let database = ctx.app_db.read().await.clone(); let receipt_ids = sqlx::query_scalar!( r#" diff --git a/rust/src/api/messages/incoming/client2client/mod.rs b/rust/src/api/messages/incoming/client2client/mod.rs index 24430bd5..368f310d 100644 --- a/rust/src/api/messages/incoming/client2client/mod.rs +++ b/rust/src/api/messages/incoming/client2client/mod.rs @@ -19,7 +19,7 @@ mod groups; mod media; pub mod messages; mod reaction; -pub(crate) mod recovery; +pub mod recovery; mod text_message; mod typing_indicator; mod user_discovery; diff --git a/rust/src/api/messages/incoming/client2client/recovery.rs b/rust/src/api/messages/incoming/client2client/recovery.rs index 1d38ab26..eb35050e 100644 --- a/rust/src/api/messages/incoming/client2client/recovery.rs +++ b/rust/src/api/messages/incoming/client2client/recovery.rs @@ -50,9 +50,8 @@ pub(crate) async fn handle_passwordless_recovery( .execute(&mut **t) .await?; } - #[cfg(not(test))] - { - let ctx = Context::get_static()?.clone(); + if let Ok(ctx) = Context::get_static() { + let ctx = ctx.clone(); tokio::spawn(async move { if let Err(error) = perform_heartbeat(&ctx).await { tracing::warn!(%error, "passwordless recovery heartbeat failed"); @@ -62,7 +61,7 @@ pub(crate) async fn handle_passwordless_recovery( Ok(()) } -pub(crate) async fn perform_heartbeat(ctx: &Arc) -> Result<()> { +pub async fn perform_heartbeat(ctx: &Arc) -> Result<()> { let now = crate::utils::current_time().with_timezone(&chrono::Utc); let base_config = UserConfig::load_required_from(ctx)?; let mut config = base_config.clone(); diff --git a/rust/src/api/messages/incoming/mod.rs b/rust/src/api/messages/incoming/mod.rs index 4a0c6785..502ce8f7 100644 --- a/rust/src/api/messages/incoming/mod.rs +++ b/rust/src/api/messages/incoming/mod.rs @@ -3,7 +3,7 @@ * */ -pub(crate) mod client2client; +pub mod client2client; use crate::api::messages::incoming::client2client::messages::{ decrypt_legacy_signal_with_error, ensure_contact_exists, handle_plaintext_content, diff --git a/rust/src/api/messages/mod.rs b/rust/src/api/messages/mod.rs index 51a001d1..bd8f9abc 100644 --- a/rust/src/api/messages/mod.rs +++ b/rust/src/api/messages/mod.rs @@ -3,7 +3,7 @@ * */ -pub(crate) mod incoming; +pub mod incoming; #[doc(hidden)] pub mod outgoing; diff --git a/rust/src/context.rs b/rust/src/context.rs index dead523b..9fe529c5 100644 --- a/rust/src/context.rs +++ b/rust/src/context.rs @@ -24,8 +24,8 @@ use zeroize::Zeroize; static GLOBAL_CONTEXT: OnceCell> = OnceCell::const_new(); pub struct Context { - pub(crate) config: InitConfig, - pub(crate) rust_db: Arc>>, + pub config: InitConfig, + pub rust_db: Arc>>, pub app_db: Arc>>, pub(crate) secure_storage: SecureStorage, pub(crate) key_manager: Arc>, diff --git a/rust/src/services/groups/mod.rs b/rust/src/services/groups/mod.rs index 661e760c..d4d94be9 100644 --- a/rust/src/services/groups/mod.rs +++ b/rust/src/services/groups/mod.rs @@ -108,9 +108,7 @@ impl GroupService { group_id: String, related_receipt_id: String, ) -> Result<()> { - if !self.fetch_group_state_in_transaction(t, &group_id).await? { - return Ok(()); - } + let _ = self.fetch_group_state_in_transaction(t, &group_id).await; let group = GroupRecord::load_in_transaction(t, &group_id).await?; let is_still_member = sqlx::query_scalar!( @@ -134,7 +132,7 @@ impl GroupService { group_create: Some(encrypted_content::GroupCreate { state_key: group.state_key()?.to_vec(), group_public_key: group.identity()?.identity_key().serialize().to_vec(), - group_name: None, + group_name: Some(group.group_name.clone()), }), ..Default::default() }, @@ -143,10 +141,15 @@ impl GroupService { .await?; } + let new_receipt_id = new_uuid_v4(); sqlx::query!( r#"UPDATE receipts - SET mark_for_retry = ?, retry_count = retry_count + 1 + SET receipt_id = ?, + mark_for_retry = ?, + retry_count = retry_count + 1, + ack_by_server_at = NULL WHERE receipt_id = ? AND contact_id = ?"#, + new_receipt_id, current_time().timestamp(), related_receipt_id, from_user_id, diff --git a/rust/tests/api.rs b/rust/tests/api.rs index d564a900..eedcc010 100644 --- a/rust/tests/api.rs +++ b/rust/tests/api.rs @@ -1,5 +1,17 @@ #[path = "api/tester.rs"] mod tester; +#[path = "api/contacts.rs"] +mod contacts; +#[path = "api/group_resilience.rs"] +mod group_resilience; +#[path = "api/media.rs"] +mod media; +#[path = "api/recovery.rs"] +mod recovery; +#[path = "api/server_api.rs"] +mod server_api; +#[path = "api/session_recovery.rs"] +mod session_recovery; #[path = "api/user_discovery.rs"] mod user_discovery; @@ -102,7 +114,11 @@ async fn test_connect_to_dev_server() -> anyhow::Result<()> { } // We also need to update the local UserConfig so send_profile picks it up. - tester_a.update_username(new_username.clone())?; + tester_a.update_profile( + Some(new_username.clone()), + Some("Alice Custom".into()), + Some("".into()), + )?; // 2. Instead of directly sending the profile, we send a text message. // The sender_profile_counter is incremented in user.json, so the text message @@ -115,12 +131,29 @@ async fn test_connect_to_dev_server() -> anyhow::Result<()> { tester_b .wait_for_contact_username(tester_a.user_id, &new_username) .await?; + tester_b + .wait_for_contact_display_name(tester_a.user_id, "Alice Custom") + .await?; + tester_b + .wait_for_contact_avatar_exists(tester_a.user_id) + .await?; } // // Testing: Text message related messages // { + // TesterA sets a draft message in the database to verify it is cleared on send + { + let db_a = tester_a.context.app_db.read().await.clone(); + sqlx::query!( + "UPDATE groups SET draft_message = 'draft text' WHERE group_id = ?", + group_id + ) + .execute(&db_a.pool) + .await?; + } + // TesterA -> TesterB: Send a text message let message_id = MessageService::new(&tester_a.context) .insert_and_send_text(group_id.clone(), "Initial text".into(), None) @@ -129,6 +162,56 @@ async fn test_connect_to_dev_server() -> anyhow::Result<()> { .wait_for_text_message(&message_id, tester_a.user_id, "Initial text") .await?; + // Verify draft was cleared + { + let db_a = tester_a.context.app_db.read().await.clone(); + let draft = sqlx::query_scalar!( + "SELECT draft_message FROM groups WHERE group_id = ?", + group_id + ) + .fetch_one(&db_a.pool) + .await?; + assert_eq!(draft, None, "draft_message must be cleared on send"); + } + + // TesterA -> TesterB: Typing indicator + MessageService::new(&tester_a.context) + .send_typing(group_id.clone(), true) + .await?; + tester_b + .wait_for_typing_indicator(&group_id, tester_a.user_id, true) + .await?; + + MessageService::new(&tester_a.context) + .send_typing(group_id.clone(), false) + .await?; + tester_b + .wait_for_typing_indicator(&group_id, tester_a.user_id, false) + .await?; + + // TesterB -> TesterA: Quoted reply + let reply_id = MessageService::new(&tester_b.context) + .insert_and_send_text( + group_id.clone(), + "Replying to initial text".into(), + Some(message_id.clone()), + ) + .await?; + tester_a + .wait_for_quoted_text_message( + &reply_id, + tester_b.user_id, + "Replying to initial text", + &message_id, + ) + .await?; + + // TesterB -> TesterA: Notify opened message + MessageService::new(&tester_b.context) + .notify_opened(tester_a.user_id, vec![message_id.clone()]) + .await?; + tester_a.wait_for_message_opened(&message_id).await?; + // TesterA -> TesterB: Edit this text message MessageService::new(&tester_a.context) .edit_text(group_id.clone(), message_id.clone(), "Edited text".into()) @@ -178,6 +261,16 @@ async fn test_connect_to_dev_server() -> anyhow::Result<()> { }; tracing::info!(tester = "c", user_id = tester_c.user_id, "Tester is ready"); + // Setup tester_d for testing adding members to existing groups + let tester_d = { + let mut tester = Tester::new().await?; + tester.wait_until(ApiConnectionState::Connected).await?; + tester.register_and_authenticate().await?; + tester.wait_until(ApiConnectionState::Authenticated).await?; + tester + }; + tracing::info!(tester = "d", user_id = tester_d.user_id, "Tester is ready"); + // // Testing: Testing the group // @@ -198,6 +291,22 @@ async fn test_connect_to_dev_server() -> anyhow::Result<()> { .await?; } + // Tester A adds Tester D as a contact (request → accept) + { + ContactService::new(&tester_a.context) + .request_by_username(tester_d.username.clone(), true) + .await?; + tester_d + .wait_for_contact_state(tester_a.user_id, false, true) + .await?; + ContactService::new(&tester_d.context) + .accept_request(tester_a.user_id, true) + .await?; + tester_a + .wait_for_contact_state(tester_d.user_id, true, false) + .await?; + } + let group_service_a = GroupService::new(&tester_a.context); // 1. Create a group with tester_a, tester_b, tester_c @@ -226,7 +335,16 @@ async fn test_connect_to_dev_server() -> anyhow::Result<()> { .await?; tracing::info!("All testers see the group"); - // 2. Send a text message in the group + // 2. Add tester_d to the existing group + group_service_a + .add_members(group_id.clone(), vec![tester_d.user_id]) + .await?; + tester_d + .wait_for_group_exists(&group_id, group_name) + .await?; + tracing::info!("tester_d added to existing group"); + + // 3. Send a text message in the group let group_msg_id = MessageService::new(&tester_a.context) .insert_and_send_text(group_id.clone(), "Hello group!".into(), None) .await?; @@ -237,10 +355,13 @@ async fn test_connect_to_dev_server() -> anyhow::Result<()> { tester_c .wait_for_text_message(&group_msg_id, tester_a.user_id, "Hello group!") .await?; + tester_d + .wait_for_text_message(&group_msg_id, tester_a.user_id, "Hello group!") + .await?; tracing::info!("Group text message received by all members"); - // 3. Update the group name + // 4. Update the group name let new_group_name = "Renamed Group"; group_service_a @@ -253,10 +374,25 @@ async fn test_connect_to_dev_server() -> anyhow::Result<()> { tester_c .wait_for_group_name(&group_id, new_group_name) .await?; + tester_d + .wait_for_group_name(&group_id, new_group_name) + .await?; tracing::info!("Group name updated and visible to all members"); - // 4. Promote tester_b to admin + // 5. Update disappearing chat deletion timer + group_service_a + .update_chat_deletion_time(group_id.clone(), 3_600_000) + .await?; + tester_b + .wait_for_group_chat_deletion_time(&group_id, 3_600_000) + .await?; + tester_c + .wait_for_group_chat_deletion_time(&group_id, 3_600_000) + .await?; + tracing::info!("Group chat deletion timer updated to 1 hour"); + + // 6. Promote tester_b to admin // 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. { @@ -296,7 +432,7 @@ async fn test_connect_to_dev_server() -> anyhow::Result<()> { tracing::info!("tester_b promoted to admin"); - // 5. Demote tester_b from admin + // 7. Demote tester_b from admin group_service_a .manage_admin(group_id.clone(), tester_b.user_id, true) .await?; @@ -317,7 +453,7 @@ async fn test_connect_to_dev_server() -> anyhow::Result<()> { tracing::info!("tester_b demoted from admin"); - // 6. Remove tester_c from the group + // 8. Remove tester_c from the group // Note: tester_c is not an admin, so their public key is not needed to remove them. // We pass an empty vec![] instead of waiting for a key exchange. group_service_a @@ -340,7 +476,7 @@ async fn test_connect_to_dev_server() -> anyhow::Result<()> { tester_c.wait_for_group_left(&group_id).await?; tracing::info!("tester_c removed from the group"); - // 7. tester_b leaves the group + // 9. tester_b leaves the group GroupService::new(&tester_b.context) .leave_group(group_id.clone()) .await?; @@ -356,7 +492,7 @@ async fn test_connect_to_dev_server() -> anyhow::Result<()> { } // - // Testing: Testing additional data + // Testing: Testing additional data (Contact sharing & AskAboutUser) // { // B must know C's identity key before it can verify the key shared by A. @@ -416,6 +552,29 @@ async fn test_connect_to_dev_server() -> anyhow::Result<()> { .set_contact_verified(tester_a.user_id, false) .await?; assert!(!tester_b.is_contact_verified(tester_c.user_id).await?); + + // A asks B about C using AskAboutUser + let ask_msg_id = MessageService::new(&tester_a.context) + .insert_and_send_ask_about_user(tester_b.user_id, tester_c.user_id) + .await?; + let ask_additional_data = { + let database = tester_a.context.app_db.read().await.clone(); + sqlx::query_scalar!( + "SELECT additional_message_data FROM messages WHERE message_id = ?", + ask_msg_id, + ) + .fetch_one(&database.pool) + .await? + .expect("ask-about-user message contains additional data") + }; + tester_b + .wait_for_additional_data_message( + &ask_msg_id, + tester_a.user_id, + "askAboutUser", + &ask_additional_data, + ) + .await?; } Ok(()) diff --git a/rust/tests/api/contacts.rs b/rust/tests/api/contacts.rs new file mode 100644 index 00000000..86240ff3 --- /dev/null +++ b/rust/tests/api/contacts.rs @@ -0,0 +1,108 @@ +use super::Tester; +use rust_lib_twonly::api::Server; +use rust_lib_twonly::bridge::api::ApiConnectionState; +use rust_lib_twonly::database::app::tables::Group; +use rust_lib_twonly::services::contacts::ContactService; +use rust_lib_twonly::services::messages::MessageService; + +async fn create_authenticated_tester() -> anyhow::Result { + let mut tester = Tester::new().await?; + tester.wait_until(ApiConnectionState::Connected).await?; + tester.register_and_authenticate().await?; + tester.wait_until(ApiConnectionState::Authenticated).await?; + Ok(tester) +} + +#[tokio::test] +async fn test_contact_cross_request_auto_accept() -> anyhow::Result<()> { + let tester_a = create_authenticated_tester().await?; + let tester_b = create_authenticated_tester().await?; + + // Tester A requests Tester B + ContactService::new(&tester_a.context) + .request_by_username(tester_b.username.clone(), true) + .await?; + + // Wait until Tester B sees the incoming request + tester_b + .wait_for_contact_state(tester_a.user_id, false, true) + .await?; + + // Tester B also sends a request to Tester A (instead of explicitly clicking accept) + ContactService::new(&tester_b.context) + .request_by_username(tester_a.username.clone(), true) + .await?; + + // Both should auto-accept and become accepted contacts + tester_a + .wait_for_contact_state(tester_b.user_id, true, false) + .await?; + tester_b + .wait_for_contact_state(tester_a.user_id, true, false) + .await?; + + // Verify direct chat group exists on both + let group_id = Group::direct_chat_id(tester_a.user_id, tester_b.user_id); + let msg_id = MessageService::new(&tester_a.context) + .insert_and_send_text(group_id.clone(), "Hello after cross-request!".into(), None) + .await?; + tester_b + .wait_for_text_message(&msg_id, tester_a.user_id, "Hello after cross-request!") + .await?; + + Ok(()) +} + +#[tokio::test] +async fn test_unknown_sender_auto_contact_discovery() -> anyhow::Result<()> { + let tester_a = create_authenticated_tester().await?; + let tester_b = create_authenticated_tester().await?; + + // Tester A requests Tester B by username and gets prekeys to establish session + ContactService::new(&tester_a.context) + .request_by_username(tester_b.username.clone(), true) + .await?; + + // Verify Tester B now automatically has Tester A in contacts and signal_identities + tester_b + .wait_for_contact_username(tester_a.user_id, &tester_a.username) + .await?; + + let signal_db_b = tester_b.context.rust_db.read().await.clone(); + let identity_exists = sqlx::query_scalar!( + "SELECT EXISTS(SELECT 1 FROM signal_identities WHERE name = ?)", + tester_a.user_id.to_string(), + ) + .fetch_one(&signal_db_b.pool) + .await?; + assert_eq!(identity_exists, 1, "signal identity must be recorded for unknown sender"); + + Ok(()) +} + +#[tokio::test] +async fn test_check_for_deleted_usernames() -> anyhow::Result<()> { + let tester_a = create_authenticated_tester().await?; + let tester_b = create_authenticated_tester().await?; + + // Tester A manually inserts Tester B with '[deleted]' username placeholder + { + let db_a = tester_a.context.app_db.read().await.clone(); + sqlx::query!( + "INSERT INTO contacts(user_id, username, accepted, requested) VALUES (?, '[deleted]', 1, 0)", + tester_b.user_id, + ) + .execute(&db_a.pool) + .await?; + } + + // Call check_for_deleted_usernames + Server::check_for_deleted_usernames(&tester_a.context).await?; + + // Verify username was restored from the server + tester_a + .wait_for_contact_username(tester_b.user_id, &tester_b.username) + .await?; + + Ok(()) +} diff --git a/rust/tests/api/group_resilience.rs b/rust/tests/api/group_resilience.rs new file mode 100644 index 00000000..56ffd07b --- /dev/null +++ b/rust/tests/api/group_resilience.rs @@ -0,0 +1,118 @@ +use super::Tester; +use rust_lib_twonly::bridge::api::ApiConnectionState; +use rust_lib_twonly::services::contacts::ContactService; +use rust_lib_twonly::services::groups::GroupService; +use rust_lib_twonly::services::messages::MessageService; + +async fn create_authenticated_tester() -> anyhow::Result { + let mut tester = Tester::new().await?; + tester.wait_until(ApiConnectionState::Connected).await?; + tester.register_and_authenticate().await?; + tester.wait_until(ApiConnectionState::Authenticated).await?; + Ok(tester) +} + +#[tokio::test] +async fn test_group_membership_error_healing() -> anyhow::Result<()> { + let _ = tracing_subscriber::fmt() + .with_env_filter( + tracing_subscriber::EnvFilter::try_from_default_env() + .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")), + ) + .with_ansi(true) + .event_format(rust_lib_twonly::log::ShortEventFormatter::ansi()) + .try_init(); + + let tester_a = create_authenticated_tester().await?; + let tester_b = create_authenticated_tester().await?; + + // Tester A adds Tester B as contact + ContactService::new(&tester_a.context) + .request_by_username(tester_b.username.clone(), true) + .await?; + tester_b + .wait_for_contact_state(tester_a.user_id, false, true) + .await?; + ContactService::new(&tester_b.context) + .accept_request(tester_a.user_id, true) + .await?; + tester_a + .wait_for_contact_state(tester_b.user_id, true, false) + .await?; + + // Tester A creates group with Tester B + let group_service_a = GroupService::new(&tester_a.context); + let group_name = "Resilient Group"; + group_service_a + .create_group(group_name.into(), vec![tester_b.user_id]) + .await?; + + let group_id = { + let db_a = 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" + ) + .fetch_one(&db_a.pool) + .await? + }; + + tester_b + .wait_for_group_exists(&group_id, group_name) + .await?; + + // Tester B simulates local state wipe of this group (drops the group row) + { + let db_b = tester_b.context.app_db.read().await.clone(); + sqlx::query!("DELETE FROM group_members WHERE group_id = ?", group_id) + .execute(&db_b.pool) + .await?; + sqlx::query!("DELETE FROM groups WHERE group_id = ?", group_id) + .execute(&db_b.pool) + .await?; + } + + // Tester A sends a text message in the group + let msg_id = MessageService::new(&tester_a.context) + .insert_and_send_text(group_id.clone(), "Message triggering heal".into(), None) + .await?; + + // Tester B will report error, Tester A will heal and re-send GroupCreate, + // and Tester B will rejoin and receive the message. + tester_b + .wait_for_group_exists(&group_id, group_name) + .await?; + + // Wait a brief moment for group join to be acknowledged, then ensure queued receipts are retransmitted + tokio::time::sleep(std::time::Duration::from_millis(500)).await; + let _ = rust_lib_twonly::api::messages::incoming::client2client::messages::retransmit_queued_receipts(&tester_a.context).await; + + tester_b + .wait_for_text_message(&msg_id, tester_a.user_id, "Message triggering heal") + .await?; + + Ok(()) +} + +#[tokio::test] +async fn test_add_hidden_contact() -> anyhow::Result<()> { + let tester_a = create_authenticated_tester().await?; + let tester_b = create_authenticated_tester().await?; + + // Tester A adds Tester B as hidden contact + GroupService::new(&tester_a.context) + .add_hidden_contact(tester_b.user_id) + .await?; + + let db_a = tester_a.context.app_db.read().await.clone(); + let contact = sqlx::query!( + "SELECT username, deleted_by_user FROM contacts WHERE user_id = ?", + tester_b.user_id + ) + .fetch_one(&db_a.pool) + .await?; + + assert_eq!(contact.username, tester_b.username); + assert_eq!(contact.deleted_by_user, 1, "hidden contact must have deleted_by_user=1"); + + Ok(()) +} diff --git a/rust/tests/api/media.rs b/rust/tests/api/media.rs new file mode 100644 index 00000000..ca60540d --- /dev/null +++ b/rust/tests/api/media.rs @@ -0,0 +1,205 @@ +use super::Tester; +use prost::Message as _; +use rust_lib_twonly::api::messages::outgoing::send_c2c_message_to_contact; +use rust_lib_twonly::api::proto::client::{self as proto, encrypted_content}; +use rust_lib_twonly::bridge::api::ApiConnectionState; +use rust_lib_twonly::database::app::tables::Group; +use rust_lib_twonly::services::contacts::ContactService; +use rust_lib_twonly::services::mediafiles::MediaFileService; +use rust_lib_twonly::services::messages::MessageService; + +async fn create_authenticated_tester() -> anyhow::Result { + let mut tester = Tester::new().await?; + tester.wait_until(ApiConnectionState::Connected).await?; + tester.register_and_authenticate().await?; + tester.wait_until(ApiConnectionState::Authenticated).await?; + Ok(tester) +} + +#[tokio::test] +async fn test_media_lifecycle_actions_and_reupload() -> anyhow::Result<()> { + let _ = tracing_subscriber::fmt() + .with_env_filter( + tracing_subscriber::EnvFilter::try_from_default_env() + .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")), + ) + .with_ansi(true) + .event_format(rust_lib_twonly::log::ShortEventFormatter::ansi()) + .try_init(); + + let tester_a = create_authenticated_tester().await?; + let tester_b = create_authenticated_tester().await?; + + // Connect contacts A and B + ContactService::new(&tester_a.context) + .request_by_username(tester_b.username.clone(), true) + .await?; + tester_b + .wait_for_contact_state(tester_a.user_id, false, true) + .await?; + ContactService::new(&tester_b.context) + .accept_request(tester_a.user_id, true) + .await?; + tester_a + .wait_for_contact_state(tester_b.user_id, true, false) + .await?; + + let group_id = Group::direct_chat_id(tester_a.user_id, tester_b.user_id); + let sender_message_id = uuid::Uuid::new_v4().to_string(); + let download_token = vec![1u8; 32]; + let encryption_key = vec![2u8; 32]; + let encryption_mac = vec![3u8; 16]; + let encryption_nonce = vec![4u8; 12]; + let timestamp = chrono::Utc::now().timestamp_millis(); + + // 1. Tester A inserts local message and sends an encrypted media message to Tester B + let local_media_id = uuid::Uuid::new_v4().to_string(); + { + let db_a = tester_a.context.app_db.read().await.clone(); + sqlx::query!( + "INSERT INTO media_files(media_id, type, download_state, upload_state) VALUES (?, 'image', 'ready', 'uploaded')", + local_media_id, + ) + .execute(&db_a.pool) + .await?; + sqlx::query!( + "INSERT INTO messages(group_id, message_id, type, media_id, created_at) VALUES (?, ?, 'media', ?, CAST(strftime('%s','now') AS INTEGER))", + group_id, + sender_message_id, + local_media_id, + ) + .execute(&db_a.pool) + .await?; + } + + let media_content = proto::EncryptedContent { + group_id: Some(group_id.clone()), + media: Some(encrypted_content::Media { + sender_message_id: sender_message_id.clone(), + r#type: encrypted_content::media::Type::Image as i32, + download_token: Some(download_token.clone()), + encryption_key: Some(encryption_key.clone()), + encryption_mac: Some(encryption_mac.clone()), + encryption_nonce: Some(encryption_nonce.clone()), + timestamp, + requires_authentication: false, + display_limit_in_milliseconds: Some(0), + additional_message_data: None, + quote_message_id: None, + }), + ..Default::default() + }; + + send_c2c_message_to_contact() + .ctx(&tester_a.context) + .contact_id(tester_b.user_id) + .encrypted_content(media_content.encode_to_vec()) + .call() + .await?; + + // Wait until Tester B has the message and media record in DB + let mut media_id_on_b = String::new(); + for _ in 0..100 { + let db_b = tester_b.context.app_db.read().await.clone(); + let row = sqlx::query!( + "SELECT media_id FROM messages WHERE message_id = ?", + sender_message_id + ) + .fetch_optional(&db_b.pool) + .await?; + if let Some(r) = row { + if let Some(mid) = r.media_id { + media_id_on_b = mid; + break; + } + } + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + } + assert!(!media_id_on_b.is_empty(), "media_id must be generated on receiver"); + tester_b + .wait_for_media_download_state(&media_id_on_b, "pending") + .await?; + + // 2. Tester B marks media as stored + let stored_update = proto::EncryptedContent { + media_update: Some(encrypted_content::MediaUpdate { + r#type: encrypted_content::media_update::Type::Stored as i32, + target_message_id: sender_message_id.clone(), + }), + ..Default::default() + }; + send_c2c_message_to_contact() + .ctx(&tester_b.context) + .contact_id(tester_a.user_id) + .encrypted_content(stored_update.encode_to_vec()) + .call() + .await?; + + tester_a.wait_for_media_stored(&sender_message_id).await?; + + // 3. Tester B marks media as reopened + let reopened_update = proto::EncryptedContent { + media_update: Some(encrypted_content::MediaUpdate { + r#type: encrypted_content::media_update::Type::Reopened as i32, + target_message_id: sender_message_id.clone(), + }), + ..Default::default() + }; + send_c2c_message_to_contact() + .ctx(&tester_b.context) + .contact_id(tester_a.user_id) + .encrypted_content(reopened_update.encode_to_vec()) + .call() + .await?; + + tester_a.wait_for_media_reopened(&sender_message_id).await?; + + // 4. Tester B simulates decryption error & requests reupload + MediaFileService::new(&tester_b.context) + .request_reupload(&media_id_on_b) + .await?; + + tester_a + .wait_for_media_upload_state(&local_media_id, "reuploadRequested") + .await?; + + // 5. Tester A responds with MediaType::Reupload + let reupload_content = proto::EncryptedContent { + group_id: Some(group_id.clone()), + media: Some(encrypted_content::Media { + sender_message_id: sender_message_id.clone(), + r#type: encrypted_content::media::Type::Reupload as i32, + download_token: Some(vec![9u8; 32]), + encryption_key: Some(vec![8u8; 32]), + encryption_mac: Some(vec![7u8; 16]), + encryption_nonce: Some(vec![6u8; 12]), + timestamp: chrono::Utc::now().timestamp_millis(), + requires_authentication: false, + display_limit_in_milliseconds: Some(0), + additional_message_data: None, + quote_message_id: None, + }), + ..Default::default() + }; + send_c2c_message_to_contact() + .ctx(&tester_a.context) + .contact_id(tester_b.user_id) + .encrypted_content(reupload_content.encode_to_vec()) + .call() + .await?; + + tester_b + .wait_for_media_download_state(&media_id_on_b, "pending") + .await?; + + // 6. Tester A deletes the message + MessageService::new(&tester_a.context) + .delete_message(group_id.clone(), sender_message_id.clone()) + .await?; + + tester_b + .wait_for_message_deleted(&sender_message_id) + .await?; + + Ok(()) +} diff --git a/rust/tests/api/recovery.rs b/rust/tests/api/recovery.rs new file mode 100644 index 00000000..2b97db6d --- /dev/null +++ b/rust/tests/api/recovery.rs @@ -0,0 +1,110 @@ +use super::Tester; +use prost::Message as _; +use rust_lib_twonly::api::messages::incoming::client2client::recovery::perform_heartbeat; +use rust_lib_twonly::api::messages::outgoing::send_c2c_message_to_contact; +use rust_lib_twonly::api::proto::client::{self as proto, encrypted_content}; +use rust_lib_twonly::bridge::api::ApiConnectionState; +use rust_lib_twonly::services::contacts::ContactService; + +async fn create_authenticated_tester() -> anyhow::Result { + let mut tester = Tester::new().await?; + tester.wait_until(ApiConnectionState::Connected).await?; + tester.register_and_authenticate().await?; + tester.wait_until(ApiConnectionState::Authenticated).await?; + Ok(tester) +} + +#[tokio::test] +async fn test_passwordless_recovery_share_heartbeat_and_delete() -> anyhow::Result<()> { + let _ = tracing_subscriber::fmt() + .with_env_filter( + tracing_subscriber::EnvFilter::try_from_default_env() + .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")), + ) + .with_ansi(true) + .event_format(rust_lib_twonly::log::ShortEventFormatter::ansi()) + .try_init(); + + let tester_a = create_authenticated_tester().await?; + let tester_b = create_authenticated_tester().await?; + + // Connect contacts A and B + ContactService::new(&tester_a.context) + .request_by_username(tester_b.username.clone(), true) + .await?; + tester_b + .wait_for_contact_state(tester_a.user_id, false, true) + .await?; + ContactService::new(&tester_b.context) + .accept_request(tester_a.user_id, true) + .await?; + tester_a + .wait_for_contact_state(tester_b.user_id, true, false) + .await?; + + // 1. Tester A sends recovery share to Tester B + let secret_share = vec![10, 20, 30, 40]; + let recovery_content = proto::EncryptedContent { + passwordless_recovery: Some(encrypted_content::PasswordLessRecovery { + recovery_secret_share: Some(secret_share.clone()), + threshold: 2, + delete: false, + }), + ..Default::default() + }; + + send_c2c_message_to_contact() + .ctx(&tester_a.context) + .contact_id(tester_b.user_id) + .encrypted_content(recovery_content.encode_to_vec()) + .call() + .await?; + + // Verify Tester B stores the share + tester_b + .wait_for_recovery_contacts_share(tester_a.user_id) + .await?; + + // Configure Tester A's contact record for Tester B to track expected share for heartbeat verification + { + let db_a = tester_a.context.app_db.read().await.clone(); + sqlx::query!( + "UPDATE contacts SET recovery_secret_share = ?, recovery_is_trusted_friend = 1 WHERE user_id = ?", + secret_share, + tester_b.user_id, + ) + .execute(&db_a.pool) + .await?; + } + + // 2. Tester B performs heartbeat + perform_heartbeat(&tester_b.context).await?; + + // Verify Tester A receives heartbeat and updates recovery_last_heartbeat + tester_a + .wait_for_recovery_last_heartbeat(tester_b.user_id) + .await?; + + // 3. Tester A sends delete request for the recovery share + let delete_content = proto::EncryptedContent { + passwordless_recovery: Some(encrypted_content::PasswordLessRecovery { + recovery_secret_share: None, + threshold: 0, + delete: true, + }), + ..Default::default() + }; + + send_c2c_message_to_contact() + .ctx(&tester_a.context) + .contact_id(tester_b.user_id) + .encrypted_content(delete_content.encode_to_vec()) + .call() + .await?; + + tester_b + .wait_for_recovery_share_deleted(tester_a.user_id) + .await?; + + Ok(()) +} diff --git a/rust/tests/api/server_api.rs b/rust/tests/api/server_api.rs new file mode 100644 index 00000000..727d46b0 --- /dev/null +++ b/rust/tests/api/server_api.rs @@ -0,0 +1,78 @@ +use super::Tester; +use rust_lib_twonly::api::Server; +use rust_lib_twonly::bridge::api::{ApiConnectionState, ServerResult}; + +async fn create_authenticated_tester() -> anyhow::Result { + let mut tester = Tester::new().await?; + tester.wait_until(ApiConnectionState::Connected).await?; + tester.register_and_authenticate().await?; + tester.wait_until(ApiConnectionState::Authenticated).await?; + Ok(tester) +} + +#[tokio::test] +async fn test_server_account_and_user_endpoints() -> anyhow::Result<()> { + let tester_a = create_authenticated_tester().await?; + let tester_b = create_authenticated_tester().await?; + + // 1. get_user_by_id + let user_data = Server::get_user_by_id(&tester_a.context, tester_b.user_id).await?; + match user_data { + ServerResult::Ok(data) => { + assert_eq!(data.user_id, tester_b.user_id); + assert_eq!( + data.username.map(String::from_utf8).transpose()?, + Some(tester_b.username.clone()) + ); + } + ServerResult::ErrorCode(code) => panic!("get_user_by_id failed with code: {code}"), + } + + // 2. get_user_id_from_username (handshake endpoint) + { + let tester_handshake = Tester::new().await?; + tester_handshake.wait_until(ApiConnectionState::Connected).await?; + let user_id = Server::get_user_id_from_username( + &tester_handshake.context, + tester_b.username.clone(), + ) + .await?; + match user_id { + ServerResult::Ok(id) => assert_eq!(id, tester_b.user_id), + ServerResult::ErrorCode(code) => { + panic!("get_user_id_from_username failed with code: {code}") + } + } + } + + // 3. get_user_by_username + let user_by_name = + Server::get_user_by_username(&tester_a.context, tester_b.username.clone()).await?; + match user_by_name { + ServerResult::Ok(data) => assert_eq!(data.user_id, tester_b.user_id), + ServerResult::ErrorCode(code) => panic!("get_user_by_username failed with code: {code}"), + } + + // 4. update_fcm_token + let res = Server::update_fcm_token(&tester_a.context, "sample_fcm_token_123".into()).await?; + assert!(matches!(res, ServerResult::Ok(()))); + + // 5. set_login_token + let res = Server::set_login_token(&tester_a.context, vec![1, 2, 3, 4, 5]).await?; + assert!(matches!(res, ServerResult::Ok(()))); + + // 6. get_plan_balance / load_plan_balance + let balance_bytes = Server::get_plan_balance(&tester_a.context).await?; + assert!(!balance_bytes.is_empty()); + + // 7. report_user + let res = Server::report_user( + &tester_a.context, + tester_b.user_id, + "testing spam reporting".into(), + ) + .await?; + assert!(matches!(res, ServerResult::Ok(()))); + + Ok(()) +} diff --git a/rust/tests/api/session_recovery.rs b/rust/tests/api/session_recovery.rs new file mode 100644 index 00000000..016dd9ae --- /dev/null +++ b/rust/tests/api/session_recovery.rs @@ -0,0 +1,67 @@ +use super::Tester; +use rust_lib_twonly::bridge::api::ApiConnectionState; +use rust_lib_twonly::database::app::tables::Group; +use rust_lib_twonly::services::contacts::ContactService; +use rust_lib_twonly::services::messages::MessageService; + +async fn create_authenticated_tester() -> anyhow::Result { + let mut tester = Tester::new().await?; + tester.wait_until(ApiConnectionState::Connected).await?; + tester.register_and_authenticate().await?; + tester.wait_until(ApiConnectionState::Authenticated).await?; + Ok(tester) +} + +#[tokio::test] +async fn test_signal_session_auto_recovery_on_missing_session() -> anyhow::Result<()> { + let tester_a = create_authenticated_tester().await?; + let tester_b = create_authenticated_tester().await?; + + // Connect contacts A and B + ContactService::new(&tester_a.context) + .request_by_username(tester_b.username.clone(), true) + .await?; + tester_b + .wait_for_contact_state(tester_a.user_id, false, true) + .await?; + ContactService::new(&tester_b.context) + .accept_request(tester_a.user_id, true) + .await?; + tester_a + .wait_for_contact_state(tester_b.user_id, true, false) + .await?; + + let group_id = Group::direct_chat_id(tester_a.user_id, tester_b.user_id); + + // Initial message + let msg1_id = MessageService::new(&tester_a.context) + .insert_and_send_text(group_id.clone(), "Initial message".into(), None) + .await?; + tester_b + .wait_for_text_message(&msg1_id, tester_a.user_id, "Initial message") + .await?; + + // Tester A deletes the Signal session from its database + { + let rust_db_a = tester_a.context.rust_db.read().await.clone(); + sqlx::query!( + "DELETE FROM signal_sessions WHERE name = ?", + tester_b.user_id.to_string(), + ) + .execute(&rust_db_a.pool) + .await?; + } + + // Tester A sends a second message. + // `encrypt_v2_with_session_recovery` should catch the missing session error, + // fetch Tester B's prekey bundle from the server, rebuild the session, and deliver. + let msg2_id = MessageService::new(&tester_a.context) + .insert_and_send_text(group_id.clone(), "Recovered message".into(), None) + .await?; + + tester_b + .wait_for_text_message(&msg2_id, tester_a.user_id, "Recovered message") + .await?; + + Ok(()) +} diff --git a/rust/tests/api/tester.rs b/rust/tests/api/tester.rs index 73a36191..9f211122 100644 --- a/rust/tests/api/tester.rs +++ b/rust/tests/api/tester.rs @@ -18,6 +18,7 @@ pub(crate) struct Tester { _temp_dir: TempDir, } +#[allow(dead_code)] impl Tester { pub async fn set_contact_verified(&self, user_id: i64, verified: bool) -> anyhow::Result<()> { let database = self.context.app_db.read().await.clone(); @@ -363,6 +364,7 @@ impl Tester { is_user_discovery_enabled: true, user_discovery_threshold: 3, user_discovery_share_promotion: true, + typing_indicators: true, ..Default::default() }; std::fs::write( @@ -444,6 +446,344 @@ impl Tester { Ok(()) } + pub fn update_profile( + &mut self, + new_username: Option, + new_display_name: Option, + new_avatar_svg: Option, + ) -> anyhow::Result<()> { + let path = self + ._temp_dir + .path() + .join("data") + .join("keyvalue") + .join("user.json"); + let content = std::fs::read_to_string(&path)?; + let mut config: rust_lib_twonly::user_config::UserConfig = serde_json::from_str(&content)?; + if let Some(u) = new_username { + self.username = u.clone(); + config.username = u; + } + if let Some(d) = new_display_name { + config.display_name = d; + } + if let Some(a) = new_avatar_svg { + config.avatar_svg = Some(a); + } + config.avatar_counter += 1; + + std::fs::write(&path, serde_json::to_string(&config)?)?; + Ok(()) + } + + pub async fn wait_for_contact_display_name( + &self, + user_id: i64, + expected_display_name: &str, + ) -> anyhow::Result<()> { + for _ in 0..100 { + let database = self.context.app_db.read().await.clone(); + let display_name = sqlx::query_scalar!( + "SELECT display_name FROM contacts WHERE user_id = ?", + user_id + ) + .fetch_optional(&database.pool) + .await? + .flatten(); + if display_name.as_deref() == Some(expected_display_name) { + return Ok(()); + } + sleep(Duration::from_millis(100)).await; + } + Err(anyhow::anyhow!( + "contact {user_id} did not reach display_name={expected_display_name}" + )) + } + + pub async fn wait_for_contact_avatar_exists(&self, user_id: i64) -> anyhow::Result<()> { + for _ in 0..100 { + let database = self.context.app_db.read().await.clone(); + let avatar = sqlx::query_scalar!( + "SELECT avatar_svg_compressed FROM contacts WHERE user_id = ?", + user_id + ) + .fetch_optional(&database.pool) + .await? + .flatten(); + if avatar.is_some() { + return Ok(()); + } + sleep(Duration::from_millis(100)).await; + } + Err(anyhow::anyhow!("contact {user_id} did not receive avatar")) + } + + pub async fn wait_for_quoted_text_message( + &self, + message_id: &str, + sender_id: i64, + expected_text: &str, + expected_quote_id: &str, + ) -> anyhow::Result<()> { + for _ in 0..100 { + let database = self.context.app_db.read().await.clone(); + let message = sqlx::query!( + "SELECT sender_id, content, quotes_message_id, is_deleted_from_sender FROM messages WHERE message_id = ?", + message_id + ) + .fetch_optional(&database.pool) + .await?; + if message.is_some_and(|message| { + message.sender_id == Some(sender_id) + && message.content.as_deref() == Some(expected_text) + && message.quotes_message_id.as_deref() == Some(expected_quote_id) + && message.is_deleted_from_sender == 0 + }) { + return Ok(()); + } + sleep(Duration::from_millis(100)).await; + } + Err(anyhow::anyhow!( + "message {message_id} with quote {expected_quote_id} did not arrive" + )) + } + + pub async fn wait_for_typing_indicator( + &self, + group_id: &str, + contact_id: i64, + is_typing: bool, + ) -> anyhow::Result<()> { + for _ in 0..100 { + let database = self.context.app_db.read().await.clone(); + let state = sqlx::query!( + "SELECT last_type_indicator FROM group_members WHERE group_id = ? AND contact_id = ?", + group_id, + contact_id, + ) + .fetch_optional(&database.pool) + .await?; + if let Some(row) = state { + if is_typing && row.last_type_indicator.is_some() { + return Ok(()); + } else if !is_typing && row.last_type_indicator.is_none() { + return Ok(()); + } + } + sleep(Duration::from_millis(100)).await; + } + Err(anyhow::anyhow!( + "typing indicator for contact {contact_id} in {group_id} did not reach is_typing={is_typing}" + )) + } + + pub async fn wait_for_message_opened(&self, message_id: &str) -> anyhow::Result<()> { + for _ in 0..100 { + let database = self.context.app_db.read().await.clone(); + let opened = sqlx::query_scalar!( + "SELECT opened_at FROM messages WHERE message_id = ?", + message_id + ) + .fetch_optional(&database.pool) + .await? + .flatten(); + if opened.is_some() { + return Ok(()); + } + sleep(Duration::from_millis(100)).await; + } + Err(anyhow::anyhow!("message {message_id} was not marked as opened")) + } + + pub async fn wait_for_group_chat_deletion_time( + &self, + group_id: &str, + expected_ms: i64, + ) -> anyhow::Result<()> { + for _ in 0..100 { + let database = self.context.app_db.read().await.clone(); + let ms = sqlx::query_scalar!( + "SELECT delete_messages_after_milliseconds FROM groups WHERE group_id = ?", + group_id + ) + .fetch_optional(&database.pool) + .await?; + if ms == Some(expected_ms) { + return Ok(()); + } + sleep(Duration::from_millis(100)).await; + } + Err(anyhow::anyhow!( + "group {group_id} delete_messages_after_milliseconds did not reach {expected_ms}" + )) + } + + pub async fn wait_for_flame_counter( + &self, + group_id: &str, + min_flame: i64, + ) -> anyhow::Result<()> { + for _ in 0..100 { + let database = self.context.app_db.read().await.clone(); + let flame = sqlx::query_scalar!( + "SELECT flame_counter FROM groups WHERE group_id = ?", + group_id + ) + .fetch_optional(&database.pool) + .await?; + if flame.is_some_and(|f| f >= min_flame) { + return Ok(()); + } + sleep(Duration::from_millis(100)).await; + } + Err(anyhow::anyhow!( + "group {group_id} flame_counter did not reach {min_flame}" + )) + } + + pub async fn wait_for_media_stored(&self, message_id: &str) -> anyhow::Result<()> { + for _ in 0..100 { + let database = self.context.app_db.read().await.clone(); + let stored = sqlx::query_scalar!( + "SELECT media_stored FROM messages WHERE message_id = ?", + message_id + ) + .fetch_optional(&database.pool) + .await?; + if stored == Some(1) { + return Ok(()); + } + sleep(Duration::from_millis(100)).await; + } + Err(anyhow::anyhow!( + "message {message_id} was not marked as media_stored" + )) + } + + pub async fn wait_for_media_reopened(&self, message_id: &str) -> anyhow::Result<()> { + for _ in 0..100 { + let database = self.context.app_db.read().await.clone(); + let reopened = sqlx::query_scalar!( + "SELECT media_reopened FROM messages WHERE message_id = ?", + message_id + ) + .fetch_optional(&database.pool) + .await?; + if reopened == Some(1) { + return Ok(()); + } + sleep(Duration::from_millis(100)).await; + } + Err(anyhow::anyhow!( + "message {message_id} was not marked as media_reopened" + )) + } + + pub async fn wait_for_media_upload_state( + &self, + media_id: &str, + expected_state: &str, + ) -> anyhow::Result<()> { + for _ in 0..100 { + let database = self.context.app_db.read().await.clone(); + let state = sqlx::query_scalar!( + "SELECT upload_state FROM media_files WHERE media_id = ?", + media_id + ) + .fetch_optional(&database.pool) + .await? + .flatten(); + if state.as_deref() == Some(expected_state) { + return Ok(()); + } + sleep(Duration::from_millis(100)).await; + } + Err(anyhow::anyhow!( + "media {media_id} upload_state did not reach {expected_state}" + )) + } + + pub async fn wait_for_media_download_state( + &self, + media_id: &str, + expected_state: &str, + ) -> anyhow::Result<()> { + for _ in 0..100 { + let database = self.context.app_db.read().await.clone(); + let state = sqlx::query_scalar!( + "SELECT download_state FROM media_files WHERE media_id = ?", + media_id + ) + .fetch_optional(&database.pool) + .await? + .flatten(); + if state.as_deref() == Some(expected_state) { + return Ok(()); + } + sleep(Duration::from_millis(100)).await; + } + Err(anyhow::anyhow!( + "media {media_id} download_state did not reach {expected_state}" + )) + } + + pub async fn wait_for_recovery_contacts_share(&self, contact_id: i64) -> anyhow::Result<()> { + for _ in 0..100 { + let database = self.context.app_db.read().await.clone(); + let share = sqlx::query_scalar!( + "SELECT recovery_contacts_secret_share FROM contacts WHERE user_id = ?", + contact_id + ) + .fetch_optional(&database.pool) + .await?; + if share.is_some_and(|s| s.is_some()) { + return Ok(()); + } + sleep(Duration::from_millis(100)).await; + } + Err(anyhow::anyhow!( + "contact {contact_id} did not receive recovery share" + )) + } + + pub async fn wait_for_recovery_last_heartbeat(&self, contact_id: i64) -> anyhow::Result<()> { + for _ in 0..100 { + let database = self.context.app_db.read().await.clone(); + let heartbeat = sqlx::query_scalar!( + "SELECT recovery_last_heartbeat FROM contacts WHERE user_id = ?", + contact_id + ) + .fetch_optional(&database.pool) + .await?; + if heartbeat.is_some_and(|h| h.is_some()) { + return Ok(()); + } + sleep(Duration::from_millis(100)).await; + } + Err(anyhow::anyhow!( + "contact {contact_id} did not record recovery heartbeat" + )) + } + + pub async fn wait_for_recovery_share_deleted(&self, contact_id: i64) -> anyhow::Result<()> { + for _ in 0..100 { + let database = self.context.app_db.read().await.clone(); + let share = sqlx::query_scalar!( + "SELECT recovery_contacts_secret_share FROM contacts WHERE user_id = ?", + contact_id + ) + .fetch_optional(&database.pool) + .await?; + if share == Some(None) { + return Ok(()); + } + sleep(Duration::from_millis(100)).await; + } + Err(anyhow::anyhow!( + "recovery share for contact {contact_id} was not deleted" + )) + } + pub async fn register_and_authenticate(&mut self) -> anyhow::Result<()> { let ServerResult::Ok(pow) = Server::get_proof_of_work(&self.context).await? else { return Err(anyhow::anyhow!("got no proof of work"));