Merge branch 'main' of https://github.com/xmtp/libxmtp into offsite/s…

…cw-libxmtp

bindings_ffi/src/mls.rs

@@ -10,6 +10,7 @@ use xmtp_api_grpc::grpc_api_helper::Client as TonicApiClient;
 use xmtp_id::associations::unverified::UnverifiedSignature;
 use xmtp_id::associations::AccountId;
 use xmtp_id::associations::AssociationState;
+use xmtp_id::associations::MemberIdentifier;
 use xmtp_id::scw_verifier::RpcSmartContractWalletVerifier;
 use xmtp_id::scw_verifier::SmartContractSignatureVerifier;
 use xmtp_id::{
@@ -461,16 +462,32 @@ impl FfiXmtpClient {
 pub struct FfiInboxState {
     pub inbox_id: String,
     pub recovery_address: String,
-    pub installation_ids: Vec<Vec<u8>>,
+    pub installations: Vec<FfiInstallation>,
     pub account_addresses: Vec<String>,
 }
 
+#[derive(uniffi::Record)]
+pub struct FfiInstallation {
+    pub id: Vec<u8>,
+    pub client_timestamp_ns: Option<u64>,
+}
+
 impl From<AssociationState> for FfiInboxState {
     fn from(state: AssociationState) -> Self {
         Self {
             inbox_id: state.inbox_id().to_string(),
             recovery_address: state.recovery_address().to_string(),
-            installation_ids: state.installation_ids(),
+            installations: state
+                .members()
+                .into_iter()
+                .filter_map(|m| match m.identifier {
+                    MemberIdentifier::Address(_) => None,
+                    MemberIdentifier::Installation(inst) => Some(FfiInstallation {
+                        id: inst,
+                        client_timestamp_ns: m.client_timestamp_ns,
+                    }),
+                })
+                .collect(),
             account_addresses: state.account_addresses(),
         }
     }
@@ -3651,8 +3668,8 @@ mod tests {
 
         let client_1_state = client_1.inbox_state(true).await.unwrap();
         let client_2_state = client_2.inbox_state(true).await.unwrap();
-        assert_eq!(client_1_state.installation_ids.len(), 2);
-        assert_eq!(client_2_state.installation_ids.len(), 2);
+        assert_eq!(client_1_state.installations.len(), 2);
+        assert_eq!(client_2_state.installations.len(), 2);
 
         let signature_request = client_1.revoke_all_other_installations().await.unwrap();
         sign_with_wallet(&wallet, &signature_request).await;
@@ -3663,22 +3680,22 @@ mod tests {
 
         let client_1_state_after_revoke = client_1.inbox_state(true).await.unwrap();
         let client_2_state_after_revoke = client_2.inbox_state(true).await.unwrap();
-        assert_eq!(client_1_state_after_revoke.installation_ids.len(), 1);
-        assert_eq!(client_2_state_after_revoke.installation_ids.len(), 1);
+        assert_eq!(client_1_state_after_revoke.installations.len(), 1);
+        assert_eq!(client_2_state_after_revoke.installations.len(), 1);
         assert_eq!(
             client_1_state_after_revoke
-                .installation_ids
+                .installations
                 .first()
                 .unwrap()
-                .clone(),
+                .id,
             client_1.installation_id()
         );
         assert_eq!(
             client_2_state_after_revoke
-                .installation_ids
+                .installations
                 .first()
                 .unwrap()
-                .clone(),
+                .id,
             client_1.installation_id()
         );
     }

xmtp_id/src/associations/state.rs

@@ -13,6 +13,12 @@ pub struct AssociationStateDiff {
     pub removed_members: Vec<MemberIdentifier>,
 }
 
+#[derive(Debug)]
+pub struct Installation {
+    pub id: Vec<u8>,
+    pub client_timestamp_ns: Option<u64>,
+}
+
 impl AssociationStateDiff {
     pub fn new_installations(&self) -> Vec<Vec<u8>> {
         self.new_members
@@ -128,6 +134,19 @@ impl AssociationState {
             .collect()
     }
 
+    pub fn installations(&self) -> Vec<Installation> {
+        self.members()
+            .into_iter()
+            .filter_map(|member| match member.identifier {
+                MemberIdentifier::Address(_) => None,
+                MemberIdentifier::Installation(id) => Some(Installation {
+                    id,
+                    client_timestamp_ns: member.client_timestamp_ns,
+                }),
+            })
+            .collect()
+    }
+
     pub fn diff(&self, new_state: &Self) -> AssociationStateDiff {
         let new_members: Vec<MemberIdentifier> = new_state
             .members

xmtp_mls/src/groups/mod.rs

@@ -340,6 +340,7 @@ impl MlsGroup {
         added_by_inbox: String,
         welcome_id: i64,
     ) -> Result<Self, GroupError> {
+        log::info!("Creating from welcome");
         let mls_welcome =
             StagedWelcome::new_from_welcome(provider, &build_group_join_config(), welcome, None)?;
 
@@ -386,6 +387,7 @@ impl MlsGroup {
         encrypted_welcome_bytes: Vec<u8>,
         welcome_id: i64,
     ) -> Result<Self, GroupError> {
+        log::info!("Trying to decrypt welcome");
         let welcome_bytes = decrypt_welcome(provider, hpke_public_key, &encrypted_welcome_bytes)?;
 
         let welcome = deserialize_welcome(&welcome_bytes)?;
@@ -396,6 +398,7 @@ impl MlsGroup {
             ProcessedWelcome::new_from_welcome(provider, &join_config, welcome.clone())?;
         let psks = processed_welcome.psks();
         if !psks.is_empty() {
+            log::error!("No PSK support for welcome");
             return Err(GroupError::NoPSKSupport);
         }
         let staged_welcome = processed_welcome.into_staged_welcome(provider, None)?;
@@ -1258,6 +1261,7 @@ async fn validate_initial_group_membership<ApiClient: XmtpApi>(
     conn: &DbConnection,
     mls_group: &OpenMlsGroup,
 ) -> Result<(), GroupError> {
+    log::info!("Validating initial group membership");
     let membership = extract_group_membership(mls_group.extensions())?;
     let needs_update = client.filter_inbox_ids_needing_updates(conn, membership.to_filters())?;
     if !needs_update.is_empty() {
@@ -1289,6 +1293,7 @@ async fn validate_initial_group_membership<ApiClient: XmtpApi>(
         return Err(GroupError::InvalidGroupMembership);
     }
 
+    log::info!("Group membership validated");
     Ok(())
 }
 

xmtp_mls/src/storage/encrypted_store/group.rs

@@ -212,6 +212,7 @@ impl DbConnection {
     }
 
     pub fn insert_or_replace_group(&self, group: StoredGroup) -> Result<StoredGroup, StorageError> {
+        log::info!("Trying to insert group");
         let stored_group = self.raw_query(|conn| {
             let maybe_inserted_group: Option<StoredGroup> = diesel::insert_into(dsl::groups)
                 .values(&group)
@@ -222,14 +223,18 @@ impl DbConnection {
             if maybe_inserted_group.is_none() {
                 let existing_group: StoredGroup = dsl::groups.find(group.id).first(conn)?;
                 if existing_group.welcome_id == group.welcome_id {
+                    log::info!("Group welcome id already exists");
                     // Error so OpenMLS db transaction are rolled back on duplicate welcomes
                     return Err(diesel::result::Error::DatabaseError(
                         diesel::result::DatabaseErrorKind::UniqueViolation,
                         Box::new("welcome id already exists".to_string()),
                     ));
                 } else {
+                    log::info!("Group already exists");
                     return Ok(existing_group);
                 }
+            } else {
+                log::info!("Group is inserted");
             }
 
             match maybe_inserted_group {

xmtp_mls/src/subscriptions.rs

@@ -77,6 +77,7 @@ where
         let creation_result = retry_async!(
             Retry::default(),
             (async {
+                log::info!("Trying to process streamed welcome");
                 let welcome_v1 = welcome_v1.clone();
                 self.context
                     .store
@@ -147,6 +148,7 @@ where
         let installation_key = self.installation_public_key();
         let id_cursor = 0;
 
+        log::info!("Setting up conversation stream");
         let subscription = self
             .api_client
             .subscribe_welcome_messages(installation_key, Some(id_cursor))
@@ -242,6 +244,7 @@ where
             futures::pin_mut!(stream);
             let _ = tx.send(());
             while let Some(convo) = stream.next().await {
+                log::info!("Trigger conversation callback");
                 convo_callback(convo)
             }
             log::debug!("`stream_conversations` stream ended, dropping stream");
@@ -298,6 +301,7 @@ where
                 .await?;
             futures::pin_mut!(messages_stream);
 
+            log::info!("Setting up conversation stream in stream_all_messages");
             let convo_stream = self.stream_conversations().await?;
             futures::pin_mut!(convo_stream);
 
@@ -321,6 +325,7 @@ where
                         yield Ok(message);
                     }
                     Some(new_group) = convo_stream.next() => {
+                        log::info!("Received new conversation inside streamAllMessages");
                         if group_id_to_info.contains_key(&new_group.group_id) {
                             continue;
                         }