Repository: dbpunk-labs/db3 (language: rust)

Issue:
Standalone indexer: apply mutations from block
## Motivations

<!--
If your feature request is related to a problem, please describe it.
-->
This is part of work of https://github.com/orgs/dbpunk-labs/projects/10#card-89107422

- Would you like to implement this feature? [y/n] y

## Solution

<!-- Describe the solution you'd like. -->

## Alternatives

<!-- Describe any alternative solutions or features you've considered. -->

## Additional context

<!-- Add any other context or screenshots about the feature request here. -->



Discussion / hints:

Fix patch:
diff --git a/src/node/src/abci_impl.rs b/src/node/src/abci_impl.rs
index cd3ff1c1..8f6e8f28 100644
--- a/src/node/src/abci_impl.rs
+++ b/src/node/src/abci_impl.rs
@@ -17,23 +17,18 @@
 
 use shadow_rs::shadow;
 shadow!(build);
+use crate::mutation_utils::MutationUtil;
 use crate::node_storage::NodeStorage;
 use bytes::Bytes;
-use db3_crypto::{
-    db3_address::DB3Address as AccountAddress, db3_verifier, id::AccountId, id::TxId,
-};
-use db3_error::{DB3Error, Result};
+use db3_crypto::{db3_address::DB3Address as AccountAddress, id::AccountId, id::TxId};
 use db3_proto::db3_mutation_proto::{
     DatabaseAction, DatabaseMutation, MintCreditsMutation, PayloadType, WriteRequest,
 };
-use db3_proto::db3_session_proto::{QuerySession, QuerySessionInfo};
+use db3_proto::db3_session_proto::QuerySessionInfo;
 use db3_session::query_session_verifier;
-use ethers::core::types::transaction::eip712::{Eip712, TypedData};
-use ethers::core::types::Bytes as EthersBytes;
 use hex;
 use prost::Message;
 use std::pin::Pin;
-use std::str::FromStr;
 use std::sync::{Arc, Mutex};
 use tendermint_abci::Application;
 use tendermint_proto::abci::{
@@ -43,29 +38,6 @@ use tendermint_proto::abci::{
 };
 use tracing::{info, span, warn, Level};
 
-macro_rules! parse_mutation {
-    ($func:ident, $type:ident) => {
-        fn $func(&self, payload: &[u8]) -> Result<$type> {
-            match $type::decode(payload) {
-                Ok(dm) => match &dm.meta {
-                    Some(_) => Ok(dm),
-                    None => {
-                        warn!("no meta for mutation");
-                        Err(DB3Error::ApplyMutationError("meta is none".to_string()))
-                    }
-                },
-                Err(e) => {
-                    //TODO add event ?
-                    warn!("invalid mutation data {e}");
-                    Err(DB3Error::ApplyMutationError(
-                        "invalid mutation data".to_string(),
-                    ))
-                }
-            }
-        }
-    };
-}
-
 #[derive(Clone)]
 pub struct AbciImpl {
     node_store: Arc<Mutex<Pin<Box<NodeStorage>>>>,
@@ -84,68 +56,6 @@ impl AbciImpl {
             pending_credits: Arc::new(Mutex::new(Vec::new())),
         }
     }
-
-    parse_mutation!(parse_database_mutation, DatabaseMutation);
-    parse_mutation!(parse_mint_credits_mutation, MintCreditsMutation);
-    parse_mutation!(parse_query_session, QuerySession);
-
-    fn unwrap_and_verify(
-        &self,
-        req: WriteRequest,
-    ) -> Result<(EthersBytes, PayloadType, AccountId)> {
-        if req.payload_type == 3 {
-            // typed data
-            match serde_json::from_slice::<TypedData>(req.payload.as_ref()) {
-                Ok(data) => {
-                    let hashed_message = data.encode_eip712().map_err(|e| {
-                        DB3Error::ApplyMutationError(format!("invalid payload type for err {e}"))
-                    })?;
-                    let account_id = db3_verifier::DB3Verifier::verify_hashed(
-                        &hashed_message,
-                        req.signature.as_ref(),
-                    )?;
-                    if let (Some(payload), Some(payload_type)) =
-                        (data.message.get("payload"), data.message.get("payloadType"))
-                    {
-                        //TODO advoid data copy
-                        let data: EthersBytes =
-                            serde_json::from_value(payload.clone()).map_err(|e| {
-                                DB3Error::ApplyMutationError(format!(
-                                    "invalid payload type for err {e}"
-                                ))
-                            })?;
-                        let internal_data_type = i32::from_str(payload_type.as_str().ok_or(
-                            DB3Error::QuerySessionVerifyError("invalid payload type".to_string()),
-                        )?)
-                        .map_err(|e| {
-                            DB3Error::QuerySessionVerifyError(format!(
-                                "fail to convert payload type to i32 {e}"
-                            ))
-                        })?;
-                        let data_type: PayloadType = PayloadType::from_i32(internal_data_type)
-                            .ok_or(DB3Error::ApplyMutationError(
-                                "invalid payload type".to_string(),
-                            ))?;
-                        Ok((data, data_type, account_id))
-                    } else {
-                        Err(DB3Error::ApplyMutationError("bad typed data".to_string()))
-                    }
-                }
-                Err(e) => Err(DB3Error::ApplyMutationError(format!(
-                    "bad typed data for err {e}"
-                ))),
-            }
-        } else {
-            let account_id =
-                db3_verifier::DB3Verifier::verify(req.payload.as_ref(), req.signature.as_ref())?;
-            let data_type: PayloadType = PayloadType::from_i32(req.payload_type).ok_or(
-                DB3Error::ApplyMutationError("invalid payload type".to_string()),
-            )?;
-            let data = Bytes::from(req.payload);
-            Ok((EthersBytes(data), data_type, account_id))
-        }
-    }
-
     fn build_check_response(&self, ok: bool, msg: &str) -> ResponseCheckTx {
         if ok {
             ResponseCheckTx {
@@ -309,10 +219,10 @@ impl Application for AbciImpl {
     fn check_tx(&self, request: RequestCheckTx) -> ResponseCheckTx {
         let wrequest = WriteRequest::decode(request.tx.as_ref());
         match wrequest {
-            Ok(req) => match self.unwrap_and_verify(req) {
+            Ok(req) => match MutationUtil::unwrap_and_verify(req) {
                 Ok((data, data_type, _)) => match data_type {
                     PayloadType::DatabasePayload => {
-                        match self.parse_database_mutation(data.as_ref()) {
+                        match MutationUtil::parse_database_mutation(data.as_ref()) {
                             Ok(_) => {
                                 return self.build_check_response(true, "");
                             }
@@ -324,7 +234,7 @@ impl Application for AbciImpl {
                         }
                     }
                     PayloadType::QuerySessionPayload => {
-                        match self.parse_query_session(data.as_ref()) {
+                        match MutationUtil::parse_query_session(data.as_ref()) {
                             Ok(qs) => {
                                 match query_session_verifier::verify_query_session(
                                     qs.payload.as_ref(),
@@ -348,7 +258,7 @@ impl Application for AbciImpl {
                         }
                     }
                     PayloadType::MintCreditsPayload => {
-                        match self.parse_mint_credits_mutation(data.as_ref()) {
+                        match MutationUtil::parse_mint_credits_mutation(data.as_ref()) {
                             Ok(_) => {
                                 return self.build_check_response(true, "");
                             }
@@ -383,10 +293,10 @@ impl Application for AbciImpl {
         let tx_id = TxId::from(request.tx.as_ref());
         let wrequest = WriteRequest::decode(request.tx.as_ref());
         match wrequest {
-            Ok(req) => match self.unwrap_and_verify(req) {
+            Ok(req) => match MutationUtil::unwrap_and_verify(req) {
                 Ok((data, data_type, account_id)) => match data_type {
                     PayloadType::DatabasePayload => {
-                        match self.parse_database_mutation(data.as_ref()) {
+                        match MutationUtil::parse_database_mutation(data.as_ref()) {
                             Ok(dm) => match self.pending_databases.lock() {
                                 Ok(mut s) => {
                                     let response = self.dispatch_database_event(&account_id, &dm);
@@ -408,7 +318,7 @@ impl Application for AbciImpl {
                         }
                     }
                     PayloadType::QuerySessionPayload => {
-                        match self.parse_query_session(data.as_ref()) {
+                        match MutationUtil::parse_query_session(data.as_ref()) {
                             Ok(qs) => match (
                                 self.pending_query_session.lock(),
                                 query_session_verifier::verify_query_session(
@@ -441,7 +351,7 @@ impl Application for AbciImpl {
                         }
                     }
                     PayloadType::MintCreditsPayload => {
-                        match self.parse_mint_credits_mutation(data.as_ref()) {
+                        match MutationUtil::parse_mint_credits_mutation(data.as_ref()) {
                             Ok(mm) => match self.pending_credits.lock() {
                                 Ok(mut s) => {
                                     s.push((account_id.addr, mm, tx_id));
diff --git a/src/node/src/indexer_impl.rs b/src/node/src/indexer_impl.rs
index ed17dd6f..6efb8895 100644
--- a/src/node/src/indexer_impl.rs
+++ b/src/node/src/indexer_impl.rs
@@ -1,48 +1,19 @@
 use crate::auth_storage::AuthStorage;
+use crate::mutation_utils::MutationUtil;
 use crate::node_storage::NodeStorage;
-use bytes::Bytes;
 use chrono::Utc;
-use db3_crypto::id::AccountId;
-use db3_crypto::{db3_address::DB3Address as AccountAddress, db3_verifier, id::TxId};
-use db3_error::DB3Error;
+use db3_crypto::{db3_address::DB3Address as AccountAddress, id::TxId};
 use db3_proto::db3_event_proto::event_message;
 use db3_proto::db3_event_proto::EventMessage;
 use db3_proto::db3_mutation_proto::{DatabaseAction, DatabaseMutation, PayloadType, WriteRequest};
 use db3_sdk::store_sdk::StoreSDK;
-use ethers::core::types::Bytes as EthersBytes;
-use ethers::types::transaction::eip712::{Eip712, TypedData};
 use prost::Message;
 use std::pin::Pin;
-use std::str::FromStr;
 use std::sync::{Arc, Mutex};
 use tendermint::block;
 use tonic::Status;
 use tracing::{debug, info, warn};
 
-/// parse mutation
-/// TODO duplicate with abci impl. will refactor later
-macro_rules! parse_mutation {
-    ($func:ident, $type:ident) => {
-        fn $func(payload: &[u8]) -> Result<$type, DB3Error> {
-            match $type::decode(payload) {
-                Ok(dm) => match &dm.meta {
-                    Some(_) => Ok(dm),
-                    None => {
-                        warn!("no meta for mutation");
-                        Err(DB3Error::ApplyMutationError("meta is none".to_string()))
-                    }
-                },
-                Err(e) => {
-                    //TODO add event ?
-                    warn!("invalid mutation data {e}");
-                    Err(DB3Error::ApplyMutationError(
-                        "invalid mutation data".to_string(),
-                    ))
-                }
-            }
-        }
-    };
-}
 pub struct IndexerImpl {
     store_sdk: StoreSDK,
     node_store: Arc<Mutex<Pin<Box<NodeStorage>>>>,
@@ -55,7 +26,6 @@ impl IndexerImpl {
             node_store,
         }
     }
-    parse_mutation!(parse_database_mutation, DatabaseMutation);
 
     /// start standalone indexer
     /// 1. subscribe db3 event
@@ -111,11 +81,8 @@ impl IndexerImpl {
                             vec![];
                         Self::parse_and_pending_mutations(&block.data, &mut pending_databases)
                             .await?;
-                        Self::apply_database_mutations(
-                            store.get_auth_store(),
-                            &mut pending_databases,
-                        )
-                        .await?;
+                        Self::apply_database_mutations(store.get_auth_store(), &pending_databases)
+                            .await?;
                         store.get_auth_store().commit().map_err(|e| {
                             Status::internal(format!("fail to commit database for {e}"))
                         })?;
@@ -147,10 +114,10 @@ impl IndexerImpl {
             let tx_id = TxId::from(tx.as_ref());
             let wrequest = WriteRequest::decode(tx.as_ref());
             match wrequest {
-                Ok(req) => match Self::unwrap_and_verify(req) {
+                Ok(req) => match MutationUtil::unwrap_and_verify(req) {
                     Ok((data, data_type, account_id)) => match data_type {
                         PayloadType::DatabasePayload => {
-                            match Self::parse_database_mutation(data.as_ref()) {
+                            match MutationUtil::parse_database_mutation(data.as_ref()) {
                                 Ok(dm) => {
                                     let action = DatabaseAction::from_i32(dm.action);
                                     info!(
@@ -189,7 +156,7 @@ impl IndexerImpl {
     }
     async fn apply_database_mutations(
         auth_store: &mut AuthStorage,
-        pending_databases: &mut Vec<(AccountAddress, DatabaseMutation, TxId)>,
+        pending_databases: &Vec<(AccountAddress, DatabaseMutation, TxId)>,
     ) -> std::result::Result<(), Status> {
         info!(
             "Pending database mutations queue size: {}",
@@ -220,63 +187,6 @@ impl IndexerImpl {
         }
         Ok(())
     }
-    /// unwrap and verify write request
-    /// TODO: duplicate with abci unwrap_and_verify, refactor it to a common/util function
-    fn unwrap_and_verify(
-        req: WriteRequest,
-    ) -> Result<(EthersBytes, PayloadType, AccountId), DB3Error> {
-        if req.payload_type == 3 {
-            // typed data
-            match serde_json::from_slice::<TypedData>(req.payload.as_ref()) {
-                Ok(data) => {
-                    let hashed_message = data.encode_eip712().map_err(|e| {
-                        DB3Error::ApplyMutationError(format!("invalid payload type for err {e}"))
-                    })?;
-                    let account_id = db3_verifier::DB3Verifier::verify_hashed(
-                        &hashed_message,
-                        req.signature.as_ref(),
-                    )?;
-                    if let (Some(payload), Some(payload_type)) =
-                        (data.message.get("payload"), data.message.get("payloadType"))
-                    {
-                        //TODO advoid data copy
-                        let data: EthersBytes =
-                            serde_json::from_value(payload.clone()).map_err(|e| {
-                                DB3Error::ApplyMutationError(format!(
-                                    "invalid payload type for err {e}"
-                                ))
-                            })?;
-                        let internal_data_type = i32::from_str(payload_type.as_str().ok_or(
-                            DB3Error::QuerySessionVerifyError("invalid payload type".to_string()),
-                        )?)
-                        .map_err(|e| {
-                            DB3Error::QuerySessionVerifyError(format!(
-                                "fail to convert payload type to i32 {e}"
-                            ))
-                        })?;
-                        let data_type: PayloadType = PayloadType::from_i32(internal_data_type)
-                            .ok_or(DB3Error::ApplyMutationError(
-                                "invalid payload type".to_string(),
-                            ))?;
-                        Ok((data, data_type, account_id))
-                    } else {
-                        Err(DB3Error::ApplyMutationError("bad typed data".to_string()))
-                    }
-                }
-                Err(e) => Err(DB3Error::ApplyMutationError(format!(
-                    "bad typed data for err {e}"
-                ))),
-            }
-        } else {
-            let account_id =
-                db3_verifier::DB3Verifier::verify(req.payload.as_ref(), req.signature.as_ref())?;
-            let data_type: PayloadType = PayloadType::from_i32(req.payload_type).ok_or(
-                DB3Error::ApplyMutationError("invalid payload type".to_string()),
-            )?;
-            let data = Bytes::from(req.payload);
-            Ok((EthersBytes(data), data_type, account_id))
-        }
-    }
 }
 
 #[cfg(test)]
diff --git a/src/node/src/lib.rs b/src/node/src/lib.rs
index 8fa7c2c4..d9115e2a 100644
--- a/src/node/src/lib.rs
+++ b/src/node/src/lib.rs
@@ -22,6 +22,7 @@ pub mod context;
 pub mod indexer_impl;
 mod json_rpc;
 pub mod json_rpc_impl;
+mod mutation_utils;
 pub mod node_key;
 pub mod node_storage;
 pub mod storage_node_impl;
diff --git a/src/node/src/mutation_utils.rs b/src/node/src/mutation_utils.rs
new file mode 100644
index 00000000..3a02c8d1
--- /dev/null
+++ b/src/node/src/mutation_utils.rs
@@ -0,0 +1,144 @@
+use bytes::Bytes;
+use db3_crypto::db3_verifier;
+use db3_crypto::id::AccountId;
+use db3_error::DB3Error;
+use db3_proto::db3_mutation_proto::{
+    DatabaseMutation, MintCreditsMutation, PayloadType, WriteRequest,
+};
+use db3_proto::db3_session_proto::QuerySession;
+use ethers::core::types::Bytes as EthersBytes;
+use ethers::types::transaction::eip712::{Eip712, TypedData};
+use prost::Message;
+use std::str::FromStr;
+use tracing::warn;
+/// parse mutation
+macro_rules! parse_mutation {
+    ($func:ident, $type:ident) => {
+        pub fn $func(payload: &[u8]) -> Result<$type, DB3Error> {
+            match $type::decode(payload) {
+                Ok(dm) => match &dm.meta {
+                    Some(_) => Ok(dm),
+                    None => {
+                        warn!("no meta for mutation");
+                        Err(DB3Error::ApplyMutationError("meta is none".to_string()))
+                    }
+                },
+                Err(e) => {
+                    //TODO add event ?
+                    warn!("invalid mutation data {e}");
+                    Err(DB3Error::ApplyMutationError(
+                        "invalid mutation data".to_string(),
+                    ))
+                }
+            }
+        }
+    };
+}
+pub struct MutationUtil {}
+
+impl MutationUtil {
+    parse_mutation!(parse_database_mutation, DatabaseMutation);
+    parse_mutation!(parse_mint_credits_mutation, MintCreditsMutation);
+    parse_mutation!(parse_query_session, QuerySession);
+    /// unwrap and verify write request
+    pub fn unwrap_and_verify(
+        req: WriteRequest,
+    ) -> Result<(EthersBytes, PayloadType, AccountId), DB3Error> {
+        if req.payload_type == 3 {
+            // typed data
+            match serde_json::from_slice::<TypedData>(req.payload.as_ref()) {
+                Ok(data) => {
+                    let hashed_message = data.encode_eip712().map_err(|e| {
+                        DB3Error::ApplyMutationError(format!("invalid payload type for err {e}"))
+                    })?;
+                    let account_id = db3_verifier::DB3Verifier::verify_hashed(
+                        &hashed_message,
+                        req.signature.as_ref(),
+                    )?;
+                    if let (Some(payload), Some(payload_type)) =
+                        (data.message.get("payload"), data.message.get("payloadType"))
+                    {
+                        //TODO advoid data copy
+                        let data: EthersBytes =
+                            serde_json::from_value(payload.clone()).map_err(|e| {
+                                DB3Error::ApplyMutationError(format!(
+                                    "invalid payload type for err {e}"
+                                ))
+                            })?;
+                        let internal_data_type = i32::from_str(payload_type.as_str().ok_or(
+                            DB3Error::QuerySessionVerifyError("invalid payload type".to_string()),
+                        )?)
+                        .map_err(|e| {
+                            DB3Error::QuerySessionVerifyError(format!(
+                                "fail to convert payload type to i32 {e}"
+                            ))
+                        })?;
+                        let data_type: PayloadType = PayloadType::from_i32(internal_data_type)
+                            .ok_or(DB3Error::ApplyMutationError(
+                                "invalid payload type".to_string(),
+                            ))?;
+                        Ok((data, data_type, account_id))
+                    } else {
+                        Err(DB3Error::ApplyMutationError("bad typed data".to_string()))
+                    }
+                }
+                Err(e) => Err(DB3Error::ApplyMutationError(format!(
+                    "bad typed data for err {e}"
+                ))),
+            }
+        } else {
+            let account_id =
+                db3_verifier::DB3Verifier::verify(req.payload.as_ref(), req.signature.as_ref())?;
+            let data_type: PayloadType = PayloadType::from_i32(req.payload_type).ok_or(
+                DB3Error::ApplyMutationError("invalid payload type".to_string()),
+            )?;
+            let data = Bytes::from(req.payload);
+            Ok((EthersBytes(data), data_type, account_id))
+        }
+    }
+}
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use bytes::BytesMut;
+    use chrono::Utc;
+    use db3_crypto::db3_signer::Db3MultiSchemeSigner;
+    use db3_proto::db3_base_proto::{BroadcastMeta, ChainId, ChainRole};
+    use db3_proto::db3_mutation_proto::DatabaseAction;
+    fn create_a_database_mutation() -> DatabaseMutation {
+        let meta = BroadcastMeta {
+            //TODO get from network
+            nonce: Utc::now().timestamp() as u64,
+            //TODO use config
+            chain_id: ChainId::DevNet.into(),
+            //TODO use config
+            chain_role: ChainRole::StorageShardChain.into(),
+        };
+        let dm = DatabaseMutation {
+            meta: Some(meta),
+            collection_mutations: vec![],
+            db_address: vec![],
+            action: DatabaseAction::CreateDb.into(),
+            document_mutations: vec![],
+            db_desc: "".to_string(),
+        };
+        dm
+    }
+    #[test]
+    pub fn unwrap_and_verify_ut() {
+        let kp = db3_cmd::keystore::KeyStore::get_keypair(None).unwrap();
+        let signer = Db3MultiSchemeSigner::new(kp);
+        let dm = create_a_database_mutation();
+        let mut mbuf = BytesMut::with_capacity(1024 * 4);
+        dm.encode(&mut mbuf).unwrap();
+        let mbuf = mbuf.freeze();
+        let signature = signer.sign(mbuf.as_ref()).unwrap();
+        let request = WriteRequest {
+            signature: signature.as_ref().to_vec(),
+            payload: mbuf.as_ref().to_vec().to_owned(),
+            payload_type: PayloadType::DatabasePayload.into(),
+        };
+        let (_, payload_type, _) = MutationUtil::unwrap_and_verify(request).unwrap();
+        assert_eq!(PayloadType::DatabasePayload, payload_type);
+    }
+}


Test patch: