Refactor message history sync, add to CLI (#930)

* fix: stubs support for SyncHistory

* fix: send the history_request

* fix: stubs handling the reply as well

* fix: amend request-history-sync command

* wip on the correct placement for the streams

* fix: handle stream out in cli

* fix: stream sync groups

* fix: cleanup wip notes

* fix: cleanup command

* fix: formatting

* fix: remove redundant commented fn call

* fix: remove redundant log

* fix: removes unused fn

* Update Cargo.lock

* Change how message history messages are stored

* Add helper to get sync group

* Only create sync group if not present

* Refactor PIN verification

* Add checks before sending history requests

* Add checks before sending history reply

* Update test_send_history_request test

* Apply patch

* Remove sync groups from stream_all_messages

* Replace bincode with serde_json

* Refactor history message checks

* Update send history reply test

* Add new_request_id test

* Remove bundle_hash and signing_key

* Simplify prepare groups

* Refactor error handling

* Add message history processing

* Make message_history public

* Remove redundant history URL check

* Add official message history URLs

* Use and require history_sync_url

* Fix FFI bindings error type

* Update CLI client for message history commands

* Remove unused fn

* Return group without assignment

* Update CLI README

* Update var name

* Remove history message processing during sync

* Add logging and more message processing

* Add more message history CLI commands

* Update README

* Update history bundle download path

* Fix test

* Prevent processing reply from self

* Add temp file cleanup

* Remove ring dependency

* Only return MlsGroup from get_sync_group

* allow_history_sync => enable_history_sync

* Add serde_json errors, remove map_err

* Expose serde_json errors

* Refactor match

* Update CLI

* Use chained calls

* Fix clippy

* Use log::error

* Add get_pending_history_request test

* Add get_latest_history_reply test

* Add reply_to_history_request test

---------

Co-authored-by: Ry Racherbaumer <ry@xmtp.com>

bindings_ffi/src/mls.rs

@@ -348,7 +348,10 @@ impl FfiXmtpClient {
     }
 
     pub async fn request_history_sync(&self) -> Result<(), GenericError> {
-        self.inner_client.send_history_request().await?;
+        self.inner_client
+            .send_history_request()
+            .await
+            .map_err(GenericError::from_error)?;
         Ok(())
     }
 

examples/cli/Cargo.toml

@@ -31,6 +31,7 @@ serde_json.workspace = true
 thiserror.workspace = true
 timeago = "0.4.1"
 tokio = "1.28.1"
+tokio-stream = "0.1.15"
 url = "2.3.1"
 xmtp_api_grpc = { path = "../../xmtp_api_grpc" }
 xmtp_cryptography = { path = "../../xmtp_cryptography" }

examples/cli/README.md

@@ -55,6 +55,30 @@ Use the CLI to send a [double ratchet message](https://github.com/xmtp/libxmtp/b
    ./xli.sh --db user2.db3 list-group-messages $GROUP_ID
    ```
 
+9. Request a message history sync
+
+   ```bash
+   ./xli.sh --db user1.db3 request-history-sync
+   ```
+
+10. Reply to the history sync request
+
+    ```bash
+    ./xli.sh --db user1.db3 reply-to-history-sync-request
+    ```
+
+11. Process the history sync reply
+
+    ```bash
+    ./xli.sh --db user1.db3 process-history-sync-reply
+    ```
+
+12. List the history sync messages
+
+    ```bash
+    ./xli.sh --db user1.db3 list-history-sync-messages
+    ```
+
 If you want to run the CLI against localhost, go to the root directory and run `dev/up` to start a local server. Then run the CLI commands using the `--local` flag.
 
 ## Structured logging

examples/cli/cli-client.rs

@@ -10,13 +10,16 @@ extern crate ethers;
 extern crate log;
 extern crate xmtp_mls;
 
+use std::iter::Iterator;
 use std::{fs, path::PathBuf, time::Duration};
 
 use clap::{Parser, Subcommand, ValueEnum};
 use ethers::signers::{coins_bip39::English, LocalWallet, MnemonicBuilder};
 use kv_log_macro::{error, info};
 use prost::Message;
 use xmtp_id::associations::RecoverableEcdsaSignature;
+use xmtp_mls::groups::message_history::MessageHistoryContent;
+use xmtp_mls::storage::group_message::GroupMessageKind;
 
 use crate::{
     json_logger::make_value,
@@ -34,7 +37,7 @@ use xmtp_mls::{
     builder::ClientBuilderError,
     client::ClientError,
     codecs::{text::TextCodec, ContentCodec},
-    groups::{GroupMetadataOptions, MlsGroup},
+    groups::{message_history::MessageHistoryUrls, GroupMetadataOptions, MlsGroup},
     identity::IdentityStrategy,
     storage::{
         group_message::StoredGroupMessage, EncryptedMessageStore, EncryptionKey, StorageError,
@@ -108,6 +111,10 @@ enum Commands {
         #[clap(short, long, value_parser, num_args = 1.., value_delimiter = ' ')]
         account_addresses: Vec<String>,
     },
+    RequestHistorySync {},
+    ReplyToHistorySyncRequest {},
+    ProcessHistorySyncReply {},
+    ListHistorySyncMessages {},
     /// Information about the account that owns the DB
     Info {},
     Clear {},
@@ -333,6 +340,68 @@ async fn main() {
             let serializable: SerializableGroup = group.into();
             info!("Group {}", group_id, { command_output: true, group_id: group_id, group_info: make_value(&serializable) })
         }
+        Commands::RequestHistorySync {} => {
+            let client = create_client(&cli, IdentityStrategy::CachedOnly)
+                .await
+                .unwrap();
+            client.sync_welcomes().await.unwrap();
+            client.enable_history_sync().await.unwrap();
+            let (group_id, _) = client.send_history_request().await.unwrap();
+            let group_id_str = hex::encode(group_id);
+            info!("Sent history sync request in sync group {group_id_str}", { group_id: group_id_str})
+        }
+        Commands::ReplyToHistorySyncRequest {} => {
+            let client = create_client(&cli, IdentityStrategy::CachedOnly)
+                .await
+                .unwrap();
+            let group = client.get_sync_group().unwrap();
+            let group_id_str = hex::encode(group.group_id);
+            let reply = client.reply_to_history_request().await.unwrap();
+
+            info!("Sent history sync reply in sync group {group_id_str}", { group_id: group_id_str});
+            info!("Reply: {:?}", reply);
+        }
+        Commands::ProcessHistorySyncReply {} => {
+            let client = create_client(&cli, IdentityStrategy::CachedOnly)
+                .await
+                .unwrap();
+            client.sync_welcomes().await.unwrap();
+            client.enable_history_sync().await.unwrap();
+            client.process_history_reply().await.unwrap();
+
+            info!("History bundle downloaded and inserted into user DB", {})
+        }
+        Commands::ListHistorySyncMessages {} => {
+            let client = create_client(&cli, IdentityStrategy::CachedOnly)
+                .await
+                .unwrap();
+            client.sync_welcomes().await.unwrap();
+            client.enable_history_sync().await.unwrap();
+            let group = client.get_sync_group().unwrap();
+            let group_id_str = hex::encode(group.group_id.clone());
+            group.sync(&client).await.unwrap();
+            let messages = group
+                .find_messages(Some(GroupMessageKind::Application), None, None, None, None)
+                .unwrap();
+            info!("Listing history sync messages", { group_id: group_id_str, messages: messages.len()});
+            for message in messages {
+                let message_history_content = serde_json::from_slice::<MessageHistoryContent>(
+                    &message.decrypted_message_bytes,
+                );
+
+                match message_history_content {
+                    Ok(MessageHistoryContent::Request(ref request)) => {
+                        info!("Request: {:?}", request);
+                    }
+                    Ok(MessageHistoryContent::Reply(ref reply)) => {
+                        info!("Reply: {:?}", reply);
+                    }
+                    _ => {
+                        info!("Unknown message type: {:?}", message);
+                    }
+                }
+            }
+        }
         Commands::Clear {} => {
             fs::remove_file(cli.db.unwrap()).unwrap();
         }
@@ -345,21 +414,27 @@ async fn create_client(cli: &Cli, account: IdentityStrategy) -> Result<Client, C
 
     if cli.local {
         info!("Using local network");
-        builder = builder.api_client(
-            ApiClient::create("http://localhost:5556".into(), false)
-                .await
-                .unwrap(),
-        );
+        builder = builder
+            .api_client(
+                ApiClient::create("http://localhost:5556".into(), false)
+                    .await
+                    .unwrap(),
+            )
+            .history_sync_url(MessageHistoryUrls::LOCAL_ADDRESS);
     } else {
         info!("Using dev network");
-        builder = builder.api_client(
-            ApiClient::create("https://grpc.dev.xmtp.network:443".into(), true)
-                .await
-                .unwrap(),
-        );
+        builder = builder
+            .api_client(
+                ApiClient::create("https://grpc.dev.xmtp.network:443".into(), true)
+                    .await
+                    .unwrap(),
+            )
+            .history_sync_url(MessageHistoryUrls::DEV_ADDRESS);
     }
 
-    builder.build().await.map_err(CliError::ClientBuilder)
+    let client = builder.build().await.map_err(CliError::ClientBuilder)?;
+
+    Ok(client)
 }
 
 async fn register(cli: &Cli, maybe_seed_phrase: Option<String>) -> Result<(), CliError> {

xmtp_mls/Cargo.toml

@@ -51,7 +51,6 @@ parking_lot = "0.12.3"
 prost = { workspace = true, features = ["prost-derive"] }
 rand = { workspace = true }
 reqwest = { version = "0.12.4", features = ["stream"] }
-ring = "0.17.8"
 serde = { workspace = true }
 serde_json.workspace = true
 sha2.workspace = true

xmtp_mls/src/client.rs

@@ -179,6 +179,8 @@ pub enum MessageProcessingError {
     ClearPendingCommit(#[from] sql_key_store::SqlKeyStoreError),
     #[error(transparent)]
     Group(#[from] Box<GroupError>),
+    #[error("Serialization/Deserialization Error {0}")]
+    Serde(#[from] serde_json::Error),
     #[error("generic:{0}")]
     Generic(String),
     #[error("intent is missing staged_commit field")]

xmtp_mls/src/configuration.rs

@@ -24,8 +24,6 @@ pub const MAX_PAST_EPOCHS: usize = 3;
 /// we leave 5 * 1024 * 1024 as extra buffer room
 pub const GRPC_DATA_LIMIT: usize = 45 * 1024 * 1024;
 
-pub const DELIMITER: char = '\x01';
-
 /// MLS Extension Types
 ///
 /// Copied from draft-ietf-mls-protocol-16: