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, InstallationId, Topic, TopicCursor,
9        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    pub async fn query_welcome_messages_with_cursors(
81        &self,
82        cursors: TopicCursor,
83    ) -> Result<Vec<WelcomeMessage>> {
84        self.query_all(cursors, self.limits().max_query_limit as u32)
85            .await?
86            .into_iter()
87            .map(|envelope| decode_welcome_message(envelope).map_err(Into::into))
88            .collect()
89    }
90    #[xmtp_common::rpc_span]
91    pub async fn upload_key_package(&self, key_package: Vec<u8>) -> Result<wire::EnvelopeMeta> {
92        self.publish_units(vec![PublishUnit::single_within(
93            wire::ClientEnvelope {
94                payload: Some(wire::client_envelope::Payload::KeyPackage(
95                    wire::KeyPackage {
96                        key_package_tls_serialized: key_package,
97                    },
98                )),
99            },
100            self.limits(),
101        )?])
102        .await?
103        .into_iter()
104        .next()
105        .ok_or(ApiError::InvalidResponse("key package metadata"))
106    }
107    // implements: API-245
108    #[xmtp_common::rpc_span]
109    pub async fn fetch_key_packages(&self, keys: &[InstallationId]) -> Result<KeyPackageMap> {
110        let mut found: KeyPackageMap = keys.iter().cloned().map(|key| (key, None)).collect();
111        let topics = found.keys().map(Topic::new_key_package).collect();
112        for result in self.newest(topics, true).await? {
113            let topic = Topic::parse(
114                &result
115                    .topic
116                    .ok_or(ApiError::InvalidResponse("key package topic"))?
117                    .topic,
118            )?;
119            let key: InstallationId = topic.identifier().try_into()?;
120            let slot = found
121                .get_mut(&key)
122                .ok_or(ApiError::InvalidResponse("unrequested key package"))?;
123            if slot.is_some() {
124                return Err(ApiError::InvalidResponse("duplicate key package"));
125            }
126            *slot = Some(decode_key_package(wire::ServerEnvelope {
127                meta: result.meta,
128                envelope: result.envelope,
129            })?);
130        }
131        Ok(found)
132    }
133    #[xmtp_common::rpc_span]
134    pub async fn send_welcome_messages(
135        &self,
136        messages: &[wire::WelcomeMessage],
137    ) -> Result<Vec<wire::EnvelopeMeta>> {
138        let units = messages
139            .iter()
140            .cloned()
141            .map(|message| {
142                PublishUnit::single_within(
143                    wire::ClientEnvelope {
144                        payload: Some(wire::client_envelope::Payload::WelcomeMessage(message)),
145                    },
146                    self.limits(),
147                )
148            })
149            .collect::<Result<Vec<_>>>()?;
150        self.publish_units(units).await
151    }
152    #[xmtp_common::rpc_span]
153    pub async fn send_group_messages(
154        &self,
155        units: Vec<PublishUnit>,
156    ) -> Result<Vec<wire::EnvelopeMeta>> {
157        self.publish_units(units).await
158    }
159    #[xmtp_common::rpc_span]
160    pub async fn publish_commit_log(
161        &self,
162        entries: Vec<wire::CommitLogEntry>,
163    ) -> Result<Vec<wire::EnvelopeMeta>> {
164        let units = entries
165            .into_iter()
166            .map(|entry| {
167                PublishUnit::single_within(
168                    wire::ClientEnvelope {
169                        payload: Some(wire::client_envelope::Payload::CommitLogEntry(entry)),
170                    },
171                    self.limits(),
172                )
173            })
174            .collect::<Result<Vec<_>>>()?;
175        self.publish_units(units).await
176    }
177    #[xmtp_common::rpc_span]
178    pub async fn query_commit_log(
179        &self,
180        cursors: TopicCursor,
181    ) -> Result<Vec<xmtp_proto::types::CommitLogEntry>> {
182        self.query_all(cursors, self.limits().max_query_limit as u32)
183            .await?
184            .into_iter()
185            .map(|envelope| decode_commit_log_entry(envelope).map_err(Into::into))
186            .collect()
187    }
188    #[xmtp_common::rpc_span]
189    pub async fn get_newest_message_metadata(
190        &self,
191        group_ids: &[GroupId],
192    ) -> Result<MessageMetadataMap> {
193        self.newest(
194            group_ids.iter().map(Topic::new_group_message).collect(),
195            false,
196        )
197        .await?
198        .into_iter()
199        .map(|result| {
200            let meta = decode_group_message_metadata(
201                result
202                    .meta
203                    .ok_or(ApiError::InvalidResponse("group metadata"))?,
204            )?;
205            Ok((meta.group_id, meta))
206        })
207        .collect()
208    }
209}
210
211impl<C: XmtpMlsStreams + XmtpBackendClient> ApiClientWrapper<C>
212where
213    C::GroupMessageStream: 'static,
214    C::WelcomeMessageStream: 'static,
215{
216    #[xmtp_common::rpc_span]
217    pub async fn subscribe_group_messages(
218        &self,
219        groups: &[&GroupId],
220    ) -> Result<<crate::preflight::GuardedApi<C> as XmtpMlsStreams>::GroupMessageStream> {
221        self.retry_call(|| self.api_client.subscribe_group_messages(groups), false)
222            .await
223            .map_err(dyn_err)
224    }
225    #[xmtp_common::rpc_span]
226    pub async fn subscribe_group_messages_with_cursors(
227        &self,
228        cursors: &TopicCursor,
229    ) -> Result<<crate::preflight::GuardedApi<C> as XmtpMlsStreams>::GroupMessageStream> {
230        self.retry_call(
231            || {
232                self.api_client
233                    .subscribe_group_messages_with_cursors(cursors)
234            },
235            false,
236        )
237        .await
238        .map_err(dyn_err)
239    }
240    #[xmtp_common::rpc_span]
241    pub async fn subscribe_welcome_messages(
242        &self,
243        installations: &[&InstallationId],
244    ) -> Result<<crate::preflight::GuardedApi<C> as XmtpMlsStreams>::WelcomeMessageStream> {
245        self.retry_call(
246            || self.api_client.subscribe_welcome_messages(installations),
247            false,
248        )
249        .await
250        .map_err(dyn_err)
251    }
252    #[xmtp_common::rpc_span]
253    pub async fn subscribe_welcome_messages_with_cursors(
254        &self,
255        cursors: &TopicCursor,
256    ) -> Result<<crate::preflight::GuardedApi<C> as XmtpMlsStreams>::WelcomeMessageStream> {
257        self.retry_call(
258            || {
259                self.api_client
260                    .subscribe_welcome_messages_with_cursors(cursors)
261            },
262            false,
263        )
264        .await
265        .map_err(dyn_err)
266    }
267}
xmtp_api/mls.rs

xmtp_api/
mls.rs
