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/.gitignore b/.gitignore
index b6daef53..de1fda0b 100644
--- a/.gitignore
+++ b/.gitignore
@@ -23,3 +23,4 @@ db3.log
 /bridge/cache/
 /bridge/artifacts/
 indexer.db
+indexer
diff --git a/src/node/Cargo.toml b/src/node/Cargo.toml
index 41dd969e..9c79311d 100644
--- a/src/node/Cargo.toml
+++ b/src/node/Cargo.toml
@@ -57,6 +57,7 @@ ctrlc = { version = "3.0", features = ["termination"] }
 ed25519-dalek = "2.0.0-rc.2"
 dirs = "5.0.0"
 redb = "0.14.0"
+chrono = "0.4.22"
 [build-dependencies]
 shadow-rs = "0.21.0"
 [dev-dependencies]
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/auth_storage.rs b/src/node/src/auth_storage.rs
index 5a2bdb7d..95147e4c 100644
--- a/src/node/src/auth_storage.rs
+++ b/src/node/src/auth_storage.rs
@@ -41,7 +41,7 @@ use tracing::info;
 
 pub const HASH_LENGTH: usize = 32;
 pub type Hash = [u8; HASH_LENGTH];
-
+#[derive(Debug)]
 pub struct NetworkState {
     pub total_storage_bytes: Arc<AtomicU64>,
     pub total_mutation_count: Arc<AtomicU64>,
diff --git a/src/node/src/command.rs b/src/node/src/command.rs
index 93e16d0e..ccafbb65 100644
--- a/src/node/src/command.rs
+++ b/src/node/src/command.rs
@@ -136,7 +136,7 @@ pub enum DB3Command {
             default_value = "http://127.0.0.1:26659"
         )]
         db3_storage_grpc_url: String,
-        #[clap(short, long, default_value = "./indexer.db")]
+        #[clap(short, long, default_value = "./indexer")]
         db_path: String,
         #[clap(long, default_value = "16")]
         db_tree_level_in_memory: u8,
diff --git a/src/node/src/indexer_impl.rs b/src/node/src/indexer_impl.rs
index d69f8d8c..6efb8895 100644
--- a/src/node/src/indexer_impl.rs
+++ b/src/node/src/indexer_impl.rs
@@ -1,52 +1,22 @@
+use crate::auth_storage::AuthStorage;
+use crate::mutation_utils::MutationUtil;
 use crate::node_storage::NodeStorage;
-use bytes::Bytes;
-use db3_crypto::id::AccountId;
-use db3_crypto::{db3_address::DB3Address as AccountAddress, db3_verifier, id::TxId};
-use db3_error::DB3Error;
+use chrono::Utc;
+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_proto::db3_session_proto::QuerySessionInfo;
-use db3_sdk::mutation_sdk::MutationSDK;
 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::{info, warn};
+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(&self, 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>>>>,
-    pending_databases: Arc<Mutex<Vec<(AccountAddress, DatabaseMutation, TxId)>>>,
 }
 
 impl IndexerImpl {
@@ -54,10 +24,8 @@ impl IndexerImpl {
         Self {
             store_sdk,
             node_store,
-            pending_databases: Arc::new(Mutex::new(Vec::new())),
         }
     }
-    parse_mutation!(parse_database_mutation, DatabaseMutation);
 
     /// start standalone indexer
     /// 1. subscribe db3 event
@@ -87,7 +55,7 @@ impl IndexerImpl {
         match event.event {
             Some(event_message::Event::BlockEvent(be)) => {
                 info!(
-                    "Block\t{}\t0x{}\t0x{}\t{}",
+                    "[Indexer] Receive BlockEvent: Block\t{}\t0x{}\t0x{}\t{}",
                     be.height,
                     hex::encode(be.block_hash),
                     hex::encode(be.app_hash),
@@ -99,129 +67,129 @@ impl IndexerImpl {
                     .await
                     .map_err(|e| Status::internal(format!("fetch block error: {:?}", e)))?;
 
-                info!("Block Id: {:?}", response.block_id);
+                debug!("Block Id: {:?}", response.block_id);
                 let block: block::Block =
                     serde_json::from_slice(response.block.as_slice()).unwrap();
-                info!("Block transaction size: {}", block.data.len());
-                for tx in block.data {
-                    let tx_id = TxId::from(tx.as_ref());
-                    let wrequest = WriteRequest::decode(tx.as_ref());
-                    info!("TxId: {}", tx_id.to_base64());
-                    match wrequest {
-                        Ok(req) => match self.unwrap_and_verify(req) {
-                            Ok((data, data_type, account_id)) => match data_type {
-                                PayloadType::DatabasePayload => {
-                                    match self.parse_database_mutation(data.as_ref()) {
-                                        Ok(dm) => match self.pending_databases.lock() {
-                                            Ok(mut s) => {
-                                                let action = DatabaseAction::from_i32(dm.action);
-                                                info!(
-                                                    "Pending database mutation size: {}",
-                                                    s.len()
-                                                );
-                                                info!("Add action: {:?}, mutation : {:?} into pending queue", action, dm);
-                                                s.push((account_id.addr, dm, tx_id));
-                                            }
-                                            _ => {
-                                                let msg = format!("lock pending_databases failed");
-                                                return Err(Status::internal(msg));
-                                            }
-                                        },
-                                        Err(e) => {
-                                            let msg = format!("{e}");
-                                            return Err(Status::internal(msg));
-                                        }
-                                    }
-                                }
-                                PayloadType::QuerySessionPayload => {
-                                    info!("Skip QuerySessionPayload");
-                                }
-                                PayloadType::MintCreditsPayload => {
-                                    info!("Skip MintCreditsPayload");
+                debug!("Block transaction size: {}", block.data.len());
+                match self.node_store.lock() {
+                    Ok(mut store) => {
+                        store.get_auth_store().begin_block(
+                            block.header.height.value(),
+                            Utc::now().timestamp() as u64,
+                        );
+                        let mut pending_databases: Vec<(AccountAddress, DatabaseMutation, TxId)> =
+                            vec![];
+                        Self::parse_and_pending_mutations(&block.data, &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}"))
+                        })?;
+                        // TODO: add show indexer state api
+                        info!(
+                            "[Indexer] last block state: {:?}",
+                            store.get_auth_store().get_last_block_state()
+                        );
+                        info!(
+                            "[Indexer] indexer state: {:?}",
+                            store.get_auth_store().get_state()
+                        );
+                    }
+                    Err(e) => {
+                        warn!("[Indexer] get node store error: {:?}", e);
+                        return Err(Status::internal(format!("get node store error: {:?}", e)));
+                    }
+                }
+            }
+            _ => {}
+        }
+        Ok(())
+    }
+    async fn parse_and_pending_mutations(
+        txs: &Vec<Vec<u8>>,
+        database_mutations: &mut Vec<(AccountAddress, DatabaseMutation, TxId)>,
+    ) -> Result<(), Status> {
+        for tx in txs {
+            let tx_id = TxId::from(tx.as_ref());
+            let wrequest = WriteRequest::decode(tx.as_ref());
+            match wrequest {
+                Ok(req) => match MutationUtil::unwrap_and_verify(req) {
+                    Ok((data, data_type, account_id)) => match data_type {
+                        PayloadType::DatabasePayload => {
+                            match MutationUtil::parse_database_mutation(data.as_ref()) {
+                                Ok(dm) => {
+                                    let action = DatabaseAction::from_i32(dm.action);
+                                    info!(
+                                        "Add account: {}, mutation : {:?}, tx: {} into pending queue",
+                                        account_id.addr.to_hex(), action, tx_id.to_base64());
+                                    database_mutations.push((account_id.addr, dm, tx_id));
                                 }
-                                _ => {
-                                    info!("Skip other payload type");
+                                Err(e) => {
+                                    let msg = format!("{e}");
+                                    return Err(Status::internal(msg));
                                 }
-                            },
-                            Err(e) => {
-                                warn!("invalid write request: {:?}", e);
-                                return Err(Status::internal(format!(
-                                    "invalid write request: {:?}",
-                                    e
-                                )));
                             }
-                        },
-                        Err(e) => {
-                            warn!("invalid write request: {:?}", e);
-                            return Err(Status::internal(format!(
-                                "invalid write request: {:?}",
-                                e
-                            )));
                         }
+                        PayloadType::QuerySessionPayload => {
+                            debug!("[Indexer] Skip QuerySessionPayload");
+                        }
+                        PayloadType::MintCreditsPayload => {
+                            debug!("[Indexer] Skip MintCreditsPayload");
+                        }
+                        _ => {
+                            debug!("[Indexer] Skip other payload type");
+                        }
+                    },
+                    Err(e) => {
+                        warn!("[Indexer] invalid write request: {:?}", e);
+                        return Err(Status::internal(format!("invalid write request: {:?}", e)));
                     }
+                },
+                Err(e) => {
+                    warn!("[Indexer] invalid write request: {:?}", e);
+                    return Err(Status::internal(format!("invalid write request: {:?}", e)));
                 }
             }
-            _ => {}
         }
         Ok(())
     }
+    async fn apply_database_mutations(
+        auth_store: &mut AuthStorage,
+        pending_databases: &Vec<(AccountAddress, DatabaseMutation, TxId)>,
+    ) -> std::result::Result<(), Status> {
+        info!(
+            "Pending database mutations queue size: {}",
+            pending_databases.len()
+        );
 
-    /// unwrap and verify write request
-    /// TODO: duplicate with abci unwrap_and_verify, refactor it to a common/util function
-    fn unwrap_and_verify(
-        &self,
-        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()))
-                    }
+        for (account_addr, mutation, tx_id) in pending_databases {
+            let action = DatabaseAction::from_i32(mutation.action);
+            info!(
+                "apply database mutation transaction tx: {}, mutation: action: {:?}",
+                &tx_id.to_base64(),
+                action
+            );
+            let nonce: u64 = match &mutation.meta {
+                Some(m) => m.nonce,
+                //TODO will not go to here
+                None => 1,
+            };
+            match auth_store.apply_database(&account_addr, nonce, &tx_id, &mutation) {
+                Ok(_) => {
+                    info!("apply transaction {} success", &tx_id.to_base64());
                 }
-                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))
+                Err(e) => {
+                    warn!("fail to apply database for {e}");
+                    return Err(Status::internal(format!("fail to apply database for {e}")));
+                }
+            };
         }
+        Ok(())
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+}
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: