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