Merge branch 'main' into cv/mutable-metadata-part-3-fixed-commits

xmtp_id/src/associations/mod.rs

@@ -14,7 +14,7 @@ pub use self::hashes::generate_inbox_id;
 pub use self::member::{Member, MemberIdentifier, MemberKind};
 pub use self::serialization::DeserializationError;
 pub use self::signature::{Signature, SignatureError, SignatureKind};
-pub use self::state::AssociationState;
+pub use self::state::{AssociationState, AssociationStateDiff};
 
 // Apply a single IdentityUpdate to an existing AssociationState
 pub fn apply_update(

xmtp_id/src/associations/state.rs

@@ -110,6 +110,15 @@ impl AssociationState {
         }
     }
 
+    /// Converts the AssociationState to a diff that represents all members
+    /// of the inbox at the current state.
+    pub fn as_diff(&self) -> AssociationStateDiff {
+        AssociationStateDiff {
+            new_members: self.members.keys().cloned().collect(),
+            removed_members: vec![],
+        }
+    }
+
     pub fn new(account_address: String, nonce: u64) -> Self {
         let inbox_id = generate_inbox_id(&account_address, &nonce);
         let identifier = MemberIdentifier::Address(account_address.clone());

xmtp_mls/src/client.rs

@@ -217,19 +217,30 @@ where
             permissions,
             self.account_address(),
         )
-        .map_err(|e| ClientError::Generic(format!("group create error {}", e)))?;
+        .map_err(|e| {
+            ClientError::Storage(StorageError::Store(format!("group create error {}", e)))
+        })?;
 
         Ok(group)
     }
 
+    pub fn create_sync_group(&self) -> Result<MlsGroup<ApiClient>, ClientError> {
+        log::info!("creating sync group");
+        let sync_group = MlsGroup::create_and_insert_sync_group(self).map_err(|e| {
+            ClientError::Storage(StorageError::Store(format!("group create error {}", e)))
+        })?;
+
+        Ok(sync_group)
+    }
+
     /// Look up a group by its ID
     /// Returns a [`MlsGroup`] if the group exists, or an error if it does not
     pub fn group(&self, group_id: Vec<u8>) -> Result<MlsGroup<ApiClient>, ClientError> {
         let conn = &mut self.store.conn()?;
         let stored_group: Option<StoredGroup> = conn.fetch(&group_id)?;
         match stored_group {
             Some(group) => Ok(MlsGroup::new(self, group.id, group.created_at_ns)),
-            None => Err(ClientError::Generic("group not found".to_string())),
+            None => Err(ClientError::Storage(StorageError::NotFound)),
         }
     }
 

xmtp_mls/src/groups/group_membership.rs

@@ -105,17 +105,17 @@ mod tests {
     use super::GroupMembership;
 
     #[test]
-    fn equality_works() {
+    fn test_equality_works() {
         let inbox_id_1 = "inbox_1".to_string();
         let sequence_id_1: u64 = 1;
         let mut member_map_1 = GroupMembership::new();
         let mut member_map_2 = GroupMembership::new();
 
-        member_map_1.add(inbox_id_1.clone(), sequence_id_1.clone());
+        member_map_1.add(inbox_id_1.clone(), sequence_id_1);
 
         assert!(member_map_1.ne(&member_map_2));
 
-        member_map_2.add(inbox_id_1.clone(), sequence_id_1.clone());
+        member_map_2.add(inbox_id_1.clone(), sequence_id_1);
         assert!(member_map_1.eq(&member_map_2));
 
         // Now change the sequence ID and make sure it is not equal again
@@ -124,7 +124,7 @@ mod tests {
     }
 
     #[test]
-    fn diff() {
+    fn test_diff() {
         let mut initial_members = GroupMembership::new();
         initial_members.add("inbox_1".into(), 1);
         initial_members.add("inbox_2".into(), 1);

xmtp_mls/src/groups/message_history.rs

@@ -97,7 +97,7 @@ mod tests {
     async fn test_send_mesage_history_request() {
         let wallet = generate_local_wallet();
         let client = ClientBuilder::new_test_client(&wallet).await;
-        let group = client.create_group(None).expect("create group");
+        let group = client.create_sync_group().expect("create group");
 
         let result = group.send_message_history_request().await;
         assert_ok!(result);
@@ -107,14 +107,28 @@ mod tests {
     async fn test_send_mesage_history_reply() {
         let wallet = generate_local_wallet();
         let client = ClientBuilder::new_test_client(&wallet).await;
-        let group = client.create_group(None).expect("create group");
-        let expiry = now_ns() + 1_000;
+        let group = client.create_sync_group().expect("create sync group");
 
         let request_id = new_request_id();
         let url = "https://test.com/abc-123";
         let backup_hash = b"ABC123".into();
+        let expiry = now_ns() + 10_000;
         let reply = new_message_history_reply(&request_id, url, backup_hash, expiry);
         let result = group.send_message_history_reply(reply).await;
         assert_ok!(result);
     }
+
+    #[tokio::test]
+    async fn test_request_reply_roundtrip() {
+        let wallet = generate_local_wallet();
+        let amal_a = ClientBuilder::new_test_client(&wallet).await;
+        let amal_b = ClientBuilder::new_test_client(&wallet).await;
+        let group = amal_a.create_group(None).expect("create group");
+        let add_members_result = group
+            .add_members_by_installation_id(vec![amal_b.installation_public_key()])
+            .await;
+        assert_ok!(add_members_result);
+
+        let _ = group.send_message_history_request().await;
+    }
 }

xmtp_mls/src/groups/mod.rs

@@ -297,13 +297,38 @@ where
         Self::create_from_welcome(client, provider, welcome, account_address)
     }
 
-    fn into_envelope(encoded_msg: &[u8], idempotency_key: &str) -> PlaintextEnvelope {
-        PlaintextEnvelope {
-            content: Some(Content::V1(V1 {
-                content: encoded_msg.to_vec(),
-                idempotency_key: idempotency_key.into(),
-            })),
-        }
+    pub(crate) fn create_and_insert_sync_group(
+        client: &'c Client<ApiClient>,
+    ) -> Result<MlsGroup<ApiClient>, GroupError> {
+        let conn = client.store.conn()?;
+        let provider = XmtpOpenMlsProvider::new(&conn);
+        let protected_metadata = build_protected_metadata_extension(
+            &client.identity,
+            PreconfiguredPolicies::default().to_policy_set(),
+        )?;
+        let mutable_metadata = build_mutable_metadata_extension_default()?;
+        let group_config = build_group_config(protected_metadata, mutable_metadata)?;
+        let mut mls_group = OpenMlsGroup::new(
+            &provider,
+            &client.identity.installation_keys,
+            &group_config,
+            CredentialWithKey {
+                credential: client.identity.credential()?,
+                signature_key: client.identity.installation_keys.to_public_vec().into(),
+            },
+        )?;
+        mls_group.save(provider.key_store())?;
+
+        let group_id = mls_group.group_id().to_vec();
+        let stored_group =
+            StoredGroup::new_sync_group(group_id.clone(), now_ns(), GroupMembershipState::Allowed);
+
+        stored_group.store(provider.conn())?;
+        Ok(Self::new(
+            client,
+            stored_group.id,
+            stored_group.created_at_ns,
+        ))
     }
 
     pub async fn send_message(&self, message: &[u8]) -> Result<Vec<u8>, GroupError> {
@@ -351,6 +376,15 @@ where
         Ok(message_id)
     }
 
+    fn into_envelope(encoded_msg: &[u8], idempotency_key: &str) -> PlaintextEnvelope {
+        PlaintextEnvelope {
+            content: Some(Content::V1(V1 {
+                content: encoded_msg.to_vec(),
+                idempotency_key: idempotency_key.into(),
+            })),
+        }
+    }
+
     // Query the database for stored messages. Optionally filtered by time, kind, delivery_status
     // and limit
     pub fn find_messages(

xmtp_mls/src/identity_updates.rs

@@ -1,5 +1,8 @@
 use prost::Message;
-use xmtp_id::associations::{get_state, AssociationError, AssociationState, IdentityUpdate};
+use xmtp_id::associations::{
+    apply_update, get_state, AssociationError, AssociationState, AssociationStateDiff,
+    IdentityUpdate,
+};
 use xmtp_proto::api_client::{XmtpIdentityClient, XmtpMlsClient};
 
 use crate::{
@@ -75,6 +78,33 @@ where
 
         Ok(get_state(updates)?)
     }
+
+    pub fn get_association_state_diff<InboxId: AsRef<str>>(
+        &self,
+        conn: &'a DbConnection<'a>,
+        inbox_id: String,
+        starting_sequence_id: Option<i64>,
+        ending_sequence_id: Option<i64>,
+    ) -> Result<AssociationStateDiff, ClientError> {
+        let initial_state = self.get_association_state(conn, &inbox_id, starting_sequence_id)?;
+        if starting_sequence_id.is_none() {
+            return Ok(initial_state.as_diff());
+        }
+
+        let incremental_updates = conn
+            .get_identity_updates(inbox_id, starting_sequence_id, ending_sequence_id)?
+            .into_iter()
+            .map(|update| update.try_into())
+            .collect::<Result<Vec<IdentityUpdate>, AssociationError>>()?;
+
+        let final_state = incremental_updates
+            .into_iter()
+            .try_fold(initial_state.clone(), |state, update| {
+                apply_update(state, update)
+            })?;
+
+        Ok(initial_state.diff(&final_state))
+    }
 }
 
 #[cfg(test)]

xmtp_mls/src/storage/encrypted_store/group.rs

@@ -64,15 +64,14 @@ impl StoredGroup {
         id: ID,
         created_at_ns: i64,
         membership_state: GroupMembershipState,
-        added_by_address: String,
     ) -> Self {
         Self {
             id,
             created_at_ns,
             membership_state,
             installations_last_checked: 0,
             purpose: Purpose::Sync,
-            added_by_address,
+            added_by_address: "".into(),
         }
     }
 }
@@ -414,12 +413,7 @@ pub(crate) mod tests {
             let created_at_ns = now_ns();
             let membership_state = GroupMembershipState::Allowed;
 
-            let sync_group = StoredGroup::new_sync_group(
-                id,
-                created_at_ns,
-                membership_state,
-                "placeholder_address".to_string(),
-            );
+            let sync_group = StoredGroup::new_sync_group(id, created_at_ns, membership_state);
             let purpose = sync_group.purpose;
             assert_eq!(purpose, Purpose::Sync);
         })

xmtp_mls/src/storage/encrypted_store/group_message.rs

@@ -170,6 +170,13 @@ impl DbConnection<'_> {
         })?)
     }
 
+    pub fn get_sync_group_messages(&self) -> Result<Vec<StoredGroupMessage>, StorageError> {
+        let query = dsl::group_messages
+            .order(dsl::sent_at_ns.asc())
+            .into_boxed();
+        Ok(self.raw_query(|conn| query.load::<StoredGroupMessage>(conn))?)
+    }
+
     pub fn set_delivery_status_to_published<MessageId: AsRef<[u8]>>(
         &self,
         msg_id: &MessageId,