only queue intent inside of a transaction

bindings_ffi/src/mls.rs

@@ -881,8 +881,8 @@ impl FfiConversations {
 
     pub async fn sync(&self) -> Result<(), GenericError> {
         let inner = self.inner_client.as_ref();
-        let conn = inner.store().conn()?;
-        inner.sync_welcomes(&conn).await?;
+        let provider = inner.mls_provider()?;
+        inner.sync_welcomes(&provider).await?;
         Ok(())
     }
 
@@ -895,9 +895,9 @@ impl FfiConversations {
 
     pub async fn sync_all_conversations(&self) -> Result<u32, GenericError> {
         let inner = self.inner_client.as_ref();
-        let conn = inner.store().conn()?;
+        let provider = inner.mls_provider()?;
 
-        let num_groups_synced: usize = inner.sync_all_welcomes_and_groups(&conn).await?;
+        let num_groups_synced: usize = inner.sync_all_welcomes_and_groups(&provider).await?;
 
         // Convert usize to u32 for compatibility with Uniffi
         let num_groups_synced: u32 = num_groups_synced
@@ -1848,6 +1848,8 @@ mod tests {
         conversations: Mutex<Vec<Arc<FfiConversation>>>,
         consent_updates: Mutex<Vec<FfiConsent>>,
         notify: Notify,
+        inbox_id: Option<String>,
+        installation_id: Option<String>,
     }
 
     impl RustStreamCallback {
@@ -1867,12 +1869,22 @@ mod tests {
             .await?;
             Ok(())
         }
+
+        pub fn from_client(client: &FfiXmtpClient) -> Self {
+            RustStreamCallback {
+                inbox_id: Some(client.inner_client.inbox_id().to_string()),
+                installation_id: Some(hex::encode(client.inner_client.installation_public_key())),
+                ..Default::default()
+            }
+        }
     }
 
     impl FfiMessageCallback for RustStreamCallback {
         fn on_message(&self, message: FfiMessage) {
             let mut messages = self.messages.lock().unwrap();
             log::info!(
+                inbox_id = self.inbox_id,
+                installation_id = self.installation_id,
                 "ON MESSAGE Received\n-------- \n{}\n----------",
                 String::from_utf8_lossy(&message.content)
             );
@@ -1888,7 +1900,11 @@ mod tests {
 
     impl FfiConversationCallback for RustStreamCallback {
         fn on_conversation(&self, group: Arc<super::FfiConversation>) {
-            log::debug!("received conversation");
+            log::debug!(
+                inbox_id = self.inbox_id,
+                installation_id = self.installation_id,
+                "received conversation"
+            );
             let _ = self.num_messages.fetch_add(1, Ordering::SeqCst);
             let mut convos = self.conversations.lock().unwrap();
             convos.push(group);
@@ -1902,7 +1918,11 @@ mod tests {
 
     impl FfiConsentCallback for RustStreamCallback {
         fn on_consent_update(&self, mut consent: Vec<FfiConsent>) {
-            log::debug!("received consent update");
+            log::debug!(
+                inbox_id = self.inbox_id,
+                installation_id = self.installation_id,
+                "received consent update"
+            );
             let mut consent_updates = self.consent_updates.lock().unwrap();
             consent_updates.append(&mut consent);
             self.notify.notify_one();
@@ -2751,7 +2771,7 @@ mod tests {
         let caro = new_test_client().await;
 
         // Alix begins a stream for all messages
-        let message_callbacks = Arc::new(RustStreamCallback::default());
+        let message_callbacks = Arc::new(RustStreamCallback::from_client(&alix));
         let stream_messages = alix
             .conversations()
             .stream_all_messages(message_callbacks.clone())
@@ -2798,12 +2818,12 @@ mod tests {
         let bo2 = new_test_client_with_wallet(bo_wallet).await;
 
         // Bo begins a stream for all messages
-        let bo_message_callbacks = Arc::new(RustStreamCallback::default());
-        let bo_stream_messages = bo2
+        let bo2_message_callbacks = Arc::new(RustStreamCallback::from_client(&bo2));
+        let bo2_stream_messages = bo2
             .conversations()
-            .stream_all_messages(bo_message_callbacks.clone())
+            .stream_all_messages(bo2_message_callbacks.clone())
             .await;
-        bo_stream_messages.wait_for_ready().await;
+        bo2_stream_messages.wait_for_ready().await;
 
         alix_group.update_installations().await.unwrap();
 
@@ -3167,7 +3187,7 @@ mod tests {
         let bo = new_test_client().await;
         let caro = new_test_client().await;
 
-        let caro_conn = caro.inner_client.store().conn().unwrap();
+        let caro_provider = caro.inner_client.mls_provider().unwrap();
 
         let alix_group = alix
             .conversations()
@@ -3197,7 +3217,11 @@ mod tests {
             )
             .await
             .unwrap();
-        let _ = caro.inner_client.sync_welcomes(&caro_conn).await.unwrap();
+        let _ = caro
+            .inner_client
+            .sync_welcomes(&caro_provider)
+            .await
+            .unwrap();
 
         bo_group.send("second".as_bytes().to_vec()).await.unwrap();
         stream_callback.wait_for_delivery(None).await.unwrap();
@@ -3216,7 +3240,7 @@ mod tests {
         let amal = new_test_client().await;
         let bola = new_test_client().await;
 
-        let bola_conn = bola.inner_client.store().conn().unwrap();
+        let bola_provider = bola.inner_client.mls_provider().unwrap();
 
         let amal_group: Arc<FfiConversation> = amal
             .conversations()
@@ -3227,7 +3251,10 @@ mod tests {
             .await
             .unwrap();
 
-        bola.inner_client.sync_welcomes(&bola_conn).await.unwrap();
+        bola.inner_client
+            .sync_welcomes(&bola_provider)
+            .await
+            .unwrap();
         let bola_group = bola.conversation(amal_group.id()).unwrap();
 
         let stream_callback = Arc::new(RustStreamCallback::default());

bindings_node/src/conversations.rs

@@ -235,30 +235,28 @@ impl Conversations {
 
   #[napi]
   pub async fn sync(&self) -> Result<()> {
-    let conn = self
+    let provider = self
       .inner_client
-      .store()
-      .conn()
+      .mls_provider()
       .map_err(ErrorWrapper::from)?;
     self
       .inner_client
-      .sync_welcomes(&conn)
+      .sync_welcomes(&provider)
       .await
       .map_err(ErrorWrapper::from)?;
     Ok(())
   }
 
   #[napi]
   pub async fn sync_all_conversations(&self) -> Result<usize> {
-    let conn = self
+    let provider = self
       .inner_client
-      .store()
-      .conn()
+      .mls_provider()
       .map_err(ErrorWrapper::from)?;
 
     let num_groups_synced = self
       .inner_client
-      .sync_all_welcomes_and_groups(&conn)
+      .sync_all_welcomes_and_groups(&provider)
       .await
       .map_err(ErrorWrapper::from)?;
 

bindings_wasm/src/conversations.rs

@@ -269,14 +269,13 @@ impl Conversations {
 
   #[wasm_bindgen]
   pub async fn sync(&self) -> Result<(), JsError> {
-    let conn = self
+    let provider = self
       .inner_client
-      .store()
-      .conn()
+      .mls_provider()
       .map_err(|e| JsError::new(format!("{}", e).as_str()))?;
     self
       .inner_client
-      .sync_welcomes(&conn)
+      .sync_welcomes(&provider)
       .await
       .map_err(|e| JsError::new(format!("{}", e).as_str()))?;
 
@@ -285,15 +284,14 @@ impl Conversations {
 
   #[wasm_bindgen(js_name = syncAllConversations)]
   pub async fn sync_all_conversations(&self) -> Result<usize, JsError> {
-    let conn = self
+    let provider = self
       .inner_client
-      .store()
-      .conn()
+      .mls_provider()
       .map_err(|e| JsError::new(format!("{}", e).as_str()))?;
 
     let num_groups_synced = self
       .inner_client
-      .sync_all_welcomes_and_groups(&conn)
+      .sync_all_welcomes_and_groups(&provider)
       .await
       .map_err(|e| JsError::new(format!("{}", e).as_str()))?;
 

examples/cli/cli-client.rs

@@ -290,9 +290,9 @@ async fn main() -> color_eyre::eyre::Result<()> {
         }
         Commands::ListGroups {} => {
             info!("List Groups");
-            let conn = client.store().conn()?;
+            let provider = client.mls_provider()?;
             client
-                .sync_welcomes(&conn)
+                .sync_welcomes(&provider)
                 .await
                 .expect("failed to sync welcomes");
 
@@ -440,9 +440,8 @@ async fn main() -> color_eyre::eyre::Result<()> {
             );
         }
         Commands::RequestHistorySync {} => {
-            let conn = client.store().conn().unwrap();
             let provider = client.mls_provider().unwrap();
-            client.sync_welcomes(&conn).await.unwrap();
+            client.sync_welcomes(&provider).await.unwrap();
             client.start_sync_worker();
             client
                 .send_sync_request(&provider, DeviceSyncKind::MessageHistory)
@@ -451,9 +450,9 @@ async fn main() -> color_eyre::eyre::Result<()> {
             info!("Sent history sync request in sync group.")
         }
         Commands::ListHistorySyncMessages {} => {
-            let conn = client.store().conn()?;
-            client.sync_welcomes(&conn).await?;
-            let group = client.get_sync_group(&conn)?;
+            let provider = client.mls_provider()?;
+            client.sync_welcomes(&provider).await?;
+            let group = client.get_sync_group(provider.conn_ref())?;
             let group_id_str = hex::encode(group.group_id.clone());
             group.sync().await?;
             let messages = group
@@ -574,8 +573,8 @@ where
 }
 
 async fn get_group(client: &Client, group_id: Vec<u8>) -> Result<MlsGroup, CliError> {
-    let conn = client.store().conn().unwrap();
-    client.sync_welcomes(&conn).await?;
+    let provider = client.mls_provider().unwrap();
+    client.sync_welcomes(&provider).await?;
     let group = client.group(group_id)?;
     group
         .sync()

xmtp_debug/src/app/generate/messages.rs

@@ -8,7 +8,6 @@ use crate::{
 use color_eyre::eyre::{self, eyre, Result};
 use rand::{rngs::SmallRng, seq::SliceRandom, Rng, SeedableRng};
 use std::sync::Arc;
-use xmtp_mls::XmtpOpenMlsProvider;
 
 mod content_type;
 
@@ -118,10 +117,9 @@ impl GenerateMessages {
                 hex::encode(inbox_id)
             ))?;
             let client = app::client_from_identity(&identity, &network).await?;
-            let conn = client.store().conn()?;
-            client.sync_welcomes(&conn).await?;
+            let provider = client.mls_provider()?;
+            client.sync_welcomes(&provider).await?;
             let group = client.group(group.id.into())?;
-            let provider: XmtpOpenMlsProvider = conn.into();
             group.maybe_update_installations(&provider, None).await?;
             group.sync_with_conn(&provider).await?;
             let words = rng.gen_range(0..*max_message_size);

xmtp_debug/src/app/send.rs

@@ -49,8 +49,8 @@ impl Send {
             .ok_or(eyre!("No Identity with inbox_id [{}]", hex::encode(member)))?;
 
         let client = crate::app::client_from_identity(&identity, network).await?;
-        let conn = client.store().conn()?;
-        client.sync_welcomes(&conn).await?;
+        let provider = client.mls_provider()?;
+        client.sync_welcomes(&provider).await?;
         let xmtp_group = client.group(group.id.to_vec())?;
         xmtp_group.send_message(data.as_bytes()).await?;
         Ok(())

xmtp_mls/src/api/mls.rs

@@ -115,9 +115,9 @@ where
     }
 
     #[tracing::instrument(level = "trace", skip_all)]
-    pub async fn query_welcome_messages(
+    pub async fn query_welcome_messages<Id: AsRef<[u8]> + Copy>(
         &self,
-        installation_id: &[u8],
+        installation_id: Id,
         id_cursor: Option<u64>,
     ) -> Result<Vec<WelcomeMessage>, ApiError> {
         tracing::debug!(
@@ -135,7 +135,7 @@ where
                 (async {
                     self.api_client
                         .query_welcome_messages(QueryWelcomeMessagesRequest {
-                            installation_key: installation_id.to_vec(),
+                            installation_key: installation_id.as_ref().to_vec(),
                             paging_info: Some(PagingInfo {
                                 id_cursor: id_cursor.unwrap_or(0),
                                 limit: page_size,
@@ -297,7 +297,7 @@ where
 
     pub async fn subscribe_welcome_messages(
         &self,
-        installation_key: Vec<u8>,
+        installation_key: &[u8],
         id_cursor: Option<u64>,
     ) -> Result<impl futures::Stream<Item = Result<WelcomeMessage, ApiError>> + '_, ApiError>
     where
@@ -307,7 +307,7 @@ where
         self.api_client
             .subscribe_welcome_messages(SubscribeWelcomeMessagesRequest {
                 filters: vec![WelcomeFilterProto {
-                    installation_key,
+                    installation_key: installation_key.to_vec(),
                     id_cursor: id_cursor.unwrap_or(0),
                 }],
             })

xmtp_mls/src/builder.rs

@@ -187,15 +187,16 @@ where
         .take()
         .ok_or(ClientBuilderError::MissingParameter { parameter: "store" })?;
 
-    debug!(
-        inbox_id = identity_strategy.inbox_id(),
-        "Initializing identity"
-    );
-
     let identity = identity_strategy
         .initialize_identity(&api_client_wrapper, &store, &scw_verifier)
         .await?;
 
+    debug!(
+        inbox_id = identity.inbox_id(),
+        installation_id = hex::encode(identity.installation_keys.public_bytes()),
+        "Initialized identity"
+    );
+
     // get sequence_id from identity updates and loaded into the DB
     load_identity_updates(
         &api_client_wrapper,