increase test coverage

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?

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<Context>) -> Result<()> {
+pub async fn retransmit_queued_receipts(ctx: &Arc<Context>) -> Result<()> {
     let database = ctx.app_db.read().await.clone();
     let receipt_ids = sqlx::query_scalar!(
         r#"

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;

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<Context>) -> Result<()> {
+pub async fn perform_heartbeat(ctx: &Arc<Context>) -> 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();

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,

rust/src/api/messages/mod.rs

@@ -3,7 +3,7 @@
  *
  */
 
-pub(crate) mod incoming;
+pub mod incoming;
 #[doc(hidden)]
 pub mod outgoing;
 

rust/src/context.rs

@@ -24,8 +24,8 @@ use zeroize::Zeroize;
 static GLOBAL_CONTEXT: OnceCell<Arc<Context>> = OnceCell::const_new();
 
 pub struct Context {
-    pub(crate) config: InitConfig,
-    pub(crate) rust_db: Arc<RwLock<Arc<Database>>>,
+    pub config: InitConfig,
+    pub rust_db: Arc<RwLock<Arc<Database>>>,
     pub app_db: Arc<RwLock<Arc<AppDatabase>>>,
     pub(crate) secure_storage: SecureStorage,
     pub(crate) key_manager: Arc<Mutex<KeyManager>>,

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,

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("<svg height='100' width='100'><circle cx='50' cy='50' r='40'/></svg>".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(())

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<Tester> {
+    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(())
+}

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<Tester> {
+    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(())
+}

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<Tester> {
+    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(())
+}

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<Tester> {
+    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(())
+}

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<Tester> {
+    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(())
+}

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<Tester> {
+    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(())
+}

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<String>,
+        new_display_name: Option<String>,
+        new_avatar_svg: Option<String>,
+    ) -> 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"));