fix!: decode key-index-list to filter stale devices from device registry

src/client/device_registry.rs

@@ -186,31 +186,120 @@ impl Client {
         debug!("Invalidated device cache for user: {} ({:?})", user, lookup);
     }
 
-    /// Granularly patch device registry after a device notification.
+    /// Patch device registry after a device add notification.
     ///
-    /// Matches WA Web's approach: read current → apply diff → write back.
-    /// Looks up the record from cache first, then falls back to the backend
-    /// DB so notifications are never silently dropped.
+    /// Matches WA Web's `handleDeviceAddNotification()` in `AdvDeviceNotificationApi`:
+    /// 1. Decode `key-index-list` signed bytes → `ADVKeyIndexList`
+    /// 2. Filter existing devices by `valid_indexes` (prune stale devices)
+    /// 3. Add the new device
+    /// 4. Replace the full device record
+    ///
+    /// If `signed_bytes` is absent, falls back to simple append (lenient).
     pub(crate) async fn patch_device_add(
         &self,
         user: &str,
         device: &wacore::stanza::devices::DeviceElement,
+        key_index_info: Option<&wacore::stanza::devices::KeyIndexInfo>,
     ) {
         let device_id = device.device_id();
 
-        if let Some(mut record) = self.load_device_record(user).await
-            && !record.devices.iter().any(|d| d.device_id == device_id)
-        {
+        let Some(mut record) = self.load_device_record(user).await else {
+            return;
+        };
+
+        let signed_bytes = key_index_info.and_then(|ki| ki.signed_bytes.as_deref());
+
+        if let Some(bytes) = signed_bytes {
+            if let Some(decoded) = wacore::adv::decode_key_index_list(bytes) {
+                // Check raw_id mismatch (identity change)
+                if let Some(stored_raw_id) = record.raw_id
+                    && stored_raw_id != decoded.raw_id
+                {
+                    info!(
+                        "raw_id mismatch for user {user}: stored={stored_raw_id}, received={}. Clearing record.",
+                        decoded.raw_id
+                    );
+                    self.clear_device_record(user, &device.jid.server, &record)
+                        .await;
+                    record.devices.clear();
+                }
+                record.raw_id = Some(decoded.raw_id);
+
+                // Filter stale devices by valid_indexes
+                record.devices =
+                    wacore::adv::filter_devices_by_key_index(&record.devices, &decoded);
+
+                // Only add the new device if its key_index is accepted by the ADV list
+                if !record.devices.iter().any(|d| d.device_id == device_id)
+                    && wacore::adv::is_key_index_valid(device.key_index, &decoded)
+                {
+                    record.devices.push(wacore::store::traits::DeviceInfo {
+                        device_id,
+                        key_index: device.key_index,
+                    });
+                }
+            } else {
+                warn!("patch_device_add: failed to decode key-index-list for user {user}");
+                self.append_device_if_new(&mut record, device_id, device.key_index);
+            }
+        } else {
+            // No signed bytes — fall back to simple append
+            self.append_device_if_new(&mut record, device_id, device.key_index);
+        }
+
+        if let Err(e) = self.update_device_list(record).await {
+            warn!("patch_device_add: failed to persist: {e}");
+        }
+    }
+
+    /// Append a device if it doesn't already exist in the record.
+    fn append_device_if_new(
+        &self,
+        record: &mut wacore::store::traits::DeviceListRecord,
+        device_id: u32,
+        key_index: Option<u32>,
+    ) {
+        if !record.devices.iter().any(|d| d.device_id == device_id) {
             record.devices.push(wacore::store::traits::DeviceInfo {
                 device_id,
-                key_index: device.key_index,
+                key_index,
             });
-            if let Err(e) = self.update_device_list(record).await {
-                warn!("patch_device_add: failed to persist: {e}");
-            }
         }
     }
 
+    /// Clear device record on raw_id mismatch (identity change).
+    ///
+    /// Matches WA Web's `clearDeviceRecord()` in `IdentityUpdateDeviceTableApi`:
+    /// - Deletes Signal sessions for non-primary devices (stale identity)
+    /// - Invalidates sender key device cache so SKDM will be redistributed
+    /// - Flushes cache to persist session deletions
+    pub(crate) async fn clear_device_record(
+        &self,
+        user: &str,
+        server: &str,
+        record: &wacore::store::traits::DeviceListRecord,
+    ) {
+        let non_primary_count = record.devices.iter().filter(|d| d.device_id != 0).count();
+        info!(
+            "Clearing device record for user {user}: removing {non_primary_count} non-primary device(s) due to raw_id change",
+        );
+
+        // Delete Signal sessions for non-primary devices from cache + DB
+        for device in record.devices.iter().filter(|d| d.device_id != 0) {
+            let mut jid = Jid::new(user, server);
+            jid.device = device.device_id as u16;
+            let addr = wacore::types::jid::JidExt::to_protocol_address(&jid);
+            self.signal_cache.delete_session(&addr).await;
+        }
+        if let Err(e) = self.flush_signal_cache().await {
+            warn!("clear_device_record: failed to flush session deletions: {e}");
+        }
+
+        // Invalidate sender_key_device_cache so stale SKDM tracking is discarded.
+        // Global invalidation because we don't track which groups a user is in.
+        self.sender_key_device_cache.invalidate_all();
+    }
+
     /// Remove a device from the registry after a device remove notification.
     pub(crate) async fn patch_device_remove(&self, user: &str, device_id: u32) {
         if let Some(mut record) = self.load_device_record(user).await {
@@ -243,7 +332,7 @@ impl Client {
     }
 
     /// Load a `DeviceListRecord` from cache or DB for patching.
-    async fn load_device_record(
+    pub(crate) async fn load_device_record(
         &self,
         user: &str,
     ) -> Option<wacore::store::traits::DeviceListRecord> {
@@ -512,6 +601,7 @@ mod tests {
             }],
             timestamp: 12345,
             phash: None,
+            raw_id: None,
         };
         client
             .device_registry_cache
@@ -546,6 +636,7 @@ mod tests {
             }],
             timestamp: 12345,
             phash: None,
+            raw_id: None,
         };
         client
             .device_registry_cache
@@ -570,6 +661,7 @@ mod tests {
             }],
             timestamp: 12346,
             phash: None,
+            raw_id: None,
         };
         client
             .device_registry_cache
@@ -597,6 +689,7 @@ mod tests {
             }],
             timestamp: 12345,
             phash: None,
+            raw_id: None,
         };
         client
             .device_registry_cache
@@ -655,6 +748,7 @@ mod tests {
             }],
             timestamp: wacore::time::now_secs(),
             phash: None,
+            raw_id: None,
         };
         client
             .device_registry_cache
@@ -663,7 +757,7 @@ mod tests {
 
         // Patch: add device 3
         let elem = make_device_element(3, Some(5));
-        client.patch_device_add("15551234567", &elem).await;
+        client.patch_device_add("15551234567", &elem, None).await;
 
         let updated = client
             .device_registry_cache
@@ -690,6 +784,7 @@ mod tests {
             }],
             timestamp: wacore::time::now_secs(),
             phash: None,
+            raw_id: None,
         };
         client
             .device_registry_cache
@@ -698,7 +793,7 @@ mod tests {
 
         // Patch: add device 3 again — should not duplicate
         let elem = make_device_element(3, None);
-        client.patch_device_add("15551234567", &elem).await;
+        client.patch_device_add("15551234567", &elem, None).await;
 
         let updated = client
             .device_registry_cache
@@ -714,7 +809,7 @@ mod tests {
 
         // No pre-populated cache — patch should be a no-op
         let elem = make_device_element(3, None);
-        client.patch_device_add("15551234567", &elem).await;
+        client.patch_device_add("15551234567", &elem, None).await;
 
         assert!(
             client
@@ -745,6 +840,7 @@ mod tests {
             ],
             timestamp: wacore::time::now_secs(),
             phash: None,
+            raw_id: None,
         };
         client
             .device_registry_cache
@@ -781,6 +877,7 @@ mod tests {
             ],
             timestamp: 1000,
             phash: None,
+            raw_id: None,
         };
         client
             .device_registry_cache
@@ -813,6 +910,7 @@ mod tests {
             }],
             timestamp: 1000,
             phash: None,
+            raw_id: None,
         };
         client
             .device_registry_cache
@@ -821,7 +919,7 @@ mod tests {
 
         // Patch: add device 3
         let elem = make_device_element(3, Some(2));
-        client.patch_device_add("15551234567", &elem).await;
+        client.patch_device_add("15551234567", &elem, None).await;
 
         let updated = client
             .device_registry_cache
@@ -857,6 +955,7 @@ mod tests {
             ],
             timestamp: wacore::time::now_secs(),
             phash: None,
+            raw_id: None,
         };
         client
             .persistence_manager
@@ -917,6 +1016,7 @@ mod tests {
             }],
             timestamp: wacore::time::now_secs(),
             phash: None,
+            raw_id: None,
         };
         client
             .device_registry_cache
@@ -969,6 +1069,7 @@ mod tests {
             }],
             timestamp: wacore::time::now_secs(),
             phash: None,
+            raw_id: None,
         };
         client
             .persistence_manager
@@ -987,7 +1088,7 @@ mod tests {
         );
 
         let elem = make_device_element(3, Some(7));
-        client.patch_device_add("15551234567", &elem).await;
+        client.patch_device_add("15551234567", &elem, None).await;
 
         // Verify patch was applied to DB (not silently dropped)
         let updated = client
@@ -1030,6 +1131,7 @@ mod tests {
             ],
             timestamp: wacore::time::now_secs(),
             phash: None,
+            raw_id: None,
         };
         client
             .persistence_manager

src/handlers/notification.rs

@@ -349,7 +349,9 @@ async fn handle_devices_notification(client: &Arc<Client>, node: &Node) {
     match op.operation_type {
         wacore::stanza::devices::DeviceNotificationType::Add => {
             for device in &op.devices {
-                client.patch_device_add(notification.user(), device).await;
+                client
+                    .patch_device_add(notification.user(), device, op.key_index.as_ref())
+                    .await;
             }
         }
         wacore::stanza::devices::DeviceNotificationType::Remove => {
@@ -487,6 +489,12 @@ async fn handle_account_sync_devices(client: &Arc<Client>, node: &Node, devices_
         .map(|v| v as i64)
         .unwrap_or_else(wacore::time::now_secs);
 
+    // Preserve existing raw_id so account_sync doesn't erase it
+    let existing_raw_id = client
+        .load_device_record(&from_jid.user)
+        .await
+        .and_then(|r| r.raw_id);
+
     // Build DeviceListRecord for storage
     // Note: update_device_list() will automatically store under LID if mapping is known
     let device_list = DeviceListRecord {
@@ -500,6 +508,7 @@ async fn handle_account_sync_devices(client: &Arc<Client>, node: &Node, devices_
             .collect(),
         timestamp,
         phash: dhash,
+        raw_id: existing_raw_id,
     };
 
     if let Err(e) = client.update_device_list(device_list).await {

src/send.rs

@@ -486,6 +486,11 @@ impl Client {
         self.update_sender_key_devices(&to_str, &prepared.skdm_devices)
             .await;
 
+        // Invalidate device registry for users whose devices returned 406
+        for user in &prepared.stale_device_users {
+            self.invalidate_device_cache(user).await;
+        }
+
         // Flush cached Signal state to DB after encryption
         if let Err(e) = self.flush_signal_cache().await {
             log::error!("Failed to flush signal cache after send_status_message: {e:?}");
@@ -798,6 +803,7 @@ impl Client {
         struct SkdmUpdate {
             to_str: String,
             devices: Vec<Jid>,
+            stale_users: Vec<String>,
         }
         let mut skdm_update: Option<SkdmUpdate> = None;
         let mut should_issue_tc_token_after_send = false;
@@ -929,6 +935,7 @@ impl Client {
                     skdm_update = Some(SkdmUpdate {
                         to_str: to_str.clone(),
                         devices: prepared.skdm_devices,
+                        stale_users: prepared.stale_device_users,
                     });
                     prepared.node
                 }
@@ -979,6 +986,7 @@ impl Client {
                         skdm_update = Some(SkdmUpdate {
                             to_str,
                             devices: retry_prepared.skdm_devices,
+                            stale_users: retry_prepared.stale_device_users,
                         });
                         retry_prepared.node
                     } else {
@@ -1108,6 +1116,11 @@ impl Client {
         if let Some(update) = skdm_update {
             self.update_sender_key_devices(&update.to_str, &update.devices)
                 .await;
+            // Invalidate device registry for users whose devices returned 406
+            // so the next send re-fetches from server (without stale devices)
+            for user in &update.stale_users {
+                self.invalidate_device_cache(user).await;
+            }
         }
 
         // Flush cached Signal state to DB after encryption

src/sender_key_device_cache.rs

@@ -83,6 +83,12 @@ impl SenderKeyDeviceCache {
         self.inner.invalidate(group_jid).await;
     }
 
+    /// Invalidate all entries. Used on raw_id mismatch (identity change)
+    /// to force SKDM redistribution for all groups.
+    pub(crate) fn invalidate_all(&self) {
+        self.inner.invalidate_all();
+    }
+
     #[cfg(feature = "debug-diagnostics")]
     pub(crate) fn entry_count(&self) -> u64 {
         self.inner.entry_count()