// Copyright 2024 Google LLC
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

//! submerge — data store designed for syncing
//!
//! submerge is an always-available, eventually consistent data store designed for use to
//! synchronize data across multiple devices.
//!
//! # Data model
//!
//! The data model for submerge consists of four types:
//! 1. `Map` — the map is the main structural type in submerge's data model. The key for the map is
//!    a string, and the value for the map is recursively another CRDT type.
//! 2. `Set` — a CRDT set (_causal-length set_) that can contain multiple unordered values.
//! 3. `Register` — a CRDT register that can contain a single atomically updated value.
//! 4. `VectorData` — a vector with an entry for each node in the system, where entries can only be
//!    written to by its owner.
//!
//! The value type is currently defined as raw bytes, but this can be changed in the future to allow
//! dynamic typing.
//!
//! # Interfaces
//! See [`NetworkInterface`] and [`StorageInterface`].

use std::{
    borrow::BorrowMut, cmp::Ordering, collections::hash_map::RandomState, fmt::Debug, hash::Hash,
};

use arbitrary::Arbitrary;
use crdt::{
    delta::DeltaMut,
    lww_crdt_container::LwwCrdtContainer,
    lww_map::LwwMap,
    register::Register,
    set::Set,
    typed_crdt::TypedCrdt,
    vector_data::{VectorData, VersionOverflow},
    ContentEq, CrdtState,
};
use derive_where::derive_where;
use distributed_time::{
    hybrid_logical_clock::HybridLogicalTimestamp, vector_clock::VectorClock, DistributedClock,
    TimestampOverflow, WallTimestampProvider,
};
use log::{debug, info};
use serde::{Deserialize, Serialize};
use thiserror::Error;

#[cfg(feature = "proto")]
pub mod proto;

#[cfg(any(test, feature = "checker"))]
pub mod testing;

/// The ID of a document.
pub type DocId = String;

type ValueType<Conf> = <Conf as DataStoreConfig>::ValueType;
type NodeId<Conf> = <Conf as DataStoreConfig>::NodeId;

/// Type alias for the map type for the given [`DataStoreConfig`].
pub type SubmergeMap<Conf> = LwwMap<
    String,
    TypedCrdt<ValueType<Conf>, NodeId<Conf>>,
    HybridLogicalTimestamp<NodeId<Conf>>,
    NodeId<Conf>,
>;
/// Type alias for the register type for the given [`DataStoreConfig`].
pub type SubmergeRegister<Conf> = Register<ValueType<Conf>, VectorClock<NodeId<Conf>>>;
/// Type alias for the set type for the given [`DataStoreConfig`].
pub type SubmergeSet<Conf> = Set<ValueType<Conf>, RandomState>;
/// Type alias for the vector data type for the given [`DataStoreConfig`].
pub type SubmergeVectorData<Conf> = VectorData<NodeId<Conf>, ValueType<Conf>>;

/// Type alias for the container type for the given [`DataStoreConfig`].
pub type SubmergeContainer<Conf> = LwwCrdtContainer<
    TypedCrdt<ValueType<Conf>, NodeId<Conf>>,
    HybridLogicalTimestamp<NodeId<Conf>>,
    NodeId<Conf>,
>;

/// A submerge document that implements the data model as described in the module documentation.
///
/// This document is a CRDT and guarantees eventual consistency. Instances of document should be
/// created by a [`DataStore`] via [`DataStore::open_or_create_document`].
#[derive(Serialize, Deserialize)]
#[derive_where(Default, Debug, Clone, PartialEq, Eq)]
#[serde(bound(
    serialize = "Conf::NodeId: Serialize, Conf::ValueType: Serialize",
    deserialize = "Conf::NodeId: for<'a> Deserialize<'a>, Conf::ValueType: for<'a> Deserialize<'a>"
))]
pub struct Document<Conf: DataStoreConfig> {
    /// The root container stored in this document.
    pub root: SubmergeContainer<Conf>,
    /// The version of this document. This is a vector clock that is incremented every time a
    /// transaction is committed, used to ensure causal consistency is not violated by incoming
    /// update messages.
    pub version: VectorClock<Conf::NodeId>,
}

// Manual implementation of Arbitrary to specify the correct trait bounds.
impl<'a, Conf: DataStoreConfig> Arbitrary<'a> for Document<Conf>
where
    Conf::NodeId: Arbitrary<'a>,
    Conf::ValueType: Arbitrary<'a>,
{
    fn arbitrary(u: &mut arbitrary::Unstructured<'a>) -> arbitrary::Result<Self> {
        Ok(Self {
            root: u.arbitrary()?,
            version: u.arbitrary()?,
        })
    }
}

impl<Conf: DataStoreConfig> Document<Conf> {
    /// Calculates the delta needed to update from the given `base_version`.
    ///
    /// The `base_version` must be a version vector from a replica of the same document. The
    /// returned document can be applied using [`DeltaDocument::into_document`] by supplying a base
    /// that is at least as new as `base_version`.
    pub fn calculate_delta(&self, base_version: VectorClock<Conf::NodeId>) -> DeltaDocument<Conf> {
        if self.version <= base_version {
            // Return an empty delta
            return DeltaDocument {
                root: SubmergeContainer::<Conf>::default(),
                base_version,
                version: self.version.clone(),
            };
        }
        DeltaDocument {
            root: self.root.calculate_delta(&base_version),
            base_version,
            version: self.version.clone(),
        }
    }

    /// Transform this document into a full delta document.
    ///
    /// This "full" delta document is going to contain all data in this document, thus allowing it
    /// to be applied from any version.
    pub fn into_full_delta(self) -> DeltaDocument<Conf> {
        DeltaDocument {
            root: self.root,
            base_version: VectorClock::default(),
            version: self.version,
        }
    }
}

/// A delta document that holds a delta update for a document.
///
/// This delta contains information about the changes between [`base_version`][Self::base_version]
/// and [`version`][Self::version], and can be merged with documents as least as new as
/// `base_version` without violating causal consistency.
///
/// This is serialized and sent over the wire when an update is made or requested by a remote peer.
#[derive(Serialize, Deserialize)]
#[derive_where(Clone, PartialEq, Eq)]
#[derive_where(Debug)]
#[serde(bound(
    serialize = "Conf::NodeId: Serialize, Conf::ValueType: Serialize",
    deserialize = "Conf::NodeId: for<'a> Deserialize<'a>, Conf::ValueType: for<'a> Deserialize<'a>"
))]
pub struct DeltaDocument<Conf: DataStoreConfig> {
    /// The root of the delta component of the document.
    pub root: SubmergeContainer<Conf>,
    /// The base version at which this delta starts. This represents the version of the base
    /// component at which this delta is being constructed, meaning that to recreate the resulting
    /// document, the base document must be at least as new as this version.
    pub base_version: VectorClock<Conf::NodeId>,
    /// The version at which the delta document is at.
    pub version: VectorClock<Conf::NodeId>,
}

impl<'a, Conf: DataStoreConfig> Arbitrary<'a> for DeltaDocument<Conf>
where
    Conf::NodeId: Arbitrary<'a>,
    Conf::ValueType: Arbitrary<'a>,
{
    fn arbitrary(u: &mut arbitrary::Unstructured<'a>) -> arbitrary::Result<Self> {
        Ok(Self {
            root: u.arbitrary()?,
            base_version: u.arbitrary()?,
            version: u.arbitrary()?,
        })
    }
}

/// Error returned when a delta is not applicable to the given base.
#[derive(Debug, Clone, Error)]
pub enum DeltaApplicationError {
    /// Delta is not applicable to the given base, base version is too old.
    #[error("Delta is not applicable to the given base, base version is too old")]
    BaseVersionTooOld,
    /// The CRDT merge failed.
    #[error("The CRDT merge failed")]
    CrdtMergeError(#[from] crdt::MergeError),
}

impl<Conf: DataStoreConfig> DeltaDocument<Conf> {
    /// Merge this delta document into the given `base`.
    ///
    /// The given `base` should be generated with [`Document::calculate_delta`] or
    /// [`DeltaOwned::into_delta`][crdt::delta::DeltaOwned::into_delta], and must be at least as new
    /// as the [`base_version`][Self::base_version], or `Err(DeltaApplicationError::BaseVersionTooOld)`
    /// will be returned.
    ///
    /// The resulting document will contain all changes from both this delta and the `base`, merged
    /// together according to the CRDT definitions.
    pub fn into_document(
        self,
        base: &Document<Conf>,
    ) -> Result<Document<Conf>, DeltaApplicationError> {
        if let None | Some(Ordering::Less) = base.version.partial_cmp(&self.base_version) {
            Err(DeltaApplicationError::BaseVersionTooOld)
        } else {
            match CrdtState::merge(&self.root, &base.root) {
                Ok(root) => Ok(Document {
                    root,
                    version: VectorClock::least_upper_bound(&self.version, &base.version),
                }),
                Err(e) => Err(DeltaApplicationError::CrdtMergeError(e)),
            }
        }
    }
}

/// Error when trying to commit a transaction to a document.
#[derive(Error)]
#[derive_where(Debug; StorageError<Conf>)]
pub enum TransactionError<Conf: DataStoreConfig> {
    /// Cannot perform the transaction because the timestamp overflowed.
    #[error("Cannot perform the transaction because the timestamp overflowed.")]
    TimestampOverflow,
    /// Cannot perform the transaction because the version number overflowed.
    #[error("Cannot perform the transaction because the version number overflowed.")]
    VersionOverflow,
    /// The storage interface failed to persist the resulting data.
    #[error(transparent)]
    StorageError(StorageError<Conf>),
    /// The CRDT merge failed.
    #[error("The CRDT merge failed.")]
    MergeError(#[from] crdt::MergeError),
}

impl<Conf: DataStoreConfig> From<TimestampOverflow> for TransactionError<Conf> {
    fn from(_: TimestampOverflow) -> Self {
        TransactionError::TimestampOverflow
    }
}

impl<Conf: DataStoreConfig> From<VersionOverflow> for TransactionError<Conf> {
    fn from(_: VersionOverflow) -> Self {
        TransactionError::VersionOverflow
    }
}

/// The network interface for a data store.
///
/// When the data store is modified, updated documents will be generated and passed to
/// `on_new_update`. The network interface should make a best-effort to send the update messages to
/// other nodes on the network, potentially retrying if certain nodes are unreachable.
///
/// It is safe for the network interface to send the same messages multiple times, or out of order.
/// Eventual consistency of the documents are guaranteed as long as the update messages are
/// eventually delivered.
///
/// The data store will not wait for the network message to be sent or delivered. If the network
/// layer wishes to get the latest state to send over the network, it can do so by calling
/// [`DataStore::snapshot`].
pub trait NetworkInterface<Conf: DataStoreConfig> {
    /// Called when a new update to the data store should be sent to to other nodes over the
    /// network.
    ///
    /// Failed updates are ignored and reconciled when the device reconnects. Alternatively, the
    /// network implementation can get the latest snapshot from the datastore and sync that.
    fn on_new_update(&mut self, doc_id: &str, document: DeltaDocument<Conf>);
}

/// The storage interface for a data store.
///
/// When the data store is modified, or when new update messages are received from the network,
/// `on_new_update` will be called to request the storage layer to persist the latest state. The
/// storage API is fallible, and failure to persist the data will result in a corresponding
/// [`TransactionError`].
pub trait StorageInterface<Conf: DataStoreConfig> {
    /// The error type for this storage.
    type Error: Debug;

    /// Called when a new update to the data store should be persisted to storage.
    fn on_new_update(&mut self, doc_id: &str, document: &Document<Conf>)
        -> Result<(), Self::Error>;

    /// Read a serialized document from storage.
    ///
    /// This function returns `None` if there are no documents with the given `doc_id`.
    fn read_from_storage(&self, doc_id: &str) -> Result<Option<Document<Conf>>, Self::Error>;
}

/// The update context for a given submerge data store.
#[derive_where(Debug, Clone)]
pub struct UpdateContext<Conf: DataStoreConfig> {
    /// The local node ID representing the device this code is running on. This ID must be unique
    /// among the group that participates in syncing this [`DataStore`].
    pub node_id: Conf::NodeId,
    /// The timestamp provider that gives the wall-clock time. See [`WallTimestampProvider`].
    pub timestamp_provider: Conf::TimestampProvider,
    /// The version of the document.
    ///
    /// The CRDT implementations assume that this context is monotonically increasing, meaning that
    /// the version may be copied into the CRDT tree, and any later operations' context must return
    /// a version newer than or equal to the copied version value.
    ///
    /// See [`crdt::UpdateContext::version`].
    pub version: VectorClock<Conf::NodeId>,
}

impl<Conf: DataStoreConfig> crdt::UpdateContext<Conf::NodeId> for UpdateContext<Conf> {
    type WallClock = Conf::TimestampProvider;

    fn updater(&self) -> &Conf::NodeId {
        &self.node_id
    }

    fn timestamp_provider(&self) -> &Self::WallClock {
        &self.timestamp_provider
    }

    fn version(&self) -> &VectorClock<Conf::NodeId> {
        &self.version
    }
}

/// A document with its metadata.
///
/// This handle holds the document together with the node-specific metadata needed to update it,
/// such as the local node ID and the wall-clock timestamp provider.
#[derive_where(Debug)]
pub struct DocumentHandle<Conf: DataStoreConfig> {
    id: DocId,
    doc: Document<Conf>,
    metadata: NodeMetadata<Conf>,
}

/// Type-level configuration for the [`DataStore`].
///
/// Implementations do not need to be instantiable, and can be implemented by unit structs or even
/// zero-variant enums.
pub trait DataStoreConfig: Sized {
    /// The type representing each node in this data store. This node ID is assigned when the
    /// [`DataStore`] is constructed, must be unique among all of the syncing devices, and must not
    /// be reused.
    type NodeId: Ord + Clone + Debug + 'static;
    /// The value type shared between all types in the data model. Values in `Set`, `Register`, and
    /// `VectorData` all store instances of this type.
    type ValueType: Eq + Hash + Clone + Debug + 'static;
    /// The network interface used to send update messages to other peer nodes.
    ///
    /// See [`NetworkInterface`].
    type Network: NetworkInterface<Self> + Debug + Clone;
    /// The storage interface used to persist new versions of the document, and read persisted
    /// documents back from disk.
    ///
    /// See [`StorageInterface`].
    type Storage: StorageInterface<Self> + Debug + Clone;
    /// The timestamp provider used to provide wall clock timestamps.
    ///
    /// See [`WallTimestampProvider`] for the format and requirements of the timestamps.
    type TimestampProvider: WallTimestampProvider + Debug + Clone;
}

/// A transaction for a given document.
///
/// This transaction temporarily accumulates mutations until it is [`commit`][Self::commit] is
/// called.
#[derive_where(Debug; Handle)]
pub struct DocumentTransaction<Handle, Conf>
where
    Handle: BorrowMut<DocumentHandle<Conf>>,
    Conf: DataStoreConfig,
{
    /// The document handle this transaction is for.
    handle: Handle,
    /// The delta component of this transaction (containing changes made during this transaction).
    delta: SubmergeContainer<Conf>,
    /// The update context for this document transaction.
    update_context: UpdateContext<Conf>,
    /// The delta from network update only. This is a subset of the overall delta.
    network_delta: SubmergeContainer<Conf>,
}

impl<Handle, Conf> DocumentTransaction<Handle, Conf>
where
    Handle: BorrowMut<DocumentHandle<Conf>>,
    Conf: DataStoreConfig,
{
    /// Create a transaction for this document.
    ///
    /// The function for both are similar, the main difference is that [`DocumentTransaction`] uses
    /// the explicit [`commit`][DocumentTransaction::commit] method, whereas
    /// [`DocumentHandle::modify`] automatically commits when the closure exits.
    pub fn new(handle: Handle) -> Self {
        let handle_borrowed = handle.borrow();
        let update_context = UpdateContext {
            node_id: handle_borrowed.metadata.node_id.clone(),
            timestamp_provider: handle_borrowed.metadata.timestamp_provider.clone(),
            version: handle_borrowed.doc.version.clone(),
        };
        Self {
            handle,
            delta: Default::default(),
            update_context,
            network_delta: Default::default(),
        }
    }

    /// Get a mutable reference to the root document and the update context.
    pub fn get_context_and_root_mut(
        &mut self,
    ) -> (&UpdateContext<Conf>, DeltaMut<'_, SubmergeContainer<Conf>>) {
        // Returns both items in one call just to work around partial borrow issues.
        let handle = self.handle.borrow_mut();
        (
            &self.update_context,
            DeltaMut {
                base: Some(&handle.doc.root),
                delta: &mut self.delta,
            },
        )
    }

    /// Commit the modifications in this transaction to the data store.
    ///
    /// ## Returns
    /// * `Ok` if the network update was successfully committed
    /// * [`VersionOverflow`][TransactionError::VersionOverflow] if the update would cause the
    ///   version timestamp to overflow. This happens when the node has made more than [`u32::MAX`]
    ///   changes to the datastore.
    /// * [`StorageError`][TransactionError::StorageError] if it failed to persist the updated data.
    ///   See the documentation for the [`StorageInterface`] implementation for details on how to
    ///   handle the error.
    pub fn commit(mut self) -> Result<(), TransactionError<Conf>> {
        let handle = self.handle.borrow_mut();
        let merged = CrdtState::merge(&handle.doc.root, &self.delta)?;
        let merged_network_only = CrdtState::merge(&handle.doc.root, &self.network_delta)?;
        // Network change should consider both content change and subtree_version change.
        let has_network_change = merged_network_only != handle.doc.root;
        // Local change should only consider content change.
        let has_local_change = !merged.content_eq(&merged_network_only);
        if !has_network_change && !has_local_change {
            info!("No updates were made");
            return Ok(());
        }
        let new_version = if has_local_change {
            self.update_context
                .version
                .increment(&handle.metadata.node_id)?
        } else {
            self.update_context.version
        };
        let new_root = if has_local_change {
            merged
        } else {
            // Use network merge if there is no local change. This is to ensure any subtree_version
            // change caused by local read won't be committed.
            merged_network_only
        };
        let updated_doc = Document {
            root: new_root,
            version: new_version.clone(),
        };
        // Send network update if there is a local change.
        if has_local_change {
            let delta_doc = DeltaDocument {
                root: self.delta,
                base_version: handle.doc.version.clone(),
                version: new_version,
            };
            handle.metadata.network.on_new_update(&handle.id, delta_doc);
        }
        handle
            .metadata
            .storage
            .on_new_update(&handle.id, &updated_doc)
            .map_err(TransactionError::StorageError)?;
        handle.doc = updated_doc;
        Ok(())
    }

    /// Get the update context for the local node in this data store.
    pub fn update_context(&self) -> &UpdateContext<Conf> {
        &self.update_context
    }

    /// Merge a network delta into this transaction.
    pub fn merge_network_delta(
        &mut self,
        delta: &DeltaDocument<Conf>,
    ) -> Result<(), CommitUpdateError<Conf>> {
        if let Some(Ordering::Less) | None =
            self.update_context.version.partial_cmp(&delta.base_version)
        {
            return Err(CommitUpdateError::OlderBaseNeeded {
                required_base_version: self.update_context.version.clone(),
            });
        }
        if self.update_context.version >= delta.version {
            debug!(
                "Discard obsolete network update. our version={:?}. their version={:?}.",
                self.update_context.version, delta.version,
            );
            return Ok(());
        }
        let new_delta = CrdtState::merge(&self.delta, &delta.root)?;
        // Attempt merging with the base now to ensure the network delta is compatible.
        let _ = CrdtState::merge(&self.handle.borrow_mut().doc.root, &new_delta)?;
        self.delta = new_delta;
        self.network_delta = CrdtState::merge(&self.network_delta, &delta.root)?;
        self.update_context.version =
            VectorClock::least_upper_bound(&self.update_context.version, &delta.version);
        Ok(())
    }
}

/// Error type used for [`DocumentHandle::commit_network_update`].
#[derive(Debug, Error)]
#[allow(missing_docs)]
pub enum CommitUpdateError<Conf: DataStoreConfig> {
    #[error(transparent)]
    StorageError(StorageError<Conf>),
    #[error("Delta update needs to cover to an older base to be applied")]
    OlderBaseNeeded {
        /// The base version needed to successfully update this data store.
        required_base_version: VectorClock<Conf::NodeId>,
    },
    /// The CRDT merge failed.
    #[error("The CRDT merge failed.")]
    MergeError(#[from] crdt::MergeError),
}

/// Error type used for [`DocumentHandle::modify`].
#[derive(Error)]
#[derive_where(Debug; E)]
pub enum ModifyError<E, Conf: DataStoreConfig> {
    /// Failed to commit the modifications. See [`TransactionError`].
    #[error(transparent)]
    TransactionFailed(#[from] TransactionError<Conf>),
    /// Custom error type returned from the caller-provided closure.
    Custom(E),
}

type StorageError<Conf> = <<Conf as DataStoreConfig>::Storage as StorageInterface<Conf>>::Error;

impl<Conf> DocumentHandle<Conf>
where
    Conf: DataStoreConfig,
{
    /// Get the document ID for the document.
    pub fn doc_id(&self) -> &DocId {
        &self.id
    }

    /// Get a snapshot of the document.
    pub fn snapshot(&self) -> &Document<Conf> {
        &self.doc
    }

    /// Modify the document.
    ///
    /// This method takes a closure that the caller can use to mutate the document. If the closure
    /// returns success, the mutated document is committed, persisting the result to disk and
    /// sending update messages over the network.
    ///
    /// If the modify closure returns an error or panics, the mutated document is discarded and the
    /// persisted data remains in the previous state.
    pub fn modify<'a, E, F>(&'a mut self, modify: F) -> Result<(), ModifyError<E, Conf>>
    where
        F: for<'d> FnOnce(
            DeltaMut<'d, SubmergeContainer<Conf>>,
            &UpdateContext<Conf>,
        ) -> Result<(), E>,
    {
        let mut transaction = DocumentTransaction::new(self);
        let (ctx, doc) = transaction.get_context_and_root_mut();
        match modify(doc, ctx) {
            Ok(_) => {
                transaction.commit()?;
                Ok(())
            }
            Err(e) => Err(ModifyError::Custom(e)),
        }
    }

    /// Handle an update message from the network.
    ///
    /// Merge any changes received from the incoming update message, and persist the result.
    pub fn commit_network_update(
        &mut self,
        update_message: &DeltaDocument<Conf>,
    ) -> Result<(), CommitUpdateError<Conf>> {
        if let Some(Ordering::Less) | None =
            self.doc.version.partial_cmp(&update_message.base_version)
        {
            // Possible violation of causal consistency: We got a delta update but we are not
            // strictly newer than its base. Return an error to ask the updater for a wider delta.
            return Err(CommitUpdateError::OlderBaseNeeded {
                required_base_version: self.doc.version.clone(),
            });
        }
        if self.doc.version >= update_message.version {
            debug!(
                "Discard obsolete network update. our version={:?}. their version={:?}.",
                self.doc.version, update_message.version,
            );
            return Ok(());
        }
        let updated_document = Document {
            root: CrdtState::merge(&self.doc.root, &update_message.root)?,
            version: VectorClock::least_upper_bound(&self.doc.version, &update_message.version),
        };
        debug!(
            "Commit network update:\norig={:?}\nupdate={:?}\nnew={:?}",
            self.doc, update_message, updated_document
        );
        self.metadata
            .storage
            .on_new_update(&self.id, &updated_document)
            .map_err(CommitUpdateError::StorageError)?;
        self.doc = updated_document;
        Ok(())
    }
}

/// Contains the necessary metadata to update a document.
#[derive_where(Debug, Clone)]
pub struct NodeMetadata<Conf: DataStoreConfig> {
    network: Conf::Network,
    storage: Conf::Storage,
    node_id: Conf::NodeId,
    timestamp_provider: Conf::TimestampProvider,
}

/// A `DataStore` manages multiple [`Document`]s.
///
/// The `DataStore` is a unit of configuration that shares the same network and storage interfaces.
/// This means that all documents in the same `DataStore` are assumed to be synced with the same
/// group and persisted to the same location.
///
/// As mentioned above, [`Document`] is the unit of causality tracking. Causal consistency is not
/// guaranteed across different [`Documents`][Document] within the same `DataStore`.
#[derive(Debug)]
pub struct DataStore<Conf: DataStoreConfig> {
    metadata: NodeMetadata<Conf>,
}

impl<Conf: DataStoreConfig> DataStore<Conf> {
    /// Create a new data store.
    pub fn new(
        local_id: Conf::NodeId,
        network: Conf::Network,
        storage: Conf::Storage,
        timestamp_provider: Conf::TimestampProvider,
    ) -> Self {
        Self {
            metadata: NodeMetadata {
                network,
                storage,
                node_id: local_id,
                timestamp_provider,
            },
        }
    }

    /// Open the document with the given ID, creating it if it doesn't already exist.
    pub fn open_or_create_document(
        &mut self,
        id: impl Into<DocId>,
    ) -> Result<DocumentHandle<Conf>, StorageError<Conf>>
    where
        SubmergeContainer<Conf>: Default,
    {
        let id = id.into();
        let doc = self
            .metadata
            .storage
            .read_from_storage(&id)?
            .unwrap_or_default();
        Ok(DocumentHandle {
            id,
            doc,
            metadata: self.metadata.clone(),
        })
    }

    /// Get a snapshot of what is currently in the document.
    pub fn snapshot(&self, id: impl Into<DocId>) -> Option<Document<Conf>> {
        let id = id.into();
        self.metadata.storage.read_from_storage(&id).ok()?
    }

    /// Gets the storage interface for this data store.
    pub fn storage(&self) -> &Conf::Storage {
        &self.metadata.storage
    }
}