[Rosetta] Support coin standard (#19018)

## Description 

Rosetta support for all coins.

## Test plan 

Unit and integration tests, CB also did a thorough manual integration test on their end.

---

## Release notes

Check each box that your changes affect. If none of the boxes relate to
your changes, release notes aren't required.

For each box you select, include information after the relevant heading
that describes the impact of your changes that a user might notice and
any actions they must take to implement updates.

- [ ] Protocol: 
- [ ] Nodes (Validators and Full nodes): 
- [ ] Indexer: 
- [ ] JSON-RPC: 
- [ ] GraphQL: 
- [ ] CLI: 
- [ ] Rust SDK:
- [ ] REST API:

---------

Co-authored-by: nikos-terzo <[email protected]>
Co-authored-by: patrick <[email protected]>

crates/sui-rosetta/Cargo.toml

@@ -49,3 +49,4 @@ test-cluster.workspace = true
 tempfile.workspace = true
 rand.workspace = true
 reqwest.workspace = true
+move-cli.workspace = true

crates/sui-rosetta/src/account.rs

@@ -4,7 +4,7 @@
 use axum::extract::State;
 use axum::{Extension, Json};
 use axum_extra::extract::WithRejection;
-use futures::StreamExt;
+use futures::{future::join_all, StreamExt};
 
 use sui_sdk::rpc_types::StakeStatus;
 use sui_sdk::{SuiClient, SUI_COIN_TYPE};
@@ -14,10 +14,12 @@ use tracing::info;
 use crate::errors::Error;
 use crate::types::{
     AccountBalanceRequest, AccountBalanceResponse, AccountCoinsRequest, AccountCoinsResponse,
-    Amount, Coin, SubAccount, SubAccountType, SubBalance,
+    Amount, Coin, Currencies, Currency, SubAccountType, SubBalance,
 };
 use crate::{OnlineServerContext, SuiEnv};
 use std::time::Duration;
+use sui_sdk::error::SuiRpcResult;
+use sui_types::messages_checkpoint::CheckpointSequenceNumber;
 
 /// Get an array of all AccountBalances for an AccountIdentifier and the BlockIdentifier
 /// at which the balance lookup was performed.
@@ -29,108 +31,97 @@ pub async fn balance(
 ) -> Result<AccountBalanceResponse, Error> {
     env.check_network_identifier(&request.network_identifier)?;
     let address = request.account_identifier.address;
+    let currencies = &request.currencies;
     let mut retry_attempts = 5;
-    if let Some(SubAccount { account_type }) = request.account_identifier.sub_account {
-        while retry_attempts > 0 {
-            let balances_first =
-                get_sub_account_balances(account_type.clone(), &ctx.client, address).await?;
-            let checkpoint1 = ctx
-                .client
-                .read_api()
-                .get_latest_checkpoint_sequence_number()
-                .await?;
-            // Get another checkpoint which is greater than current
-            let mut checkpoint2 = ctx
-                .client
-                .read_api()
-                .get_latest_checkpoint_sequence_number()
-                .await?;
-
-            while checkpoint2 <= checkpoint1 {
-                checkpoint2 = ctx
-                    .client
-                    .read_api()
-                    .get_latest_checkpoint_sequence_number()
-                    .await?;
-                tokio::time::sleep(Duration::from_secs(1)).await;
-            }
-            let balances_second =
-                get_sub_account_balances(account_type.clone(), &ctx.client, address).await?;
-            if balances_first.eq(&balances_second) {
-                return Ok(AccountBalanceResponse {
-                    block_identifier: ctx.blocks().create_block_identifier(checkpoint2).await?,
-                    balances: balances_first,
-                });
-            } else {
-                // retry logic needs to be aaded
-                retry_attempts -= 1;
-            }
+    while retry_attempts > 0 {
+        let balances_first = get_balances(&ctx, &request, address, currencies.clone()).await?;
+        let checkpoint1 = get_checkpoint(&ctx).await?;
+        let mut checkpoint2 = get_checkpoint(&ctx).await?;
+        while checkpoint2 <= checkpoint1 {
+            checkpoint2 = get_checkpoint(&ctx).await?;
+            tokio::time::sleep(Duration::from_secs(1)).await;
         }
-        Err(Error::RetryExhausted(String::from("retry")))
-    } else {
-        // Get current live balance
-        while retry_attempts > 0 {
-            let balances_first = ctx
-                .client
-                .coin_read_api()
-                .get_balance(address, Some(SUI_COIN_TYPE.to_string()))
-                .await?
-                .total_balance as i128;
-
-            // Get current latest checkpoint
-            let checkpoint1 = ctx
-                .client
-                .read_api()
-                .get_latest_checkpoint_sequence_number()
-                .await?;
+        let balances_second = get_balances(&ctx, &request, address, currencies.clone()).await?;
+        if balances_first.eq(&balances_second) {
+            info!(
+                "same balance for account {} at checkpoint {}",
+                address, checkpoint2
+            );
+            return Ok(AccountBalanceResponse {
+                block_identifier: ctx.blocks().create_block_identifier(checkpoint2).await?,
+                balances: balances_first,
+            });
+        } else {
+            info!(
+                "different balance for account {} at checkpoint {}",
+                address, checkpoint2
+            );
+            retry_attempts -= 1;
+        }
+    }
+    Err(Error::RetryExhausted(String::from("retry")))
+}
 
-            // Get another checkpoint which is greater than current
-            let mut checkpoint2 = ctx
-                .client
-                .read_api()
-                .get_latest_checkpoint_sequence_number()
-                .await?;
+async fn get_checkpoint(ctx: &OnlineServerContext) -> SuiRpcResult<CheckpointSequenceNumber> {
+    ctx.client
+        .read_api()
+        .get_latest_checkpoint_sequence_number()
+        .await
+}
 
-            while checkpoint2 <= checkpoint1 {
-                checkpoint2 = ctx
-                    .client
-                    .read_api()
-                    .get_latest_checkpoint_sequence_number()
-                    .await?;
-                tokio::time::sleep(Duration::from_secs(1)).await;
+async fn get_balances(
+    ctx: &OnlineServerContext,
+    request: &AccountBalanceRequest,
+    address: SuiAddress,
+    currencies: Currencies,
+) -> Result<Vec<Amount>, Error> {
+    if let Some(sub_account) = &request.account_identifier.sub_account {
+        let account_type = sub_account.account_type.clone();
+        get_sub_account_balances(account_type, &ctx.client, address).await
+    } else if !currencies.0.is_empty() {
+        let balance_futures = currencies.0.iter().map(|currency| {
+            let coin_type = currency.metadata.clone().coin_type.clone();
+            async move {
+                (
+                    currency.clone(),
+                    get_account_balances(ctx, address, &coin_type).await,
+                )
             }
-
-            // Get live balance again
-            let balances_second = ctx
-                .client
-                .coin_read_api()
-                .get_balance(address, Some(SUI_COIN_TYPE.to_string()))
-                .await?
-                .total_balance as i128;
-
-            // if those two live balances are equal then that is the current balance for checkpoint2
-            if balances_first.eq(&balances_second) {
-                info!(
-                    "same balance for account {} at checkpoint {}",
-                    address, checkpoint2
-                );
-                return Ok(AccountBalanceResponse {
-                    block_identifier: ctx.blocks().create_block_identifier(checkpoint2).await?,
-                    balances: vec![Amount::new(balances_first)],
-                });
-            } else {
-                // balances are different so we need to try again.
-                info!(
-                    "different balance for account {} at checkpoint {}",
-                    address, checkpoint2
-                );
-                retry_attempts -= 1;
+        });
+        let balances: Vec<(Currency, Result<i128, Error>)> = join_all(balance_futures).await;
+        let mut amounts = Vec::new();
+        for (currency, balance_result) in balances {
+            match balance_result {
+                Ok(value) => amounts.push(Amount::new(value, Some(currency))),
+                Err(_e) => {
+                    return Err(Error::InvalidInput(format!(
+                        "{:?}",
+                        currency.metadata.coin_type
+                    )))
+                }
             }
         }
-        Err(Error::RetryExhausted(String::from("retry")))
+        Ok(amounts)
+    } else {
+        Err(Error::InvalidInput(
+            "Coin type is required for this request".to_string(),
+        ))
     }
 }
 
+async fn get_account_balances(
+    ctx: &OnlineServerContext,
+    address: SuiAddress,
+    coin_type: &String,
+) -> Result<i128, Error> {
+    Ok(ctx
+        .client
+        .coin_read_api()
+        .get_balance(address, Some(coin_type.to_string()))
+        .await?
+        .total_balance as i128)
+}
+
 async fn get_sub_account_balances(
     account_type: SubAccountType,
     client: &SuiClient,
@@ -187,7 +178,7 @@ async fn get_sub_account_balances(
 
     // Make sure there are always one amount returned
     Ok(if amounts.is_empty() {
-        vec![Amount::new(0)]
+        vec![Amount::new(0, None)]
     } else {
         vec![Amount::new_from_sub_balances(amounts)]
     })

crates/sui-rosetta/src/block.rs

@@ -6,6 +6,7 @@ use axum::{Extension, Json};
 use axum_extra::extract::WithRejection;
 use tracing::debug;
 
+use crate::operations::Operations;
 use crate::types::{
     BlockRequest, BlockResponse, BlockTransactionRequest, BlockTransactionResponse, Transaction,
     TransactionIdentifier,
@@ -57,7 +58,7 @@ pub async fn transaction(
         .await?;
     let hash = response.digest;
 
-    let operations = response.try_into()?;
+    let operations = Operations::try_from_response(response, &context.coin_metadata_cache).await?;
 
     let transaction = Transaction {
         transaction_identifier: TransactionIdentifier { hash },

crates/sui-rosetta/src/construction.rs

@@ -16,7 +16,7 @@ use sui_json_rpc_types::{
     SuiTransactionBlockResponseOptions,
 };
 use sui_sdk::rpc_types::SuiExecutionStatus;
-use sui_types::base_types::SuiAddress;
+use sui_types::base_types::{ObjectRef, SuiAddress};
 use sui_types::crypto::{DefaultHash, SignatureScheme, ToFromBytes};
 use sui_types::error::SuiError;
 use sui_types::signature::{GenericSignature, VerifyParams};
@@ -198,7 +198,6 @@ pub async fn preprocess(
     let internal_operation = request.operations.into_internal()?;
     let sender = internal_operation.sender();
     let budget = request.metadata.and_then(|m| m.budget);
-
     Ok(ConstructionPreprocessResponse {
         options: Some(MetadataOptions {
             internal_operation,
@@ -239,6 +238,12 @@ pub async fn metadata(
     let option = request.options.ok_or(Error::MissingMetadata)?;
     let budget = option.budget;
     let sender = option.internal_operation.sender();
+    let currency = match &option.internal_operation {
+        InternalOperation::PayCoin { currency, .. } => Some(currency.clone()),
+        _ => None,
+    };
+    let coin_type = currency.as_ref().map(|c| c.metadata.coin_type.clone());
+
     let mut gas_price = context
         .client
         .governance_api()
@@ -253,6 +258,20 @@ pub async fn metadata(
             let amount = amounts.iter().sum::<u64>();
             (Some(amount), vec![])
         }
+        InternalOperation::PayCoin { amounts, .. } => {
+            let amount = amounts.iter().sum::<u64>();
+            let coin_objs: Vec<ObjectRef> = context
+                .client
+                .coin_read_api()
+                .select_coins(sender, coin_type, amount.into(), vec![])
+                .await
+                .ok()
+                .unwrap_or_default()
+                .iter()
+                .map(|coin| coin.object_ref())
+                .collect();
+            (Some(0), coin_objs) // amount is 0 for gas coin
+        }
         InternalOperation::Stake { amount, .. } => (*amount, vec![]),
         InternalOperation::WithdrawStake { sender, stake_ids } => {
             let stake_ids = if stake_ids.is_empty() {
@@ -313,6 +332,7 @@ pub async fn metadata(
                     gas_price,
                     // MAX BUDGET
                     budget: 50_000_000_000,
+                    currency: currency.clone(),
                 })?;
 
             let dry_run = context
@@ -329,7 +349,7 @@ pub async fn metadata(
         }
     };
 
-    // Try select coins for required amounts
+    // Try select gas coins for required amounts
     let coins = if let Some(amount) = total_required_amount {
         let total_amount = amount + budget;
         context
@@ -369,8 +389,9 @@ pub async fn metadata(
             total_coin_value,
             gas_price,
             budget,
+            currency,
         },
-        suggested_fee: vec![Amount::new(budget as i128)],
+        suggested_fee: vec![Amount::new(budget as i128, None)],
     })
 }