[Fix] User deletion & migration

src/anonymous_clients/anonymous_client_connection.rs

@@ -302,7 +302,6 @@ impl AnonymousClientConnection {
 
             if cv.is_type(CommunicationType::GetUserData)
                 || cv.is_type(CommunicationType::GetIotaData)
-                || cv.is_type(CommunicationType::DeleteUser)
             {
                 self.handle_omega_forward(cv).await;
                 return;

src/omega/omega_connection.rs

@@ -624,6 +624,22 @@ impl OmegaConnection {
                                     }
                                 }
                             }
+                            if cv.is_type(CommunicationType::EraseHostedUserData) {
+                                let Some(iota_id) = cv
+                                    .get_data(DataType::IotaId)
+                                    .as_number()
+                                    .and_then(|id| i64::try_from(id).ok())
+                                else {
+                                    log_err!(0, PrintType::Omega, "Discarded hosted-data erasure without IotaId");
+                                    continue;
+                                };
+                                if let Some(rho) = self.rho.get_by_iota(iota_id).await {
+                                    rho.get_iota_connection().send_message(&cv).await;
+                                } else {
+                                    log_err!(iota_id, PrintType::Omega, "Hosted-data erasure is pending because the Iota is offline");
+                                }
+                                continue;
+                            }
                             if cv.is_type(CommunicationType::ClientChanged) {
                                 let Some((receiver, session_id)) = client_changed_target(&cv)
                                 else {

src/rho/client_connection.rs

@@ -77,6 +77,12 @@ impl ClientConnection {
         *self.rho_connection.write().await = Some(rho_connection);
     }
 
+    /// A membership snapshot can remove an account from an Iota while keeping
+    /// its authenticated Omega connection alive.
+    pub async fn clear_rho_connection(&self) {
+        *self.rho_connection.write().await = None;
+    }
+
     /// Send a CommunicationValue to the client
     pub async fn send_message(self: Arc<Self>, cv: &CommunicationValue) {
         if !*self.is_open.read().await {

src/rho/iota_connection.rs

@@ -442,6 +442,12 @@ impl IotaConnection {
             || cv.is_type(CommunicationType::GetUserData)
             || cv.is_type(CommunicationType::GetIotaData)
             || cv.is_type(CommunicationType::DeleteIota)
+            || cv.is_type(CommunicationType::AttachUserBegin)
+            || cv.is_type(CommunicationType::AttachUserComplete)
+            || cv.is_type(CommunicationType::ReleaseUserFromIota)
+            || cv.is_type(CommunicationType::DeleteUserCredentialBegin)
+            || cv.is_type(CommunicationType::DeleteUserCredentialComplete)
+            || cv.is_type(CommunicationType::EraseHostedUserDataAck)
         {
             let sender = self.get_iota_id().await;
 

src/rho/rho_connection.rs

@@ -86,6 +86,14 @@ impl RhoConnection {
             .collect()
     }
 
+    pub async fn detach_user_clients(&self, user_id: i64) {
+        let clients = self.get_client_connections_for_user(user_id).await;
+        for client in clients {
+            client.clear_rho_connection().await;
+        }
+        self.client_connections.retain(|(id, _), _| *id != user_id as u64);
+    }
+
     #[allow(unused)]
     pub async fn get_app_connections(
         &self,

src/rho/rho_manager.rs

@@ -66,7 +66,11 @@ impl RhoManager {
             )
             .await;
 
+        let replacement = user_ids.iter().copied().collect::<std::collections::HashSet<_>>();
         for user_id in previous_users {
+            if !replacement.contains(&user_id) {
+                rho.detach_user_clients(user_id).await;
+            }
             if self
                 .users
                 .get(&user_id)