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 std::collections::HashMap;
8use xmtp_proto::{
9    api::grpc_status,
10    api_client::XmtpBackendClient,
11    backend_v1 as wire,
12    types::{ApiIdentifier, Cursor, IdentityUpdateLog, Topic, TopicCursor},
13    xmtp::identity::associations::{IdentifierKind, IdentityUpdate},
14};
15
16/// The handle rejection that a refused identity publish reports in its
17/// structured detail.
18fn handle_rejection(error: &ApiError) -> Option<HandleRejection> {
19    use wire::publish_error::Reason;
20    match publish_error_reason(error)? {
21        Reason::HandleTaken => Some(HandleRejection::Taken),
22        Reason::HandleNotAllowed => Some(HandleRejection::NotAllowed),
23        Reason::HandleTooSoon => Some(HandleRejection::TooSoon),
24        _ => None,
25    }
26}
27
28/// Read updates after the exclusive sequence id on this inbox topic.
29#[derive(Debug)]
30pub struct GetIdentityUpdatesV2Filter {
31    pub inbox_id: String,
32    pub sequence_id: Option<u64>,
33}
34
35impl<C: XmtpBackendClient> ApiClientWrapper<C> {
36    #[xmtp_common::rpc_span]
37    pub async fn publish_identity_update<U: Into<IdentityUpdate>>(
38        &self,
39        update: U,
40    ) -> Result<Cursor> {
41        let unit = PublishUnit::single_within(
42            wire::ClientEnvelope {
43                payload: Some(wire::client_envelope::Payload::IdentityUpdate(
44                    update.into(),
45                )),
46            },
47            self.limits(),
48        )?;
49        match self.publish_units(vec![unit]).await {
50            Ok(metas) => metas
51                .into_iter()
52                .next()
53                .and_then(|meta| meta.cursor)
54                .map(Into::into)
55                .ok_or(ApiError::InvalidResponse("identity publish cursor")),
56            Err(error)
57                if grpc_status(&error)
58                    .is_some_and(|status| status.code() == tonic::Code::Aborted) =>
59            {
60                Err(ApiError::IdentityUpdateConflict)
61            }
62            Err(error) => Err(handle_rejection(&error)
63                .map(ApiError::HandleRejected)
64                .unwrap_or(error)),
65        }
66    }
67    #[xmtp_common::rpc_span]
68    pub async fn get_identity_updates_v2(
69        &self,
70        filters: Vec<GetIdentityUpdatesV2Filter>,
71    ) -> Result<HashMap<String, Vec<IdentityUpdateLog>>> {
72        let mut cursors = TopicCursor::new();
73        let mut result = HashMap::new();
74        for filter in filters {
75            let bytes =
76                hex::decode(&filter.inbox_id).map_err(|_| ApiError::InvalidRequest("inbox id"))?;
77            let topic = Topic::new_identity_update(bytes);
78            Topic::parse(&topic)?;
79            cursors
80                .entry(topic)
81                .and_modify(|cursor| {
82                    *cursor = (*cursor).min(Cursor(filter.sequence_id.unwrap_or(0)))
83                })
84                .or_insert(Cursor(filter.sequence_id.unwrap_or(0)));
85            result.entry(filter.inbox_id).or_insert_with(Vec::new);
86        }
87        for envelope in self
88            .query_all(cursors, self.limits().max_query_limit as u32)
89            .await?
90        {
91            let update = xmtp_api_backend::envelope::decode_identity_update(envelope)?;
92            result
93                .get_mut(&update.update.inbox_id)
94                .ok_or(ApiError::InvalidResponse("unrequested inbox"))?
95                .push(update);
96        }
97        Ok(result)
98    }
99    /// Return one optional inbox id for every input, in the same order.
100    #[xmtp_common::rpc_span]
101    pub async fn get_inbox_ids(
102        &self,
103        identifiers: Vec<ApiIdentifier>,
104    ) -> Result<Vec<Option<String>>> {
105        if identifiers
106            .iter()
107            .any(|id| id.identifier_kind == IdentifierKind::Unspecified)
108        {
109            return Err(ApiError::InvalidResponse("unspecified identifier kind"));
110        }
111        let requests: Vec<_> = identifiers
112            .chunks(self.limits().max_lookup_identifiers)
113            .map(<[_]>::to_vec)
114            .collect();
115        let mut chunks: Vec<_> = stream::iter(requests.into_iter().enumerate().map(
116            |(index, chunk)| async move {
117                let request = wire::GetInboxIdsRequest {
118                    requests: chunk
119                        .iter()
120                        .map(|id| wire::get_inbox_ids_request::Request {
121                            identifier: id.identifier.clone(),
122                            identifier_kind: id.identifier_kind as i32,
123                        })
124                        .collect(),
125                };
126                let response = self
127                    .retry_call(|| self.api_client.get_inbox_ids(request.clone()), false)
128                    .await
129                    .map_err(dyn_err)?;
130                if response.responses.len() != chunk.len() {
131                    return Err(ApiError::InvalidResponse("inbox result count"));
132                }
133                let mut values = Vec::with_capacity(chunk.len());
134                for (response, input) in response.responses.into_iter().zip(&chunk) {
135                    if IdentifierKind::try_from(response.identifier_kind)
136                        .ok()
137                        .filter(|kind| *kind != IdentifierKind::Unspecified)
138                        != Some(input.identifier_kind)
139                        || response.identifier != input.identifier
140                    {
141                        return Err(ApiError::InvalidResponse("inbox result identity"));
142                    }
143                    values.push(response.inbox_id);
144                }
145                Ok((index, values))
146            },
147        ))
148        .buffer_unordered(MAX_READ_CHUNKS_IN_FLIGHT)
149        .try_collect()
150        .await?;
151        chunks.sort_by_key(|(index, _)| *index);
152        Ok(chunks.into_iter().flat_map(|(_, values)| values).collect())
153    }
154    #[xmtp_common::rpc_span]
155    pub async fn verify_smart_contract_wallet_signatures(
156        &self,
157        request: wire::VerifySmartContractWalletSignaturesRequest,
158    ) -> Result<wire::VerifySmartContractWalletSignaturesResponse> {
159        let requests: Vec<_> = request
160            .signatures
161            .chunks(self.limits().max_scw_signatures)
162            .map(<[_]>::to_vec)
163            .collect();
164        let mut chunks: Vec<_> = stream::iter(requests.into_iter().enumerate().map(
165            |(index, chunk)| async move {
166                let request = wire::VerifySmartContractWalletSignaturesRequest {
167                    signatures: chunk.to_vec(),
168                };
169                let response = self
170                    .retry_call(
171                        || {
172                            self.api_client
173                                .verify_smart_contract_wallet_signatures(request.clone())
174                        },
175                        false,
176                    )
177                    .await
178                    .map_err(dyn_err)?;
179                if response.responses.len() != chunk.len() {
180                    return Err(ApiError::InvalidResponse("signature result count"));
181                }
182                Ok((index, response.responses))
183            },
184        ))
185        .buffer_unordered(MAX_READ_CHUNKS_IN_FLIGHT)
186        .try_collect()
187        .await?;
188        chunks.sort_by_key(|(index, _)| *index);
189        Ok(wire::VerifySmartContractWalletSignaturesResponse {
190            responses: chunks.into_iter().flat_map(|(_, values)| values).collect(),
191        })
192    }
193}
xmtp_api/identity.rs

xmtp_api/
identity.rs
