use std::{
    sync::Arc,
    time::{Duration, SystemTime},
};

use arc_swap::ArcSwapOption;
use async_trait::async_trait;
use tokio::{runtime::Handle, sync::Notify, time::Instant};
use kameleoon_trace_logging::trace_method;

use crate::{
    data::{BaseData, Conversion, CustomData},
    datafile::{custom_data_info::CustomDataInfo, datafile::DataFile, datafile_manager::DataFileManager},
    info_log,
    logging::LogLevel,
    storage::visitor::Visitor,
    tracking::trackable::{Trackable, TrackingState},
    warn_log,
};

#[async_trait]
pub(crate) trait VisitorManager: Send + Sync {
    fn peek(&self, visitor_code: &str) -> Option<Arc<Visitor>>;
    fn get(&self, visitor_code: &str) -> Option<Arc<Visitor>>;
    fn get_or_create(&self, visitor_code: &str) -> Arc<Visitor>;
    fn add_data(
        &self,
        visitor_code: &str,
        data: impl Iterator<Item = BaseData>,
        options: AddDataOptions,
    ) -> Arc<Visitor>;
    fn start_purging(&self, expiration_period: Duration);

    // fn count(&self) -> usize;
    #[cfg(test)]
    fn enumerate_visitor_codes(&self, f: impl FnMut(&String) -> bool);
}

pub(crate) struct VisitorManagerImpl<D> {
    datafile_manager: Arc<D>,
    visitors: Arc<scc::HashMap<String, Arc<Visitor>>>,
    runtime_handle: Handle,
    stop_purging_notify: ArcSwapOption<Notify>,
}

impl<D: DataFileManager> VisitorManagerImpl<D> {
    #[trace_method(LogLevel::Debug, skip(retval, datafile_manager, runtime_handle))]
    pub fn new(datafile_manager: Arc<D>, runtime_handle: Handle) -> Self {
        return Self {
            datafile_manager,
            visitors: Arc::new(scc::HashMap::new()),
            runtime_handle,
            stop_purging_notify: ArcSwapOption::empty(),
        };
    }

    fn process_data_to_add(
        &self,
        visitor_code: &str,
        visitor: &Arc<Visitor>,
        data: impl Iterator<Item = BaseData>,
        track: bool,
    ) -> Vec<BaseData> {
        let datafile = self.datafile_manager.datafile();
        let cdi = datafile.custom_data_info();
        data.filter_map(|data| {
            let data = match data {
                BaseData::CustomData(cd) => {
                    BaseData::CustomData(self.process_custom_data(cd, cdi, visitor_code, visitor)?)
                }
                BaseData::Conversion(conversion) => BaseData::Conversion(Self::process_conversion(conversion, cdi)),
                _ => data,
            };
            if !track {
                if let Some(trackable_base) = data.trackable_base() {
                    trackable_base.state_cell().change_state(TrackingState::Sent);
                }
            }
            Some(data)
        })
        .collect()
    }

    fn process_custom_data(
        &self,
        custom_data: CustomData,
        cdi: &impl CustomDataInfo,
        visitor_code: &str,
        visitor: &Arc<Visitor>,
    ) -> Option<CustomData> {
        let mut custom_data = Self::try_map_custom_data_index_by_name(custom_data, cdi)?;
        // We shouldn't send custom data with local only type
        if cdi.is_local_only(custom_data.index) {
            custom_data.state_cell().change_state(TrackingState::Sent);
        }
        // If mappingIdentifier is passed, we should link anonymous visitor with real unique userId.
        // After authorization, customer must be able to continue work with userId, but hash for variation
        // should be calculated based on anonymous visitor code, that's why set MappingIdentifier to visitor.
        if (cdi.mapping_identifier_index() == Some(custom_data.index))
            && custom_data.values.first().map_or(false, |v| !v.is_empty())
        {
            visitor.wdata().set_mapping_identifier(visitor_code);
            let user_id = custom_data.values[0].as_str();
            if visitor_code != user_id {
                self.visitors.upsert_sync(user_id.to_owned(), Arc::new(visitor.as_ref().clone()));
            }
            info_log!("Linked anonymous visitor '{visitor_code}' with user '{user_id}'");
            custom_data = custom_data.into_mapping_identifier();
        }
        Some(custom_data)
    }

    fn process_conversion(mut conversion: Conversion, cdi: &impl CustomDataInfo) -> Conversion {
        conversion.metadata = conversion
            .metadata
            .into_iter()
            .filter_map(|metadata| Self::try_map_custom_data_index_by_name(metadata, cdi))
            .collect();
        conversion
    }

    fn try_map_custom_data_index_by_name(custom_data: CustomData, cdi: &impl CustomDataInfo) -> Option<CustomData> {
        let Some(name) = custom_data.name.as_deref() else {
            return Some(custom_data); // No name -> pass through unchanged
        };
        let Some(index) = cdi.get_custom_data_index_by_name(name) else {
            warn_log!("{custom_data:?} is invalid and will be ignored");
            return None;
        };
        Some(custom_data.named_to_indexed(index))
    }

    fn purge(visitors: &Arc<scc::HashMap<String, Arc<Visitor>>>, expiration_period: Duration) {
        let expired_time = SystemTime::now() - expiration_period;
        visitors.retain_sync(|_, v| v.rdata().last_activity_time() >= expired_time);
    }
}

impl<D> Drop for VisitorManagerImpl<D> {
    //~ to cover drop with tests
    #[trace_method(LogLevel::Debug, skip(self))]
    fn drop(&mut self) {
        if let Some(notify) = self.stop_purging_notify.swap(None) {
            notify.notify_one();
        }
    }
}

#[async_trait]
impl<D: DataFileManager> VisitorManager for VisitorManagerImpl<D> {
    #[trace_method(LogLevel::Debug, skip(self, retval))]
    fn peek(&self, visitor_code: &str) -> Option<Arc<Visitor>> {
        self.visitors.read_sync(visitor_code, |_, v| v.clone())
    }

    #[trace_method(LogLevel::Debug, skip(self, retval))]
    fn get(&self, visitor_code: &str) -> Option<Arc<Visitor>> {
        self.visitors.read_sync(visitor_code, |_, v| {
            v.update_last_activity_time();
            v.clone()
        })
    }

    #[trace_method(LogLevel::Debug, skip(self, retval))]
    fn get_or_create(&self, visitor_code: &str) -> Arc<Visitor> {
        match self.get(visitor_code) {
            Some(v) => v,
            None => {
                self.visitors.entry_sync(visitor_code.to_owned()).or_insert_with(|| Arc::new(Visitor::new())).clone()
            }
        }
    }

    #[trace_method(LogLevel::Debug, skip(self, retval, data))]
    fn add_data(
        &self,
        visitor_code: &str,
        data: impl Iterator<Item = BaseData>,
        options: AddDataOptions,
    ) -> Arc<Visitor> {
        let visitor = self.get_or_create(visitor_code);
        // `process_data_to_add` may call `visitor.wdata()` (e.g. mappingIdentifier).
        // `visitor.add_data` holds that same write lock while consuming the iterator,
        // so we must eagerly materialize the transformed data to avoid a self-deadlock.
        // The logic could be refactored later
        let data = self.process_data_to_add(visitor_code, &visitor, data, options.track);
        visitor.add_data(data.into_iter(), options.overwrite);
        return visitor.clone();
    }

    fn start_purging(&self, expiration_period: Duration) {
        let notify = Arc::new(Notify::new());
        self.stop_purging_notify.store(Some(notify.clone()));
        let visitors = Arc::downgrade(&self.visitors);
        self.runtime_handle.spawn(async move {
            let mut interval = tokio::time::interval_at(Instant::now() + expiration_period, expiration_period);
            loop {
                tokio::select! {
                    _ = notify.notified() => break,
                    _ = interval.tick() => match visitors.upgrade() {
                        Some(visitors) => Self::purge(&visitors, expiration_period),
                        None => break,
                    }
                }
            }
        });
    }

    // fn count(&self) -> usize {
    //     return self.visitors.len();
    // }

    #[cfg(test)]
    fn enumerate_visitor_codes(&self, mut f: impl FnMut(&String) -> bool) {
        self.visitors.iter_sync(|vc, _| f(vc));
    }
}

#[derive(Debug, Clone)]
pub struct AddDataOptions {
    pub overwrite: bool,
    pub track: bool,
}

#[cfg(test)]
mod tests {
    use std::{collections::HashSet, thread::sleep};

    use crate::{
        data::{Conversion, ConversionOpts, CustomData},
        datafile::{
            datafile::DataFileImpl, datafile_manager::MockDataFileManager, datafile_test_utils::make_cdi_datafile_dto,
        },
        network::dto::CustomDataDto,
    };

    use super::*;
    use arc_swap::Guard;
    use rstest::rstest;

    fn test_runtime() -> tokio::runtime::Runtime {
        tokio::runtime::Runtime::new().unwrap()
    }

    fn checkpoint_datafile_manager_mock(mut vm: VisitorManagerImpl<MockDataFileManager>) {
        let dfm = std::mem::replace(&mut vm.datafile_manager, Arc::new(MockDataFileManager::new()));
        let mut dfm = Arc::into_inner(dfm).unwrap();
        dfm.checkpoint();
    }

    #[rstest]
    fn test_new() {
        let runtime = test_runtime();
        let dfm = Arc::new(MockDataFileManager::new());
        let vm = VisitorManagerImpl::new(dfm, runtime.handle().clone());
        assert!(vm.visitors.is_empty());
    }

    #[rstest]
    fn test_enumerate_visitor_codes() {
        let runtime = test_runtime();
        // arrange
        let dfm = Arc::new(MockDataFileManager::new());
        let vm = VisitorManagerImpl::new(dfm, runtime.handle().clone());
        let visitor_codes = ["vc1", "vc2", "vc3"];
        for vc in visitor_codes.iter() {
            assert!(vm.visitors.insert_sync((*vc).to_owned(), Arc::new(Visitor::new())).is_ok());
        }
        // act-assert
        let mut actual_visitor_codes: HashSet<String> = HashSet::new();
        vm.enumerate_visitor_codes(|vc| actual_visitor_codes.insert(vc.clone()));
        assert_eq!(visitor_codes.len(), actual_visitor_codes.len());
        for vc in visitor_codes {
            assert!(actual_visitor_codes.remove(vc));
        }
        assert!(actual_visitor_codes.is_empty());
    }

    #[test]
    fn test_peek() {
        let runtime = test_runtime();
        let dfm = Arc::new(MockDataFileManager::new());
        let vm = VisitorManagerImpl::new(dfm, runtime.handle().clone());
        const VC1: &str = "vc1";
        let v1 = Arc::new(Visitor::new());
        const VC2: &str = "vc2";
        let v2 = Arc::new(Visitor::new());
        let v2_activity_time0 = v2.rdata().last_activity_time();
        const VC3: &str = "vc3";
        assert!(vm.visitors.insert_sync(VC1.to_owned(), v1.clone()).is_ok());
        assert!(vm.visitors.insert_sync(VC2.to_owned(), v2.clone()).is_ok());
        // present visitor
        sleep(Duration::from_millis(1));
        assert!(Arc::ptr_eq(&v2, &vm.peek(VC2).unwrap()));
        assert_eq!(v2_activity_time0, v2.rdata().last_activity_time());
        // absent visitor
        assert!(vm.peek(VC3).is_none());
    }

    #[test]
    fn test_get() {
        let runtime = test_runtime();
        let dfm = Arc::new(MockDataFileManager::new());
        let vm = VisitorManagerImpl::new(dfm, runtime.handle().clone());
        const VC1: &str = "vc1";
        let v1 = Arc::new(Visitor::new());
        const VC2: &str = "vc2";
        let v2 = Arc::new(Visitor::new());
        let v2_activity_time0 = v2.rdata().last_activity_time();
        const VC3: &str = "vc3";
        assert!(vm.visitors.insert_sync(VC1.to_owned(), v1.clone()).is_ok());
        assert!(vm.visitors.insert_sync(VC2.to_owned(), v2.clone()).is_ok());
        // present visitor
        sleep(Duration::from_millis(1));
        assert!(Arc::ptr_eq(&v2, &vm.get(VC2).unwrap()));
        assert!(v2_activity_time0 < v2.rdata().last_activity_time());
        // absent visitor
        assert!(vm.get(VC3).is_none());
    }

    #[test]
    fn test_get_or_create() {
        let runtime = test_runtime();
        let dfm = Arc::new(MockDataFileManager::new());
        let vm = VisitorManagerImpl::new(dfm, runtime.handle().clone());
        const VC1: &str = "vc1";
        let v1 = Arc::new(Visitor::new());
        const VC2: &str = "vc2";
        let v2 = Arc::new(Visitor::new());
        const VC3: &str = "vc3";
        assert!(vm.visitors.insert_sync(VC1.to_owned(), v1.clone()).is_ok());
        assert!(vm.visitors.insert_sync(VC2.to_owned(), v2.clone()).is_ok());
        // present visitor
        assert!(Arc::ptr_eq(&v2, &vm.get_or_create(VC2)));
        assert_eq!(2, vm.visitors.len());
        // absent visitor
        let v3 = vm.get_or_create(VC3);
        assert!(!Arc::ptr_eq(&v1, &v3));
        assert!(!Arc::ptr_eq(&v2, &v3));
        assert_eq!(3, vm.visitors.len());
    }

    #[test]
    fn test_add_data() {
        let runtime = test_runtime();
        let mut dfm = MockDataFileManager::new();
        dfm.expect_datafile().returning(|| Guard::from_inner(Arc::new(DataFileImpl::new_nullum(None))));
        let dfm = Arc::new(dfm);
        let vm = VisitorManagerImpl::new(dfm, runtime.handle().clone());
        const VC1: &str = "vc1";
        let v1 = Arc::new(Visitor::new());
        const VC2: &str = "vc2";
        let v2 = Arc::new(Visitor::new());
        const VC3: &str = "vc3";
        assert!(vm.visitors.insert_sync(VC1.to_owned(), v1.clone()).is_ok());
        assert!(vm.visitors.insert_sync(VC2.to_owned(), v2.clone()).is_ok());
        let data = CustomData::new_with_index(10, vec!["v1".to_owned()]);
        // present visitor
        assert!(Arc::ptr_eq(
            &v2,
            &vm.add_data(
                VC2,
                std::iter::once(BaseData::CustomData(data.clone())),
                AddDataOptions {
                    overwrite: true,
                    track: true
                }
            )
        ));
        let v2_data = v2.rdata();
        let mut v2_cd_iter = v2_data.custom_data_map.r().iter();
        assert_eq!(Some(&data), v2_cd_iter.next());
        assert_eq!(None, v2_cd_iter.next());
        // absent visitor
        let v3 = vm.add_data(
            VC3,
            std::iter::once(BaseData::CustomData(data.clone())),
            AddDataOptions {
                overwrite: true,
                track: true,
            },
        );
        assert!(!Arc::ptr_eq(&v1, &v3));
        assert!(!Arc::ptr_eq(&v2, &v3));
        let v3_data = v3.rdata();
        let mut v3_cd_iter = v3_data.custom_data_map.r().iter();
        assert_eq!(Some(&data), v3_cd_iter.next());
        assert_eq!(None, v3_cd_iter.next());
    }

    #[rstest]
    fn test_add_data_custom_data_local_only(#[values(false, true)] local_only: bool) {
        let runtime = test_runtime();
        // arrange
        const VISITOR_CODE: &str = "vc1";
        const CD_VAL: &str = "val";
        const INDEX: u32 = 123;
        let mut dfm = MockDataFileManager::new();
        dfm.expect_datafile().once().returning(move || {
            let cdi = vec![CustomDataDto {
                id: 1234,
                index: INDEX,
                name: "".to_owned(),
                local_only,
                is_mapping_identifier: false,
                scope: "".to_owned(),
            }];
            Guard::from_inner(Arc::new(DataFileImpl::new(make_cdi_datafile_dto(cdi), None, None)))
        });
        let vm = VisitorManagerImpl::new(Arc::new(dfm), runtime.handle().clone());
        let cd = BaseData::CustomData(CustomData::new_with_index(INDEX, vec![CD_VAL.to_owned()]));
        // act
        let visitor = vm.add_data(
            VISITOR_CODE,
            std::iter::once(cd),
            AddDataOptions {
                overwrite: true,
                track: true,
            },
        );
        // assert
        let visitor_data = visitor.rdata();
        assert_eq!(1, visitor_data.custom_data_map.r().len());
        let expected_cd = CustomData::new_with_index(INDEX, vec![CD_VAL.to_owned()]);
        if local_only {
            expected_cd.state_cell().change_state(TrackingState::Sent);
        }
        assert_eq!(Some(&expected_cd), visitor_data.custom_data_map.r().get(&INDEX));
        checkpoint_datafile_manager_mock(vm);
    }

    #[test]
    fn test_add_data_custom_data_name_mapping() {
        let runtime = test_runtime();
        // arrange
        const VISITOR_CODE: &str = "vc_named";
        const CD_NAME: &str = "status";
        const CD_VAL: &str = "active";
        const INDEX: u32 = 123;
        let mut dfm = MockDataFileManager::new();
        dfm.expect_datafile().once().returning(move || {
            let cdi = vec![CustomDataDto {
                id: 1234,
                index: INDEX,
                name: CD_NAME.to_owned(),
                local_only: false,
                is_mapping_identifier: false,
                scope: "".to_owned(),
            }];
            Guard::from_inner(Arc::new(DataFileImpl::new(make_cdi_datafile_dto(cdi), None, None)))
        });
        let vm = VisitorManagerImpl::new(Arc::new(dfm), runtime.handle().clone());
        let cd = BaseData::CustomData(CustomData::new_with_name(CD_NAME.to_owned(), vec![CD_VAL.to_owned()]));
        // act
        let visitor = vm.add_data(
            VISITOR_CODE,
            std::iter::once(cd),
            AddDataOptions {
                overwrite: true,
                track: true,
            },
        );
        // assert
        let visitor_data = visitor.rdata();
        assert_eq!(1, visitor_data.custom_data_map.r().len());
        let expected_cd =
            CustomData::new_with_name(CD_NAME.to_owned(), vec![CD_VAL.to_owned()]).named_to_indexed(INDEX);
        assert_eq!(Some(&expected_cd), visitor_data.custom_data_map.r().get(&INDEX));
        checkpoint_datafile_manager_mock(vm);
    }

    #[test]
    fn test_add_data_invalid_named_custom_data_is_ignored() {
        let runtime = test_runtime();
        // arrange
        const VISITOR_CODE: &str = "vc_invalid_named";
        let mut dfm = MockDataFileManager::new();
        dfm.expect_datafile()
            .once()
            .returning(|| Guard::from_inner(Arc::new(DataFileImpl::new(make_cdi_datafile_dto(vec![]), None, None))));
        let vm = VisitorManagerImpl::new(Arc::new(dfm), runtime.handle().clone());
        let cd = BaseData::CustomData(CustomData::new_with_name("missing".to_owned(), vec!["v1".to_owned()]));
        // act
        let visitor = vm.add_data(
            VISITOR_CODE,
            std::iter::once(cd),
            AddDataOptions {
                overwrite: true,
                track: true,
            },
        );
        // assert
        assert_eq!(0, visitor.rdata().custom_data_map.r().len());
        checkpoint_datafile_manager_mock(vm);
    }

    #[rstest]
    fn test_add_data_custom_data_mapping_identifier(#[values(false, true)] is_mapping_identifier: bool) {
        let runtime = test_runtime();
        // arrange
        const VISITOR_CODE: &str = "vc1";
        const USER_ID: &str = "uid1";
        const INDEX: u32 = 123;
        let mut dfm = MockDataFileManager::new();
        dfm.expect_datafile().once().returning(move || {
            let cdi = vec![CustomDataDto {
                id: 1234,
                index: INDEX,
                name: "".to_owned(),
                local_only: false,
                is_mapping_identifier,
                scope: "".to_owned(),
            }];
            Guard::from_inner(Arc::new(DataFileImpl::new(make_cdi_datafile_dto(cdi), None, None)))
        });
        let vm = VisitorManagerImpl::new(Arc::new(dfm), runtime.handle().clone());
        let cd = BaseData::CustomData(CustomData::new_with_index(INDEX, vec![USER_ID.to_owned()]));
        // act
        let visitor = vm.add_data(
            VISITOR_CODE,
            std::iter::once(cd),
            AddDataOptions {
                overwrite: true,
                track: true,
            },
        );
        // assert
        let visitor_data = visitor.rdata();
        assert_eq!(1, visitor_data.custom_data_map.r().len());
        let expected_cd = CustomData::new_with_index(INDEX, vec![USER_ID.to_owned()]);
        let expected_cd = if is_mapping_identifier { expected_cd.into_mapping_identifier() } else { expected_cd };
        assert_eq!(Some(&expected_cd), visitor_data.custom_data_map.r().get(&INDEX));
        assert_eq!(if is_mapping_identifier { Some(VISITOR_CODE) } else { None }, visitor_data.mapping_identifier());
        assert_eq!(if is_mapping_identifier { 2 } else { 1 }, vm.visitors.len());
        if is_mapping_identifier {
            let reg_visitor = vm.get(USER_ID).unwrap();
            assert!(Arc::ptr_eq(visitor.data(), reg_visitor.data()));
        }
        checkpoint_datafile_manager_mock(vm);
    }

    #[test]
    fn test_add_data_conversion_metadata_name_mapping() {
        let runtime = test_runtime();
        // arrange
        const VISITOR_CODE: &str = "vc_conversion";
        const CD_NAME: &str = "status";
        const INDEX: u32 = 123;
        let mut dfm = MockDataFileManager::new();
        dfm.expect_datafile().once().returning(move || {
            let cdi = vec![CustomDataDto {
                id: 1234,
                index: INDEX,
                name: CD_NAME.to_owned(),
                local_only: false,
                is_mapping_identifier: false,
                scope: "".to_owned(),
            }];
            Guard::from_inner(Arc::new(DataFileImpl::new(make_cdi_datafile_dto(cdi), None, None)))
        });
        let vm = VisitorManagerImpl::new(Arc::new(dfm), runtime.handle().clone());
        let conversion = BaseData::Conversion(Conversion::new_with_opts(
            1,
            ConversionOpts::new().metadata(vec![
                CustomData::new_with_name(CD_NAME.to_owned(), vec!["mapped".to_owned()]),
                CustomData::new_with_name("missing".to_owned(), vec!["ignored".to_owned()]),
            ]),
        ));
        // act
        let visitor = vm.add_data(
            VISITOR_CODE,
            std::iter::once(conversion),
            AddDataOptions {
                overwrite: true,
                track: true,
            },
        );
        // assert
        let conversions = &visitor.rdata().conversions;
        assert_eq!(1, conversions.len());
        assert_eq!(
            vec![CustomData::new_with_name(CD_NAME.to_owned(), vec!["mapped".to_owned()]).named_to_indexed(INDEX)],
            conversions.iter().next().unwrap().metadata
        );
        checkpoint_datafile_manager_mock(vm);
    }

    #[rstest]
    fn test_add_data_marks_trackable(#[values(false, true)] track: bool) {
        let runtime = test_runtime();
        use crate::{
            data::{
                Browser, BrowserKind, Conversion, Device, DeviceKind, OperatingSystem, OperatingSystemKind, PageView,
                assigned_variation::AssignedVariation, geolocation::GeolocationBuilder, visitor_visits::VisitorVisits,
            },
            datafile::rule::RuleType,
        };

        // arrange
        const VISITOR_CODE: &str = "vc_track";
        const CD_INDEX: u32 = 10;
        let mut dfm = MockDataFileManager::new();
        dfm.expect_datafile()
            .once()
            .returning(|| Guard::from_inner(Arc::new(DataFileImpl::new(make_cdi_datafile_dto(vec![]), None, None))));
        let vm = VisitorManagerImpl::new(Arc::new(dfm), runtime.handle().clone());

        let data = vec![
            BaseData::Device(Device::new(DeviceKind::Desktop)),
            BaseData::Browser(Browser::new(BrowserKind::Chrome, None)),
            BaseData::OperatingSystem(OperatingSystem::new(OperatingSystemKind::Linux)),
            BaseData::Geolocation(GeolocationBuilder::default().country("co").build().unwrap()),
            BaseData::VisitorVisits(VisitorVisits::new(6, vec![], SystemTime::UNIX_EPOCH, Duration::ZERO)),
            BaseData::CustomData(CustomData::new_with_index(CD_INDEX, vec!["v1".to_owned()])),
            BaseData::PageView(PageView::new_with_url("url1")),
            BaseData::AssignedVariation(AssignedVariation::new(1, 10, RuleType::Unknown)),
            BaseData::Conversion(Conversion::new(1)),
        ];

        // act
        let visitor = vm.add_data(VISITOR_CODE, data.into_iter(), AddDataOptions { overwrite: true, track });

        // assert
        let vdata = visitor.rdata();
        let expected_state = if track { TrackingState::Unsent } else { TrackingState::Sent };

        let trackable_states: &[(&dyn Fn() -> TrackingState, &str)] = &[
            (&|| vdata.device.r().get().unwrap().state_cell().state(), "device"),
            (&|| vdata.browser.r().get().unwrap().state_cell().state(), "browser"),
            (&|| vdata.operating_system.r().get().unwrap().state_cell().state(), "operating_system"),
            (&|| vdata.geolocation.r().get().unwrap().as_ref().state_cell().state(), "geolocation"),
            (&|| vdata.visitor_visits.r().get().unwrap().as_ref().state_cell().state(), "visitor_visits"),
            (&|| vdata.custom_data_map.r().get(&CD_INDEX).unwrap().state_cell().state(), "custom_data"),
            (&|| vdata.page_view_visits.r().iter().next().unwrap().page_view().state_cell().state(), "page_view"),
            (&|| vdata.assigned_variations.r().iter().next().unwrap().state_cell().state(), "assigned_variation"),
            (&|| vdata.conversions.iter().next().unwrap().state_cell().state(), "conversion"),
        ];

        for (get_state, label) in trackable_states {
            assert_eq!(expected_state, get_state(), "{label} state mismatch");
        }

        checkpoint_datafile_manager_mock(vm);
    }

    #[rstest]
    fn test_purge() {
        let runtime = test_runtime();
        // arrange
        const K: f32 = 1.0;
        let period = Duration::from_millis(100).mul_f32(K);
        let s = Duration::from_millis(25).mul_f32(K);
        const VC_TO_REMOVE: &str = "vc2r";
        const VC_TO_KEEP: &str = "vc2k";
        let dfm = Arc::new(MockDataFileManager::new());
        let vm = VisitorManagerImpl::new(dfm, runtime.handle().clone());
        vm.get_or_create(VC_TO_REMOVE);
        let v_to_keep = vm.get_or_create(VC_TO_KEEP);
        // t = 0
        <VisitorManagerImpl<MockDataFileManager>>::purge(&vm.visitors, period);
        assert_eq!(2, vm.visitors.len());
        let start = SystemTime::now();
        while SystemTime::now().duration_since(start).unwrap() < period - s {}
        // t = 75k
        <VisitorManagerImpl<MockDataFileManager>>::purge(&vm.visitors, period);
        assert_eq!(2, vm.visitors.len());
        v_to_keep.rdata().update_last_activity_time();
        while SystemTime::now().duration_since(start).unwrap() < period + s {}
        // t = 125k
        <VisitorManagerImpl<MockDataFileManager>>::purge(&vm.visitors, period);
        assert_eq!(1, vm.visitors.len());
        assert!(Arc::ptr_eq(&v_to_keep, &vm.visitors.get_sync(VC_TO_KEEP).unwrap()));
        while SystemTime::now().duration_since(start).unwrap() < 2 * period {}
        // t = 200k
        <VisitorManagerImpl<MockDataFileManager>>::purge(&vm.visitors, period);
        assert!(vm.visitors.is_empty());
    }
}
