use crate::codec::CodecError;
use crate::migration::AppliedStep;
use crate::migration::set::MigrationSet;
use crate::store::builder::Backend;
use crate::store::error::{StorageError, StorageResult};
use crate::store::facts::{Facts, ValueBytes};
use crate::store::reading::{ReadResult, ReadValue};
use amethystate_core::path::{IntoStorePath, PathRef, StorePath, StorePathError};
use amethystate_core::primitives::error::WriteValue;
use std::fmt;
use std::path::{Path, PathBuf};
use crate::store::meta::{PrefixMeta, SchemaSnapshot};
use crate::store::{CodecFormat, Kv, StoreCallback, SubscriptionId, Writer};
use crate::{MigrationReport, Store, SubscriptionKind};
use error_stack::{Report, ResultExt};
use serde::{Serialize, de::DeserializeOwned};
use uuid::Uuid;
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub enum StoreLayout {
Single { data: PathBuf },
Sidecars {
data: PathBuf,
meta: PathBuf,
data_backup: PathBuf,
meta_backup: PathBuf,
},
InMemory,
PageStorage { under: StorePath },
}
impl StoreLayout {
pub fn of(path: impl AsRef<Path>, backend: Backend) -> Self {
let mut path = path.as_ref().to_path_buf();
if path.extension().is_none() {
path.set_extension(backend.extension());
}
match backend {
#[cfg(feature = "redb")]
Backend::Redb => Self::Single { data: path },
#[cfg(feature = "sqlite")]
Backend::Sqlite => Self::Single { data: path },
#[cfg(feature = "json")]
Backend::Json => Self::sidecars(path),
#[cfg(feature = "toml")]
Backend::Toml => Self::sidecars(path),
#[cfg(feature = "ron")]
Backend::Ron => Self::sidecars(path),
#[cfg(feature = "memory")]
Backend::Memory => Self::InMemory,
#[cfg(feature = "localstorage")]
Backend::LocalStorage => Self::PageStorage {
under: crate::store::backend::local_storage::root_of(&path),
},
}
}
pub fn rewrite_copy_of(file: &Path) -> PathBuf {
appended(file, "bak")
}
#[cfg_attr(
not(any(feature = "json", feature = "toml", feature = "ron")),
allow(dead_code)
)]
pub(crate) fn bookkeeping_of(data: &Path) -> PathBuf {
appended(data, "meta")
}
#[cfg_attr(
not(any(feature = "json", feature = "toml", feature = "ron")),
allow(dead_code)
)]
pub(crate) fn former_bookkeeping_of(data: &Path) -> PathBuf {
data.with_extension("meta")
}
#[cfg_attr(
not(any(feature = "json", feature = "toml", feature = "ron")),
allow(dead_code)
)]
pub(crate) fn set_aside_copy_of(file: &Path) -> PathBuf {
appended(file, "unreadable")
}
pub fn names(&self) -> Vec<PathBuf> {
match self {
Self::Single { data } => vec![data.clone()],
Self::Sidecars {
data,
meta,
data_backup,
meta_backup,
} => vec![
data.clone(),
meta.clone(),
data_backup.clone(),
meta_backup.clone(),
],
Self::InMemory | Self::PageStorage { .. } => Vec::new(),
}
}
pub fn present(&self) -> Vec<PathBuf> {
self.names()
.into_iter()
.filter(|file| file.exists())
.collect()
}
#[cfg_attr(
not(any(feature = "json", feature = "toml", feature = "ron")),
allow(dead_code)
)]
fn sidecars(data: PathBuf) -> Self {
let meta = Self::bookkeeping_of(&data);
Self::Sidecars {
data_backup: Self::rewrite_copy_of(&data),
meta_backup: Self::rewrite_copy_of(&meta),
data,
meta,
}
}
}
#[cfg(test)]
mod layout_tests {
use super::*;
#[test]
fn the_two_copies_of_a_sidecar_store_are_two_files() {
let layout = StoreLayout::sidecars(PathBuf::from("app/settings.db"));
let StoreLayout::Sidecars {
data,
meta,
data_backup,
meta_backup,
} = layout
else {
unreachable!("built as sidecars")
};
assert_eq!(data_backup, PathBuf::from("app/settings.db.bak"));
assert_eq!(meta_backup, PathBuf::from("app/settings.db.meta.bak"));
assert_ne!(
data_backup, meta_backup,
"a copy that keeps only the extension names one file for both, so the second \
lands on the first and the data is left with the metadata's bytes"
);
assert!(
![&data, &meta].contains(&&data_backup) && ![&data, &meta].contains(&&meta_backup),
"a copy took the name of a file the store writes"
);
}
#[test]
fn a_copy_keeps_the_whole_name_it_was_made_from() {
assert_eq!(
StoreLayout::rewrite_copy_of(Path::new("settings.conf")),
PathBuf::from("settings.conf.bak"),
"an extension the caller spelled is theirs, and the copy is beside it rather \
than over a neighbour's `settings.bak`"
);
}
}
fn appended(file: &Path, suffix: &str) -> PathBuf {
match file.file_name() {
Some(name) => {
let mut name = name.to_os_string();
name.push(".");
name.push(suffix);
file.with_file_name(name)
}
None => file.with_extension(suffix),
}
}
pub fn to_path(path: impl IntoStorePath) -> Result<StorePath, StorePathError> {
path.into_store_path()
}
pub(crate) fn entry_path(path: &StorePath, key: impl AsRef<str>) -> StorePath {
path.push(key.as_ref())
}
pub trait MigrationBackendAdapter {
fn format(&self) -> CodecFormat;
fn get(&self, key: &StorePath) -> StorageResult<Option<Vec<u8>>>;
fn set(&mut self, key: &StorePath, value: &[u8]) -> StorageResult<()>;
fn delete(&mut self, key: &StorePath) -> StorageResult<()>;
fn scan_prefix(&self, prefix: &StorePath) -> StorageResult<Vec<(StorePath, Vec<u8>)>>;
fn bookkeeping_is_lost(&self) -> bool {
false
}
fn holds_no_data(&self) -> StorageResult<bool> {
Ok(self.scan_prefix(&StorePath::root())?.is_empty())
}
fn delete_prefix(&mut self, prefix: &StorePath) -> StorageResult<()> {
let under = self.scan_prefix(prefix)?;
for (path, _) in under {
self.delete(&path)?;
}
self.delete(prefix)
}
fn get_meta(&self, prefix: &StorePath) -> StorageResult<Option<PrefixMeta>>;
fn set_meta(&mut self, prefix: &StorePath, meta: &PrefixMeta) -> StorageResult<()>;
fn get_schema_snapshots(&self, prefix: &StorePath) -> StorageResult<Vec<SchemaSnapshot>>;
fn set_schema_snapshots(
&mut self,
prefix: &StorePath,
trees: &[SchemaSnapshot],
) -> StorageResult<()>;
fn get_migration_log(&self, prefix: &StorePath) -> StorageResult<Option<Vec<AppliedStep>>>;
fn set_migration_log(&mut self, prefix: &StorePath, log: &[AppliedStep]) -> StorageResult<()>;
}
pub trait SchemaAwareStore: StoreBackend {
fn run_migrations(&self, mset: MigrationSet) -> StorageResult<MigrationReport>;
}
pub trait StoreBackend: Send + Sync + 'static {
fn get_raw(&self, path: &StorePath) -> StorageResult<Option<Vec<u8>>>;
fn set_erased(
&self,
path: &StorePath,
value: &dyn erased_serde::Serialize,
by: Writer,
) -> StorageResult<()>;
fn set_owned_erased(
&self,
path: StorePath,
value: &dyn erased_serde::Serialize,
by: Writer,
) -> StorageResult<()>;
fn get_erased(
&self,
path: &StorePath,
f: &mut dyn FnMut(&mut dyn erased_serde::Deserializer) -> StorageResult<()>,
) -> StorageResult<bool>;
fn decode_erased(
&self,
bytes: &[u8],
f: &mut dyn FnMut(&mut dyn erased_serde::Deserializer) -> StorageResult<()>,
) -> StorageResult<()>;
fn delete_with_source(&self, path: &StorePath, source: Option<Uuid>) -> StorageResult<()>;
fn delete(&self, path: &StorePath) -> StorageResult<()>;
fn delete_prefix_with_source(
&self,
prefix: &StorePath,
source: Option<Uuid>,
) -> StorageResult<()>;
fn delete_prefix(&self, prefix: &StorePath) -> StorageResult<()> {
self.delete_prefix_with_source(prefix, None)
}
fn scan_prefix(&self, prefix: &StorePath) -> StorageResult<Vec<(StorePath, Vec<u8>)>>;
fn visit_prefix(
&self,
prefix: &StorePath,
visit: &mut dyn FnMut(PathRef<'_>, &[u8]) -> StorageResult<()>,
) -> StorageResult<()> {
for (path, bytes) in self.scan_prefix(prefix)? {
visit(PathRef::from(&path), &bytes)?;
}
Ok(())
}
fn scan_keys(&self, prefix: &StorePath) -> StorageResult<Vec<StorePath>>;
fn parallel_reads(&self) -> bool {
false
}
fn files_layout(&self) -> Option<StoreLayout> {
None
}
fn persist_health(&self) -> Option<std::sync::Arc<crate::store::durable::PersistHealth>> {
None
}
#[cfg(feature = "test-utils")]
fn format_record(&self) -> Option<&dyn crate::store::format::TestFormatRecord> {
None
}
#[cfg(feature = "test-utils")]
fn reread_from_disk(&self) {}
fn save_now(&self) -> StorageResult<()>;
fn close(&self) -> StorageResult<()> {
self.save_now()
}
fn is_closed(&self) -> bool {
false
}
fn subscribe(&self, kind: SubscriptionKind, callback: StoreCallback) -> SubscriptionId;
fn unsubscribe(&self, id: SubscriptionId) -> bool;
fn flush_prefix(&self, prefix: &StorePath) -> StorageResult<()>;
fn flush_async(&self) -> crate::store::durable::Commit;
fn save_async(&self) -> crate::store::durable::Commit {
self.flush_async()
}
fn close_async(&self) -> crate::store::durable::Commit {
crate::store::durable::Commit::ready(self.close())
}
fn is_initialized(&self, namespace: &StorePath) -> StorageResult<bool>;
fn set_initialized(&self, namespace: &StorePath, state: InitState) -> StorageResult<()>;
fn record_schema(&self, at: &StorePath, schema: &SchemaSnapshot) -> StorageResult<()> {
let _ = (at, schema);
Ok(())
}
fn mark_initialized(&self, namespace: &StorePath) -> StorageResult<()> {
self.set_initialized(namespace, InitState::Seeded)
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum InitState {
Seeded,
Fresh,
}
impl InitState {
pub fn is_seeded(self) -> bool {
matches!(self, InitState::Seeded)
}
}
pub trait StoreExt: StoreBackend {
fn get<T: DeserializeOwned>(&self, path: impl IntoStorePath) -> ReadResult<Option<T>> {
let path = to_path(path)?;
let mut out = None;
let found = self
.get_erased(&path, &mut |d| {
out = Some(
erased_serde::deserialize::<T>(d)
.map_err(CodecError::from)
.change_context(StorageError::Codec)
.attach_key(&path)?,
);
Ok(())
})
.map_err(|why| ReadValue::from_store(&path, why))?;
Ok(if found { out } else { None })
}
fn set<T: Serialize>(&self, path: impl IntoStorePath, value: &T) -> Result<(), WriteValue> {
let path = to_path(path)?;
self.set_erased(&path, &value, Writer::default())
.map_err(|why| WriteValue::from_store(&path, why))
}
fn set_owned<T: Serialize>(&self, path: StorePath, value: &T) -> Result<(), WriteValue> {
self.set_owned_erased(path.clone(), &value, Writer::default())
.map_err(|why| WriteValue::from_store(&path, why))
}
fn set_with_source<T: Serialize>(
&self,
path: impl IntoStorePath,
value: &T,
source: Option<Uuid>,
) -> Result<(), WriteValue> {
let path = to_path(path)?;
self.set_erased(&path, &value, source.into())
.map_err(|why| WriteValue::from_store(&path, why))
}
fn set_owned_with_source<T: Serialize>(
&self,
path: StorePath,
value: &T,
source: Option<Uuid>,
) -> Result<(), WriteValue> {
self.set_owned_erased(path.clone(), &value, source.into())
.map_err(|why| WriteValue::from_store(&path, why))
}
fn decode<T: DeserializeOwned>(&self, bytes: &[u8]) -> StorageResult<T> {
let mut out = None;
self.decode_erased(bytes, &mut |d| {
out = Some(
erased_serde::deserialize::<T>(d)
.map_err(CodecError::from)
.change_context(StorageError::Codec)
.attach_with(|| format!("as: {}", std::any::type_name::<T>()))?,
);
Ok(())
})?;
out.ok_or_else(|| {
Report::new(StorageError::Codec)
.attach("the backend accepted the bytes without producing a value")
.attach(ValueBytes(bytes.len()))
})
}
fn decode_with<T>(&self, at: &StorePath, bytes: &[u8], read: ReadAs<T>) -> StorageResult<T> {
let mut out = None;
self.decode_erased(bytes, &mut |d| {
out = Some(
read(d)
.map_err(CodecError::from)
.change_context(StorageError::Codec)
.attach_with(|| format!("as: {}", std::any::type_name::<T>()))
.attach_key(at)?,
);
Ok(())
})?;
out.ok_or_else(|| {
Report::new(StorageError::Codec)
.attach("the backend accepted the bytes without producing a value")
.attach(ValueBytes(bytes.len()))
})
}
}
pub type ReadAs<T> =
for<'de> fn(&mut dyn erased_serde::Deserializer<'de>) -> Result<T, erased_serde::Error>;
pub type WriteAs<T> =
fn(&T, &mut dyn FnMut(&dyn erased_serde::Serialize) -> StorageResult<()>) -> StorageResult<()>;
pub struct StoredAs<T> {
pub write: Option<WriteAs<T>>,
pub read: Option<ReadAs<T>>,
}
impl<T> Default for StoredAs<T> {
fn default() -> Self {
Self {
write: None,
read: None,
}
}
}
impl<T> Clone for StoredAs<T> {
fn clone(&self) -> Self {
*self
}
}
impl<T> Copy for StoredAs<T> {}
impl<T> fmt::Debug for StoredAs<T> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("StoredAs")
.field("write", &self.write.is_some())
.field("read", &self.read.is_some())
.finish()
}
}
impl<S: StoreBackend + ?Sized> StoreExt for S {}
impl Store {
pub fn kv(&self) -> Kv {
Kv::new(self.clone())
}
}