I think I want these functions in these places instead

bindings_ffi/src/mls.rs

@@ -220,7 +220,7 @@ impl FfiConversations {
         Ok(out)
     }
 
-    pub fn process_streamed_welcome_message(&self, envelope_bytes: Vec<u8>) -> Result<FfiMessage, GenericError> {
+    pub fn process_streamed_welcome_message(&self, envelope_bytes: Vec<u8>) -> Result<FfiGroup, GenericError> {
         let message = process_streamed_welcome_message(envelope).into();
 
         Ok(message)

xmtp_mls/src/client.rs

@@ -7,7 +7,7 @@ use openmls::{
     prelude::TlsSerializeTrait,
 };
 use openmls_traits::OpenMlsProvider;
-use prost::{EncodeError, Message};
+use prost::EncodeError;
 use thiserror::Error;
 use tls_codec::{Deserialize, Error as TlsSerializationError};
 
@@ -405,16 +405,6 @@ where
             .collect::<Result<_, _>>()?)
     }
 
-    pub fn process_streamed_welcome_message(
-        &self, 
-        envelope_bytes: Vec<u8>
-    ) -> Result<WelcomeMessage, MessageProcessingError> {
-        let envelope = WelcomeMessage::decode(envelope_bytes.as_slice())
-            .map_err(|e| MessageProcessingError::WelcomeProcessing(e.to_string()))?;
-
-        Ok(envelope)
-    }
-
     /// Download all unread welcome messages and convert to groups.
     /// Returns any new groups created in the operation
     pub async fn sync_welcomes(&self) -> Result<Vec<MlsGroup<ApiClient>>, ClientError> {

xmtp_mls/src/groups/mod.rs

@@ -322,16 +322,6 @@ where
         Ok(messages)
     }
 
-    pub fn process_streamed_group_message(
-        &self, 
-        envelope_bytes: Vec<u8>
-    ) -> Result<GroupMessage, GroupError> {
-        let envelope = GroupMessage::decode(envelope_bytes.as_slice())
-            .map_err(|e| GroupError::GroupMessageNotFound(e.to_string()))?;
-
-        Ok(envelope)
-    }
-
     pub async fn add_members(
         &self,
         account_addresses_to_add: Vec<String>,

xmtp_mls/src/groups/subscriptions.rs

@@ -4,6 +4,7 @@ use std::sync::Arc;
 
 use futures::Stream;
 
+use prost::Message;
 use xmtp_proto::{api_client::XmtpMlsClient, xmtp::mls::api::v1::GroupMessage};
 
 use super::{extract_message_v1, GroupError, MlsGroup};
@@ -44,6 +45,17 @@ where
         Ok(new_message)
     }
 
+    pub async fn process_streamed_group_message(
+        &self, 
+        envelope_bytes: Vec<u8>
+    ) -> Result<Option<StoredGroupMessage>, GroupError> {
+        let envelope = GroupMessage::decode(envelope_bytes.as_slice())
+            .map_err(|e| GroupError::GroupMessageNotFound(e.to_string()))?;
+
+        let message = self.process_stream_entry(envelope).await.map_err(|e| e)?;
+        Ok(message)
+    }
+
     pub async fn stream(
         &'c self,
     ) -> Result<Pin<Box<dyn Stream<Item = StoredGroupMessage> + 'c + Send>>, GroupError> {

xmtp_mls/src/subscriptions.rs

@@ -9,6 +9,7 @@ use std::{
 
 use futures::{Stream, StreamExt};
 use tokio::sync::oneshot::{self, Sender};
+use prost::Message;
 use xmtp_proto::{api_client::XmtpMlsClient, xmtp::mls::api::v1::WelcomeMessage};
 
 use crate::{
@@ -69,6 +70,17 @@ where
         .map_err(|e| ClientError::Generic(e.to_string()))
     }
 
+    pub fn process_streamed_welcome_message(
+        &self, 
+        envelope_bytes: Vec<u8>
+    ) -> Result<MlsGroup<ApiClient>, ClientError> {
+        let envelope = WelcomeMessage::decode(envelope_bytes.as_slice())
+            .map_err(|e| ClientError::Generic(e.to_string()))?;
+
+        let welcome = self.process_streamed_welcome(envelope).map_err(|e| e)?;
+        Ok(welcome)
+    }
+
     pub async fn stream_conversations(
         &'a self,
     ) -> Result<Pin<Box<dyn Stream<Item = MlsGroup<ApiClient>> + Send + 'a>>, ClientError> {