xmtp_api/
identity.rs

1use crate::{
2    ApiClientWrapper, ApiError, HandleRejection, PublishUnit, Result,
3    chunk::{MAX_READ_CHUNKS_IN_FLIGHT, publish_error_reason},
4    dyn_err,
5};
6use futures::{StreamExt, TryStreamExt, stream};
7use prost::Message;
8use std::collections::{HashMap, HashSet};
9use xmtp_configuration::MAX_SCW_SIGNATURES_PER_REQUEST;
10use xmtp_proto::{
11    api::grpc_status,
12    api_client::XmtpBackendClient,
13    backend_v1 as wire,
14    types::{ApiIdentifier, Cursor, IdentityUpdateLog, IncomingBatchLimits, Topic, TopicCursor},
15    xmtp::identity::associations::{IdentifierKind, IdentityUpdate},
16};
17
18/// The handle rejection that a refused identity publish reports in its
19/// structured detail.
20fn handle_rejection(error: &ApiError) -> Option<HandleRejection> {
21    use wire::publish_error::Reason;
22    match publish_error_reason(error)? {
23        Reason::HandleTaken => Some(HandleRejection::Taken),
24        Reason::HandleNotAllowed => Some(HandleRejection::NotAllowed),
25        Reason::HandleTooSoon => Some(HandleRejection::TooSoon),
26        _ => None,
27    }
28}
29
30/// Read updates after the exclusive sequence id on this inbox topic.
31#[derive(Debug)]
32pub struct GetIdentityUpdatesV2Filter {
33    pub inbox_id: String,
34    pub sequence_id: Option<u64>,
35}
36
37impl<C: XmtpBackendClient> ApiClientWrapper<C> {
38    #[xmtp_common::rpc_span]
39    pub async fn publish_identity_update<U: Into<IdentityUpdate>>(
40        &self,
41        update: U,
42    ) -> Result<Cursor> {
43        let unit = PublishUnit::single_within(
44            wire::ClientEnvelope {
45                payload: Some(wire::client_envelope::Payload::IdentityUpdate(
46                    update.into(),
47                )),
48            },
49            self.limits(),
50        )?;
51        match self.publish_units(vec![unit]).await {
52            Ok(metas) => metas
53                .into_iter()
54                .next()
55                .and_then(|meta| meta.cursor)
56                .map(Into::into)
57                .ok_or(ApiError::InvalidResponse("identity publish cursor")),
58            Err(error)
59                if grpc_status(&error)
60                    .is_some_and(|status| status.code() == tonic::Code::Aborted) =>
61            {
62                Err(ApiError::IdentityUpdateConflict)
63            }
64            Err(error) => Err(handle_rejection(&error)
65                .map(ApiError::HandleRejected)
66                .unwrap_or(error)),
67        }
68    }
69    /// Read every update the filters name and return them by inbox, in
70    /// sequence order. Use [`Self::fold_identity_updates`] to store a long
71    /// history without retaining it all.
72    #[xmtp_common::rpc_span]
73    pub async fn get_identity_updates_v2(
74        &self,
75        filters: Vec<GetIdentityUpdatesV2Filter>,
76    ) -> Result<HashMap<String, Vec<IdentityUpdateLog>>> {
77        let mut result: HashMap<_, Vec<_>> = filters
78            .iter()
79            .map(|filter| (filter.inbox_id.clone(), Vec::new()))
80            .collect();
81        self.fold_identity_updates(filters, IncomingBatchLimits::STREAM_BATCH, |updates| {
82            for update in updates {
83                result
84                    .entry(update.update.inbox_id.clone())
85                    .or_default()
86                    .push(update);
87            }
88            Ok::<_, ApiError>(())
89        })
90        .await?;
91        Ok(result)
92    }
93
94    /// Read the updates the filters name one bounded page at a time, in
95    /// sequence order for each inbox, without retaining earlier pages. Every
96    /// update names one of the filters' inboxes exactly. An API or fold error
97    /// stops the read; the caller stores each page before the next request.
98    #[xmtp_common::rpc_span]
99    pub async fn fold_identity_updates<E: From<ApiError> + std::fmt::Display>(
100        &self,
101        filters: Vec<GetIdentityUpdatesV2Filter>,
102        limits: IncomingBatchLimits,
103        mut fold: impl FnMut(Vec<IdentityUpdateLog>) -> std::result::Result<(), E>,
104    ) -> std::result::Result<(), E> {
105        let mut cursors = TopicCursor::new();
106        for filter in &filters {
107            let bytes =
108                hex::decode(&filter.inbox_id).map_err(|_| ApiError::InvalidRequest("inbox id"))?;
109            let topic = Topic::new_identity_update(bytes);
110            Topic::parse(&topic).map_err(ApiError::from)?;
111            let after = Cursor(filter.sequence_id.unwrap_or(0));
112            cursors
113                .entry(topic)
114                .and_modify(|cursor| *cursor = (*cursor).min(after))
115                .or_insert(after);
116        }
117        let requested: HashSet<_> = filters.iter().map(|filter| &filter.inbox_id).collect();
118        self.fold_ordered_pages(cursors, limits, |batches, _| {
119            let updates = batches
120                .into_iter()
121                .flat_map(|batch| batch.envelopes)
122                .map(|envelope| {
123                    let update =
124                        xmtp_api_backend::envelope::decode_identity_update(envelope.decode()?)?;
125                    if !requested.contains(&update.update.inbox_id) {
126                        return Err(ApiError::InvalidResponse("unrequested inbox"));
127                    }
128                    Ok(update)
129                })
130                .collect::<Result<Vec<_>>>()?;
131            fold(updates)
132        })
133        .await
134    }
135    /// Return one optional inbox id for every input, in the same order.
136    #[xmtp_common::rpc_span]
137    pub async fn get_inbox_ids(
138        &self,
139        identifiers: Vec<ApiIdentifier>,
140    ) -> Result<Vec<Option<String>>> {
141        if identifiers
142            .iter()
143            .any(|id| id.identifier_kind == IdentifierKind::Unspecified)
144        {
145            return Err(ApiError::InvalidResponse("unspecified identifier kind"));
146        }
147        let requests: Vec<_> = identifiers
148            .chunks(self.limits().max_lookup_identifiers)
149            .map(<[_]>::to_vec)
150            .collect();
151        let mut chunks: Vec<_> = stream::iter(requests.into_iter().enumerate().map(
152            |(index, chunk)| async move {
153                let request = wire::GetInboxIdsRequest {
154                    requests: chunk
155                        .iter()
156                        .map(|id| wire::get_inbox_ids_request::Request {
157                            identifier: id.identifier.clone(),
158                            identifier_kind: id.identifier_kind as i32,
159                        })
160                        .collect(),
161                };
162                let response = self
163                    .retry_call(|| self.api_client.get_inbox_ids(request.clone()), false)
164                    .await
165                    .map_err(dyn_err)?;
166                if response.responses.len() != chunk.len() {
167                    return Err(ApiError::InvalidResponse("inbox result count"));
168                }
169                let mut values = Vec::with_capacity(chunk.len());
170                for (response, input) in response.responses.into_iter().zip(&chunk) {
171                    if IdentifierKind::try_from(response.identifier_kind)
172                        .ok()
173                        .filter(|kind| *kind != IdentifierKind::Unspecified)
174                        != Some(input.identifier_kind)
175                        || response.identifier != input.identifier
176                    {
177                        return Err(ApiError::InvalidResponse("inbox result identity"));
178                    }
179                    values.push(response.inbox_id);
180                }
181                Ok((index, values))
182            },
183        ))
184        .buffer_unordered(MAX_READ_CHUNKS_IN_FLIGHT)
185        .try_collect()
186        .await?;
187        chunks.sort_by_key(|(index, _)| *index);
188        Ok(chunks.into_iter().flat_map(|(_, values)| values).collect())
189    }
190    /// Verify every signature and return one verdict per signature, in
191    /// request order. The signatures go out in requests of at most the
192    /// published `max_scw_signatures`, never more than
193    /// [`MAX_SCW_SIGNATURES_PER_REQUEST`], and within `max_request_bytes`,
194    /// with up to [`MAX_READ_CHUNKS_IN_FLIGHT`] requests in flight. A
195    /// response whose verdict count differs from its request's is a
196    /// retryable [`ApiError::VerdictCount`].
197    #[xmtp_common::rpc_span]
198    pub async fn verify_smart_contract_wallet_signatures(
199        &self,
200        request: wire::VerifySmartContractWalletSignaturesRequest,
201    ) -> Result<wire::VerifySmartContractWalletSignaturesResponse> {
202        let limits = self.limits();
203        let per_request = limits
204            .max_scw_signatures
205            .min(MAX_SCW_SIGNATURES_PER_REQUEST);
206        let requests: Vec<_> =
207            verify_requests(&request.signatures, per_request, limits.max_request_bytes)
208                .into_iter()
209                .map(<[_]>::to_vec)
210                .collect();
211        let chunks: Vec<Vec<_>> = stream::iter(
212            requests
213                .into_iter()
214                .map(|signatures| self.verify_chunk(signatures)),
215        )
216        .buffered(MAX_READ_CHUNKS_IN_FLIGHT)
217        .try_collect()
218        .await?;
219        Ok(wire::VerifySmartContractWalletSignaturesResponse {
220            responses: chunks.into_iter().flatten().collect(),
221        })
222    }
223
224    /// Verify the signatures of one request. The snapshot it was sized by
225    /// can be above the deployment's current limits, so a refusal of the
226    /// whole request as too large sends its halves instead, one after the
227    /// other, down to single signatures. A single signature's refusal is
228    /// returned.
229    async fn verify_chunk(
230        &self,
231        signatures: Vec<wire::verify_smart_contract_wallet_signatures_request::Signature>,
232    ) -> Result<Vec<wire::verify_smart_contract_wallet_signatures_response::Response>> {
233        let mut verdicts = Vec::with_capacity(signatures.len());
234        // The next request is the last one pushed, so halves go out in order.
235        let mut pending = vec![signatures];
236        while let Some(signatures) = pending.pop() {
237            let request = wire::VerifySmartContractWalletSignaturesRequest { signatures };
238            let response = match self
239                .retry_call(
240                    || {
241                        self.api_client
242                            .verify_smart_contract_wallet_signatures(request.clone())
243                    },
244                    false,
245                )
246                .await
247            {
248                Ok(response) => response,
249                Err(error) if request.signatures.len() > 1 && refused_as_too_large(&error) => {
250                    let mut first = request.signatures;
251                    let second = first.split_off(first.len() / 2);
252                    pending.extend([second, first]);
253                    continue;
254                }
255                Err(error) => return Err(dyn_err(error)),
256            };
257            if response.responses.len() != request.signatures.len() {
258                return Err(ApiError::VerdictCount {
259                    expected: request.signatures.len(),
260                    got: response.responses.len(),
261                });
262            }
263            verdicts.extend(response.responses);
264        }
265        Ok(verdicts)
266    }
267}
268
269/// A refusal of a whole verify request that a smaller request can pass: the
270/// count limit (`INVALID_ARGUMENT`) or the byte limit (`OUT_OF_RANGE`). The
271/// backend also answers `INVALID_ARGUMENT` for one signature it refuses, so
272/// splitting ends at that signature with its own refusal.
273fn refused_as_too_large(error: &(dyn std::error::Error + 'static)) -> bool {
274    crate::preflight::failure(error).is_none()
275        && grpc_status(error).is_some_and(|status| {
276            matches!(
277                status.code(),
278                tonic::Code::InvalidArgument | tonic::Code::OutOfRange
279            )
280        })
281}
282
283/// Split `signatures` into verify requests the deployment accepts. A request
284/// ends at `per_request` signatures, or before the signature whose framed
285/// length would take its encoding past `max_request_bytes`. Every request
286/// holds at least one signature, so one that alone exceeds the byte bound is
287/// sent by itself.
288// implements: CONF-073
289fn verify_requests(
290    signatures: &[wire::verify_smart_contract_wallet_signatures_request::Signature],
291    per_request: usize,
292    max_request_bytes: usize,
293) -> Vec<&[wire::verify_smart_contract_wallet_signatures_request::Signature]> {
294    let mut requests = Vec::new();
295    let (mut start, mut bytes) = (0, 0);
296    for (end, signature) in signatures.iter().enumerate() {
297        let len = signature.encoded_len();
298        let framed = 1 + prost::length_delimiter_len(len) + len;
299        if end > start && (end - start >= per_request || bytes + framed > max_request_bytes) {
300            requests.push(&signatures[start..end]);
301            (start, bytes) = (end, 0);
302        }
303        bytes += framed;
304    }
305    if start < signatures.len() {
306        requests.push(&signatures[start..]);
307    }
308    requests
309}
xmtp_api/identity.rs

xmtp_api/
identity.rs
