fix http stream with stream all messages

.github/workflows/test-webassembly.yml

@@ -39,8 +39,7 @@ jobs:
       - name: test with chrome
         run: |
           cargo test --release --target wasm32-unknown-unknown -p xmtp_mls -p xmtp_id -p xmtp_api_http -p xmtp_cryptography -- \
-            --skip xmtp_mls::subscriptions --skip xmtp_mls::groups::subscriptions \
-            --skip xmtp_mls::storage::encrypted_store::group_message::tests::it_cannot_insert_message_without_group \
+             -skip xmtp_mls::storage::encrypted_store::group_message::tests::it_cannot_insert_message_without_group \
             --skip xmtp_mls::groups::tests::process_messages_abort_on_retryable_error \
             --skip xmtp_mls::storage::encrypted_store::group::tests::test_find_groups \
             --skip xmtp_mls::storage::encrypted_store::group::tests::test_installations_last_checked_is_updated

.vscode/settings.json

@@ -10,6 +10,7 @@
   ],
   "rust-analyzer.procMacro.attributes.enable": true,
   "rust-analyzer.procMacro.ignored": {
+    "tracing": ["instrument"],
     "async-trait": ["async_trait"],
     "napi-derive": ["napi"],
     "async-recursion": ["async_recursion"],

Cargo.toml

@@ -100,6 +100,8 @@ wasm-bindgen-test = "0.3.50"
 web-sys = "0.3"
 zeroize = "1.8"
 pin-project-lite = "0.2"
+reqwest = { version = "0.12.12", features = ["json", "stream"] }
+bytes = "1.9"
 
 # Internal Crate Dependencies
 xmtp_api_grpc = { path = "xmtp_api_grpc" }
@@ -156,3 +158,6 @@ diesel = { git = "https://github.com/diesel-rs/diesel", branch = "master" }
 diesel_derives = { git = "https://github.com/diesel-rs/diesel", branch = "master" }
 diesel_migrations = { git = "https://github.com/diesel-rs/diesel", branch = "master" }
 sqlite-web = { git = "https://github.com/xmtp/sqlite-web-rs", branch = "main" }
+reqwest = { path = "/Users/insipx/Projects/seanmonster/reqwest" }
+wasm-streams = { path = "/Users/insipx/Projects/wasm-streams" }
+

bindings_ffi/src/lib.rs

@@ -56,6 +56,8 @@ pub enum GenericError {
     DeviceSync(#[from] xmtp_mls::groups::device_sync::DeviceSyncError),
     #[error(transparent)]
     Identity(#[from] xmtp_mls::identity::IdentityError),
+    #[error(transparent)]
+    Subscription(#[from] xmtp_mls::subscriptions::SubscribeError),
 }
 
 #[derive(uniffi::Error, thiserror::Error, Debug)]

bindings_ffi/src/mls.rs

@@ -27,7 +27,7 @@ use xmtp_mls::storage::group_message::{SortDirection, StoredGroupMessageWithReac
 use xmtp_mls::{
     api::ApiClientWrapper,
     builder::ClientBuilder,
-    client::{Client as MlsClient, ClientError},
+    client::Client as MlsClient,
     groups::{
         group_metadata::GroupMetadata,
         group_mutable_metadata::MetadataField,
@@ -47,6 +47,7 @@ use xmtp_mls::{
         group_message::{DeliveryStatus, GroupMessageKind, StoredGroupMessage},
         EncryptedMessageStore, EncryptionKey, StorageOption,
     },
+    subscriptions::SubscribeError,
     AbortHandle, GenericStreamHandle, StreamHandle,
 };
 use xmtp_proto::xmtp::mls::message_contents::content_types::ReactionV2;
@@ -1770,15 +1771,13 @@ impl FfiConversation {
     }
 
     pub async fn stream(&self, message_callback: Arc<dyn FfiMessageCallback>) -> FfiStreamCloser {
-        let handle = MlsGroup::stream_with_callback(
-            self.inner.client.clone(),
-            self.id(),
-            self.inner.created_at_ns,
-            move |message| match message {
-                Ok(m) => message_callback.on_message(m.into()),
-                Err(e) => message_callback.on_error(e.into()),
-            },
-        );
+        let handle =
+            MlsGroup::stream_with_callback(self.inner.client.clone(), self.id(), move |message| {
+                match message {
+                    Ok(m) => message_callback.on_message(m.into()),
+                    Err(e) => message_callback.on_error(e.into()),
+                }
+            });
 
         FfiStreamCloser::new(handle)
     }
@@ -2065,7 +2064,7 @@ impl From<FfiConsent> for StoredConsentRecord {
     }
 }
 
-type FfiHandle = Box<GenericStreamHandle<Result<(), ClientError>>>;
+type FfiHandle = Box<GenericStreamHandle<Result<(), SubscribeError>>>;
 
 #[derive(uniffi::Object, Clone)]
 pub struct FfiStreamCloser {
@@ -2076,7 +2075,10 @@ pub struct FfiStreamCloser {
 
 impl FfiStreamCloser {
     pub fn new(
-        stream_handle: impl StreamHandle<StreamOutput = Result<(), ClientError>> + Send + Sync + 'static,
+        stream_handle: impl StreamHandle<StreamOutput = Result<(), SubscribeError>>
+            + Send
+            + Sync
+            + 'static,
     ) -> Self {
         Self {
             abort_handle: Arc::new(stream_handle.abort_handle()),

bindings_node/src/conversation.rs

@@ -559,7 +559,6 @@ impl Conversation {
     let stream_closer = MlsGroup::stream_with_callback(
       self.inner_client.clone(),
       self.group_id.clone(),
-      self.created_at_ns,
       move |message| {
         tsfn.call(
           message

bindings_node/src/streams.rs

@@ -2,13 +2,13 @@ use napi::bindgen_prelude::Error;
 use std::sync::Arc;
 use tokio::sync::Mutex;
 use xmtp_mls::{
-  client::ClientError, AbortHandle, GenericStreamHandle, StreamHandle as XmtpStreamHandle,
-  StreamHandleError,
+  subscriptions::SubscribeError, AbortHandle, GenericStreamHandle,
+  StreamHandle as XmtpStreamHandle, StreamHandleError,
 };
 
 use napi_derive::napi;
 
-type StreamHandle = Box<GenericStreamHandle<Result<(), ClientError>>>;
+type StreamHandle = Box<GenericStreamHandle<Result<(), SubscribeError>>>;
 
 #[napi]
 pub struct StreamCloser {
@@ -18,7 +18,7 @@ pub struct StreamCloser {
 
 impl StreamCloser {
   pub fn new(
-    handle: impl XmtpStreamHandle<StreamOutput = Result<(), ClientError>> + Send + Sync + 'static,
+    handle: impl XmtpStreamHandle<StreamOutput = Result<(), SubscribeError>> + Send + Sync + 'static,
   ) -> Self {
     let abort = handle.abort_handle();
     Self {

common/Cargo.toml

@@ -25,6 +25,7 @@ console_error_panic_hook = { version =  "0.1", optional = true }
 js-sys.workspace = true
 web-sys = { workspace = true, features = ["Window"] }
 wasm-bindgen-futures.workspace = true
+wasm-bindgen.workspace = true
 
 [dev-dependencies]
 thiserror.workspace = true

common/src/lib.rs

@@ -13,11 +13,8 @@ pub mod bench;
 pub mod retry;
 pub use retry::*;
 
-/// Global Marker trait for WebAssembly
-#[cfg(target_arch = "wasm32")]
-pub trait Wasm {}
-#[cfg(target_arch = "wasm32")]
-impl<T> Wasm for T {}
+pub mod wasm;
+pub use wasm::*;
 
 pub mod time;
 
@@ -36,15 +33,3 @@ pub fn rand_array<const N: usize>() -> [u8; N] {
     crypto_utils::rng().fill_bytes(&mut buffer);
     buffer
 }
-
-/// Yield back control to the async runtime
-#[cfg(not(target_arch = "wasm32"))]
-pub async fn yield_() {
-    tokio::task::yield_now().await
-}
-
-/// Yield back control to the async runtime
-#[cfg(target_arch = "wasm32")]
-pub async fn yield_() {
-    time::sleep(crate::time::Duration::from_millis(1)).await;
-}

common/src/test.rs

@@ -73,9 +73,9 @@ pub fn logger() {
     use tracing_subscriber::EnvFilter;
 
     INIT.get_or_init(|| {
-        let filter = EnvFilter::builder()
-            .with_default_directive(tracing::metadata::LevelFilter::DEBUG.into())
-            .from_env_lossy();
+        let filter = EnvFilter::builder().parse_lossy("xmtp_mls::subscriptions=debug");
+        // .parse_lossy("xmtp_mls::subscriptions=TRACE,xmtp_api_http=TRACE,xmtp_common=TRACE,wasm_streams=TRACE,reqwest=TRACE");
+        // .with_default_directive(tracing::metadata::LevelFilter::TRACE.into());
 
         tracing_subscriber::registry()
             .with(tracing_wasm::WASMLayer::default())

common/src/time.rs

@@ -80,7 +80,6 @@ pub async fn sleep(duration: Duration) {
         let worker = js_sys::global()
             .dyn_into::<WorkerGlobalScope>()
             .expect("xmtp_mls should always act in worker in browser");
-
         worker
             .set_timeout_with_callback_and_timeout_and_arguments_0(
                 &resolve,