ic-query 0.54.0

//! Module: icrc::account_transaction_cache::refresh
//!
//! Responsibility: orchestrate complete account-history refreshes and atomic publication.
//! Does not own: index collection, cache identity, strict cache reads, or report views.
//! Boundary: publishes only validated API-exhausted collections under one refresh lock.

use super::{
    ICRC_ACCOUNT_TRANSACTION_CACHE_COMPONENT, ICRC_ACCOUNT_TRANSACTION_FETCHED_BY,
    attempt::{write_complete_attempt, write_failed_attempt, write_starting_attempt},
    storage::{
        ICRC_ACCOUNT_TRANSACTION_CACHE_SCHEMA_VERSION, cache_paths,
        icrc_account_transaction_cache_path, load_cached_icrc_account_transactions,
        normalize_cache_request, snapshot_is_stale, validate_collection_timestamps,
    },
};
use crate::cache_file::write_managed_json_pretty_atomically;
use crate::{
    HostCacheError, QueryProgress,
    cache::{CacheCollectionCompleteness, validate_cache_collection_completeness},
    cache_file::{CacheRefreshReason, host_cache_refresh_reason, load_or_refresh_cache},
    icrc::{
        ledger::principal_from_text,
        live::{
            ICRC_ACCOUNT_TRANSACTION_MAX_PAGE_SIZE, IcrcAccountTransactionCollectionSource,
            LiveIcrcSource, account_transactions::validate_canonical_account_transactions,
        },
        model::{
            CachedIcrcAccountTransactionSnapshot, IcrcAccountTransactionCollectionData,
            IcrcAccountTransactionError, IcrcAccountTransactionRefreshReport,
            IcrcAccountTransactionRefreshRequest, IcrcAccountTransactionSnapshot, IcrcError,
        },
    },
    progress::IgnoreQueryProgress,
    snapshot_cache::{
        LockedSnapshotRefreshRequest, SnapshotJsonPaths, publish_snapshot_with_attempt,
        run_snapshot_refresh_with_attempts, with_locked_snapshot_refresh,
    },
    subnet_catalog::{MAINNET_NETWORK, format_utc_timestamp_secs},
};
use std::time::{Duration, Instant};

const ICRC_ACCOUNT_TRANSACTION_REFRESH_REPORT_SCHEMA_VERSION: u32 = 1;

/// Force a complete live refresh and atomically replace its cache.
pub fn refresh_icrc_account_transaction_cache(
    request: &IcrcAccountTransactionRefreshRequest,
) -> Result<IcrcAccountTransactionRefreshReport, IcrcAccountTransactionError> {
    refresh_icrc_account_transaction_cache_with_source(request, &LiveIcrcSource)
}

/// Force a complete live refresh while emitting structured progress.
pub fn refresh_icrc_account_transaction_cache_with_progress(
    request: &IcrcAccountTransactionRefreshRequest,
    progress: &mut (dyn QueryProgress + Send),
) -> Result<IcrcAccountTransactionRefreshReport, IcrcAccountTransactionError> {
    refresh_icrc_account_transaction_cache_with_source_and_progress(
        request,
        &LiveIcrcSource,
        progress,
    )
}

/// Refresh with a caller-supplied complete-history source.
pub fn refresh_icrc_account_transaction_cache_with_source(
    request: &IcrcAccountTransactionRefreshRequest,
    source: &dyn IcrcAccountTransactionCollectionSource,
) -> Result<IcrcAccountTransactionRefreshReport, IcrcAccountTransactionError> {
    let mut progress = IgnoreQueryProgress;
    refresh_icrc_account_transaction_cache_with_source_and_progress(request, source, &mut progress)
}

fn refresh_icrc_account_transaction_cache_with_source_and_progress(
    request: &IcrcAccountTransactionRefreshRequest,
    source: &dyn IcrcAccountTransactionCollectionSource,
    progress: &mut (dyn QueryProgress + Send),
) -> Result<IcrcAccountTransactionRefreshReport, IcrcAccountTransactionError> {
    let request = normalize_refresh_request(request)?;
    let paths = cache_paths(&request.cache);
    let collection_started = Instant::now();
    with_locked_snapshot_refresh(
        LockedSnapshotRefreshRequest {
            cache_root: &request.cache.cache_root,
            snapshot_path: &paths.snapshot_path,
            refresh_lock_path: &paths.refresh_lock_path,
            network: MAINNET_NETWORK,
            now_unix_secs: request.now_unix_secs,
            lock_stale_after_seconds: request.lock_stale_after_seconds,
        },
        cache_operation_error,
        |state| {
            run_snapshot_refresh_with_attempts(
                || write_starting_attempt(&paths.refresh_attempt_path, &request),
                || {
                    let complete =
                        source.fetch_complete_account_transactions(&request, progress)?;
                    publish_complete_snapshot(
                        &request,
                        &paths,
                        state.replaced_existing_snapshot,
                        complete,
                        collection_started.elapsed(),
                    )
                },
                |error| write_failed_attempt(&paths.refresh_attempt_path, &request, error),
            )
        },
    )
}

/// Load a complete cache, refreshing when it is missing or invalid.
pub fn load_or_refresh_missing_icrc_account_transactions(
    request: &IcrcAccountTransactionRefreshRequest,
) -> Result<CachedIcrcAccountTransactionSnapshot, IcrcAccountTransactionError> {
    load_or_refresh_missing_icrc_account_transactions_with_source(request, &LiveIcrcSource)
}

/// Load a complete cache or replace missing or invalid content with the supplied source.
pub fn load_or_refresh_missing_icrc_account_transactions_with_source(
    request: &IcrcAccountTransactionRefreshRequest,
    source: &dyn IcrcAccountTransactionCollectionSource,
) -> Result<CachedIcrcAccountTransactionSnapshot, IcrcAccountTransactionError> {
    let expected_path = icrc_account_transaction_cache_path(&request.cache)?;
    load_or_refresh_cache(
        || load_cached_icrc_account_transactions(&request.cache),
        |_| false,
        |error| account_transaction_cache_refresh_reason(error, &expected_path),
        |_| {
            refresh_icrc_account_transaction_cache_with_source(request, source)?;
            Ok(())
        },
    )
}

/// Load a complete cache, refreshing when it is missing, invalid, or stale.
pub fn load_or_refresh_stale_icrc_account_transactions(
    request: &IcrcAccountTransactionRefreshRequest,
    stale_after_seconds: u64,
) -> Result<CachedIcrcAccountTransactionSnapshot, IcrcAccountTransactionError> {
    load_or_refresh_stale_icrc_account_transactions_with_source(
        request,
        stale_after_seconds,
        &LiveIcrcSource,
    )
}

/// Load a complete cache or use the supplied source when it is unusable or stale.
pub fn load_or_refresh_stale_icrc_account_transactions_with_source(
    request: &IcrcAccountTransactionRefreshRequest,
    stale_after_seconds: u64,
    source: &dyn IcrcAccountTransactionCollectionSource,
) -> Result<CachedIcrcAccountTransactionSnapshot, IcrcAccountTransactionError> {
    let expected_path = icrc_account_transaction_cache_path(&request.cache)?;
    load_or_refresh_cache(
        || load_cached_icrc_account_transactions(&request.cache),
        |snapshot| {
            snapshot_is_stale(
                &snapshot.snapshot,
                request.now_unix_secs,
                stale_after_seconds,
            )
        },
        |error| account_transaction_cache_refresh_reason(error, &expected_path),
        |_| {
            refresh_icrc_account_transaction_cache_with_source(request, source)?;
            Ok(())
        },
    )
}

fn publish_complete_snapshot(
    request: &IcrcAccountTransactionRefreshRequest,
    paths: &SnapshotJsonPaths,
    replaced_existing_cache: bool,
    complete: IcrcAccountTransactionCollectionData,
    collection_elapsed: Duration,
) -> Result<IcrcAccountTransactionRefreshReport, IcrcAccountTransactionError> {
    let completeness = validate_collection_data(request, &complete)?;
    let collection_started_at = format_utc_timestamp_secs(request.now_unix_secs);
    let collection_completed_unix_secs = request
        .now_unix_secs
        .checked_add(collection_elapsed.as_secs())
        .ok_or_else(|| IcrcAccountTransactionError::InvalidCache {
            path: paths.snapshot_path.clone(),
            reason: "collection completion timestamp exceeds supported seconds".to_string(),
        })?;
    let collection_completed_at = format_utc_timestamp_secs(collection_completed_unix_secs);
    let newest_transaction_id = complete
        .transactions
        .first()
        .map(|transaction| transaction.id.clone());
    let oldest_transaction_id = complete
        .transactions
        .last()
        .map(|transaction| transaction.id.clone());
    let snapshot = IcrcAccountTransactionSnapshot {
        schema_version: ICRC_ACCOUNT_TRANSACTION_CACHE_SCHEMA_VERSION,
        source_endpoint: request.cache.source_endpoint.clone(),
        collection_started_at: collection_started_at.clone(),
        collection_completed_at: collection_completed_at.clone(),
        fetched_by: ICRC_ACCOUNT_TRANSACTION_FETCHED_BY.to_string(),
        ledger_canister_id: request.cache.ledger_canister_id.clone(),
        index_canister_id: complete.index_canister_id.clone(),
        account_owner: request.cache.account_owner.clone(),
        subaccount_hex: request.cache.subaccount_hex.clone(),
        balance: complete.balance,
        token_symbol: complete.token_symbol,
        decimals: complete.decimals,
        newest_transaction_id: newest_transaction_id.clone(),
        oldest_transaction_id: oldest_transaction_id.clone(),
        completeness,
        transactions: complete.transactions,
    };
    validate_collection_timestamps(
        &paths.snapshot_path,
        &snapshot.collection_started_at,
        &snapshot.collection_completed_at,
    )?;
    let transaction_count = snapshot.transactions.len();
    let attempt_finalization_error = publish_snapshot_with_attempt(
        || {
            write_managed_json_pretty_atomically(
                &request.cache.cache_root,
                &paths.snapshot_path,
                &snapshot,
                crate::cache_file::MAX_COLLECTION_SNAPSHOT_BYTES,
                |path, source| {
                    HostCacheError::serialize_cache(
                        ICRC_ACCOUNT_TRANSACTION_CACHE_COMPONENT,
                        path,
                        source,
                    )
                    .into()
                },
                cache_operation_error,
            )
        },
        || {
            write_complete_attempt(
                &paths.refresh_attempt_path,
                request,
                &snapshot.index_canister_id,
                complete.last_cursor,
                complete.page_count,
                transaction_count,
            )
        },
    )?;

    Ok(IcrcAccountTransactionRefreshReport {
        schema_version: ICRC_ACCOUNT_TRANSACTION_REFRESH_REPORT_SCHEMA_VERSION,
        ledger_canister_id: snapshot.ledger_canister_id,
        index_canister_id: snapshot.index_canister_id,
        account_owner: snapshot.account_owner,
        subaccount_hex: snapshot.subaccount_hex,
        transaction_count,
        newest_transaction_id,
        oldest_transaction_id,
        page_size: snapshot.completeness.page_size,
        page_count: snapshot.completeness.page_count,
        point_in_time_guaranteed: false,
        replaced_existing_cache,
        attempt_finalization_error,
        collection_started_at,
        collection_completed_at,
        source_endpoint: snapshot.source_endpoint,
        fetched_by: snapshot.fetched_by,
        cache_path: paths.snapshot_path.display().to_string(),
        refresh_attempt_path: paths.refresh_attempt_path.display().to_string(),
        refresh_lock_path: paths.refresh_lock_path.display().to_string(),
    })
}

fn validate_collection_data(
    request: &IcrcAccountTransactionRefreshRequest,
    complete: &IcrcAccountTransactionCollectionData,
) -> Result<CacheCollectionCompleteness, IcrcAccountTransactionError> {
    let completeness = CacheCollectionCompleteness::api_exhausted(
        request.page_size,
        complete.page_count,
        complete.transactions.len(),
        false,
    );
    validate_cache_collection_completeness(&completeness, complete.transactions.len())
        .map_err(|reason| incomplete_collection_error(complete, reason))?;
    if request
        .max_pages
        .is_some_and(|maximum| complete.page_count > maximum)
    {
        return Err(incomplete_collection_error(
            complete,
            "source returned a complete collection beyond the requested max pages",
        ));
    }
    let actual_index =
        principal_from_text::<IcrcError>(&complete.index_canister_id, "index_canister_id")?;
    if let Some(expected_index) = request.index_canister_id.as_deref()
        && expected_index != actual_index.to_text()
    {
        return Err(IcrcAccountTransactionError::CollectionIndexMismatch {
            expected_index_canister_id: expected_index.to_string(),
            actual_index_canister_id: actual_index.to_text(),
        });
    }
    let expected_last_cursor = complete
        .transactions
        .last()
        .map(|transaction| transaction.id.as_str());
    if complete.last_cursor.as_deref() != expected_last_cursor {
        return Err(incomplete_collection_error(
            complete,
            "final cursor does not match the oldest collected transaction",
        ));
    }
    validate_canonical_account_transactions(&complete.transactions)
        .map_err(|reason| incomplete_collection_error(complete, reason))?;
    Ok(completeness)
}

fn incomplete_collection_error(
    complete: &IcrcAccountTransactionCollectionData,
    reason: impl Into<String>,
) -> IcrcAccountTransactionError {
    IcrcAccountTransactionError::IncompleteCollection {
        index_canister_id: Some(complete.index_canister_id.clone()),
        pages_fetched: complete.page_count,
        rows_fetched: complete.transactions.len(),
        last_cursor: complete.last_cursor.clone(),
        reason: reason.into(),
    }
}

fn normalize_refresh_request(
    request: &IcrcAccountTransactionRefreshRequest,
) -> Result<IcrcAccountTransactionRefreshRequest, IcrcAccountTransactionError> {
    if !(1..=ICRC_ACCOUNT_TRANSACTION_MAX_PAGE_SIZE).contains(&request.page_size) {
        return Err(IcrcAccountTransactionError::InvalidPageSize {
            page_size: request.page_size,
            max_page_size: ICRC_ACCOUNT_TRANSACTION_MAX_PAGE_SIZE,
        });
    }
    if request.max_pages == Some(0) {
        return Err(IcrcAccountTransactionError::InvalidMaxPages { max_pages: 0 });
    }
    let index_canister_id = request
        .index_canister_id
        .as_deref()
        .map(|value| principal_from_text::<IcrcError>(value, "index_canister_id"))
        .transpose()?
        .map(|principal| principal.to_text());
    Ok(IcrcAccountTransactionRefreshRequest {
        cache: normalize_cache_request(&request.cache)?,
        now_unix_secs: request.now_unix_secs,
        index_canister_id,
        page_size: request.page_size,
        max_pages: request.max_pages,
        lock_stale_after_seconds: request.lock_stale_after_seconds,
    })
}

fn cache_operation_error(source: crate::CacheFileError) -> IcrcAccountTransactionError {
    HostCacheError::operation(ICRC_ACCOUNT_TRANSACTION_CACHE_COMPONENT, source).into()
}

fn account_transaction_cache_refresh_reason(
    error: IcrcAccountTransactionError,
    expected_path: &std::path::Path,
) -> Result<CacheRefreshReason, IcrcAccountTransactionError> {
    match error {
        IcrcAccountTransactionError::Cache(error) => {
            host_cache_refresh_reason(error, expected_path)
                .map_err(IcrcAccountTransactionError::Cache)
        }
        IcrcAccountTransactionError::InvalidCache { path, .. } => {
            Ok(CacheRefreshReason::Invalid(path))
        }
        error => Err(error),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{icrc::IcrcAccountTransactionCacheRequest, test_support::temp_dir};

    #[test]
    fn publication_uses_the_callers_clock_and_elapsed_collection_duration() {
        let root = temp_dir("ic-query-icrc-account-collection-duration");
        let cache = IcrcAccountTransactionCacheRequest::new(
            &root,
            "https://icp-api.io",
            "ryjl3-tyaaa-aaaaa-aaaba-cai",
            "aaaaa-aa",
        );
        let request = IcrcAccountTransactionRefreshRequest::new(cache, 2_000_000_000, 100, 1_800);
        let paths = cache_paths(&request.cache);
        let complete = IcrcAccountTransactionCollectionData {
            index_canister_id: "qhbym-qaaaa-aaaaa-aaafq-cai".to_string(),
            balance: "0".to_string(),
            token_symbol: "ICP".to_string(),
            decimals: 8,
            transactions: Vec::new(),
            page_count: 1,
            last_cursor: None,
        };
        let report =
            publish_complete_snapshot(&request, &paths, false, complete, Duration::from_secs(23))
                .unwrap();
        let cached = load_cached_icrc_account_transactions(&request.cache).unwrap();
        assert_eq!(
            crate::subnet_catalog::parse_utc_timestamp_secs(&report.collection_started_at),
            Some(2_000_000_000)
        );
        assert_eq!(
            crate::subnet_catalog::parse_utc_timestamp_secs(&report.collection_completed_at),
            Some(2_000_000_023)
        );
        assert_eq!(
            cached.snapshot.collection_completed_at,
            report.collection_completed_at
        );
        std::fs::remove_dir_all(root).unwrap();
    }
}