use crate::data::Id; use crate::data::entities::{CardState, EntityData, EntityHandle}; use crate::data_provider::sqlite::SqliteConnection; 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, Index, IndexMut}; use std::sync::Arc; pub trait Repository where D: DbLayer, { fn get(&self, handle: &mut EntityHandle) -> Option; fn get_ref(&self, handle: &mut EntityHandle) -> Option<&T>; fn get_mut(&mut self, handle: &mut EntityHandle) -> Option<&mut T>; fn get_static(&self, handle: EntityHandle) -> Option; fn get_ref_static(&self, handle: EntityHandle) -> Option<&T>; fn get_mut_static(&mut self, handle: EntityHandle) -> Option<&mut T>; fn set(&mut self, handle: &mut EntityHandle, value: T); fn set_static(&mut self, handle: EntityHandle, value: T); fn handle_by_id(&self, id: Id) -> EntityHandle; fn create(&mut self, new: T) -> EntityHandle; fn create_now(&mut self, new: T) -> Result, NotificationData>; fn save(&mut self, entity: &mut EntityHandle) -> MaybeError; fn delete(&mut self, handle: EntityHandle) -> MaybeError; fn find(&self, predicate: F) -> Option> where F : Fn(&T) -> bool; fn init(&mut self) -> MaybeError; fn start(&mut self) -> MaybeError; fn reload(&mut self) -> MaybeError; fn shifts(&self) -> usize; fn iter<'a>(&'a self) -> impl Iterator where T: 'a; fn iter_mut<'a>(&'a mut self) -> impl Iterator where T: 'a; } struct QueryManger where R: Repository, D: DbLayer, { queries: Vec>>>, } impl QueryManger where R: Repository, D: DbLayer, T: EntityData, { pub(crate) fn handle_save(&self, old: EntityHandle, new: &mut EntityHandle, 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, 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 { 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) { for q in &self.queries { let mut query = q.lock(); query.cache.clear(); } } } #[derive(Clone)] pub struct Query where R: Repository, D: DbLayer, { cache: Vec>, rule: Arc bool + Send + Sync>, parallel: bool, _marker: PhantomData<(R, D)>, shift_stamp: usize, } impl, D: DbLayer> Query { pub fn iter<'a>(&mut self, data_repository_inner: &'a DataRepositoryInner) -> impl Iterator where T: 'a, { if self.shift_stamp != data_repository_inner.shifts() { self.load(data_repository_inner); } self.cache .iter_mut() .filter_map(move |e| data_repository_inner.get_ref(e)) } pub fn iter_with_handles<'a>(&mut self, data_repository_inner: &'a DataRepositoryInner) -> impl Iterator)> where T: 'a, { if self.shift_stamp != data_repository_inner.shifts() { self.load(data_repository_inner); } self.cache .iter_mut() .filter_map(move |e| Some((data_repository_inner.get_ref(e)?, e.clone()))) } pub fn for_each_mut(&mut self, repo: &mut DataRepositoryInner, mut f: F) where F: FnMut(&mut T), { if self.shift_stamp != repo.shifts() { self.load(repo); } for handle in &mut self.cache { if let Some(entity) = repo.get_mut(handle) { f(entity); } } } pub fn load(&mut self, data_repository: &DataRepositoryInner) where T: Send + Sync, { self.cache.clear(); if self.parallel { self.cache = data_repository .cache .par_iter() .enumerate() .filter(|(_, entity)| self.rule.as_ref()(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::>(), ); } pub fn handles(&self) -> impl Iterator> { self.cache.iter().cloned() } pub fn len(&self) -> usize { self.cache.len() } } impl, D: DbLayer> Index for Query { type Output = EntityHandle; fn index(&self, index: usize) -> &Self::Output { &self.cache[index] } } impl, D: DbLayer> IndexMut for Query { fn index_mut(&mut self, index: usize) -> &mut Self::Output { &mut self.cache[index] } } pub trait BatchedRepository where D: BatchedDbLayer, T: EntityData, { fn create_now_many(&mut self, entities: &mut [T]) -> Result>, NotificationData>; fn save_many(&mut self, handles: &mut [EntityHandle]) -> MaybeError; fn delete_many(&mut self, handles: &mut [EntityHandle]) -> MaybeError; } pub trait DbLayer { 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, connection: &Connection) -> MaybeError; } pub trait BatchedDbLayer 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> { inner: Arc>>, } pub struct DataRepositoryInner> { cache: Vec, temporal: Vec, next_temporal: u32, connection: Arc>, data_layer: D, query_manger: QueryManger, version: u16, shifts: usize, } impl + Default> DataRepository { pub fn new(connection: Arc>) -> DataRepository { Self{ inner: Arc::new(RwLock::new(DataRepositoryInner{ cache: vec![], temporal: vec![], next_temporal: 0, connection, data_layer: Default::default(), query_manger: QueryManger { queries: vec![] }, version: 0, shifts: 0, })), } } } impl> Deref for DataRepository { type Target = Arc>>; fn deref(&self) -> &Self::Target { &self.inner } } impl> DataRepositoryInner { pub fn query_par( &self, rule: Arc bool + Send + Sync>, ) -> Query, D> where T: Send + Sync, { let mut cache: Vec> = 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::>(), ); Query { cache, rule, parallel: true, _marker: PhantomData, shift_stamp: self.shifts, } } pub fn query( &self, rule: Arc bool + Send + Sync>, ) -> Query, D> where T: Send + Sync, { let mut cache: Vec> = 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::>(), ); Query { cache, rule, parallel: false, _marker: PhantomData, shift_stamp: self.shifts, } } pub fn empty_query(&self) -> Query, D> { Query{ cache: vec![], rule: Arc::new(|_| false), parallel: false, _marker: Default::default(), shift_stamp: 0, } } pub fn empty_query_par(&self) -> Query, D> { Query{ cache: vec![], rule: Arc::new(|_| false), parallel: true, _marker: Default::default(), shift_stamp: 0, } } pub fn len(&self) -> usize { self.cache.len() + self.temporal.len() } } impl> Repository for DataRepositoryInner { fn get(&self, handle: &mut EntityHandle) -> Option { 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) -> 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) -> 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 get_static(&self, mut handle: EntityHandle) -> Option { self.get(&mut handle) } fn get_ref_static(&self, mut handle: EntityHandle) -> Option<&T> { self.get_ref(&mut handle) } fn get_mut_static(&mut self, mut handle: EntityHandle) -> Option<&mut T> { self.get_mut(&mut handle) } fn set(&mut self, handle: &mut EntityHandle, value: T) { let slice = if handle.real() { self.cache.as_mut_slice() } else { self.temporal.as_mut_slice() }; let index = fast_position_of(slice, handle).unwrap(); slice[index] = value; } fn set_static(&mut self, mut handle: EntityHandle, value: T) { self.set(&mut handle, value) } fn handle_by_id(&self, id: Id) -> EntityHandle { EntityHandle::new(id, id.0 as usize, true) } fn create(&mut self, mut new: T) -> EntityHandle { 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, NotificationData> { self.data_layer .add_entity(&mut new, &self.connection.lock())?; 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) -> MaybeError { let original_state = entity.clone(); if entity.real() { let data = self.get_ref(entity).unwrap(); self.data_layer .write_entity(data, &self.connection.lock())?; } 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.lock())?; 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) -> MaybeError { let vec = if handle.real() { self.data_layer .delete_entity(handle.id, &self.connection.lock())?; &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 find(&self, predicate: F) -> Option> where F: Fn(&T) -> bool { if let Some(index) = self.cache.iter().position(|entity| predicate(entity)){ let value = &self.cache[index]; return Some(EntityHandle::new(value.id(), index, true)) } if let Some(index) = self.temporal.iter().position(|entity| predicate(entity)){ let value = &self.temporal[index]; return Some(EntityHandle::new(value.id(), index, false)) } None } fn init(&mut self) -> MaybeError { self.data_layer .load_all(&mut self.cache, &self.connection.lock())?; self.version += 1; Ok(()) } fn start(&mut self) -> MaybeError { if self.version != 0 { return Ok(()); } self.init()?; Ok(()) } fn reload(&mut self) -> MaybeError { self.cache.clear(); self.temporal.clear(); self.init()?; Ok(()) } fn shifts(&self) -> usize { self.shifts } fn iter<'a>(&'a self) -> impl Iterator where T: 'a { self.cache.iter().chain(self.temporal.iter()) } fn iter_mut<'a>(&'a mut self) -> impl Iterator where T: 'a { self.cache.iter_mut().chain(self.temporal.iter_mut()) } } impl + DbLayer> BatchedRepository for DataRepositoryInner { fn create_now_many(&mut self, entities: &mut [T]) -> Result>, NotificationData> { self.data_layer.add_many_entities(entities, &mut self.connection.lock())?; let len = self.cache.len(); let handles = entities.iter().enumerate().map(|(index, entity)| {EntityHandle::new(entity.id(), index + len, true)}).collect(); self.cache.extend(entities.to_vec()); Ok(handles) } fn save_many(&mut self, handles: &mut [EntityHandle]) -> 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.lock())?; 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.lock())?; } self.shifts += 1; Ok(()) } fn delete_many(&mut self, handles: &mut [EntityHandle]) -> 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.lock())?; 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(handles: &mut [EntityHandle], getter: F) -> (Vec, Vec) where F: Fn(&EntityHandle) -> 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( storage: &[E], handle: &mut EntityHandle, ) -> Option { 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) }