aptos_sdk/api/
indexer.rs

1//! GraphQL indexer client.
2//!
3//! This module provides a client for querying the Aptos Indexer GraphQL API.
4//! The indexer provides access to indexed blockchain data including tokens,
5//! events, transaction history, and more.
6//!
7//! # Example
8//!
9//! ```rust,no_run
10//! use aptos_sdk::api::{IndexerClient, PaginationParams};
11//! use aptos_sdk::config::AptosConfig;
12//! use aptos_sdk::types::AccountAddress;
13//!
14//! #[tokio::main]
15//! async fn main() -> anyhow::Result<()> {
16//!     let config = AptosConfig::testnet();
17//!     let client = IndexerClient::new(&config)?;
18//!     
19//!     // Get fungible asset balances
20//!     let balances = client.get_fungible_asset_balances(AccountAddress::ONE).await?;
21//!     
22//!     // Get account tokens with pagination
23//!     let tokens = client.get_account_tokens_paginated(
24//!         AccountAddress::ONE,
25//!         Some(PaginationParams { limit: 10, offset: 0 }),
26//!     ).await?;
27//!     
28//!     Ok(())
29//! }
30//! ```
31
32use crate::config::AptosConfig;
33use crate::error::{AptosError, AptosResult};
34use crate::retry::{RetryConfig, RetryExecutor};
35use crate::types::AccountAddress;
36use reqwest::Client;
37use serde::{Deserialize, Serialize};
38use std::sync::Arc;
39use url::Url;
40
41/// Maximum indexer response size: 10 MB.
42const MAX_INDEXER_RESPONSE_SIZE: usize = 10 * 1024 * 1024;
43
44/// Client for the Aptos indexer GraphQL API.
45///
46/// The indexer provides access to indexed blockchain data including
47/// tokens, events, and transaction history. Queries are automatically
48/// retried with exponential backoff for transient failures.
49///
50/// # Example
51///
52/// ```rust,no_run
53/// use aptos_sdk::api::IndexerClient;
54/// use aptos_sdk::config::AptosConfig;
55///
56/// #[tokio::main]
57/// async fn main() -> anyhow::Result<()> {
58///     let config = AptosConfig::testnet();
59///     let client = IndexerClient::new(&config)?;
60///     // Use the client for GraphQL queries
61///     Ok(())
62/// }
63/// ```
64#[derive(Debug, Clone)]
65pub struct IndexerClient {
66    indexer_url: Url,
67    client: Client,
68    retry_config: Arc<RetryConfig>,
69}
70
71/// GraphQL request body.
72#[derive(Debug, Serialize)]
73struct GraphQLRequest {
74    query: String,
75    #[serde(skip_serializing_if = "Option::is_none")]
76    variables: Option<serde_json::Value>,
77}
78
79/// GraphQL response body.
80#[derive(Debug, Deserialize)]
81struct GraphQLResponse<T> {
82    data: Option<T>,
83    errors: Option<Vec<GraphQLError>>,
84}
85
86/// GraphQL error.
87#[derive(Debug, Deserialize)]
88struct GraphQLError {
89    message: String,
90}
91
92impl IndexerClient {
93    /// Creates a new indexer client.
94    ///
95    /// # TLS Security
96    ///
97    /// This client uses `reqwest` with its default TLS configuration, which
98    /// validates server certificates against the system's certificate store.
99    /// All Aptos indexer endpoints use HTTPS with valid certificates.
100    ///
101    /// # Errors
102    ///
103    /// Returns an error if the indexer URL is not configured in the config, or if the HTTP client
104    /// fails to build (e.g., invalid TLS configuration).
105    pub fn new(config: &AptosConfig) -> AptosResult<Self> {
106        let indexer_url = config
107            .indexer_url()
108            .cloned()
109            .ok_or_else(|| AptosError::Config("indexer URL not configured".into()))?;
110
111        let pool = config.pool_config();
112
113        // SECURITY: TLS certificate validation is enabled by default via reqwest.
114        let mut builder = Client::builder()
115            .timeout(config.timeout)
116            .pool_max_idle_per_host(pool.max_idle_per_host.unwrap_or(usize::MAX))
117            .pool_idle_timeout(pool.idle_timeout)
118            .tcp_nodelay(pool.tcp_nodelay);
119
120        if let Some(keepalive) = pool.tcp_keepalive {
121            builder = builder.tcp_keepalive(keepalive);
122        }
123
124        let client = builder.build().map_err(AptosError::Http)?;
125
126        let retry_config = Arc::new(config.retry_config().clone());
127
128        Ok(Self {
129            indexer_url,
130            client,
131            retry_config,
132        })
133    }
134
135    /// Creates an indexer client with a custom URL.
136    ///
137    /// # Errors
138    ///
139    /// Returns an error if the URL cannot be parsed.
140    pub fn with_url(url: &str) -> AptosResult<Self> {
141        let indexer_url = Url::parse(url)?;
142        // SECURITY: Validate URL scheme to prevent SSRF via dangerous protocols
143        crate::config::validate_url_scheme(&indexer_url)?;
144        let client = Client::new();
145        Ok(Self {
146            indexer_url,
147            client,
148            retry_config: Arc::new(RetryConfig::default()),
149        })
150    }
151
152    /// Executes a GraphQL query.
153    ///
154    /// # Errors
155    ///
156    /// Returns an error if the HTTP request fails, the API returns an error status code,
157    /// the response cannot be parsed as JSON, the GraphQL query contains errors, or the
158    /// response data is missing.
159    pub async fn query<T: for<'de> Deserialize<'de> + Send + 'static>(
160        &self,
161        query: &str,
162        variables: Option<serde_json::Value>,
163    ) -> AptosResult<T> {
164        let request = GraphQLRequest {
165            query: query.to_string(),
166            variables,
167        };
168
169        let client = self.client.clone();
170        let url = self.indexer_url.clone();
171        let retry_config = self.retry_config.clone();
172
173        let executor = RetryExecutor::from_shared(retry_config);
174        executor
175            .execute(|| {
176                let client = client.clone();
177                let url = url.clone();
178                let request = GraphQLRequest {
179                    query: request.query.clone(),
180                    variables: request.variables.clone(),
181                };
182                async move {
183                    let response = client.post(url.as_str()).json(&request).send().await?;
184
185                    if response.status().is_success() {
186                        // SECURITY: Stream body with size limit to prevent OOM
187                        // from malicious responses (including chunked encoding).
188                        let bytes = crate::config::read_response_bounded(
189                            response,
190                            MAX_INDEXER_RESPONSE_SIZE,
191                        )
192                        .await?;
193                        let graphql_response: GraphQLResponse<T> = serde_json::from_slice(&bytes)?;
194
195                        if let Some(errors) = graphql_response.errors {
196                            // Build error message directly without intermediate Vec
197                            let mut message = String::new();
198                            for (i, e) in errors.iter().enumerate() {
199                                if i > 0 {
200                                    message.push_str("; ");
201                                }
202                                message.push_str(&e.message);
203                            }
204                            return Err(AptosError::Api {
205                                status_code: 400,
206                                message,
207                                error_code: Some("GRAPHQL_ERROR".into()),
208                                vm_error_code: None,
209                            });
210                        }
211
212                        graphql_response.data.ok_or_else(|| {
213                            AptosError::Internal("no data in GraphQL response".into())
214                        })
215                    } else {
216                        let status = response.status();
217                        let body = response.text().await.unwrap_or_default();
218                        Err(AptosError::api(status.as_u16(), body))
219                    }
220                }
221            })
222            .await
223    }
224
225    /// Gets the account's fungible asset balances.
226    ///
227    /// # Errors
228    ///
229    /// Returns an error if the GraphQL query fails (see [`query`](Self::query) for details).
230    pub async fn get_fungible_asset_balances(
231        &self,
232        address: AccountAddress,
233    ) -> AptosResult<Vec<FungibleAssetBalance>> {
234        #[derive(Deserialize)]
235        struct Response {
236            current_fungible_asset_balances: Vec<FungibleAssetBalance>,
237        }
238
239        let query = r"
240            query GetFungibleAssetBalances($address: String!) {
241                current_fungible_asset_balances(
242                    where: { owner_address: { _eq: $address } }
243                ) {
244                    asset_type
245                    amount
246                    metadata {
247                        name
248                        symbol
249                        decimals
250                    }
251                }
252            }
253        ";
254
255        let variables = serde_json::json!({
256            "address": address.to_string()
257        });
258
259        let response: Response = self.query(query, Some(variables)).await?;
260        Ok(response.current_fungible_asset_balances)
261    }
262
263    /// Gets the account's token (NFT) holdings.
264    ///
265    /// # Errors
266    ///
267    /// Returns an error if the GraphQL query fails (see [`query`](Self::query) for details).
268    pub async fn get_account_tokens(
269        &self,
270        address: AccountAddress,
271    ) -> AptosResult<Vec<TokenBalance>> {
272        #[derive(Deserialize)]
273        struct Response {
274            current_token_ownerships_v2: Vec<TokenBalance>,
275        }
276
277        let query = r"
278            query GetAccountTokens($address: String!) {
279                current_token_ownerships_v2(
280                    where: { owner_address: { _eq: $address }, amount: { _gt: 0 } }
281                ) {
282                    token_data_id
283                    amount
284                    current_token_data {
285                        token_name
286                        description
287                        token_uri
288                        current_collection {
289                            collection_name
290                        }
291                    }
292                }
293            }
294        ";
295
296        let variables = serde_json::json!({
297            "address": address.to_string()
298        });
299
300        let response: Response = self.query(query, Some(variables)).await?;
301        Ok(response.current_token_ownerships_v2)
302    }
303
304    /// Gets recent transactions for an account.
305    ///
306    /// # Errors
307    ///
308    /// Returns an error if the GraphQL query fails (see [`query`](Self::query) for details).
309    pub async fn get_account_transactions(
310        &self,
311        address: AccountAddress,
312        limit: Option<u32>,
313    ) -> AptosResult<Vec<Transaction>> {
314        #[derive(Deserialize)]
315        struct Response {
316            account_transactions: Vec<Transaction>,
317        }
318
319        let query = r"
320            query GetAccountTransactions($address: String!, $limit: Int!) {
321                account_transactions(
322                    where: { account_address: { _eq: $address } }
323                    order_by: { transaction_version: desc }
324                    limit: $limit
325                ) {
326                    transaction_version
327                    coin_activities {
328                        activity_type
329                        amount
330                        coin_type
331                    }
332                }
333            }
334        ";
335
336        let variables = serde_json::json!({
337            "address": address.to_string(),
338            "limit": limit.unwrap_or(25)
339        });
340
341        let response: Response = self.query(query, Some(variables)).await?;
342        Ok(response.account_transactions)
343    }
344}
345
346/// Fungible asset balance from the indexer.
347#[derive(Debug, Clone, Deserialize)]
348pub struct FungibleAssetBalance {
349    /// The asset type.
350    pub asset_type: String,
351    /// The balance amount.
352    pub amount: String,
353    /// Asset metadata.
354    pub metadata: Option<FungibleAssetMetadata>,
355}
356
357/// Fungible asset metadata from the indexer.
358#[derive(Debug, Clone, Deserialize)]
359pub struct FungibleAssetMetadata {
360    /// Asset name.
361    pub name: String,
362    /// Asset symbol.
363    pub symbol: String,
364    /// Number of decimals.
365    pub decimals: u8,
366}
367
368/// Token (NFT) balance from the indexer.
369#[derive(Debug, Clone, Deserialize)]
370pub struct TokenBalance {
371    /// The token data ID.
372    pub token_data_id: String,
373    /// Amount owned.
374    pub amount: String,
375    /// Token data.
376    pub current_token_data: Option<TokenData>,
377}
378
379/// Token data from the indexer.
380#[derive(Debug, Clone, Deserialize)]
381pub struct TokenData {
382    /// Token name.
383    pub token_name: String,
384    /// Token description.
385    pub description: String,
386    /// Token URI.
387    pub token_uri: String,
388    /// Collection data.
389    pub current_collection: Option<CollectionData>,
390}
391
392/// Collection data from the indexer.
393#[derive(Debug, Clone, Deserialize)]
394pub struct CollectionData {
395    /// Collection name.
396    pub collection_name: String,
397}
398
399/// Transaction from the indexer.
400#[derive(Debug, Clone, Deserialize)]
401pub struct Transaction {
402    /// Transaction version.
403    pub transaction_version: String,
404    /// Coin activities in this transaction.
405    pub coin_activities: Vec<CoinActivity>,
406}
407
408/// Coin activity from the indexer.
409#[derive(Debug, Clone, Deserialize)]
410pub struct CoinActivity {
411    /// Activity type.
412    pub activity_type: String,
413    /// Amount.
414    pub amount: Option<String>,
415    /// Coin type.
416    pub coin_type: String,
417}
418
419/// Pagination parameters for indexer queries.
420#[derive(Debug, Clone, Default)]
421pub struct PaginationParams {
422    /// Maximum number of items to return.
423    pub limit: u32,
424    /// Number of items to skip.
425    pub offset: u32,
426}
427
428impl PaginationParams {
429    /// Creates new pagination parameters.
430    pub fn new(limit: u32, offset: u32) -> Self {
431        Self { limit, offset }
432    }
433
434    /// Creates pagination for the first page.
435    pub fn first(limit: u32) -> Self {
436        Self { limit, offset: 0 }
437    }
438}
439
440/// A paginated response.
441#[derive(Debug, Clone)]
442pub struct Page<T> {
443    /// The items in this page.
444    pub items: Vec<T>,
445    /// Whether there are more items.
446    pub has_more: bool,
447    /// Total count if available.
448    pub total_count: Option<u64>,
449}
450
451/// Event from the indexer.
452#[derive(Debug, Clone, Deserialize)]
453pub struct Event {
454    /// Event sequence number.
455    pub sequence_number: String,
456    /// Event type.
457    #[serde(rename = "type")]
458    pub event_type: String,
459    /// Event data.
460    pub data: serde_json::Value,
461    /// Transaction version that emitted this event.
462    pub transaction_version: Option<String>,
463    /// Account address associated with the event.
464    pub account_address: Option<String>,
465    /// Creation number.
466    pub creation_number: Option<String>,
467}
468
469/// Collection data from the indexer.
470#[derive(Debug, Clone, Deserialize)]
471pub struct Collection {
472    /// Collection address.
473    pub collection_id: String,
474    /// Collection name.
475    pub collection_name: String,
476    /// Creator address.
477    pub creator_address: String,
478    /// Current supply.
479    pub current_supply: String,
480    /// Maximum supply (0 = unlimited).
481    pub max_supply: Option<String>,
482    /// Collection URI.
483    pub uri: String,
484    /// Description.
485    pub description: String,
486}
487
488/// Coin balance from the indexer (legacy coin module).
489#[derive(Debug, Clone, Deserialize)]
490pub struct CoinBalance {
491    /// Coin type.
492    pub coin_type: String,
493    /// Balance amount.
494    pub amount: String,
495}
496
497/// Processor status from the indexer.
498#[derive(Debug, Clone, Deserialize)]
499pub struct ProcessorStatus {
500    /// Processor name.
501    pub processor: String,
502    /// Last successfully processed version.
503    pub last_success_version: u64,
504    /// Last updated timestamp.
505    pub last_updated: Option<String>,
506}
507
508impl IndexerClient {
509    // ... existing methods ...
510
511    /// Gets the account's token (NFT) holdings with pagination.
512    ///
513    /// # Errors
514    ///
515    /// Returns an error if the GraphQL query fails (see [`query`](Self::query) for details).
516    pub async fn get_account_tokens_paginated(
517        &self,
518        address: AccountAddress,
519        pagination: Option<PaginationParams>,
520    ) -> AptosResult<Page<TokenBalance>> {
521        #[derive(Deserialize)]
522        struct AggregateCount {
523            count: u64,
524        }
525
526        #[derive(Deserialize)]
527        struct Aggregate {
528            aggregate: Option<AggregateCount>,
529        }
530
531        #[derive(Deserialize)]
532        struct Response {
533            current_token_ownerships_v2: Vec<TokenBalance>,
534            current_token_ownerships_v2_aggregate: Aggregate,
535        }
536
537        let pagination = pagination.unwrap_or(PaginationParams {
538            limit: 25,
539            offset: 0,
540        });
541
542        let query = r"
543            query GetAccountTokens($address: String!, $limit: Int!, $offset: Int!) {
544                current_token_ownerships_v2(
545                    where: { owner_address: { _eq: $address }, amount: { _gt: 0 } }
546                    limit: $limit
547                    offset: $offset
548                ) {
549                    token_data_id
550                    amount
551                    current_token_data {
552                        token_name
553                        description
554                        token_uri
555                        current_collection {
556                            collection_name
557                        }
558                    }
559                }
560                current_token_ownerships_v2_aggregate(
561                    where: { owner_address: { _eq: $address }, amount: { _gt: 0 } }
562                ) {
563                    aggregate {
564                        count
565                    }
566                }
567            }
568        ";
569
570        let variables = serde_json::json!({
571            "address": address.to_string(),
572            "limit": pagination.limit,
573            "offset": pagination.offset
574        });
575
576        let response: Response = self.query(query, Some(variables)).await?;
577        let total_count = response
578            .current_token_ownerships_v2_aggregate
579            .aggregate
580            .map(|a| a.count);
581        let has_more = total_count.is_some_and(|total| {
582            (u64::from(pagination.offset) + response.current_token_ownerships_v2.len() as u64)
583                < total
584        });
585
586        Ok(Page {
587            items: response.current_token_ownerships_v2,
588            has_more,
589            total_count,
590        })
591    }
592
593    /// Gets the account's transaction history with pagination.
594    ///
595    /// # Errors
596    ///
597    /// Returns an error if the GraphQL query fails (see [`query`](Self::query) for details).
598    pub async fn get_account_transactions_paginated(
599        &self,
600        address: AccountAddress,
601        pagination: Option<PaginationParams>,
602    ) -> AptosResult<Page<Transaction>> {
603        #[derive(Deserialize)]
604        struct AggregateCount {
605            count: u64,
606        }
607
608        #[derive(Deserialize)]
609        struct Aggregate {
610            aggregate: Option<AggregateCount>,
611        }
612
613        #[derive(Deserialize)]
614        struct Response {
615            account_transactions: Vec<Transaction>,
616            account_transactions_aggregate: Aggregate,
617        }
618
619        let pagination = pagination.unwrap_or(PaginationParams {
620            limit: 25,
621            offset: 0,
622        });
623
624        let query = r"
625            query GetAccountTransactions($address: String!, $limit: Int!, $offset: Int!) {
626                account_transactions(
627                    where: { account_address: { _eq: $address } }
628                    order_by: { transaction_version: desc }
629                    limit: $limit
630                    offset: $offset
631                ) {
632                    transaction_version
633                    coin_activities {
634                        activity_type
635                        amount
636                        coin_type
637                    }
638                }
639                account_transactions_aggregate(
640                    where: { account_address: { _eq: $address } }
641                ) {
642                    aggregate {
643                        count
644                    }
645                }
646            }
647        ";
648
649        let variables = serde_json::json!({
650            "address": address.to_string(),
651            "limit": pagination.limit,
652            "offset": pagination.offset
653        });
654
655        let response: Response = self.query(query, Some(variables)).await?;
656        let total_count = response
657            .account_transactions_aggregate
658            .aggregate
659            .map(|a| a.count);
660        let has_more = total_count.is_some_and(|total| {
661            (u64::from(pagination.offset) + response.account_transactions.len() as u64) < total
662        });
663
664        Ok(Page {
665            items: response.account_transactions,
666            has_more,
667            total_count,
668        })
669    }
670
671    /// Gets events by type.
672    ///
673    /// # Errors
674    ///
675    /// Returns an error if the GraphQL query fails (see [`query`](Self::query) for details).
676    pub async fn get_events_by_type(
677        &self,
678        event_type: &str,
679        limit: Option<u32>,
680    ) -> AptosResult<Vec<Event>> {
681        #[derive(Deserialize)]
682        struct Response {
683            events: Vec<Event>,
684        }
685
686        let query = r"
687            query GetEventsByType($type: String!, $limit: Int!) {
688                events(
689                    where: { type: { _eq: $type } }
690                    order_by: { transaction_version: desc }
691                    limit: $limit
692                ) {
693                    sequence_number
694                    type
695                    data
696                    transaction_version
697                    account_address
698                    creation_number
699                }
700            }
701        ";
702
703        let variables = serde_json::json!({
704            "type": event_type,
705            "limit": limit.unwrap_or(25)
706        });
707
708        let response: Response = self.query(query, Some(variables)).await?;
709        Ok(response.events)
710    }
711
712    /// Gets events involving an account.
713    ///
714    /// # Errors
715    ///
716    /// Returns an error if the GraphQL query fails (see [`query`](Self::query) for details).
717    pub async fn get_events_by_account(
718        &self,
719        address: AccountAddress,
720        limit: Option<u32>,
721    ) -> AptosResult<Vec<Event>> {
722        #[derive(Deserialize)]
723        struct Response {
724            events: Vec<Event>,
725        }
726
727        let query = r"
728            query GetEventsByAccount($address: String!, $limit: Int!) {
729                events(
730                    where: { account_address: { _eq: $address } }
731                    order_by: { transaction_version: desc }
732                    limit: $limit
733                ) {
734                    sequence_number
735                    type
736                    data
737                    transaction_version
738                    account_address
739                    creation_number
740                }
741            }
742        ";
743
744        let variables = serde_json::json!({
745            "address": address.to_string(),
746            "limit": limit.unwrap_or(25)
747        });
748
749        let response: Response = self.query(query, Some(variables)).await?;
750        Ok(response.events)
751    }
752
753    /// Gets a collection by its address.
754    ///
755    /// # Errors
756    ///
757    /// Returns an error if the GraphQL query fails (see [`query`](Self::query) for details),
758    /// or if the collection is not found.
759    pub async fn get_collection(
760        &self,
761        collection_address: AccountAddress,
762    ) -> AptosResult<Collection> {
763        #[derive(Deserialize)]
764        struct Response {
765            current_collections_v2: Vec<Collection>,
766        }
767
768        let query = r"
769            query GetCollection($address: String!) {
770                current_collections_v2(
771                    where: { collection_id: { _eq: $address } }
772                    limit: 1
773                ) {
774                    collection_id
775                    collection_name
776                    creator_address
777                    current_supply
778                    max_supply
779                    uri
780                    description
781                }
782            }
783        ";
784
785        let variables = serde_json::json!({
786            "address": collection_address.to_string()
787        });
788
789        let response: Response = self.query(query, Some(variables)).await?;
790        response
791            .current_collections_v2
792            .into_iter()
793            .next()
794            .ok_or_else(|| {
795                AptosError::NotFound(format!("Collection not found: {collection_address}"))
796            })
797    }
798
799    /// Gets tokens in a collection.
800    ///
801    /// # Errors
802    ///
803    /// Returns an error if the GraphQL query fails (see [`query`](Self::query) for details).
804    pub async fn get_collection_tokens(
805        &self,
806        collection_address: AccountAddress,
807        pagination: Option<PaginationParams>,
808    ) -> AptosResult<Page<TokenBalance>> {
809        #[derive(Deserialize)]
810        struct Response {
811            current_token_ownerships_v2: Vec<TokenBalance>,
812        }
813
814        let pagination = pagination.unwrap_or(PaginationParams {
815            limit: 25,
816            offset: 0,
817        });
818
819        let query = r"
820            query GetCollectionTokens($address: String!, $limit: Int!, $offset: Int!) {
821                current_token_ownerships_v2(
822                    where: { 
823                        current_token_data: { 
824                            current_collection: { 
825                                collection_id: { _eq: $address } 
826                            } 
827                        }
828                        amount: { _gt: 0 }
829                    }
830                    limit: $limit
831                    offset: $offset
832                ) {
833                    token_data_id
834                    amount
835                    current_token_data {
836                        token_name
837                        description
838                        token_uri
839                        current_collection {
840                            collection_name
841                        }
842                    }
843                }
844            }
845        ";
846
847        let variables = serde_json::json!({
848            "address": collection_address.to_string(),
849            "limit": pagination.limit,
850            "offset": pagination.offset
851        });
852
853        let response: Response = self.query(query, Some(variables)).await?;
854        let items_count = response.current_token_ownerships_v2.len();
855
856        Ok(Page {
857            items: response.current_token_ownerships_v2,
858            has_more: items_count == pagination.limit as usize,
859            total_count: None,
860        })
861    }
862
863    /// Gets coin balances for an account (legacy coin module).
864    ///
865    /// # Errors
866    ///
867    /// Returns an error if the GraphQL query fails (see [`query`](Self::query) for details).
868    pub async fn get_coin_balances(
869        &self,
870        address: AccountAddress,
871    ) -> AptosResult<Vec<CoinBalance>> {
872        #[derive(Deserialize)]
873        struct Response {
874            current_coin_balances: Vec<CoinBalance>,
875        }
876
877        let query = r"
878            query GetCoinBalances($address: String!) {
879                current_coin_balances(
880                    where: { owner_address: { _eq: $address } }
881                ) {
882                    coin_type
883                    amount
884                }
885            }
886        ";
887
888        let variables = serde_json::json!({
889            "address": address.to_string()
890        });
891
892        let response: Response = self.query(query, Some(variables)).await?;
893        Ok(response.current_coin_balances)
894    }
895
896    /// Gets coin activities for an account.
897    ///
898    /// # Errors
899    ///
900    /// Returns an error if the GraphQL query fails (see [`query`](Self::query) for details).
901    pub async fn get_coin_activities(
902        &self,
903        address: AccountAddress,
904        limit: Option<u32>,
905    ) -> AptosResult<Vec<CoinActivity>> {
906        #[derive(Deserialize)]
907        struct Response {
908            coin_activities: Vec<CoinActivity>,
909        }
910
911        let query = r"
912            query GetCoinActivities($address: String!, $limit: Int!) {
913                coin_activities(
914                    where: { owner_address: { _eq: $address } }
915                    order_by: { transaction_version: desc }
916                    limit: $limit
917                ) {
918                    activity_type
919                    amount
920                    coin_type
921                }
922            }
923        ";
924
925        let variables = serde_json::json!({
926            "address": address.to_string(),
927            "limit": limit.unwrap_or(25)
928        });
929
930        let response: Response = self.query(query, Some(variables)).await?;
931        Ok(response.coin_activities)
932    }
933
934    /// Gets the processor status to check indexer health.
935    ///
936    /// # Errors
937    ///
938    /// Returns an error if the GraphQL query fails (see [`query`](Self::query) for details).
939    pub async fn get_processor_status(&self) -> AptosResult<Vec<ProcessorStatus>> {
940        #[derive(Deserialize)]
941        struct Response {
942            processor_status: Vec<ProcessorStatus>,
943        }
944
945        let query = r"
946            query GetProcessorStatus {
947                processor_status {
948                    processor
949                    last_success_version
950                    last_updated
951                }
952            }
953        ";
954
955        let response: Response = self.query(query, None).await?;
956        Ok(response.processor_status)
957    }
958
959    /// Gets the current indexer version (last processed transaction).
960    ///
961    /// # Errors
962    ///
963    /// Returns an error if the processor status cannot be fetched, or if no processor status
964    /// is available.
965    pub async fn get_indexer_version(&self) -> AptosResult<u64> {
966        let statuses = self.get_processor_status().await?;
967        statuses
968            .into_iter()
969            .map(|s| s.last_success_version)
970            .max()
971            .ok_or_else(|| AptosError::Internal("No processor status available".into()))
972    }
973
974    /// Checks if the indexer is healthy by comparing with a reference version.
975    ///
976    /// # Errors
977    ///
978    /// Returns an error if the indexer version cannot be fetched (see [`get_indexer_version`](Self::get_indexer_version) for details).
979    pub async fn check_indexer_lag(
980        &self,
981        reference_version: u64,
982        max_lag: u64,
983    ) -> AptosResult<bool> {
984        let indexer_version = self.get_indexer_version().await?;
985        Ok(reference_version.saturating_sub(indexer_version) <= max_lag)
986    }
987}
988
989#[cfg(test)]
990mod tests {
991    use super::*;
992    use wiremock::{Mock, MockServer, ResponseTemplate, matchers::method};
993
994    /// Builds an [`IndexerClient`] pointed at a wiremock server.
995    fn mock_client(server: &MockServer) -> IndexerClient {
996        IndexerClient::with_url(&server.uri()).unwrap()
997    }
998
999    /// Mounts a single mock returning `body` as a 200 JSON response to any POST.
1000    async fn mount_json(server: &MockServer, body: serde_json::Value) {
1001        Mock::given(method("POST"))
1002            .respond_with(ResponseTemplate::new(200).set_body_json(body))
1003            .expect(1)
1004            .mount(server)
1005            .await;
1006    }
1007
1008    #[tokio::test]
1009    async fn test_get_fungible_asset_balances() {
1010        let server = MockServer::start().await;
1011        mount_json(
1012            &server,
1013            serde_json::json!({
1014                "data": {
1015                    "current_fungible_asset_balances": [
1016                        {
1017                            "asset_type": "0x1::aptos_coin::AptosCoin",
1018                            "amount": "1000000",
1019                            "metadata": {
1020                                "name": "Aptos Coin",
1021                                "symbol": "APT",
1022                                "decimals": 8
1023                            }
1024                        }
1025                    ]
1026                }
1027            }),
1028        )
1029        .await;
1030
1031        let client = mock_client(&server);
1032        let balances = client
1033            .get_fungible_asset_balances(AccountAddress::ONE)
1034            .await
1035            .unwrap();
1036
1037        assert_eq!(balances.len(), 1);
1038        assert_eq!(balances[0].asset_type, "0x1::aptos_coin::AptosCoin");
1039        assert_eq!(balances[0].amount, "1000000");
1040        let metadata = balances[0].metadata.as_ref().unwrap();
1041        assert_eq!(metadata.name, "Aptos Coin");
1042        assert_eq!(metadata.symbol, "APT");
1043        assert_eq!(metadata.decimals, 8);
1044    }
1045
1046    #[tokio::test]
1047    async fn test_get_account_tokens() {
1048        let server = MockServer::start().await;
1049        mount_json(
1050            &server,
1051            serde_json::json!({
1052                "data": {
1053                    "current_token_ownerships_v2": [
1054                        {
1055                            "token_data_id": "0xabc",
1056                            "amount": "1",
1057                            "current_token_data": {
1058                                "token_name": "Cool NFT",
1059                                "description": "A cool token",
1060                                "token_uri": "https://example.com/1",
1061                                "current_collection": {
1062                                    "collection_name": "Cool Collection"
1063                                }
1064                            }
1065                        }
1066                    ]
1067                }
1068            }),
1069        )
1070        .await;
1071
1072        let client = mock_client(&server);
1073        let tokens = client
1074            .get_account_tokens(AccountAddress::ONE)
1075            .await
1076            .unwrap();
1077
1078        assert_eq!(tokens.len(), 1);
1079        assert_eq!(tokens[0].token_data_id, "0xabc");
1080        assert_eq!(tokens[0].amount, "1");
1081        let data = tokens[0].current_token_data.as_ref().unwrap();
1082        assert_eq!(data.token_name, "Cool NFT");
1083        assert_eq!(data.description, "A cool token");
1084        assert_eq!(data.token_uri, "https://example.com/1");
1085        assert_eq!(
1086            data.current_collection.as_ref().unwrap().collection_name,
1087            "Cool Collection"
1088        );
1089    }
1090
1091    #[tokio::test]
1092    async fn test_get_account_transactions() {
1093        let server = MockServer::start().await;
1094        mount_json(
1095            &server,
1096            serde_json::json!({
1097                "data": {
1098                    "account_transactions": [
1099                        {
1100                            "transaction_version": "12345",
1101                            "coin_activities": [
1102                                {
1103                                    "activity_type": "0x1::coin::WithdrawEvent",
1104                                    "amount": "500",
1105                                    "coin_type": "0x1::aptos_coin::AptosCoin"
1106                                }
1107                            ]
1108                        }
1109                    ]
1110                }
1111            }),
1112        )
1113        .await;
1114
1115        let client = mock_client(&server);
1116        let txns = client
1117            .get_account_transactions(AccountAddress::ONE, Some(10))
1118            .await
1119            .unwrap();
1120
1121        assert_eq!(txns.len(), 1);
1122        assert_eq!(txns[0].transaction_version, "12345");
1123        assert_eq!(txns[0].coin_activities.len(), 1);
1124        let activity = &txns[0].coin_activities[0];
1125        assert_eq!(activity.activity_type, "0x1::coin::WithdrawEvent");
1126        assert_eq!(activity.amount.as_deref(), Some("500"));
1127        assert_eq!(activity.coin_type, "0x1::aptos_coin::AptosCoin");
1128    }
1129
1130    #[tokio::test]
1131    async fn test_get_account_tokens_paginated_has_more() {
1132        let server = MockServer::start().await;
1133        mount_json(
1134            &server,
1135            serde_json::json!({
1136                "data": {
1137                    "current_token_ownerships_v2": [
1138                        {
1139                            "token_data_id": "0x1",
1140                            "amount": "1",
1141                            "current_token_data": null
1142                        },
1143                        {
1144                            "token_data_id": "0x2",
1145                            "amount": "1",
1146                            "current_token_data": null
1147                        }
1148                    ],
1149                    "current_token_ownerships_v2_aggregate": {
1150                        "aggregate": { "count": 5 }
1151                    }
1152                }
1153            }),
1154        )
1155        .await;
1156
1157        let client = mock_client(&server);
1158        let page = client
1159            .get_account_tokens_paginated(AccountAddress::ONE, Some(PaginationParams::new(2, 0)))
1160            .await
1161            .unwrap();
1162
1163        assert_eq!(page.items.len(), 2);
1164        assert_eq!(page.total_count, Some(5));
1165        // offset 0 + 2 items < 5 => more pages remain.
1166        assert!(page.has_more);
1167    }
1168
1169    #[tokio::test]
1170    async fn test_get_account_tokens_paginated_no_more() {
1171        let server = MockServer::start().await;
1172        mount_json(
1173            &server,
1174            serde_json::json!({
1175                "data": {
1176                    "current_token_ownerships_v2": [
1177                        {
1178                            "token_data_id": "0x1",
1179                            "amount": "1",
1180                            "current_token_data": null
1181                        }
1182                    ],
1183                    "current_token_ownerships_v2_aggregate": {
1184                        "aggregate": { "count": 1 }
1185                    }
1186                }
1187            }),
1188        )
1189        .await;
1190
1191        let client = mock_client(&server);
1192        // Default pagination (None) is used here.
1193        let page = client
1194            .get_account_tokens_paginated(AccountAddress::ONE, None)
1195            .await
1196            .unwrap();
1197
1198        assert_eq!(page.items.len(), 1);
1199        assert_eq!(page.total_count, Some(1));
1200        assert!(!page.has_more);
1201    }
1202
1203    #[tokio::test]
1204    async fn test_get_account_transactions_paginated() {
1205        let server = MockServer::start().await;
1206        mount_json(
1207            &server,
1208            serde_json::json!({
1209                "data": {
1210                    "account_transactions": [
1211                        {
1212                            "transaction_version": "100",
1213                            "coin_activities": []
1214                        }
1215                    ],
1216                    "account_transactions_aggregate": {
1217                        "aggregate": { "count": 3 }
1218                    }
1219                }
1220            }),
1221        )
1222        .await;
1223
1224        let client = mock_client(&server);
1225        let page = client
1226            .get_account_transactions_paginated(
1227                AccountAddress::ONE,
1228                Some(PaginationParams::first(1)),
1229            )
1230            .await
1231            .unwrap();
1232
1233        assert_eq!(page.items.len(), 1);
1234        assert_eq!(page.items[0].transaction_version, "100");
1235        assert_eq!(page.total_count, Some(3));
1236        assert!(page.has_more);
1237    }
1238
1239    #[tokio::test]
1240    async fn test_get_events_by_type() {
1241        let server = MockServer::start().await;
1242        mount_json(
1243            &server,
1244            serde_json::json!({
1245                "data": {
1246                    "events": [
1247                        {
1248                            "sequence_number": "7",
1249                            "type": "0x1::coin::DepositEvent",
1250                            "data": { "amount": "42" },
1251                            "transaction_version": "9001",
1252                            "account_address": "0x1",
1253                            "creation_number": "3"
1254                        }
1255                    ]
1256                }
1257            }),
1258        )
1259        .await;
1260
1261        let client = mock_client(&server);
1262        let events = client
1263            .get_events_by_type("0x1::coin::DepositEvent", Some(5))
1264            .await
1265            .unwrap();
1266
1267        assert_eq!(events.len(), 1);
1268        assert_eq!(events[0].sequence_number, "7");
1269        assert_eq!(events[0].event_type, "0x1::coin::DepositEvent");
1270        assert_eq!(events[0].data["amount"], "42");
1271        assert_eq!(events[0].transaction_version.as_deref(), Some("9001"));
1272        assert_eq!(events[0].account_address.as_deref(), Some("0x1"));
1273        assert_eq!(events[0].creation_number.as_deref(), Some("3"));
1274    }
1275
1276    #[tokio::test]
1277    async fn test_get_events_by_account() {
1278        let server = MockServer::start().await;
1279        mount_json(
1280            &server,
1281            serde_json::json!({
1282                "data": {
1283                    "events": [
1284                        {
1285                            "sequence_number": "1",
1286                            "type": "0x1::account::CoinRegisterEvent",
1287                            "data": {},
1288                            "transaction_version": null,
1289                            "account_address": null,
1290                            "creation_number": null
1291                        }
1292                    ]
1293                }
1294            }),
1295        )
1296        .await;
1297
1298        let client = mock_client(&server);
1299        let events = client
1300            .get_events_by_account(AccountAddress::ONE, None)
1301            .await
1302            .unwrap();
1303
1304        assert_eq!(events.len(), 1);
1305        assert_eq!(events[0].event_type, "0x1::account::CoinRegisterEvent");
1306        assert!(events[0].transaction_version.is_none());
1307        assert!(events[0].account_address.is_none());
1308        assert!(events[0].creation_number.is_none());
1309    }
1310
1311    #[tokio::test]
1312    async fn test_get_collection() {
1313        let server = MockServer::start().await;
1314        mount_json(
1315            &server,
1316            serde_json::json!({
1317                "data": {
1318                    "current_collections_v2": [
1319                        {
1320                            "collection_id": "0xcol",
1321                            "collection_name": "My Collection",
1322                            "creator_address": "0x1",
1323                            "current_supply": "10",
1324                            "max_supply": "100",
1325                            "uri": "https://example.com/collection",
1326                            "description": "A test collection"
1327                        }
1328                    ]
1329                }
1330            }),
1331        )
1332        .await;
1333
1334        let client = mock_client(&server);
1335        let collection = client.get_collection(AccountAddress::ONE).await.unwrap();
1336
1337        assert_eq!(collection.collection_id, "0xcol");
1338        assert_eq!(collection.collection_name, "My Collection");
1339        assert_eq!(collection.creator_address, "0x1");
1340        assert_eq!(collection.current_supply, "10");
1341        assert_eq!(collection.max_supply.as_deref(), Some("100"));
1342        assert_eq!(collection.uri, "https://example.com/collection");
1343        assert_eq!(collection.description, "A test collection");
1344    }
1345
1346    #[tokio::test]
1347    async fn test_get_collection_not_found() {
1348        let server = MockServer::start().await;
1349        mount_json(
1350            &server,
1351            serde_json::json!({
1352                "data": { "current_collections_v2": [] }
1353            }),
1354        )
1355        .await;
1356
1357        let client = mock_client(&server);
1358        let result = client.get_collection(AccountAddress::ONE).await;
1359
1360        assert!(result.is_err());
1361        assert!(result.unwrap_err().is_not_found());
1362    }
1363
1364    #[tokio::test]
1365    async fn test_get_collection_tokens() {
1366        let server = MockServer::start().await;
1367        mount_json(
1368            &server,
1369            serde_json::json!({
1370                "data": {
1371                    "current_token_ownerships_v2": [
1372                        {
1373                            "token_data_id": "0xt1",
1374                            "amount": "1",
1375                            "current_token_data": null
1376                        },
1377                        {
1378                            "token_data_id": "0xt2",
1379                            "amount": "1",
1380                            "current_token_data": null
1381                        }
1382                    ]
1383                }
1384            }),
1385        )
1386        .await;
1387
1388        let client = mock_client(&server);
1389        let page = client
1390            .get_collection_tokens(AccountAddress::ONE, Some(PaginationParams::new(2, 0)))
1391            .await
1392            .unwrap();
1393
1394        assert_eq!(page.items.len(), 2);
1395        assert_eq!(page.total_count, None);
1396        // items_count == limit => there may be more pages.
1397        assert!(page.has_more);
1398    }
1399
1400    #[tokio::test]
1401    async fn test_get_coin_balances() {
1402        let server = MockServer::start().await;
1403        mount_json(
1404            &server,
1405            serde_json::json!({
1406                "data": {
1407                    "current_coin_balances": [
1408                        {
1409                            "coin_type": "0x1::aptos_coin::AptosCoin",
1410                            "amount": "9999"
1411                        }
1412                    ]
1413                }
1414            }),
1415        )
1416        .await;
1417
1418        let client = mock_client(&server);
1419        let balances = client.get_coin_balances(AccountAddress::ONE).await.unwrap();
1420
1421        assert_eq!(balances.len(), 1);
1422        assert_eq!(balances[0].coin_type, "0x1::aptos_coin::AptosCoin");
1423        assert_eq!(balances[0].amount, "9999");
1424    }
1425
1426    #[tokio::test]
1427    async fn test_get_coin_activities() {
1428        let server = MockServer::start().await;
1429        mount_json(
1430            &server,
1431            serde_json::json!({
1432                "data": {
1433                    "coin_activities": [
1434                        {
1435                            "activity_type": "0x1::coin::DepositEvent",
1436                            "amount": "250",
1437                            "coin_type": "0x1::aptos_coin::AptosCoin"
1438                        }
1439                    ]
1440                }
1441            }),
1442        )
1443        .await;
1444
1445        let client = mock_client(&server);
1446        let activities = client
1447            .get_coin_activities(AccountAddress::ONE, Some(50))
1448            .await
1449            .unwrap();
1450
1451        assert_eq!(activities.len(), 1);
1452        assert_eq!(activities[0].activity_type, "0x1::coin::DepositEvent");
1453        assert_eq!(activities[0].amount.as_deref(), Some("250"));
1454        assert_eq!(activities[0].coin_type, "0x1::aptos_coin::AptosCoin");
1455    }
1456
1457    #[tokio::test]
1458    async fn test_get_processor_status() {
1459        let server = MockServer::start().await;
1460        mount_json(
1461            &server,
1462            serde_json::json!({
1463                "data": {
1464                    "processor_status": [
1465                        {
1466                            "processor": "default_processor",
1467                            "last_success_version": 12345,
1468                            "last_updated": "2024-01-01T00:00:00Z"
1469                        }
1470                    ]
1471                }
1472            }),
1473        )
1474        .await;
1475
1476        let client = mock_client(&server);
1477        let statuses = client.get_processor_status().await.unwrap();
1478
1479        assert_eq!(statuses.len(), 1);
1480        assert_eq!(statuses[0].processor, "default_processor");
1481        assert_eq!(statuses[0].last_success_version, 12345);
1482        assert_eq!(
1483            statuses[0].last_updated.as_deref(),
1484            Some("2024-01-01T00:00:00Z")
1485        );
1486    }
1487
1488    #[tokio::test]
1489    async fn test_get_indexer_version() {
1490        let server = MockServer::start().await;
1491        mount_json(
1492            &server,
1493            serde_json::json!({
1494                "data": {
1495                    "processor_status": [
1496                        { "processor": "a", "last_success_version": 100, "last_updated": null },
1497                        { "processor": "b", "last_success_version": 250, "last_updated": null }
1498                    ]
1499                }
1500            }),
1501        )
1502        .await;
1503
1504        let client = mock_client(&server);
1505        let version = client.get_indexer_version().await.unwrap();
1506
1507        // Returns the maximum last_success_version across processors.
1508        assert_eq!(version, 250);
1509    }
1510
1511    #[tokio::test]
1512    async fn test_check_indexer_lag() {
1513        let server = MockServer::start().await;
1514        Mock::given(method("POST"))
1515            .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
1516                "data": {
1517                    "processor_status": [
1518                        { "processor": "a", "last_success_version": 900, "last_updated": null }
1519                    ]
1520                }
1521            })))
1522            .mount(&server)
1523            .await;
1524
1525        let client = mock_client(&server);
1526        // reference 1000, indexer at 900 => lag of 100.
1527        assert!(client.check_indexer_lag(1000, 200).await.unwrap());
1528        assert!(!client.check_indexer_lag(1000, 50).await.unwrap());
1529    }
1530
1531    #[tokio::test]
1532    async fn test_query_graphql_error() {
1533        let server = MockServer::start().await;
1534        mount_json(
1535            &server,
1536            serde_json::json!({
1537                "errors": [
1538                    { "message": "field not found" },
1539                    { "message": "syntax error" }
1540                ]
1541            }),
1542        )
1543        .await;
1544
1545        let client = mock_client(&server);
1546        let result = client.get_coin_balances(AccountAddress::ONE).await;
1547
1548        assert!(result.is_err());
1549        match result.unwrap_err() {
1550            AptosError::Api {
1551                status_code,
1552                message,
1553                error_code,
1554                ..
1555            } => {
1556                assert_eq!(status_code, 400);
1557                assert_eq!(error_code.as_deref(), Some("GRAPHQL_ERROR"));
1558                // Multiple errors are joined with "; ".
1559                assert_eq!(message, "field not found; syntax error");
1560            }
1561            other => panic!("expected Api error, got {other:?}"),
1562        }
1563    }
1564
1565    #[tokio::test]
1566    async fn test_query_missing_data() {
1567        let server = MockServer::start().await;
1568        // Neither `data` nor `errors` present => Internal error.
1569        mount_json(&server, serde_json::json!({})).await;
1570
1571        let client = mock_client(&server);
1572        let result = client.get_coin_balances(AccountAddress::ONE).await;
1573
1574        assert!(result.is_err());
1575        assert!(matches!(result.unwrap_err(), AptosError::Internal(_)));
1576    }
1577
1578    #[test]
1579    fn test_indexer_client_creation() {
1580        let client = IndexerClient::new(&AptosConfig::testnet());
1581        assert!(client.is_ok());
1582    }
1583
1584    #[test]
1585    fn test_pagination_params() {
1586        let params = PaginationParams::new(10, 20);
1587        assert_eq!(params.limit, 10);
1588        assert_eq!(params.offset, 20);
1589
1590        let first_page = PaginationParams::first(50);
1591        assert_eq!(first_page.limit, 50);
1592        assert_eq!(first_page.offset, 0);
1593    }
1594
1595    #[test]
1596    fn test_page_has_more() {
1597        let page: Page<u32> = Page {
1598            items: vec![1, 2, 3],
1599            has_more: true,
1600            total_count: Some(100),
1601        };
1602        assert!(page.has_more);
1603        assert_eq!(page.items.len(), 3);
1604        assert_eq!(page.total_count, Some(100));
1605    }
1606
1607    #[test]
1608    fn test_custom_url() {
1609        let client = IndexerClient::with_url("https://custom-indexer.example.com/v1/graphql");
1610        assert!(client.is_ok());
1611    }
1612}
aptos_sdk/api/indexer.rs

aptos_sdk/api/
indexer.rs
