Consent Sync (#1152)

* wip

* reorganize

* reuse group

* wip

* type fixes

* wip

* cleanup wip

* get and set

* fix diesel issue

* cleanup

* sync reply

* wip

* more generalization

* processing reply

* skip the fs

* syncables

* cleanup

* wip

* rearrange a bit more

* no more fs in history sync

* all fs stuff gone for real

* fix verify pin

* update encrypt/decrypt test

* cleanup tests

* fix

* more cleanup

* kv query optional

* this test was bad

* explain

* check the pin

* update cli client

* fix a test

* fix another test

* fix kvstore get

* remove feature flag

* lint

* lint

* query update

* wip

* consent sync test

* test cleanup

* borrow consent records

* borrow, borrow, borrow

* finish message sync test

* lint and naming

* add to cli

* bench

* lint

* lint

* revert

* update protos

* remove kv store

* revert

* add time slip limit

* dedup

* better serialization and parsing

* update enc

* maybe insert consent record

* add a test

* test a bit more

* do not ignore tests

* revert

* revert

* Simplify logic

* function name update

* tweak reply flow

* make diesel happy

* conn as a param

* cleanup

* lint

* more store refs

* nits

* cleanup

* fix test

* cleanup

* lint

bindings_ffi/Cargo.toml

@@ -17,11 +17,10 @@ uniffi = { version = "0.28.0", default-features = false, features = ["tokio"] }
 xmtp_api_grpc = { path = "../xmtp_api_grpc" }
 xmtp_cryptography = { path = "../xmtp_cryptography" }
 xmtp_id = { path = "../xmtp_id" }
-xmtp_mls = { path = "../xmtp_mls", features = ["message-history"] }
+xmtp_mls = { path = "../xmtp_mls" }
 xmtp_proto = { path = "../xmtp_proto", features = ["proto_full"] }
 xmtp_user_preferences = { path = "../xmtp_user_preferences" }
 xmtp_v2 = { path = "../xmtp_v2" }
-tonic.workspace = true
 
 [build-dependencies]
 uniffi = { version = "0.28.0", features = ["build"] }
@@ -37,7 +36,6 @@ rand.workspace = true
 tokio = { workspace = true, features = ["rt-multi-thread"] }
 uniffi = { version = "0.28.0", features = ["bindgen-tests"] }
 uuid = { workspace = true, features = ["v4", "fast-rng"] }
+xmtp_api_grpc = { path = "../xmtp_api_grpc", features = ["test-utils"] }
 xmtp_mls = { path = "../xmtp_mls", features = ["test-utils"] }
 xmtp_proto = { path = "../xmtp_proto", features = ["test-utils"] }
-xmtp_api_grpc = { path = "../xmtp_api_grpc", features = ["test-utils"] }
-

bindings_ffi/src/mls.rs

@@ -15,6 +15,7 @@ use xmtp_id::{
     },
     InboxId,
 };
+use xmtp_mls::groups::scoped_client::LocalScopedGroupClient;
 use xmtp_mls::storage::group_message::MsgQueryArgs;
 use xmtp_mls::storage::group_message::SortDirection;
 use xmtp_mls::{
@@ -281,7 +282,7 @@ impl FfiXmtpClient {
     ) -> Result<HashMap<String, bool>, GenericError> {
         let inner = self.inner_client.as_ref();
 
-        let results: HashMap<String, bool> = inner.can_message(account_addresses).await?;
+        let results: HashMap<String, bool> = inner.can_message(&account_addresses).await?;
 
         Ok(results)
     }
@@ -353,7 +354,7 @@ impl FfiXmtpClient {
         let stored_records: Vec<StoredConsentRecord> =
             records.into_iter().map(StoredConsentRecord::from).collect();
 
-        inner.set_consent_states(stored_records).await?;
+        inner.set_consent_states(&stored_records).await?;
         Ok(())
     }
 
@@ -397,10 +398,15 @@ impl FfiXmtpClient {
     }
 
     pub async fn request_history_sync(&self) -> Result<(), GenericError> {
+        let provider = self
+            .inner_client
+            .mls_provider()
+            .map_err(GenericError::from_error)?;
         self.inner_client
-            .send_history_request()
+            .send_history_sync_request(&provider)
             .await
             .map_err(GenericError::from_error)?;
+
         Ok(())
     }
 
@@ -790,7 +796,7 @@ impl FfiConversations {
                 .create_group(group_permissions, metadata_options)?
         } else {
             self.inner_client
-                .create_group_with_members(account_addresses, group_permissions, metadata_options)
+                .create_group_with_members(&account_addresses, group_permissions, metadata_options)
                 .await?
         };
 
@@ -822,7 +828,8 @@ impl FfiConversations {
 
     pub async fn sync(&self) -> Result<(), GenericError> {
         let inner = self.inner_client.as_ref();
-        inner.sync_welcomes().await?;
+        let conn = inner.store().conn()?;
+        inner.sync_welcomes(&conn).await?;
         Ok(())
     }
 
@@ -1206,7 +1213,8 @@ impl FfiConversation {
 
     pub async fn add_members(&self, account_addresses: Vec<String>) -> Result<(), GenericError> {
         log::info!("adding members: {}", account_addresses.join(","));
-        self.inner.add_members(account_addresses).await?;
+
+        self.inner.add_members(&account_addresses).await?;
 
         Ok(())
     }
@@ -1216,15 +1224,16 @@ impl FfiConversation {
         inbox_ids: Vec<String>,
     ) -> Result<(), GenericError> {
         log::info!("adding members by inbox id: {}", inbox_ids.join(","));
+
         self.inner
-            .add_members_by_inbox_id(inbox_ids)
+            .add_members_by_inbox_id(&inbox_ids)
             .await
             .map_err(Into::into)
     }
 
     pub async fn remove_members(&self, account_addresses: Vec<String>) -> Result<(), GenericError> {
         self.inner
-            .remove_members(account_addresses)
+            .remove_members(&account_addresses)
             .await
             .map_err(Into::into)
     }
@@ -1233,7 +1242,7 @@ impl FfiConversation {
         &self,
         inbox_ids: Vec<String>,
     ) -> Result<(), GenericError> {
-        self.inner.remove_members_by_inbox_id(inbox_ids).await?;
+        self.inner.remove_members_by_inbox_id(&inbox_ids).await?;
         Ok(())
     }
 
@@ -3017,6 +3026,8 @@ mod tests {
         let bo = new_test_client().await;
         let caro = new_test_client().await;
 
+        let caro_conn = caro.inner_client.store().conn().unwrap();
+
         let alix_group = alix
             .conversations()
             .create_group(
@@ -3045,7 +3056,7 @@ mod tests {
             )
             .await
             .unwrap();
-        let _ = caro.inner_client.sync_welcomes().await.unwrap();
+        let _ = caro.inner_client.sync_welcomes(&caro_conn).await.unwrap();
 
         bo_group.send("second".as_bytes().to_vec()).await.unwrap();
         stream_callback.wait_for_delivery(None).await.unwrap();
@@ -3064,6 +3075,8 @@ mod tests {
         let amal = new_test_client().await;
         let bola = new_test_client().await;
 
+        let bola_conn = bola.inner_client.store().conn().unwrap();
+
         let amal_group: Arc<FfiConversation> = amal
             .conversations()
             .create_group(
@@ -3073,7 +3086,7 @@ mod tests {
             .await
             .unwrap();
 
-        bola.inner_client.sync_welcomes().await.unwrap();
+        bola.inner_client.sync_welcomes(&bola_conn).await.unwrap();
         let bola_group = bola.conversation(amal_group.id()).unwrap();
 
         let stream_callback = RustStreamCallback::default();

bindings_node/Cargo.toml

@@ -9,16 +9,19 @@ crate-type = ["cdylib"]
 [dependencies]
 # Default enable napi4 feature, see https://nodejs.org/api/n-api.html#node-api-version-matrix
 hex.workspace = true
-napi = { version = "2.12.2", default-features = false, features = ["napi6", "async"] }
+napi = { version = "2.12.2", default-features = false, features = [
+  "napi6",
+  "async",
+] }
 napi-derive = "2.12.2"
-tracing.workspace = true
-tracing-subscriber = { workspace = true, features = ["env-filter", "fmt"] }
 prost.workspace = true
-tokio = { workspace = true, features = ["sync"]}
+tokio = { workspace = true, features = ["sync"] }
+tracing-subscriber = { workspace = true, features = ["env-filter", "fmt"] }
+tracing.workspace = true
 xmtp_api_grpc = { path = "../xmtp_api_grpc" }
 xmtp_cryptography = { path = "../xmtp_cryptography" }
 xmtp_id = { path = "../xmtp_id" }
-xmtp_mls = { path = "../xmtp_mls", features = ["message-history"]}
+xmtp_mls = { path = "../xmtp_mls" }
 xmtp_proto = { path = "../xmtp_proto", features = ["proto_full"] }
 
 [build-dependencies]

bindings_node/src/consent_state.rs

@@ -74,7 +74,7 @@ impl NapiClient {
 
     self
       .inner_client()
-      .set_consent_states(stored_records)
+      .set_consent_states(&stored_records)
       .await
       .map_err(ErrorWrapper::from)?;
     Ok(())

bindings_node/src/conversations.rs

@@ -165,7 +165,7 @@ impl NapiConversations {
     } else {
       self
         .inner_client
-        .create_group_with_members(account_addresses, group_permissions, metadata_options)
+        .create_group_with_members(&account_addresses, group_permissions, metadata_options)
         .await
         .map_err(|e| Error::from_reason(format!("ClientError: {}", e)))?
     };
@@ -234,9 +234,14 @@ impl NapiConversations {
 
   #[napi]
   pub async fn sync(&self) -> Result<()> {
+    let conn = self
+      .inner_client
+      .store()
+      .conn()
+      .map_err(ErrorWrapper::from)?;
     self
       .inner_client
-      .sync_welcomes()
+      .sync_welcomes(&conn)
       .await
       .map_err(ErrorWrapper::from)?;
     Ok(())

bindings_node/src/groups.rs

@@ -291,7 +291,7 @@ impl NapiGroup {
     );
 
     group
-      .add_members(account_addresses)
+      .add_members(&account_addresses)
       .await
       .map_err(ErrorWrapper::from)?;
 
@@ -380,7 +380,7 @@ impl NapiGroup {
     );
 
     group
-      .add_members_by_inbox_id(inbox_ids)
+      .add_members_by_inbox_id(&inbox_ids)
       .await
       .map_err(ErrorWrapper::from)?;
 
@@ -396,7 +396,7 @@ impl NapiGroup {
     );
 
     group
-      .remove_members(account_addresses)
+      .remove_members(&account_addresses)
       .await
       .map_err(ErrorWrapper::from)?;
 
@@ -412,7 +412,7 @@ impl NapiGroup {
     );
 
     group
-      .remove_members_by_inbox_id(inbox_ids)
+      .remove_members_by_inbox_id(&inbox_ids)
       .await
       .map_err(ErrorWrapper::from)?;
 

bindings_node/src/mls_client.rs

@@ -13,6 +13,7 @@ pub use xmtp_api_grpc::grpc_api_helper::Client as TonicApiClient;
 use xmtp_cryptography::signature::ed25519_public_key_to_address;
 use xmtp_id::associations::builder::SignatureRequest;
 use xmtp_mls::builder::ClientBuilder;
+use xmtp_mls::groups::scoped_client::LocalScopedGroupClient;
 use xmtp_mls::identity::IdentityStrategy;
 use xmtp_mls::storage::{EncryptedMessageStore, EncryptionKey, StorageOption};
 use xmtp_mls::Client as MlsClient;
@@ -140,7 +141,7 @@ impl NapiClient {
   pub async fn can_message(&self, account_addresses: Vec<String>) -> Result<HashMap<String, bool>> {
     let results: HashMap<String, bool> = self
       .inner_client
-      .can_message(account_addresses)
+      .can_message(&account_addresses)
       .await
       .map_err(ErrorWrapper::from)?;
 
@@ -179,11 +180,15 @@ impl NapiClient {
 
   #[napi]
   pub async fn request_history_sync(&self) -> Result<()> {
+    let provider = self
+      .inner_client
+      .mls_provider()
+      .map_err(ErrorWrapper::from)?;
     let _ = self
       .inner_client
-      .send_history_request()
+      .send_history_sync_request(&provider)
       .await
-      .map_err(ErrorWrapper::from);
+      .map_err(ErrorWrapper::from)?;
 
     Ok(())
   }

bindings_wasm/Cargo.toml

@@ -18,18 +18,9 @@ wasm-bindgen.workspace = true
 xmtp_api_http = { path = "../xmtp_api_http" }
 xmtp_cryptography = { path = "../xmtp_cryptography" }
 xmtp_id = { path = "../xmtp_id" }
-xmtp_mls = { path = "../xmtp_mls", features = [
-  "message-history",
-  "test-utils",
-  "http-api"
-] }
+xmtp_mls = { path = "../xmtp_mls", features = ["test-utils", "http-api"] }
 xmtp_proto = { path = "../xmtp_proto", features = ["proto_full"] }
 
 [dev-dependencies]
 wasm-bindgen-test.workspace = true
-xmtp_mls = { path = "../xmtp_mls", features = [
-  "message-history",
-  "test-utils",
-  "http-api"
-] }
-
+xmtp_mls = { path = "../xmtp_mls", features = ["test-utils", "http-api"] }

bindings_wasm/src/conversations.rs

@@ -122,7 +122,7 @@ impl WasmConversations {
     } else {
       self
         .inner_client
-        .create_group_with_members(account_addresses, group_permissions, metadata_options)
+        .create_group_with_members(&account_addresses, group_permissions, metadata_options)
         .await
         .map_err(|e| JsError::new(format!("{}", e).as_str()))?
     };
@@ -167,9 +167,14 @@ impl WasmConversations {
 
   #[wasm_bindgen]
   pub async fn sync(&self) -> Result<(), JsError> {
+    let conn = self
+      .inner_client
+      .store()
+      .conn()
+      .map_err(|e| JsError::new(format!("{}", e).as_str()))?;
     self
       .inner_client
-      .sync_welcomes()
+      .sync_welcomes(&conn)
       .await
       .map_err(|e| JsError::new(format!("{}", e).as_str()))?;
     Ok(())

bindings_wasm/src/groups.rs

@@ -296,7 +296,7 @@ impl WasmGroup {
     );
 
     group
-      .add_members(account_addresses)
+      .add_members(&account_addresses)
       .await
       .map_err(|e| JsError::new(&format!("{e}")))?;
 
@@ -387,7 +387,7 @@ impl WasmGroup {
     );
 
     group
-      .add_members_by_inbox_id(inbox_ids)
+      .add_members_by_inbox_id(&inbox_ids)
       .await
       .map_err(|e| JsError::new(&format!("{e}")))?;
 
@@ -403,7 +403,7 @@ impl WasmGroup {
     );
 
     group
-      .remove_members(account_addresses)
+      .remove_members(&account_addresses)
       .await
       .map_err(|e| JsError::new(&format!("{e}")))?;
 
@@ -419,7 +419,7 @@ impl WasmGroup {
     );
 
     group
-      .remove_members_by_inbox_id(inbox_ids)
+      .remove_members_by_inbox_id(&inbox_ids)
       .await
       .map_err(|e| JsError::new(&format!("{e}")))?;