xmtp_api/
mls.rs

1use crate::{ApiClientWrapper, ApiError, PublishUnit, Result, dyn_err};
2use std::collections::HashMap;
3use xmtp_api_backend::envelope::*;
4use xmtp_proto::{
5    api_client::{XmtpBackendClient, XmtpMlsStreams},
6    backend_v1 as wire,
7    types::{
8        Cursor, GroupId, GroupMessage, GroupMessageMetadata, IncomingBatchLimits, InstallationId,
9        Topic, TopicCursor, WelcomeMessage,
10    },
11};
12
13#[derive(Clone, Debug)]
14pub struct GroupFilter {
15    pub group_id: GroupId,
16    pub id_cursor: Option<u64>,
17}
18impl GroupFilter {
19    pub fn new(group_id: GroupId, id_cursor: Option<u64>) -> Self {
20        Self {
21            group_id,
22            id_cursor,
23        }
24    }
25}
26
27pub type KeyPackageMap = HashMap<InstallationId, Option<wire::KeyPackage>>;
28type MessageMetadataMap = HashMap<GroupId, GroupMessageMetadata>;
29
30impl<C: XmtpBackendClient> ApiClientWrapper<C> {
31    #[xmtp_common::rpc_span]
32    pub async fn query_group_messages(&self, group_id: GroupId) -> Result<Vec<GroupMessage>> {
33        self.query_group_messages_with_cursors(HashMap::from([(
34            Topic::new_group_message(group_id),
35            Cursor(0),
36        )]))
37        .await
38    }
39    pub async fn query_group_messages_with_cursors(
40        &self,
41        cursors: TopicCursor,
42    ) -> Result<Vec<GroupMessage>> {
43        self.query_all(cursors, self.limits().max_query_limit as u32)
44            .await?
45            .into_iter()
46            .map(|envelope| decode_group_message(envelope).map_err(Into::into))
47            .collect()
48    }
49    #[xmtp_common::rpc_span]
50    pub async fn query_latest_group_message(
51        &self,
52        group_id: GroupId,
53    ) -> Result<Option<GroupMessage>> {
54        let result = self
55            .newest(vec![Topic::new_group_message(group_id)], true)
56            .await?
57            .into_iter()
58            .next();
59        result
60            .map(|result| {
61                decode_group_message(wire::ServerEnvelope {
62                    meta: result.meta,
63                    envelope: result.envelope,
64                })
65                .map_err(Into::into)
66            })
67            .transpose()
68    }
69    #[xmtp_common::rpc_span]
70    pub async fn query_welcome_messages<Id: AsRef<[u8]> + Copy>(
71        &self,
72        installation_id: Id,
73    ) -> Result<Vec<WelcomeMessage>> {
74        self.query_welcome_messages_with_cursors(HashMap::from([(
75            Topic::new_welcome_message(installation_id.as_ref().try_into()?),
76            Cursor(0),
77        )]))
78        .await
79    }
80    /// The lowest-sequence envelope on the Welcome topic of `destination`,
81    /// with validated metadata and an undecoded payload. One `Query` with
82    /// limit 1 from cursor 0, so later envelopes are never fetched.
83    // implements: JOIN-032
84    #[xmtp_common::rpc_span]
85    pub async fn query_first_welcome(
86        &self,
87        destination: InstallationId,
88        max_bytes: usize,
89    ) -> Result<Option<wire::ServerEnvelope>> {
90        let page = self
91            .query_ordered_page(
92                HashMap::from([(Topic::new_welcome_message(destination), Cursor(0))]),
93                1,
94                IncomingBatchLimits {
95                    max_rows: 1,
96                    max_bytes,
97                },
98            )
99            .await?;
100        Ok(page
101            .batches
102            .into_iter()
103            .flat_map(|batch| batch.envelopes)
104            .next()
105            .map(|envelope| envelope.decode())
106            .transpose()?)
107    }
108    pub async fn query_welcome_messages_with_cursors(
109        &self,
110        cursors: TopicCursor,
111    ) -> Result<Vec<WelcomeMessage>> {
112        self.query_all(cursors, self.limits().max_query_limit as u32)
113            .await?
114            .into_iter()
115            .map(|envelope| decode_welcome_message(envelope).map_err(Into::into))
116            .collect()
117    }
118    #[xmtp_common::rpc_span]
119    pub async fn upload_key_package(&self, key_package: Vec<u8>) -> Result<wire::EnvelopeMeta> {
120        self.publish_units(vec![PublishUnit::single_within(
121            wire::ClientEnvelope {
122                payload: Some(wire::client_envelope::Payload::KeyPackage(
123                    wire::KeyPackage {
124                        key_package_tls_serialized: key_package,
125                    },
126                )),
127            },
128            self.limits(),
129        )?])
130        .await?
131        .into_iter()
132        .next()
133        .ok_or(ApiError::InvalidResponse("key package metadata"))
134    }
135    // implements: API-245
136    #[xmtp_common::rpc_span]
137    pub async fn fetch_key_packages(&self, keys: &[InstallationId]) -> Result<KeyPackageMap> {
138        let mut found: KeyPackageMap = keys.iter().cloned().map(|key| (key, None)).collect();
139        let topics = found.keys().map(Topic::new_key_package).collect();
140        for result in self.newest(topics, true).await? {
141            let topic = Topic::parse(
142                &result
143                    .topic
144                    .ok_or(ApiError::InvalidResponse("key package topic"))?
145                    .topic,
146            )?;
147            let key: InstallationId = topic.identifier().try_into()?;
148            let slot = found
149                .get_mut(&key)
150                .ok_or(ApiError::InvalidResponse("unrequested key package"))?;
151            if slot.is_some() {
152                return Err(ApiError::InvalidResponse("duplicate key package"));
153            }
154            *slot = Some(decode_key_package(wire::ServerEnvelope {
155                meta: result.meta,
156                envelope: result.envelope,
157            })?);
158        }
159        Ok(found)
160    }
161    /// The newest package on the key package topic of `id`, with its sequence id.
162    // implements: JOIN-074
163    #[xmtp_common::rpc_span]
164    pub async fn newest_key_package(
165        &self,
166        id: &InstallationId,
167    ) -> Result<Option<(wire::KeyPackage, Cursor)>> {
168        let newest = self.newest(vec![Topic::new_key_package(id)], true).await?;
169        let Some(result) = newest.into_iter().next() else {
170            return Ok(None);
171        };
172        let cursor = result.meta.as_ref().and_then(|meta| meta.cursor.as_ref());
173        let cursor = Cursor(cursor.map_or(0, |cursor| cursor.sequence_id));
174        let package = decode_key_package(wire::ServerEnvelope {
175            meta: result.meta,
176            envelope: result.envelope,
177        })?;
178        Ok(Some((package, cursor)))
179    }
180    #[xmtp_common::rpc_span]
181    pub async fn send_welcome_messages(
182        &self,
183        messages: &[wire::WelcomeMessage],
184    ) -> Result<Vec<wire::EnvelopeMeta>> {
185        let units = messages
186            .iter()
187            .cloned()
188            .map(|message| {
189                PublishUnit::single_within(
190                    wire::ClientEnvelope {
191                        payload: Some(wire::client_envelope::Payload::WelcomeMessage(message)),
192                    },
193                    self.limits(),
194                )
195            })
196            .collect::<Result<Vec<_>>>()?;
197        self.publish_units(units).await
198    }
199    #[xmtp_common::rpc_span]
200    pub async fn send_group_messages(
201        &self,
202        units: Vec<PublishUnit>,
203    ) -> Result<Vec<wire::EnvelopeMeta>> {
204        self.publish_units(units).await
205    }
206    /// Publish commit-log entries in the given order, one request at a time,
207    /// and stop at the first failed request. Each stored entry's metadata is
208    /// appended to `stored`, so on an error `stored` names the stored prefix
209    /// of `entries` and a retry starts from the first entry after it.
210    // implements: FORK-021
211    #[xmtp_common::rpc_span]
212    pub async fn publish_commit_log(
213        &self,
214        entries: Vec<wire::SealedCommitLogEntry>,
215        stored: &mut Vec<wire::EnvelopeMeta>,
216    ) -> Result<()> {
217        let units = entries
218            .into_iter()
219            .map(|entry| {
220                PublishUnit::single_within(
221                    wire::ClientEnvelope {
222                        payload: Some(wire::client_envelope::Payload::SealedCommitLogEntry(entry)),
223                    },
224                    self.limits(),
225                )
226            })
227            .collect::<Result<Vec<_>>>()?;
228        self.publish_units_in_order(units, stored).await
229    }
230    /// Read commit-log topics without opening any entry. An envelope whose
231    /// metadata does not decode is logged and left out; the read still
232    /// returns every other envelope, so one bad envelope never fails it.
233    #[xmtp_common::rpc_span]
234    pub async fn query_commit_log(
235        &self,
236        cursors: TopicCursor,
237    ) -> Result<Vec<xmtp_proto::types::CommitLogEntry>> {
238        Ok(self
239            .query_all(cursors, self.limits().max_query_limit as u32)
240            .await?
241            .into_iter()
242            .filter_map(|envelope| {
243                decode_commit_log_entry(envelope)
244                    .inspect_err(|error| {
245                        tracing::warn!(%error, "discarding a commit-log envelope");
246                    })
247                    .ok()
248            })
249            .collect())
250    }
251    #[xmtp_common::rpc_span]
252    pub async fn get_newest_message_metadata(
253        &self,
254        group_ids: &[GroupId],
255    ) -> Result<MessageMetadataMap> {
256        self.newest(
257            group_ids.iter().map(Topic::new_group_message).collect(),
258            false,
259        )
260        .await?
261        .into_iter()
262        .map(|result| {
263            let meta = decode_group_message_metadata(
264                result
265                    .meta
266                    .ok_or(ApiError::InvalidResponse("group metadata"))?,
267            )?;
268            Ok((meta.group_id, meta))
269        })
270        .collect()
271    }
272}
273
274impl<C: XmtpMlsStreams + XmtpBackendClient> ApiClientWrapper<C>
275where
276    C::GroupMessageStream: 'static,
277    C::WelcomeMessageStream: 'static,
278{
279    #[xmtp_common::rpc_span]
280    pub async fn subscribe_group_messages(
281        &self,
282        groups: &[&GroupId],
283    ) -> Result<<crate::preflight::GuardedApi<C> as XmtpMlsStreams>::GroupMessageStream> {
284        self.retry_call(|| self.api_client.subscribe_group_messages(groups), false)
285            .await
286            .map_err(dyn_err)
287    }
288    #[xmtp_common::rpc_span]
289    pub async fn subscribe_group_messages_with_cursors(
290        &self,
291        cursors: &TopicCursor,
292    ) -> Result<<crate::preflight::GuardedApi<C> as XmtpMlsStreams>::GroupMessageStream> {
293        self.retry_call(
294            || {
295                self.api_client
296                    .subscribe_group_messages_with_cursors(cursors)
297            },
298            false,
299        )
300        .await
301        .map_err(dyn_err)
302    }
303    #[xmtp_common::rpc_span]
304    pub async fn subscribe_welcome_messages(
305        &self,
306        installations: &[&InstallationId],
307    ) -> Result<<crate::preflight::GuardedApi<C> as XmtpMlsStreams>::WelcomeMessageStream> {
308        self.retry_call(
309            || self.api_client.subscribe_welcome_messages(installations),
310            false,
311        )
312        .await
313        .map_err(dyn_err)
314    }
315    #[xmtp_common::rpc_span]
316    pub async fn subscribe_welcome_messages_with_cursors(
317        &self,
318        cursors: &TopicCursor,
319    ) -> Result<<crate::preflight::GuardedApi<C> as XmtpMlsStreams>::WelcomeMessageStream> {
320        self.retry_call(
321            || {
322                self.api_client
323                    .subscribe_welcome_messages_with_cursors(cursors)
324            },
325            false,
326        )
327        .await
328        .map_err(dyn_err)
329    }
330}
xmtp_api/mls.rs

xmtp_api/
mls.rs
