diamond_types_extended/

lib.rs

//! # Diamond Types Extended - Unified CRDT Library
//!
//! Diamond Types Extended provides high-performance CRDTs (Conflict-free Replicated Data Types) for
//! collaborative applications. It's a fork of [diamond-types](https://github.com/josephg/diamond-types)
//! with an ergonomic, unified API.
//!
//! ## Quick Start
//!
//! ```
//! use diamond_types_extended::{Document, Uuid, Value};
//!
//! // Create a document
//! let mut doc = Document::new();
//! let alice = doc.create_agent(Uuid::now_v7());
//!
//! // All mutations happen in transactions
//! doc.transact(alice, |tx| {
//!     tx.root().set("title", "My Document");
//!     tx.root().set("count", 42);
//!     tx.root().create_text("content");
//! });
//!
//! // Access the text CRDT
//! doc.transact(alice, |tx| {
//!     if let Some(mut text) = tx.get_text_mut(&["content"]) {
//!         text.insert(0, "Hello, world!");
//!     }
//! });
//!
//! // Read values directly
//! assert_eq!(doc.root().get("title").unwrap().as_str(), Some("My Document"));
//! assert_eq!(doc.root().get_text("content").unwrap().content(), "Hello, world!");
//! ```
//!
//! ## CRDT Types
//!
//! - **Map**: Key-value container with LWW (Last-Writer-Wins) registers per key
//! - **Text**: Sequence CRDT for collaborative text editing
//! - **Set**: OR-Set (Observed-Remove) with add-wins semantics
//! - **Register**: Single-value LWW container
//!
//! All types can be nested within Maps.
//!
//! ## Replication
//!
//! ```ignore
//! // Peer A creates some changes
//! let ops = doc_a.ops_since(&Frontier::root()).into_owned();
//!
//! // Peer B merges them
//! doc_b.merge_ops(ops)?;
//! ```
//!
//! ## Attribution
//!
//! Diamond Types Extended is built on diamond-types by Joseph Gentle. See ATTRIBUTION.md for details.

#![allow(clippy::module_inception)]

use std::collections::{BTreeMap, BTreeSet};
use std::fmt::{Debug, Formatter};

use jumprope::JumpRopeBuf;
use serde::{Deserialize, Serialize};
use smallvec::SmallVec;
use smartstring::alias::String as SmartString;

// HasLength is re-exported from our vendored rle module for internal use.
use causalgraph::graph::Graph;
pub use frontier::Frontier;

pub use uuid::Uuid;
pub use ordered_float::NotNan;
pub use crate::causalgraph::agent_assignment::remote_ids::{RemoteVersion, RemoteFrontier};
use crate::causalgraph::agent_span::AgentVersion;
pub(crate) use crate::causalgraph::CausalGraph;
pub(crate) use crate::dtrange::DTRange;
use crate::list::op_metrics::{ListOperationCtx, ListOpMetrics};

use crate::rle::{KVPair, RleVec};
use crate::textinfo::TextInfo;

// use crate::list::internal_op::OperationInternal as TextOpInternal;

// ============ New Diamond Types Extended Public API ============
mod value;
mod document;
mod refs;
mod muts;

pub use value::{Value, PrimitiveValue, MaterializedValue, CrdtId, Conflicted};
pub use document::{Document, DocumentWriter, Transaction};
pub use refs::{MapRef, TextRef, SetRef, RegisterRef};
pub use muts::{MapMut, TextMut, SetMut};
pub use encoding::parseerror::ParseError;

// ============ Original diamond-types modules (upstream internals) ============
// These modules contain upstream diamond-types code. Many have WIP features
// and unused items that we preserve for compatibility. Dead code warnings are
// suppressed at the module level rather than per-item.
#[allow(dead_code)]
pub(crate) mod list;
#[allow(dead_code)]
pub(crate) mod register;
#[allow(dead_code)]
pub(crate) mod map;
#[allow(dead_code)]
pub(crate) mod set;

#[allow(dead_code)]
pub(crate) mod rle;
#[allow(dead_code)]
mod dtrange;
mod unicount;
#[allow(dead_code)]
mod rev_range;
#[allow(dead_code)]
pub(crate) mod frontier;
mod check;
#[allow(dead_code)]
pub(crate) mod encoding;
#[allow(dead_code)]
pub(crate) mod causalgraph;
#[allow(dead_code)]
mod wal;
#[allow(dead_code)]
mod ost;

#[allow(dead_code, unused_imports)]
pub(crate) mod serde_helpers;

#[allow(dead_code)]
mod listmerge;

#[cfg(any(test, feature = "gen_test_data"))]
mod list_fuzzer_tools;
#[cfg(test)]
mod fuzzer;
#[allow(dead_code)]
mod branch;
mod textinfo;
#[allow(dead_code)]
mod oplog;
#[cfg(feature = "storage")]
#[allow(dead_code)]
mod storage;
#[allow(dead_code)]
mod simple_checkout;
#[allow(dead_code)]
mod stats;

pub type AgentId = u32;

// TODO: Consider changing this to u64 to add support for very long lived documents even on 32 bit
// systems like wasm32
/// An LV (LocalVersion) is used all over the place internally to identify a single operation.
///
/// A local version (as the name implies) is local-only. Local versions generally need to be
/// converted to RawVersions before being sent over the wire or saved to disk.
pub type LV = usize;

#[derive(Clone, Eq, PartialEq, Ord, PartialOrd)]
#[derive(Serialize, Deserialize)]
pub(crate) enum Primitive {
    Nil,
    Bool(bool),
    I64(i64),
    F64(NotNan<f64>),
    Str(SmartString),

    #[serde(skip)]
    #[allow(dead_code)]
    InvalidUninitialized,
}

impl Debug for Primitive {
    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
        match self {
            Primitive::Nil => f.debug_struct("Nil").finish(),
            Primitive::Bool(val) => val.fmt(f),
            // Primitive::I64(val) => f.debug_tuple("I64").field(val).finish(),
            Primitive::I64(val) => val.fmt(f),
            Primitive::F64(val) => val.fmt(f),
            Primitive::Str(val) => val.fmt(f),
            Primitive::InvalidUninitialized => f.debug_tuple("InvalidUninitialized").finish()
        }
    }
}

// #[derive(Debug, Eq, PartialEq, Copy, Clone, TryFromPrimitive)]
#[derive(Debug, Eq, PartialEq, Copy, Clone)]
#[derive(Serialize, Deserialize)]
pub(crate) enum CRDTKind {
    Map,        // String => Register (like a JS object)
    Register,   // Single LWW value
    #[allow(dead_code)]
    Collection, // SQL table / mongo collection
    Text,       // Text/sequence CRDT
    Set,        // OR-Set (add-wins semantics)
}

#[derive(Debug, Clone, Eq, PartialEq)]
#[derive(Serialize, Deserialize)]
pub(crate) enum CreateValue {
    Primitive(Primitive),
    NewCRDT(CRDTKind),
    // Deleted, // Marks that the key / contents should be deleted.
}

// #[derive(Debug, Clone, Eq, PartialEq)]
// pub enum CollectionOp {
//     Insert(CreateValue),
//     Remove(LV),
// }

// #[derive(Debug, Clone, Eq, PartialEq)]
// pub(crate) enum OpContents {
//     RegisterSet(CreateValue),
//     MapSet(SmartString, CreateValue), // TODO: Inline the index here.
//     MapDelete(SmartString), // TODO: And here.
//     Collection(CollectionOp), // TODO: Consider just inlining this.
//     Text(ListOpMetrics),
//
//
//     // The other way to write this would be:
//
//
//     // SetInsert(CRDTKind),
//     // SetRemove(Time),
//
//     // TextInsert(..),
//     // TextRemove(..)
// }

// #[derive(Debug, Clone, Eq, PartialEq)]
// pub(crate) struct Op {
//     pub target_id: LV,
//     pub contents: OpContents,
// }

// #[derive(Debug, Clone, Eq, PartialEq, Default)]
// pub(crate) struct Ops {
//     /// Local version + op pairs
//     ops: RleVec<KVPair<Op>>,
//     list_ctx: ListOperationCtx,
// }

pub(crate) const ROOT_CRDT_ID: LV = usize::MAX;
#[allow(dead_code)]
pub(crate) const ROOT_CRDT_ID_AV: AgentVersion = (AgentId::MAX, 0);


// #[derive(Debug, Clone, Eq, PartialEq)]
// pub enum SnapshotValue {
//     Primitive(Primitive),
//     InnerCRDT(LV),
//     // Ref(LV),
// }
//
// #[derive(Debug, Clone, Eq, PartialEq)]
// struct RegisterState {
//     value: SnapshotValue,
//     version: LV,
// }

// /// Guaranteed to always have at least 1 value inside.
// type MVRegister = SmallVec<RegisterState, 1>;

// // TODO: Probably should also store a dirty flag for when we flush to disk.
// #[derive(Debug, Clone, Eq, PartialEq)]
// enum OverlayValue {
//     Register(MVRegister),
//     Map(BTreeMap<SmartString, MVRegister>),
//     Collection(BTreeMap<LV, SnapshotValue>),
//     Text(Box<JumpRope>),
// }

// type Pair<T> = (LV, T);
type ValPair = (LV, CreateValue);
// type RawPair<'a, T> = (RemoteVersion<'a>, T);
type LVKey = LV;


#[derive(Debug, Clone, Default)]
pub(crate) struct RegisterInfo {
    // I bet there's a clever way to use RLE to optimize this. Right now this contains the full
    // history of values this register has ever held.
    ops: Vec<ValPair>,

    /// Cached version(s) which together store the current HEAD for this register.
    supremum: SmallVec<usize, 2>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum RegisterValue {
    Primitive(Primitive),
    OwnedCRDT(CRDTKind, LVKey),
}


#[derive(Debug, Clone, Default)]
pub(crate) struct OpLog {
    pub(crate) cg: CausalGraph,


    // cg_storage: Option<CGStorage>,
    // wal_storage: Option<WriteAheadLog>,

    // Information about whether the map still exists!
    // maps: BTreeMap<LVKey, MapInfo>,

    /// (CRDT ID, key) -> MVRegister.
    map_keys: BTreeMap<(LVKey, SmartString), RegisterInfo>,
    /// CRDT ID -> Text CRDT.
    texts: BTreeMap<LVKey, TextInfo>,

    // These are always inserted at the end, but items in the middle are removed. There's probably
    // a better data structure to accomplish this.
    map_index: BTreeMap<LV, (LVKey, SmartString)>,
    text_index: BTreeMap<LV, LVKey>,

    /// Standalone registers (not inside maps).
    registers: BTreeMap<LVKey, RegisterInfo>,
    /// Index from operation LV to register CRDT ID.
    #[allow(dead_code)]
    register_index: BTreeMap<LV, LVKey>,

    /// OR-Sets storing Primitive values.
    sets: BTreeMap<LVKey, set::SetInfo<Primitive>>,
    /// Index from operation LV to set CRDT ID.
    set_index: BTreeMap<LV, LVKey>,

    // The set of CRDTs which have been deleted or superseded in the current version. This data is
    // pretty similar to the _index data, in that its mainly just useful for branches doing
    // checkouts.
    deleted_crdts: BTreeSet<LVKey>,
}

#[derive(Debug, Clone, Eq, PartialEq)]
pub(crate) struct Branch {
    pub(crate) frontier: Frontier,

    // Objects are always created at the highest version ID, but can be deleted anywhere in the
    // range.
    //
    // TODO: Replace BTreeMap with something more appropriate later.
    maps: BTreeMap<LVKey, BTreeMap<SmartString, RegisterState>>, // any objects.
    pub(crate) texts: BTreeMap<LVKey, JumpRopeBuf>,
    /// Standalone registers (not inside maps).
    pub(crate) registers: BTreeMap<LVKey, RegisterState>,
    /// OR-Sets storing Primitive values.
    pub(crate) sets: BTreeMap<LVKey, BTreeSet<Primitive>>,
}

/// The register stores the specified value, but if conflicts_with is not empty, it has some
/// conflicting concurrent values too. The `value` field will be consistent across all peers.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct RegisterState {
    /// The winning value according to LWW semantics.
    pub(crate) value: RegisterValue,
    /// Any concurrent values that lost the LWW tie-break.
    pub(crate) conflicts_with: Vec<RegisterValue>,
}

#[derive(Debug, Clone)]
#[derive(Serialize, Deserialize)]
pub struct SerializedOps<'a> {
    cg_changes: Vec<u8>,

    // The version of the op, and the name of the containing CRDT.
    #[serde(borrow)]
    map_ops: Vec<(RemoteVersion, RemoteVersion, &'a str, CreateValue)>,
    text_ops: Vec<(RemoteVersion, RemoteVersion, ListOpMetrics)>,
    text_context: ListOperationCtx,
    /// OR-Set operations: (crdt_name, op_version, `SerializedSetOp<Primitive>`)
    set_ops: Vec<(RemoteVersion, RemoteVersion, set::SerializedSetOp<Primitive>)>,
}

impl<'a> From<SerializedOps<'a>> for SerializedOpsOwned {
    fn from(ops: SerializedOps<'a>) -> Self {
        Self {
            cg_changes: ops.cg_changes,
            map_ops: ops.map_ops.into_iter().map(|(crdt_name, rv, key, val)| {
                (crdt_name, rv, SmartString::from(key), val)
            }).collect(),
            text_ops: ops.text_ops,
            text_context: ops.text_context,
            set_ops: ops.set_ops,
        }
    }
}

impl<'a> SerializedOps<'a> {
    /// Convert to an owned version that can be sent across threads.
    pub fn into_owned(self) -> SerializedOpsOwned {
        self.into()
    }

    /// Check if this contains no operations.
    pub fn is_empty(&self) -> bool {
        self.cg_changes.is_empty()
            && self.map_ops.is_empty()
            && self.text_ops.is_empty()
            && self.set_ops.is_empty()
    }
}

#[derive(Debug, Clone)]
#[derive(Serialize, Deserialize)]
pub struct SerializedOpsOwned {
    cg_changes: Vec<u8>,

    // The version of the op, and the name of the containing CRDT.
    map_ops: Vec<(RemoteVersion, RemoteVersion, SmartString, CreateValue)>,
    text_ops: Vec<(RemoteVersion, RemoteVersion, ListOpMetrics)>,
    text_context: ListOperationCtx,
    /// OR-Set operations: (crdt_name, op_version, `SerializedSetOp<Primitive>`)
    set_ops: Vec<(RemoteVersion, RemoteVersion, set::SerializedSetOp<Primitive>)>,
}

impl SerializedOpsOwned {
    /// Check if this contains no operations.
    pub fn is_empty(&self) -> bool {
        self.cg_changes.is_empty()
            && self.map_ops.is_empty()
            && self.text_ops.is_empty()
            && self.set_ops.is_empty()
    }
}

/// This is used for checkouts. This is a value tree.
#[derive(Debug, Clone, Eq, PartialEq)]
#[derive(Serialize, Deserialize)]
pub(crate) enum DTValue {
    Primitive(Primitive),
    /// A register containing a value (which could be a nested CRDT).
    Register(Box<DTValue>),
    Map(BTreeMap<SmartString, Box<DTValue>>),
    // Collection(BTreeMap<LV, Box<DTValue>>),
    Text(String),
    /// An OR-Set containing primitive values.
    Set(BTreeSet<Primitive>),
}