storage: refactor and rocksdb implementation

Cargo.toml

@@ -14,15 +14,22 @@ keywords = ["crypto", "key-transparency"]
 readme = "README.md"
 
 [workspace]
+default-members = [
+    "crates/prism",
+    "crates/common",
+    "crates/errors",
+    "crates/storage",
+    "crates/da",
+]
+
 members = [
     "crates/prism",
     "crates/common",
-    "crates/nova",
-    "crates/groth16",
     "crates/errors",
     "crates/sp1",
+    "crates/storage",
+    "crates/da",
 ]
-default-members = ["crates/prism", "crates/common", "crates/errors"]
 resolver = "2"
 
 [workspace.dependencies]
@@ -72,10 +79,14 @@ secp256k1 = "0.29.0"
 sp1-zkvm = { version = "1.2.0" }
 sp1-sdk = { version = "1.2.0" }
 prism-common = { path = "crates/common" }
+prism-storage = { path = "crates/storage" }
 prism-nova = { path = "crates/nova" }
+prism-da = { path = "crates/da" }
 prism-errors = { path = "crates/errors" }
 prism-main = { path = "crates/prism" }
 prism-groth16 = { path = "crates/groth16" }
+rocksdb = { version = "0.21.0", features = ["multi-threaded-cf"] }
+
 
 [patch.crates-io]
 sha2-v0-10-8 = { git = "https://github.com/sp1-patches/RustCrypto-hashes", package = "sha2", branch = "patch-sha2-v0.10.8" }

crates/common/src/hashchain.rs

@@ -9,7 +9,7 @@ use std::{
 
 use crate::{
     operation::{CreateAccountArgs, Operation, PublicKey, ServiceChallengeInput},
-    tree::{hash, Digest, Hasher},
+    tree::{Digest, Hasher},
 };
 
 #[derive(Clone, Serialize, Deserialize, Debug, PartialEq)]
@@ -280,12 +280,11 @@ pub struct HashchainEntry {
 
 impl HashchainEntry {
     pub fn new(operation: Operation, previous_hash: Digest) -> Self {
-        let hash = {
-            let mut data = Vec::new();
-            data.extend_from_slice(operation.to_string().as_bytes());
-            data.extend_from_slice(previous_hash.as_ref());
-            hash(&data)
-        };
+        let mut data = Vec::new();
+        data.extend_from_slice(operation.to_string().as_bytes());
+        data.extend_from_slice(previous_hash.as_ref());
+        let hash = Digest::hash(data);
+
         Self {
             hash,
             previous_hash,

crates/common/src/tree.rs

@@ -8,6 +8,7 @@ use jmt::{
 };
 use prism_errors::DatabaseError;
 use serde::{ser::SerializeTupleStruct, Deserialize, Serialize};
+use std::convert::{From, Into};
 use std::sync::Arc;
 
 use crate::{
@@ -70,21 +71,29 @@ impl SimpleHasher for Hasher {
     }
 }
 
-pub fn hash(data: &[u8]) -> Digest {
-    let mut hasher = Hasher::new();
-    hasher.update(data);
-    Digest(hasher.finalize())
-}
-
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Copy)]
-pub struct Digest([u8; 32]);
+pub struct Digest(pub [u8; 32]);
 
 impl Digest {
-    pub fn to_bytes(&self) -> [u8; 32] {
-        self.0
+    pub fn hash(data: impl AsRef<[u8]>) -> Self {
+        let mut hasher = Hasher::new();
+        hasher.update(data.as_ref());
+        Self(hasher.finalize())
+    }
+}
+
+// serializer and deserializer for rocksdb
+// converts from bytearrays into digests
+// padds it with zero if it is too small
+impl <const N: usize> From<[u8; N]> for Digest {
+    fn from(value: [u8; N]) -> Self {
+        let mut digest = [0u8; 32];
+        digest[..N].copy_from_slice(&value);
+        Self(digest)
     }
 }
 
+
 // implementing it for now to get things to compile, curve choice will be made later
 impl TryFrom<Digest> for Scalar {
     type Error = anyhow::Error;
@@ -316,7 +325,7 @@ where
         match operation {
             Operation::AddKey(KeyOperationArgs { id, .. })
             | Operation::RevokeKey(KeyOperationArgs { id, .. }) => {
-                let hashed_id = hash(id.as_bytes());
+                let hashed_id = Digest::hash(id);
                 let key_hash = KeyHash::with::<Hasher>(hashed_id);
 
                 let mut current_chain = self
@@ -337,7 +346,7 @@ where
                 service_id,
                 challenge,
             }) => {
-                let hashed_id = hash(id.as_bytes());
+                let hashed_id = Digest::hash(id);
                 let key_hash = KeyHash::with::<Hasher>(hashed_id);
 
                 match &challenge {

crates/da/Cargo.toml

@@ -0,0 +1,52 @@
+[package]
+name = "prism-da"
+version.workspace = true
+authors.workspace = true
+edition.workspace = true
+description.workspace = true
+homepage.workspace = true
+repository.workspace = true
+license.workspace = true
+keywords.workspace = true
+readme.workspace = true
+
+[dependencies]
+axum = { workspace = true }
+borsh = { workspace = true }
+tower-http = { workspace = true }
+utoipa = { workspace = true }
+utoipa-swagger-ui = { workspace = true }
+async-trait = { workspace = true }
+serde = { workspace = true }
+serde_json = { workspace = true }
+redis = { workspace = true }
+ed25519-dalek = { workspace = true }
+base64 = { workspace = true }
+tokio = { workspace = true }
+bellman = { workspace = true }
+bincode = { workspace = true }
+bls12_381 = { workspace = true }
+rand = { workspace = true }
+hex = { workspace = true }
+ff = { workspace = true }
+log = { workspace = true }
+pretty_env_logger = { workspace = true }
+clap = { workspace = true }
+config = { workspace = true }
+thiserror = { workspace = true }
+indexed-merkle-tree = { workspace = true }
+dotenvy = { workspace = true }
+celestia-rpc = { workspace = true }
+celestia-types = { workspace = true }
+mockall = { workspace = true }
+keystore-rs = { workspace = true }
+toml = { workspace = true }
+dirs = { workspace = true }
+anyhow = { workspace = true }
+jmt = { workspace = true }
+sha2 = { workspace = true }
+auto_impl = { workspace = true }
+prism-common = { workspace = true }
+prism-errors = { workspace = true }
+prism-groth16 = { workspace = true }
+sp1-sdk = { workspace = true }

crates/prism/src/da/celestia.rs → crates/da/src/celestia.rs

@@ -1,13 +1,12 @@
-use crate::{
-    cfg::CelestiaConfig,
-    da::{DataAvailabilityLayer, FinalizedEpoch},
-};
+use crate::{DataAvailabilityLayer, FinalizedEpoch};
 use anyhow::{anyhow, Context, Result};
 use async_trait::async_trait;
 use celestia_rpc::{BlobClient, Client, HeaderClient};
 use celestia_types::{blob::GasPrice, nmt::Namespace, Blob};
+use log::{debug, error, trace, warn};
 use prism_common::operation::Operation;
 use prism_errors::{DataAvailabilityError, GeneralError};
+use serde::{Deserialize, Serialize};
 use std::{
     self,
     sync::{
@@ -29,6 +28,25 @@ impl TryFrom<&Blob> for FinalizedEpoch {
     }
 }
 
+#[derive(Debug, Serialize, Deserialize, Clone)]
+pub struct CelestiaConfig {
+    pub connection_string: String,
+    pub start_height: u64,
+    pub snark_namespace_id: String,
+    pub operation_namespace_id: Option<String>,
+}
+
+impl Default for CelestiaConfig {
+    fn default() -> Self {
+        CelestiaConfig {
+            connection_string: "ws://localhost:26658".to_string(),
+            start_height: 0,
+            snark_namespace_id: "00000000000000de1008".to_string(),
+            operation_namespace_id: Some("00000000000000de1009".to_string()),
+        }
+    }
+}
+
 pub struct CelestiaConnection {
     pub client: celestia_rpc::Client,
     pub snark_namespace: Namespace,

crates/da/src/lib.rs

@@ -0,0 +1,75 @@
+use anyhow::Result;
+use async_trait::async_trait;
+use ed25519_dalek::{Signature, Signer, SigningKey, VerifyingKey};
+use prism_common::{operation::Operation, tree::Digest};
+use serde::{Deserialize, Serialize};
+use sp1_sdk::SP1ProofWithPublicValues;
+use tokio::sync::broadcast;
+
+pub mod celestia;
+pub mod consts;
+pub mod memory;
+
+// FinalizedEpoch is the data structure that represents the finalized epoch data, and is posted to the DA layer.
+#[derive(Serialize, Deserialize, Clone, Debug)]
+pub struct FinalizedEpoch {
+    pub height: u64,
+    pub prev_commitment: Digest,
+    pub current_commitment: Digest,
+    pub proof: SP1ProofWithPublicValues,
+    pub signature: Option<String>,
+}
+
+impl FinalizedEpoch {
+    pub fn insert_signature(&mut self, key: &SigningKey) {
+        let plaintext = bincode::serialize(&self).unwrap();
+        let signature = key.sign(&plaintext);
+        self.signature = Some(hex::encode(signature.to_bytes()));
+    }
+
+    pub fn verify_signature(&self, vk: VerifyingKey) -> Result<()> {
+        let epoch_without_signature = FinalizedEpoch {
+            height: self.height,
+            prev_commitment: self.prev_commitment,
+            current_commitment: self.current_commitment,
+            proof: self.proof.clone(),
+            signature: None,
+        };
+
+        let message = bincode::serialize(&epoch_without_signature)
+            .map_err(|e| anyhow::anyhow!("Failed to serialize epoch: {}", e))?;
+
+        let signature = self
+            .signature
+            .as_ref()
+            .ok_or_else(|| anyhow::anyhow!("No signature present"))?;
+
+        let signature_bytes = hex::decode(signature)
+            .map_err(|e| anyhow::anyhow!("Failed to decode signature: {}", e))?;
+
+        if signature_bytes.len() != 64 {
+            return Err(anyhow::anyhow!("Invalid signature length"));
+        }
+
+        let signature: Signature = signature_bytes
+            .as_slice()
+            .try_into()
+            .map_err(|_| anyhow::anyhow!("Invalid signature length"))?;
+
+        vk.verify_strict(&message, &signature)
+            .map_err(|e| anyhow::anyhow!("Signature verification failed: {}", e))?;
+        Ok(())
+    }
+}
+
+#[async_trait]
+pub trait DataAvailabilityLayer: Send + Sync {
+    async fn get_latest_height(&self) -> Result<u64>;
+    async fn initialize_sync_target(&self) -> Result<u64>;
+    async fn get_finalized_epoch(&self, height: u64) -> Result<Option<FinalizedEpoch>>;
+    async fn submit_finalized_epoch(&self, epoch: FinalizedEpoch) -> Result<u64>;
+    async fn get_operations(&self, height: u64) -> Result<Vec<Operation>>;
+    async fn submit_operations(&self, operations: Vec<Operation>) -> Result<u64>;
+    async fn start(&self) -> Result<()>;
+    fn subscribe_to_heights(&self) -> broadcast::Receiver<u64>;
+}

crates/prism/src/da/memory.rs → crates/da/src/memory.rs

@@ -1,6 +1,7 @@
-use crate::da::{DataAvailabilityLayer, FinalizedEpoch};
+use crate::{DataAvailabilityLayer, FinalizedEpoch};
 use anyhow::Result;
 use async_trait::async_trait;
+use log::debug;
 use prism_common::operation::Operation;
 use std::{collections::VecDeque, sync::Arc};
 use tokio::{

crates/prism/Cargo.toml

@@ -49,7 +49,9 @@ jmt = { workspace = true }
 sha2 = { workspace = true }
 auto_impl = { workspace = true }
 prism-common = { workspace = true, features = ["test_utils"] }
+prism-storage = { workspace = true }
 prism-errors = { workspace = true }
+prism-da = { workspace = true }
 prism-groth16 = { workspace = true }
 sp1-sdk = { workspace = true }
 

crates/prism/src/cfg.rs

@@ -1,17 +1,20 @@
-use crate::{
-    consts::{DA_RETRY_COUNT, DA_RETRY_INTERVAL},
-    da::memory::InMemoryDataAvailabilityLayer,
-};
 use anyhow::{anyhow, Context, Result};
 use clap::{Parser, Subcommand};
 use config::{builder::DefaultState, ConfigBuilder, File};
 use dirs::home_dir;
 use dotenvy::dotenv;
+use log::{error, warn};
 use prism_errors::{DataAvailabilityError, GeneralError, PrismError};
+use prism_storage::redis::RedisConfig;
 use serde::{Deserialize, Serialize};
 use std::{fs, path::Path, sync::Arc};
 
-use crate::da::{celestia::CelestiaConnection, DataAvailabilityLayer};
+use prism_da::{
+    celestia::{CelestiaConfig, CelestiaConnection},
+    consts::{DA_RETRY_COUNT, DA_RETRY_INTERVAL},
+    memory::InMemoryDataAvailabilityLayer,
+    DataAvailabilityLayer,
+};
 
 #[derive(Clone, Debug, Subcommand, Deserialize)]
 pub enum Commands {
@@ -97,38 +100,6 @@ impl Default for WebServerConfig {
     }
 }
 
-#[derive(Debug, Serialize, Deserialize, Clone)]
-pub struct RedisConfig {
-    pub connection_string: String,
-}
-
-impl Default for RedisConfig {
-    fn default() -> Self {
-        RedisConfig {
-            connection_string: "redis://127.0.0.1/".to_string(),
-        }
-    }
-}
-
-#[derive(Debug, Serialize, Deserialize, Clone)]
-pub struct CelestiaConfig {
-    pub connection_string: String,
-    pub start_height: u64,
-    pub snark_namespace_id: String,
-    pub operation_namespace_id: Option<String>,
-}
-
-impl Default for CelestiaConfig {
-    fn default() -> Self {
-        CelestiaConfig {
-            connection_string: "ws://localhost:26658".to_string(),
-            start_height: 0,
-            snark_namespace_id: "00000000000000de1008".to_string(),
-            operation_namespace_id: Some("00000000000000de1009".to_string()),
-        }
-    }
-}
-
 impl Default for Config {
     fn default() -> Self {
         Config {
@@ -300,7 +271,7 @@ pub async fn initialize_da_layer(
             unreachable!() // This line should never be reached due to the return in the last iteration
         }
         DALayerOption::InMemory => {
-            let (da_layer, _height_rx, _block_rx) = InMemoryDataAvailabilityLayer::new(1);
+            let (da_layer, _height_rx, _block_rx) = InMemoryDataAvailabilityLayer::new(30);
             Ok(Arc::new(da_layer) as Arc<dyn DataAvailabilityLayer + 'static>)
         }
         DALayerOption::None => Err(anyhow!(PrismError::ConfigError(