xmtp_api/
identity.rs

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

xmtp_api/
identity.rs
