Skip to content

Commit

Permalink
Add token filter to API
Browse files Browse the repository at this point in the history
  • Loading branch information
@perekopskiy
perekopskiy committed Oct 6, 2021
1 parent 03899b7 commit 28fa1fd
Show file tree
Hide file tree
Showing 7 changed files with 316 additions and 203 deletions.
36 changes: 30 additions & 6 deletions core/bin/zksync_api/src/api_server/rest/v02/account.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -9,15 +9,15 @@ use actix_web::{web, Scope};

// Workspace uses
use zksync_api_types::v02::{
account::{Account, AccountAddressOrId, AccountState},
account::{Account, AccountAddressOrId, AccountState, IncomingAccountTxsQuery},
pagination::{
parse_query, AccountTxsRequest, ApiEither, Paginated, PaginationQuery, PendingOpsRequest,
},
transaction::{Transaction, TxHashSerializeWrapper},
};
use zksync_crypto::params::{MIN_NFT_TOKEN_ID, NFT_TOKEN_ID_VAL};
use zksync_storage::{ConnectionPool, StorageProcessor};
use zksync_types::{tx::TxHash, AccountId, Address, BlockNumber, SerialId};
use zksync_types::{tx::TxHash, AccountId, Address, BlockNumber, SerialId, TokenLike};

// Local uses
use super::{
Expand Down Expand Up @@ -303,12 +303,25 @@ impl ApiAccountData {
&self,
query: PaginationQuery<ApiEither<TxHash>>,
address: Address,
token_like: Option<TokenLike>,
) -> Result<Paginated<Transaction, TxHashSerializeWrapper>, Error> {
let mut storage = self.pool.access_storage().await.map_err(Error::storage)?;
let token = if let Some(token_like) = token_like {
Some(
self.tokens
.get_token(&mut storage, token_like.clone())
.await
.map_err(Error::storage)?
.ok_or_else(|| Error::from(PriceError::token_not_found(token_like)))?,
)
} else {
None
};
let new_query = PaginationQuery {
from: AccountTxsRequest {
address,
tx_hash: query.from,
address,
token: token.map(|token| token.id),
},
limit: query.limit,
direction: query.direction,
Expand Down Expand Up @@ -382,12 +395,23 @@ async fn account_full_info(
async fn account_txs(
data: web::Data<ApiAccountData>,
account_id_or_address: web::Path<String>,
web::Query(query): web::Query<PaginationQuery<String>>,
web::Query(query): web::Query<IncomingAccountTxsQuery>,
) -> ApiResult<Paginated<Transaction, TxHashSerializeWrapper>> {
let query = api_try!(parse_query(query).map_err(Error::from));
let pagination = api_try!(parse_query(PaginationQuery {
from: query.from,
limit: query.limit,
direction: query.direction
})
.map_err(Error::from));

let address_or_id = api_try!(data.parse_account_id_or_address(&account_id_or_address));
let address = api_try!(data.get_address_by_address_or_id(address_or_id).await);
data.account_txs(query, address).await.into()

let token_like = query.token.map(|token| TokenLike::parse(&token));

data.account_txs(pagination, address, token_like)
.await
.into()
}

async fn account_pending_txs(
Expand Down
4 changes: 2 additions & 2 deletions core/bin/zksync_api/src/api_server/rest/v02/paginate_impl.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -229,8 +229,8 @@ impl Paginate<AccountTxsRequest> for StorageProcessor<'_> {

let query = PaginationQuery {
from: AccountTxsRequest {
address: query.from.address,
tx_hash: ApiEither::from(tx_hash),
..query.from
},
limit: query.limit,
direction: query.direction,
Expand All @@ -246,7 +246,7 @@ impl Paginate<AccountTxsRequest> for StorageProcessor<'_> {
let count = transaction
.chain()
.operations_ext_schema()
.get_account_transactions_count(query.from.address)
.get_account_transactions_count(query.from.address, query.from.token)
.await
.map_err(Error::storage)?;

Expand Down
10 changes: 10 additions & 0 deletions core/lib/api_types/src/v02/account.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@ use zksync_types::{
};
use zksync_utils::{BigUintSerdeAsRadix10Str, BigUintSerdeWrapper};

use super::pagination::PaginationDirection;
use super::token::NFT;

#[derive(Debug, Serialize, Deserialize, PartialEq, Clone, Default)]
Expand Down Expand Up @@ -104,3 +105,12 @@ pub struct DepositingFunds {
pub struct DepositingAccountBalances {
pub balances: HashMap<String, DepositingFunds>,
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct IncomingAccountTxsQuery {
pub token: Option<String>,
pub from: String,
pub limit: u32,
pub direction: PaginationDirection,
}
3 changes: 2 additions & 1 deletion core/lib/api_types/src/v02/pagination.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@ use either::Either;
use serde::{Deserialize, Serialize, Serializer};
use std::str::FromStr;
use thiserror::Error;
use zksync_types::{tx::TxHash, AccountId, Address, BlockNumber, SerialId};
use zksync_types::{tx::TxHash, AccountId, Address, BlockNumber, SerialId, TokenId};

pub const MAX_LIMIT: u32 = 100;

Expand Down Expand Up @@ -136,4 +136,5 @@ pub struct PendingOpsRequest {
pub struct AccountTxsRequest {
pub address: Address,
pub tx_hash: ApiEither<TxHash>,
pub token: Option<TokenId>,
}
378 changes: 193 additions & 185 deletions core/lib/storage/sqlx-data.json
View file

Large diffs are not rendered by default.

77 changes: 70 additions & 7 deletions core/lib/storage/src/chain/operations_ext/mod.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1020,6 +1020,7 @@ impl<'a, 'c> OperationsExtSchema<'a, 'c> {
.account_id_by_address(query.from.address)
.await?;
let account_id_value = serde_json::to_value(account_id).unwrap();
let token_id_value = serde_json::to_value(query.from.token).unwrap();

let txs = if let Some((time_from, _)) = created_at_and_block {
let raw_txs: Vec<TransactionItem> = match query.direction {
Expand Down Expand Up @@ -1058,6 +1059,20 @@ impl<'a, 'c> OperationsExtSchema<'a, 'c> {
operation->'recipients'->1 = $2
)
)
) AND (
$4::jsonb = $5::jsonb
OR
tx->'token' = $4
OR
tx->'feeToken' = $4
OR (
tx->'type' = '"Swap"'
AND (
tx->'orders'->0->'tokenBuy' = $4
OR
tx->'orders'->0->'tokenSell' = $4
)
)
)
AND created_at >= $3
), priority_ops AS (
Expand All @@ -1073,7 +1088,9 @@ impl<'a, 'c> OperationsExtSchema<'a, 'c> {
block_index,
Null::bigint as batch_id
FROM executed_priority_operations
WHERE (from_account = $1 OR to_account = $1) AND created_at >= $3
WHERE (from_account = $1 OR to_account = $1)
AND ($4::jsonb = $5::jsonb OR operation->'priority_op'->'token' = $4)
AND created_at >= $3
), everything AS (
SELECT * FROM transactions
UNION ALL
Expand All @@ -1091,11 +1108,13 @@ impl<'a, 'c> OperationsExtSchema<'a, 'c> {
batch_id as "batch_id?"
FROM everything
ORDER BY created_at ASC, block_index ASC
LIMIT $4
LIMIT $6
"#,
query.from.address.as_bytes(),
account_id_value,
time_from,
token_id_value,
serde_json::Value::Null,
i64::from(query.limit),
)
.fetch_all(transaction.conn())
Expand Down Expand Up @@ -1136,6 +1155,20 @@ impl<'a, 'c> OperationsExtSchema<'a, 'c> {
operation->'recipients'->1 = $2
)
)
) AND (
$4::jsonb = $5::jsonb
OR
tx->'token' = $4
OR
tx->'feeToken' = $4
OR (
tx->'type' = '"Swap"'
AND (
tx->'orders'->0->'tokenBuy' = $4
OR
tx->'orders'->0->'tokenSell' = $4
)
)
)
AND created_at <= $3
), priority_ops AS (
Expand All @@ -1151,7 +1184,9 @@ impl<'a, 'c> OperationsExtSchema<'a, 'c> {
block_index,
Null::bigint as batch_id
FROM executed_priority_operations
WHERE (from_account = $1 OR to_account = $1) AND created_at <= $3
WHERE (from_account = $1 OR to_account = $1)
AND ($4::jsonb = $5::jsonb OR operation->'priority_op'->'token' = $4)
AND created_at <= $3
), everything AS (
SELECT * FROM transactions
UNION ALL
Expand All @@ -1169,11 +1204,13 @@ impl<'a, 'c> OperationsExtSchema<'a, 'c> {
batch_id as "batch_id"
FROM everything
ORDER BY created_at DESC, block_index DESC
LIMIT $4
LIMIT $6
"#,
query.from.address.as_bytes(),
account_id_value,
time_from,
token_id_value,
serde_json::Value::Null,
i64::from(query.limit),
)
.fetch_all(transaction.conn())
Expand Down Expand Up @@ -1313,7 +1350,11 @@ impl<'a, 'c> OperationsExtSchema<'a, 'c> {
Ok(record.map(|record| TxHash::from_slice(&record.tx_hash).unwrap()))
}

pub async fn get_account_transactions_count(&mut self, address: Address) -> QueryResult<u32> {
pub async fn get_account_transactions_count(
&mut self,
address: Address,
token: Option<TokenId>,
) -> QueryResult<u32> {
let start = Instant::now();
let mut transaction = self.0.start_transaction().await?;
let last_committed = transaction
Expand All @@ -1327,6 +1368,7 @@ impl<'a, 'c> OperationsExtSchema<'a, 'c> {
.account_id_by_address(address)
.await?;
let account_id_value = serde_json::to_value(account_id).unwrap();
let token_id_value = serde_json::to_value(token).unwrap();
let tx_count = sqlx::query!(
r#"
SELECT COUNT(*) as "count!" FROM executed_transactions
Expand All @@ -1348,23 +1390,44 @@ impl<'a, 'c> OperationsExtSchema<'a, 'c> {
operation->'recipients'->1 = $3
)
)
) AND (
$4::jsonb = $5::jsonb
OR
tx->'token' = $4
OR
tx->'feeToken' = $4
OR (
tx->'type' = '"Swap"'
AND (
tx->'orders'->0->'tokenBuy' = $4
OR
tx->'orders'->0->'tokenSell' = $4
)
)
)
"#,
i64::from(*last_committed),
address.as_bytes(),
account_id_value
account_id_value,
token_id_value,
serde_json::Value::Null
)
.fetch_one(transaction.conn())
.await?
.count;

println!("{:?}", token_id_value);

let priority_op_count = sqlx::query!(
r#"
SELECT COUNT(*) as "count!" FROM executed_priority_operations
WHERE block_number <= $1 AND (from_account = $2 OR to_account = $2)
AND ($3::jsonb = $4::jsonb OR operation->'priority_op'->'token' = $3)
"#,
i64::from(*last_committed),
address.as_bytes()
address.as_bytes(),
token_id_value,
serde_json::Value::Null
)
.fetch_one(transaction.conn())
.await?
Expand Down
11 changes: 9 additions & 2 deletions core/lib/storage/src/tests/chain/operations_ext/mod.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -338,6 +338,7 @@ async fn get_account_transactions(mut storage: StorageProcessor<'_>) -> QueryRes
from: AccountTxsRequest {
address: from,
tx_hash: ApiEither::from(setup.get_tx_hash(0, 0)),
token: None,
},
limit: 1,
direction: PaginationDirection::Newer,
Expand Down Expand Up @@ -464,6 +465,7 @@ async fn get_account_transactions(mut storage: StorageProcessor<'_>) -> QueryRes
from: AccountTxsRequest {
address: from,
tx_hash: ApiEither::from(request.tx_hash),
token: None,
},
limit: request.limit,
direction: request.direction,
Expand All @@ -481,6 +483,7 @@ async fn get_account_transactions(mut storage: StorageProcessor<'_>) -> QueryRes
from: AccountTxsRequest {
address: from,
tx_hash: ApiEither::from(setup.get_tx_hash(1, 2)),
token: None,
},
limit: 1,
direction: PaginationDirection::Newer,
Expand All @@ -497,6 +500,7 @@ async fn get_account_transactions(mut storage: StorageProcessor<'_>) -> QueryRes
from: AccountTxsRequest {
address: from,
tx_hash: ApiEither::from(setup.get_tx_hash(0, 9)),
token: None,
},
limit: 2,
direction: PaginationDirection::Newer,
Expand All @@ -514,6 +518,7 @@ async fn get_account_transactions(mut storage: StorageProcessor<'_>) -> QueryRes
from: AccountTxsRequest {
address: from,
tx_hash: ApiEither::from(setup.get_tx_hash(0, 2)),
token: None,
},
limit: 1,
direction: PaginationDirection::Newer,
Expand All @@ -527,6 +532,7 @@ async fn get_account_transactions(mut storage: StorageProcessor<'_>) -> QueryRes
from: AccountTxsRequest {
address: to,
tx_hash: ApiEither::from(setup.get_tx_hash(0, 2)),
token: None,
},
limit: 1,
direction: PaginationDirection::Newer,
Expand Down Expand Up @@ -715,7 +721,7 @@ async fn account_transactions_count(mut storage: StorageProcessor<'_>) -> QueryR
let count_before_commit = storage
.chain()
.operations_ext_schema()
.get_account_transactions_count(setup.from_zksync_account.address)
.get_account_transactions_count(setup.from_zksync_account.address, None)
.await?;
assert_eq!(count_before_commit, 0);

Expand All @@ -724,7 +730,7 @@ async fn account_transactions_count(mut storage: StorageProcessor<'_>) -> QueryR
let count_after_commit = storage
.chain()
.operations_ext_schema()
.get_account_transactions_count(setup.from_zksync_account.address)
.get_account_transactions_count(setup.from_zksync_account.address, None)
.await?;
assert_eq!(count_after_commit, 10);

Expand Down Expand Up @@ -1164,6 +1170,7 @@ async fn test_getting_swap_for_acc(mut storage: StorageProcessor<'_>) -> QueryRe
from: AccountTxsRequest {
address: *address,
tx_hash: ApiEither::from(tx_hash),
token: None,
},
limit: 1,
direction: PaginationDirection::Older,
Expand Down

0 comments on commit 28fa1fd

Please sign in to comment.