pdk-cache-lib 1.10.0

// Copyright (c) 2026, Salesforce, Inc.,
// All rights reserved.
// For full license text, see the LICENSE.txt file

//! Implementation of the [`crate::Cache`] using a FIFO approach.

use log;
use serde::{Deserialize, Serialize};
use std::collections::VecDeque;

use super::{error::CacheError, Cache};
use pdk_core::classy::shared_data::concurrent_shared_data::TransactionStatus;

cfg_if::cfg_if! {
    if #[cfg(test)] {
        use crate::mocks::MockConcurrentSharedData as ConcurrentSharedData;
    } else {
        use pdk_core::classy::shared_data::concurrent_shared_data::ConcurrentSharedData;
    }
}

enum OperationType {
    Insert,
    Remove,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
struct InternalData {
    max_entries_size: usize,
    fifo: VecDeque<String>,
    last_modified_item_key: String,
}

impl InternalData {
    fn new(max_entries_size: usize, fifo: VecDeque<String>, last_modified_item_key: &str) -> Self {
        Self {
            max_entries_size,
            fifo,
            last_modified_item_key: last_modified_item_key.to_string(),
        }
    }
}

/// A FIFO cache, that shares global state along all incoming requests of a policy.
///
/// As the name suggests, a FIFO eviction strategy is used, meaning that when the maximum number
/// of entries is reached, the first added cache entries will be the first to be evicted.
pub struct FifoCache {
    cache_id: String,
    cache_id_internal_data: String,
    concurrent_shared_data: ConcurrentSharedData,
    max_entries_size: usize,
}

impl FifoCache {
    pub(crate) fn new(
        cache_id: String,
        max_entries_size: usize,
        concurrent_shared_data: ConcurrentSharedData,
    ) -> Self {
        Self {
            cache_id_internal_data: format!("{}-internal_data", &cache_id),
            cache_id,
            max_entries_size,
            concurrent_shared_data,
        }
    }

    /// Get the cache's maximum number of entries available.
    pub fn max_entries_size(&self) -> usize {
        self.max_entries_size
    }

    /// Returns cache ID.
    pub fn cache_id(&self) -> &str {
        self.cache_id.as_str()
    }

    /// Get a reference to the fifo shared data cache's cache id internal data.
    fn cache_id_internal_data(&self) -> &str {
        self.cache_id_internal_data.as_ref()
    }

    /// Creates a key to be used by cache to store data, using cache instance ID as prefix.
    fn cache_key(&self, item_key: &str) -> String {
        format!("{}-{}", self.cache_id(), item_key)
    }

    fn new_internal_data(
        &self,
        fifo: VecDeque<String>,
        last_modified_item_key: &str,
    ) -> InternalData {
        InternalData::new(self.max_entries_size(), fifo, last_modified_item_key)
    }

    fn get_stored_internal_data(&self) -> Option<InternalData> {
        self.concurrent_shared_data
            .get::<InternalData>(self.cache_id_internal_data())
    }

    fn update_internal_data(&self, last_in: String, is_update: bool) {
        log::trace!(
            "Updating cache state data for key: {} (updated: {})",
            &last_in,
            is_update
        );

        if is_update {
            self.remove_item_from_internal_data(&last_in);
        }

        // update variable with shared_data info; otherwise init as empty.
        let mut internal_data: InternalData;
        if let Some(stored_internal_data) = self.get_stored_internal_data() {
            internal_data = stored_internal_data;
        } else {
            internal_data =
                self.new_internal_data(VecDeque::with_capacity(self.max_entries_size()), &last_in);
        }

        // insert in queue
        internal_data.fifo.push_back(last_in.clone());
        internal_data.last_modified_item_key = last_in;

        // Save updated internal data
        self.save_internal_data(&internal_data, OperationType::Insert);

        // check eviction
        self.check_eviction(&mut internal_data);
    }

    fn check_eviction(&self, internal_data: &mut InternalData) {
        let mut is_evicted = false;
        while internal_data.fifo.len() > self.max_entries_size() {
            let first_in = internal_data.fifo.pop_front().unwrap();

            self.evict(&first_in);
            internal_data.last_modified_item_key = first_in;
            is_evicted = true;
        }

        if is_evicted {
            self.save_internal_data(internal_data, OperationType::Remove);
            log::trace!("Cache evicted");
        }
    }

    fn evict(&self, key: &str) {
        let _deleted: Option<Vec<u8>> = self.concurrent_shared_data.remove(key);
    }

    fn save_internal_data(&self, new_internal_data: &InternalData, operation_type: OperationType) {
        // declare unique variable since destructuring assignments are unstable
        // see issue #71126 <https://github.com/rust-lang/rust/issues/71126> for more information
        // related issue: https://github.com/rust-lang/rust/issues/90095
        let insert_result: (TransactionStatus, InternalData) = match operation_type {
            OperationType::Insert => self.concurrent_shared_data.insert(
                self.cache_id_internal_data().to_string(),
                new_internal_data.to_owned(),
                handle_cache_state_consistency_when_insert,
            ),
            OperationType::Remove => self.concurrent_shared_data.insert(
                self.cache_id_internal_data().to_string(),
                new_internal_data.to_owned(),
                handle_cache_state_consistency_when_remove_or_evict,
            ),
        };

        if !matches!(insert_result.0, TransactionStatus::Complete) {
            log::error!(
                "Error updating internal cache state {}: {:?}",
                self.cache_id(),
                insert_result.0
            );
        }
    }

    fn remove_item_from_internal_data(&self, deleted_key: &str) {
        if let Some(mut stored_internal_data) = self.get_stored_internal_data() {
            // Remove from queue
            if let Some(found_index) =
                stored_internal_data
                    .fifo
                    .iter()
                    .enumerate()
                    .find_map(|(index, key)| {
                        if key.as_str() == deleted_key {
                            Some(index)
                        } else {
                            None
                        }
                    })
            {
                stored_internal_data.fifo.remove(found_index);
                stored_internal_data.last_modified_item_key = deleted_key.to_string();

                // Store updated value
                self.save_internal_data(&stored_internal_data, OperationType::Remove);
            }
        } else {
            log::trace!("Stored cache state data not found, nothing to clean from cache state");
        }
    }
}

/// In any case, if prev_value was already inserted or not, deleted or not; we decide override existing value with new one.
fn handle_cache_value_consistency(
    _stored_value: Vec<u8>,
    candidate_value: Vec<u8>,
) -> Option<Vec<u8>> {
    Some(candidate_value)
}

/// This method applies for insertion operations, and adds last inserted into stored state.
fn handle_cache_state_consistency_when_insert(
    stored_state: InternalData,
    candidate_state: InternalData,
) -> Option<InternalData> {
    let mut new_fifo: VecDeque<String> = stored_state.fifo;

    if let Some(index) = new_fifo.iter().enumerate().find_map(|(index, key)| {
        if key.as_str() == candidate_state.last_modified_item_key.as_str() {
            Some(index)
        } else {
            None
        }
    }) {
        new_fifo.remove(index);
    }
    new_fifo.push_back(candidate_state.last_modified_item_key.clone());

    Some(InternalData::new(
        stored_state.max_entries_size,
        new_fifo,
        candidate_state.last_modified_item_key.as_str(),
    ))
}

/// This method applies for removing operations, and removes all elements not present in candidate state from stored state.
fn handle_cache_state_consistency_when_remove_or_evict(
    stored_state: InternalData,
    candidate_state: InternalData,
) -> Option<InternalData> {
    let mut new_fifo: VecDeque<String> = stored_state.fifo;
    new_fifo.retain(|key| key.as_str() != candidate_state.last_modified_item_key.as_str());

    Some(InternalData::new(
        stored_state.max_entries_size,
        new_fifo,
        candidate_state.last_modified_item_key.as_str(),
    ))
}

impl Cache for FifoCache {
    fn save(&self, key: &str, value: Vec<u8>) -> Result<(), CacheError> {
        // store data in shared data
        let cache_key = self.cache_key(key);

        // First, verify if key is already stored in cache,
        // in order to see if we need to remove element from FIFO and then update cache item correctly.
        let is_update = self
            .concurrent_shared_data
            .get::<Vec<u8>>(cache_key.as_str())
            .is_some();

        // After cleaning up FIFO, let's udpate cache item.
        let (tx_status, _saved_data) =
            self.concurrent_shared_data
                .insert(cache_key, value, handle_cache_value_consistency);

        match tx_status {
            TransactionStatus::Complete => {
                self.update_internal_data(self.cache_key(key), is_update);
            }
            _ => {
                return Err(CacheError::SaveError {
                    status: format!("{tx_status:?}"),
                    cache_id: String::from(self.cache_id()),
                })
            }
        }
        Ok(())
    }

    fn get(&self, key: &str) -> Option<Vec<u8>> {
        let cache_key = self.cache_key(key);
        self.concurrent_shared_data
            .get::<Vec<u8>>(cache_key.as_str())
    }

    fn delete(&self, key: &str) -> Option<Vec<u8>> {
        log::trace!("Delete item '{key}' from cache");
        let delete_key = self.cache_key(key);
        let deleted = self.concurrent_shared_data.remove(&delete_key);
        self.remove_item_from_internal_data(&delete_key);
        deleted
    }

    fn purge(&self) {
        if let Some(cache_internal_data) = self.get_stored_internal_data() {
            // Clear cache items
            cache_internal_data.fifo.iter().for_each(|cache_key| {
                let _removed: Option<Vec<u8>> = self.concurrent_shared_data.remove(cache_key);
            });
            // Clear internal data
            let _internal_data: Option<InternalData> = self
                .concurrent_shared_data
                .remove(self.cache_id_internal_data());
        }
    }
}

/// Shared value store backing the stateful test mock: maps stored key to bytes.
#[cfg(test)]
pub(crate) type SharedValueStore =
    std::rc::Rc<std::cell::RefCell<std::collections::HashMap<String, Vec<u8>>>>;

/// Builds a stateful mock `ConcurrentSharedData` for tests, backed by the given
/// shared value store so TTL-framed bytes survive a save/get roundtrip and are
/// removed on delete. This lives at module scope (not in `mod tests`) because it
/// must name the private [`InternalData`] type; the `build()`-path test in
/// builder.rs reuses it through [`ConcurrentSharedData::new`].
#[cfg(test)]
pub(crate) fn stateful_shared_data_mock(store: SharedValueStore) -> ConcurrentSharedData {
    use std::rc::Rc;

    let mut mock = ConcurrentSharedData::default();

    // Value reads: return whatever was persisted for the key. `returning_st` is
    // used because the shared store is `Rc<RefCell<..>>` (not `Send`); these
    // tests run single-threaded.
    let get_store = Rc::clone(&store);
    mock.expect_get::<Vec<u8>>()
        .times(0..)
        .returning_st(move |key: &str| get_store.borrow().get(key).cloned());

    // Value writes: persist the framed bytes exactly as the cache sends them.
    let insert_store = Rc::clone(&store);
    mock.expect_insert::<Vec<u8>>().times(0..).returning_st(
        move |key: String, value: Vec<u8>, _handle_consistency| {
            insert_store.borrow_mut().insert(key, value.clone());
            (TransactionStatus::Complete, value)
        },
    );

    // Value removals: drop the key from the shared store.
    let remove_store = Rc::clone(&store);
    mock.expect_remove::<Vec<u8>>()
        .times(0..)
        .returning_st(move |key: &str| remove_store.borrow_mut().remove(key));

    // Internal FIFO bookkeeping: a fresh empty state is fine for a single-key
    // roundtrip, and accepting every write keeps the real FIFO save/delete
    // paths satisfied without asserting internal layout.
    mock.expect_get::<InternalData>()
        .times(0..)
        .returning(|_key: &str| None);
    mock.expect_insert::<InternalData>().times(0..).returning(
        |_key: String, value: InternalData, _handle_consistency| {
            (TransactionStatus::Complete, value)
        },
    );
    mock.expect_remove::<InternalData>()
        .times(0..)
        .returning(|_key: &str| None);

    mock
}

#[cfg(test)]
mod tests {
    use std::collections::{HashMap, VecDeque};

    use mockall::{
        predicate::{always, eq},
        Sequence,
    };
    use serde::{Deserialize, Serialize};

    use super::Cache;
    use super::ConcurrentSharedData as MockConcurrentSharedData;
    use super::{
        handle_cache_state_consistency_when_insert,
        handle_cache_state_consistency_when_remove_or_evict, handle_cache_value_consistency,
        FifoCache, InternalData,
    };
    use pdk_core::classy::shared_data::concurrent_shared_data::TransactionStatus;

    #[derive(Debug, Serialize, Deserialize, PartialEq, Copy, Clone)]
    struct SerializableObject {
        property_one: u64,
        property_two: u64,
    }

    #[test]
    fn save_in_cache_first_time() {
        let obj = SerializableObject {
            property_one: 1,
            property_two: 10,
        };
        let serialized_obj = serde_fixint::to_vec(&obj).unwrap();

        let mut mock_concurrent_shared_data = MockConcurrentSharedData::default();
        let mut sequence = Sequence::new();

        mock_concurrent_shared_data
            .expect_get::<Vec<u8>>()
            .with(eq("test-cache-key1"))
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x: &str| None);

        mock_concurrent_shared_data
            .expect_insert()
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x, value: Vec<u8>, _handle_consistency| {
                (TransactionStatus::Complete, value)
            });

        mock_concurrent_shared_data
            .expect_get::<InternalData>()
            .with(eq("test-cache-internal_data"))
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x: &str| None);

        mock_concurrent_shared_data
            .expect_insert()
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x, value: InternalData, _handle_consistency| {
                (TransactionStatus::Complete, value)
            });

        let cache = FifoCache::new(String::from("test-cache"), 3, mock_concurrent_shared_data);
        let value = serialized_obj.as_slice().to_vec();
        assert!(cache.save("key1", value).is_ok());
    }

    #[test]
    fn save_in_cache_already_exists() {
        let obj = SerializableObject {
            property_one: 1,
            property_two: 10,
        };
        let serialized_obj = serde_fixint::to_vec(&obj).unwrap();

        let stored_obj = SerializableObject {
            property_one: 1,
            property_two: 10,
        };
        let serialized_stored_obj = serde_fixint::to_vec(&stored_obj).unwrap();

        let stored_fifo = VecDeque::from([String::from("test-cache-key1")]);
        let stored_internal_data = InternalData::new(3, stored_fifo, "test-cache-key1");

        let mut mock_concurrent_shared_data = MockConcurrentSharedData::default();
        let mut sequence = Sequence::new();

        mock_concurrent_shared_data
            .expect_get::<Vec<u8>>()
            .with(eq("test-cache-key1"))
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x: &str| Some(serialized_stored_obj.clone()));

        mock_concurrent_shared_data
            .expect_insert()
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x, value: Vec<u8>, _handle_consistency| {
                (TransactionStatus::Complete, value)
            });

        mock_concurrent_shared_data
            .expect_get::<InternalData>()
            .with(eq("test-cache-internal_data"))
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x: &str| Some(stored_internal_data.clone()));

        mock_concurrent_shared_data
            .expect_insert()
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x, value: InternalData, _handle_consistency| {
                (TransactionStatus::Complete, value)
            });

        let stored_internal_data = InternalData::new(3, VecDeque::new(), "test-cache-key1");

        mock_concurrent_shared_data
            .expect_get::<InternalData>()
            .with(eq("test-cache-internal_data"))
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x: &str| Some(stored_internal_data.clone()));

        mock_concurrent_shared_data
            .expect_insert()
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x, value: InternalData, _handle_consistency| {
                (TransactionStatus::Complete, value)
            });

        let cache = FifoCache::new(String::from("test-cache"), 3, mock_concurrent_shared_data);
        let value = serialized_obj.as_slice().to_vec();
        assert!(cache.save("key1", value).is_ok());
    }

    #[test]
    fn save_in_cache_when_shared_data_fails_storing_value() {
        let obj = SerializableObject {
            property_one: 1,
            property_two: 10,
        };
        let serialized_obj = serde_fixint::to_vec(&obj).unwrap();

        let mut mock_concurrent_shared_data = MockConcurrentSharedData::default();
        let mut sequence = Sequence::new();

        mock_concurrent_shared_data
            .expect_get::<Vec<u8>>()
            .with(eq("test-cache-key1"))
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x: &str| None);

        mock_concurrent_shared_data
            .expect_insert()
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x, value: Vec<u8>, _handle_consistency| {
                (TransactionStatus::InternalError, value)
            });

        let cache = FifoCache::new(String::from("test-cache"), 3, mock_concurrent_shared_data);
        let value = serialized_obj.as_slice().to_vec();
        assert!(cache.save("key1", value).is_err());
    }

    // TODO: W-10712428 Update test when implement rollback for errors when save internal data.
    #[test]
    fn save_in_cache_when_shared_data_fails_storing_internaldata() {
        let obj = SerializableObject {
            property_one: 1,
            property_two: 10,
        };
        let serialized_obj = serde_fixint::to_vec(&obj).unwrap();

        let stored_obj = SerializableObject {
            property_one: 1,
            property_two: 10,
        };
        let serialized_stored_obj = serde_fixint::to_vec(&stored_obj).unwrap();

        let stored_fifo = VecDeque::from([String::from("test-cache-key1")]);
        let stored_internal_data = InternalData::new(3, stored_fifo, "test-cache-key1");

        let mut mock_concurrent_shared_data = MockConcurrentSharedData::default();
        let mut sequence = Sequence::new();

        mock_concurrent_shared_data
            .expect_get::<Vec<u8>>()
            .with(eq("test-cache-key1"))
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x: &str| Some(serialized_stored_obj.clone()));

        mock_concurrent_shared_data
            .expect_insert()
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x, value: Vec<u8>, _handle_consistency| {
                (TransactionStatus::Complete, value)
            });

        mock_concurrent_shared_data
            .expect_get::<InternalData>()
            .with(eq("test-cache-internal_data"))
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x: &str| Some(stored_internal_data.clone()));

        mock_concurrent_shared_data
            .expect_insert()
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x, value: InternalData, _handle_consistency| {
                (TransactionStatus::InternalError, value)
            });

        let stored_internal_data = InternalData::new(3, VecDeque::new(), "test-cache-key1");

        mock_concurrent_shared_data
            .expect_get::<InternalData>()
            .with(eq("test-cache-internal_data"))
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x: &str| Some(stored_internal_data.clone()));

        mock_concurrent_shared_data
            .expect_insert()
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x, value: InternalData, _handle_consistency| {
                (TransactionStatus::Complete, value)
            });

        let cache = FifoCache::new(String::from("test-cache"), 3, mock_concurrent_shared_data);
        let value = serialized_obj.as_slice().to_vec();
        assert!(cache.save("key1", value).is_ok());
    }

    #[test]
    fn save_in_cache_first_time_when_cache_has_elements() {
        let obj = SerializableObject {
            property_one: 1,
            property_two: 10,
        };
        let serialized_obj = serde_fixint::to_vec(&obj).unwrap();

        let stored_fifo = VecDeque::from([
            String::from("test-cache-keyA"),
            String::from("test-cache-keyB"),
        ]);
        let stored_internal_data = InternalData::new(3, stored_fifo, "test-cache-keyB");

        let mut mock_concurrent_shared_data = MockConcurrentSharedData::default();
        let mut sequence = Sequence::new();

        mock_concurrent_shared_data
            .expect_get::<Vec<u8>>()
            .with(eq("test-cache-key1"))
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x: &str| None);

        mock_concurrent_shared_data
            .expect_insert()
            .times(1)
            .with(eq(String::from("test-cache-key1")), always(), always())
            .in_sequence(&mut sequence)
            .returning(move |_x, value: Vec<u8>, _handle_consistency| {
                (TransactionStatus::Complete, value)
            });

        mock_concurrent_shared_data
            .expect_get::<InternalData>()
            .with(eq("test-cache-internal_data"))
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x: &str| Some(stored_internal_data.clone()));

        mock_concurrent_shared_data
            .expect_insert()
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x, value: InternalData, _handle_consistency| {
                (TransactionStatus::InternalError, value)
            });

        let cache = FifoCache::new(String::from("test-cache"), 3, mock_concurrent_shared_data);
        let value = serialized_obj.as_slice().to_vec();
        assert!(cache.save("key1", value).is_ok());
    }

    #[test]
    fn save_in_cache_when_cache_is_full_needs_eviction() {
        let obj = SerializableObject {
            property_one: 1,
            property_two: 10,
        };
        let serialized_obj = serde_fixint::to_vec(&obj).unwrap();

        let stored_fifo = VecDeque::from([
            String::from("test-cache-keyA"),
            String::from("test-cache-keyB"),
            String::from("test-cache-keyC"),
        ]);
        let stored_internal_data = InternalData::new(3, stored_fifo, "test-cache-keyC");

        let mut mock_concurrent_shared_data = MockConcurrentSharedData::default();
        let mut sequence = Sequence::new();

        mock_concurrent_shared_data
            .expect_get::<Vec<u8>>()
            .with(eq("test-cache-key1"))
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x: &str| None);

        mock_concurrent_shared_data
            .expect_insert()
            .times(1)
            .with(eq("test-cache-key1".to_string()), always(), always())
            .in_sequence(&mut sequence)
            .returning(move |_x, value: Vec<u8>, _handle_consistency| {
                (TransactionStatus::Complete, value)
            });

        mock_concurrent_shared_data
            .expect_get::<InternalData>()
            .with(eq("test-cache-internal_data"))
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x: &str| Some(stored_internal_data.clone()));

        mock_concurrent_shared_data
            .expect_insert()
            .times(1)
            .with(
                eq(String::from("test-cache-internal_data")),
                always(),
                always(),
            )
            .in_sequence(&mut sequence)
            .returning(move |_x, value: InternalData, _handle_consistency| {
                (TransactionStatus::Complete, value)
            });

        // eviction value
        mock_concurrent_shared_data
            .expect_remove::<Vec<u8>>()
            .times(1)
            .with(eq("test-cache-keyA"))
            .in_sequence(&mut sequence)
            .returning(move |_key_to_evict| None);

        let stored_fifo = VecDeque::from([
            String::from("test-cache-keyA"),
            String::from("test-cache-keyB"),
            String::from("test-cache-keyC"),
            String::from("test-cache-key1"),
        ]);
        let _stored_internal_data = InternalData::new(3, stored_fifo, "test-cache-key1");

        // update after eviction
        mock_concurrent_shared_data
            .expect_insert()
            .times(1)
            .with(
                eq(String::from("test-cache-internal_data")),
                always(),
                always(),
            )
            .in_sequence(&mut sequence)
            .returning(move |_x, value: InternalData, _handle_consistency| {
                (TransactionStatus::Complete, value)
            });

        let cache = FifoCache::new(String::from("test-cache"), 3, mock_concurrent_shared_data);
        let value = serialized_obj.as_slice().to_vec();
        assert!(cache.save("key1", value).is_ok());
    }

    #[test]
    fn save_in_cache_when_cache_item_has_version_mismatch() {
        let obj = SerializableObject {
            property_one: 1,
            property_two: 10,
        };
        let serialized_obj = serde_fixint::to_vec(&obj).unwrap();

        let mut mock_concurrent_shared_data = MockConcurrentSharedData::default();
        let mut sequence = Sequence::new();

        mock_concurrent_shared_data
            .expect_get::<Vec<u8>>()
            .with(eq("test-cache-key1"))
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x: &str| None);

        mock_concurrent_shared_data
            .expect_insert()
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x, value: Vec<u8>, _handle_consistency| {
                (TransactionStatus::Rejected, value)
            });

        let cache = FifoCache::new(String::from("test-cache"), 3, mock_concurrent_shared_data);
        let value = serialized_obj.as_slice().to_vec();
        assert!(cache.save("key1", value).is_err());
    }

    #[test]
    fn test_handle_cache_value_consistency() {
        let vec1 = "item1".as_bytes().to_vec();
        let vec2 = "item2".as_bytes().to_vec();
        assert_eq!(
            handle_cache_value_consistency(vec1, vec2.clone()).unwrap(),
            vec2
        );
    }

    #[test]
    fn test_handle_cache_value_consistency_when_remove_or_evict() {
        let vec1 = "item1".as_bytes().to_vec();
        let vec2 = "".as_bytes().to_vec();
        assert_eq!(
            handle_cache_value_consistency(vec1, vec2.clone()).unwrap(),
            vec2
        );
    }

    #[test]
    fn test_handle_cache_state_consistency_when_insert() {
        let fifo1 = VecDeque::from(["A".to_string(), "B".to_string(), "C".to_string()]);
        let state1: InternalData = InternalData::new(3, fifo1, "C");

        let fifo2 = VecDeque::from(["A".to_string(), "B".to_string(), "D".to_string()]);
        let state2: InternalData = InternalData::new(3, fifo2, "D");

        let merged_state = handle_cache_state_consistency_when_insert(state1, state2);
        assert!(merged_state.is_some());

        let merged_internal_data = merged_state.unwrap();
        assert_eq!(merged_internal_data.fifo.len(), 4);
        assert!(merged_internal_data.fifo.contains(&"A".to_string()));
        assert!(merged_internal_data.fifo.contains(&"B".to_string()));
        assert!(merged_internal_data.fifo.contains(&"C".to_string()));
        assert_eq!(merged_internal_data.fifo.back().unwrap(), &"D".to_string());
        assert_eq!(merged_internal_data.last_modified_item_key, "D".to_string());
    }

    #[test]
    fn test_handle_cache_state_consistency_when_insert_with_repeated_keys() {
        let fifo1 = VecDeque::from(["A".to_string(), "B".to_string(), "D".to_string()]);
        let state1: InternalData = InternalData::new(3, fifo1, "D");

        let fifo2 = VecDeque::from(["A".to_string(), "B".to_string(), "D".to_string()]);
        let state2: InternalData = InternalData::new(3, fifo2, "D");

        let merged_state = handle_cache_state_consistency_when_insert(state1, state2);
        assert!(merged_state.is_some());

        let merged_internal_data = merged_state.unwrap();
        assert_eq!(merged_internal_data.fifo.len(), 3);
        assert!(merged_internal_data.fifo.contains(&"A".to_string()));
        assert!(merged_internal_data.fifo.contains(&"B".to_string()));
        assert_eq!(merged_internal_data.fifo.back().unwrap(), &"D".to_string());
        assert_eq!(merged_internal_data.last_modified_item_key, "D".to_string());
    }

    #[test]
    fn test_handle_cache_state_consistency_when_insert_with_initial_less_keys() {
        let fifo1 = VecDeque::from(["A".to_string(), "C".to_string()]);
        let state1: InternalData = InternalData::new(3, fifo1, "C");

        let fifo2 = VecDeque::from(["A".to_string(), "B".to_string(), "D".to_string()]);
        let state2: InternalData = InternalData::new(3, fifo2, "D");

        let merged_state = handle_cache_state_consistency_when_insert(state1, state2);
        assert!(merged_state.is_some());

        let merged_internal_data = merged_state.unwrap();
        assert_eq!(merged_internal_data.fifo.len(), 3);
        assert!(merged_internal_data.fifo.contains(&"A".to_string()));
        assert!(merged_internal_data.fifo.contains(&"C".to_string()));
        assert_eq!(merged_internal_data.fifo.back().unwrap(), &"D".to_string());
        assert_eq!(merged_internal_data.last_modified_item_key, "D".to_string());
    }

    #[test]
    fn test_handle_cache_state_consistency_when_insert_with_initial_more_keys() {
        let fifo1 = VecDeque::from([
            "A".to_string(),
            "C".to_string(),
            "D".to_string(),
            "E".to_string(),
        ]);
        let state1: InternalData = InternalData::new(3, fifo1, "C");

        let fifo2 = VecDeque::from(["A".to_string(), "B".to_string(), "F".to_string()]);
        let state2: InternalData = InternalData::new(3, fifo2, "F");

        let merged_state = handle_cache_state_consistency_when_insert(state1, state2);
        assert!(merged_state.is_some());

        let merged_internal_data = merged_state.unwrap();
        assert_eq!(merged_internal_data.fifo.len(), 5);
        assert!(merged_internal_data.fifo.contains(&"A".to_string()));
        assert!(merged_internal_data.fifo.contains(&"C".to_string()));
        assert!(merged_internal_data.fifo.contains(&"D".to_string()));
        assert!(merged_internal_data.fifo.contains(&"E".to_string()));
        assert_eq!(merged_internal_data.fifo.back().unwrap(), &"F".to_string());
        assert_eq!(merged_internal_data.last_modified_item_key, "F".to_string());
    }

    #[test]
    fn test_handle_cache_state_consistency_when_insert_with_initial_more_keys_and_repeated() {
        let fifo1 = VecDeque::from([
            "A".to_string(),
            "C".to_string(),
            "D".to_string(),
            "E".to_string(),
        ]);
        let state1: InternalData = InternalData::new(3, fifo1, "C");

        let fifo2 = VecDeque::from(["A".to_string(), "B".to_string(), "D".to_string()]);
        let state2: InternalData = InternalData::new(3, fifo2, "D");

        let merged_state = handle_cache_state_consistency_when_insert(state1, state2);
        assert!(merged_state.is_some());

        let merged_internal_data = merged_state.unwrap();
        assert_eq!(merged_internal_data.fifo.len(), 4);
        assert!(merged_internal_data.fifo.contains(&"A".to_string()));
        assert!(merged_internal_data.fifo.contains(&"C".to_string()));
        assert!(merged_internal_data.fifo.contains(&"E".to_string()));
        assert_eq!(merged_internal_data.fifo.back().unwrap(), &"D".to_string());
        assert_eq!(merged_internal_data.last_modified_item_key, "D".to_string());
    }

    #[test]
    fn test_handle_cache_state_consistency_when_remove_or_evict() {
        let fifo1 = VecDeque::from(["A".to_string(), "B".to_string(), "C".to_string()]);
        let state1: InternalData = InternalData::new(3, fifo1, "C");

        let fifo2 = VecDeque::from(["A".to_string(), "C".to_string()]);
        let state2: InternalData = InternalData::new(3, fifo2, "B");

        let merged_state = handle_cache_state_consistency_when_remove_or_evict(state1, state2);
        assert!(merged_state.is_some());

        let merged_internal_data = merged_state.unwrap();
        assert_eq!(merged_internal_data.fifo.len(), 2);
        assert!(merged_internal_data.fifo.contains(&"A".to_string()));
        assert!(merged_internal_data.fifo.contains(&"C".to_string()));
        assert_eq!(merged_internal_data.last_modified_item_key, "B".to_string());
    }

    #[test]
    fn test_handle_cache_state_consistency_when_remove_or_evict_with_prev_insert() {
        let fifo1 = VecDeque::from([
            "A".to_string(),
            "B".to_string(),
            "C".to_string(),
            "D".to_string(),
        ]);
        let state1: InternalData = InternalData::new(3, fifo1, "C");

        let fifo2 = VecDeque::from(["A".to_string(), "C".to_string()]);
        let state2: InternalData = InternalData::new(3, fifo2, "B");

        let merged_state = handle_cache_state_consistency_when_remove_or_evict(state1, state2);
        assert!(merged_state.is_some());

        let merged_internal_data = merged_state.unwrap();
        assert_eq!(merged_internal_data.fifo.len(), 3);
        assert!(merged_internal_data.fifo.contains(&"A".to_string()));
        assert!(merged_internal_data.fifo.contains(&"C".to_string()));
        assert!(merged_internal_data.fifo.contains(&"D".to_string()));
        assert_eq!(merged_internal_data.last_modified_item_key, "B".to_string());
    }

    #[test]
    fn remove_from_cache_when_item_exists() {
        let stored_fifo = VecDeque::from([
            String::from("test-cache-keyA"),
            String::from("test-cache-key1"),
        ]);
        let stored_internal_data = InternalData::new(3, stored_fifo, "test-cache-key1");

        let mut mock_concurrent_shared_data = MockConcurrentSharedData::default();
        let mut sequence = Sequence::new();

        mock_concurrent_shared_data
            .expect_remove()
            .times(1)
            .with(eq("test-cache-key1"))
            .in_sequence(&mut sequence)
            .returning(move |_key| Some(Vec::<u8>::new()));

        mock_concurrent_shared_data
            .expect_get::<InternalData>()
            .with(eq("test-cache-internal_data"))
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x: &str| Some(stored_internal_data.clone()));

        mock_concurrent_shared_data
            .expect_insert()
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x, value: InternalData, _handle_consistency| {
                (TransactionStatus::Complete, value)
            });

        let cache = FifoCache::new(String::from("test-cache"), 3, mock_concurrent_shared_data);
        cache.delete("key1");
    }

    #[test]
    fn remove_from_cache_when_item_is_deleted() {
        let stored_fifo = VecDeque::from([
            String::from("test-cache-keyA"),
            String::from("test-cache-keyB"),
        ]);
        let stored_internal_data = InternalData::new(3, stored_fifo, "test-cache-keyB");

        let mut mock_concurrent_shared_data = MockConcurrentSharedData::default();
        let mut sequence = Sequence::new();

        mock_concurrent_shared_data
            .expect_remove::<Vec<u8>>()
            .times(1)
            .with(eq("test-cache-key1"))
            .in_sequence(&mut sequence)
            .returning(move |_key| None);

        mock_concurrent_shared_data
            .expect_get::<InternalData>()
            .with(eq("test-cache-internal_data"))
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x: &str| Some(stored_internal_data.clone()));

        mock_concurrent_shared_data
            .expect_insert()
            .times(0)
            .in_sequence(&mut sequence)
            .returning(move |_x, value: InternalData, _handle_consistency| {
                (TransactionStatus::Complete, value)
            });

        let cache = FifoCache::new(String::from("test-cache"), 3, mock_concurrent_shared_data);
        cache.delete("key1");
    }

    #[test]
    fn remove_from_cache_when_item_is_deleted_but_it_is_already_removed_from_state() {
        let stored_fifo = VecDeque::from([
            String::from("test-cache-keyA"),
            String::from("test-cache-keyB"),
        ]);
        let stored_internal_data = InternalData::new(3, stored_fifo, "test-cache-keyB");

        let mut mock_concurrent_shared_data = MockConcurrentSharedData::default();
        let mut sequence = Sequence::new();

        mock_concurrent_shared_data
            .expect_remove()
            .times(1)
            .with(eq("test-cache-key1"))
            .in_sequence(&mut sequence)
            .returning(move |_key| Some(Vec::<u8>::new()));

        mock_concurrent_shared_data
            .expect_get::<InternalData>()
            .with(eq("test-cache-internal_data"))
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x: &str| Some(stored_internal_data.clone()));

        mock_concurrent_shared_data
            .expect_insert()
            .times(0)
            .in_sequence(&mut sequence)
            .returning(move |_x, value: InternalData, _handle_consistency| {
                (TransactionStatus::Complete, value)
            });

        let cache = FifoCache::new(String::from("test-cache"), 3, mock_concurrent_shared_data);
        cache.delete("key1");
    }

    // TODO: W-10712428 Update test when implement rollback for errors when save internal data.
    #[test]
    fn remove_from_cache_when_shared_data_fails_updating_state() {
        let stored_fifo = VecDeque::from([String::from("test-cache-key1")]);
        let stored_internal_data = InternalData::new(3, stored_fifo, "test-cache-key1");

        let mut mock_concurrent_shared_data = MockConcurrentSharedData::default();
        let mut sequence = Sequence::new();

        mock_concurrent_shared_data
            .expect_remove()
            .times(1)
            .with(eq("test-cache-key1"))
            .in_sequence(&mut sequence)
            .returning(move |_key| Some(Vec::<u8>::new()));

        mock_concurrent_shared_data
            .expect_get::<InternalData>()
            .with(eq("test-cache-internal_data"))
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x: &str| Some(stored_internal_data.clone()));

        mock_concurrent_shared_data
            .expect_insert()
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x, value: InternalData, _handle_consistency| {
                (TransactionStatus::Rejected, value)
            });

        let cache = FifoCache::new(String::from("test-cache"), 3, mock_concurrent_shared_data);
        cache.delete("key1");
    }

    #[test]
    fn get_item_from_cache_ok() {
        let obj = SerializableObject {
            property_one: 1,
            property_two: 10,
        };
        let serialized_obj = serde_fixint::to_vec(&obj).unwrap();

        let mut mock_concurrent_shared_data = MockConcurrentSharedData::default();

        mock_concurrent_shared_data
            .expect_get::<Vec<u8>>()
            .with(eq("test-cache-key1"))
            .times(1)
            .returning(move |_x: &str| Some(serialized_obj.clone()));

        let cache = FifoCache::new(String::from("test-cache"), 3, mock_concurrent_shared_data);

        let value = cache.get("key1");
        assert!(value.is_some());
        assert_eq!(value.unwrap(), serde_fixint::to_vec(&obj).unwrap());
    }

    #[test]
    fn get_item_from_cache_empty() {
        let mut mock_concurrent_shared_data = MockConcurrentSharedData::default();

        mock_concurrent_shared_data
            .expect_get::<Vec<u8>>()
            .with(eq("test-cache-key1"))
            .times(1)
            .returning(move |_x: &str| None);

        let cache = FifoCache::new(String::from("test-cache"), 3, mock_concurrent_shared_data);
        assert!(cache.get("key1").is_none());
    }

    #[test]
    fn test_purge() {
        let stored_fifo = VecDeque::from([
            String::from("test-cache-key1"),
            String::from("test-cache-key2"),
        ]);
        let stored_internal_data = InternalData::new(3, stored_fifo, "test-cache-key2");

        let mut mock_concurrent_shared_data = MockConcurrentSharedData::default();
        let mut sequence = Sequence::new();

        mock_concurrent_shared_data
            .expect_get::<InternalData>()
            .with(eq("test-cache-internal_data"))
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x: &str| Some(stored_internal_data.clone()));

        mock_concurrent_shared_data
            .expect_remove()
            .times(1)
            .with(eq("test-cache-key1"))
            .in_sequence(&mut sequence)
            .returning(move |_key| Some(Vec::<u8>::new()));

        mock_concurrent_shared_data
            .expect_remove()
            .times(1)
            .with(eq("test-cache-key2"))
            .in_sequence(&mut sequence)
            .returning(move |_key| Some(Vec::<u8>::new()));

        mock_concurrent_shared_data
            .expect_remove()
            .times(1)
            .with(eq("test-cache-internal_data"))
            .in_sequence(&mut sequence)
            .returning(move |_key| Some(InternalData::new(3, VecDeque::new(), "test-cache-key2")));

        let cache = FifoCache::new(String::from("test-cache"), 3, mock_concurrent_shared_data);
        cache.purge();
    }

    #[test]
    fn test_purge_cache_empty() {
        let mut mock_concurrent_shared_data = MockConcurrentSharedData::default();
        let mut sequence = Sequence::new();

        mock_concurrent_shared_data
            .expect_get::<InternalData>()
            .with(eq("test-cache-internal_data"))
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x: &str| None);

        mock_concurrent_shared_data
            .expect_remove::<Vec<u8>>()
            .times(0)
            .in_sequence(&mut sequence);

        mock_concurrent_shared_data
            .expect_remove::<InternalData>()
            .times(0)
            .in_sequence(&mut sequence);

        let cache = FifoCache::new(String::from("test-cache"), 3, mock_concurrent_shared_data);
        cache.purge();
    }

    // TtlCache wrapping a real FifoCache: verifies TTL decoration composes with
    // the genuine FIFO backend. The FifoCache internal data type is private to
    // this module, so this composition test lives here rather than in builder.rs.
    // The mock is stateful (backed by a shared HashMap) so TTL-framed bytes
    // survive a save/get roundtrip and are removed on the post-expiry delete.

    use crate::ttl_cache::TtlCache;
    use pdk_core::classy::Clock;
    use std::cell::RefCell;
    use std::rc::Rc;
    use std::time::{Duration, SystemTime};

    // A clock whose time can be advanced manually, to drive TTL expiry.
    struct ManualClock {
        current: RefCell<SystemTime>,
    }

    impl ManualClock {
        fn new() -> Self {
            Self {
                current: RefCell::new(SystemTime::now()),
            }
        }

        fn move_forward(&self, duration: Duration) {
            let new_time = self.get_current_time() + duration;
            *self.current.borrow_mut() = new_time;
        }
    }

    impl Clock for ManualClock {
        fn get_current_time(&self) -> SystemTime {
            *self.current.borrow()
        }
    }

    // Shared value store backing the stateful mock: maps stored key to bytes.
    type SharedStore = Rc<RefCell<HashMap<String, Vec<u8>>>>;

    // Builds a real FifoCache backed by the shared stateful mock store. The
    // returned `SharedStore` is the shared value store so tests can observe what
    // actually persisted through the real FIFO paths.
    fn real_fifo_with_stateful_store(
        cache_id: &str,
        max_entries: usize,
    ) -> (FifoCache, SharedStore) {
        let store: SharedStore = Rc::new(RefCell::new(HashMap::new()));
        let mock = super::stateful_shared_data_mock(Rc::clone(&store));
        let cache = FifoCache::new(cache_id.to_string(), max_entries, mock);
        (cache, store)
    }

    #[test]
    fn ttl_wrapping_real_fifo_cache_present_before_ttl() {
        let manual_clock = Rc::new(ManualClock::new());
        let clock: Rc<dyn Clock> = Rc::clone(&manual_clock) as Rc<dyn Clock>;

        let (fifo, _store) = real_fifo_with_stateful_store("test-ttl-fifo", 3);
        let ttl_cache = TtlCache::new(fifo, Rc::clone(&clock), Duration::from_secs(10));

        ttl_cache.save("key1", b"value1".to_vec()).unwrap();

        // Within the TTL window the real FIFO backend still holds the value.
        manual_clock.move_forward(Duration::from_secs(5));
        assert_eq!(ttl_cache.get("key1"), Some(b"value1".to_vec()));
    }

    #[test]
    fn ttl_wrapping_real_fifo_cache_gone_after_ttl() {
        let manual_clock = Rc::new(ManualClock::new());
        let clock: Rc<dyn Clock> = Rc::clone(&manual_clock) as Rc<dyn Clock>;

        let (fifo, store) = real_fifo_with_stateful_store("test-ttl-fifo", 3);
        let ttl_cache = TtlCache::new(fifo, Rc::clone(&clock), Duration::from_secs(10));

        ttl_cache.save("key1", b"value1".to_vec()).unwrap();

        // Past the TTL the get() reports expiry AND lazily deletes from the real
        // FIFO backend, so the framed entry is physically gone from the store.
        manual_clock.move_forward(Duration::from_secs(11));
        assert!(ttl_cache.get("key1").is_none());
        assert!(store.borrow().get("test-ttl-fifo-key1").is_none());
    }
}