xmtp_proto/
api_client.rs

1use crate::api::IsConnectedCheck;
2pub use crate::backend_v1::{
3    GetConfigurationRequest, GetConfigurationResponse, GetInboxIdsRequest, GetInboxIdsResponse,
4    PublishRequest, PublishResponse, QueryNewestRequest, QueryNewestResponse, QueryRequest,
5    QueryResponse, RecipientState, RegisterRequest, ServerEnvelope, UnregisterRequest,
6    UnregisterResponse, UpdateSubscriptionsRequest, VerifySmartContractWalletSignaturesRequest,
7    VerifySmartContractWalletSignaturesResponse,
8};
9use crate::types::{
10    GroupId, GroupMessage, IncomingBatchLimits, IncomingEvent, IncomingSubscription,
11    InstallationId, 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::backend_v1::SubscribeResponse, 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    async fn publish(&self, request: PublishRequest) -> Result<PublishResponse, Self::Error>;
87    async fn query(&self, request: QueryRequest) -> Result<QueryResponse, Self::Error>;
88    async fn query_newest(
89        &self,
90        request: QueryNewestRequest,
91    ) -> Result<QueryNewestResponse, Self::Error>;
92    async fn get_inbox_ids(
93        &self,
94        request: GetInboxIdsRequest,
95    ) -> Result<GetInboxIdsResponse, Self::Error>;
96    /// Read what this deployment publishes about itself. Served without a
97    /// credential, so it is the one call a client can make before it knows
98    /// whether the deployment requires one.
99    async fn get_configuration(
100        &self,
101        request: GetConfigurationRequest,
102    ) -> Result<GetConfigurationResponse, Self::Error>;
103
104    /// The backend URL this client sends to, when the transport knows it.
105    ///
106    /// The stored configuration row records the URL its copy came from, and
107    /// `build` compares the two (CFG-042, CFG-055). A transport that cannot
108    /// name a URL — a test double, for instance — returns `None`, and the
109    /// comparison is skipped.
110    fn backend_url(&self) -> Option<&str> {
111        None
112    }
113
114    /// Whether a credential source — an auth callback or an auth handle — was
115    /// configured on this transport (CFG-062). A transport with no auth
116    /// middleware reports `false` and `build` refuses a deployment that
117    /// requires authentication. Defaults to `true` so a test double, which has
118    /// no transport stack to ask, is never the thing that refuses a build.
119    fn has_credential_source(&self) -> bool {
120        true
121    }
122
123    /// Install the shapes the deployment publishes (CFG-064), so the stream and
124    /// metadata chunking that happens below `ApiClientWrapper` uses them too.
125    /// Called once, by `build`, before any stream opens. A transport with
126    /// nothing to chunk ignores it.
127    fn set_limits(&self, limits: std::sync::Arc<xmtp_configuration::LimitsConfiguration>) {
128        let _ = limits;
129    }
130    async fn verify_smart_contract_wallet_signatures(
131        &self,
132        request: VerifySmartContractWalletSignaturesRequest,
133    ) -> Result<VerifySmartContractWalletSignaturesResponse, Self::Error>;
134    async fn register(&self, request: RegisterRequest) -> Result<RecipientState, Self::Error>;
135    async fn unregister(
136        &self,
137        request: UnregisterRequest,
138    ) -> Result<UnregisterResponse, Self::Error>;
139    async fn update_subscriptions(
140        &self,
141        request: UpdateSubscriptionsRequest,
142    ) -> Result<RecipientState, Self::Error>;
143}
144
145/// Represents the backend API required for an MLS Delivery Service
146/// to be compatible with XMTP streaming
147#[xmtp_common::async_trait]
148pub trait XmtpMlsStreams: MaybeSend + MaybeSync {
149    type GroupMessageStream: Stream<Item = Result<GroupMessage, Self::Error>> + MaybeSend;
150
151    type WelcomeMessageStream: Stream<Item = Result<WelcomeMessage, Self::Error>> + MaybeSend;
152
153    type Error: RetryableError + 'static;
154
155    /// Subscribe after committed receipt positions without decoding MLS payloads.
156    /// Acknowledge new positions only after the raw envelopes commit to storage.
157    async fn subscribe_envelopes_with_cursors(
158        &self,
159        cursors: &TopicCursor,
160        limits: IncomingBatchLimits,
161    ) -> Result<IncomingSubscription<Self::Error>, Self::Error>;
162
163    async fn subscribe_group_messages(
164        &self,
165        group_ids: &[&GroupId],
166    ) -> Result<Self::GroupMessageStream, Self::Error>;
167    async fn subscribe_group_messages_with_cursors(
168        &self,
169        groups_with_cursors: &TopicCursor,
170    ) -> Result<Self::GroupMessageStream, Self::Error>;
171    async fn subscribe_welcome_messages(
172        &self,
173        installations: &[&InstallationId],
174    ) -> Result<Self::WelcomeMessageStream, Self::Error>;
175    async fn subscribe_welcome_messages_with_cursors(
176        &self,
177        cursors: &TopicCursor,
178    ) -> Result<Self::WelcomeMessageStream, Self::Error>;
179}
180
181xmtp_common::if_native! {
182    /// The XIP-83 bidirectional subscription: one long-lived stream carrying
183    /// group and welcome messages, mutated in place (no reconnect on membership
184    /// change) and kept alive with WebSocket-style ping/pong. Native-only —
185    /// gRPC-Web transports cannot speak full-duplex, so browsers stay on
186    /// [`XmtpMlsStreams`] with a client-side watchdog.
187    #[xmtp_common::async_trait]
188    pub trait XmtpMlsBidiStreams: MaybeSend + MaybeSync {
189        type SubscribeStream: Stream<Item = Result<crate::backend_v1::SubscribeResponse, Self::Error>>
190            + MaybeSend;
191
192        type Error: RetryableError + 'static;
193
194        /// The frame shapes this deployment accepts (CFG-064), as installed by
195        /// [`XmtpBackendClient::set_limits`]. The bidi ledger chunks interest
196        /// updates below `ApiClientWrapper`, so it cannot read the wrapper's
197        /// copy and asks the transport instead. A transport that was never told
198        /// reports the compiled defaults.
199        fn bidi_limits(&self) -> std::sync::Arc<xmtp_configuration::LimitsConfiguration> {
200            std::sync::Arc::default()
201        }
202
203        /// Return the URL used for bidi connections. Combine it with the API
204        /// client's Arc identity when selecting a shared connection.
205        fn host(&self) -> &str;
206
207        /// Open the bidirectional stream. `requests` is the outbound
208        /// client→server frame stream (typically fed by a channel; the first
209        /// frame is an `Update` naming the initial topic set); the
210        /// returned stream yields the server→client frames.
211        async fn subscribe_bidi(
212            &self,
213            requests: futures::stream::BoxStream<'static, crate::backend_v1::SubscribeRequest>,
214        ) -> Result<Self::SubscribeStream, Self::Error>;
215    }
216}
217
218/// describe how to create a single network
219/// connection.
220/// Implement this trait if your type connects to a single
221/// network channel/connection (like gRPc or HTTP)
222pub trait NetConnectConfig: ApiBuilder + MaybeSend + MaybeSync {
223    /// set the libxmtp version (required)
224    fn set_libxmtp_version(&mut self, version: String) -> Result<(), Self::Error>;
225
226    /// set the sdk app version (required)
227    fn set_app_version(&mut self, version: crate::types::AppVersion) -> Result<(), Self::Error>;
228
229    /// Set the libxmtp host (required)
230    fn set_host(&mut self, host: url::Url);
231
232    /// Set the retry strategy for this client
233    fn set_retry(&mut self, retry: Retry);
234
235    /// Set the rate limit per minute for this client
236    fn rate_per_minute(&mut self, limit: u32);
237
238    /// The port this api builder is using
239    fn port(&self) -> Result<Option<String>, Self::Error>;
240
241    /// Host of the builder
242    fn host(&self) -> Option<&str>;
243}
244
245/// Build an API from its parts for the XMTP Backend
246pub trait ApiBuilder: MaybeSend + MaybeSync {
247    type Output: MaybeSend + MaybeSync;
248    type Error: MaybeSend + MaybeSync + std::fmt::Debug;
249    /// Build the api client
250    fn build(self) -> Result<Self::Output, Self::Error>;
251}
xmtp_proto/api_client.rs

xmtp_proto/
api_client.rs
