ic-query 0.74.2

//! Module: snapshot_cache::attempt
//!
//! Responsibility: read and validate refresh-attempt sidecar files.
//! Does not own: command-specific attempt metadata or refresh execution.
//! Boundary: validates generic refresh-attempt JSON through cache-file primitives.

use crate::{
    cache::CacheRefreshAttemptStatus,
    cache_file::{
        BoundedManagedFileReadError, CacheFileError, ManagedReadBudget, read_bounded_managed_file,
    },
};
use ic_host_text::timestamp::format_utc_timestamp_secs;
use serde::{Deserialize as SerdeDeserialize, Serialize, de::DeserializeOwned};
use std::{
    path::{Path, PathBuf},
    time::{SystemTime, UNIX_EPOCH},
};

pub const SNAPSHOT_REFRESH_ATTEMPT_SCHEMA_VERSION: u32 = 1;

///
/// SnapshotRefreshAttemptReadError
///
/// Strict refresh-attempt sidecar read or parse failure.
///

#[derive(Debug)]
pub enum SnapshotRefreshAttemptReadError {
    /// Capability-rooted cache access failed.
    Operation(CacheFileError),
    Parse {
        path: PathBuf,
        source: serde_json::Error,
    },
    Invalid {
        path: PathBuf,
        reason: String,
    },
}

impl SnapshotRefreshAttemptReadError {
    pub fn into_error<E>(
        self,
        component: &'static str,
        cache: impl FnOnce(crate::HostCacheError) -> E,
        invalid: impl FnOnce(PathBuf, String) -> E,
    ) -> E {
        match self {
            Self::Operation(source) => cache(crate::HostCacheError::operation(component, source)),
            Self::Parse { path, source } => {
                cache(crate::HostCacheError::parse_cache(component, path, source))
            }
            Self::Invalid { path, reason } => invalid(path, reason),
        }
    }
}

///
/// SnapshotRefreshProgress
///
/// Page, row, and cursor progress shared by complete-snapshot refresh attempts.
///

#[derive(Clone, Debug, Default, Eq, PartialEq)]
pub struct SnapshotRefreshProgress {
    pub pages_fetched: u32,
    pub rows_fetched: usize,
    pub last_cursor: Option<String>,
}

impl SnapshotRefreshProgress {
    #[must_use]
    pub const fn new(pages_fetched: u32, rows_fetched: usize, last_cursor: Option<String>) -> Self {
        Self {
            pages_fetched,
            rows_fetched,
            last_cursor,
        }
    }
}

///
/// SnapshotRefreshAttempt
///
/// Sidecar status for an in-progress or failed complete snapshot refresh.
///

#[derive(Clone, Debug, Eq, PartialEq, SerdeDeserialize, Serialize)]
#[serde(deny_unknown_fields)]
pub struct SnapshotRefreshAttempt<Metadata> {
    pub schema_version: u32,
    pub network: String,
    pub source_endpoint: String,
    pub started_at: String,
    pub updated_at: String,
    pub metadata: Metadata,
    /// Current lifecycle state, serialized as its stable lowercase label.
    pub status: CacheRefreshAttemptStatus,
    pub page_size: u32,
    pub pages_fetched: u32,
    pub rows_fetched: usize,
    pub last_cursor: Option<String>,
    pub last_error: Option<String>,
}

impl<Metadata> From<SnapshotRefreshAttempt<Metadata>> for SnapshotRefreshProgress {
    fn from(attempt: SnapshotRefreshAttempt<Metadata>) -> Self {
        Self::new(
            attempt.pages_fetched,
            attempt.rows_fetched,
            attempt.last_cursor,
        )
    }
}

pub fn read_snapshot_refresh_attempt_strict<T>(
    cache_root: &Path,
    path: &Path,
    budget: Option<&mut ManagedReadBudget>,
) -> Result<Option<T>, SnapshotRefreshAttemptReadError>
where
    T: DeserializeOwned,
{
    let Some(data) = read_bounded_managed_file(
        cache_root,
        path,
        crate::cache_file::MAX_REFRESH_ATTEMPT_BYTES,
        budget,
    )
    .map_err(|error| snapshot_attempt_read_error(cache_root, error))?
    else {
        return Ok(None);
    };
    serde_json::from_slice(&data).map(Some).map_err(|source| {
        SnapshotRefreshAttemptReadError::Parse {
            path: path.to_path_buf(),
            source,
        }
    })
}

fn snapshot_attempt_read_error(
    cache_root: &Path,
    error: BoundedManagedFileReadError,
) -> SnapshotRefreshAttemptReadError {
    match error {
        BoundedManagedFileReadError::Operation(source) => {
            SnapshotRefreshAttemptReadError::Operation(source)
        }
        BoundedManagedFileReadError::Read { path, source } => {
            SnapshotRefreshAttemptReadError::Operation(CacheFileError::OpenManagedPath {
                root: cache_root.to_path_buf(),
                path,
                source,
            })
        }
        BoundedManagedFileReadError::LimitExceeded {
            path,
            actual,
            maximum,
        } => SnapshotRefreshAttemptReadError::Invalid {
            path,
            reason: format!("refresh-attempt file is {actual} bytes, maximum is {maximum}"),
        },
        BoundedManagedFileReadError::Accounting { path } => {
            SnapshotRefreshAttemptReadError::Invalid {
                path,
                reason: "refresh-attempt byte length cannot be represented safely".to_string(),
            }
        }
    }
}

pub fn current_attempt_timestamp(fallback: &str) -> String {
    SystemTime::now().duration_since(UNIX_EPOCH).map_or_else(
        |_| fallback.to_string(),
        |duration| format_utc_timestamp_secs(duration.as_secs()),
    )
}

/// Validate shared identity, progress and lifecycle/error consistency.
pub fn validate_snapshot_refresh_attempt<Metadata>(
    attempt: &SnapshotRefreshAttempt<Metadata>,
    expected_network: &str,
) -> Result<(), String> {
    if attempt.schema_version != SNAPSHOT_REFRESH_ATTEMPT_SCHEMA_VERSION {
        return Err(format!(
            "schema_version is {}, expected {}",
            attempt.schema_version, SNAPSHOT_REFRESH_ATTEMPT_SCHEMA_VERSION
        ));
    }
    if attempt.network != expected_network {
        return Err(format!(
            "network is {}, expected {expected_network}",
            attempt.network
        ));
    }
    if attempt.source_endpoint.is_empty() {
        return Err("source_endpoint must not be empty".to_string());
    }
    if attempt.started_at.is_empty() || attempt.updated_at.is_empty() {
        return Err("attempt timestamps must not be empty".to_string());
    }
    if attempt.page_size == 0 {
        return Err("page_size must be greater than zero".to_string());
    }
    if attempt.pages_fetched == 0 && (attempt.rows_fetched != 0 || attempt.last_cursor.is_some()) {
        return Err("zero-page attempt contains row or cursor progress".to_string());
    }
    let status = attempt.status;
    match status {
        CacheRefreshAttemptStatus::Running | CacheRefreshAttemptStatus::Complete
            if attempt.last_error.is_none() =>
        {
            Ok(())
        }
        CacheRefreshAttemptStatus::Failed
            if attempt
                .last_error
                .as_deref()
                .is_some_and(|error| !error.trim().is_empty()) =>
        {
            Ok(())
        }
        CacheRefreshAttemptStatus::Running | CacheRefreshAttemptStatus::Complete => {
            Err(format!("{status} attempt contains last_error"))
        }
        CacheRefreshAttemptStatus::Failed => {
            Err("failed attempt must contain last_error".to_string())
        }
    }
}