diff --git a/src/retry.rs b/src/retry.rs index f2c70416f..bc3af41bf 100644 --- a/src/retry.rs +++ b/src/retry.rs @@ -15,8 +15,8 @@ use wacore::libsignal::store::PreKeyStore; use wacore::protocol::ProtocolNode; use wacore::types::jid::JidExt; use wacore_binary::JidExt as _; -use wacore_binary::OwnedNodeRef; use wacore_binary::builder::NodeBuilder; +use wacore_binary::{Jid, OwnedNodeRef}; #[cfg(test)] use wacore_binary::{Node, NodeContent}; use wacore_binary::{NodeContentRef, NodeRef}; @@ -82,6 +82,16 @@ const MAX_RETRY_COUNT: u8 = 5; /// WhatsApp Web saves base key on retry 2, checks on retry > 2. const MIN_RETRY_FOR_BASE_KEY_CHECK: u8 = 2; +fn build_retry_dedupe_key(chat: &Jid, message_id: &str, participant_jid: &Jid) -> String { + let mut key = String::with_capacity(message_id.len() + 64); + chat.push_to(&mut key); + key.push(':'); + key.push_str(message_id); + key.push(':'); + participant_jid.push_to(&mut key); + key +} + impl Client { pub(crate) async fn handle_retry_receipt( self: &Arc, @@ -128,21 +138,10 @@ impl Client { receipt.source.sender.clone() }; - // Deduplicate retry receipts to prevent processing the same retry multiple times. - // For groups/status: key includes participant since each device retries independently. - // For DMs: key is (chat, msg_id) since there's only one sender. - // Uses atomic entry API to avoid race conditions between check and insert. - let dedupe_key = { - let mut key = String::with_capacity(64); - receipt.source.chat.push_to(&mut key); - key.push(':'); - key.push_str(&message_id); - if is_group_or_status { - key.push(':'); - participant_jid.push_to(&mut key); - } - key - }; + // Deduplicate retry receipts per requesting device. + // DMs can also receive retries from multiple companion devices. + let dedupe_key = + build_retry_dedupe_key(&receipt.source.chat, &message_id, &participant_jid); if self.retried_group_messages.get(&dedupe_key).await.is_some() { log::debug!( @@ -188,13 +187,10 @@ impl Client { } }; - // Re-add for groups/status so other participants can also retry. - // take_recent_message consumed it; without this a second participant's - // retry would silently fail with "not found in cache". - if is_group_or_status { - self.add_recent_message(&receipt.source.chat, &message_id, &original_msg) - .await; - } + // take_recent_message consumes the cached message; re-add it so other + // devices for the same chat can still request a retry. + self.add_recent_message(&receipt.source.chat, &message_id, &original_msg) + .await; // Resolved JID for session operations; keep original for stanza addressing let resolved_jid = self.resolve_encryption_jid(&participant_jid).await; @@ -387,86 +383,8 @@ impl Client { ); } } else { - // For DMs, handle base key tracking for collision detection (matches WhatsApp Web). - // This detects when we haven't regenerated our session despite receiving retry receipts, - // which can cause infinite retry loops where both sides are stuck with stale keys. - let signal_address = resolved_jid.to_protocol_address(); - let device_store = self.persistence_manager.get_device_arc().await; - - // Check for base key collision before deleting the session. - // Read session through cache for consistent state. - { - let device_guard = device_store.read().await; - let session = self - .signal_cache - .peek_session(&signal_address, &*device_guard.backend) - .await - .ok() - .flatten(); - - if let Some(session) = session - && let Ok(current_base_key) = session.alice_base_key() - { - let addr_str = signal_address.as_str(); - if retry_count == MIN_RETRY_FOR_BASE_KEY_CHECK { - // On retry 2: Save the base key for later comparison - if let Err(e) = device_guard - .backend - .save_base_key(addr_str, &message_id, current_base_key) - .await - { - warn!("Failed to save base key for {}: {}", signal_address, e); - } else { - info!( - "Saved base key for {} at retry #{} for collision detection", - signal_address, retry_count - ); - } - } else if retry_count > MIN_RETRY_FOR_BASE_KEY_CHECK { - // On retry > 2: Check if base key is the same (collision detection) - match device_guard - .backend - .has_same_base_key(addr_str, &message_id, current_base_key) - .await - { - Ok(true) => { - // Collision detected! We haven't regenerated our session. - warn!( - "Base key collision detected for {} at retry #{}. \ - Session hasn't been regenerated. Forcing fresh session.", - signal_address, retry_count - ); - // Clean up base key entry since we're deleting the session - let _ = device_guard - .backend - .delete_base_key(addr_str, &message_id) - .await; - } - Ok(false) => { - // Base key changed, session was regenerated - good! - info!( - "Base key changed for {} at retry #{} - session regenerated", - signal_address, retry_count - ); - // Clean up old base key entry - let _ = device_guard - .backend - .delete_base_key(addr_str, &message_id) - .await; - } - Err(e) => { - warn!("Failed to check base key for {}: {}", signal_address, e); - } - } - } - } - } - - // Delete the old session through the signal cache so encryption uses a fresh session. - // IMPORTANT: Must go through cache, not backend, to avoid stale cached sessions. - self.signal_cache.delete_session(&signal_address).await; - self.flush_signal_cache().await?; - info!("Deleted session for {signal_address} due to retry receipt"); + self.delete_dm_retry_session_target(&resolved_jid, &message_id, retry_count) + .await?; } // Status broadcasts can't resend (requires explicit recipient list). @@ -495,6 +413,9 @@ impl Client { .map(|g| g.addressing_mode) .unwrap_or_default(); + let signal_address = resolved_jid.to_protocol_address(); + let session_mutex = self.session_lock_for(signal_address.as_str()).await; + let _session_guard = session_mutex.lock().await; let mut store_adapter = self.signal_adapter().await; let stanza = wacore::send::prepare_group_retry_stanza( @@ -514,22 +435,123 @@ impl Client { self.send_node(stanza).await?; self.flush_signal_cache().await?; } else { - // DM retry: re-encrypt via normal send path (already targets single recipient) - self.send_message_impl( + // DM retry: pairwise resend to the requesting device only. + self.ensure_e2e_sessions(std::slice::from_ref(&resolved_jid)) + .await?; + + let device_snapshot = self.persistence_manager.get_device_snapshot().await; + let signal_address = resolved_jid.to_protocol_address(); + let session_mutex = self.session_lock_for(signal_address.as_str()).await; + let _session_guard = session_mutex.lock().await; + let mut store_adapter = self.signal_adapter().await; + + let stanza = wacore::send::prepare_dm_retry_stanza( + &mut store_adapter.session_store, + &mut store_adapter.identity_store, receipt.source.chat.clone(), + participant_jid, + resolved_jid.clone(), &original_msg, - Some(message_id), - false, - true, - None, - vec![], + message_id, + retry_count, + device_snapshot.account.as_ref(), ) .await?; + + self.send_node(stanza).await?; + self.flush_signal_cache().await?; } Ok(()) } + async fn delete_dm_retry_session_target( + &self, + device_jid: &Jid, + message_id: &str, + retry_count: u8, + ) -> Result<(), anyhow::Error> { + // Base key collision detection prevents stale-session retry loops. + let signal_address = device_jid.to_protocol_address(); + let device_store = self.persistence_manager.get_device_arc().await; + + // Check for base key collision before deleting the session. + // Read session through cache for consistent state. + { + let device_guard = device_store.read().await; + let session = self + .signal_cache + .peek_session(&signal_address, &*device_guard.backend) + .await + .ok() + .flatten(); + + if let Some(session) = session + && let Ok(current_base_key) = session.alice_base_key() + { + let addr_str = signal_address.as_str(); + if retry_count == MIN_RETRY_FOR_BASE_KEY_CHECK { + // Save retry #2's base key so later retries can prove regeneration happened. + if let Err(e) = device_guard + .backend + .save_base_key(addr_str, message_id, current_base_key) + .await + { + warn!("Failed to save base key for {}: {}", signal_address, e); + } else { + info!( + "Saved base key for {} at retry #{} for collision detection", + signal_address, retry_count + ); + } + } else if retry_count > MIN_RETRY_FOR_BASE_KEY_CHECK { + // An unchanged base key means we never rebuilt the session. + match device_guard + .backend + .has_same_base_key(addr_str, message_id, current_base_key) + .await + { + Ok(true) => { + // Collision detected! We haven't regenerated our session. + warn!( + "Base key collision detected for {} at retry #{}. \ + Session hasn't been regenerated. Forcing fresh session.", + signal_address, retry_count + ); + // Clean up base key entry since we're deleting the session + let _ = device_guard + .backend + .delete_base_key(addr_str, message_id) + .await; + } + Ok(false) => { + // Base key changed, session was regenerated - good! + info!( + "Base key changed for {} at retry #{} - session regenerated", + signal_address, retry_count + ); + // Clean up old base key entry + let _ = device_guard + .backend + .delete_base_key(addr_str, message_id) + .await; + } + Err(e) => { + warn!("Failed to check base key for {}: {}", signal_address, e); + } + } + } + } + } + + // Delete through the cache so resend can't revive a stale in-memory session. + // The retry path flushes once after the resend so the session transition + // stays visible through the cache for the whole targeted rebuild. + self.signal_cache.delete_session(&signal_address).await; + info!("Deleted session for {signal_address} due to retry receipt"); + Ok(()) + } + /// Extracts and processes the key bundle from a retry receipt. /// This allows us to establish a new session with the requester using their fresh prekeys. /// @@ -913,6 +935,8 @@ mod tests { use crate::store::persistence_manager::PersistenceManager; use crate::test_utils::MockHttpClient; use std::borrow::Cow; + use std::sync::Arc; + use wacore::types::jid::JidExt as _; use wacore_binary::{Jid, JidExt}; use waproto::whatsapp as wa; @@ -1427,6 +1451,79 @@ mod tests { ); } + #[tokio::test] + async fn dm_retry_deletes_only_requested_session() { + let client = + crate::test_utils::create_test_client_with_failing_http("retry_dm_devices").await; + let user = "100000000000088".to_string(); + let resolved_jid = Jid::lid_device(user.clone(), 33); + + let backend = client.persistence_manager.backend(); + let device_0 = Jid::lid_device(user.clone(), 0).to_protocol_address(); + let device_33 = Jid::lid_device(user, 33).to_protocol_address(); + + backend + .put_session(device_0.as_str(), b"invalid-session") + .await + .unwrap(); + backend + .put_session(device_33.as_str(), b"invalid-session") + .await + .unwrap(); + + client + .delete_dm_retry_session_target(&resolved_jid, "MSG-ONE-DEVICE", 1) + .await + .unwrap(); + client.flush_signal_cache().await.unwrap(); + + assert!( + backend + .get_session(device_0.as_str()) + .await + .unwrap() + .is_some(), + "non-requesting device session should be preserved" + ); + assert!( + backend + .get_session(device_33.as_str()) + .await + .unwrap() + .is_none(), + "requesting device session should be deleted" + ); + } + + #[tokio::test] + async fn dm_retry_deletes_resolved_session_without_registry_devices() { + let client = + crate::test_utils::create_test_client_with_failing_http("retry_dm_fallback").await; + let resolved_jid = Jid::lid("100000000000099"); + let signal_address = resolved_jid.to_protocol_address(); + let backend = client.persistence_manager.backend(); + + backend + .put_session(signal_address.as_str(), b"invalid-session") + .await + .unwrap(); + + client + .delete_dm_retry_session_target(&resolved_jid, "MSG-FALLBACK", 1) + .await + .unwrap(); + client.flush_signal_cache().await.unwrap(); + + assert!( + backend + .get_session(signal_address.as_str()) + .await + .unwrap() + .is_none(), + "resolved session should be deleted when registry has no device list" + ); + } + #[test] fn bot_jid_detection() { // Test bot JID detection for bot message filtering @@ -1783,32 +1880,46 @@ mod tests { assert!(participant_jid.is_status_broadcast()); } - /// Test that dedupe keys are correctly differentiated per-participant - /// for status broadcast retries. + /// Test that retry dedupe keys are differentiated per requesting device, + /// including DMs where multiple companion devices can retry the same message. #[test] - fn status_broadcast_dedupe_key_per_participant() { - let chat = "status@broadcast"; + fn retry_dedupe_key_per_participant() { let msg_id = "3EB06D00CAB92340790621"; - let key_a = format!("{}:{}:{}", chat, msg_id, "236395184570386@lid"); - let key_b = format!("{}:{}:{}", chat, msg_id, "559985213786@s.whatsapp.net"); - + let status_chat = Jid::status_broadcast(); + let status_participant_a: Jid = "236395184570386@lid".parse().unwrap(); + let status_participant_b: Jid = "559985213786@s.whatsapp.net".parse().unwrap(); + let status_key_a = build_retry_dedupe_key(&status_chat, msg_id, &status_participant_a); + let status_key_b = build_retry_dedupe_key(&status_chat, msg_id, &status_participant_b); assert_ne!( - key_a, key_b, - "Different participants should have different dedupe keys" + status_key_a, status_key_b, + "Different status participants should have different dedupe keys" + ); + assert_eq!( + status_key_a, + build_retry_dedupe_key(&status_chat, msg_id, &status_participant_a), + "Same status participant should have the same dedupe key" ); - let key_a2 = format!("{}:{}:{}", chat, msg_id, "236395184570386@lid"); + let dm_chat = Jid::pn("559911112222"); + let dm_device_a = Jid::pn_device("559922223333", 1); + let dm_device_b = Jid::pn_device("559922223333", 2); + let dm_key_a = build_retry_dedupe_key(&dm_chat, msg_id, &dm_device_a); + let dm_key_b = build_retry_dedupe_key(&dm_chat, msg_id, &dm_device_b); + assert_ne!( + dm_key_a, dm_key_b, + "Different DM requester devices should have different dedupe keys" + ); assert_eq!( - key_a, key_a2, - "Same participant should have same dedupe key" + dm_key_a, + build_retry_dedupe_key(&dm_chat, msg_id, &dm_device_a), + "Same DM requester device should have the same dedupe key" ); } /// Test that the recent message cache supports re-addition after take. - /// This is critical for status broadcasts where we take the message, - /// mark the participant for fresh SKDM, then re-add so other devices - /// can also retry. + /// This is critical for multi-device retries where another device can + /// ask for the same message after the first retry already consumed it. #[tokio::test] async fn recent_message_cache_readd_after_take() { let _ = env_logger::builder().is_test(true).try_init(); @@ -1832,8 +1943,6 @@ mod tests { ) .await; - let chat = Jid::status_broadcast(); - let msg_id = "STATUS_MSG_001".to_string(); let msg = wa::Message { extended_text_message: Some(Box::new(wa::message::ExtendedTextMessage { text: Some("status text".to_string()), @@ -1842,32 +1951,33 @@ mod tests { ..Default::default() }; - // Add message to cache - client.add_recent_message(&chat, &msg_id, &msg).await; + for (chat, msg_id) in [ + (Jid::status_broadcast(), "STATUS_MSG_001".to_string()), + (Jid::pn("559911112222"), "DM_MSG_001".to_string()), + ] { + client.add_recent_message(&chat, &msg_id, &msg).await; - // First device takes the message - let taken = client.take_recent_message(&chat, &msg_id).await; - assert!(taken.is_some(), "First take should succeed"); + let taken = client.take_recent_message(&chat, &msg_id).await; + assert!(taken.is_some(), "First take should succeed for {chat}"); - // Re-add for subsequent retries (simulating the status broadcast fix) - let taken_msg = taken.unwrap(); - client.add_recent_message(&chat, &msg_id, &taken_msg).await; + let taken_msg = taken.unwrap(); + client.add_recent_message(&chat, &msg_id, &taken_msg).await; - // Second device should also be able to take the message - let taken2 = client.take_recent_message(&chat, &msg_id).await; - assert!( - taken2.is_some(), - "Second take should succeed after re-add (status broadcast multi-device retry)" - ); - assert_eq!( - taken2 - .unwrap() - .extended_text_message - .as_ref() - .unwrap() - .text - .as_deref(), - Some("status text") - ); + let taken2 = client.take_recent_message(&chat, &msg_id).await; + assert!( + taken2.is_some(), + "Second take should succeed after re-add for {chat}" + ); + assert_eq!( + taken2 + .unwrap() + .extended_text_message + .as_ref() + .unwrap() + .text + .as_deref(), + Some("status text") + ); + } } } diff --git a/tests/e2e/tests/retry_dm_multidevice.rs b/tests/e2e/tests/retry_dm_multidevice.rs new file mode 100644 index 000000000..bcf613d0a --- /dev/null +++ b/tests/e2e/tests/retry_dm_multidevice.rs @@ -0,0 +1,139 @@ +//! DM retry recovery after session deletion. + +use e2e_tests::{TestClient, send_and_expect_text, text_msg}; +use log::info; +use wacore::types::events::Event; +use wacore_binary::node::Node; +use whatsapp_rust::{NodeFilter, SendOptions}; + +fn participant_target_count(message_node: &Node) -> usize { + message_node + .get_optional_child("participants") + .and_then(|participants| participants.children()) + .map(|children| children.iter().filter(|child| child.tag == "to").count()) + .unwrap_or_default() +} + +fn retry_enc_count(message_node: &Node) -> Option { + let participants = message_node.get_optional_child("participants")?; + let target = participants.children()?.first()?; + let enc = target.get_optional_child("enc")?; + enc.attrs().optional_string("count").map(|s| s.into_owned()) +} + +#[tokio::test] +async fn test_dm_retry_recovers_after_session_deletion() -> anyhow::Result<()> { + let _ = env_logger::builder().is_test(true).try_init(); + + let mut client_a = TestClient::connect("e2e_retry_dm_a").await?; + let mut client_b = TestClient::connect("e2e_retry_dm_b").await?; + + let jid_a = client_a.jid().await; + let jid_b = client_b.jid().await; + assert_ne!( + jid_a.user, jid_b.user, + "Clients must use different accounts" + ); + + send_and_expect_text( + &client_a.client, + &mut client_b, + &jid_b, + "retry-setup-a2b", + 30, + ) + .await?; + send_and_expect_text( + &client_b.client, + &mut client_a, + &jid_a, + "retry-setup-b2a", + 30, + ) + .await?; + info!("Baseline roundtrip established sessions"); + + client_b + .client + .signal() + .delete_sessions(std::slice::from_ref(&jid_a)) + .await?; + + let message_id = format!("E2ERETRY{}", uuid::Uuid::new_v4().simple()); + let initial_waiter = client_a + .client + .wait_for_sent_node(NodeFilter::tag("message").attr("id", &message_id)); + client_a + .client + .send_message_with_options( + jid_b.clone(), + text_msg("retry-recover"), + SendOptions { + message_id: Some(message_id.clone()), + ..Default::default() + }, + ) + .await?; + + let initial_node = tokio::time::timeout(tokio::time::Duration::from_secs(10), initial_waiter) + .await + .map_err(|_| anyhow::anyhow!("Timed out waiting for initial DM send node"))? + .map_err(|_| anyhow::anyhow!("initial DM send waiter was canceled"))?; + assert!( + retry_enc_count(&initial_node).is_none(), + "Initial DM send should not carry a retry count" + ); + + let retry_waiter = client_a + .client + .wait_for_sent_node(NodeFilter::tag("message").attr("id", &message_id)); + + client_b + .wait_for_event(30, |e| matches!(e, Event::UndecryptableMessage(_))) + .await?; + client_b.wait_for_text("retry-recover", 30).await?; + + let retry_node = tokio::time::timeout(tokio::time::Duration::from_secs(10), retry_waiter) + .await + .map_err(|_| anyhow::anyhow!("Timed out waiting for retry DM send node"))? + .map_err(|_| anyhow::anyhow!("retry DM send waiter was canceled"))?; + assert_eq!( + participant_target_count(&retry_node), + 1, + "Retry resend should target exactly one device" + ); + assert_eq!( + retry_enc_count(&retry_node).as_deref(), + Some("1"), + "Retry resend should mark the payload with count=1" + ); + let jid_b_str = jid_b.to_string(); + assert_eq!( + retry_node.attrs().optional_string("to").as_deref(), + Some(jid_b_str.as_str()), + "Retry resend should keep the user-level chat target" + ); + info!("Retry recovered after B deleted its session with A"); + + send_and_expect_text( + &client_a.client, + &mut client_b, + &jid_b, + "retry-followup-a2b", + 15, + ) + .await?; + send_and_expect_text( + &client_b.client, + &mut client_a, + &jid_a, + "retry-followup-b2a", + 15, + ) + .await?; + info!("Messaging still works after retry recovery"); + + client_a.disconnect().await; + client_b.disconnect().await; + Ok(()) +} diff --git a/wacore/src/send.rs b/wacore/src/send.rs index 59af35d50..44ad70363 100644 --- a/wacore/src/send.rs +++ b/wacore/src/send.rs @@ -863,6 +863,69 @@ where Ok(stanza) } +/// Pairwise-encrypted retry stanza for a single DM recipient device. +/// WA Web retries target only the failing device, not a full DM fanout. +#[allow(clippy::too_many_arguments)] +pub async fn prepare_dm_retry_stanza( + session_store: &mut S, + identity_store: &mut I, + to_jid: Jid, + requester_jid: Jid, + encryption_jid: Jid, + message: &wa::Message, + message_id: String, + retry_count: u8, + account: Option<&wa::AdvSignedDeviceIdentity>, +) -> Result +where + S: crate::libsignal::protocol::SessionStore, + I: crate::libsignal::protocol::IdentityKeyStore, +{ + let plaintext = MessageUtils::encode_and_pad(message); + let signal_address = encryption_jid.to_protocol_address(); + + let encrypted = + message_encrypt(&plaintext, &signal_address, session_store, identity_store).await?; + + let (enc_type, is_prekey, serialized) = extract_ciphertext(encrypted) + .ok_or_else(|| anyhow!("Unexpected encryption message type for DM retry"))?; + + let mut enc_builder = NodeBuilder::new("enc") + .attr("v", stanza::ENC_VERSION) + .attr("type", enc_type) + .attr("count", retry_count.to_string()); + if let Some(mt) = media_type_from_message(message) { + enc_builder = enc_builder.attr("mediatype", mt); + } + let enc_node = enc_builder.bytes(serialized).build(); + + let participant_node = NodeBuilder::new("to") + .attr("jid", requester_jid) + .children([enc_node]) + .build(); + + let mut children = vec![ + NodeBuilder::new("participants") + .children([participant_node]) + .build(), + ]; + + if is_prekey && let Some(acc) = account { + children.push( + NodeBuilder::new("device-identity") + .bytes(acc.encode_to_vec()) + .build(), + ); + } + + Ok(NodeBuilder::new("message") + .attr("to", to_jid) + .attr("id", message_id) + .attr("type", stanza_type_from_message(message)) + .children(children) + .build()) +} + /// Pairwise-encrypted retry stanza for a single group participant. /// WA Web sends retries to the failing device only (RetryMsgJob.js:71), /// NOT as a sender-key broadcast to all participants. @@ -2360,6 +2423,95 @@ mod tests { assert!(n.get_optional_child("device-identity").is_none()); } + #[tokio::test] + async fn dm_retry_pkmsg_targets_single_device() { + let (mut ss, mut is, jid) = setup_session().await; + let to: Jid = "559922223333@s.whatsapp.net".parse().unwrap(); + let requester: Jid = jid.to_string().parse().unwrap(); + let encryption = requester.clone(); + + let n = prepare_dm_retry_stanza( + &mut ss, + &mut is, + to.clone(), + requester.clone(), + encryption, + &wa::Message::default(), + "dm-retry-1".into(), + 1, + None, + ) + .await + .unwrap(); + + assert_eq!(n.tag, "message"); + let mut attrs = n.attrs(); + assert_eq!( + attrs.optional_string("to").unwrap().as_ref(), + to.to_string() + ); + assert_eq!(attrs.optional_string("id").unwrap().as_ref(), "dm-retry-1"); + assert_eq!( + attrs.optional_string("type").unwrap().as_ref(), + stanza::MSG_TYPE_MEDIA + ); + assert!(attrs.optional_string("participant").is_none()); + assert!(attrs.optional_string("addressing_mode").is_none()); + + let participants = n.get_optional_child("participants").unwrap(); + let targets = participants.children().unwrap(); + assert_eq!(targets.len(), 1); + assert_eq!(targets[0].tag, "to"); + assert_eq!( + targets[0].attrs().optional_string("jid").unwrap().as_ref(), + requester.to_string() + ); + + let enc = targets[0].get_optional_child("enc").unwrap(); + let mut enc_attrs = enc.attrs(); + assert_eq!( + enc_attrs.optional_string("type").unwrap().as_ref(), + stanza::ENC_TYPE_PKMSG + ); + assert_eq!(enc_attrs.optional_string("count").unwrap().as_ref(), "1"); + assert!(n.get_optional_child("device-identity").is_none()); + } + + #[tokio::test] + async fn dm_retry_pkmsg_with_account_has_device_identity() { + let (mut ss, mut is, jid) = setup_session().await; + let requester: Jid = jid.to_string().parse().unwrap(); + let acc = wa::AdvSignedDeviceIdentity { + details: Some(b"t".to_vec()), + ..Default::default() + }; + + let n = prepare_dm_retry_stanza( + &mut ss, + &mut is, + "559922223333@s.whatsapp.net".parse().unwrap(), + requester.clone(), + requester, + &wa::Message::default(), + "dm-retry-2".into(), + 2, + Some(&acc), + ) + .await + .unwrap(); + + let participants = n.get_optional_child("participants").unwrap(); + let enc = participants.children().unwrap()[0] + .get_optional_child("enc") + .unwrap(); + assert_eq!( + enc.attrs().optional_string("type").unwrap().as_ref(), + stanza::ENC_TYPE_PKMSG + ); + assert_eq!(enc.attrs().optional_string("count").unwrap().as_ref(), "2"); + assert!(n.get_optional_child("device-identity").is_some()); + } + #[tokio::test] async fn pkmsg_with_account_has_device_identity() { let (mut ss, mut is, jid) = setup_session().await;