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

Issue:
WIP Standalone-indexer: support query API in indexer RPC and indexer sdk
## Motivations

<!--
If your feature request is related to a problem, please describe it.
-->

- Would you like to implement this feature? [y/n]

## Solution

<!-- Describe the solution you'd like. -->

- [x] step 1. Move/Add basic document/database query RCP API to indexer
- [x] step 2. Add unit/it for indexer query 
## 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/.github/workflows/ci.yml b/.github/workflows/ci.yml
index aac257c2..8955c1f8 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -74,6 +74,8 @@ jobs:
           name: db3-log-artifact
           path: |
             tools/db3.log
+            tools/indexer.log
+            tools/tm.log
       - uses: codecov/codecov-action@v3
         with:
           token: ${{secrets.COV_TOKEN}}
diff --git a/src/node/src/command.rs b/src/node/src/command.rs
index ccafbb65..e66f975c 100644
--- a/src/node/src/command.rs
+++ b/src/node/src/command.rs
@@ -18,7 +18,7 @@
 use crate::abci_impl::AbciImpl;
 use crate::auth_storage::AuthStorage;
 use crate::context::Context;
-use crate::indexer_impl::IndexerImpl;
+use crate::indexer_impl::{IndexerBlockSyncer, IndexerNodeImpl};
 use crate::json_rpc_impl;
 use crate::node_storage::NodeStorage;
 use crate::storage_node_impl::StorageNodeImpl;
@@ -36,6 +36,7 @@ use db3_faucet::{
 };
 use db3_proto::db3_event_proto::{EventMessage, Subscription};
 use db3_proto::db3_faucet_proto::faucet_node_server::FaucetNodeServer;
+use db3_proto::db3_indexer_proto::indexer_node_server::IndexerNodeServer;
 use db3_proto::db3_node_proto::storage_node_client::StorageNodeClient;
 use db3_proto::db3_node_proto::storage_node_server::StorageNodeServer;
 use db3_sdk::mutation_sdk::MutationSDK;
@@ -130,6 +131,12 @@ pub enum DB3Command {
     /// Start db3 indexer
     #[clap(name = "indexer")]
     Indexer {
+        /// Bind the gprc server to this .
+        #[clap(long, default_value = "127.0.0.1")]
+        public_host: String,
+        /// The port of grpc api
+        #[clap(long, default_value = "26639")]
+        public_grpc_port: u16,
         /// the db3 storage chain grpc url
         #[clap(
             long = "db3_storage_grpc_url",
@@ -413,6 +420,8 @@ impl DB3Command {
             }
 
             DB3Command::Indexer {
+                public_host,
+                public_grpc_port,
                 db3_storage_grpc_url,
                 db_path,
                 db_tree_level_in_memory,
@@ -445,8 +454,29 @@ impl DB3Command {
                     }
                     _ => todo!(),
                 }
-                let mut indexer_impl = IndexerImpl::new(ctx.store_sdk.unwrap(), node_store);
-                indexer_impl.start().await.unwrap();
+                let indexer_node_impl = IndexerNodeImpl::new(node_store.clone());
+                let addr = format!("{public_host}:{public_grpc_port}");
+                let listen = tokio::spawn(async move {
+                    info!("start db3 indexer listener ...");
+                    let mut indexer_block_syncer =
+                        IndexerBlockSyncer::new(ctx.store_sdk.unwrap(), node_store.clone());
+                    indexer_block_syncer.start().await.unwrap();
+                });
+                info!("start db3 indexer node on public addr {}", addr);
+                let cors_layer = CorsLayer::new()
+                    .allow_methods([Method::GET, Method::POST, Method::OPTIONS])
+                    .allow_headers(Any)
+                    .allow_origin(Any);
+                Server::builder()
+                    .accept_http1(true)
+                    .layer(cors_layer)
+                    .layer(tonic_web::GrpcWebLayer::new())
+                    .add_service(IndexerNodeServer::new(indexer_node_impl))
+                    .serve(addr.parse().unwrap())
+                    .await
+                    .unwrap();
+                let (r1,) = tokio::join!(listen);
+                r1.unwrap();
                 info!("exit standalone indexer")
             }
 
diff --git a/src/node/src/indexer_impl.rs b/src/node/src/indexer_impl.rs
index 6efb8895..913e6733 100644
--- a/src/node/src/indexer_impl.rs
+++ b/src/node/src/indexer_impl.rs
@@ -2,24 +2,31 @@ use crate::auth_storage::AuthStorage;
 use crate::mutation_utils::MutationUtil;
 use crate::node_storage::NodeStorage;
 use chrono::Utc;
-use db3_crypto::{db3_address::DB3Address as AccountAddress, id::TxId};
+use db3_crypto::db3_address::DB3Address;
+use db3_crypto::id::{DbId, DocumentId, TxId};
 use db3_proto::db3_event_proto::event_message;
 use db3_proto::db3_event_proto::EventMessage;
+use db3_proto::db3_indexer_proto::indexer_node_server::IndexerNode;
+use db3_proto::db3_indexer_proto::{
+    GetDocumentRequest, GetDocumentResponse, IndexerStatus, RunQueryRequest, RunQueryResponse,
+    ShowDatabaseRequest, ShowDatabaseResponse, ShowIndexerStatusRequest,
+};
 use db3_proto::db3_mutation_proto::{DatabaseAction, DatabaseMutation, PayloadType, WriteRequest};
 use db3_sdk::store_sdk::StoreSDK;
 use prost::Message;
 use std::pin::Pin;
+use std::sync::atomic::Ordering;
 use std::sync::{Arc, Mutex};
 use tendermint::block;
-use tonic::Status;
+use tonic::{Request, Response, Status};
 use tracing::{debug, info, warn};
 
-pub struct IndexerImpl {
+pub struct IndexerBlockSyncer {
     store_sdk: StoreSDK,
     node_store: Arc<Mutex<Pin<Box<NodeStorage>>>>,
 }
 
-impl IndexerImpl {
+impl IndexerBlockSyncer {
     pub fn new(store_sdk: StoreSDK, node_store: Arc<Mutex<Pin<Box<NodeStorage>>>>) -> Self {
         Self {
             store_sdk,
@@ -27,22 +34,22 @@ impl IndexerImpl {
         }
     }
 
-    /// start standalone indexer
+    /// start standalone indexer block syncer
     /// 1. subscribe db3 event
-    /// 2. handle event
+    /// 2. handle event to sync db3 node block
     pub async fn start(&mut self) -> std::result::Result<(), Status> {
-        info!("[Indexer] start indexer ...");
-        info!("[Indexer] subscribe event from db3 network");
+        info!("[IndexerBlockSyncer] start indexer ...");
+        info!("[IndexerBlockSyncer] subscribe event from db3 network");
         let mut stream = self
             .store_sdk
             .subscribe_event_message(true)
             .await?
             .into_inner();
         info!("listen and handle event message");
-        while let Some(event) = stream.message().await? {
+        while let Some(event) = stream.message().await.unwrap() {
             match self.handle_event(event).await {
                 Err(e) => {
-                    info!("[Indexer] handle event error: {:?}", e);
+                    info!("[IndexerBlockSyncer] handle event error: {:?}", e);
                 }
                 _ => {}
             }
@@ -55,7 +62,7 @@ impl IndexerImpl {
         match event.event {
             Some(event_message::Event::BlockEvent(be)) => {
                 info!(
-                    "[Indexer] Receive BlockEvent: Block\t{}\t0x{}\t0x{}\t{}",
+                    "[IndexerBlockSyncer] Receive BlockEvent: Block\t{}\t0x{}\t0x{}\t{}",
                     be.height,
                     hex::encode(be.block_hash),
                     hex::encode(be.app_hash),
@@ -77,27 +84,25 @@ impl IndexerImpl {
                             block.header.height.value(),
                             Utc::now().timestamp() as u64,
                         );
-                        let mut pending_databases: Vec<(AccountAddress, DatabaseMutation, TxId)> =
+                        let mut pending_databases: Vec<(DB3Address, 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?;
+                        Self::parse_and_pending_mutations(&block.data, &mut pending_databases)?;
+                        Self::apply_database_mutations(store.get_auth_store(), &pending_databases)?;
                         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: {:?}",
+                            "[IndexerBlockSyncer] last block state: {:?}",
                             store.get_auth_store().get_last_block_state()
                         );
                         info!(
-                            "[Indexer] indexer state: {:?}",
+                            "[IndexerBlockSyncer] indexer state: {:?}",
                             store.get_auth_store().get_state()
                         );
                     }
                     Err(e) => {
-                        warn!("[Indexer] get node store error: {:?}", e);
+                        warn!("[IndexerBlockSyncer] get node store error: {:?}", e);
                         return Err(Status::internal(format!("get node store error: {:?}", e)));
                     }
                 }
@@ -106,9 +111,9 @@ impl IndexerImpl {
         }
         Ok(())
     }
-    async fn parse_and_pending_mutations(
+    fn parse_and_pending_mutations(
         txs: &Vec<Vec<u8>>,
-        database_mutations: &mut Vec<(AccountAddress, DatabaseMutation, TxId)>,
+        database_mutations: &mut Vec<(DB3Address, DatabaseMutation, TxId)>,
     ) -> Result<(), Status> {
         for tx in txs {
             let tx_id = TxId::from(tx.as_ref());
@@ -132,31 +137,31 @@ impl IndexerImpl {
                             }
                         }
                         PayloadType::QuerySessionPayload => {
-                            debug!("[Indexer] Skip QuerySessionPayload");
+                            debug!("[IndexerBlockSyncer] Skip QuerySessionPayload");
                         }
                         PayloadType::MintCreditsPayload => {
-                            debug!("[Indexer] Skip MintCreditsPayload");
+                            debug!("[IndexerBlockSyncer] Skip MintCreditsPayload");
                         }
                         _ => {
-                            debug!("[Indexer] Skip other payload type");
+                            debug!("[IndexerBlockSyncer] Skip other payload type");
                         }
                     },
                     Err(e) => {
-                        warn!("[Indexer] invalid write request: {:?}", e);
+                        warn!("[IndexerBlockSyncer] invalid write request: {:?}", e);
                         return Err(Status::internal(format!("invalid write request: {:?}", e)));
                     }
                 },
                 Err(e) => {
-                    warn!("[Indexer] invalid write request: {:?}", e);
+                    warn!("[IndexerBlockSyncer] invalid write request: {:?}", e);
                     return Err(Status::internal(format!("invalid write request: {:?}", e)));
                 }
             }
         }
         Ok(())
     }
-    async fn apply_database_mutations(
+    fn apply_database_mutations(
         auth_store: &mut AuthStorage,
-        pending_databases: &Vec<(AccountAddress, DatabaseMutation, TxId)>,
+        pending_databases: &Vec<(DB3Address, DatabaseMutation, TxId)>,
     ) -> std::result::Result<(), Status> {
         info!(
             "Pending database mutations queue size: {}",
@@ -188,7 +193,122 @@ impl IndexerImpl {
         Ok(())
     }
 }
+pub struct IndexerNodeImpl {
+    node_store: Arc<Mutex<Pin<Box<NodeStorage>>>>,
+}
+impl IndexerNodeImpl {
+    pub fn new(node_store: Arc<Mutex<Pin<Box<NodeStorage>>>>) -> Self {
+        Self { node_store }
+    }
+}
+#[tonic::async_trait]
+impl IndexerNode for IndexerNodeImpl {
+    /// show indexer statuc
+    async fn show_indexer_status(
+        &self,
+        _request: Request<ShowIndexerStatusRequest>,
+    ) -> Result<Response<IndexerStatus>, Status> {
+        match self.node_store.lock() {
+            Ok(node_store) => {
+                let state = node_store.get_state();
+                let status = IndexerStatus {
+                    total_database_count: state.total_database_count.load(Ordering::Relaxed),
+                    total_collection_count: state.total_collection_count.load(Ordering::Relaxed),
+                    total_document_count: state.total_document_count.load(Ordering::Relaxed),
+                    total_account_count: state.total_account_count.load(Ordering::Relaxed),
+                    total_mutation_count: state.total_mutation_count.load(Ordering::Relaxed),
+                    total_storage_in_bytes: state.total_storage_bytes.load(Ordering::Relaxed),
+                };
+                Ok(Response::new(status))
+            }
+            Err(e) => Err(Status::internal(format!("{}", e))),
+        }
+    }
+
+    /// show databases info
+    async fn show_database(
+        &self,
+        request: Request<ShowDatabaseRequest>,
+    ) -> std::result::Result<Response<ShowDatabaseResponse>, Status> {
+        let show_database_req = request.into_inner();
+        match self.node_store.lock() {
+            Ok(mut node_store) => {
+                if show_database_req.address.len() > 0 {
+                    // get database id
+                    let address_ref: &str = show_database_req.address.as_ref();
+                    let db_id = DbId::try_from(address_ref)
+                        .map_err(|e| Status::internal(format!("invalid database address {e}")))?;
+                    if let Some(db) = node_store
+                        .get_auth_store()
+                        .get_database(&db_id)
+                        .map_err(|e| Status::internal(format!("{:?}", e)))?
+                    {
+                        Ok(Response::new(ShowDatabaseResponse { dbs: vec![db] }))
+                    } else {
+                        Ok(Response::new(ShowDatabaseResponse { dbs: vec![] }))
+                    }
+                } else {
+                    let address_ref: &str = show_database_req.owner_address.as_str();
+                    let owner = DB3Address::try_from(address_ref)
+                        .map_err(|e| Status::internal(format!("invalid database address {e}")))?;
+                    let dbs = node_store
+                        .get_auth_store()
+                        .get_my_database(&owner)
+                        .map_err(|e| Status::internal(format!("{:?}", e)))?;
+                    Ok(Response::new(ShowDatabaseResponse { dbs }))
+                }
+            }
+            Err(e) => Err(Status::internal(format!("Fail to get lock {}", e))),
+        }
+    }
+
+    /// get document with given id
+    async fn get_document(
+        &self,
+        request: Request<GetDocumentRequest>,
+    ) -> std::result::Result<Response<GetDocumentResponse>, Status> {
+        let get_document_request = request.into_inner();
+        let id = DocumentId::try_from_base64(get_document_request.id.as_str())
+            .map_err(|e| Status::internal(format!("{:?}", e)))?;
+        match self.node_store.lock() {
+            Ok(mut node_store) => {
+                // get database id
+                match node_store.get_auth_store().get_document(&id) {
+                    Ok(document) => Ok(Response::new(GetDocumentResponse { document })),
+                    Err(e) => Err(Status::internal(format!("fail to get document {:?}", e))),
+                }
+            }
+            Err(e) => Err(Status::internal(format!("Fail to get lock {}", e))),
+        }
+    }
 
+    /// run document query with structure query
+    async fn run_query(
+        &self,
+        request: Request<RunQueryRequest>,
+    ) -> std::result::Result<Response<RunQueryResponse>, Status> {
+        let run_query_req = request.into_inner();
+        match self.node_store.lock() {
+            Ok(mut node_store) => {
+                // get database id
+                let address_ref: &str = run_query_req.address.as_ref();
+                let db_id = DbId::try_from(address_ref)
+                    .map_err(|e| Status::internal(format!("invalid database address {e}")))?;
+                match &run_query_req.query {
+                    Some(query) => {
+                        let documents = node_store
+                            .get_auth_store()
+                            .run_query(&db_id, &query)
+                            .map_err(|e| Status::internal(format!("{:?}", e)))?;
+                        Ok(Response::new(RunQueryResponse { documents }))
+                    }
+                    None => return Err(Status::internal("Fail to run with none query")),
+                }
+            }
+            Err(e) => Err(Status::internal(format!("Fail to get lock {}", e))),
+        }
+    }
+}
 #[cfg(test)]
 mod tests {
     use super::*;
diff --git a/src/proto/build.rs b/src/proto/build.rs
index 194f5e9a..c34a67f6 100644
--- a/src/proto/build.rs
+++ b/src/proto/build.rs
@@ -32,6 +32,7 @@ fn main() {
                 "proto/db3_message.proto",
                 "proto/db3_faucet.proto",
                 "proto/db3_event.proto",
+                "proto/db3_indexer.proto",
             ],
             &["proto"],
         )
diff --git a/src/proto/proto/db3_indexer.proto b/src/proto/proto/db3_indexer.proto
new file mode 100644
index 00000000..db46a0df
--- /dev/null
+++ b/src/proto/proto/db3_indexer.proto
@@ -0,0 +1,73 @@
+//
+//
+// db3_node.proto
+// Copyright (C) 2022 db3.network Author imotai <codego.me@gmail.com>
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+syntax = "proto3";
+import "db3_database.proto";
+package db3_indexer_proto;
+
+message IndexerStatus {
+    uint64 total_database_count = 1;
+    uint64 total_collection_count = 2;
+    uint64 total_document_count = 3;
+    uint64 total_account_count = 4;
+    uint64 total_mutation_count = 5;
+    uint64 total_storage_in_bytes = 7;
+}
+
+message ShowDatabaseRequest {
+    // the database address in hex format
+    string address = 2;
+    // the owner address in hex format
+    string owner_address = 3;
+}
+
+message ShowDatabaseResponse {
+    repeated db3_database_proto.Database dbs = 1;
+}
+
+message RunQueryRequest {
+    // a hex string
+    string address = 2;
+    db3_database_proto.StructuredQuery query = 3;
+    // TODO: support paging setting (e.g page_size, page_token)
+    // TODO: support document mask
+}
+message GetDocumentRequest {
+    // a hex string
+    string id = 2;
+}
+
+message GetDocumentResponse {
+    db3_database_proto.Document document = 1;
+}
+
+message RunQueryResponse{
+    repeated db3_database_proto.Document documents = 2;
+    // TODO: support next_page_token for page loading
+}
+message ShowIndexerStatusRequest {}
+
+service IndexerNode {
+    // method for show indexer status
+    rpc ShowIndexerStatus(ShowIndexerStatusRequest) returns (IndexerStatus){}
+    // method for show database
+    rpc ShowDatabase(ShowDatabaseRequest) returns (ShowDatabaseResponse) {}
+    // method for query document
+    rpc RunQuery(RunQueryRequest) returns (RunQueryResponse) {}
+    // method for get document
+    rpc GetDocument(GetDocumentRequest) returns (GetDocumentResponse) {}
+}
diff --git a/src/proto/src/lib.rs b/src/proto/src/lib.rs
index d1392135..473a399c 100644
--- a/src/proto/src/lib.rs
+++ b/src/proto/src/lib.rs
@@ -46,3 +46,6 @@ pub mod db3_faucet_proto {
 pub mod db3_event_proto {
     tonic::include_proto!("db3_event_proto");
 }
+pub mod db3_indexer_proto {
+    tonic::include_proto!("db3_indexer_proto");
+}
diff --git a/src/sdk/src/indexer_sdk.rs b/src/sdk/src/indexer_sdk.rs
new file mode 100644
index 00000000..f0943d18
--- /dev/null
+++ b/src/sdk/src/indexer_sdk.rs
@@ -0,0 +1,271 @@
+//
+// bill_sdk.rs
+// Copyright (C) 2022 db3.network Author imotai <codego.me@gmail.com>
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+
+use db3_proto::db3_database_proto::structured_query::{Limit, Projection};
+use db3_proto::db3_database_proto::{Database, Document, StructuredQuery};
+use db3_proto::db3_indexer_proto::{
+    indexer_node_client::IndexerNodeClient, GetDocumentRequest, IndexerStatus, RunQueryRequest,
+    RunQueryResponse, ShowDatabaseRequest, ShowIndexerStatusRequest,
+};
+use std::sync::Arc;
+use tonic::Status;
+
+pub struct IndexerSDK {
+    client: Arc<IndexerNodeClient<tonic::transport::Channel>>,
+}
+
+impl IndexerSDK {
+    pub fn new(client: Arc<IndexerNodeClient<tonic::transport::Channel>>) -> Self {
+        Self { client }
+    }
+
+    /// show document with given db addr and collection name
+    pub async fn list_documents(
+        &mut self,
+        addr: &str,
+        collection_name: &str,
+        limit: Option<i32>,
+    ) -> std::result::Result<RunQueryResponse, Status> {
+        self.run_query(
+            addr,
+            StructuredQuery {
+                collection_name: collection_name.to_string(),
+                limit: match limit {
+                    Some(v) => Some(Limit { limit: v }),
+                    None => None,
+                },
+                select: Some(Projection { fields: vec![] }),
+                r#where: None,
+            },
+        )
+        .await
+    }
+
+    /// get the document with a base64 format id
+    pub async fn get_document(
+        &mut self,
+        id: &str,
+    ) -> std::result::Result<Option<Document>, Status> {
+        let r = GetDocumentRequest { id: id.to_string() };
+
+        let request = tonic::Request::new(r);
+        let mut client = self.client.as_ref().clone();
+        let response = client.get_document(request).await?.into_inner();
+        Ok(response.document)
+    }
+
+    ///
+    /// get the information of database with a hex format address
+    ///
+    pub async fn get_database(
+        &mut self,
+        addr: &str,
+    ) -> std::result::Result<Option<Database>, Status> {
+        let r = ShowDatabaseRequest {
+            address: addr.to_string(),
+            owner_address: "".to_string(),
+        };
+        let request = tonic::Request::new(r);
+        let mut client = self.client.as_ref().clone();
+        let response = client.show_database(request).await?.into_inner();
+        if response.dbs.len() > 0 {
+            Ok(Some(response.dbs[0].clone()))
+        } else {
+            Ok(None)
+        }
+    }
+
+    ///
+    /// get the information of database with a hex format address
+    ///
+    pub async fn get_my_database(
+        &mut self,
+        addr: &str,
+    ) -> std::result::Result<Vec<Database>, Status> {
+        let r = ShowDatabaseRequest {
+            address: "".to_string(),
+            owner_address: addr.to_string(),
+        };
+        let request = tonic::Request::new(r);
+        let mut client = self.client.as_ref().clone();
+        let response = client.show_database(request).await?.into_inner();
+        Ok(response.dbs)
+    }
+
+    /// query the document with structure query
+    pub async fn run_query(
+        &mut self,
+        addr: &str,
+        query: StructuredQuery,
+    ) -> std::result::Result<RunQueryResponse, Status> {
+        let r = RunQueryRequest {
+            address: addr.to_string(),
+            query: Some(query),
+        };
+        let request = tonic::Request::new(r);
+        let mut client = self.client.as_ref().clone();
+        let response = client.run_query(request).await?.into_inner();
+        Ok(response)
+    }
+
+    pub async fn get_state(&self) -> std::result::Result<IndexerStatus, Status> {
+        let r = ShowIndexerStatusRequest {};
+        let request = tonic::Request::new(r);
+        let mut client = self.client.as_ref().clone();
+        let status = client.show_indexer_status(request).await?.into_inner();
+        Ok(status)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+
+    use super::*;
+    use crate::mutation_sdk::MutationSDK;
+    use crate::sdk_test;
+
+    use db3_proto::db3_database_proto::structured_query::field_filter::Operator;
+    use db3_proto::db3_database_proto::structured_query::filter::FilterType;
+    use db3_proto::db3_database_proto::structured_query::value::ValueType;
+    use db3_proto::db3_database_proto::structured_query::{FieldFilter, Filter, Projection, Value};
+    use db3_proto::db3_indexer_proto::indexer_node_client::IndexerNodeClient;
+    use db3_proto::db3_node_proto::storage_node_client::StorageNodeClient;
+    use std::sync::Arc;
+    use std::time;
+    use tendermint::block;
+    use tonic::transport::Endpoint;
+    use uuid::Uuid;
+
+    async fn run_doc_crud_happy_path(
+        storage_client: Arc<StorageNodeClient<tonic::transport::Channel>>,
+        indexer_client: Arc<IndexerNodeClient<tonic::transport::Channel>>,
+        counter: i64,
+    ) {
+        let (addr1, signer) = sdk_test::gen_secp256k1_signer(counter);
+        let msdk = MutationSDK::new(storage_client.clone(), signer, true);
+        let dm = sdk_test::create_a_database_mutation();
+        let result = msdk.submit_database_mutation(&dm).await;
+        assert!(result.is_ok(), "{:?}", result.err());
+        let sleep_seconds = time::Duration::from_millis(3000);
+        std::thread::sleep(sleep_seconds);
+        // add a collection
+        let (db_id, _) = result.unwrap();
+        println!("db id {}", db_id.to_hex());
+        let cm = sdk_test::create_a_collection_mutataion("collection1", db_id.address());
+        let result = msdk.submit_database_mutation(&cm).await;
+        assert!(result.is_ok());
+        std::thread::sleep(sleep_seconds);
+        let (addr, signer) = sdk_test::gen_secp256k1_signer(counter);
+        let mut sdk = IndexerSDK::new(indexer_client.clone());
+        let my_dbs = sdk.get_my_database(addr1.to_hex().as_str()).await.unwrap();
+        assert_eq!(true, my_dbs.len() > 0);
+        let database = sdk.get_database(db_id.to_hex().as_str()).await;
+        if let Ok(Some(db)) = database {
+            assert_eq!(&db.address, db_id.address().as_ref());
+            assert_eq!(&db.sender, addr.as_ref());
+            assert_eq!(db.tx.len(), 2);
+            assert_eq!(db.collections.len(), 1);
+        } else {
+            assert!(false);
+        }
+        // add 4 documents
+        let docm = sdk_test::add_documents(
+            "collection1",
+            db_id.address(),
+            &vec![
+                r#"{"name": "John Doe","age": 43,"phones": ["+44 1234567","+44 2345678"]}"#,
+                r#"{"name": "Mike","age": 44,"phones": ["+44 1234567","+44 2345678"]}"#,
+                r#"{"name": "Bill","age": 44,"phones": ["+44 1234567","+44 2345678"]}"#,
+                r#"{"name": "Bill","age": 45,"phones": ["+44 1234567","+44 2345678"]}"#,
+            ],
+        );
+        let result = msdk.submit_database_mutation(&docm).await;
+        assert!(result.is_ok());
+        std::thread::sleep(sleep_seconds);
+
+        // show all documents
+        let documents = sdk
+            .list_documents(db_id.to_hex().as_str(), "collection1", None)
+            .await
+            .unwrap();
+        assert_eq!(documents.documents.len(), 4);
+
+        // list documents with limit=3
+        let documents = sdk
+            .list_documents(db_id.to_hex().as_str(), "collection1", Some(3))
+            .await
+            .unwrap();
+        assert_eq!(documents.documents.len(), 3);
+
+        // run query equivalent to SQL: select * from collection1 where name = "Bill"
+        let query = StructuredQuery {
+            collection_name: "collection1".to_string(),
+            select: Some(Projection { fields: vec![] }),
+            r#where: Some(Filter {
+                filter_type: Some(FilterType::FieldFilter(FieldFilter {
+                    field: "name".to_string(),
+                    op: Operator::Equal.into(),
+                    value: Some(Value {
+                        value_type: Some(ValueType::StringValue("Bill".to_string())),
+                    }),
+                })),
+            }),
+            limit: None,
+        };
+        println!("{}", serde_json::to_string(&query).unwrap());
+
+        let documents = sdk.run_query(db_id.to_hex().as_str(), query).await.unwrap();
+        assert_eq!(documents.documents.len(), 2);
+        std::thread::sleep(sleep_seconds);
+    }
+
+    fn create_storage_node_client() -> Arc<StorageNodeClient<tonic::transport::Channel>> {
+        let ep = "http://127.0.0.1:26659";
+        let rpc_endpoint = Endpoint::new(ep.to_string()).unwrap();
+        let channel = rpc_endpoint.connect_lazy();
+        Arc::new(StorageNodeClient::new(channel))
+    }
+
+    fn create_indexer_node_client() -> Arc<IndexerNodeClient<tonic::transport::Channel>> {
+        let ep = "http://127.0.0.1:26639";
+        let rpc_endpoint = Endpoint::new(ep.to_string()).unwrap();
+        let channel = rpc_endpoint.connect_lazy();
+        Arc::new(IndexerNodeClient::new(channel))
+    }
+
+    #[tokio::test]
+    async fn typed_data_doc_curd_happy_path_smoke_test() {
+        run_doc_crud_happy_path(
+            create_storage_node_client(),
+            create_indexer_node_client(),
+            131,
+        )
+        .await;
+    }
+
+    #[tokio::test]
+    async fn indexer_status_test() {
+        let ep = "http://127.0.0.1:26639";
+        let rpc_endpoint = Endpoint::new(ep.to_string()).unwrap();
+        let channel = rpc_endpoint.connect_lazy();
+        let client = Arc::new(IndexerNodeClient::new(channel));
+        let (_addr, signer) = sdk_test::gen_ed25519_signer(150);
+        let sdk = IndexerSDK::new(client.clone());
+        let result = sdk.get_state().await;
+        assert!(result.is_ok());
+    }
+}
diff --git a/src/sdk/src/lib.rs b/src/sdk/src/lib.rs
index 38d6f861..61edbf70 100644
--- a/src/sdk/src/lib.rs
+++ b/src/sdk/src/lib.rs
@@ -16,6 +16,7 @@
 //
 
 pub mod faucet_sdk;
+pub mod indexer_sdk;
 pub mod mutation_sdk;
 #[cfg(test)]
 pub mod sdk_test;
diff --git a/tools/start_localnet.sh b/tools/start_localnet.sh
index 3e713eea..d7e5022d 100644
--- a/tools/start_localnet.sh
+++ b/tools/start_localnet.sh
@@ -45,6 +45,10 @@ sleep 1
 echo "start tendermint node..."
 ./tendermint unsafe_reset_all >> tm.log 2>&1  && ./tendermint start >> tm.log 2>&1 &
 sleep 1
+echo "start db3 indexer..."
+../target/${BUILD_MODE}/db3 indexer >indexer.log 2>&1  &
+sleep 1
+
 
 if [[ $RUN_L1_CHAIN == 'OK' ]]; then
     echo "start evm chain network..."


Test patch:
diff --git a/tools/start_localtestnet.sh b/tools/start_localtestnet.sh
index e5ea66c7..10161fdc 100644
--- a/tools/start_localtestnet.sh
+++ b/tools/start_localtestnet.sh
@@ -71,6 +71,9 @@ export RUST_BACKTRACE=1
 sleep 1
 ./tendermint start --home node3 > tm3.log 2>&1 &
 sleep 1
+echo "start db3 indexer..."
+../target/${BUILD_MODE}/db3 indexer >indexer.log 2>&1  &
+sleep 1
 
 if [[ $RUN_L1_CHAIN == 'OK' ]]; then
     echo "start evm chain network..."