amethystate 0.26.0

use crate::store::backend;

#[cfg(feature = "json")]
pub use backend::text::JsonStore;

#[cfg(feature = "sqlite")]
pub use backend::sqlite::SqliteStore;

#[cfg(feature = "redb")]
pub use backend::redb::RedbStore;

#[cfg(feature = "memory")]
pub use backend::memory::MemoryStore;

#[cfg(feature = "localstorage")]
pub use backend::local_storage::LocalStorageStore;

#[cfg(feature = "toml")]
pub use backend::text::TomlStore;

#[cfg(feature = "ron")]
pub use backend::text::RonStore;

use crate::MigrationReport;
use crate::migration::set::MigrationSet;
use crate::store::config::{PersistEvent, StoreConfig};
use crate::store::durable::{Commit, PersistWatch};
#[cfg(feature = "test-utils")]
use crate::store::format::TestFormatRecord;
use crate::store::meta::SchemaSnapshot;
use crate::store::places::Places;
use crate::store::traits::StoreLayout;
use crate::store::{
    Fallbacks, Flush, FlushResult, InitState, ReadResult, RuleContext, ScanKeys, ScanResult,
    StorageError, StorageResult, StoreBackend, StoreCallback, StoreExt, StoreSubscription,
    SubscriptionId, WriteValue, Writer, to_path,
};
use amethystate_core::path::{IntoStorePath, PathRef, StorePath};
use error_stack::Report;
use std::sync::Arc;

/// A handle on an open store.
///
/// One type over every engine - which one is behind it is settled by
/// [`StoreBuilder`](crate::StoreBuilder) when the store is opened, and nothing
/// downstream is generic over it.
///
/// Cheap to clone and shared by every clone: the file stays open as long as one
/// handle is alive, and closes when the last is dropped.
///
/// Dropping the last handle writes what is still buffered, but it is the one
/// write whose failure it cannot report: `Drop` has no caller to return an
/// error to, and by then there is rarely anyone left to tell. An application
/// that would rather find out calls [`Store::close`] on the way out and reads
/// the result.
///
/// Finding out is all it buys. A close that failed has closed the store
/// anyway, a second one answers `Ok` without flushing again, and the writes
/// are gone - a retry belongs to [`Store::save_now`], while the store is still
/// open.
#[derive(Clone)]
pub struct Store {
    backend: Arc<dyn StoreBackend>,
    places: Arc<Places>,
    context: Arc<RuleContext>,
    fallbacks: Fallbacks,
}

impl Store {
    /// Wraps a backend that was built by hand.
    ///
    /// [`StoreBuilder`](crate::StoreBuilder) is the ordinary way in; this is
    /// for a [`StoreBackend`] implemented outside the crate.
    pub fn from_arc(inner: Arc<dyn StoreBackend>) -> Self {
        Self {
            backend: inner,
            places: Arc::new(Places::default()),
            context: Arc::new(RuleContext::default()),
            fallbacks: Fallbacks::default(),
        }
    }

    /// What a field falls back to here when neither it nor its struct said,
    /// put there by [`StoreBuilder::rules`](crate::StoreBuilder::rules).
    pub fn fallbacks(&self) -> Fallbacks {
        self.fallbacks
    }

    pub(crate) fn with_fallbacks(mut self, fallbacks: Fallbacks) -> Self {
        self.fallbacks = fallbacks;
        self
    }

    /// What the application handed this store for its declared rules, put
    /// there by [`StoreBuilder::context`](crate::StoreBuilder::context).
    ///
    /// A store opened any other way carries an empty one, and a rule asking
    /// it for something is refused with what was on offer.
    pub fn context(&self) -> &RuleContext {
        &self.context
    }

    pub(crate) fn with_context(mut self, context: Arc<RuleContext>) -> Self {
        self.context = context;
        self
    }

    /// The erased backend underneath, for code that is generic over
    /// [`StoreBackend`] rather than over this handle.
    pub fn as_dyn(&self) -> &Arc<dyn StoreBackend> {
        &self.backend
    }

    /// The places that are spoken for and by whom, shared by every clone of
    /// this handle.
    pub fn places(&self) -> &Places {
        &self.places
    }

    /// Attaches an observer to this store's background saving, for as long as
    /// the returned watch is kept.
    ///
    /// It hears every failing streak that outlives the retry budget, and the
    /// save that lands after one. It decides nothing - writes keep landing in
    /// memory whatever the disk does - and any number can be attached, from
    /// anywhere that holds the store. It runs on the flush thread, where
    /// closing the store is refused; one that panics is logged and the store
    /// saves on.
    ///
    /// A backend that saves on its own terms has nothing to report, and its
    /// watch hears nothing.
    pub fn on_persist_failure(
        &self,
        observer: impl Fn(&PersistEvent<'_>) + Send + Sync + 'static,
    ) -> PersistWatch {
        match self.backend.persist_health() {
            Some(health) => health.observe(Arc::new(observer)),
            None => PersistWatch::detached(),
        }
    }

    /// The failure the last failing streak gave up with, until a save lands
    /// again.
    pub fn persist_failure(&self) -> Option<Arc<Report<StorageError>>> {
        self.backend.persist_health()?.last_failure()
    }

    /// Opens the store with the first engine this build has, and refuses in a
    /// build with none.
    pub fn open(config: StoreConfig, mset: MigrationSet) -> StorageResult<(Self, MigrationReport)> {
        crate::store::builder::built_in()
            .ok_or_else(crate::store::builder::no_engine_built_in)?
            .open(config, mset)
    }
}

impl std::fmt::Debug for Store {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str("Store")
    }
}

impl PartialEq for Store {
    fn eq(&self, other: &Self) -> bool {
        Arc::ptr_eq(&self.backend, &other.backend)
    }
}
impl Eq for Store {}

impl std::ops::Deref for Store {
    type Target = dyn StoreBackend;
    fn deref(&self) -> &Self::Target {
        &*self.backend
    }
}

/// The typed surface, inherent so a call site needs no trait in scope.
impl Store {
    /// Hears the changes `kind` names for as long as what comes back is held.
    ///
    /// Dropping the [`StoreSubscription`] is how it stops, and it can only
    /// stop its own: it carries the store it was made on.
    /// [`StoreBackend::subscribe`] is the same without the guard, for whoever
    /// keeps the id and ends it by hand.
    ///
    /// ```
    /// # use amethystate::{StoreBuilder, SubscriptionKind};
    /// # use std::sync::Arc;
    /// # use std::sync::atomic::{AtomicU32, Ordering};
    /// # let path = amethystate_core::test_utils::TempPath::new("doc");
    /// # let store = StoreBuilder::new(&*path).build().unwrap();
    /// let heard = Arc::new(AtomicU32::new(0));
    /// let counter = Arc::clone(&heard);
    /// let listening = store.subscribe(
    ///     SubscriptionKind::Any,
    ///     Arc::new(move |_| {
    ///         counter.fetch_add(1, Ordering::Relaxed);
    ///         Ok(())
    ///     }),
    /// );
    ///
    /// store.set(["ui", "width"], &1280u32).unwrap();
    /// drop(listening);
    /// store.set(["ui", "width"], &800u32).unwrap();
    ///
    /// assert_eq!(heard.load(Ordering::Relaxed), 1);
    /// ```
    #[must_use = "the subscription ends when this is dropped"]
    pub fn subscribe(
        &self,
        kind: crate::SubscriptionKind,
        callback: StoreCallback,
    ) -> StoreSubscription {
        let id = StoreBackend::subscribe(self, kind, callback);
        StoreSubscription::new(self.clone(), id)
    }

    /// Reads a value by path, or `None` if nothing is stored there.
    ///
    /// Sees buffered writes as well as committed ones. The type is whatever
    /// you ask for and nothing remembers it - [`Kv::cell`](crate::store::Kv::cell)
    /// is the variant that does.
    ///
    /// The levels are given as a list, and a string is not one of them - a name
    /// holding a separator is a name, and nesting is spelled out:
    ///
    /// ```
    /// # use amethystate::StoreBuilder;
    /// # let path = amethystate_core::test_utils::TempPath::new("doc");
    /// # let store = StoreBuilder::new(&*path).build().unwrap();
    /// store.set(["ui", "width"], &1280u32).unwrap();
    ///
    /// assert_eq!(store.get::<u32>(["ui", "width"]).unwrap(), Some(1280));
    /// assert_eq!(store.get::<u32>(["ui", "height"]).unwrap(), None);
    ///
    /// // One level called `ui.width`, which is somewhere else entirely.
    /// store.set(["ui.width"], &7u32).unwrap();
    /// assert_eq!(store.get::<u32>(["ui", "width"]).unwrap(), Some(1280));
    /// ```
    pub fn get<T: serde::de::DeserializeOwned>(
        &self,
        path: impl IntoStorePath,
    ) -> ReadResult<Option<T>> {
        StoreExt::get(self, path)
    }
    /// Writes a value at `path`, creating it or replacing what was there.
    ///
    /// The write lands in the buffer and notifies subscribers; the debouncer
    /// commits it later. Nothing here carries provenance, so a subscription
    /// cannot tell this apart from its own write - use
    /// [`Store::set_with_source`] when it must.
    pub fn set<T: serde::Serialize>(
        &self,
        path: impl IntoStorePath,
        value: &T,
    ) -> Result<(), WriteValue> {
        StoreExt::set(self, path, value)
    }
    /// [`Store::set`] for a path already owned, saving the clone the borrowed
    /// form would make.
    pub fn set_owned<T: serde::Serialize>(
        &self,
        path: StorePath,
        value: &T,
    ) -> Result<(), WriteValue> {
        StoreExt::set_owned(self, path, value)
    }
    /// [`Store::set`] tagged with who made the write.
    ///
    /// Subscribers receive the id, which is how a component ignores the echo
    /// of its own change instead of reacting to it.
    pub fn set_with_source<T: serde::Serialize>(
        &self,
        path: impl IntoStorePath,
        value: &T,
        source: Option<uuid::Uuid>,
    ) -> Result<(), WriteValue> {
        StoreExt::set_with_source(self, path, value, source)
    }
    /// [`Store::set_with_source`] for a path already owned.
    pub fn set_owned_with_source<T: serde::Serialize>(
        &self,
        path: StorePath,
        value: &T,
        source: Option<uuid::Uuid>,
    ) -> Result<(), WriteValue> {
        StoreExt::set_owned_with_source(self, path, value, source)
    }
    /// Removes whatever is at `path`. Removing an absent path is not an error.
    pub fn delete(&self, path: impl IntoStorePath) -> Result<(), WriteValue> {
        let path = to_path(path)?;
        StoreBackend::delete(self, &path).map_err(|why| WriteValue::from_store(&path, why))
    }

    /// Removes every level under `prefix`, and the value at `prefix` itself.
    pub fn delete_prefix(&self, prefix: impl IntoStorePath) -> Result<(), WriteValue> {
        let prefix = to_path(prefix)?;
        StoreBackend::delete_prefix(self, &prefix)
            .map_err(|why| WriteValue::from_store(&prefix, why))
    }

    /// Commits what is buffered under `prefix`.
    ///
    /// How much else goes with it is the engine's answer: redb and sqlite
    /// commit what is under the prefix and leave the rest buffered, a document
    /// engine rewrites the whole file and so commits everything. See
    /// [`Backend::a_commit_covers_the_whole_store`](crate::store::builder::Backend::a_commit_covers_the_whole_store).
    pub fn flush_prefix(&self, prefix: impl IntoStorePath) -> FlushResult<()> {
        let prefix = to_path(prefix).map_err(|why| {
            Flush::from_store(Report::new(why).change_context(StorageError::Path))
        })?;
        StoreBackend::flush_prefix(self, &prefix).map_err(Flush::from_store)
    }

    /// Writes everything buffered, and waits for it to land.
    pub fn save_now(&self) -> FlushResult<()> {
        StoreBackend::save_now(self).map_err(Flush::from_store)
    }

    /// Writes what is buffered, stops the background threads and lets go of
    /// the file.
    pub fn close(&self) -> FlushResult<()> {
        StoreBackend::close(self).map_err(Flush::from_store)
    }

    /// The keys under `prefix`, sorted, without reading their values.
    pub fn scan_keys(&self, prefix: impl IntoStorePath) -> ScanResult<Vec<StorePath>> {
        let prefix = to_path(prefix)?;
        StoreBackend::scan_keys(self, &prefix).map_err(|why| ScanKeys::from_store(&prefix, why))
    }

    /// Every key under `prefix` with its bytes, sorted by key.
    pub fn scan_prefix(&self, prefix: impl IntoStorePath) -> ScanResult<Vec<(StorePath, Vec<u8>)>> {
        let prefix = to_path(prefix)?;
        StoreBackend::scan_prefix(self, &prefix).map_err(|why| ScanKeys::from_store(&prefix, why))
    }

    /// Decodes bytes that arrived in a [`StoreEvent`](crate::StoreEvent),
    /// in whatever format this backend writes.
    ///
    /// Bytes that will not decode - because the file was edited by hand, or the
    /// field changed type - are an error naming the type that was asked for and
    /// what the codec made of them, so a caller can tell a stored default from
    /// a value it could not read.
    pub fn decode<T: serde::de::DeserializeOwned>(&self, bytes: &[u8]) -> StorageResult<T> {
        StoreExt::decode(self, bytes)
    }
}

/// Addressing the store by path, with nothing in the way - and what that costs
/// is written on [`StoreBackend`] itself, which is worth reading before
/// reaching for any of it. The short of it: a write here lands wherever it is
/// aimed, declared paths included, and
/// [`Kv`](crate::store::Kv) is the surface that refuses to.
///
/// [`Store::get`] and [`Store::set`] are the same operations under a shorter
/// name, so they carry the same warning.
impl StoreBackend for Store {
    fn get_raw(&self, path: &StorePath) -> StorageResult<Option<Vec<u8>>> {
        self.backend.get_raw(path)
    }
    fn set_erased(
        &self,
        path: &StorePath,
        value: &dyn erased_serde::Serialize,
        by: Writer,
    ) -> StorageResult<()> {
        self.backend.set_erased(path, value, by)
    }
    fn set_owned_erased(
        &self,
        path: StorePath,
        value: &dyn erased_serde::Serialize,
        by: Writer,
    ) -> StorageResult<()> {
        self.backend.set_owned_erased(path, value, by)
    }
    fn get_erased(
        &self,
        path: &StorePath,
        f: &mut dyn FnMut(&mut dyn erased_serde::Deserializer) -> StorageResult<()>,
    ) -> StorageResult<bool> {
        self.backend.get_erased(path, f)
    }
    fn decode_erased(
        &self,
        bytes: &[u8],
        f: &mut dyn FnMut(&mut dyn erased_serde::Deserializer) -> StorageResult<()>,
    ) -> StorageResult<()> {
        self.backend.decode_erased(bytes, f)
    }
    fn delete_with_source(
        &self,
        path: &StorePath,
        source: Option<uuid::Uuid>,
    ) -> StorageResult<()> {
        self.backend.delete_with_source(path, source)
    }
    fn delete(&self, path: &StorePath) -> StorageResult<()> {
        self.backend.delete(path)
    }
    fn delete_prefix_with_source(
        &self,
        prefix: &StorePath,
        source: Option<uuid::Uuid>,
    ) -> StorageResult<()> {
        self.backend.delete_prefix_with_source(prefix, source)
    }
    fn scan_prefix(&self, prefix: &StorePath) -> StorageResult<Vec<(StorePath, Vec<u8>)>> {
        self.backend.scan_prefix(prefix)
    }
    fn visit_prefix(
        &self,
        prefix: &StorePath,
        visit: &mut dyn FnMut(PathRef<'_>, &[u8]) -> StorageResult<()>,
    ) -> StorageResult<()> {
        self.backend.visit_prefix(prefix, visit)
    }
    fn scan_keys(&self, prefix: &StorePath) -> StorageResult<Vec<StorePath>> {
        self.backend.scan_keys(prefix)
    }
    fn parallel_reads(&self) -> bool {
        self.backend.parallel_reads()
    }
    fn files_layout(&self) -> Option<StoreLayout> {
        self.backend.files_layout()
    }

    fn persist_health(&self) -> Option<Arc<crate::store::durable::PersistHealth>> {
        self.backend.persist_health()
    }

    #[cfg(feature = "test-utils")]
    fn format_record(&self) -> Option<&dyn TestFormatRecord> {
        self.backend.format_record()
    }

    #[cfg(feature = "test-utils")]
    fn reread_from_disk(&self) {
        self.backend.reread_from_disk()
    }

    fn save_now(&self) -> StorageResult<()> {
        self.backend.save_now()
    }
    fn close(&self) -> StorageResult<()> {
        self.backend.close()
    }
    fn is_closed(&self) -> bool {
        self.backend.is_closed()
    }
    fn subscribe(&self, kind: crate::SubscriptionKind, callback: StoreCallback) -> SubscriptionId {
        self.backend.subscribe(kind, callback)
    }
    fn unsubscribe(&self, id: SubscriptionId) -> bool {
        self.backend.unsubscribe(id)
    }
    fn flush_prefix(&self, prefix: &StorePath) -> StorageResult<()> {
        self.backend.flush_prefix(prefix)
    }
    fn flush_async(&self) -> Commit {
        self.backend.flush_async()
    }
    fn save_async(&self) -> Commit {
        self.backend.save_async()
    }
    fn close_async(&self) -> Commit {
        self.backend.close_async()
    }
    fn is_initialized(&self, namespace: &StorePath) -> StorageResult<bool> {
        self.backend.is_initialized(namespace)
    }
    fn record_schema(&self, at: &StorePath, schema: &SchemaSnapshot) -> StorageResult<()> {
        self.backend.record_schema(at, schema)
    }

    fn set_initialized(&self, namespace: &StorePath, state: InitState) -> StorageResult<()> {
        self.backend.set_initialized(namespace, state)
    }
}