Simplify external API interface (#206)

1. Make all methods of `Conversations` static, and accept a `Client` argument, so that `Conversations` doesn't need to be instantiated anywhere. `Conversations` doesn't really hold state anyway and effectively functions as a wrapper for `Client` right now, so it doesn't make sense for consumers to have to instantiate it and hold onto it.
2. Call `process_outbound_messages()` on init and on message send, so that consumers don't have to manually call it.
3. Limit the connection pool size to 1 for the ephemeral DB instance, because when multiple connections are made, each one operates on a separate database. In the event that a connection is requested from the pool while another one is ongoing, it will wait for a time limit before expiring with an error. An added plus of this is it's much easier for tests to detect deadlocks.

This simplifies the client API quite a bit, as can be seen in `bindings_ffi` and in the `CLI`.

Ideally, we could have spawned a separate thread to call `process_outbound_messages` so that the existing thread is not blocked, but that requires additional refactoring. This requires us to enforce `XmtpApiClient` is `Send + Sync` (which is easy), but also comes up against issues with lifetimes for `Client`, for which a reference is needed in the other thread that may outlive the current thread. This possibly requires us to start using `Arcs` again.

Author: richardhuaaa

bindings_ffi/src/lib.rs

@@ -6,7 +6,8 @@ use log::info;
 use logger::FfiLogger;
 use std::error::Error;
 use std::sync::Arc;
-use xmtp::conversation::ListMessagesOptions;
+use xmtp::conversation::{ListMessagesOptions, SecretConversation};
+use xmtp::conversations::Conversations;
 use xmtp::storage::StoredMessage;
 use xmtp::types::Address;
 use xmtp_networking::grpc_api_helper::Client as TonicApiClient;
@@ -111,9 +112,7 @@ impl FfiConversations {
         &self,
         wallet_address: String,
     ) -> Result<Arc<FfiConversation>, GenericError> {
-        let conversations = xmtp::conversations::Conversations::new(self.inner_client.as_ref());
-        let convo = conversations
-            .new_secret_conversation(wallet_address)
+        let convo = SecretConversation::new(self.inner_client.as_ref(), wallet_address)
             .map_err(|e| e.to_string())?;
         // TODO: This should happen as part of `new_secret_conversation` and should only send to new participants
         convo.initialize().await.map_err(|e| e.to_string())?;
@@ -128,8 +127,9 @@ impl FfiConversations {
 
     pub async fn list(&self) -> Result<Vec<Arc<FfiConversation>>, GenericError> {
         let inner = self.inner_client.as_ref();
-        let conversations = xmtp::conversations::Conversations::new(inner);
-        let convo_list = conversations.list(true).await.map_err(|e| e.to_string())?;
+        let convo_list = Conversations::list(inner, true)
+            .await
+            .map_err(|e| e.to_string())?;
         let out: Vec<Arc<FfiConversation>> = convo_list
             .into_iter()
             .map(|convo| {
@@ -173,15 +173,10 @@ impl FfiConversation {
         let conversation = xmtp::conversation::SecretConversation::new(
             self.inner_client.as_ref(),
             self.peer_address.clone(),
-        );
-        let conversations = xmtp::conversations::Conversations::new(self.inner_client.as_ref());
-
+        )
+        .map_err(|e| e.to_string())?;
         conversation
             .send(content_bytes)
-            .map_err(|e| e.to_string())?;
-
-        conversations
-            .process_outbound_messages()
             .await
             .map_err(|e| e.to_string())?;
 
@@ -195,7 +190,8 @@ impl FfiConversation {
         let conversation = xmtp::conversation::SecretConversation::new(
             self.inner_client.as_ref(),
             self.peer_address.clone(),
-        );
+        )
+        .map_err(|e| e.to_string())?;
         let options: ListMessagesOptions = opts.to_options();
 
         let messages: Vec<Arc<FfiMessage>> = conversation

examples/cli/cli-client.rs

@@ -160,8 +160,7 @@ async fn main() {
                 .unwrap();
 
             recv(&client).await.unwrap();
-            let conversations = Conversations::new(&client);
-            let convo_list = conversations.list(true).await.unwrap();
+            let convo_list = Conversations::list(&client, true).await.unwrap();
 
             for (index, convo) in convo_list.iter().enumerate() {
                 info!(
@@ -253,23 +252,16 @@ async fn register(cli: &Cli, use_local_db: bool, wallet_seed: &u64) -> Result<()
 }
 
 async fn send(client: Client, addr: &str, msg: &String) -> Result<(), CliError> {
-    let conversations = Conversations::new(&client);
-    let conversation = conversations
-        .new_secret_conversation(addr.to_string())
-        .unwrap();
+    let conversation = SecretConversation::new(&client, addr.to_string()).unwrap();
     conversation.initialize().await.unwrap();
-    conversation.send_text(msg).unwrap();
-    conversations.process_outbound_messages().await.unwrap();
+    conversation.send_text(msg).await.unwrap();
     info!("Message successfully sent");
 
     Ok(())
 }
 
 async fn recv(client: &Client) -> Result<(), CliError> {
-    let conversations = Conversations::new(client);
-    conversations.save_inbound_messages()?;
-    conversations.process_inbound_messages()?;
-
+    Conversations::receive(client)?;
     Ok(())
 }
 

xmtp/src/client.rs

@@ -9,6 +9,7 @@ use vodozemac::olm::PreKeyMessage;
 use crate::{
     account::Account,
     contact::{Contact, ContactError},
+    conversations::Conversations,
     session::SessionManager,
     storage::{
         now, DbConnection, EncryptedMessageStore, StorageError, StoredInstallation, StoredSession,
@@ -119,6 +120,12 @@ where
         }
 
         self.is_initialized = true;
+
+        // Send any unsent messages
+        if let Err(err) = Conversations::process_outbound_messages(&self).await {
+            log::error!("Could not process outbound messages on init: {:?}", err)
+        }
+
         Ok(())
     }
 

xmtp/src/conversation.rs

@@ -2,6 +2,7 @@ use crate::{
     client::ClientError,
     codecs::{text::TextCodec, CodecError, ContentCodec},
     contact::Contact,
+    conversations::Conversations,
     invitation::{Invitation, InvitationError},
     message::PayloadError,
     session::SessionError,
@@ -103,19 +104,12 @@ impl<'c, A> SecretConversation<'c, A>
 where
     A: XmtpApiClient,
 {
-    pub fn new(client: &'c Client<A>, peer_address: Address) -> Self {
-        Self {
+    // Instantiate the conversation and insert all the necessary records into the database
+    pub fn new(client: &'c Client<A>, peer_address: Address) -> Result<Self, ConversationError> {
+        let obj = Self {
             client,
             peer_address,
-        }
-    }
-
-    // Instantiate the conversation and insert all the necessary records into the database
-    pub(crate) fn create(
-        client: &'c Client<A>,
-        peer_address: Address,
-    ) -> Result<Self, ConversationError> {
-        let obj = Self::new(client, peer_address);
+        };
         let conn = &mut client.store.conn()?;
 
         obj.client.store.insert_or_ignore_user_with_conn(
@@ -148,7 +142,7 @@ where
         self.peer_address.clone()
     }
 
-    pub fn send(&self, content_bytes: Vec<u8>) -> Result<(), ConversationError> {
+    pub async fn send(&self, content_bytes: Vec<u8>) -> Result<(), ConversationError> {
         NewStoredMessage::new(
             self.convo_id(),
             self.client.account.addr(),
@@ -158,15 +152,19 @@ where
         )
         .store(&mut self.client.store.conn().unwrap())?;
 
+        if let Err(err) = Conversations::process_outbound_messages(&self.client).await {
+            log::error!("Could not process outbound messages on init: {:?}", err)
+        }
+
         Ok(())
     }
 
-    pub fn send_text(&self, text: &str) -> Result<(), ConversationError> {
+    pub async fn send_text(&self, text: &str) -> Result<(), ConversationError> {
         // TODO support other codecs
         let encoded_content = TextCodec::encode(text.to_string())?;
         let content_bytes = encoded_content.encode_to_vec();
 
-        self.send(content_bytes)
+        self.send(content_bytes).await
     }
 
     pub async fn list_messages(
@@ -247,10 +245,7 @@ mod tests {
         codecs::{text::TextCodec, ContentCodec},
         conversation::ListMessagesOptions,
         conversations::Conversations,
-        mock_xmtp_api_client::MockXmtpApiClient,
-        test_utils::test_utils::{
-            gen_test_client, gen_test_client_internal, gen_test_conversation,
-        },
+        test_utils::test_utils::{gen_test_client, gen_test_conversation, gen_two_test_clients},
     };
 
     #[tokio::test]
@@ -260,18 +255,16 @@ mod tests {
         let convo_id = format!(":{}:{}", peer_address, client.wallet_address());
         assert!(client.store.get_conversation(&convo_id).unwrap().is_none());
 
-        let conversations = Conversations::new(&client);
-        let conversation = gen_test_conversation(&conversations, peer_address).await;
+        let conversation = gen_test_conversation(&client, peer_address).await;
         assert!(conversation.peer_address() == peer_address);
         assert!(client.store.get_conversation(&convo_id).unwrap().is_some());
     }
 
     #[tokio::test]
     async fn test_send_text() {
         let client = gen_test_client().await;
-        let conversations = Conversations::new(&client);
-        let conversation = gen_test_conversation(&conversations, "0x000").await;
-        conversation.send_text("Hello, world!").unwrap();
+        let conversation = gen_test_conversation(&client, "0x000").await;
+        conversation.send_text("Hello, world!").await.unwrap();
 
         let message = &client.store.get_unprocessed_messages().unwrap()[0];
         let content = EncodedContent::decode(&message.content[..]).unwrap();
@@ -280,17 +273,15 @@ mod tests {
 
     #[tokio::test]
     async fn test_list_messages() {
-        let api_client = MockXmtpApiClient::new();
-        let client = gen_test_client_internal(api_client.clone()).await;
-        let recipient = gen_test_client_internal(api_client.clone()).await;
-        let conversations = Conversations::new(&client);
-        let conversation =
-            gen_test_conversation(&conversations, recipient.account.addr().as_str()).await;
+        let (client, recipient) = gen_two_test_clients().await;
+        let conversation = gen_test_conversation(&client, recipient.account.addr().as_str()).await;
         conversation.initialize().await.unwrap();
-        conversation.send_text("Hello, world!").unwrap();
-        conversation.send_text("Hello, again").unwrap();
+        conversation.send_text("Hello, world!").await.unwrap();
+        conversation.send_text("Hello, again").await.unwrap();
 
-        conversations.process_outbound_messages().await.unwrap();
+        Conversations::process_outbound_messages(&client)
+            .await
+            .unwrap();
 
         let results = conversation
             .list_messages(&ListMessagesOptions::default())