feat: add AB props cache and privacy tokens conditional

src/client.rs

@@ -324,6 +324,7 @@ pub struct Client {
     /// This allows us to reuse existing LID-based sessions when sending replies.
     /// The cache is backed by persistent storage and warmed up on client initialization.
     pub(crate) lid_pn_cache: Arc<LidPnCache>,
+    pub(crate) ab_props: Arc<wacore::store::ab_props::AbPropsCache>,
 
     /// Per-chat mutex for serializing message enqueue operations.
     /// This ensures messages are enqueued in the order they arrive,
@@ -614,6 +615,7 @@ impl Client {
                 &cache_config.lid_pn_cache,
                 cache_config.cache_stores.lid_pn_cache.clone(),
             )),
+            ab_props: Arc::new(wacore::store::ab_props::AbPropsCache::new()),
             message_enqueue_locks: Cache::builder()
                 .max_capacity(cache_config.message_enqueue_locks_capacity.max(1))
                 .build(),
@@ -1558,13 +1560,14 @@ impl Client {
             .props_hash
             .clone();
 
+        // Deltas only contain changed props, so they're invalid against an empty cache.
         let spec = match &stored_hash {
-            Some(hash) => {
+            Some(hash) if self.ab_props.is_seeded() => {
                 debug!("Fetching props with hash for delta update...");
                 PropsSpec::with_hash(hash)
             }
-            None => {
-                debug!("Fetching props (full, no stored hash)...");
+            _ => {
+                debug!("Fetching props (full)...");
                 PropsSpec::new()
             }
         };
@@ -1584,6 +1587,8 @@ impl Client {
             );
         }
 
+        self.ab_props.apply_response(&response).await;
+
         if let Some(new_hash) = response.hash {
             self.persistence_manager
                 .process_command(DeviceCommand::SetPropsHash(Some(new_hash)))
@@ -1593,6 +1598,10 @@ impl Client {
         Ok(())
     }
 
+    pub(crate) fn ab_props(&self) -> &wacore::store::ab_props::AbPropsCache {
+        &self.ab_props
+    }
+
     pub async fn fetch_privacy_settings(
         &self,
     ) -> Result<wacore::iq::privacy::PrivacySettingsResponse, crate::request::IqError> {

src/features/groups.rs

@@ -205,6 +205,16 @@ impl<'a> Groups<'a> {
 
         options.participants = normalize_participants(&resolved_participants);
 
+        if self
+            .client
+            .ab_props()
+            .is_enabled(wacore::iq::props::config_codes::PRIVACY_TOKEN_ON_GROUP_CREATE)
+            .await
+        {
+            self.attach_tokens_to_participants(&mut options.participants)
+                .await;
+        }
+
         let gid = self.client.execute(GroupCreateIq::new(options)).await?;
 
         Ok(CreateGroupResult { gid })
@@ -244,10 +254,19 @@ impl<'a> Groups<'a> {
         jid: &Jid,
         participants: &[Jid],
     ) -> Result<Vec<ParticipantChangeResponse>, anyhow::Error> {
-        let result = self
+        let iq = if self
             .client
-            .execute(AddParticipantsIq::new(jid, participants))
-            .await?;
+            .ab_props()
+            .is_enabled(wacore::iq::props::config_codes::PRIVACY_TOKEN_ON_GROUP_PARTICIPANT_ADD)
+            .await
+        {
+            let options = self.resolve_participant_tokens(participants).await;
+            AddParticipantsIq::with_options(jid, options)
+        } else {
+            AddParticipantsIq::new(jid, participants)
+        };
+
+        let result = self.client.execute(iq).await?;
         // Patch cache with only the participants the server accepted (status 200).
         // Note: the get→mutate→insert is not atomic; a concurrent notification
         // for the same group could race.  This is acceptable — the cache is
@@ -452,6 +471,105 @@ impl<'a> Groups<'a> {
             .execute(SetMemberAddModeIq::new(jid, mode))
             .await?)
     }
+
+    async fn resolve_participant_tokens(&self, jids: &[Jid]) -> Vec<GroupParticipantOptions> {
+        if jids.is_empty() {
+            return Vec::new();
+        }
+        let only_lid = self.only_check_lid().await;
+        let futs = jids.iter().map(|jid| async move {
+            let mut opt = GroupParticipantOptions::new(jid.clone());
+            if let Some(token_key) = self.resolve_token_key(jid, only_lid).await
+                && let Some(token) = self.lookup_valid_token(&token_key).await
+            {
+                opt = opt.with_privacy(token);
+            }
+            opt
+        });
+        futures::future::join_all(futs).await
+    }
+
+    /// Skips participants that already have a token set by the caller.
+    async fn attach_tokens_to_participants(&self, participants: &mut [GroupParticipantOptions]) {
+        if participants.is_empty() {
+            return;
+        }
+        let only_lid = self.only_check_lid().await;
+        let futs = participants.iter().enumerate().map(|(i, p)| async move {
+            if p.privacy.is_some() {
+                return (i, None);
+            }
+            let Some(token_key) = self.resolve_token_key(&p.jid, only_lid).await else {
+                log::debug!(
+                    target: "Client/Groups",
+                    "No LID mapping for participant {}, skipping privacy attachment",
+                    p.jid
+                );
+                return (i, None);
+            };
+            let token = self.lookup_valid_token(&token_key).await;
+            if token.is_none() {
+                log::debug!(
+                    target: "Client/Groups",
+                    "No valid tc_token for participant {} (key={}), skipping privacy attachment",
+                    p.jid, token_key
+                );
+            }
+            (i, token)
+        });
+        for (i, token) in futures::future::join_all(futs).await {
+            if token.is_some() {
+                participants[i].privacy = token;
+            }
+        }
+    }
+
+    async fn only_check_lid(&self) -> bool {
+        self.client
+            .ab_props()
+            .is_enabled(wacore::iq::props::config_codes::PRIVACY_TOKEN_ONLY_CHECK_LID)
+            .await
+    }
+
+    /// Resolve JID to tc_token store key. When `only_lid`, PN JIDs without a
+    /// LID mapping return `None` instead of falling back to the PN user.
+    async fn resolve_token_key(&self, jid: &Jid, only_lid: bool) -> Option<String> {
+        if jid.is_lid() {
+            Some(jid.user.clone())
+        } else if only_lid {
+            self.client.lid_pn_cache.get_current_lid(&jid.user).await
+        } else {
+            Some(
+                self.client
+                    .lid_pn_cache
+                    .get_current_lid(&jid.user)
+                    .await
+                    .unwrap_or_else(|| jid.user.clone()),
+            )
+        }
+    }
+
+    /// Returns the tc_token if present and not expired.
+    async fn lookup_valid_token(&self, token_key: &str) -> Option<Vec<u8>> {
+        use wacore::iq::tctoken::is_tc_token_expired;
+        let backend = self.client.persistence_manager.backend();
+        match backend.get_tc_token(token_key).await {
+            Ok(Some(entry))
+                if !entry.token.is_empty() && !is_tc_token_expired(entry.token_timestamp) =>
+            {
+                Some(entry.token)
+            }
+            Ok(_) => None,
+            Err(e) => {
+                log::warn!(
+                    target: "Client/Groups",
+                    "Failed to get tc_token for {}: {e}",
+                    token_key
+                );
+                None
+            }
+        }
+    }
 }
 
 impl Client {

tests/e2e/src/lib.rs

@@ -105,7 +105,16 @@ impl TestClient {
         client.register_handler(event_handler);
         let run_handle = bot.run().await?;
 
-        // Wait for PairSuccess + Connected
+        // Wait for PairSuccess + Connected.
+        //
+        // PairSuccess arrives quickly (handshake only), but Connected is dispatched
+        // only after the critical app-state sync completes (sync_collections_batched).
+        // Under CI load with many concurrent clients, the mock server may be slow to
+        // serve app-state IQs, so Connected can take significantly longer than pairing.
+        //
+        // We use a two-phase timeout: 30s for pairing, then an additional 30s for
+        // Connected (which includes critical sync). This avoids a single shared timeout
+        // where a slow sync eats into the pairing budget.
         let timeout = tokio::time::Duration::from_secs(30);
         let mut got_pair = false;
         let mut got_connected = false;
@@ -144,11 +153,43 @@ impl TestClient {
 
         match wait_result {
             Err(_) => {
-                client.disconnect().await;
-                drop(run_handle); // aborts task via AbortHandle drop
-                return Err(anyhow::anyhow!(
-                    "Timed out waiting for PairSuccess + Connected"
-                ));
+                // If we got PairSuccess but not Connected, the critical sync is slow.
+                // Give it extra time via wait_for_startup_sync instead of failing immediately.
+                if got_pair && !got_connected {
+                    eprintln!(
+                        "WARN: Got PairSuccess but Connected timed out after {timeout:?}, \
+                         waiting for startup sync..."
+                    );
+                    if let Err(e) = client
+                        .wait_for_startup_sync(tokio::time::Duration::from_secs(30))
+                        .await
+                    {
+                        client.disconnect().await;
+                        drop(run_handle);
+                        return Err(anyhow::anyhow!(
+                            "Timed out waiting for Connected after PairSuccess: {e}"
+                        ));
+                    }
+                    // Drain the Connected event that should now be available
+                    let connected_timeout = tokio::time::Duration::from_secs(5);
+                    let _ = tokio::time::timeout(connected_timeout, async {
+                        loop {
+                            match event_rx.recv().await {
+                                Ok(Event::Connected(_)) => break,
+                                Ok(_) => continue,
+                                Err(_) => break,
+                            }
+                        }
+                    })
+                    .await;
+                } else {
+                    client.disconnect().await;
+                    drop(run_handle);
+                    return Err(anyhow::anyhow!(
+                        "Timed out waiting for PairSuccess + Connected \
+                         (got_pair={got_pair}, got_connected={got_connected})"
+                    ));
+                }
             }
             Ok(Err(e)) => {
                 client.disconnect().await;
@@ -158,9 +199,6 @@ impl TestClient {
             Ok(Ok(())) => {}
         }
 
-        assert!(got_pair, "Should have received PairSuccess");
-        assert!(got_connected, "Should have received Connected");
-
         if let Err(e) = client
             .wait_for_startup_sync(tokio::time::Duration::from_secs(15))
             .await