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

Issue:
Standalone indexer: sync the latest block from layer2 in realtime
## Motivations

<!--
If your feature request is related to a problem, please describe it.
-->

This is part of work of **Readonly and Standalone indexer** phase 1

- 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/error/src/lib.rs b/src/error/src/lib.rs
index 35e1add2..70a822ec 100644
--- a/src/error/src/lib.rs
+++ b/src/error/src/lib.rs
@@ -115,6 +115,8 @@ pub enum DB3Error {
     InvalidJson(String),
     #[error("fail to request faucet for {0}")]
     RequestFaucetError(String),
+    #[error("fail to fetch faucet for {0}")]
+    FetchBlockError(String),
 }
 
 pub type Result<T> = std::result::Result<T, DB3Error>;
diff --git a/src/node/src/command.rs b/src/node/src/command.rs
index 8731977d..93e16d0e 100644
--- a/src/node/src/command.rs
+++ b/src/node/src/command.rs
@@ -447,7 +447,7 @@ impl DB3Command {
                 }
                 let mut indexer_impl = IndexerImpl::new(ctx.store_sdk.unwrap(), node_store);
                 indexer_impl.start().await.unwrap();
-                println!("exit standalone indexer")
+                info!("exit standalone indexer")
             }
 
             DB3Command::Client {
diff --git a/src/node/src/indexer_impl.rs b/src/node/src/indexer_impl.rs
index 940f0d3a..d69f8d8c 100644
--- a/src/node/src/indexer_impl.rs
+++ b/src/node/src/indexer_impl.rs
@@ -1,21 +1,51 @@
 use crate::node_storage::NodeStorage;
-use db3_cmd::command::DB3ClientContext;
-use db3_crypto::{db3_address::DB3Address as AccountAddress, id::TxId};
+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 db3_proto::db3_event_proto::event_message;
-use db3_proto::db3_event_proto::mutation_event::MutationEventStatus;
 use db3_proto::db3_event_proto::EventMessage;
-use db3_proto::db3_mutation_proto::DatabaseMutation;
+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;
+use tracing::{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_query_session:
-        Arc<Mutex<Vec<(AccountAddress, AccountAddress, TxId, QuerySessionInfo)>>>,
     pending_databases: Arc<Mutex<Vec<(AccountAddress, DatabaseMutation, TxId)>>>,
 }
 
@@ -24,10 +54,10 @@ impl IndexerImpl {
         Self {
             store_sdk,
             node_store,
-            pending_query_session: Arc::new(Mutex::new(Vec::new())),
             pending_databases: Arc::new(Mutex::new(Vec::new())),
         }
     }
+    parse_mutation!(parse_database_mutation, DatabaseMutation);
 
     /// start standalone indexer
     /// 1. subscribe db3 event
@@ -63,9 +93,135 @@ impl IndexerImpl {
                     hex::encode(be.app_hash),
                     be.gas
                 );
+                let response = self
+                    .store_sdk
+                    .fetch_block_by_height(be.height)
+                    .await
+                    .map_err(|e| Status::internal(format!("fetch block error: {:?}", e)))?;
+
+                info!("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");
+                                }
+                                _ => {
+                                    info!("Skip other payload type");
+                                }
+                            },
+                            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
+                            )));
+                        }
+                    }
+                }
             }
             _ => {}
         }
         Ok(())
     }
+
+    /// 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()))
+                    }
+                }
+                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))
+        }
+    }
 }
diff --git a/src/node/src/storage_node_impl.rs b/src/node/src/storage_node_impl.rs
index 62e58fd5..62d38bd9 100644
--- a/src/node/src/storage_node_impl.rs
+++ b/src/node/src/storage_node_impl.rs
@@ -31,12 +31,12 @@ use db3_proto::db3_event_proto::{
 };
 use db3_proto::db3_mutation_proto::{PayloadType, WriteRequest};
 use db3_proto::db3_node_proto::{
-    storage_node_server::StorageNode, BroadcastRequest, BroadcastResponse, CloseSessionRequest,
-    CloseSessionResponse, GetAccountRequest, GetAccountResponse, GetDocumentRequest,
-    GetDocumentResponse, GetSessionInfoRequest, GetSessionInfoResponse, NetworkStatus,
-    OpenSessionRequest, OpenSessionResponse, QueryBillRequest, QueryBillResponse, RunQueryRequest,
-    RunQueryResponse, ShowDatabaseRequest, ShowDatabaseResponse, ShowNetworkStatusRequest,
-    SubscribeRequest,
+    storage_node_server::StorageNode, BlockRequest, BlockResponse, BroadcastRequest,
+    BroadcastResponse, CloseSessionRequest, CloseSessionResponse, GetAccountRequest,
+    GetAccountResponse, GetDocumentRequest, GetDocumentResponse, GetSessionInfoRequest,
+    GetSessionInfoResponse, NetworkStatus, OpenSessionRequest, OpenSessionResponse,
+    QueryBillRequest, QueryBillResponse, RunQueryRequest, RunQueryResponse, ShowDatabaseRequest,
+    ShowDatabaseResponse, ShowNetworkStatusRequest, SubscribeRequest,
 };
 use db3_proto::db3_session_proto::{OpenSessionPayload, QuerySession, QuerySessionInfo};
 use db3_session::query_session_verifier;
@@ -59,6 +59,7 @@ use tendermint_rpc::{
 
 use std::collections::BTreeMap;
 use std::collections::HashSet;
+use tendermint::block;
 use tokio::sync::mpsc::{Receiver, Sender};
 use tokio::time::{sleep, Duration};
 use tokio_stream::wrappers::ReceiverStream;
@@ -767,6 +768,25 @@ impl StorageNode for StorageNodeImpl {
         }))
     }
 
+    async fn get_block(
+        &self,
+        request: Request<BlockRequest>,
+    ) -> std::result::Result<Response<BlockResponse>, Status> {
+        let r = request.into_inner();
+        let response = self
+            .context
+            .client
+            .block(block::Height::try_from(r.block_height).unwrap())
+            .await
+            .map_err(|e| Status::internal(format!("{}", e)))?;
+
+        let block_encoded =
+            serde_json::to_vec(&response.block).map_err(|e| Status::internal(format!("{}", e)))?;
+        Ok(Response::new(BlockResponse {
+            block_id: response.block_id.hash.as_bytes().to_vec(),
+            block: block_encoded,
+        }))
+    }
     async fn show_network_status(
         &self,
         _request: Request<ShowNetworkStatusRequest>,
diff --git a/src/proto/proto/db3_node.proto b/src/proto/proto/db3_node.proto
index 194dc76d..97d15a01 100644
--- a/src/proto/proto/db3_node.proto
+++ b/src/proto/proto/db3_node.proto
@@ -108,6 +108,21 @@ message BroadcastResponse {
     bytes hash = 2;
 }
 
+enum BlockType {
+    BlockByHeight = 0;
+    BlockByHash = 1;
+    LatestBlock = 2;
+}
+message BlockRequest {
+    BlockType block_type = 1;
+    uint64 block_height= 2;
+    bytes block_hash = 3;
+}
+
+message BlockResponse {
+    bytes block_id = 1;
+    bytes block = 2;
+}
 message ShowDatabaseRequest {
     string session_token = 1;
     // the database address in hex format
@@ -172,4 +187,6 @@ service StorageNode {
     rpc ShowNetworkStatus(ShowNetworkStatusRequest) returns (NetworkStatus){}
     // method for subscribution
     rpc Subscribe(SubscribeRequest) returns (stream db3_event_proto.EventMessage) {}
+    // method for get block
+    rpc GetBlock(BlockRequest) returns (BlockResponse) {}
 }
diff --git a/src/sdk/Cargo.toml b/src/sdk/Cargo.toml
index aa825336..43e034b3 100644
--- a/src/sdk/Cargo.toml
+++ b/src/sdk/Cargo.toml
@@ -27,6 +27,7 @@ num-traits = "^0.2"
 rand = "0.8.5"
 serde_json = "1.0.88"
 hex = "0.4.3"
+tendermint = "0.31.1"
 [dev-dependencies]
 db3-base={path="../base", version="0.1.0"}
 db3-cmd={path="../cmd", version="0.1.0"}
diff --git a/src/sdk/src/store_sdk.rs b/src/sdk/src/store_sdk.rs
index 90e0e606..c2479e76 100644
--- a/src/sdk/src/store_sdk.rs
+++ b/src/sdk/src/store_sdk.rs
@@ -25,10 +25,11 @@ use db3_proto::db3_database_proto::structured_query::{Limit, Projection};
 use db3_proto::db3_database_proto::{Database, Document, StructuredQuery};
 use db3_proto::db3_mutation_proto::PayloadType;
 use db3_proto::db3_node_proto::{
-    storage_node_client::StorageNodeClient, CloseSessionRequest, GetAccountRequest,
-    GetDocumentRequest, GetSessionInfoRequest, NetworkStatus, OpenSessionRequest,
-    OpenSessionResponse, QueryBillKey, QueryBillRequest, RunQueryRequest, RunQueryResponse,
-    SessionIdentifier, ShowDatabaseRequest, ShowNetworkStatusRequest, SubscribeRequest,
+    storage_node_client::StorageNodeClient, BlockRequest, BlockResponse, BlockType,
+    CloseSessionRequest, GetAccountRequest, GetDocumentRequest, GetSessionInfoRequest,
+    NetworkStatus, OpenSessionRequest, OpenSessionResponse, QueryBillKey, QueryBillRequest,
+    RunQueryRequest, RunQueryResponse, SessionIdentifier, ShowDatabaseRequest,
+    ShowNetworkStatusRequest, SubscribeRequest,
 };
 
 use db3_proto::db3_event_proto::{
@@ -701,6 +702,21 @@ impl StoreSDK {
             SessionStatus::from_i32(response.session_status).unwrap(),
         ))
     }
+
+    pub async fn fetch_block_by_height(&self, height: u64) -> Result<BlockResponse, Status> {
+        let request = tonic::Request::new(BlockRequest {
+            block_height: height,
+            block_hash: vec![],
+            block_type: BlockType::BlockByHeight.into(),
+        });
+        let mut client = self.client.as_ref().clone();
+        let response = client
+            .get_block(request)
+            .await
+            .map_err(|e| Status::internal(format!("fail to get block from node service: {e}")))?
+            .into_inner();
+        Ok(response)
+    }
 }
 
 #[cfg(test)]
@@ -721,9 +737,9 @@ mod tests {
     use db3_proto::db3_session_proto::OpenSessionPayload;
     use std::sync::Arc;
     use std::time;
+    use tendermint::block;
     use tonic::transport::Endpoint;
     use uuid::Uuid;
-
     async fn run_get_bills_flow(
         use_typed_format: bool,
         client: Arc<StorageNodeClient<tonic::transport::Channel>>,
@@ -751,6 +767,18 @@ mod tests {
         assert!(result.is_ok());
     }
 
+    async fn run_fetch_block_flow(
+        use_typed_format: bool,
+        client: Arc<StorageNodeClient<tonic::transport::Channel>>,
+        counter: i64,
+    ) {
+        let (_, signer) = sdk_test::gen_secp256k1_signer(counter);
+        let mut sdk = StoreSDK::new(client, signer, use_typed_format);
+        let res = sdk.fetch_block_by_height(1).await;
+        assert!(res.is_ok(), "{:?}", res);
+        let block: block::Block = serde_json::from_slice(res.unwrap().block.as_slice()).unwrap();
+        assert_eq!(block.header.height.value(), 1);
+    }
     #[tokio::test]
     async fn get_bills_smoke_test() {
         let ep = "http://127.0.0.1:26659";
@@ -947,4 +975,14 @@ mod tests {
     /// write a test case for method get_my_database
     #[tokio::test]
     async fn test_get_my_database() {}
+
+    #[tokio::test]
+    async fn fetch_block_by_height() {
+        let ep = "http://127.0.0.1:26659";
+        let rpc_endpoint = Endpoint::new(ep.to_string()).unwrap();
+        let channel = rpc_endpoint.connect_lazy();
+        let client = Arc::new(StorageNodeClient::new(channel));
+
+        run_fetch_block_flow(false, client.clone(), 200).await;
+    }
 }


Test patch: