rullst-core 12.1.0

use super::json_limit::{BoundedJsonError, encode_bounded, validate_bounded};
use super::types::validate_mutation;
use super::validation::validate_server_page;
use super::{OFFLINE_SNAPSHOT_VERSION, OfflineSyncError, OfflineSyncPolicy, OfflineSyncState};

impl OfflineSyncState {
    pub(super) fn encode(&self, policy: OfflineSyncPolicy) -> Result<Vec<u8>, OfflineSyncError> {
        self.validate_snapshot(policy)?;
        match encode_bounded(self, policy.max_snapshot_bytes()) {
            Ok(bytes) => Ok(bytes),
            Err(BoundedJsonError::LimitExceeded) => Err(OfflineSyncError::SnapshotTooLarge {
                maximum: policy.max_snapshot_bytes(),
            }),
            Err(BoundedJsonError::Encode(error)) => Err(OfflineSyncError::EncodeJson(error)),
        }
    }

    pub(super) fn validate_encoded_size(
        &self,
        policy: OfflineSyncPolicy,
    ) -> Result<(), OfflineSyncError> {
        self.validate_snapshot(policy)?;
        match validate_bounded(self, policy.max_snapshot_bytes()) {
            Ok(()) => Ok(()),
            Err(BoundedJsonError::LimitExceeded) => Err(OfflineSyncError::SnapshotTooLarge {
                maximum: policy.max_snapshot_bytes(),
            }),
            Err(BoundedJsonError::Encode(error)) => Err(OfflineSyncError::EncodeJson(error)),
        }
    }

    pub(super) fn decode(
        policy: OfflineSyncPolicy,
        bytes: &[u8],
    ) -> Result<Self, OfflineSyncError> {
        if bytes.len() > policy.max_snapshot_bytes() {
            return Err(OfflineSyncError::SnapshotTooLarge {
                maximum: policy.max_snapshot_bytes(),
            });
        }
        let state = serde_json::from_slice::<Self>(bytes).map_err(OfflineSyncError::InvalidJson)?;
        state.validate_snapshot(policy)?;
        Ok(state)
    }

    fn validate_snapshot(&self, policy: OfflineSyncPolicy) -> Result<(), OfflineSyncError> {
        if self.schema_version != OFFLINE_SNAPSHOT_VERSION {
            return Err(OfflineSyncError::UnsupportedSnapshotVersion(
                self.schema_version,
            ));
        }
        if self.records.len() > policy.max_records() {
            return Err(OfflineSyncError::RecordQuotaExceeded {
                maximum: policy.max_records(),
            });
        }
        if self.pending.len() > policy.max_pending_mutations() {
            return Err(OfflineSyncError::PendingQuotaExceeded {
                maximum: policy.max_pending_mutations(),
            });
        }
        if self.conflicts.len() > policy.max_conflicts() {
            return Err(OfflineSyncError::ConflictQuotaExceeded {
                maximum: policy.max_conflicts(),
            });
        }
        validate_server_page(&self.records, policy)?;
        for mutation in &self.pending {
            validate_mutation(mutation, policy)?;
        }
        for conflict in &self.conflicts {
            validate_mutation(&conflict.mutation, policy)?;
            if conflict.authoritative_revision == Some(0) {
                return Err(OfflineSyncError::InvalidRevision);
            }
        }
        let mut replay_keys = std::collections::HashSet::with_capacity(
            self.pending.len().saturating_add(self.conflicts.len()),
        );
        for mutation in &self.pending {
            if !replay_keys.insert(mutation.idempotency_key().as_str()) {
                return Err(OfflineSyncError::DuplicateIdempotencyKey);
            }
        }
        for conflict in &self.conflicts {
            if !replay_keys.insert(conflict.mutation.idempotency_key().as_str()) {
                return Err(OfflineSyncError::DuplicateIdempotencyKey);
            }
        }
        Ok(())
    }
}