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

xmtp_proto/
api_client.rs
