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

aptos_sdk/api/
indexer.rs
