jap_learn/src/data/repositories.rs

use crate::data::Id;
use crate::data::entities::{EntityData, EntityHandle};
use crate::helpers::into_result;
use crate::notification::{MaybeError, NotificationData};
use parking_lot::{Mutex, RwLock};
use rayon::iter::IndexedParallelIterator;
use rayon::iter::IntoParallelRefIterator;
use rayon::iter::ParallelIterator;
use rusqlite::Connection;
use std::marker::PhantomData;
use std::ops::Deref;
use std::sync::Arc;

pub trait Repository<T, D>
where
    D: DbLayer<T>,
{
    fn get(&self, handle: &mut EntityHandle<T>) -> Option<T>;
    fn get_ref(&self, handle: &mut EntityHandle<T>) -> Option<&T>;
    fn get_mut(&mut self, handle: &mut EntityHandle<T>) -> Option<&mut T>;
    fn handle_by_id(&self, id: Id) -> EntityHandle<T>;
    fn create(&mut self, new: T) -> EntityHandle<T>;
    fn create_now(&mut self, new: T) -> Result<EntityHandle<T>, NotificationData>;
    fn save(&mut self, entity: &mut EntityHandle<T>) -> MaybeError;
    fn delete(&mut self, handle: EntityHandle<T>) -> MaybeError;
    fn init(&mut self) -> MaybeError;
    fn reload(&mut self) -> MaybeError;
}

struct QueryManger<T, R, D>
where
    R: Repository<T, D>,
    D: DbLayer<T>,
{
    queries: Vec<Arc<Mutex<Query<T, R, D>>>>,
}

impl<T, R, D> QueryManger<T, R, D>
where
    R: Repository<T, D>,
    D: DbLayer<T>,
    T: EntityData,
{
    pub(crate) fn handle_save(&self, old: EntityHandle<T>, new: &mut EntityHandle<T>, data: &T) {
        return;
        if old.real() && new.real() {
            return;
        }

        let affected = self.get_affected_queries(data);
        for q in affected {
            let mut query = self.queries[q].lock();
            let index = query.cache.binary_search(&old).unwrap();
            query.cache.remove(index);
            let real_end = query.cache.iter().rposition(|r| r.real()).unwrap() + 1;
            query.cache.insert(real_end, new.clone());
        }
    }
    pub(crate) fn handle_delete(&self, handle: &EntityHandle<T>, data: &T) {
        return;
        let affected = self.get_affected_queries(data);
        for q in affected {
            let mut query = self.queries[q].lock();
            let slice = query.cache.as_slice();
            if let Ok(index) = slice.binary_search(&handle) {
                query.cache.remove(index);
            }
        }
    }

    fn get_affected_queries(&self, data: &T) -> Vec<usize> {
        self.queries
            .iter()
            .enumerate()
            .filter(|(_, q)| {
                let query = q.lock();
                let rule = &query.rule;
                return rule(data);

                false
            })
            .map(|(i, _)| i)
            .collect()
    }

    fn reload_queries(&mut self, data_repository: impl Repository<T, D>) {
        for q in &self.queries {
            let mut query = q.lock();
            query.cache.clear();
        }
    }
}
pub struct Query<T, R, D>
where
    R: Repository<T, D>,
    D: DbLayer<T>,
{
    cache: Vec<EntityHandle<T>>,
    rule: Box<dyn Fn(&T) -> bool + Send + Sync>,
    parallel: bool,
    _marker: PhantomData<(R, D)>,
    shift_stamp: u16,
}

impl<T: EntityData + Clone, R: Repository<T, D>, D: DbLayer<T>> Query<T, R, D> {
    pub fn iter<'a>(&mut self, data_repository_inner: &'a R) -> impl Iterator<Item = &'a T>
    where
        T: 'a,
    {
        self.cache
            .iter_mut()
            .filter_map(move |e| data_repository_inner.get_ref(e))
    }

    pub fn for_each_mut<F>(&mut self, repo: &mut R, mut f: F)
    where
        F: FnMut(&mut T),
    {
        for handle in &mut self.cache {
            if let Some(entity) = repo.get_mut(handle) {
                f(entity);
            }
        }
    }

    pub fn load(&mut self, data_repository: &DataRepositoryInner<T, D>)
    where
        T: Send + Sync,
    {
        self.cache.clear();
        if self.parallel {
            self.cache = data_repository
                .cache
                .par_iter()
                .enumerate()
                .filter(|(_, entity)| (self.rule)(entity))
                .map(|(i, entity)| EntityHandle::new(entity.id(), i, true))
                .collect();
        } else {
            self.cache = data_repository
                .cache
                .iter()
                .enumerate()
                .filter(|(_, entity)| (self.rule)(entity))
                .map(|(i, entity)| EntityHandle::new(entity.id(), i, true))
                .collect();
        }

        self.cache.extend(
            data_repository
                .temporal
                .iter()
                .enumerate()
                .filter(|(_, entity)| (self.rule)(entity))
                .map(|(i, entity)| EntityHandle::new(entity.id(), i, false))
                .collect::<Vec<_>>(),
        );
    }
}

pub trait BatchedRepository<T, D>
where
    D: BatchedDbLayer<T>,
    T: EntityData,
{
    fn save_many(&mut self, handles: &mut [EntityHandle<T>]) -> MaybeError;
    fn delete_many(&mut self, handles: &mut [EntityHandle<T>]) -> MaybeError;
}

pub trait DbLayer<T> {
    fn write_entity(&self, entity: &T, connection: &Connection) -> MaybeError;
    fn add_entity(&mut self, entity: &mut T, connection: &Connection) -> MaybeError;
    fn delete_entity(&mut self, id: Id, connection: &Connection) -> MaybeError;
    fn load_all(&self, storage: &mut Vec<T>, connection: &Connection) -> MaybeError;
}

pub trait BatchedDbLayer<T>
where
    T: EntityData,
{
    fn delete_many_entities(&mut self, ids: &[Id], connection: &mut Connection) -> MaybeError;
    fn add_many_entities(&mut self, entity: &mut [T], connection: &mut Connection) -> MaybeError;
}

#[derive(Clone)]
pub struct DataRepository<T: std::clone::Clone + EntityData, D: DbLayer<T>> {
    inner: Arc<RwLock<DataRepositoryInner<T, D>>>,
}

pub struct DataRepositoryInner<T: std::clone::Clone + EntityData, D: DbLayer<T>> {
    cache: Vec<T>,
    temporal: Vec<T>,
    next_temporal: u32,
    connection: Arc<RwLock<Connection>>,
    data_layer: D,
    query_manger: QueryManger<T, Self, D>,
    version: u16,
    shifts: u16,
}

impl<T: std::clone::Clone + EntityData, D: DbLayer<T>> Deref for DataRepository<T, D> {
    type Target = Arc<RwLock<DataRepositoryInner<T, D>>>;

    fn deref(&self) -> &Self::Target {
        &self.inner
    }
}

impl<T: EntityData + Clone, D: DbLayer<T>> DataRepositoryInner<T, D> {
    pub fn query_par(
        &self,
        rule: Box<dyn Fn(&T) -> bool + Send + Sync>,
    ) -> Query<T, DataRepositoryInner<T, D>, D>
    where
        T: Send + Sync,
    {
        let mut cache: Vec<EntityHandle<T>> = self
            .cache
            .par_iter()
            .enumerate()
            .filter(|(_, entity)| rule(entity))
            .map(|(i, entity)| EntityHandle::new(entity.id(), i, true))
            .collect();

        cache.extend(
            self.temporal
                .iter()
                .enumerate()
                .filter(|(_, entity)| rule(entity))
                .map(|(i, entity)| EntityHandle::new(entity.id(), i, false))
                .collect::<Vec<_>>(),
        );

        Query {
            cache,
            rule,
            parallel: true,
            _marker: PhantomData,
            shift_stamp: self.shifts,
        }
    }

    pub fn query(
        &self,
        rule: Box<dyn Fn(&T) -> bool + Send + Sync>,
    ) -> Query<T, DataRepositoryInner<T, D>, D>
    where
        T: Send + Sync,
    {
        let mut cache: Vec<EntityHandle<T>> = self
            .cache
            .iter()
            .enumerate()
            .filter(|(_, entity)| rule(entity))
            .map(|(i, entity)| EntityHandle::new(entity.id(), i, true))
            .collect();

        cache.extend(
            self.temporal
                .iter()
                .enumerate()
                .filter(|(_, entity)| rule(entity))
                .map(|(i, entity)| EntityHandle::new(entity.id(), i, false))
                .collect::<Vec<_>>(),
        );

        Query {
            cache,
            rule,
            parallel: false,
            _marker: PhantomData,
            shift_stamp: self.shifts,
        }
    }
}

impl<T: EntityData + Clone, D: DbLayer<T>> Repository<T, D> for DataRepositoryInner<T, D> {
    fn get(&self, handle: &mut EntityHandle<T>) -> Option<T> {
        let slice = if handle.real() {
            self.cache.as_slice()
        } else {
            self.temporal.as_slice()
        };
        let index = fast_position_of(slice, handle)?;
        Some(slice[index].clone())
    }

    fn get_ref(&self, handle: &mut EntityHandle<T>) -> Option<&T> {
        let slice = if handle.real() {
            self.cache.as_slice()
        } else {
            self.temporal.as_slice()
        };
        let index = fast_position_of(slice, handle)?;
        Some(&slice[index])
    }

    fn get_mut(&mut self, handle: &mut EntityHandle<T>) -> Option<&mut T> {
        let slice = if handle.real() {
            self.cache.as_mut_slice()
        } else {
            self.temporal.as_mut_slice()
        };
        let index = fast_position_of(slice, handle)?;
        Some(&mut slice[index])
    }

    fn handle_by_id(&self, id: Id) -> EntityHandle<T> {
        EntityHandle::new(id, id.0 as usize, true)
    }

    fn create(&mut self, mut new: T) -> EntityHandle<T> {
        let next_temporal = self.next_temporal;
        self.next_temporal += 1;
        let entity = EntityHandle::new(next_temporal.into(), self.temporal.len(), false);
        new.set_id(entity.id);
        self.temporal.push(new);
        entity
    }

    fn create_now(&mut self, mut new: T) -> Result<EntityHandle<T>, NotificationData> {
        self.data_layer.add_entity(&mut new, &self.connection.read())?;
        let index = self.cache.len();
        let handle = EntityHandle::new(new.id(), index, true);
        self.cache.push(new);
        Ok(handle)
    }

    fn save(&mut self, entity: &mut EntityHandle<T>) -> MaybeError {
        let original_state = entity.clone();
        if entity.real() {
            let data = self.get_ref(entity).unwrap();
            self.data_layer
                .write_entity(data, &self.connection.read())?;
        } else {
            let index = fast_position_of(self.temporal.as_slice(), entity).unwrap();
            let mut data = self.temporal.remove(index);
            self.data_layer
                .add_entity(&mut data, &self.connection.read())?;
            entity.set_real(true);
            entity.id = data.id();
            entity.set_index(self.cache.len());
            self.query_manger.handle_save(original_state, entity, &data);

            self.cache.push(data);
            self.shifts += 1;
        }

        Ok(())
    }

    fn delete(&mut self, mut handle: EntityHandle<T>) -> MaybeError {
        let vec = if handle.real() {
            self.data_layer
                .delete_entity(handle.id, &self.connection.read())?;
            &mut self.cache
        } else {
            &mut self.temporal
        };

        let index = into_result(
            fast_position_of(vec.as_slice(), &mut handle),
            NotificationData::DbSaveError,
        )?;
        let value = vec.remove(index);
        self.query_manger.handle_delete(&handle, &value);
        self.shifts += 1;
        Ok(())
    }

    fn init(&mut self) -> MaybeError {
        self.data_layer
            .load_all(&mut self.cache, &self.connection.read())?;
        self.version += 1;
        Ok(())
    }

    fn reload(&mut self) -> MaybeError {
        self.cache.clear();
        self.temporal.clear();
        self.init()?;
        Ok(())
    }
}
impl<T: EntityData + Clone, D: BatchedDbLayer<T> + DbLayer<T>> BatchedRepository<T, D>
    for DataRepositoryInner<T, D>
{
    fn save_many(&mut self, handles: &mut [EntityHandle<T>]) -> MaybeError {
        let mut temp = Vec::with_capacity(handles.len() / 2);
        let mut cached = Vec::with_capacity(handles.len() / 2);
        handles.sort_by_key(|a| a.id);
        handles.iter_mut().for_each(|handle| {
            if handle.real() {
                cached.push(handle)
            } else {
                temp.push(handle)
            }
        });

        let mut temp_values = Vec::with_capacity(temp.len());
        let mut cell = 0;
        let len = temp.len();
        self.temporal.retain(|x| {
            if cell == len {
                return true;
            }
            let find = temp[cell].id;
            let remove = x.id() == find;
            if remove {
                cell += 1;
                temp_values.push(x.clone());
            }
            !remove
        });

        self.data_layer
            .add_many_entities(&mut temp_values, &mut self.connection.write())?;

        let start = self.cache.len();

        for (i, (handle, data)) in temp.into_iter().zip(&mut temp_values).enumerate() {
            let old = handle.clone();
            handle.id = data.id();
            handle.set_index(start + i);
            handle.set_real(true);
            self.query_manger.handle_save(old, handle, data);
        }

        self.cache.extend(temp_values);

        for handle in cached.into_iter() {
            let h = handle;
            let idx = fast_position_of(self.cache.as_slice(), h).unwrap();
            self.data_layer
                .write_entity(&self.cache[idx], &self.connection.read())?;
        }
        self.shifts += 1;

        Ok(())
    }

    fn delete_many(&mut self, handles: &mut [EntityHandle<T>]) -> MaybeError {
        let (cached, temp) = split_temp_real(handles, |e| e.id);

        let mut cell = 0;
        let len = temp.len();
        self.temporal.retain(|x| {
            if cell == len {
                return true;
            }
            let find = temp[cell];
            let remove = x.id() == find;
            if remove {
                cell += 1;
            }
            !remove
        });
        self.data_layer
            .delete_many_entities(cached.as_slice(), &mut self.connection.write())?;
        let len = cached.len();
        cell = 0;
        self.cache.retain(|x| {
            if cell == len {
                return true;
            }
            let find = cached[cell];
            let remove = x.id() == find;
            if remove {
                cell += 1;
            }

            !remove
        });
        self.shifts += 1;

        Ok(())
    }
}

fn split_temp_real<T, F, V>(handles: &mut [EntityHandle<T>], getter: F) -> (Vec<V>, Vec<V>)
where
    F: Fn(&EntityHandle<T>) -> V,
{
    let mut temp = Vec::with_capacity(handles.len() / 2);
    let mut cached = Vec::with_capacity(handles.len() / 2);
    handles.sort_by(|a, b| a.id.cmp(&b.id));
    handles.iter().for_each(|handle| {
        if handle.real() {
            cached.push(getter(handle))
        } else {
            temp.push(getter(handle))
        }
    });
    (cached, temp)
}

pub fn fast_position_of<E: EntityData>(
    storage: &[E],
    handle: &mut EntityHandle<E>,
) -> Option<usize> {
    if storage.is_empty() {
        return None;
    }

    let max_idx = storage.len() - 1;
    let upper = (handle.index()).min(max_idx);
    let peek = storage[upper].id();

    if peek == handle.id {
        handle.set_index(upper);
        return Some(upper);
    }

    if peek.0 < handle.id.0 {
        return None;
    }

    let target_id_val = handle.id.0 as i64;
    let peek_val = peek.0 as i64;
    let delta = peek_val - target_id_val;

    let new_index = (upper as i64 - delta).max(0) as usize;

    if storage[new_index].id() == handle.id {
        handle.set_index(new_index);
        return Some(new_index);
    }

    if new_index == upper {
        return None;
    }

    let index = storage[new_index..upper]
        .binary_search_by(|e| e.id().cmp(&handle.id))
        .ok()?
        + new_index;

    handle.set_index(index);
    Some(index)
}