ic-query 0.76.5

//! Module: icrc::live::account_transactions
//!
//! Responsibility: resolve one ICRC index and query a page or complete account history.
//! Does not own: wire-specific response conversion, collection validation, caching, or rendering.
//! Boundary: validates ledger/index identity once and delegates each protocol concern explicitly.

use ic_platform::icrc_ledger_types::Account as LedgerAccount;
mod collection;
mod cursor;
mod request;
mod wire;

use self::collection::{AccountTransactionCollectionState, AccountTransactionsPage};
use self::cursor::parse_transaction_cursor;
pub(in crate::icrc) use self::cursor::{
    normalize_transaction_cursor, validate_account_transaction_page,
    validate_canonical_account_transactions, validate_transaction_cursor_text,
};
pub(in crate::icrc) use self::request::{
    normalize_account_transaction_identity, normalize_optional_index_canister_id,
    validate_account_transaction_page_size,
};
use super::fetch::{account_from_parts, live_query_context, query_token_display_fields};
use crate::candid_call::query_no_args;
use crate::{
    QueryProgress, QueryProgressState,
    icrc::{
        ledger::{index_principal_error_text, principal_from_text, query_index_principal},
        model::{
            IcrcAccountTransactionCollectionData, IcrcAccountTransactionError,
            IcrcAccountTransactionPageData, IcrcAccountTransactionPageRequest,
            IcrcAccountTransactionRefreshRequest, IcrcError,
        },
    },
    snapshot_cache::{
        PagedCollectionPage, PagedSnapshotRefresh, run_paged_snapshot_refresh_with_progress,
    },
};
use ic_platform::agent::Agent;
use ic_platform::candid::contract::Principal;

const INDEX_LEDGER_ID_METHOD: &str = "ledger_id";

pub(super) async fn fetch_account_transaction_page_async(
    source: &crate::icrc::LiveIcrcSource,
    request: &IcrcAccountTransactionPageRequest,
) -> Result<IcrcAccountTransactionPageData, IcrcAccountTransactionError> {
    let context = resolve_account_transaction_context(
        source,
        &request.ledger.source_endpoint,
        &request.ledger.ledger_canister_id,
        request.index_canister_id.as_deref(),
        &request.account_owner,
        request.subaccount_hex.as_deref(),
    )
    .await?;
    let page =
        query_account_transaction_page(&context, request.start.as_deref(), request.limit).await?;

    let next_start = page.next_start().map(str::to_owned);
    Ok(IcrcAccountTransactionPageData {
        index_canister_id: context.index_canister.to_text(),
        balance: page.balance,
        oldest_transaction_id: page.oldest_transaction_id,
        next_start,
        token_symbol: context.token_symbol,
        decimals: context.decimals,
        transactions: page.transactions,
    })
}

pub(super) async fn fetch_complete_account_transactions_async(
    source: &crate::icrc::LiveIcrcSource,
    request: &IcrcAccountTransactionRefreshRequest,
    progress: &mut (dyn QueryProgress + Send),
) -> Result<IcrcAccountTransactionCollectionData, IcrcAccountTransactionError> {
    let context = resolve_account_transaction_context(
        source,
        &request.cache.source_endpoint,
        &request.cache.ledger_canister_id,
        request.index_canister_id.as_deref(),
        &request.cache.account_owner,
        request.cache.subaccount_hex.as_deref(),
    )
    .await?;
    let state = run_paged_snapshot_refresh_with_progress(
        AccountTransactionRefresh {
            request,
            context: &context,
            state: AccountTransactionCollectionState::new(context.index_canister.to_text()),
        },
        progress,
    )
    .await?;

    state.into_complete(context.token_symbol, context.decimals)
}

struct AccountTransactionQueryContext {
    agent: Agent,
    index_canister: Principal,
    account: LedgerAccount,
    token_symbol: String,
    decimals: u8,
}

async fn resolve_account_transaction_context(
    source: &crate::icrc::LiveIcrcSource,
    source_endpoint: &str,
    ledger_canister_id: &str,
    index_canister_id: Option<&str>,
    account_owner: &str,
    subaccount_hex: Option<&str>,
) -> Result<AccountTransactionQueryContext, IcrcAccountTransactionError> {
    let (agent, ledger_canister) = live_query_context(source, source_endpoint, ledger_canister_id)?;
    let index_canister = match index_canister_id {
        Some(index_canister_id) => principal_from_text(index_canister_id, "index_canister_id")?,
        None => resolve_index_canister(&agent, &ledger_canister).await?,
    };
    let (actual_ledger, token_display) = futures_util::try_join!(
        query_no_args::<Principal, IcrcError>(&agent, &index_canister, INDEX_LEDGER_ID_METHOD),
        query_token_display_fields(&agent, &ledger_canister),
    )?;
    if actual_ledger != ledger_canister {
        return Err(IcrcAccountTransactionError::IndexLedgerMismatch {
            index_canister_id: index_canister.to_text(),
            expected_ledger_canister_id: ledger_canister.to_text(),
            actual_ledger_canister_id: actual_ledger.to_text(),
        });
    }
    let account = account_from_parts(account_owner, subaccount_hex, "account_owner")?;
    let (token_symbol, decimals) = token_display;

    Ok(AccountTransactionQueryContext {
        agent,
        index_canister,
        account,
        token_symbol,
        decimals,
    })
}

async fn query_account_transaction_page(
    context: &AccountTransactionQueryContext,
    start: Option<&str>,
    limit: u32,
) -> Result<AccountTransactionsPage, IcrcAccountTransactionError> {
    wire::query_account_transaction_page(
        &context.agent,
        &context.index_canister,
        &context.account,
        start.map(parse_transaction_cursor).transpose()?,
        limit,
    )
    .await
}

async fn resolve_index_canister(
    agent: &Agent,
    ledger_canister: &Principal,
) -> Result<Principal, IcrcAccountTransactionError> {
    let result = query_index_principal::<IcrcError>(agent, ledger_canister)
        .await
        .map_err(|source| IcrcAccountTransactionError::IndexDiscovery {
            ledger_canister_id: ledger_canister.to_text(),
            source,
        })?;
    match result {
        Ok(index_canister) => Ok(index_canister),
        Err(error) => Err(IcrcAccountTransactionError::IndexUnavailable {
            ledger_canister_id: ledger_canister.to_text(),
            reason: index_principal_error_text(error),
        }),
    }
}

struct AccountTransactionRefresh<'a> {
    request: &'a IcrcAccountTransactionRefreshRequest,
    context: &'a AccountTransactionQueryContext,
    state: AccountTransactionCollectionState,
}

impl PagedSnapshotRefresh for AccountTransactionRefresh<'_> {
    type Complete = AccountTransactionCollectionState;
    type Error = IcrcAccountTransactionError;

    const PAGE_LIMIT_REASON: &'static str = "max pages reached before index exhaustion";

    fn progress_text(&self) -> String {
        self.state.progress_text()
    }

    fn progress_text_for_state(&self, _state: QueryProgressState) -> String {
        self.state.progress_text()
    }

    fn max_pages_reached(&self) -> bool {
        self.request
            .max_pages
            .is_some_and(|limit| self.state.page_count() >= limit)
    }

    fn incomplete_refresh_error(&self, reason: &'static str) -> Self::Error {
        self.state.incomplete(reason)
    }

    fn fetch_next_page(
        &mut self,
    ) -> crate::report::SourceFuture<'_, Result<PagedCollectionPage, Self::Error>> {
        Box::pin(async move {
            let page = query_account_transaction_page(
                self.context,
                self.state.next_cursor(),
                self.request.page_size,
            )
            .await
            .map_err(|error| self.state.page_error(error))?;
            let page_len = page.transactions.len();
            let exhausted = self.state.ingest(page, self.request.page_size)?;
            Ok(PagedCollectionPage::new(page_len, exhausted))
        })
    }

    fn into_complete(self) -> Self::Complete {
        self.state
    }
}

#[cfg(test)]
mod tests;