feat(xmtp_api_d14n): XmtpMlsBidiStreams trait + v3 Subscribe transport

crates/xmtp_api_d14n/src/middleware/auth.rs

@@ -181,6 +181,20 @@ impl<C: Client> Client for AuthMiddleware<C> {
         let request = self.modify_request(request).await?;
         self.inner.stream(request, path, body).await
     }
+
+    async fn bidi_stream(
+        &self,
+        request: http::request::Builder,
+        path: http::uri::PathAndQuery,
+        body: xmtp_common::BoxDynStream<'static, Bytes>,
+    ) -> Result<http::Response<BytesStream>, ApiClientError> {
+        // Inject the auth header exactly as `stream`/`request` do — otherwise an
+        // authenticated client's bidi subscription would either skip auth or, if
+        // we forgot to override at all, fall through to the trait's default
+        // "unsupported" error.
+        let request = self.modify_request(request).await?;
+        self.inner.bidi_stream(request, path, body).await
+    }
 }
 
 #[xmtp_common::async_trait]
@@ -273,6 +287,29 @@ mod tests {
                 futures::stream::once(Box::pin(async move { Ok(body) })),
             )))
         }
+
+        async fn bidi_stream(
+            &self,
+            request: http::request::Builder,
+            _path: http::uri::PathAndQuery,
+            _body: xmtp_common::BoxDynStream<'static, Bytes>,
+        ) -> Result<http::Response<BytesStream>, ApiClientError> {
+            // Same header assertion as `stream`: this only passes if the
+            // middleware actually forwarded `bidi_stream` (rather than inheriting
+            // the trait's "unsupported" default) AND injected the auth header.
+            let headers = request.headers_ref().unwrap();
+            if let Some(expected_credential) = &self.expected_credential {
+                assert_eq!(
+                    headers.get(&expected_credential.name).unwrap(),
+                    &expected_credential.value
+                );
+            } else {
+                assert!(headers.is_empty());
+            }
+            Ok(http::Response::new(BytesStream::new(
+                futures::stream::empty(),
+            )))
+        }
     }
 
     impl<C: Client> AuthMiddleware<C> {
@@ -297,7 +334,28 @@ mod tests {
             }
 
             let request = http::request::Builder::new();
-            let result = self.stream(request, path, body).await;
+            let result = self.stream(request, path.clone(), body).await;
+            match (&expected, result) {
+                (Ok(()), Ok(response)) => {
+                    assert_eq!(response.status(), http::StatusCode::OK);
+                }
+                (Err(e), Ok(_)) => {
+                    panic!("Expected error: {e}, got Ok");
+                }
+                (Ok(()), Err(e)) => {
+                    panic!("Expected Ok, got error: {e}");
+                }
+                (Err(e), Err(res)) => {
+                    assert_eq!(e, &res.to_string());
+                }
+            }
+
+            let request = http::request::Builder::new();
+            // `Box::pin` (not `.boxed()`) so this coerces to `BoxDynStream` on
+            // both native (`BoxStream`) and wasm (`LocalBoxStream`) targets.
+            let bidi_body: xmtp_common::BoxDynStream<'static, Bytes> =
+                Box::pin(futures::stream::empty());
+            let result = self.bidi_stream(request, path, bidi_body).await;
             match (&expected, result) {
                 (Ok(()), Ok(response)) => {
                     assert_eq!(response.status(), http::StatusCode::OK);

crates/xmtp_api_d14n/src/middleware/multi_node_client/client.rs

@@ -78,6 +78,17 @@ impl<T: Client> Client for MultiNodeClient<T> {
 
         inner.stream(request, path, body).await
     }
+
+    async fn bidi_stream(
+        &self,
+        request: http::request::Builder,
+        path: http::uri::PathAndQuery,
+        body: xmtp_common::BoxDynStream<'static, Bytes>,
+    ) -> Result<http::Response<BytesStream>, ApiClientError> {
+        let inner = self.init_inner().await?;
+
+        inner.bidi_stream(request, path, body).await
+    }
 }
 
 #[xmtp_common::async_trait]

crates/xmtp_api_d14n/src/middleware/read_write_client/client.rs

@@ -62,6 +62,19 @@ where
             self.read.stream(request, path, body).await
         }
     }
+
+    async fn bidi_stream(
+        &self,
+        request: http::request::Builder,
+        path: http::uri::PathAndQuery,
+        body: xmtp_common::BoxDynStream<'static, Bytes>,
+    ) -> Result<http::Response<BytesStream>, ApiClientError> {
+        if path.path().contains(&self.filter) {
+            self.write.bidi_stream(request, path, body).await
+        } else {
+            self.read.bidi_stream(request, path, body).await
+        }
+    }
 }
 
 #[xmtp_common::async_trait]

crates/xmtp_api_d14n/src/middleware/readonly_client.rs

@@ -66,6 +66,20 @@ where
 
         self.inner.stream(request, path, body).await
     }
+
+    async fn bidi_stream(
+        &self,
+        request: http::request::Builder,
+        path: http::uri::PathAndQuery,
+        body: xmtp_common::BoxDynStream<'static, Bytes>,
+    ) -> Result<http::Response<BytesStream>, ApiClientError> {
+        let p = path.path();
+        if DENY.iter().any(|d| p.contains(d)) {
+            return Err(ApiClientError::WritesDisabled);
+        }
+
+        self.inner.bidi_stream(request, path, body).await
+    }
 }
 
 #[xmtp_common::async_trait]

crates/xmtp_api_d14n/src/queries/v3.rs

@@ -1,3 +1,5 @@
+#[cfg(not(target_arch = "wasm32"))]
+mod bidi;
 mod client;
 mod identity;
 mod mls;

crates/xmtp_api_d14n/src/queries/v3/bidi.rs

@@ -0,0 +1,175 @@
+//! XIP-83 bidirectional Subscribe transport for the v3 backend (native-only).
+
+use crate::V3Client;
+use crate::protocol::CursorStore;
+use futures::StreamExt;
+use prost::Message;
+use prost::bytes::Bytes;
+use xmtp_proto::ApiEndpoint;
+use xmtp_proto::api::{ApiClientError, Client, XmtpStream};
+use xmtp_proto::api_client::XmtpMlsBidiStreams;
+use xmtp_proto::mls_v1::{SubscribeRequest, SubscribeResponse};
+
+const SUBSCRIBE_PATH: &str = "/xmtp.mls.api.v1.MlsApi/Subscribe";
+
+#[xmtp_common::async_trait]
+impl<C, Store> XmtpMlsBidiStreams for V3Client<C, Store>
+where
+    C: Client,
+    Store: CursorStore,
+{
+    type SubscribeStream = XmtpStream<SubscribeResponse>;
+
+    type Error = ApiClientError;
+
+    // Spans the open handshake (not the stream's lifetime) as `rpc.subscribe_bidi`.
+    // Bidi is consumed directly by `xmtp_mls` with no `xmtp_api` wrapper in front,
+    // so this transport impl is the RPC boundary — the same layer the other
+    // `rpc_span`s live at for unary calls. Move it up if a wrapper is ever added.
+    #[xmtp_common::rpc_span]
+    async fn subscribe_bidi(
+        &self,
+        requests: futures::stream::BoxStream<'static, SubscribeRequest>,
+    ) -> Result<Self::SubscribeStream, Self::Error> {
+        tracing::debug!("opening bidirectional subscription");
+        let outbound = requests.map(|frame| Bytes::from(frame.encode_to_vec()));
+        let response = self
+            .client
+            .bidi_stream(
+                http::Request::builder(),
+                http::uri::PathAndQuery::from_static(SUBSCRIBE_PATH),
+                Box::pin(outbound),
+            )
+            .await
+            .map_err(|e| e.endpoint(SUBSCRIBE_PATH.to_string()))?;
+        Ok(XmtpStream::new(
+            response.into_body(),
+            ApiEndpoint::Path(SUBSCRIBE_PATH.to_string()),
+        ))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    #![allow(clippy::unwrap_used)]
+    use super::*;
+    use crate::protocol::NoCursorStore;
+    use futures::stream;
+    use xmtp_common::BoxDynStream;
+    use xmtp_proto::api::BytesStream;
+    use xmtp_proto::api::mock::MockNetworkClient;
+    use xmtp_proto::mls_v1::subscribe_request::v1::Mutate;
+    use xmtp_proto::mls_v1::{Ping, Pong, subscribe_request, subscribe_response};
+
+    fn req(request: subscribe_request::v1::Request) -> SubscribeRequest {
+        SubscribeRequest {
+            version: Some(subscribe_request::Version::V1(subscribe_request::V1 {
+                request: Some(request),
+            })),
+        }
+    }
+
+    fn resp(response: subscribe_response::v1::Response) -> SubscribeResponse {
+        SubscribeResponse {
+            version: Some(subscribe_response::Version::V1(subscribe_response::V1 {
+                response: Some(response),
+            })),
+        }
+    }
+
+    fn ping_req(nonce: u64) -> SubscribeRequest {
+        req(subscribe_request::v1::Request::Ping(Ping { nonce }))
+    }
+
+    /// `subscribe_bidi` must prost-encode each outbound `SubscribeRequest` in
+    /// order, dial the `Subscribe` path, and decode the inbound byte frames back
+    /// into `SubscribeResponse`s through `XmtpStream`.
+    #[xmtp_common::test(unwrap_try = true)]
+    async fn encodes_outbound_and_decodes_inbound() {
+        let captured: std::sync::Arc<std::sync::Mutex<Option<BoxDynStream<'static, Bytes>>>> =
+            Default::default();
+        let sink = captured.clone();
+
+        let mut mock = MockNetworkClient::new();
+        mock.expect_bidi_stream()
+            .return_once(move |_req, path, body| {
+                assert_eq!(path.path(), "/xmtp.mls.api.v1.MlsApi/Subscribe");
+                *sink.lock().unwrap() = Some(body);
+                let frames: Vec<Result<Bytes, ApiClientError>> = vec![
+                    Ok(Bytes::from(
+                        resp(subscribe_response::v1::Response::Ping(Ping { nonce: 7 }))
+                            .encode_to_vec(),
+                    )),
+                    Ok(Bytes::from(
+                        resp(subscribe_response::v1::Response::Pong(Pong { nonce: 9 }))
+                            .encode_to_vec(),
+                    )),
+                ];
+                Ok(http::Response::new(BytesStream::new(stream::iter(frames))))
+            });
+
+        let client = V3Client::new(mock, NoCursorStore);
+        let outbound = stream::iter(vec![
+            req(subscribe_request::v1::Request::Mutate(Mutate::default())),
+            ping_req(3),
+        ])
+        .boxed();
+
+        let inbound = client.subscribe_bidi(outbound).await?;
+        let decoded: Vec<SubscribeResponse> = inbound.map(|r| r.unwrap()).collect().await;
+        assert_eq!(
+            decoded,
+            vec![
+                resp(subscribe_response::v1::Response::Ping(Ping { nonce: 7 })),
+                resp(subscribe_response::v1::Response::Pong(Pong { nonce: 9 })),
+            ],
+        );
+
+        // The outbound stream handed to the transport carries the same requests,
+        // prost-encoded, in order. Take it out of the mutex before awaiting so no
+        // lock is held across the `.await`.
+        let captured_outbound = captured.lock().unwrap().take().unwrap();
+        let sent_bytes: Vec<Bytes> = captured_outbound.collect().await;
+        let sent: Vec<SubscribeRequest> = sent_bytes
+            .iter()
+            .map(|b| SubscribeRequest::decode(b.clone()).unwrap())
+            .collect();
+        assert_eq!(
+            sent,
+            vec![
+                req(subscribe_request::v1::Request::Mutate(Mutate::default())),
+                ping_req(3),
+            ],
+        );
+    }
+
+    /// A transport error opening the stream is surfaced as a `ClientWithEndpoint`
+    /// tagged with the `Subscribe` path, not a bare client error.
+    #[xmtp_common::test(unwrap_try = true)]
+    async fn tags_open_error_with_subscribe_endpoint() {
+        #[derive(Debug, thiserror::Error)]
+        #[error("boom")]
+        struct Boom;
+        impl xmtp_common::RetryableError for Boom {
+            fn is_retryable(&self) -> bool {
+                false
+            }
+        }
+
+        let mut mock = MockNetworkClient::new();
+        mock.expect_bidi_stream()
+            .return_once(|_req, _path, _body| Err(ApiClientError::client(Boom)));
+
+        let client = V3Client::new(mock, NoCursorStore);
+        let outbound = stream::iter(vec![ping_req(1)]).boxed();
+
+        // `XmtpStream` isn't `Debug`, so match instead of `unwrap_err`.
+        match client.subscribe_bidi(outbound).await {
+            Err(ApiClientError::ClientWithEndpoint { endpoint, .. }) => {
+                assert_eq!(endpoint, "/xmtp.mls.api.v1.MlsApi/Subscribe");
+            }
+            Err(other) => panic!("expected ClientWithEndpoint, got {other:?}"),
+            Ok(_) => panic!("subscribe_bidi should error when the transport fails to open"),
+        }
+    }
+}

crates/xmtp_api_grpc/src/grpc_client/client.rs

@@ -80,12 +80,17 @@ impl GrpcClient {
         }
     }
 
-    /// Builds a tonic request from a body and a generic HTTP Request
-    fn build_tonic_request(
+    /// Builds a tonic request from a body and a generic HTTP Request.
+    ///
+    /// Generic over the body type `B` so the same path serves both the
+    /// unary / server-streaming transports (where `B` is `Bytes`) and the
+    /// XIP-83 bidirectional transport (where `B` is a `BoxDynStream` of
+    /// outbound frames). `tonic::Request<B>` is happy with either.
+    fn build_tonic_request<B>(
         &self,
         request: http::request::Builder,
-        body: Bytes,
-    ) -> Result<tonic::Request<Bytes>, Status> {
+        body: B,
+    ) -> Result<tonic::Request<B>, Status> {
         let request = request
             .body(body)
             .map_err(|e| tonic::Status::from_error(Box::new(e)))?;
@@ -187,6 +192,36 @@ impl Client for GrpcClient {
         });
         Ok(response.to_http().map(Into::into))
     }
+
+    // Full-duplex needs a real HTTP/2 transport; the gRPC-Web service used on
+    // wasm cannot carry it, so the browser keeps the trait's default error.
+    #[cfg(not(target_arch = "wasm32"))]
+    async fn bidi_stream(
+        &self,
+        request: request::Builder,
+        path: PathAndQuery,
+        body: xmtp_common::BoxDynStream<'static, Bytes>,
+    ) -> Result<http::Response<BytesStream>, ApiClientError> {
+        let this = self.clone();
+        // client requires to be moved so it lives long enough for streaming response future.
+        let response = async move {
+            let mut client = this.inner.clone();
+            this.wait_for_ready(&mut client).await?;
+            let request = this.build_tonic_request(request, body)?;
+            let codec = TransparentCodec::default();
+            client.streaming(request, path, codec).await
+        };
+        let req = crate::streams::NonBlockingStreamRequest::new(
+            Box::pin(response) as crate::streams::ResponseFuture
+        );
+        let response = crate::streams::send(req).await.map_err(GrpcError::from)?;
+        let response = response.map(|body| {
+            BytesStream::new(GrpcStream {
+                inner: EscapableTonicStream::new(body),
+            })
+        });
+        Ok(response.to_http().map(Into::into))
+    }
 }
 
 #[xmtp_common::async_trait]
@@ -311,7 +346,7 @@ pub mod tests {
         let request = client
             .build_tonic_request(
                 Default::default(),
-                PublishRequest { envelopes: vec![] }.encode_to_vec().into(),
+                prost::bytes::Bytes::from(PublishRequest { envelopes: vec![] }.encode_to_vec()),
             )
             .unwrap();