xmtp_proto/
api_client.rs

1use crate::api::IsConnectedCheck;
2pub use crate::backend_v1::{
3    CreateUploadRequest, CreateUploadResponse, GetConfigurationRequest, GetConfigurationResponse,
4    GetInboxIdsRequest, GetInboxIdsResponse, PublishRequest, PublishResponse, QueryNewestRequest,
5    QueryNewestResponse, QueryRequest, QueryResponse, RecipientState, RegisterRequest,
6    ServerEnvelope, UnregisterRequest, UnregisterResponse, UpdateSubscriptionsRequest,
7    VerifySmartContractWalletSignaturesRequest, VerifySmartContractWalletSignaturesResponse,
8};
9use crate::types::{
10    GroupId, GroupMessage, IncomingBatchLimits, IncomingEvent, IncomingSubscription,
11    InstallationId, ReceivedQueryResponse, TopicCursor, WelcomeMessage,
12};
13use futures::Stream;
14use std::pin::Pin;
15use std::sync::Arc;
16use xmtp_common::{MaybeSend, MaybeSync};
17use xmtp_common::{Retry, RetryableError};
18
19mod impls;
20mod stats;
21pub use stats::*;
22
23xmtp_common::if_test! {
24    mod tests;
25    pub use tests::*;
26}
27
28/// A type-erased version of the Xmtp Api in a [`Box`]
29pub type BoxedXmtpApi<Error> = Box<dyn BoxableXmtpApi<Error>>;
30/// A type-erased version of the Xntp Api in a [`Arc`]
31pub type ArcedXmtpApi<Error> = Arc<dyn BoxableXmtpApi<Error>>;
32
33/// Owned transport events. Durable receipt remains the caller's responsibility.
34pub type BoxedIncomingS<Err> = xmtp_common::BoxDynStream<'static, Result<IncomingEvent, Err>>;
35
36xmtp_common::if_native! {
37    pub type BoxedGroupS<Err> = Pin<Box<dyn Stream<Item = Result<GroupMessage, Err>> + Send>>;
38    pub type BoxedWelcomeS<Err> = Pin<Box<dyn Stream<Item = Result<WelcomeMessage, Err>> + Send>>;
39    pub type BoxedSubscribeS<Err> =
40        Pin<Box<dyn Stream<Item = Result<crate::types::ReceivedSubscribeResponse, Err>> + Send>>;
41}
42
43xmtp_common::if_wasm! {
44    pub type BoxedGroupS<Err> = Pin<Box<dyn Stream<Item = Result<GroupMessage, Err>>>>;
45    pub type BoxedWelcomeS<Err> = Pin<Box<dyn Stream<Item = Result<WelcomeMessage, Err>>>>;
46}
47
48pub trait BoxableXmtpApi<Err>
49where
50    Self: XmtpBackendClient<Error = Err>
51        + XmtpMlsStreams<
52            Error = Err,
53            WelcomeMessageStream = BoxedWelcomeS<Err>,
54            GroupMessageStream = BoxedGroupS<Err>,
55        > + IsConnectedCheck
56        + MaybeSend
57        + MaybeSync,
58{
59}
60
61impl<T, Err> BoxableXmtpApi<Err> for T where
62    T: XmtpBackendClient<Error = Err>
63        + XmtpMlsStreams<
64            Error = Err,
65            WelcomeMessageStream = BoxedWelcomeS<Err>,
66            GroupMessageStream = BoxedGroupS<Err>,
67        > + IsConnectedCheck
68        + MaybeSend
69        + MaybeSync
70        + ?Sized
71{
72}
73
74pub trait XmtpApi
75where
76    Self: XmtpBackendClient,
77{
78}
79
80impl<T> XmtpApi for T where T: XmtpBackendClient + ?Sized {}
81
82/// The backend unary RPCs. Callers own retries and chunking.
83#[xmtp_common::async_trait]
84pub trait XmtpBackendClient: MaybeSend + MaybeSync {
85    type Error: RetryableError + MaybeSend + MaybeSync + 'static;
86    /// Register this client's public event writer with a shared credential.
87    fn register_client_event_writer(&self, _writer: &Arc<dyn xmtp_events::EventWriter<()>>) {}
88    fn unregister_client_event_writer(&self, _writer: &Arc<dyn xmtp_events::EventWriter<()>>) {}
89    async fn publish(&self, request: PublishRequest) -> Result<PublishResponse, Self::Error>;
90    async fn create_upload(
91        &self,
92        request: CreateUploadRequest,
93    ) -> Result<CreateUploadResponse, Self::Error>;
94    async fn query(&self, request: QueryRequest) -> Result<QueryResponse, Self::Error>;
95    /// Query and keep each `ServerEnvelope` byte-for-byte as received.
96    async fn query_raw_envelopes(
97        &self,
98        request: QueryRequest,
99    ) -> Result<ReceivedQueryResponse, Self::Error>;
100    async fn query_newest(
101        &self,
102        request: QueryNewestRequest,
103    ) -> Result<QueryNewestResponse, Self::Error>;
104    async fn get_inbox_ids(
105        &self,
106        request: GetInboxIdsRequest,
107    ) -> Result<GetInboxIdsResponse, Self::Error>;
108    /// Read what this deployment publishes about itself. Served without a
109    /// credential, so it is the one call a client can make before it knows
110    /// whether the deployment requires one.
111    async fn get_configuration(
112        &self,
113        request: GetConfigurationRequest,
114    ) -> Result<GetConfigurationResponse, Self::Error>;
115
116    /// The backend URL this client sends to, when the transport knows it.
117    ///
118    /// The stored configuration row records the URL its copy came from, and
119    /// `build` compares the two. A transport that cannot
120    /// name a URL — a test double, for instance — returns `None`, and the
121    /// comparison is skipped.
122    fn backend_url(&self) -> Option<&str> {
123        None
124    }
125
126    /// Whether a credential source — an auth callback or an auth handle — was
127    /// configured on this transport. A transport with no auth
128    /// middleware reports `false` and `build` refuses a deployment that
129    /// requires authentication. Defaults to `true` so a test double, which has
130    /// no transport stack to ask, is never the thing that refuses a build.
131    fn has_credential_source(&self) -> bool {
132        true
133    }
134
135    /// Install the shapes the deployment publishes, so the stream and
136    /// metadata chunking that happens below `ApiClientWrapper` uses them too.
137    /// Called once, by `build`, before any stream opens. A transport with
138    /// nothing to chunk ignores it.
139    fn set_limits(&self, limits: std::sync::Arc<xmtp_configuration::LimitsConfiguration>) {
140        let _ = limits;
141    }
142    async fn verify_smart_contract_wallet_signatures(
143        &self,
144        request: VerifySmartContractWalletSignaturesRequest,
145    ) -> Result<VerifySmartContractWalletSignaturesResponse, Self::Error>;
146    async fn register(&self, request: RegisterRequest) -> Result<RecipientState, Self::Error>;
147    async fn unregister(
148        &self,
149        request: UnregisterRequest,
150    ) -> Result<UnregisterResponse, Self::Error>;
151    async fn update_subscriptions(
152        &self,
153        request: UpdateSubscriptionsRequest,
154    ) -> Result<RecipientState, Self::Error>;
155}
156
157/// Represents the backend API required for an MLS Delivery Service
158/// to be compatible with XMTP streaming
159#[xmtp_common::async_trait]
160pub trait XmtpMlsStreams: MaybeSend + MaybeSync {
161    type GroupMessageStream: Stream<Item = Result<GroupMessage, Self::Error>> + MaybeSend;
162
163    type WelcomeMessageStream: Stream<Item = Result<WelcomeMessage, Self::Error>> + MaybeSend;
164
165    type Error: RetryableError + 'static;
166
167    /// Subscribe after committed receipt positions without decoding MLS payloads.
168    /// Acknowledge new positions only after the raw envelopes commit to storage.
169    async fn subscribe_envelopes_with_cursors(
170        &self,
171        cursors: &TopicCursor,
172        limits: IncomingBatchLimits,
173    ) -> Result<IncomingSubscription<Self::Error>, Self::Error>;
174
175    async fn subscribe_group_messages(
176        &self,
177        group_ids: &[&GroupId],
178    ) -> Result<Self::GroupMessageStream, Self::Error>;
179    async fn subscribe_group_messages_with_cursors(
180        &self,
181        groups_with_cursors: &TopicCursor,
182    ) -> Result<Self::GroupMessageStream, Self::Error>;
183    async fn subscribe_welcome_messages(
184        &self,
185        installations: &[&InstallationId],
186    ) -> Result<Self::WelcomeMessageStream, Self::Error>;
187    async fn subscribe_welcome_messages_with_cursors(
188        &self,
189        cursors: &TopicCursor,
190    ) -> Result<Self::WelcomeMessageStream, Self::Error>;
191}
192
193xmtp_common::if_native! {
194    /// The XIP-83 bidirectional subscription: one long-lived stream carrying
195    /// group and welcome messages, mutated in place (no reconnect on membership
196    /// change) and kept alive with WebSocket-style ping/pong. Native-only —
197    /// gRPC-Web transports cannot speak full-duplex, so browsers stay on
198    /// [`XmtpMlsStreams`] with a client-side watchdog.
199    #[xmtp_common::async_trait]
200    pub trait XmtpMlsBidiStreams: MaybeSend + MaybeSync {
201        /// Server frames that keep each `ServerEnvelope` byte-for-byte as received.
202        type SubscribeStream: Stream<Item = Result<crate::types::ReceivedSubscribeResponse, Self::Error>>
203            + MaybeSend;
204
205        type Error: RetryableError + 'static;
206
207        /// The frame shapes this deployment accepts, as installed by
208        /// [`XmtpBackendClient::set_limits`]. The bidi ledger chunks interest
209        /// updates below `ApiClientWrapper`, so it cannot read the wrapper's
210        /// copy and asks the transport instead. A transport that was never told
211        /// reports the compiled defaults.
212        fn bidi_limits(&self) -> std::sync::Arc<xmtp_configuration::LimitsConfiguration> {
213            std::sync::Arc::default()
214        }
215
216        /// Return the URL used for bidi connections. Combine it with the API
217        /// client's Arc identity when selecting a shared connection.
218        fn host(&self) -> &str;
219
220        /// Open the bidirectional stream. `requests` is the outbound
221        /// client→server frame stream (typically fed by a channel; the first
222        /// frame is an `Update` naming the initial topic set); the
223        /// returned stream yields the server→client frames.
224        async fn subscribe_bidi(
225            &self,
226            requests: futures::stream::BoxStream<'static, crate::backend_v1::SubscribeRequest>,
227        ) -> Result<Self::SubscribeStream, Self::Error>;
228    }
229}
230
231/// describe how to create a single network
232/// connection.
233/// Implement this trait if your type connects to a single
234/// network channel/connection (like gRPc or HTTP)
235pub trait NetConnectConfig: ApiBuilder + MaybeSend + MaybeSync {
236    /// set the libxmtp version (required)
237    fn set_libxmtp_version(&mut self, version: String) -> Result<(), Self::Error>;
238
239    /// set the sdk app version (required)
240    fn set_app_version(&mut self, version: crate::types::AppVersion) -> Result<(), Self::Error>;
241
242    /// Set the libxmtp host (required)
243    fn set_host(&mut self, host: url::Url);
244
245    /// Set the retry strategy for this client
246    fn set_retry(&mut self, retry: Retry);
247
248    /// Set the rate limit per minute for this client
249    fn rate_per_minute(&mut self, limit: u32);
250
251    /// The port this api builder is using
252    fn port(&self) -> Result<Option<String>, Self::Error>;
253
254    /// Host of the builder
255    fn host(&self) -> Option<&str>;
256}
257
258/// Build an API from its parts for the XMTP Backend
259pub trait ApiBuilder: MaybeSend + MaybeSync {
260    type Output: MaybeSend + MaybeSync;
261    type Error: MaybeSend + MaybeSync + std::fmt::Debug;
262    /// Build the api client
263    fn build(self) -> Result<Self::Output, Self::Error>;
264}
xmtp_proto/api_client.rs

xmtp_proto/
api_client.rs
