Fix/leveldb transaction

examples/cli.rs

@@ -9,6 +9,7 @@ use crsl_lib::crdt::{
 use crsl_lib::dasl::cid::ContentId;
 use crsl_lib::graph::{dag::DagGraph, storage::LeveldbNodeStorage};
 use crsl_lib::repo::Repo;
+use crsl_lib::storage::SharedLeveldb;
 use std::collections::{HashMap, HashSet};
 use std::error::Error;
 use std::path::{Path, PathBuf};
@@ -63,12 +64,11 @@ fn main() -> Result<(), Box<dyn Error>> {
     match cli.cmd {
         Commands::Init { path } => {
             std::fs::create_dir_all(&path)?;
-            std::fs::create_dir_all(path.join("ops"))?;
-            std::fs::create_dir_all(path.join("nodes"))?;
+            std::fs::create_dir_all(path.join("store"))?;
 
             std::fs::write(path.join(".crsl"), "")?;
 
-            println!("Initialized CRSL repository at {path:?}");
+            println!("Initialized CRSL repository at {path:?} (single LevelDB store)");
         }
         other_command => {
             let repo_path = Path::new(DEFAULT_REPO_PATH);
@@ -215,10 +215,9 @@ fn main() -> Result<(), Box<dyn Error>> {
 }
 
 fn open_repo(repo_path: &Path) -> Result<CliRepo, Box<dyn Error>> {
-    let op_storage = LeveldbStorage::open(repo_path.join("ops"))?;
-    let node_storage = LeveldbNodeStorage::open(repo_path.join("nodes"));
-    let state = CrdtState::new(op_storage);
-    let dag = DagGraph::new(node_storage);
+    let shared = SharedLeveldb::open(repo_path.join("store"))?;
+    let state = CrdtState::new(LeveldbStorage::new(shared.clone()));
+    let dag = DagGraph::new(LeveldbNodeStorage::new(shared));
     Ok(Repo::new(state, dag))
 }
 

examples/content_versioning.rs

@@ -14,6 +14,7 @@ use crsl_lib::{
         storage::LeveldbStorage as OpStore,
     },
     graph::{dag::DagGraph, storage::LeveldbNodeStorage as NodeStorage},
+    storage::SharedLeveldb,
 };
 use tempfile::tempdir;
 
@@ -23,8 +24,9 @@ type ContentState = CrdtState<String, Content, Store, LwwReducer>;
 
 fn main() {
     let tmp = tempdir().expect("tmp dir");
-    let op_store = OpStore::open(tmp.path().join("ops")).unwrap();
-    let node_store = NodeStorage::open(tmp.path().join("nodes"));
+    let shared = SharedLeveldb::open(tmp.path().join("store")).unwrap();
+    let op_store = OpStore::new(shared.clone());
+    let node_store = NodeStorage::new(shared);
     let state = ContentState::new(op_store);
     let mut _dag = DagGraph::<_, Content, ()>::new(node_store);
 

readme.md

@@ -24,6 +24,7 @@ use crsl_lib::{
     },
     graph::{dag::DagGraph, storage::LeveldbNodeStorage as NodeStorage},
     repo::Repo,
+    storage::SharedLeveldb,
 };
 use tempfile::tempdir;
 use cid::Cid;
@@ -34,10 +35,9 @@ struct Content(String);
 fn main() {
     // Initialize storage
     let tmp = tempdir().expect("tmp dir");
-    let op_store = OpStore::open(tmp.path().join("ops")).unwrap();
-    let node_store = NodeStorage::open(tmp.path().join("nodes"));
-    let state = CrdtState::new(op_store);
-    let dag = DagGraph::new(node_store);
+    let shared = SharedLeveldb::open(tmp.path().join("store")).unwrap();
+    let state = CrdtState::new(OpStore::new(shared.clone()));
+    let dag = DagGraph::new(NodeStorage::new(shared));
     let mut repo = Repo::new(state, dag);
 
     // Create a content ID (in practice, you'd use a proper CID)

src/crdt/crdt_state.rs

@@ -4,6 +4,7 @@ use crate::crdt::reducer::Reducer;
 use crate::crdt::storage::OperationStorage;
 use std::fmt::Debug;
 use std::marker::PhantomData;
+use ulid::Ulid;
 /// A generic CRDT state container that manages operations on content.
 ///
 /// `CrdtState` provides a high-level interface for applying operations to content
@@ -39,6 +40,10 @@ where
             _marker: PhantomData,
         }
     }
+
+    pub fn storage(&self) -> &S {
+        &self.storage
+    }
     /// Applies an operation to the CRDT state without validation.
     ///
     /// This method directly saves the operation to storage without checking its validity.
@@ -84,6 +89,14 @@ where
         self.storage.load_operations(genesis)
     }
 
+    pub fn get_operation(&self, op_id: &Ulid) -> Result<Option<Operation<ContentId, T>>> {
+        self.storage.get_operation(op_id)
+    }
+
+    pub fn delete_operation(&self, op_id: &Ulid) -> Result<()> {
+        self.storage.delete_operation(op_id)
+    }
+
     /// Validates whether an operation is logically valid to apply.
     ///
     /// This method performs the following checks:
@@ -124,6 +137,7 @@ mod tests {
     #[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
     struct DummyPayload(String);
 
+    /// Helper for constructing operations with deterministic timestamps.
     fn make_op(
         id: u64,
         ts: u64,